Add class that runs Mistral workflow

New class is created that is responsible to run Mistral workflows.
The workflow that can be run hare is already uploaded workflow.

Partially Implements: blueprint murano-mistral-integration
Change-Id: I6b6251d06aa3ce9bf103a67fe354b8c49f52d193
This commit is contained in:
Natasha Beck
2015-01-20 10:18:30 +02:00
parent e5026e098c
commit 90759fa693
3 changed files with 82 additions and 0 deletions

View File

@@ -0,0 +1,5 @@
# Blank class for in-python system libraries, for caching purposes
Namespaces:
=: io.murano.system
Name: MistralClient

View File

@@ -0,0 +1,75 @@
# Copyright (c) 2015 OpenStack Foundation.
#
# 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 eventlet
import json
import murano.dsl.helpers as helpers
import murano.dsl.murano_class as murano_class
import murano.dsl.murano_object as murano_object
import murano.openstack.common.log as logging
LOG = logging.getLogger(__name__)
class MistralError(Exception):
pass
@murano_class.classname('io.murano.system.MistralClient')
class MistralClient(murano_object.MuranoObject):
def initialize(self, _context):
self._clients = helpers.get_environment(_context).clients
def run(self, _context, name, timeout=600, inputs=None, params=None):
mistral_client = self._clients.get_mistral_client(_context)
execution = mistral_client.executions.create(workflow_name=name,
workflow_input=inputs,
params=params)
# For the fire and forget functionality - when we do not want to wait
# for the result of the run.
if timeout == 0:
return execution.id
state = execution.state
try:
# While the workflow is running we continue to wait until timeout.
with eventlet.timeout.Timeout(timeout):
while state not in ('ERROR', 'SUCCESS'):
eventlet.sleep(2)
execution = mistral_client.executions.get(execution.id)
state = execution.state
except eventlet.timeout.Timeout:
error_message = (
'Mistral run timed out. Execution id: {0}.').format(
execution.id)
raise MistralError(error_message)
if state == 'ERROR':
error_message = ('Mistral execution completed with ERROR.'
' Execution id: {0}. Output: {1}').format(
execution.id, execution.output)
raise MistralError(error_message)
# Load the JSON we got from Mistral client to dictionary.
output = json.loads(execution.output)
# Clean the returned dictionary from unnecessary data.
# We want to keep only flow level outputs.
output.pop('openstack', None)
output.pop('__execution', None)
output.pop('task', None)
return output

View File

@@ -20,6 +20,7 @@ from murano.engine.system import agent
from murano.engine.system import agent_listener
from murano.engine.system import heat_stack
from murano.engine.system import instance_reporter
from murano.engine.system import mistralclient
from murano.engine.system import net_explorer
from murano.engine.system import resource_manager
from murano.engine.system import status_reporter
@@ -48,6 +49,7 @@ def register(class_loader, package_loader):
class_loader.import_class(agent.Agent)
class_loader.import_class(agent_listener.AgentListener)
class_loader.import_class(heat_stack.HeatStack)
class_loader.import_class(mistralclient.MistralClient)
class_loader.import_class(ResourceManagerWrapper)
class_loader.import_class(instance_reporter.InstanceReportNotifier)
class_loader.import_class(status_reporter.StatusReporter)