# main.py

import sys
import logging
from cliff.app import App
from cliff.commandmanager import CommandManager


class MyApp(App):
    log = logging.getLogger(__name__)

    def __init__(self):
        super(MyApp, self).__init__(
            description='Cliff Demo App',
            version='1.0',
            command_manager=CommandManager('cliff.cliffdemo')
        )

    def initialize_app(self, argv):
        self.LOG.debug('initialize_app')

    def prepare_to_run_command(self, cmd):
        self.LOG.debug('prepare_to_run_command %s' % cmd.__class__.__name__)

    def clean_up(self, cmd, result, err):
        self.LOG.debug('clean_up %s' % cmd.__class__.__name__)
        if err:
            self.LOG.debug('got an error: %s' % err)


def main(argv=sys.argv[1:]):
    my_app = MyApp()
    return my_app.run(argv)


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))

class cliff.app.App(description, 
                    version, 
                    command_manager,
                    stdin=None, 
                    stdout=None, 
                    stderr=None,
                    interactive_app_factory=<class cliff.interactive.InteractiveApp>,
                    deferred_help=False):
    pass

class cliff.commandmanager.CommandManager(namespace,
                                          convert_underscores=True):
    pass

# simple.py

import logging
from cliff.command import Command


class Simple(Command):

    log = logging.getLogger(__name__)

    def take_action(self, parsed_args):
        self.log.info('sending greeting')
        self.log.debug('debugging')
        self.app.stdout.write('Hello world \n')

# setup.py

from setuptools import setup, find_packages

setup(
    name='cliffdemo',
    version='1.0',
    install_requires=['cliff'],
    namespace_packages=[],
    packages=find_packages(),

    entry_points={
        'console_scripts': [
            'cliffdemo = cliffdemo.main:main'
        ],
        'cliff.cliffdemo': [
            'simple = cliffdemo.simple:Simple'
        ]
    }
)

service.py

import os
from oslo_config import cfg

opts = [
    cfg.StrOpt(
        'host',
        default='127.0.0.1',
        help='host of node'
    ),
    cfg.IntOpt(
        'collector_workers',
        default=2,
        help='Number of workers for collector service.'
    ),
]

# 注册配置选项
cfg.CONF.register_opts(opts)

# 将配置选项注册为命令行选项
cli_opts = [
    cfg.StrOpt(
        'os-tenant-id',
        deprecated_group='DEFAULT',
        default=os.environ.get('OS_TENANT_ID', ''),
        help='Tenant ID to use for Openstack service access.'
    ),
    cfg.BoolOpt(
        'insecure',
        default=False,
        help='xxx'
    ),
]

cfg.CONF.register_cli_opts(cli_opts, group='service_credentials')

from oslo_config import cfg
from oslo_config import types

PortType = types.Integer(1, 65535)

common_opts = [
    cfg.Opt(
        'bind_port',
        type=PortType(),
        default=8080,
        help='Port number to listen on.'
    )
]

from olso_config import cfg

cfg.CONF(sys.argv[1:], project='xyz')

--service_credentials-os-tenant-id abc123

from oslo_config import cfg

cfg.CONF.import_opt('host', 'service')
host = cfg.CONF.host

from oslo_db.sqlalchemy import enginefacade


class SomeClass(object):
    def __init__(self, x, y):
        self.x = x
        self.y = y


class MyContext(object):
    pass
    'User-defined context class'


def read(context):
    with enginefacade.reader.using(context) as session:
        return session.query(SomeClass).all()


def write(context, x, y):
    with enginefacade.writer.using(context) as session:
        session.add(SomeClass(x, y))


def main():
    context = MyContext()
    results = read(context)
    write(context, 0, 0)

from oslo_db.sqlalchemy import enginefacade


class SomeClass(object):
    def __init__(self, x, y):
        self.x = x
        self.y = y


@enginefacade.transaction_context_provider
class MyContext(object):
    pass
    'User-defined context class'


