python类StrOpt()的实例源码

def load_repositories_options():
        repo_opts = [
            cfg.StrOpt(
                'offsets',
                default='monasca_transform.offset_specs:JSONOffsetSpecs',
                help='Repository for offset persistence'
            ),
            cfg.StrOpt(
                'data_driven_specs',
                default='monasca_transform.data_driven_specs.'
                        'json_data_driven_specs_repo:JSONDataDrivenSpecsRepo',
                help='Repository for metric and event data_driven_specs'
            ),
            cfg.IntOpt('offsets_max_revisions', default=10,
                       help="Max revisions of offsets for each application")
        ]
        repo_group = cfg.OptGroup(name='repositories', title='repositories')
        cfg.CONF.register_group(repo_group)
        cfg.CONF.register_opts(repo_opts, group=repo_group)

def load_messaging_options():
        messaging_options = [
            cfg.StrOpt('adapter',
                       default='monasca_transform.messaging.adapter:'
                       'KafkaMessageAdapter',
                       help='Message adapter implementation'),
            cfg.StrOpt('topic', default='metrics',
                       help='Messaging topic'),
            cfg.StrOpt('brokers',
                       default='192.168.10.4:9092',
                       help='Messaging brokers'),
            cfg.StrOpt('publish_kafka_project_id',
                       default='111111',
                       help='publish aggregated metrics tenant'),
            cfg.StrOpt('adapter_pre_hourly',
                       default='monasca_transform.messaging.adapter:'
                       'KafkaMessageAdapterPreHourly',
                       help='Message adapter implementation'),
            cfg.StrOpt('topic_pre_hourly', default='metrics_pre_hourly',
                       help='Messaging topic pre hourly')
        ]
        messaging_group = cfg.OptGroup(name='messaging', title='messaging')
        cfg.CONF.register_group(messaging_group)
        cfg.CONF.register_opts(messaging_options, group=messaging_group)

def load_pre_hourly_processor_options():
        app_opts = [
            cfg.IntOpt('late_metric_slack_time', default=600),
            cfg.StrOpt('data_provider',
                       default='monasca_transform.processor.'
                               'pre_hourly_processor:'
                               'PreHourlyProcessorDataProvider'),
            cfg.BoolOpt('enable_instance_usage_df_cache'),
            cfg.StrOpt('instance_usage_df_cache_storage_level'),
            cfg.BoolOpt('enable_batch_time_filtering'),
            cfg.IntOpt('effective_batch_revision', default=2)
        ]
        app_group = cfg.OptGroup(name='pre_hourly_processor',
                                 title='pre_hourly_processor')
        cfg.CONF.register_group(app_group)
        cfg.CONF.register_opts(app_opts, group=app_group)

def list_opts():
    return [
        ('DEFAULT',
         [
             # FIXME(jd) Move to [api]
             cfg.StrOpt('api_paste_config',
                        default="api_paste.ini",
                        help="Configuration file for WSGI definition of API."),
         ]),
        ('api',
         [
             cfg.IntOpt('default_api_return_limit',
                        min=1,
                        default=100,
                        help='Default maximum number of '
                        'items returned by API request.'),
         ]),
        ('database', panko.storage.OPTS),
        ('storage', STORAGE_OPTS),
    ]

def setUp(self):
        super(SqliteConfFixture, self).setUp()

        self.register_opt(cfg.StrOpt('connection', default=''),
                          group='database')
        self.register_opt(cfg.IntOpt('max_pool_size', default=None),
                          group='database')
        self.register_opt(cfg.IntOpt('idle_timeout', default=None),
                          group='database')
        self.register_opts(cli.MIGRATION_OPTS)
        self.url = db_test_utils.get_connect_string("sqlite")
        self.set_default('connection', self.url, group='database')
        self.set_default('disable_microsecond_data_migration', False)
        lockutils.set_defaults(lock_path='/tmp')
        self._drop_db()
        self.addCleanup(self.cleanup)

