stackforge-attic/murano-docs
Code Issues Proposed changes

Licenses added

Browse Source
This commit is contained in:
Stan Lagun
2013-04-04 18:10:25 +04:00
parent f30a833cca
commit c8887c6b66
16 changed files with 966 additions and 790 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
conductor/bin/conductor
View File

@@ -1,20 +1,17 @@
#!/usr/bin/env python # Copyright (c) 2013 Mirantis Inc.
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack LLC.
# All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # Licensed under the Apache License, Version 2.0 (the "License");
# not use this file except in compliance with the License. You may obtain # you may not use this file except in compliance with the License.
# a copy of the License at # You may obtain a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS,
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# License for the specific language governing permissions and limitations # implied.
# under the License. # See the License for the specific language governing permissions and
# limitations under the License.
import sys import sys

195
conductor/conductor/app.py
View File

@@ -1,90 +1,105 @@
import datetime # Copyright (c) 2013 Mirantis Inc.
import glob #
import sys # Licensed under the Apache License, Version 2.0 (the "License");
import traceback # you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
import anyjson #
from conductor.openstack.common import service # http://www.apache.org/licenses/LICENSE-2.0
from workflow import Workflow #
from commands.dispatcher import CommandDispatcher # Unless required by applicable law or agreed to in writing, software
from openstack.common import log as logging # distributed under the License is distributed on an "AS IS" BASIS,
from config import Config # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
import reporting # implied.
import rabbitmq # See the License for the specific language governing permissions and
# limitations under the License.
import windows_agent
import cloud_formation import datetime
import glob
config = Config(sys.argv[1] if len(sys.argv) > 1 else None) import sys
import traceback
log = logging.getLogger(__name__)
import anyjson
from conductor.openstack.common import service
def task_received(task, message_id): from workflow import Workflow
with rabbitmq.RmqClient() as rmqclient: from commands.dispatcher import CommandDispatcher
try: from openstack.common import log as logging
log.info('Starting processing task {0}: {1}'.format( from config import Config
message_id, anyjson.dumps(task))) import reporting
reporter = reporting.Reporter(rmqclient, message_id, task['id']) import rabbitmq
command_dispatcher = CommandDispatcher( import windows_agent
task['name'], rmqclient, task['token'], task['tenant_id']) import cloud_formation
workflows = []
for path in glob.glob("data/workflows/*.xml"): config = Config(sys.argv[1] if len(sys.argv) > 1 else None)
log.debug('Loading XML {0}'.format(path))
workflow = Workflow(path, task, command_dispatcher, config, log = logging.getLogger(__name__)
reporter)
workflows.append(workflow)
def task_received(task, message_id):
while True: with rabbitmq.RmqClient() as rmqclient:
try: try:
while True: log.info('Starting processing task {0}: {1}'.format(
result = False message_id, anyjson.dumps(task)))
for workflow in workflows: reporter = reporting.Reporter(rmqclient, message_id, task['id'])
if workflow.execute():
result = True command_dispatcher = CommandDispatcher(
if not result: task['name'], rmqclient, task['token'], task['tenant_id'])
break workflows = []
if not command_dispatcher.execute_pending(): for path in glob.glob("data/workflows/*.xml"):
break log.debug('Loading XML {0}'.format(path))
except Exception as ex: workflow = Workflow(path, task, command_dispatcher, config,
log.exception(ex) reporter)
break workflows.append(workflow)
command_dispatcher.close() while True:
finally: try:
del task['token'] while True:
result_msg = rabbitmq.Message() result = False
result_msg.body = task for workflow in workflows:
result_msg.id = message_id if workflow.execute():
result = True
rmqclient.send(message=result_msg, key='task-results') if not result:
log.info('Finished processing task {0}. Result = {1}'.format( break
message_id, anyjson.dumps(task))) if not command_dispatcher.execute_pending():
break
except Exception as ex:
class ConductorWorkflowService(service.Service): log.exception(ex)
def __init__(self): break
super(ConductorWorkflowService, self).__init__()
command_dispatcher.close()
def start(self): finally:
super(ConductorWorkflowService, self).start() del task['token']
self.tg.add_thread(self._start_rabbitmq) result_msg = rabbitmq.Message()
result_msg.body = task
def stop(self): result_msg.id = message_id
super(ConductorWorkflowService, self).stop()
rmqclient.send(message=result_msg, key='task-results')
def _start_rabbitmq(self): log.info('Finished processing task {0}. Result = {1}'.format(
while True: message_id, anyjson.dumps(task)))
try:
with rabbitmq.RmqClient() as rmq:
rmq.declare('tasks', 'tasks') class ConductorWorkflowService(service.Service):
rmq.declare('task-results') def __init__(self):
with rmq.open('tasks') as subscription: super(ConductorWorkflowService, self).__init__()
while True:
msg = subscription.get_message() def start(self):
self.tg.add_thread( super(ConductorWorkflowService, self).start()
task_received, msg.body, msg.id) self.tg.add_thread(self._start_rabbitmq)
except Exception as ex:
log.exception(ex) def stop(self):
super(ConductorWorkflowService, self).stop()
def _start_rabbitmq(self):
while True:
try:
with rabbitmq.RmqClient() as rmq:
rmq.declare('tasks', 'tasks')
rmq.declare('task-results')
with rmq.open('tasks') as subscription:
while True:
msg = subscription.get_message()
self.tg.add_thread(
task_received, msg.body, msg.id)
except Exception as ex:
log.exception(ex)

211
conductor/conductor/cloud_formation.py
View File