@enginefacade.reader
def read(context):
    return context.session.query(SomeClass).all()


@enginefacade.writer
def write(context, x, y):
    context.session.add(SomeClass(x, y))


def main():
    context = MyContext()
    results = read(context)
    write(context, 0, 0)

from oslo_db.sqlalchemy.models import TimestampMixin, ModelBase
from oslo_db.sqlalchemy.models import types
from oslo_db.sqlalchemy.models import Column


class Project(TimestampMixin, ModelBase):
    id = Column(types.Integer, primary_key=True)
    name = Column(types.String)

from oslo_db.sqlalchemy.models import ModelBase, SoftDeleteMixin
from oslo_db.sqlalchemy.models import types
from oslo_db.sqlalchemy.models import Column
from oslo_db.sqlalchemy.utils import model_query

class Bar(SoftDeleteMixin, ModelBase):
    id = Column(types.Integer, primary_key=True)
    name = Column(types.String)


count = model_query(Bar).find(id=1).soft_delete()
if count == 0:
    raise Exception("0 entries were soft deleted")

from oslo_config import cfg
from oslo_db import api as db_api

# 定义不同backend所对应的实现，如果配置选项conf.database.backend的值为sqlalchemy，
# 就用project.db.sqlalchemy.api模块中的实现。

_BACKEND_MAPPING = {
    'sqlalchemy': 'project.db.sqlalchemy.api'
}

IMPL = db_api.DBAPI.from_config(cfg.CONF,
                                backend_mapping=_BACKEND_MAPPING)


def get_engine():
    return IMPL.get_engine()


def get_session():
    return IMPL.get_session()


# DB-API method
def do_something(id):
    return IMPL.do_something(id)

def get_backend():
    return MyImplementationClass

{ 
    'message_id': six.text_type(uuid.uuid4()), # 消息id
    'publisher_id': 'compute.host1'            # 发送者id
    'timestamp': timeutils.utcnow(),           # 时间戳
    'priority': 'WARN',                        # 通知优先级
    'event_type': 'compute.create_instance',   # 通知类型
    'payload': {'instance_id': '123'}          # 通知内容    
}

from oslo import messaging

Class ErrorEndpoint(object):
    def error(self, ctxt, pulisher_id, event_type, payload, metadata):
        do_something(payload)
        return messaging.NotificationResult.HANDLED

# server.py 服务器端

from oslo_config import cfg
import oslo_messaging as messaging
import time


class ServerControllerEndpoint(object):
    target = messaging.Target(
        namespace='control',
        version='2.0'
    )

    def __init__(self, server):
        self.server = server

    def stop(self, context):
        if self.server:
            self.server.stop()


class TestEndpoint(object):
    def test(self, context, arg):
        print arg
        return arg


transport = messaging.get_transport(cfg.CONF)
target = messaging.Target(topic='test', server='server-1')
endpoints = [
    ServerControllerEndpoint(None),
    TestEndpoint()
]
server = messaging.get_rpc_server(
    transport=transport,
    target=target,
    endpoints=endpoints,
    executor='blocking'
)

try:
    server.start()
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print 'Stop server.'

server.wait()

# client.py 客户端

from oslo_config import cfg
import oslo_messaging as messaging

transport = messaging.get_transport(cfg.CONF)
target = messaging.Target(topic='test')
client = messaging.RPCClient(
    transport=transport,
    target=target
)

ret = client.call(
    ctxt={},
    method='test',
    arg={'password': 'Hello world'}

)
cctxt = client.prepare(
    namespace='control',
    version='2.0'
)

cctxt.cast({}, 'stop')

# notification_listener.py 消息通知处理

from oslo_config import cfg
import oslo_messaging as messaging


class NotificationEndpoint(object):
    filter_rule = messaging.NotificationFilter(
        publisher_id='^compute.*'
    )

    def warn(self, context, publisher_id, event_type, payload, metadata):
        print context, publisher_id, event_type, payload, metadata


