dcae-policy/dcaepolicyplugin/discovery.py

   1 # ================================================================================
   2 # Copyright (c) 2017-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 #
  17 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
  18
  19 """client to talk to consul on standard port 8500"""
  20
  21 import base64
  22 import json
  23
  24 import requests
  25 from cloudify import ctx
  26 from cloudify.exceptions import NonRecoverableError
  27
  28 # it is safe to assume that consul agent is at consul:8500
  29 # define consul alis in /etc/hosts on cloudify manager vm
  30 # $ cat /etc/hosts
  31 # 127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4 consul
  32
  33 CONSUL_SERVICE_URL = "http://consul:8500/v1/catalog/service/{0}"
  34 CONSUL_KV_MASK = "http://consul:8500/v1/kv/{0}"
  35
  36
  37 def discover_service_url(service_name):
  38     """find the service record in consul"""
  39     service_url = CONSUL_SERVICE_URL.format(service_name)
  40     ctx.logger.info("getting service_url at {0}".format(service_url))
  41
  42     try:
  43         response = requests.get(service_url, timeout=60)
  44     except requests.ConnectionError as ex:
  45         raise NonRecoverableError(
  46             "ConnectionError - failed to get {0}: {1}".format(service_url, str(ex)))
  47
  48     ctx.logger.info("got {0} for service_url at {1} response: {2}"
  49                     .format(response.status_code, service_url, response.text))
  50
  51     if response.status_code != requests.codes.ok:
  52         return
  53
  54     resp_json = response.json()
  55     if resp_json:
  56         service = resp_json[0]
  57         return "http://{0}:{1}".format(service["ServiceAddress"], service["ServicePort"])
  58
  59
  60 def discover_value(key):
  61     """get the value for the key from consul-kv"""
  62     kv_url = CONSUL_KV_MASK.format(key)
  63     ctx.logger.info("getting kv at {0}".format(kv_url))
  64
  65     try:
  66         response = requests.get(kv_url, timeout=60)
  67     except requests.ConnectionError as ex:
  68         raise NonRecoverableError(
  69             "ConnectionError - failed to get {0}: {1}".format(kv_url, str(ex)))
  70
  71     ctx.logger.info("got {0} for kv at {1} response: {2}"
  72                     .format(response.status_code, kv_url, response.text))
  73
  74     if response.status_code != requests.codes.ok:
  75         return
  76
  77     data = response.json()
  78     if not data:
  79         ctx.logger.error("failed discover_value %s", key)
  80         return
  81     value = base64.b64decode(data[0]["Value"]).decode("utf-8")
  82     ctx.logger.info("consul-kv key=%s value(%s) data=%s",
  83                     key, value, json.dumps(data))
  84     return json.loads(value)