@@ -1,98 +1,113 @@
import base64 # Copyright (c) 2013 Mirantis Inc.
#
import xml_code_engine # Licensed under the Apache License, Version 2.0 (the "License");
import config # you may not use this file except in compliance with the License.
from random import choice # You may obtain a copy of the License at
import time #
import string # http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
def update_cf_stack(engine, context, body, template, # distributed under the License is distributed on an "AS IS" BASIS,
mappings, arguments, **kwargs): # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
command_dispatcher = context['/commandDispatcher'] # implied.
# See the License for the specific language governing permissions and
callback = lambda result: engine.evaluate_content( # limitations under the License.
body.find('success'), context)
import base64
command_dispatcher.execute(
name='cf', command='CreateOrUpdate', template=template, import xml_code_engine
mappings=mappings, arguments=arguments, callback=callback) import config
from random import choice
import time
def delete_cf_stack(engine, context, body, **kwargs): import string
command_dispatcher = context['/commandDispatcher']
callback = lambda result: engine.evaluate_content( def update_cf_stack(engine, context, body, template,
body.find('success'), context) mappings, arguments, **kwargs):
command_dispatcher = context['/commandDispatcher']
command_dispatcher.execute(
name='cf', command='Delete', callback=callback) callback = lambda result: engine.evaluate_content(
body.find('success'), context)
def prepare_user_data(context, hostname, service, unit, template='Default', **kwargs): command_dispatcher.execute(
settings = config.CONF.rabbitmq name='cf', command='CreateOrUpdate', template=template,
mappings=mappings, arguments=arguments, callback=callback)
with open('data/init.ps1') as init_script_file:
with open('data/templates/agent-config/{0}.template'.format(
template)) as template_file: def delete_cf_stack(engine, context, body, **kwargs):
init_script = init_script_file.read() command_dispatcher = context['/commandDispatcher']
template_data = template_file.read()
template_data = template_data.replace( callback = lambda result: engine.evaluate_content(
'%RABBITMQ_HOST%', settings.host) body.find('success'), context)
template_data = template_data.replace(
'%RABBITMQ_INPUT_QUEUE%', command_dispatcher.execute(
'-'.join([str(context['/dataSource']['name']), name='cf', command='Delete', callback=callback)
str(service), str(unit)]).lower()
)
template_data = template_data.replace( def prepare_user_data(context, hostname, service, unit,
'%RESULT_QUEUE%', template='Default', **kwargs):
'-execution-results-{0}'.format( settings = config.CONF.rabbitmq
str(context['/dataSource']['name'])).lower())
with open('data/init.ps1') as init_script_file:
init_script = init_script.replace( with open('data/templates/agent-config/{0}.template'.format(
'%WINDOWS_AGENT_CONFIG_BASE64%', template)) as template_file:
base64.b64encode(template_data)) init_script = init_script_file.read()
template_data = template_file.read()
init_script = init_script.replace('%INTERNAL_HOSTNAME%', hostname) template_data = template_data.replace(
'%RABBITMQ_HOST%', settings.host)
return init_script template_data = template_data.replace(
'%RABBITMQ_INPUT_QUEUE%',
counter = 0 '-'.join([str(context['/dataSource']['name']),
str(service), str(unit)]).lower()
)
def int2base(x, base): template_data = template_data.replace(
digs = string.digits + string.lowercase '%RESULT_QUEUE%',
if x < 0: sign = -1 '-execution-results-{0}'.format(
elif x==0: return '0' str(context['/dataSource']['name'])).lower())
else: sign = 1
x *= sign init_script = init_script.replace(
digits = [] '%WINDOWS_AGENT_CONFIG_BASE64%',
while x: base64.b64encode(template_data))
digits.append(digs[x % base])
x /= base init_script = init_script.replace('%INTERNAL_HOSTNAME%', hostname)
if sign < 0:
digits.append('-') return init_script
digits.reverse()
return ''.join(digits) counter = 0
def int2base(x, base):
def generate_hostname(**kwargs): digs = string.digits + string.lowercase
global counter if x < 0: sign = -1
prefix = ''.join(choice(string.lowercase) for _ in range(5)) elif x==0: return '0'
timestamp = int2base(int(time.time() * 1000), 36)[:8] else: sign = 1
suffix = int2base(counter, 36) x *= sign
counter = (counter + 1) % 1296 digits = []
return prefix + timestamp + suffix while x:
digits.append(digs[x % base])
x /= base
xml_code_engine.XmlCodeEngine.register_function( if sign < 0:
update_cf_stack, "update-cf-stack") digits.append('-')
digits.reverse()
xml_code_engine.XmlCodeEngine.register_function( return ''.join(digits)
delete_cf_stack, "delete-cf-stack")
xml_code_engine.XmlCodeEngine.register_function( def generate_hostname(**kwargs):
prepare_user_data, "prepare-user-data") global counter
prefix = ''.join(choice(string.lowercase) for _ in range(5))
xml_code_engine.XmlCodeEngine.register_function( timestamp = int2base(int(time.time() * 1000), 36)[:8]
generate_hostname, "generate-hostname") suffix = int2base(counter, 36)
counter = (counter + 1) % 1296
return prefix + timestamp + suffix
xml_code_engine.XmlCodeEngine.register_function(
update_cf_stack, "update-cf-stack")
xml_code_engine.XmlCodeEngine.register_function(
delete_cf_stack, "delete-cf-stack")
xml_code_engine.XmlCodeEngine.register_function(
prepare_user_data, "prepare-user-data")
xml_code_engine.XmlCodeEngine.register_function(
generate_hostname, "generate-hostname")

