a1/controller.py

   1 # ==================================================================================
   2 #       Copyright (c) 2019 Nokia
   3 #       Copyright (c) 2018-2019 AT&T Intellectual Property.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #          http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 # ==================================================================================
  17 import json
  18 from flask import Response
  19 from jsonschema import validate
  20 import connexion
  21 from jsonschema.exceptions import ValidationError
  22 from a1 import get_module_logger
  23 from a1 import a1rmr, exceptions, data
  24
  25
  26 logger = get_module_logger(__name__)
  27
  28
  29 def _try_func_return(func):
  30     """
  31     generic caller that returns the apporp http response if exceptions are raised
  32     """
  33     try:
  34         return func()
  35     except ValidationError as exc:
  36         logger.exception(exc)
  37         return "", 400
  38     except exceptions.PolicyTypeAlreadyExists as exc:
  39         logger.exception(exc)
  40         return "", 400
  41     except exceptions.PolicyTypeNotFound as exc:
  42         logger.exception(exc)
  43         return "", 404
  44     except exceptions.PolicyInstanceNotFound as exc:
  45         logger.exception(exc)
  46         return "", 404
  47     except exceptions.MissingManifest as exc:
  48         logger.exception(exc)
  49         return "A1 was unable to find the required RIC manifest. report this!", 500
  50     except exceptions.MissingRmrString as exc:
  51         logger.exception(exc)
  52         return "A1 does not have a mapping for the desired rmr string. report this!", 500
  53     except BaseException as exc:
  54         # catch all, should never happen...
  55         logger.exception(exc)
  56         return Response(status=500)
  57
  58
  59 # Healthcheck
  60
  61
  62 def get_healthcheck():
  63     """
  64     Handles healthcheck GET
  65     Currently, this basically checks the server is alive.a1rmr
  66     """
  67     return "", 200
  68
  69
  70 # Policy types
  71
  72
  73 def get_all_policy_types():
  74     """
  75     Handles GET /a1-p/policytypes
  76     """
  77     return "", 501
  78
  79
  80 def create_policy_type(policy_type_id):
  81     """
  82     Handles PUT /a1-p/policytypes/policy_type_id
  83     """
  84
  85     def _put_type_handler(policy_type_id, body):
  86         data.store_policy_type(policy_type_id, body)
  87         return "", 201
  88
  89     body = connexion.request.json
  90     return _try_func_return(lambda: _put_type_handler(policy_type_id, body))
  91
  92
  93 def get_policy_type(policy_type_id):
  94     """
  95     Handles GET /a1-p/policytypes/policy_type_id
  96     """
  97     return _try_func_return(lambda: data.get_policy_type(policy_type_id))
  98
  99
 100 def delete_policy_type(policy_type_id):
 101     """
 102     Handles DELETE /a1-p/policytypes/policy_type_id
 103     """
 104     return "", 501
 105
 106
 107 # Policy instances
 108
 109
 110 def get_all_instances_for_type(policy_type_id):
 111     """
 112     Handles GET /a1-p/policytypes/policy_type_id/policies
 113     """
 114     return "", 501
 115
 116
 117 def get_policy_instance(policy_type_id, policy_instance_id):
 118     """
 119     Handles GET /a1-p/policytypes/polidyid/policies/policy_instance_id
 120     """
 121     # 200 is automatic here
 122     return _try_func_return(lambda: data.get_policy_instance(policy_type_id, policy_instance_id))
 123
 124
 125 def get_policy_instance_status(policy_type_id, policy_instance_id):
 126     """
 127     Handles GET /a1-p/policytypes/polidyid/policies/policy_instance_id/status
 128     """
 129
 130     def _get_status_handler(policy_type_id, policy_instance_id):
 131         """
 132         Pop trough A1s mailbox, insert the latest status updates into the database, and then return the status vector
 133
 134         NOTE: this is done lazily. Meaning, when someone performs a GET on this API, we pop through a1s mailbox.
 135         THis may not work in the future if there are "thousands" of policy acknowledgements that hit a1 before this is called,
 136         because the rmr mailbox may fill. However, in the near term, we do not expect this to happen.
 137         """
 138         # check validity to 404 first:
 139         data.type_is_valid(policy_type_id)
 140         data.instance_is_valid(policy_type_id, policy_instance_id)
 141
 142         # pop a1s mailbox, looking for policy notifications
 143         new_messages = a1rmr.dequeue_all_waiting_messages(21024)
 144
 145         # try to parse the messages as responses. Drop those that are malformed
 146         for msg in new_messages:
 147             # note, we don't use the parameters "policy_type_id, policy_instance" from above here,
 148             # because we are popping the whole mailbox, which might include other statuses
 149             pay = json.loads(msg["payload"])
 150             if "policy_type_id" in pay and "policy_instance_id" in pay and "handler_id" in pay and "status" in pay:
 151                 data.set_policy_instance_status(
 152                     pay["policy_type_id"], pay["policy_instance_id"], pay["handler_id"], pay["status"]
 153                 )
 154             else:
 155                 logger.debug("Dropping message")
 156                 logger.debug(pay)
 157
 158         # return the status vector
 159         return data.get_policy_instance_statuses(policy_type_id, policy_instance_id)
 160
 161     return _try_func_return(lambda: _get_status_handler(policy_type_id, policy_instance_id))
 162
 163
 164 def create_or_replace_policy_instance(policy_type_id, policy_instance_id):
 165     """
 166     Handles PUT /a1-p/policytypes/polidyid/policies/policy_instance_id
 167     """
 168
 169     def _put_instance_handler(policy_type_id, policy_instance_id, instance):
 170         """
 171         Handles policy instance put
 172
 173         For now, policy_type_id is used as the message type
 174         """
 175         #  validate the PUT against the schema
 176         schema = data.get_policy_type(policy_type_id)["create_schema"]
 177         validate(instance=instance, schema=schema)
 178
 179         # store the instance
 180         data.store_policy_instance(policy_type_id, policy_instance_id, instance)
 181
 182         body = {
 183             "operation": "CREATE",
 184             "policy_type_id": policy_type_id,
 185             "policy_instance_id": policy_instance_id,
 186             "payload": instance,
 187         }
 188
 189         # send rmr (best effort)
 190         a1rmr.send(json.dumps(body), message_type=policy_type_id)
 191
 192         return "", 201
 193
 194     instance = connexion.request.json
 195     return _try_func_return(lambda: _put_instance_handler(policy_type_id, policy_instance_id, instance))
 196
 197
 198 def delete_policy_instance(policy_type_id, policy_instance_id):
 199     """
 200     Handles DELETE /a1-p/policytypes/polidyid/policies/policy_instance_id
 201     """
 202     return "", 501