def _register_opts(self):
        """Register oslo.config options"""
        opts = []
        option = cfg.StrOpt('url', default=None,
                            help=_('API endpoint url'))
        opts.append(option)
        option = cfg.IntOpt('read_timeout', default=5,
                            help=_('API read timeout in seconds'))
        opts.append(option)
        option = cfg.IntOpt('retries', default=3,
                            help=_('API request retries'))
        opts.append(option)

        opt_group = cfg.OptGroup('valet')
        CONF.register_group(opt_group)
        CONF.register_opts(opts, group=opt_group)

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

        self.opt_group = cfg.OptGroup(name='endpoint',
                                 title='Get the endpoints for keystone')

        self.endpoint_opts = [cfg.StrOpt('endpoint', default='None',
            help=('URL or IP address where OpenStack keystone runs.'))
        ]

        CONF = cfg.CONF
        CONF.register_group(self.opt_group)
        CONF.register_opts(self.endpoint_opts, self.opt_group)

        CONF(default_config_files=[self.conf_path])
        self.AUTH_ENDPOINT = CONF.endpoint.endpoint

def _register_opts(self):
        '''Register Options'''
        opts = []
        option = cfg.StrOpt(self.opt_failure_mode_str, choices=['reject', 'yield'], default='reject',
                            help=_('Mode to operate in if Valet planning fails for any reason.'))
        opts.append(option)
        option = cfg.StrOpt(self.opt_project_name_str, default=None, help=_('Valet Project Name'))
        opts.append(option)
        option = cfg.StrOpt(self.opt_username_str, default=None, help=_('Valet Username'))
        opts.append(option)
        option = cfg.StrOpt(self.opt_password_str, default=None, help=_('Valet Password'))
        opts.append(option)
        option = cfg.StrOpt(self.opt_auth_uri_str, default=None, help=_('Keystone Authorization API Endpoint'))
        opts.append(option)

        opt_group = cfg.OptGroup(self.opt_group_str)
        cfg.CONF.register_group(opt_group)
        cfg.CONF.register_opts(opts, group=opt_group)

    # TODO(JD): Factor out common code between this and the cinder filter

def load_config(self):
        """Load config options from nova config file or command line (for example: /etc/nova/nova.conf)

        Sample settings in nova config:
            [kozinaki_EC2]
            user=AKIAJR7NAEIZPWSTFBEQ
            key=zv9zSem8OE+k/axFkPCgZ3z3tLrhvFBaIIa0Ik0j
        """

        provider_opts = [
            cfg.StrOpt('aws_secret_access_key', help='AWS secret key', secret=True),
            cfg.StrOpt('aws_access_key_id', help='AWS access key id', secret=True),
            cfg.StrOpt('region', help='AWS region name'),
        ]

        cfg.CONF.register_opts(provider_opts, self.config_name)
        return cfg.CONF[self.config_name]

def load_config(self):
        """Load config options from nova config file or command line (for example: /etc/nova/nova.conf)

        Sample settings in nova config:
            [kozinaki_EC2]
            user=AKIAJR7NAEIZPWSTFBEQ
            key=zv9zSem8OE+k/axFkPCgZ3z3tLrhvFBaIIa0Ik0j
        """

        provider_opts = [
            cfg.StrOpt('path_to_json_token', help='Google API json token file', secret=True),
            cfg.StrOpt('project', help='Google project id'),
            cfg.StrOpt('zone', help='Google zone name'),
        ]

        cfg.CONF.register_opts(provider_opts, self.config_name)
        return cfg.CONF[self.config_name]

def load_config(self):
        """Load config options from nova config file or command line (for example: /etc/nova/nova.conf)

        Sample settings in nova config:
            [kozinaki_EC2]
            user=AKIAJR7NAEIZPWSTFBEQ
            key=zv9zSem8OE+k/axFkPCgZ3z3tLrhvFBaIIa0Ik0j
        """

        provider_cls = get_libcloud_driver(getattr(Provider, self.provider_name))
        provider_cls_info = inspect.getargspec(provider_cls.__init__)

        provider_opts = [cfg.StrOpt(arg) for arg in provider_cls_info.args]
        provider_opts.append(cfg.StrOpt('location'))
        provider_opts.append(cfg.StrOpt('root_password'))
        provider_opts.append(cfg.StrOpt('project_id'))

        cfg.CONF.register_opts(provider_opts, self.config_name)
        return cfg.CONF[self.config_name]

