--- /dev/null
+#
+# ============LICENSE_START===================================================
+# Copyright (c) 2017 Cloudify.co. All rights reserved.
+# ===================================================================
+# 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.
+# ============LICENSE_END====================================================
+#
+
+
+import os
+from cStringIO import StringIO
+from flask import Flask, render_template, request, jsonify
+from flask_autodoc.autodoc import Autodoc
+from aria import install_aria_extensions
+from aria.parser import consumption
+from aria.utils import formatting, collections
+from aria.cli.core import aria
+from aria.cli import utils
+from aria.exceptions import ParsingError, DependentServicesError
+from aria.core import Core
+from aria.cli import service_template_utils
+from aria.storage import exceptions as storage_exceptions
+from aria.utils import threading
+from aria.orchestrator.workflow_runner import WorkflowRunner
+from aria.orchestrator.workflows.executor.dry import DryExecutor
+import util
+
+version_id = "0.1"
+route_base = "/api/" + version_id + "/"
+app = Flask("onap-aria-rest")
+auto = Autodoc(app)
+
+# TODO Garbage collect this dict somehow
+execution_state = util.SafeDict()
+
+
+def main():
+ install_aria_extensions()
+ app.run(host='0.0.0.0', port=5000, threaded=True)
+
+
+@app.route("/")
+@app.route("/api")
+@app.route("/docs")
+def index():
+ return auto.html()
+
+
+###
+# TEMPLATES
+###
+
+# add template
+@app.route(route_base + "templates/<template_name>", methods=['PUT'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_resource_storage
+@aria.pass_plugin_manager
+@aria.pass_logger
+def install_template(template_name, model_storage, resource_storage,
+ plugin_manager, logger):
+ """
+ installs a template in Aria storage
+ """
+ body = request.json
+
+ # Check body
+ if "service_template_path" in body:
+ service_template_path = body["service_template_path"]
+ else:
+ return "request body missing service_template_path", 501
+
+ if "service_template_filename" in body:
+ service_template_filename = body["service_template_filename"]
+ else:
+ service_template_filename = "service-template.yaml"
+
+ service_template_path = service_template_utils.get(
+ service_template_path, service_template_filename)
+
+ core = Core(model_storage, resource_storage, plugin_manager)
+
+ try:
+ core.create_service_template(service_template_path,
+ os.path.dirname(service_template_path),
+ template_name)
+ except storage_exceptions.StorageError as e:
+ logger.error("storage exception")
+ utils.check_overriding_storage_exceptions(
+ e, 'service template', template_name)
+ return e.message, 500
+ except Exception as e:
+ logger.error("catchall exception")
+ return e.message, 500
+
+ return "service template installed", 200
+
+# validate template
+@app.route(route_base + "templates", methods=['POST'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_resource_storage
+@aria.pass_plugin_manager
+@aria.pass_logger
+def validate_template(model_storage, resource_storage, plugin_manager, logger):
+ """
+ Validates a TOSCA template
+ """
+ body = request.json
+
+ # Check body
+ if "service_template_path" in body:
+ service_template_path = body["service_template_path"]
+ else:
+ return "request body missing service_template_path", 501
+ if "service_template_filename" in body:
+ service_template_filename = body["service_template_filename"]
+ else:
+ service_template_filename = "service-template.yaml"
+
+ service_template_path = service_template_utils.get(
+ service_template_path, service_template_filename)
+
+ core = Core(model_storage, resource_storage, plugin_manager)
+ try:
+ context = core.validate_service_template(service_template_path)
+ except ParsingError as e:
+ return e.message, 400
+
+ logger.info('Service template {} validated'.format(service_template_path))
+ return "", 200
+
+
+# delete template
+@app.route(route_base + "templates/<template_id>", methods=['DELETE'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_resource_storage
+@aria.pass_plugin_manager
+@aria.pass_logger
+def delete_template(
+ template_id,
+ model_storage,
+ resource_storage,
+ plugin_manager,
+ logger):
+ """
+ Deletes a template from Aria storage
+ """
+
+ logger.info('Deleting service template {}'.format(template_id))
+ core = Core(model_storage, resource_storage, plugin_manager)
+ try:
+ core.delete_service_template(template_id)
+ except DependentServicesError as e:
+ logger.error("dependent services error")
+ return e.message, 400
+ except Exception as e:
+ logger.error("failed")
+ return "Failed to delete template", 500
+
+ logger.info('Service template {} deleted'.format(template_id))
+ return "", 200
+
+
+# get template json
+@app.route(route_base + "templates/<template_id>/json", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def get_template_json(template_id, model_storage, logger):
+ """ get JSON representation of template """
+ template = model_storage.service_template.get(template_id)
+ consumption.ConsumptionContext()
+ body = formatting.json_dumps(collections.prune(template.as_raw))
+ return body
+
+
+# list templates
+@app.route(route_base + "templates", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def list_templates(model_storage, logger):
+ """
+ Lists templates installed in Aria storage
+ """
+ list = model_storage.service_template.list()
+ templates = []
+ for item in list:
+ templates.append({"name": item.name,
+ "id": item.id,
+ "description": item.description
+ })
+ return jsonify(templates)
+
+
+# list nodes
+@app.route(route_base + "templates/<template_id>/nodes", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def list_nodes_by_template(template_id, model_storage, logger):
+ """
+ Lists node templates in specified Aria template
+ """
+ service_template = model_storage.service_template.get(template_id)
+ filters = dict(service_template=service_template)
+ nodes = model_storage.node_template.list(filters=filters)
+ nodelist = []
+
+ for node in nodes:
+ nodelist.append({
+ "id": node.id,
+ "name": node.name,
+ "description": node.description,
+ "service_template_id": service_template.id,
+ "type_name": node.type_name
+ })
+ return jsonify(nodelist), 200
+
+
+# show node details
+@app.route(route_base + "nodes/<node_id>", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def get_node(node_id, model_storage, logger):
+ """
+ Get node details
+ """
+ node_template = model_storage.node_template.get(node_id)
+ service_template = model_storage.service_template.get_by_name(
+ node_template.service_template_name)
+ retmap = {}
+ retmap['id'] = node_id
+ retmap['name'] = node_template.name
+ retmap['description'] = node_template.description
+ retmap['service_template_id'] = service_template.id
+ retmap['type_name'] = node_template.type_name
+ return jsonify(retmap), 200
+
+###
+# SERVICES
+###
+
+
+# list services
+@app.route(route_base + "services", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def list_services(model_storage, logger):
+ """
+ Lists all services
+ """
+ services_list = model_storage.service.list()
+ outlist = []
+ for service in services_list:
+ outlist.append({"id": service.id,
+ "description": service.description,
+ "name": service.name,
+ "service_template": service.service_template.name,
+ "created": service.created_at,
+ "updated": service.updated_at})
+ return jsonify(outlist), 200
+
+
+# show service
+@app.route(route_base + "services/<service_id>", methods=['GET'])
+def show_service(service_id):
+ """
+ Returns details for specified servie
+ """
+ return "not implemented", 501
+
+
+# get service outputs
+@app.route(route_base + "services/<service_id>/outputs", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def get_service_outputs(service_id, model_storage, logger):
+ """
+ Gets outputs for specified service
+ """
+ service = model_storage.service.get(service_id)
+ outlist = []
+ for output_name, output in service.outputs.iteritems():
+ outlist.append({"name": output_name, "description": output.description,
+ "value": output.value})
+ return jsonify(outlist)
+
+
+# get service inputs
+@app.route(route_base + "services/<service_id>/inputs", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def get_service_inputs(service_id, model_storage, logger):
+ """
+ Gets inputs for specified service
+ """
+ service = model_storage.service.get(service_id)
+ outlist = []
+ for input_name, input in service.inputs.iteritems():
+ outlist.append({"name": input_name, "description": input.description,
+ "value": input.value})
+ return jsonify(outlist)
+
+
+# create service
+@app.route(route_base + "templates/<template_id>/services/<service_name>",
+ methods=['POST'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_resource_storage
+@aria.pass_plugin_manager
+@aria.pass_logger
+def create_service(template_id, service_name, model_storage, resource_storage,
+ plugin_manager, logger):
+ """
+ Creates a service from the specified service template
+ """
+ body = request.json
+ inputs = {}
+ if 'inputs' in body:
+ inputs = body['inputs']
+ core = Core(model_storage, resource_storage, plugin_manager)
+ service = core.create_service(template_id, inputs, service_name)
+
+ logger.info("service {} created".format(service.name))
+ return "service {} created".format(service.name), 200
+
+
+# delete service
+@app.route(route_base + "services/<service_id>", methods=['DELETE'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_resource_storage
+@aria.pass_plugin_manager
+@aria.pass_logger
+def delete_service(
+ service_id,
+ model_storage,
+ resource_storage,
+ plugin_manager,
+ logger):
+ """
+ Deletes the specified servi e
+ """
+ service = model_storage.service.get(service_id)
+ core = Core(model_storage, resource_storage, plugin_manager)
+ core.delete_service(service_id, force=True)
+ return "service {} deleted".format(service.id), 200
+
+
+###
+# WORKFLOWS
+###
+
+
+# list workflows
+@app.route(route_base + "services/<service_id>/workflows", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def list_workflows(service_id, model_storage, logger):
+ """
+ Lists all defined user workflows for the specified service
+ """
+ service = model_storage.service.get(service_id)
+ workflows = service.workflows.itervalues()
+ outlist = []
+ for workflow in workflows:
+ outlist.append(workflow.name)
+ return jsonify(outlist), 200
+
+
+# show workflow
+@app.route(
+ route_base +
+ "services/<service_id>/workflow/<workflow_name>",
+ methods=['GET'])
+def show_workflow(service_name, workflow_name):
+ """
+ Returns details of specified workflow
+ """
+ return "not implemented", 501
+
+###
+# EXECUTIONS
+###
+
+
+# list all executions
+@app.route(route_base + "executions", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def list_executions(model_storage, logger):
+ """
+ Return all executions
+ """
+ elist = model_storage.execution.list()
+ outlist = []
+ for execution in elist:
+ outlist.append(
+ {"execution_id": execution.id,
+ "workflow_name": execution.workflow_name,
+ "service_template_name": execution.service_template_name,
+ "service_name": execution.service_name,
+ "status": execution.status})
+ return jsonify(outlist), 200
+
+
+# list executions for service
+@app.route(route_base + "services/<service_id>/executions", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def list_service_executions(service_id, model_storage, logger):
+ """
+ Return all executions for specified service
+ """
+ service = model_storage.service.get(service_id)
+ elist = model_storage.execution.list(filters=dict(service=service))
+ outlist = []
+ for execution in elist:
+ outlist.append(
+ {"execution_id": execution.id,
+ "workflow_name": execution.workflow_name,
+ "service_template_name": execution.service_template_name,
+ "service_name": execution.service_name,
+ "status": execution.status})
+ return jsonify(outlist), 200
+
+
+# show execution
+@app.route(route_base + "executions/<execution_id>", methods=['GET'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def show_execution(execution_id, model_storage, logger):
+ """
+ Return details of specified execution
+ """
+ try:
+ execution = model_storage.execution.get(execution_id)
+ except BaseException:
+ return "Execution {} not found".format(execution_id), 404
+
+ return jsonify({"execution_id": execution_id,
+ "service_name": execution.service_name,
+ "service_template_name": execution.service_template_name,
+ "workflow_name": execution.workflow_name,
+ "status": execution.status}), 200
+
+# start execution
+
+
+# TODO allow executors other than default and dry to be used
+@app.route(
+ route_base +
+ "services/<service_id>/executions/<workflow_name>",
+ methods=['POST'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_resource_storage
+@aria.pass_plugin_manager
+@aria.pass_logger
+def start_execution(
+ service_id,
+ workflow_name,
+ model_storage,
+ resource_storage,
+ plugin_manager,
+ logger):
+ """
+ Start an execution for the specified service
+ """
+ body = request.json
+ executor = DryExecutor(
+ ) if 'executor' in body and body['executor'] == 'dry' else None
+
+ inputs = body['inputs'] if 'inputs' in body else None
+ task_max_attempts = (body['task_max_attempts']
+ if 'task_max_attempts' in body else 30)
+ task_retry_interval = (body['task_retry_interval']
+ if 'task_retry_interval' in body else 30)
+
+ runner = WorkflowRunner(model_storage, resource_storage, plugin_manager,
+ service_id=service_id,
+ workflow_name=workflow_name,
+ inputs=inputs,
+ executor=executor,
+ task_max_attempts=task_max_attempts,
+ task_retry_interval=task_retry_interval)
+
+ service = model_storage.service.get(service_id)
+ tname = '{}_{}_{}'.format(service.name, workflow_name, runner.execution_id)
+ thread = threading.ExceptionThread(target=runner.execute,
+ name=tname)
+ thread.start()
+ execution_state[str(runner.execution_id)] = [runner, thread]
+ return jsonify({"id": runner.execution_id}), 202
+
+
+# resume execution
+@app.route(route_base + "executions/<execution_id>", methods=['POST'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_resource_storage
+@aria.pass_plugin_manager
+@aria.pass_logger
+def resume_execution(
+ execution_id,
+ model_storage,
+ resource_storage,
+ plugin_manager,
+ logger):
+ """
+ Resume the specified execution
+ """
+ body = request.json
+ execution = model_storage.execution.get(execution_id)
+ if execution.status != execution.status.CANCELLED:
+ return "cancelled execution cannot be resumed", 400
+ executor = DryExecutor(
+ ) if 'executor' in body and body['executor'] == 'dry' else None
+ retry_failed_tasks = body['retry_failed_tasks'] \
+ if 'retry_failed_tasks' in body else False
+
+ runner = WorkflowRunner(model_storage, resource_storage, plugin_manager,
+ execution_id=execution_id,
+ executor=executor,
+ retry_failed_tasks=retry_failed_tasks)
+
+ tname = '{}_{}_{}'.format(execution.service.name, execution.workflow_name,
+ runner.execution_id)
+ thread = threading.ExceptionThread(target=runner.execute,
+ name=tname,
+ daemon=True)
+ thread.start()
+ execution_state[str(runner.execution_id)] = [runner, thread]
+ return jsonify({"id": runner.execution_id}), 202
+
+
+# cancel execution
+@app.route(route_base + "executions/<execution_id>", methods=['DELETE'])
+@auto.doc()
+@aria.pass_model_storage
+@aria.pass_logger
+def cancel_execution(execution_id, model_storage, logger):
+ """
+ Cancel the specified execution
+ """
+ logger.info("cancelling execution {}".format(execution_id))
+ body = request.json
+
+ try:
+ execution = model_storage.execution.get(execution_id)
+ except BaseException:
+ return "Execution {} not found".format(execution_id), 404
+
+ if (not execution.status == execution.PENDING and
+ not execution.status == execution.STARTED):
+ return "Cancel ignored. Execution state = {}".format(
+ execution.status), 200
+
+ if execution_id not in execution_state:
+ logger.error("id {} not found".format(execution_id))
+ return "execution id {} not found".format(execution_id), 400
+
+ einfo = execution_state[execution_id]
+ runner = einfo[0]
+ thread = einfo[1]
+ timeout = 30 # seconds to wait for thread death
+ if 'timeout' in body:
+ timeout = body['timeout']
+
+ runner.cancel()
+ while thread.is_alive() and timeout > 0:
+ thread.join(1)
+ if not thread.is_alive():
+ return "execution {} cancelled".format(execution_id), 200
+ timeout = timeout - 1
+ if timeout == 0:
+ return "execution cancel timed out", 500
+ return "execution {} cancelled".format(execution_id), 200
+
+
+if __name__ == "__main__":
+ app.run(host='0.0.0.0', port=5000, threaded=True)
Code Review / so.git / commitdiff