15
conductor/conductor/commands/__init__.py
View File

@@ -1 +1,16 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import command import command

357
conductor/conductor/commands/cloud_formation.py
View File

@@ -1,169 +1,188 @@
import anyjson # Copyright (c) 2013 Mirantis Inc.
import eventlet #
# Licensed under the Apache License, Version 2.0 (the "License");
import jsonpath # you may not use this file except in compliance with the License.
from conductor.openstack.common import log as logging # You may obtain a copy of the License at
import conductor.helpers #
from command import CommandBase # http://www.apache.org/licenses/LICENSE-2.0
import conductor.config #
from heatclient.client import Client # Unless required by applicable law or agreed to in writing, software
import heatclient.exc # distributed under the License is distributed on an "AS IS" BASIS,
from keystoneclient.v2_0 import client as ksclient # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
import types # implied.
# See the License for the specific language governing permissions and
log = logging.getLogger(__name__) # limitations under the License.
import anyjson
class HeatExecutor(CommandBase): import eventlet
def __init__(self, stack, token, tenant_id): import types
self._update_pending_list = [] import jsonpath
self._delete_pending_list = []
self._stack = stack from conductor.openstack.common import log as logging
settings = conductor.config.CONF.heat import conductor.helpers
from command import CommandBase
client = ksclient.Client(endpoint=settings.auth_url) import conductor.config
auth_data = client.tokens.authenticate( from heatclient.client import Client
tenant_id=tenant_id, import heatclient.exc
token=token) from keystoneclient.v2_0 import client as ksclient
scoped_token = auth_data.id log = logging.getLogger(__name__)
heat_url = jsonpath.jsonpath(auth_data.serviceCatalog,
"$[?(@.name == 'heat')].endpoints[0].publicURL")[0] class HeatExecutor(CommandBase):
def __init__(self, stack, token, tenant_id):
self._heat_client = Client('1', heat_url, self._update_pending_list = []
token_only=True, token=scoped_token) self._delete_pending_list = []
self._stack = stack
def execute(self, command, callback, **kwargs): settings = conductor.config.CONF.heat
log.debug('Got command {0} on stack {1}'.format(command, self._stack))
client = ksclient.Client(endpoint=settings.auth_url)
if command == 'CreateOrUpdate': auth_data = client.tokens.authenticate(
return self._execute_create_update( tenant_id=tenant_id,
kwargs['template'], token=token)
kwargs['mappings'],
kwargs['arguments'], scoped_token = auth_data.id
callback)
elif command == 'Delete': heat_url = jsonpath.jsonpath(
return self._execute_delete(callback) auth_data.serviceCatalog,
"$[?(@.name == 'heat')].endpoints[0].publicURL")[0]
def _execute_create_update(self, template, mappings, arguments, callback):
with open('data/templates/cf/%s.template' % template) as template_file: self._heat_client = Client(
template_data = template_file.read() '1',
heat_url,
template_data = conductor.helpers.transform_json( token_only=True,
anyjson.loads(template_data), mappings) token=scoped_token)
self._update_pending_list.append({ def execute(self, command, callback, **kwargs):
'template': template_data, log.debug('Got command {0} on stack {1}'.format(command, self._stack))
'arguments': arguments,
'callback': callback if command == 'CreateOrUpdate':
}) return self._execute_create_update(
kwargs['template'],
def _execute_delete(self, callback): kwargs['mappings'],
self._delete_pending_list.append({ kwargs['arguments'],
'callback': callback callback)
}) elif command == 'Delete':
return self._execute_delete(callback)
def has_pending_commands(self):
return len(self._update_pending_list) + \ def _execute_create_update(self, template, mappings, arguments, callback):
len(self._delete_pending_list) > 0 with open('data/templates/cf/%s.template' % template) as template_file:
template_data = template_file.read()
def execute_pending(self):
r1 = self._execute_pending_updates() template_data = conductor.helpers.transform_json(
r2 = self._execute_pending_deletes() anyjson.loads(template_data), mappings)
return r1 or r2
self._update_pending_list.append({
def _execute_pending_updates(self): 'template': template_data,
if not len(self._update_pending_list): 'arguments': arguments,
return False 'callback': callback
})
template, arguments = self._get_current_template()
stack_exists = (template != {}) def _execute_delete(self, callback):
self._delete_pending_list.append({
for t in self._update_pending_list: 'callback': callback
template = conductor.helpers.merge_dicts( })
template, t['template'], max_levels=2)
arguments = conductor.helpers.merge_dicts( def has_pending_commands(self):
arguments, t['arguments'], max_levels=1) return len(self._update_pending_list) + \
len(self._delete_pending_list) > 0
log.info(
'Executing heat template {0} with arguments {1} on stack {2}' def execute_pending(self):
.format(anyjson.dumps(template), arguments, self._stack)) r1 = self._execute_pending_updates()
r2 = self._execute_pending_deletes()
if stack_exists: return r1 or r2
self._heat_client.stacks.update(
stack_id=self._stack, def _execute_pending_updates(self):
parameters=arguments, if not len(self._update_pending_list):
template=template) return False
log.debug(
'Waiting for the stack {0} to be update'.format(self._stack)) template, arguments = self._get_current_template()
self._wait_state('UPDATE_COMPLETE') stack_exists = (template != {})
log.info('Stack {0} updated'.format(self._stack))
else: for t in self._update_pending_list:
self._heat_client.stacks.create( template = conductor.helpers.merge_dicts(
stack_name=self._stack, template, t['template'], max_levels=2)
parameters=arguments, arguments = conductor.helpers.merge_dicts(
template=template) arguments, t['arguments'], max_levels=1)
log.debug('Waiting for the stack {0} to be create'.format(
self._stack)) log.info(
self._wait_state('CREATE_COMPLETE') 'Executing heat template {0} with arguments {1} on stack {2}'
log.info('Stack {0} created'.format(self._stack)) .format(anyjson.dumps(template), arguments, self._stack))
pending_list = self._update_pending_list if stack_exists:
self._update_pending_list = [] self._heat_client.stacks.update(
stack_id=self._stack,
for item in pending_list: parameters=arguments,
item['callback'](True) template=template)
log.debug(
return True 'Waiting for the stack {0} to be update'.format(self._stack))
self._wait_state('UPDATE_COMPLETE')
def _execute_pending_deletes(self): log.info('Stack {0} updated'.format(self._stack))
if not len(self._delete_pending_list): else:
return False self._heat_client.stacks.create(
stack_name=self._stack,
log.debug('Deleting stack {0}'.format(self._stack)) parameters=arguments,
try: template=template)
self._heat_client.stacks.delete( log.debug('Waiting for the stack {0} to be create'.format(
stack_id=self._stack) self._stack))
log.debug( self._wait_state('CREATE_COMPLETE')
'Waiting for the stack {0} to be deleted'.format(self._stack)) log.info('Stack {0} created'.format(self._stack))
self._wait_state(['DELETE_COMPLETE', ''])
log.info('Stack {0} deleted'.format(self._stack)) pending_list = self._update_pending_list
except Exception as ex: self._update_pending_list = []
log.exception(ex)
for item in pending_list:
pending_list = self._delete_pending_list item['callback'](True)
self._delete_pending_list = []
return True
for item in pending_list:
item['callback'](True) def _execute_pending_deletes(self):
return True if not len(self._delete_pending_list):
return False
def _get_current_template(self):
try: log.debug('Deleting stack {0}'.format(self._stack))
stack_info = self._heat_client.stacks.get(stack_id=self._stack) try:
template = self._heat_client.stacks.template( self._heat_client.stacks.delete(
stack_id='{0}/{1}'.format(stack_info.stack_name, stack_info.id)) stack_id=self._stack)
return template, stack_info.parameters log.debug(
except heatclient.exc.HTTPNotFound: 'Waiting for the stack {0} to be deleted'.format(self._stack))
return {}, {} self._wait_state(['DELETE_COMPLETE', ''])
log.info('Stack {0} deleted'.format(self._stack))
def _wait_state(self, state): except Exception as ex:
if isinstance(state, types.ListType): log.exception(ex)
states = state
else: pending_list = self._delete_pending_list
states = [state] self._delete_pending_list = []
while True: for item in pending_list:
try: item['callback'](True)
status = self._heat_client.stacks.get( return True
stack_id=self._stack).stack_status
except heatclient.exc.HTTPNotFound: def _get_current_template(self):
status = '' try:
stack_info = self._heat_client.stacks.get(stack_id=self._stack)
if 'IN_PROGRESS' in status: template = self._heat_client.stacks.template(
eventlet.sleep(1) stack_id='{0}/{1}'.format(stack_info.stack_name, stack_info.id))
continue return template, stack_info.parameters
if status not in states: except heatclient.exc.HTTPNotFound:
raise EnvironmentError() return {}, {}
return
def _wait_state(self, state):
if isinstance(state, types.ListType):
states = state
else:
states = [state]
while True:
try:
status = self._heat_client.stacks.get(
stack_id=self._stack).stack_status
except heatclient.exc.HTTPNotFound:
status = ''
if 'IN_PROGRESS' in status:
eventlet.sleep(1)
continue
if status not in states:
raise EnvironmentError()
return

