dcae-cli/dcae_cli/util/dmaap.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   3 # ================================================================================
   4 # Copyright (c) 2017 AT&T Intellectual Property. All rights reserved.
   5 # ================================================================================
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #      http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17 # ============LICENSE_END=========================================================
  18 #
  19 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20
  21 """
  22 Functions for DMaaP integration
  23 """
  24 import six
  25 import logging
  26 from jsonschema import validate, ValidationError
  27 from dcae_cli.util import reraise_with_msg
  28 from dcae_cli.util.logger import get_logger
  29 from dcae_cli.catalog.mock.schema import apply_defaults
  30
  31
  32 logger = get_logger('Dmaap')
  33
  34 _SCHEMA = {
  35       "$schema": "http://json-schema.org/draft-04/schema#",
  36       "title": "Schema for dmaap inputs",
  37       "type": "object",
  38       "oneOf": [
  39         { "$ref": "#/definitions/message_router" },
  40         { "$ref": "#/definitions/data_router_publisher" },
  41         { "$ref": "#/definitions/data_router_subscriber" }
  42       ],
  43       "definitions": {
  44         "message_router": {
  45           "type": "object",
  46           "properties": {
  47             "type": {
  48               "type": "string",
  49               "enum": ["message_router"]
  50             },
  51             "aaf_username": {
  52               "type": "string",
  53               "default": None
  54             },
  55             "aaf_password": {
  56               "type": "string",
  57               "default": None
  58             },
  59             "dmaap_info": {
  60               "type": "object",
  61               "properties": {
  62                 "client_role": {
  63                   "type": "string",
  64                   "default": None
  65                 },
  66                 "client_id": {
  67                   "type": "string",
  68                   "default": None
  69                 },
  70                 "location": {
  71                   "type": "string",
  72                   "default": None
  73                 },
  74                 "topic_url": {
  75                   "type": "string"
  76                 }
  77               },
  78               "required": [
  79                 "topic_url"
  80               ],
  81               "additionalProperties": False
  82             }
  83           },
  84           "required": [
  85             "type",
  86             "dmaap_info"
  87           ],
  88           "additionalProperties": False
  89         },
  90         "data_router_publisher": {
  91           "type": "object",
  92           "properties": {
  93             "type": {
  94               "type": "string",
  95               "enum": ["data_router"]
  96             },
  97             "dmaap_info": {
  98               "type": "object",
  99               "properties": {
 100                 "location": {
 101                   "type": "string",
 102                   "default": None,
 103                   "description": "the DCAE location for the publisher, used to set up routing"
 104                 },
 105                 "publish_url": {
 106                   "type": "string",
 107                   "description": "the URL to which the publisher makes Data Router publish requests"
 108                 },
 109                 "log_url": {
 110                   "type": "string",
 111                   "default": None,
 112                   "description": "the URL from which log data for the feed can be obtained"
 113                 },
 114                 "username": {
 115                   "type": "string",
 116                   "default": None,
 117                   "description": "the username the publisher uses to authenticate to Data Router"
 118                 },
 119                 "password": {
 120                   "type": "string",
 121                   "default": None,
 122                   "description": "the password the publisher uses to authenticate to Data Router"
 123                 },
 124                 "publisher_id": {
 125                   "type": "string",
 126                   "default": ""
 127                 }
 128               },
 129               "required": [
 130                 "publish_url"
 131               ],
 132               "additionalProperties": False
 133             }
 134           },
 135           "required": [
 136             "type",
 137             "dmaap_info"
 138           ],
 139           "additionalProperties": False
 140       },
 141       "data_router_subscriber": {
 142           "type": "object",
 143           "properties": {
 144             "type": {
 145               "type": "string",
 146               "enum": ["data_router"]
 147             },
 148             "dmaap_info": {
 149               "type": "object",
 150               "properties": {
 151                 "location": {
 152                   "type": "string",
 153                   "default": None,
 154                   "description": "the DCAE location for the publisher, used to set up routing"
 155                 },
 156                 "delivery_url": {
 157                   "type": "string",
 158                   "description": "the URL to which the Data Router should deliver files"
 159                 },
 160                 "username": {
 161                   "type": "string",
 162                   "default": None,
 163                   "description": "the username Data Router uses to authenticate to the subscriber when delivering files"
 164                 },
 165                 "password": {
 166                   "type": "string",
 167                   "default": None,
 168                   "description": "the username Data Router uses to authenticate to the subscriber when delivering file"
 169                 },
 170                 "subscriber_id": {
 171                   "type": "string",
 172                   "default": ""
 173                 }
 174               },
 175               "additionalProperties": False
 176             }
 177           },
 178           "required": [
 179             "type",
 180             "dmaap_info"
 181           ],
 182           "additionalProperties": False
 183       }
 184     }
 185 }
 186
 187
 188 _validation_msg = """
 189 Is your DMaaP client object a valid json?
 190 Does your DMaaP client object follow this format?
 191
 192 Message router:
 193
 194     {
 195         "aaf_username": <string, optional>,
 196         "aaf_password": <string, optional>,
 197         "type": "message_router",
 198         "dmaap_info": {
 199             "client_role": <string, optional>,
 200             "client_id": <string, optional>,
 201             "location": <string, optional>,
 202             "topic_url": <string, required>
 203         }
 204     }
 205
 206 Data router (publisher):
 207
 208     {
 209         "type": "data_router",
 210         "dmaap_info": {
 211             "location": <string, optional>,
 212             "publish_url": <string, required>,
 213             "log_url": <string, optional>,
 214             "username": <string, optional>,
 215             "password": <string, optional>,
 216             "publisher_id": <string, optional>
 217         }
 218     }
 219
 220 Data router (subscriber):
 221
 222     {
 223         "type": "data_router",
 224         "dmaap_info": {
 225             "location": <string, optional>,
 226             "delivery_url": <string, optional>,
 227             "username": <string, optional>,
 228             "password": <string, optional>,
 229             "subscriber_id": <string, optional>
 230         }
 231     }
 232
 233 """
 234
 235 def validate_dmaap_map_schema(dmaap_map):
 236     """Validate the dmaap map schema"""
 237     for k, v in six.iteritems(dmaap_map):
 238         try:
 239             validate(v, _SCHEMA)
 240         except ValidationError as e:
 241             logger.error("DMaaP validation issue with \"{k}\"".format(k=k))
 242             logger.error(_validation_msg)
 243             reraise_with_msg(e, as_dcae=True)
 244
 245
 246 class DMaaPValidationError(RuntimeError):
 247     pass
 248
 249 def _find_matching_definition(instance):
 250     """Find and return matching definition given an instance"""
 251     for subsection in ["message_router", "data_router_publisher",
 252             "data_router_subscriber"]:
 253         try:
 254             validate(instance, _SCHEMA["definitions"][subsection])
 255             return _SCHEMA["definitions"][subsection]
 256         except ValidationError:
 257             pass
 258
 259     # You should never get here but just in case..
 260     logger.error("No matching definition: {0}".format(instance))
 261     raise DMaaPValidationError("No matching definition")
 262
 263 def apply_defaults_dmaap_map(dmaap_map):
 264     """Apply the defaults to the dmaap map"""
 265     def grab_properties(instance):
 266         return _find_matching_definition(instance)["properties"]
 267
 268     return { k: apply_defaults(grab_properties(v), v) for k,v in
 269             six.iteritems(dmaap_map) }
 270
 271
 272 def validate_dmaap_map_entries(dmaap_map, mr_config_keys, dr_config_keys):
 273     """Validate dmaap map entries
 274
 275     Validate dmaap map to make sure all config keys are there and that there's
 276     no additional config keys beceause this map is used in generating the
 277     configuration json.
 278
 279     Returns:
 280     --------
 281     True when dmaap_map is ok and False when it is not
 282     """
 283     # Catch when there is no dmaap_map when there should be
 284     if len(mr_config_keys) + len(dr_config_keys) > 0 and len(dmaap_map) == 0:
 285         logger.error("You have dmaap streams defined in your specification")
 286         logger.error("You must provide a dmaap json to resolve those dmaap streams.")
 287         logger.error("Please use the \"--dmaap-file\" option")
 288         return False
 289
 290     # Look for missing keys
 291     is_missing = lambda config_key: config_key not in dmaap_map
 292     missing_keys = list(filter(is_missing, mr_config_keys))
 293
 294     if missing_keys:
 295         logger.error("Missing config keys in dmaap json: {0}".format(
 296             ",".join(missing_keys)))
 297         logger.error("Re-edit your dmaap json")
 298         return False
 299
 300     # Look for unexpected keys
 301     is_unexpected = lambda config_key: config_key not in mr_config_keys
 302     unexpected_keys = list(filter(is_unexpected, dmaap_map.keys()))
 303
 304     if unexpected_keys:
 305         # NOTE: Changed this to a non-error in order to support the feature of
 306         # developer having a master dmaap map
 307         logger.warn("Unexpected config keys in dmaap json: {0}".format(
 308             ",".join(unexpected_keys)))
 309         return True
 310
 311     return True
 312
 313
 314 def update_delivery_urls(get_route_func, target_base_url, dmaap_map):
 315     """Update delivery urls for dmaap map
 316
 317     This method picks out all the data router connections for subscribers and
 318     updates the delivery urls with the supplied base target url concatentated
 319     with the user specified route (or path).
 320
 321     Args:
 322     -----
 323     get_route_func (func): Function that takes a config_key and returns the route
 324         used for the data router subscriber
 325     target_base_url (string): "{http|https}://<hostname>:<port>"
 326     dmaap_map (dict): DMaaP map is map of inputs that is config_key to provisioned
 327         data router feed or message router topic connection details
 328
 329     Returns:
 330     --------
 331     Returns the updated DMaaP map
 332     """
 333     def update_delivery_url(config_key, dm):
 334         route = get_route_func(config_key)
 335         dm["dmaap_info"]["delivery_url"] = "{base}{tween}{path}".format(base=target_base_url,
 336                 path=route, tween="" if route[0] == "/" else "/")
 337         return dm
 338
 339     def is_dr_subscriber(dm):
 340         return dm["type"] == "data_router" and "publish_url" not in dm["dmaap_info"]
 341
 342     updated_map = { config_key: update_delivery_url(config_key, dm)
 343             for config_key, dm in six.iteritems(dmaap_map) if is_dr_subscriber(dm) }
 344     dmaap_map.update(updated_map)
 345
 346     return dmaap_map
 347
 348
 349 def list_delivery_urls(dmaap_map):
 350     """List delivery urls
 351
 352     Returns:
 353     --------
 354     List of tuples (config_key, deliery_url)
 355     """
 356     return [(config_key, dm["dmaap_info"]["delivery_url"]) \
 357             for config_key, dm in six.iteritems(dmaap_map) if "delivery_url" in dm["dmaap_info"]]