mod/distributorapi/distributor/http.py

   1 # ============LICENSE_START=======================================================
   2 # Copyright (c) 2019 AT&T Intellectual Property. All rights reserved.
   3 # ================================================================================
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 # ============LICENSE_END=========================================================
  16 """Code for http interface"""
  17
  18 import logging, json
  19 import uuid
  20 from flask import Flask
  21 from flask_cors import CORS
  22 import flask_restplus as frp
  23 from flask_restplus import Api, Resource, fields
  24 from distributor.version import __version__
  25 from distributor import data_access as da
  26 from distributor import config
  27 from distributor import registry_client as rc
  28 from distributor import onboarding_client as oc
  29 from distributor import runtime_client as runc
  30 from distributor import transform as tr
  31
  32
  33 _log = logging.getLogger("distributor.http")
  34
  35 _app = Flask(__name__)
  36 CORS(_app)
  37 # Try to bundle as many errors together
  38 # https://flask-restplus.readthedocs.io/en/stable/parsing.html#error-handling
  39 _app.config['BUNDLE_ERRORS'] = True
  40 _api = Api(_app, version=__version__, title="Distributor HTTP API",
  41         description="HTTP API to manage distribution targets for DCAE design. Distribution targets are DCAE runtime environments that have been registered and are enabled to accept flow design changes that are to be orchestrated in that environment",
  42     contact="", default_mediatype="application/json"
  43     , prefix="/distributor", doc="/distributor", default="distributor"
  44     )
  45 # REVIEW: Do I need a namespace?
  46 ns = _api
  47
  48 model_pg = _api.model("ProcessGroup", {
  49     "id": fields.String(required=True, description="Id for this process group"
  50         , attribute="processGroupId")
  51     , "version": fields.Integer(required=True
  52         , description="Version of the process group")
  53     , "processed": fields.DateTime(required=True
  54         , description="When this process group was processed by this API")
  55     , "runtimeResponse": fields.String(required=True
  56         , description="Full response from the runtime API")
  57     })
  58
  59 model_dt = _api.model("DistributionTarget", {
  60     "selfUrl": fields.Url("resource_distribution_target", absolute=True)
  61     , "id": fields.String(required=True, description="Id for this distribution target"
  62         , attribute="dt_id")
  63     , "name": fields.String(required=True, description="Name for this distribution target"
  64         , attribute="name")
  65     , "runtimeApiUrl": fields.String(required=True
  66         , description="Url to the runtime API for this distribution target"
  67         , attribute="runtimeApiUrl")
  68     , "description": fields.String(required=False
  69         , description="Description for this distribution target"
  70         , attribute="description")
  71     , "nextDistributionTargetId": fields.String(required=False
  72         , description="Id to the next distribution target. Distribution targets can be linked together and have a progression order. Specifying the id of the next distribution target defines the next element int the order."
  73         , attribute="nextDistributionTargetId")
  74     , "created": fields.String(required=True
  75         , description="When this distribution target was created in UTC"
  76         , attribute="created")
  77     , "modified": fields.String(required=True
  78         , description="When this distribution target was last modified in UTC"
  79         , attribute="modified")
  80     , "processGroups": fields.List(fields.Nested(model_pg))
  81     })
  82
  83 model_dts = _api.model("DistributionTargets", {
  84     "distributionTargets": fields.List(fields.Nested(model_dt))
  85     })
  86
  87
  88 parser_dt_req = ns.parser()
  89 parser_dt_req.add_argument("name", required=True, trim=True,
  90         location="json", help="Name for this new distribution target")
  91 parser_dt_req.add_argument("runtimeApiUrl", required=True, trim=True,
  92         location="json", help="Url to the runtime API for this distribution target")
  93 parser_dt_req.add_argument("description", required=False, trim=True,
  94         location="json", help="Description for this distribution target")
  95 parser_dt_req.add_argument("nextDistributionTargetId", required=False, trim=True,
  96         location="json", help="Id of the next distribution target. Distribution targets can be linked together and have a progression order. Specifying the id of the next distribution target defines the next element int the order.")
  97
  98
  99 @ns.route("/distribution-targets", endpoint="resource_distribution_targets")
 100 class DistributionTargets(Resource):
 101     @ns.doc("get_distribution_targets", description="List distribution targets")
 102     @ns.marshal_with(model_dts)
 103     def get(self):
 104         return { "distributionTargets": da.get_distribution_targets() }, 200
 105
 106     @ns.doc("post_distribution_targets", description="Create a new distribution target")
 107     @ns.expect(parser_dt_req)
 108     @ns.marshal_with(model_dt)
 109     def post(self):
 110         req = parser_dt_req.parse_args()
 111         req = da.transform_request(req)
 112         resp = da.add_distribution_target(req)
 113         return resp, 200
 114
 115 @ns.route("/distribution-targets/<string:dt_id>", endpoint="resource_distribution_target")
 116 class DistributionTarget(Resource):
 117     @ns.doc("get_distribution_target", description="Get a distribution target instance")
 118     @ns.response(404, 'Distribution target not found')
 119     @ns.response(500, 'Internal Server Error')
 120     @ns.marshal_with(model_dt)
 121     def get(self, dt_id):
 122         result = da.get_distribution_target(dt_id)
 123
 124         if result:
 125             return result, 200
 126         else:
 127             frp.abort(code=404, message="Unknown distribution target")
 128
 129     @ns.doc("put_distribution_target", description="Update an existing distribution target")
 130     @ns.response(404, 'Distribution target not found')
 131     @ns.response(500, 'Internal Server Error')
 132     @ns.expect(parser_dt_req)
 133     @ns.marshal_with(model_dt)
 134     def put(self, dt_id):
 135         result = da.get_distribution_target(dt_id)
 136
 137         if not result:
 138             frp.abort(code=404, message="Unknown distribution target")
 139
 140         req = parser_dt_req.parse_args()
 141         updated_dt = da.merge_request(result, req)
 142
 143         if da.update_distribution_target(updated_dt):
 144             return updated_dt, 200
 145         else:
 146             frp.abort(code=500, message="Problem with storing the update")
 147
 148     @ns.response(404, 'Distribution target not found')
 149     @ns.response(500, 'Internal Server Error')
 150     @ns.doc("delete_distribution_target", description="Delete an existing distribution target")
 151     def delete(self, dt_id):
 152         if da.delete_distribution_target(dt_id):
 153             return
 154         else:
 155             frp.abort(code=404, message="Unknown distribution target")
 156
 157
 158 parser_post_process_group = ns.parser()
 159 parser_post_process_group.add_argument("processGroupId", required=True,
 160         trim=True, location="json", help="Process group ID that exists in Nifi")
 161
 162 @ns.route("/distribution-targets/<string:dt_id>/process-groups", endpoint="resource_target_process_groups")
 163 class DTargetProcessGroups(Resource):
 164
 165     @ns.response(404, 'Distribution target not found')
 166     @ns.response(501, 'Feature is not supported right now')
 167     @ns.response(500, 'Internal Server Error')
 168     @ns.expect(parser_post_process_group)
 169     def post(self, dt_id):
 170         # TODO: Need bucket ID but for now will simply scan through all buckets
 171         # TODO: Current impl doesn't take into consideration the last state of
 172         # the distribution target e.g. what was the last design processed
 173
 174         req = parser_post_process_group.parse_args()
 175
 176         # Check existence of distribution target
 177
 178         dtarget = da.get_distribution_target(dt_id)
 179
 180         if not dtarget:
 181             frp.abort(code=404, message="Unknown distribution target")
 182
 183         runtime_url = dtarget["runtimeApiUrl"]
 184         pg_id = req["processGroupId"]
 185
 186         # Find flow from Nifi registry
 187
 188         try:
 189             flow = rc.find_flow(config.nifi_registry_url, pg_id)
 190         except Exception as e:
 191             # TODO: Switch to logging
 192             print(e)
 193             # Assuming it'll be 404
 194             frp.abort(code=404, message="Process group not found in registry")
 195
 196         pg_name = flow["name"]
 197
 198         # Make sure graph is setup in runtime api
 199
 200         if runc.ensure_graph(runtime_url, pg_id, pg_name) == False:
 201             frp.abort(code=501 , message="Runtime API: Graph could not be created")
 202
 203         # Graph diffing using Nifi registry
 204
 205         flow_diff = rc.get_flow_diff_latest(config.nifi_registry_url, flow["selfUrl"])
 206
 207         if flow_diff:
 208             # TODO: Not really processing diff right now instead just processing
 209             # latest. Later process the diffs instead and send out the changes.
 210             flow_latest = rc.get_flow_version_latest(config.nifi_registry_url, flow["selfUrl"])
 211         else:
 212             flow_latest = rc.get_flow_version(config.nifi_registry_url, flow["selfUrl"], 1)
 213
 214         # Get component data from onboarding API
 215
 216         components = tr.extract_components_from_flow(flow_latest)
 217
 218         try:
 219             components = oc.get_components_indexed(config.onboarding_api_url, components)
 220         except Exception as e:
 221             # TODO: Switch to logging
 222             print(e)
 223             # Assuming it'll be 404
 224             frp.abort(code=404, message="Component not found in onboarding API")
 225
 226         #
 227         # Put everything together, post to runtime API, save
 228         #
 229
 230         actions = tr.make_fbp_from_flow(flow_latest, components)
 231
 232         resp = dict(req)
 233         resp["version"] = flow_latest["snapshotMetadata"]["version"]
 234         resp["runtimeResponse"] = json.dumps(runc.post_graph(runtime_url, pg_id, actions))
 235         resp = da.add_process_group(dt_id, resp)
 236
 237         if resp:
 238             return resp, 200
 239         else:
 240             frp.abort(code=500, message="Could not store process group")
 241
 242
 243 def start_http_server():
 244     config.init()
 245
 246     def is_debug():
 247         import os
 248         if os.environ.get("DISTRIBUTOR_DEBUG", "1") == "1":
 249             return True
 250         else:
 251             return False
 252
 253     if is_debug():
 254         _app.run(debug=True)
 255     else:
 256         _app.run(host="0.0.0.0", port=8080, debug=False)