16
conductor/conductor/commands/command.py
View File

@@ -1,3 +1,19 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
class CommandBase(object): class CommandBase(object):
def execute(self, **kwargs): def execute(self, **kwargs):
pass pass

81
conductor/conductor/commands/dispatcher.py
View File

@@ -1,33 +1,48 @@
import command # Copyright (c) 2013 Mirantis Inc.
import cloud_formation #
import windows_agent # Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
class CommandDispatcher(command.CommandBase): #
def __init__(self, environment_id, rmqclient, token, tenant_id): # http://www.apache.org/licenses/LICENSE-2.0
self._command_map = { #
'cf': cloud_formation.HeatExecutor(environment_id, token, tenant_id), # Unless required by applicable law or agreed to in writing, software
'agent': windows_agent.WindowsAgentExecutor( # distributed under the License is distributed on an "AS IS" BASIS,
environment_id, rmqclient) # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
} # implied.
# See the License for the specific language governing permissions and
def execute(self, name, **kwargs): # limitations under the License.
self._command_map[name].execute(**kwargs)
import command
def execute_pending(self): import cloud_formation
result = False import windows_agent
for command in self._command_map.values():
result |= command.execute_pending()
class CommandDispatcher(command.CommandBase):
return result def __init__(self, environment, rmqclient, token, tenant_id):
self._command_map = {
def has_pending_commands(self): 'cf': cloud_formation.HeatExecutor(environment, token, tenant_id),
result = False 'agent': windows_agent.WindowsAgentExecutor(
for command in self._command_map.values(): environment, rmqclient)
result |= command.has_pending_commands() }
return result def execute(self, name, **kwargs):
self._command_map[name].execute(**kwargs)
def close(self):
for t in self._command_map.values(): def execute_pending(self):
t.close() result = False
for command in self._command_map.values():
result |= command.execute_pending()
return result
def has_pending_commands(self):
result = False
for command in self._command_map.values():
result |= command.has_pending_commands()
return result
def close(self):
for t in self._command_map.values():
t.close()