class ErrorEndpoint(object):
    filter_rule = messaging.NotificationFilter(
        publisher_id='^instance\..*\.start$',
        context={'ctxt_key': 'regexp'}
    )

    def error(self, context, publisher_id, event_type, payload, metadata):
        print context, publisher_id, event_type, payload, metadata


transport = messaging.get_transport(cfg.CONF)
targets = [
    messaging.Target(topic='notifications'),
    messaging.Target(topic='notifications_test')
]
endpoints = [
    NotificationEndpoint(),
    ErrorEndpoint()
]
pool = 'listener-workers'
listener = messaging.get_notification_listener(
    transport=transport,
    targets=targets,
    endpoints=endpoints,
    pool=pool
)
listener.start()
listener.wait()

# notifier_send.py

from oslo_config import cfg
import oslo_messaging as messaging

transport = messaging.get_transport(cfg.CONF)
notifier = messaging.Notifier(
    transport=transport,
    driver='messaging',
    topics=['notifications']
)

notifier_2 = notifier.prepare(publisher_id='compute')
notifier_2.error(
    ctxt={},
    event_type='my_type',
    payload={'content': 'Hello world'}
)

# ceilometer/compute/virt/inspector.py

class Inspector(object):

    def __init__(self, conf):
        self.conf = conf

    def inspect_instance(self, instance, duration):
        """Inspect the CPU statistics for an instance.

        :param instance: the target instance
        :param duration: the last 'n' seconds, over which the value should be
               inspected
        :return: the instance stats
        """
        raise ceilometer.NotImplementedError

from ceilometer.compute.virt import inspector as virt_inspector

class LibvirtInspector(virt_inspector.Inspector):

    def __init__(self, conf):
        super(LibvirtInspector, self).__init__(conf)
        # NOTE(sileht): create a connection on startup
        self.connection

    @property
    def connection(self):
        return libvirt_utils.refresh_libvirt_connection(self.conf, self)

    @libvirt_utils.raise_nodata_if_unsupported
    @libvirt_utils.retry_on_disconnect
    def inspect_instance(self, instance, duration=None):
        domain = self._get_domain_not_shut_off_or_raise(instance)

        memory_used = memory_resident = None
        memory_swap_in = memory_swap_out = None
        memory_stats = domain.memoryStats()
        # Stat provided from libvirt is in KB, converting it to MB.
        if 'available' in memory_stats and 'unused' in memory_stats:
            memory_used = (memory_stats['available'] -
                           memory_stats['unused']) / units.Ki
        if 'rss' in memory_stats:
            memory_resident = memory_stats['rss'] / units.Ki
        if 'swap_in' in memory_stats and 'swap_out' in memory_stats:
            memory_swap_in = memory_stats['swap_in'] / units.Ki
            memory_swap_out = memory_stats['swap_out'] / units.Ki

        # TODO(sileht): stats also have the disk/vnic info
        # we could use that instead of the old method for Queen
        stats = self.connection.domainListGetStats([domain], 0)[0][1]
        cpu_time = 0
        current_cpus = stats.get('vcpu.current')
        # Iterate over the maximum number of CPUs here, and count the
        # actual number encountered, since the vcpu.x structure can
        # have holes according to
        # https://libvirt.org/git/?p=libvirt.git;a=blob;f=src/libvirt-domain.c
        # virConnectGetAllDomainStats()
        for vcpu in six.moves.range(stats.get('vcpu.maximum', 0)):
            try:
                cpu_time += (stats.get('vcpu.%s.time' % vcpu) +
                             stats.get('vcpu.%s.wait' % vcpu))
                current_cpus -= 1
            except TypeError:
                # pass here, if there are too many holes, the cpu count will
                # not match, so don't need special error handling.
                pass

        if current_cpus:
            # There wasn't enough data, so fall back
            cpu_time = stats.get('cpu.time')

        return virt_inspector.InstanceStats(
            cpu_number=stats.get('vcpu.current'),
            cpu_time=cpu_time,
            memory_usage=memory_used,
            memory_resident=memory_resident,
            memory_swap_in=memory_swap_in,
            memory_swap_out=memory_swap_out,
            cpu_cycles=stats.get("perf.cpu_cycles"),
            instructions=stats.get("perf.instructions"),
            cache_references=stats.get("perf.cache_references"),
            cache_misses=stats.get("perf.cache_misses"),
            memory_bandwidth_total=stats.get("perf.mbmt"),
            memory_bandwidth_local=stats.get("perf.mbml"),
            cpu_l3_cache_usage=stats.get("perf.cmt"),
        )