def load_database_options():
        db_opts = [
            cfg.StrOpt('server_type'),
            cfg.StrOpt('host'),
            cfg.StrOpt('database_name'),
            cfg.StrOpt('username'),
            cfg.StrOpt('password'),
            cfg.BoolOpt('use_ssl', default=False),
            cfg.StrOpt('ca_file')
        ]
        mysql_group = cfg.OptGroup(name='database', title='database')
        cfg.CONF.register_group(mysql_group)
        cfg.CONF.register_opts(db_opts, group=mysql_group)

def load_service_options():
        service_opts = [
            cfg.StrOpt('coordinator_address'),
            cfg.StrOpt('coordinator_group'),
            cfg.FloatOpt('election_polling_frequency'),
            cfg.BoolOpt('enable_debug_log_entries', default='false'),
            cfg.StrOpt('setup_file'),
            cfg.StrOpt('setup_target'),
            cfg.StrOpt('spark_driver'),
            cfg.StrOpt('service_log_path'),
            cfg.StrOpt('service_log_filename',
                       default='monasca-transform.log'),
            cfg.StrOpt('spark_event_logging_dest'),
            cfg.StrOpt('spark_event_logging_enabled'),
            cfg.StrOpt('spark_jars_list'),
            cfg.StrOpt('spark_master_list'),
            cfg.StrOpt('spark_python_files'),
            cfg.IntOpt('stream_interval'),
            cfg.StrOpt('work_dir'),
            cfg.StrOpt('spark_home'),
            cfg.BoolOpt('enable_record_store_df_cache'),
            cfg.StrOpt('record_store_df_cache_storage_level')
        ]
        service_group = cfg.OptGroup(name='service', title='service')
        cfg.CONF.register_group(service_group)
        cfg.CONF.register_opts(service_opts, group=service_group)

def get_source_opts(type_=None, location=None, reference=None):
    return [cfg.StrOpt('type', choices=['local', 'git', 'url'],
                       default=type_,
                       help='Source location type'),
            cfg.StrOpt('location', default=location,
                       help='The location for source install'),
            cfg.StrOpt('reference', default=reference,
                       help=('Git reference to pull, commit sha, tag '
                             'or branch name'))]

def __metric_opts(self):
        """List of options to be used in metric defined sections"""
        return [
            cfg.StrOpt("metric_name",
                       help="Metric Name used to log monitoring information"
                            " in Monasca",
                       required=True),
            cfg.DictOpt("dimensions",
                        default={},
                        help="Dict that contains dimensions information. "
                             "component:nova-compute,service:compute",
                        ),
            cfg.StrOpt("aggregate",
                       choices=["any", "all"],
                       help="How to consider the compute node is down. If you "
                            "metric reports many states, like checking "
                            "different services on the compute host, should we"
                            " consider if one component down all are down or"
                            " only if all components are down. Default is all."
                            " This means if all components fail, freezer-dr"
                            " will consider the host failed",
                       default='all'
                       ),
            cfg.StrOpt("undetermined",
                       choices=['OK', 'ALARM'],
                       default='ALARM',
                       help="How to handle UNDETERMINED states. It can be "
                            "ignored, will be considered OK state or can be "
                            "considered ALARM. Default is ALARM")

        ]

def setUp(self):
        super(MySQLConfFixture, self).setUp()
        self.register_opt(cfg.IntOpt('max_pool_size', default=20),
                          group='database')
        self.register_opt(cfg.IntOpt('idle_timeout', default=3600),
                          group='database')
        self.register_opt(cfg.StrOpt('connection', default=''),
                          group='database')
        self.url = db_test_utils.get_connect_string("mysql")
        self.set_default('connection', self.url, group='database')
        lockutils.set_defaults(lock_path='/tmp')
        self._drop_db()

def setUp(self):
        super(PostgresConfFixture, self).setUp()
        self.register_opt(cfg.StrOpt('connection', default=''),
                          group='database')
        self.register_opt(cfg.IntOpt('max_pool_size', default=20),
                          group='database')
        self.register_opt(cfg.IntOpt('idle_timeout', default=3600),
                          group='database')
        self.register_opts(cli.MIGRATION_OPTS)
        self.url = db_test_utils.get_connect_string("postgres")
        self.set_default('connection', self.url, group='database')
        self.set_default('disable_microsecond_data_migration', False)
        lockutils.set_defaults(lock_path='/tmp')
        self._drop_db()