423
conductor/conductor/config.py
View File

@@ -1,213 +1,210 @@
#!/usr/bin/env python # Copyright 2011 OpenStack LLC.
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # All Rights Reserved.
#
# Copyright 2011 OpenStack LLC. # Licensed under the Apache License, Version 2.0 (the "License"); you may
# All Rights Reserved. # not use this file except in compliance with the License. You may obtain
# # a copy of the License at
# Licensed under the Apache License, Version 2.0 (the "License"); you may #
# not use this file except in compliance with the License. You may obtain # http://www.apache.org/licenses/LICENSE-2.0
# a copy of the License at #
# # Unless required by applicable law or agreed to in writing, software
# http://www.apache.org/licenses/LICENSE-2.0 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# Unless required by applicable law or agreed to in writing, software # License for the specific language governing permissions and limitations
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # under the License.
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations """
# under the License. Routines for configuring Glance
"""
"""
Routines for configuring Glance import logging
""" import logging.config
import logging.handlers
import logging import os
import logging.config import sys
import logging.handlers
import os from oslo.config import cfg
import sys from paste import deploy
from oslo.config import cfg from conductor.version import version_info as version
from paste import deploy from ConfigParser import SafeConfigParser
from conductor.version import version_info as version paste_deploy_opts = [
from ConfigParser import SafeConfigParser cfg.StrOpt('flavor'),
cfg.StrOpt('config_file'),
paste_deploy_opts = [ ]
cfg.StrOpt('flavor'),
cfg.StrOpt('config_file'), rabbit_opts = [
] cfg.StrOpt('host', default='localhost'),
cfg.IntOpt('port', default=5672),
rabbit_opts = [ cfg.StrOpt('login', default='guest'),
cfg.StrOpt('host', default='localhost'), cfg.StrOpt('password', default='guest'),
cfg.IntOpt('port', default=5672), cfg.StrOpt('virtual_host', default='/'),
cfg.StrOpt('login', default='guest'), ]
cfg.StrOpt('password', default='guest'),
cfg.StrOpt('virtual_host', default='/'), heat_opts = [
] cfg.StrOpt('auth_url'),
]
heat_opts = [
cfg.StrOpt('auth_url'), CONF = cfg.CONF
] CONF.register_opts(paste_deploy_opts, group='paste_deploy')
CONF.register_opts(rabbit_opts, group='rabbitmq')
CONF = cfg.CONF CONF.register_opts(heat_opts, group='heat')
CONF.register_opts(paste_deploy_opts, group='paste_deploy')
CONF.register_opts(rabbit_opts, group='rabbitmq')
CONF.register_opts(heat_opts, group='heat') CONF.import_opt('verbose', 'conductor.openstack.common.log')
CONF.import_opt('debug', 'conductor.openstack.common.log')
CONF.import_opt('log_dir', 'conductor.openstack.common.log')
CONF.import_opt('verbose', 'conductor.openstack.common.log') CONF.import_opt('log_file', 'conductor.openstack.common.log')
CONF.import_opt('debug', 'conductor.openstack.common.log') CONF.import_opt('log_config', 'conductor.openstack.common.log')
CONF.import_opt('log_dir', 'conductor.openstack.common.log') CONF.import_opt('log_format', 'conductor.openstack.common.log')
CONF.import_opt('log_file', 'conductor.openstack.common.log') CONF.import_opt('log_date_format', 'conductor.openstack.common.log')
CONF.import_opt('log_config', 'conductor.openstack.common.log') CONF.import_opt('use_syslog', 'conductor.openstack.common.log')
CONF.import_opt('log_format', 'conductor.openstack.common.log') CONF.import_opt('syslog_log_facility', 'conductor.openstack.common.log')
CONF.import_opt('log_date_format', 'conductor.openstack.common.log')
CONF.import_opt('use_syslog', 'conductor.openstack.common.log')
CONF.import_opt('syslog_log_facility', 'conductor.openstack.common.log') def parse_args(args=None, usage=None, default_config_files=None):
CONF(args=args,
project='conductor',
def parse_args(args=None, usage=None, default_config_files=None): version=version.cached_version_string(),
CONF(args=args, usage=usage,
project='conductor', default_config_files=default_config_files)
version=version.cached_version_string(),
usage=usage,
default_config_files=default_config_files) def setup_logging():
"""
Sets up the logging options for a log with supplied name
def setup_logging(): """
"""
Sets up the logging options for a log with supplied name if CONF.log_config:
""" # Use a logging configuration file for all settings...
if os.path.exists(CONF.log_config):
if CONF.log_config: logging.config.fileConfig(CONF.log_config)
# Use a logging configuration file for all settings... return
if os.path.exists(CONF.log_config): else:
logging.config.fileConfig(CONF.log_config) raise RuntimeError("Unable to locate specified logging "
return "config file: %s" % CONF.log_config)
else:
raise RuntimeError("Unable to locate specified logging " root_logger = logging.root
"config file: %s" % CONF.log_config) if CONF.debug:
root_logger.setLevel(logging.DEBUG)
root_logger = logging.root elif CONF.verbose:
if CONF.debug: root_logger.setLevel(logging.INFO)
root_logger.setLevel(logging.DEBUG) else:
elif CONF.verbose: root_logger.setLevel(logging.WARNING)
root_logger.setLevel(logging.INFO)
else: formatter = logging.Formatter(CONF.log_format, CONF.log_date_format)
root_logger.setLevel(logging.WARNING)
if CONF.use_syslog:
formatter = logging.Formatter(CONF.log_format, CONF.log_date_format) try:
facility = getattr(logging.handlers.SysLogHandler,
if CONF.use_syslog: CONF.syslog_log_facility)
try: except AttributeError:
facility = getattr(logging.handlers.SysLogHandler, raise ValueError(_("Invalid syslog facility"))
CONF.syslog_log_facility)
except AttributeError: handler = logging.handlers.SysLogHandler(address='/dev/log',
raise ValueError(_("Invalid syslog facility")) facility=facility)
elif CONF.log_file:
handler = logging.handlers.SysLogHandler(address='/dev/log', logfile = CONF.log_file
facility=facility) if CONF.log_dir:
elif CONF.log_file: logfile = os.path.join(CONF.log_dir, logfile)
logfile = CONF.log_file handler = logging.handlers.WatchedFileHandler(logfile)
if CONF.log_dir: else:
logfile = os.path.join(CONF.log_dir, logfile) handler = logging.StreamHandler(sys.stdout)
handler = logging.handlers.WatchedFileHandler(logfile)
else: handler.setFormatter(formatter)
handler = logging.StreamHandler(sys.stdout) root_logger.addHandler(handler)
handler.setFormatter(formatter)
root_logger.addHandler(handler) def _get_deployment_flavor():
"""
Retrieve the paste_deploy.flavor config item, formatted appropriately
def _get_deployment_flavor(): for appending to the application name.
""" """
Retrieve the paste_deploy.flavor config item, formatted appropriately flavor = CONF.paste_deploy.flavor
for appending to the application name. return '' if not flavor else ('-' + flavor)
"""
flavor = CONF.paste_deploy.flavor
return '' if not flavor else ('-' + flavor) def _get_paste_config_path():
paste_suffix = '-paste.ini'
conf_suffix = '.conf'
def _get_paste_config_path(): if CONF.config_file:
paste_suffix = '-paste.ini' # Assume paste config is in a paste.ini file corresponding
conf_suffix = '.conf' # to the last config file
if CONF.config_file: path = CONF.config_file[-1].replace(conf_suffix, paste_suffix)
# Assume paste config is in a paste.ini file corresponding else:
# to the last config file path = CONF.prog + '-paste.ini'
path = CONF.config_file[-1].replace(conf_suffix, paste_suffix) return CONF.find_file(os.path.basename(path))
else:
path = CONF.prog + '-paste.ini'
return CONF.find_file(os.path.basename(path)) def _get_deployment_config_file():
"""
Retrieve the deployment_config_file config item, formatted as an
def _get_deployment_config_file(): absolute pathname.
""" """
Retrieve the deployment_config_file config item, formatted as an path = CONF.paste_deploy.config_file
absolute pathname. if not path:
""" path = _get_paste_config_path()
path = CONF.paste_deploy.config_file if not path:
if not path: msg = "Unable to locate paste config file for %s." % CONF.prog
path = _get_paste_config_path() raise RuntimeError(msg)
if not path: return os.path.abspath(path)
msg = "Unable to locate paste config file for %s." % CONF.prog
raise RuntimeError(msg)
return os.path.abspath(path) def load_paste_app(app_name=None):
"""
Builds and returns a WSGI app from a paste config file.
def load_paste_app(app_name=None):
""" We assume the last config file specified in the supplied ConfigOpts
Builds and returns a WSGI app from a paste config file. object is the paste config file.
We assume the last config file specified in the supplied ConfigOpts :param app_name: name of the application to load
object is the paste config file.
:raises RuntimeError when config file cannot be located or application
:param app_name: name of the application to load cannot be loaded from config file
"""
:raises RuntimeError when config file cannot be located or application if app_name is None:
cannot be loaded from config file app_name = CONF.prog
"""
if app_name is None: # append the deployment flavor to the application name,
app_name = CONF.prog # in order to identify the appropriate paste pipeline
app_name += _get_deployment_flavor()
# append the deployment flavor to the application name,
# in order to identify the appropriate paste pipeline conf_file = _get_deployment_config_file()
app_name += _get_deployment_flavor()
try:
conf_file = _get_deployment_config_file() logger = logging.getLogger(__name__)
logger.debug(_("Loading %(app_name)s from %(conf_file)s"),
try: {'conf_file': conf_file, 'app_name': app_name})
logger = logging.getLogger(__name__)
logger.debug(_("Loading %(app_name)s from %(conf_file)s"), app = deploy.loadapp("config:%s" % conf_file, name=app_name)
{'conf_file': conf_file, 'app_name': app_name})
# Log the options used when starting if we're in debug mode...
app = deploy.loadapp("config:%s" % conf_file, name=app_name) if CONF.debug:
CONF.log_opt_values(logger, logging.DEBUG)
# Log the options used when starting if we're in debug mode...
if CONF.debug: return app
CONF.log_opt_values(logger, logging.DEBUG) except (LookupError, ImportError), e:
msg = _("Unable to load %(app_name)s from "
return app "configuration file %(conf_file)s."
except (LookupError, ImportError), e: "\nGot: %(e)r") % locals()
msg = _("Unable to load %(app_name)s from " logger.error(msg)
"configuration file %(conf_file)s." raise RuntimeError(msg)
"\nGot: %(e)r") % locals()
logger.error(msg)
raise RuntimeError(msg) class Config(object):
CONFIG_PATH = './etc/app.config'
class Config(object): def __init__(self, filename=None):
CONFIG_PATH = './etc/app.config' self.config = SafeConfigParser()
self.config.read(filename or self.CONFIG_PATH)
def __init__(self, filename=None):
self.config = SafeConfigParser() def get_setting(self, section, name, default=None):
self.config.read(filename or self.CONFIG_PATH) if not self.config.has_option(section, name):
return default
def get_setting(self, section, name, default=None): return self.config.get(section, name)
if not self.config.has_option(section, name):
return default def __getitem__(self, item):
return self.config.get(section, name) parts = item.rsplit('.', 1)
return self.get_setting(
def __getitem__(self, item): parts[0] if len(parts) == 2 else 'DEFAULT', parts[-1])
parts = item.rsplit('.', 1)
return self.get_setting(
parts[0] if len(parts) == 2 else 'DEFAULT', parts[-1])