# ceilometer/compute/virt/inspector.py

def get_hypervisor_inspector(conf):
    try:
        namespace = 'ceilometer.compute.virt'
        mgr = driver.DriverManager(namespace,
                                   conf.hypervisor_inspector,
                                   invoke_on_load=True,
                                   invoke_args=(conf, ))
        return mgr.driver
    except ImportError as e:
        LOG.error("Unable to load the hypervisor inspector: %s" % e)
        return Inspector(conf)

from taskflow import engines
from taskflow.patterns import linear_flow as lf
from taskflow import task


class CallJim(task.Task):
    def execute(self, jim_number, *args, **kwargs):
        print 'Calling Jim %s.' % jim_number

    def revert(self, jim_number, *args, **kwargs):
        print 'Calling %s and apologizing.' % jim_number


class CallJoe(task.Task):
    def execute(self, joe_number, *args, **kwargs):
        print 'Calling Joe %s.' % joe_number

    def revert(self, joe_number, *args, **kwargs):
        print 'Calling %s and apologizing.' % joe_number


class CallHY(task.Task):
    def execute(self, hy_number, *args, **kwargs):
        raise IOError('HY not at home right now.')


flow = lf.Flow('simple-linear').add(
    CallJim(),
    CallJoe(),
    CallHY()
)

try:
    engines.run(
        flow=flow,
        engine_conf={'engine': 'serial'},
        store=dict(jim_number=444,
                   joe_number=555,
                   hy_number=666)
    )
except Exception as e:
    print 'Flow failed: %s' % e

class EchoTask(task.Task):
    def execute(self, name, *args, **kwargs):
        print 'my name is %s' % name

    def revert(self, name, *args, **kwargs):
        print 'sorry, my name is %s' % name


flow = lf.Flow('f1').add(
    EchoTask('t1'),
    lf.Flow(
        'f2',
        retry=retry.ForEach(
            values=['a', 'b', 'c'],
            name='r1',
            provides='value'
        )
    ).add(
        EchoTask('t2'),
        EchoTask('t3', requires='value')
    ),
    EchoTask('t4')

)

from taskflow.patterns import graph_flow as gf
from taskflow.patterns import linear_flow as lf
from taskflow import task
from taskflow import engines


class Adder(task.Task):
    def execute(self, x, y, *args, **kwargs):
        return x + y


flow = gf.Flow('root').add(
    lf.Flow('nested_linear').add(
        # 从后台存储中读取名为y3和y4的参数值，并以参数x，y传递给execute方法
        # x2 = y3 + y4
        Adder('add2', provides='x2', rebind=['y3', 'y4']),
        # x1 = y1 + y2
        Adder('add1', provides='x1', rebind=['y1', 'y2']),
    ),
    # x5 = x1 + x3
    Adder('add5', provides='x5', rebind=['x1', 'x3']),
    # x3 = x1 + x2
    Adder('add3', provides='x3', rebind=['x1', 'x2']),
    # x4 = x2 + y5
    Adder('add4', provides='x4', rebind=['x2', 'y5']),
    # x6 = x5 + x4
    Adder('add6', provides='x6', rebind=['x5', 'x4']),
    # x7 = x6 + x6
    Adder('add7', provides='x7', rebind=['x6', 'x6'])
)

store = {
    'y1': 1,
    'y2': 3,
    'y3': 5,
    'y4': 7,
    'y5': 9
}