def __init__(self):
        super(QueueManager, self).__init__(name="QueueManager")

        self.config_opts = [
            cfg.StrOpt("db_connection", help="the DB url", required=True),
            cfg.IntOpt('db_pool_size', default=10, required=False),
            cfg.IntOpt('db_pool_recycle', default=30, required=False),
            cfg.IntOpt('db_max_overflow', default=5, required=False)
        ]
        self.queues = {}

def __init__(self):
        super(ProjectManager, self).__init__(name="ProjectManager")

        self.config_opts = [
            cfg.IntOpt("default_TTL", default=1440, required=False),
            cfg.FloatOpt("default_share", default=10.0, required=False),
            cfg.StrOpt("db_connection", help="the DB url", required=True),
            cfg.IntOpt('db_pool_size', default=10, required=False),
            cfg.IntOpt('db_pool_recycle', default=30, required=False),
            cfg.IntOpt('db_max_overflow', default=5, required=False)
        ]

        self.projects = {}

def get_options(self):
        options = super(TokenEndpoint, self).get_options()

        options.extend([
            # Maintain name 'url' for compatibility
            cfg.StrOpt('url',
                       help='Specific service endpoint to use'),
            cfg.StrOpt('token',
                       secret=True,
                       help='Authentication token to use'),
        ])

        return options

def logger_conf(logger_name):
    return [
        cfg.StrOpt('output_format',
                   default="%(asctime)s - %(levelname)s - %(message)s"),
        cfg.BoolOpt('store', default=True),
        cfg.StrOpt('logging_level', default='debug'),
        cfg.StrOpt('logging_dir', default='/var/log/valet/'),
        cfg.StrOpt('logger_name', default=logger_name + ".log"),
        cfg.IntOpt('max_main_log_size', default=5000000),
        cfg.IntOpt('max_log_size', default=1000000),
        cfg.IntOpt('max_num_of_logs', default=3),
    ]

def _register_opts(self):
        """Register additional options specific to this filter plugin"""
        opts = []
        option = cfg.StrOpt('failure_mode',
                            choices=['reject', 'yield'], default='reject',
                            help=_('Mode to operate in if Valet '
                                   'planning fails for any reason.'))

        # In the filter plugin space, there's no access to Nova's
        # keystone credentials, so we have to specify our own.
        # This also means we can't act as the user making the request
        # at scheduling-time.
        opts.append(option)
        option = cfg.StrOpt('admin_tenant_name', default=None,
                            help=_('Valet Project Name'))
        opts.append(option)
        option = cfg.StrOpt('admin_username', default=None,
                            help=_('Valet Username'))
        opts.append(option)
        option = cfg.StrOpt('admin_password', default=None,
                            help=_('Valet Password'))
        opts.append(option)
        option = cfg.StrOpt('admin_auth_url', default=None,
                            help=_('Keystone Authorization API Endpoint'))
        opts.append(option)

        opt_group = cfg.OptGroup('valet')
        cfg.CONF.register_group(opt_group)
        cfg.CONF.register_opts(opts, group=opt_group)

def get_config_parser():
    conf = cfg.ConfigOpts()
    conf.register_cli_opt(
        cfg.StrOpt(
            'repo_root',
            default='.',
            help='directory containing the git repositories',
        )
    )
    return conf

def injector():
    conf = cfg.ConfigOpts()
    conf.register_cli_opts([
        cfg.IntOpt("metrics", default=1, min=1),
        cfg.StrOpt("archive-policy-name", default="low"),
        cfg.StrOpt("creator", default="admin"),
        cfg.IntOpt("batch-of-measures", default=1000),
        cfg.IntOpt("measures-per-batch", default=10),
    ])
    conf = service.prepare_service(conf=conf)
    index = indexer.get_driver(conf)
    instore = incoming.get_driver(conf)

    def todo():
        metric = index.create_metric(
            uuid.uuid4(),
            creator=conf.creator,
            archive_policy_name=conf.archive_policy_name)

        for _ in six.moves.range(conf.batch_of_measures):
            measures = [
                incoming.Measure(
                    utils.dt_in_unix_ns(utils.utcnow()), random.random())
                for __ in six.moves.range(conf.measures_per_batch)]
            instore.add_measures(metric, measures)

    with futures.ThreadPoolExecutor(max_workers=conf.metrics) as executor:
        for m in six.moves.range(conf.metrics):
            executor.submit(todo)