16
conductor/conductor/function_context.py
View File

@@ -1,3 +1,19 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
class Context(object): class Context(object):
def __init__(self, parent=None): def __init__(self, parent=None):
self._parent = parent self._parent = parent

15
conductor/conductor/helpers.py
View File

@@ -1,3 +1,18 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import types import types

268
conductor/conductor/rabbitmq.py
View File

@@ -1,127 +1,141 @@
from eventlet import patcher # Copyright (c) 2013 Mirantis Inc.
puka = patcher.import_patched('puka') #
#import puka # Licensed under the Apache License, Version 2.0 (the "License");
import anyjson # you may not use this file except in compliance with the License.
import config # You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
class RmqClient(object): #
def __init__(self): # Unless required by applicable law or agreed to in writing, software
settings = config.CONF.rabbitmq # distributed under the License is distributed on an "AS IS" BASIS,
self._client = puka.Client('amqp://{0}:{1}@{2}:{3}/{4}'.format( # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
settings.login, # implied.
settings.password, # See the License for the specific language governing permissions and
settings.host, # limitations under the License.
settings.port,
settings.virtual_host from eventlet import patcher
)) puka = patcher.import_patched('puka')
self._connected = False #import puka
import anyjson
def __enter__(self): import config
self.connect()
return self
class RmqClient(object):
def __exit__(self, exc_type, exc_val, exc_tb): def __init__(self):
self.close() settings = config.CONF.rabbitmq
return False self._client = puka.Client('amqp://{0}:{1}@{2}:{3}/{4}'.format(
settings.login,
def connect(self): settings.password,
if not self._connected: settings.host,
promise = self._client.connect() settings.port,
self._client.wait(promise, timeout=10000) settings.virtual_host
self._connected = True ))
self._connected = False
def close(self):
if self._connected: def __enter__(self):
self._client.close() self.connect()
self._connected = False return self
def declare(self, queue, exchange=None): def __exit__(self, exc_type, exc_val, exc_tb):
promise = self._client.queue_declare(str(queue), durable=True) self.close()
self._client.wait(promise) return False
if exchange: def connect(self):
promise = self._client.exchange_declare(str(exchange), durable=True) if not self._connected:
self._client.wait(promise) promise = self._client.connect()
promise = self._client.queue_bind( self._client.wait(promise, timeout=10000)
str(queue), str(exchange), routing_key=str(queue)) self._connected = True
self._client.wait(promise)
def close(self):
def send(self, message, key, exchange='', timeout=None): if self._connected:
if not self._connected: self._client.close()
raise RuntimeError('Not connected to RabbitMQ') self._connected = False
headers = { 'message_id': message.id } def declare(self, queue, exchange=None):
promise = self._client.queue_declare(str(queue), durable=True)
promise = self._client.basic_publish( self._client.wait(promise)
exchange=str(exchange),
routing_key=str(key), if exchange:
body=anyjson.dumps(message.body), promise = self._client.exchange_declare(str(exchange), durable=True)
headers=headers) self._client.wait(promise)
self._client.wait(promise, timeout=timeout) promise = self._client.queue_bind(
str(queue), str(exchange), routing_key=str(queue))
def open(self, queue): self._client.wait(promise)
if not self._connected:
raise RuntimeError('Not connected to RabbitMQ') def send(self, message, key, exchange='', timeout=None):
if not self._connected:
return Subscription(self._client, queue) raise RuntimeError('Not connected to RabbitMQ')
headers = { 'message_id': message.id }
class Subscription(object):
def __init__(self, client, queue): promise = self._client.basic_publish(
self._client = client exchange=str(exchange),
self._queue = queue routing_key=str(key),
self._promise = None body=anyjson.dumps(message.body),
self._lastMessage = None headers=headers)
self._client.wait(promise, timeout=timeout)
def __enter__(self):
self._promise = self._client.basic_consume( def open(self, queue):
queue=self._queue, if not self._connected:
prefetch_count=1) raise RuntimeError('Not connected to RabbitMQ')
return self
return Subscription(self._client, queue)
def __exit__(self, exc_type, exc_val, exc_tb):
self._ack_last()
promise = self._client.basic_cancel(self._promise) class Subscription(object):
self._client.wait(promise) def __init__(self, client, queue):
return False self._client = client
self._queue = queue
def _ack_last(self): self._promise = None
if self._lastMessage: self._lastMessage = None
self._client.basic_ack(self._lastMessage)
self._lastMessage = None def __enter__(self):
self._promise = self._client.basic_consume(
def get_message(self, timeout=None): queue=self._queue,
if not self._promise: prefetch_count=1)
raise RuntimeError( return self
"Subscription object must be used within 'with' block")
self._ack_last() def __exit__(self, exc_type, exc_val, exc_tb):
self._lastMessage = self._client.wait(self._promise, timeout=timeout) self._ack_last()
#print self._lastMessage promise = self._client.basic_cancel(self._promise)
msg = Message() self._client.wait(promise)
msg.body = anyjson.loads(self._lastMessage['body']) return False
msg.id = self._lastMessage['headers'].get('message_id')
return msg def _ack_last(self):
if self._lastMessage:
self._client.basic_ack(self._lastMessage)
class Message(object): self._lastMessage = None
def __init__(self):
self._body = {} def get_message(self, timeout=None):
self._id = '' if not self._promise:
raise RuntimeError(
@property "Subscription object must be used within 'with' block")
def body(self): self._ack_last()
return self._body self._lastMessage = self._client.wait(self._promise, timeout=timeout)
msg = Message()
@body.setter msg.body = anyjson.loads(self._lastMessage['body'])
def body(self, value): msg.id = self._lastMessage['headers'].get('message_id')
self._body = value return msg
@property
def id(self): class Message(object):
return self._id def __init__(self):
self._body = {}
@id.setter self._id = ''
def id(self, value):
self._id = value or '' @property
def body(self):
return self._body
@body.setter
def body(self, value):
self._body = value
@property
def id(self):
return self._id
@id.setter
def id(self, value):
self._id = value or ''