result = engines.run(
    flow=flow,
    store=store,
    engine_conf='serial'
)

print 'single thread engine result %s' % result

result = engines.run(
    flow=flow,
    store=store,
    engine_conf='parallel'
)

print 'multi thread engine result %s' % result

Single thread engine result {'x1': 4, 'y5': 9, 'y4': 7, 'y1': 1, 'x2': 12, 'x3': 16, 'y3': 5, 'y2': 3, 'x6': 41, 'x7': 82, 'x4': 21, 'x5': 20}.
Multi thread engine result {'x1': 4, 'y5': 9, 'y4': 7, 'y1': 1, 'x2': 12, 'x3': 16, 'y3': 5, 'y2': 3, 'x6': 41, 'x7': 82, 'x4': 21, 'x5': 20}.

Adder().requires

OrderedSet(['x', 'y'])

class MyTask(task.Task):
    def execute(self, spam, eggs=(), *args, **kwargs):
        return spam + eggs


MyTask().requires
OrderedSet(['spam'])

class UniTask(task.Task):
    def execute(self, *args, **kwargs):
        pass


UniTask().requires
OrderedSet([])

class Dog(task.Task):
    def execute(self, food, *args, **kwargs):
        pass


dog = Dog(requires=('water', 'grass'))
dog.requires
OrderedSet(['food', 'water', 'grass'])

from taskflow import task


class SpawnVMTask(task.Task):
    def execute(self, vm_name, vm_image_id, *args, **kwargs):
        pass


# engine执行下面这个task时，会从后台存储中获取到名为'name'的参数值
# 然后把它当做vm_name参数传递给task的execute()方法
s = SpawnVMTask(rebind={'vm_name': 'name'})
s.requires
OrderedSet(['name', 'vm_image_id'])

# engine执行下面这个task时，会从后台存储中获取到名为'name', 'image_id'
# 和'admin_key_name'的参数值，把name和image_id分别当做vm_name
# 和vm_image_id参数，把admin_key_name当做args参数中的某一项传递
# 给task的execute()方法
s = SpawnVMTask(rebind=('name', 'image_id', 'admin_key_name'))
s.requires
OrderedSet(['name', 'image_id', 'admin_key_name'])

def authorize(context, action, target, do_raise=True, exc=None):
    """Verifies that the action is valid on the target in this context.

       :param context: nova context
       :param action: string representing the action to be checked
           this should be colon separated for clarity.
           i.e. ``compute:create_instance``,
           ``compute:attach_volume``,
           ``volume:attach_volume``
       :param target: dictionary representing the object of the action
           for object creation this should be a dictionary representing the
           location of the object e.g. ``{'project_id': context.project_id}``
       :param do_raise: if True (the default), raises PolicyNotAuthorized;
           if False, returns False
       :param exc: Class of the exception to raise if the check fails.
                   Any remaining arguments passed to :meth:`authorize` (both
                   positional and keyword arguments) will be passed to
                   the exception class. If not specified,
                   :class:`PolicyNotAuthorized` will be used.

       :raises nova.exception.PolicyNotAuthorized: if verification fails
           and do_raise is True. Or if 'exc' is specified it will raise an
           exception of that type.

       :return: returns a non-False value (not necessarily "True") if
           authorized, and the exact value False if not authorized and
           do_raise is False.
    """
    init()
    credentials = context.to_policy_values()
    if not exc:
        exc = exception.PolicyNotAuthorized
    try:
        result = _ENFORCER.authorize(action, target, credentials,
                                     do_raise=do_raise, exc=exc, action=action)
    except policy.PolicyNotRegistered:
        with excutils.save_and_reraise_exception():
            LOG.exception(_LE('Policy not registered'))
    except Exception:
        with excutils.save_and_reraise_exception():
            LOG.debug('Policy check for %(action)s failed with credentials '
                      '%(credentials)s',
                      {'action': action, 'credentials': credentials})
    return result