def init_config(conf):
    opts = [
        cfg.StrOpt('endpoint_type', default='publicURL')
    ]
    conf.register_opts(opts, group="networking_sfc")
    return conf.networking_sfc

def logger_conf(logger_name):
    return [
        cfg.StrOpt('output_format', default="%(asctime)s - %(levelname)s - %(message)s"),  # dict
        cfg.BoolOpt('store', default=True),
        cfg.StrOpt('logging_level', default='debug'),
        cfg.StrOpt('logging_dir', default='/var/log/valet/'),
        cfg.StrOpt('logger_name', default=logger_name + ".log"),
        cfg.IntOpt('max_main_log_size', default=5000000),
        cfg.IntOpt('max_log_size', default=1000000),
        cfg.IntOpt('max_num_of_logs', default=3),
    ]

def _register_opts(self):
        '''Register options'''
        opts = []
        option = cfg.StrOpt(self.opt_name_str, default=None, help=_('Valet API endpoint'))
        opts.append(option)
        option = cfg.IntOpt(self.opt_conn_timeout, default=3, help=_('Valet Plugin Connect Timeout'))
        opts.append(option)
        option = cfg.IntOpt(self.opt_read_timeout, default=5, help=_('Valet Plugin Read Timeout'))
        opts.append(option)

        opt_group = cfg.OptGroup(self.opt_group_str)
        cfg.CONF.register_group(opt_group)
        cfg.CONF.register_opts(opts, group=opt_group)

    # TODO(JD): Keep stack param for now. We may need it again.

def get_source_opts(type_=None, location=None, reference=None):
    return [cfg.StrOpt('type', choices=['local', 'git', 'url'],
                       default=type_,
                       help='Source location type'),
            cfg.StrOpt('location', default=location,
                       help='The location for source install'),
            cfg.StrOpt('reference', default=reference,
                       help=('Git reference to pull, commit sha, tag '
                             'or branch name'))]

def get_user_opts(uid, gid):
    return [
        cfg.StrOpt('uid', default=uid, help='The user id'),
        cfg.StrOpt('gid', default=gid, help='The group id'),
    ]

def load_config(self):
        """Load config options from nova config file or command line (for example: /etc/nova/nova.conf)

        Sample settings in nova config:
            [kozinaki_EC2]
            user=AKIAJR7NAEIZPWSTFBEQ
            key=zv9zSem8OE+k/axFkPCgZ3z3tLrhvFBaIIa0Ik0j
        """

        provider_opts = [
            cfg.StrOpt('subscription_id', help='Subscribe is from azure portal settings'),
            cfg.StrOpt('key_file', help='API key to work with the cloud provider', secret=True),
            cfg.StrOpt('username', help='Default vm username'),
            cfg.StrOpt('password', help='Azure: default instance password. '
                                        'Password must be 6-72 characters long'),
            cfg.StrOpt('app_client_id', help='Azure app client id'),
            cfg.StrOpt('app_secret', help='Azure app secret'),
            cfg.StrOpt('app_tenant', help='Azure app tenant'),
            cfg.StrOpt('resource_group_name', help='Azure resource group name'),
            cfg.StrOpt('location', help='VM location'),
            cfg.StrOpt('storage_account_name', help='Azure storage account name'),
            cfg.StrOpt('os_disk_name', help='VM default disk name'),
            cfg.StrOpt('vnet_name', help='Azure default virtual network'),
            cfg.StrOpt('subnet_name', help='Azure default subnet name'),
            cfg.StrOpt('ip_config_name', help='Azure default ip config name'),
            cfg.StrOpt('nic_name', help='Azure default nic name'),
            cfg.StrOpt('cloud_service_name', help='Azure default cloud service name'),
            cfg.StrOpt('deployment_name', help='Azure default deployment name'),
        ]

        cfg.CONF.register_opts(provider_opts, self.config_name)
        return cfg.CONF[self.config_name]