15
conductor/conductor/reporting.py
View File

@@ -1,3 +1,18 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import xml_code_engine import xml_code_engine
import rabbitmq import rabbitmq

2
conductor/conductor/version.py
View File

@@ -1,5 +1,3 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 OpenStack Foundation # Copyright 2012 OpenStack Foundation
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # Licensed under the Apache License, Version 2.0 (the "License"); you may

71
conductor/conductor/windows_agent.py
View File

@@ -1,29 +1,44 @@
import xml_code_engine # Copyright (c) 2013 Mirantis Inc.
#
from openstack.common import log as logging # Licensed under the Apache License, Version 2.0 (the "License");
log = logging.getLogger(__name__) # you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
def send_command(engine, context, body, template, service, host, mappings=None, # http://www.apache.org/licenses/LICENSE-2.0
result=None, **kwargs): #
if not mappings: # Unless required by applicable law or agreed to in writing, software
mappings = {} # distributed under the License is distributed on an "AS IS" BASIS,
command_dispatcher = context['/commandDispatcher'] # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
def callback(result_value): # See the License for the specific language governing permissions and
log.info( # limitations under the License.
'Received result from {2} for {0}: {1}'.format(
template, result_value, host)) import xml_code_engine
if result is not None:
context[result] = result_value['Result'] from openstack.common import log as logging
log = logging.getLogger(__name__)
success_handler = body.find('success')
if success_handler is not None:
engine.evaluate_content(success_handler, context) def send_command(engine, context, body, template, service, host, mappings=None,
result=None, **kwargs):
command_dispatcher.execute( if not mappings:
name='agent', template=template, mappings=mappings, mappings = {}
host=host, service=service, callback=callback) command_dispatcher = context['/commandDispatcher']
def callback(result_value):
log.info(
'Received result from {2} for {0}: {1}'.format(
template, result_value, host))
if result is not None:
context[result] = result_value['Result']
success_handler = body.find('success')
if success_handler is not None:
engine.evaluate_content(success_handler, context)
command_dispatcher.execute(
name='agent', template=template, mappings=mappings,
host=host, service=service, callback=callback)
xml_code_engine.XmlCodeEngine.register_function(send_command, "send-command") xml_code_engine.XmlCodeEngine.register_function(send_command, "send-command")

18
conductor/conductor/workflow.py
View File

@@ -1,3 +1,18 @@
# Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import jsonpath import jsonpath
import types import types
import re import re
@@ -124,9 +139,6 @@ class Workflow(object):
def _rule_func(match, context, body, engine, limit=0, name=None, **kwargs): def _rule_func(match, context, body, engine, limit=0, name=None, **kwargs):
position = context['__dataSource_currentPosition'] or [] position = context['__dataSource_currentPosition'] or []
# data = context['__dataSource_currentObj']
# if data is None:
# data = context['/dataSource']
position, match = Workflow._get_relative_position(match, context) position, match = Workflow._get_relative_position(match, context)
data = Workflow._get_path(context['/dataSource'], position) data = Workflow._get_path(context['/dataSource'], position)
match = re.sub(r'@\.([\w.]+)', match = re.sub(r'@\.([\w.]+)',

28
conductor/conductor/xml_code_engine.py
View File

@@ -1,4 +1,18 @@
#from lxml import etree # Copyright (c) 2013 Mirantis Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import xml.etree.ElementTree as etree import xml.etree.ElementTree as etree
import types import types
@@ -121,15 +135,3 @@ XmlCodeEngine.register_function(_function_func, "function")
XmlCodeEngine.register_function(_null_func, "null") XmlCodeEngine.register_function(_null_func, "null")
XmlCodeEngine.register_function(_true_func, "true") XmlCodeEngine.register_function(_true_func, "true")
XmlCodeEngine.register_function(_false_func, "false") XmlCodeEngine.register_function(_false_func, "false")
def xprint(context, body, **kwargs):
print "------------------------ start ------------------------"
for arg in kwargs:
print "%s = %s" % (arg, kwargs[arg])
print 'context = ', context
print 'body = %s (%s)' % (body, body.text)
print "------------------------- end -------------------------"
XmlCodeEngine.register_function(xprint, "print")