osm_nbi/subscriptions.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 """
  17 This module implements a thread that reads from kafka bus implementing all the subscriptions.
  18 It is based on asyncio.
  19 To avoid race conditions it uses same engine class as the main module for database changes
  20 For the moment this module only deletes NS instances when they are terminated with the autoremove flag
  21 """
  22
  23 import logging
  24 import threading
  25 import asyncio
  26 from http import HTTPStatus
  27
  28 from osm_common import dbmongo, dbmemory, msglocal, msgkafka
  29 from osm_common.dbbase import DbException
  30 from osm_common.msgbase import MsgException
  31 from osm_nbi.engine import EngineException
  32 from osm_nbi.notifications import NsLcmNotification
  33
  34 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  35
  36
  37 class SubscriptionException(Exception):
  38
  39     def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
  40         self.http_code = http_code
  41         Exception.__init__(self, message)
  42
  43
  44 class SubscriptionThread(threading.Thread):
  45
  46     def __init__(self, config, engine):
  47         """
  48         Constructor of class
  49         :param config: configuration parameters of database and messaging
  50         :param engine: an instance of Engine class, used for deleting instances
  51         """
  52         threading.Thread.__init__(self)
  53         self.to_terminate = False
  54         self.config = config
  55         self.db = None
  56         self.msg = None
  57         self.engine = engine
  58         self.loop = None
  59         self.logger = logging.getLogger("nbi.subscriptions")
  60         self.aiomain_task_admin = None  # asyncio task for receiving admin actions from kafka bus
  61         self.aiomain_task = None  # asyncio task for receiving normal actions from kafka bus
  62         self.internal_session = {  # used for a session to the engine methods
  63             "project_id": (),
  64             "set_project": (),
  65             "admin": True,
  66             "force": False,
  67             "public": None,
  68             "method": "delete",
  69         }
  70         self.nslcm = None
  71
  72     async def start_kafka(self):
  73         # timeout_wait_for_kafka = 3*60
  74         kafka_working = True
  75         while not self.to_terminate:
  76             try:
  77                 # bug 710 635. The library aiokafka does not recieve anything when the topci at kafka has not been
  78                 # created.
  79                 # Before subscribe, send dummy messages
  80                 await self.msg.aiowrite("admin", "echo", "dummy message", loop=self.loop)
  81                 await self.msg.aiowrite("ns", "echo", "dummy message", loop=self.loop)
  82                 await self.msg.aiowrite("nsi", "echo", "dummy message", loop=self.loop)
  83                 if not kafka_working:
  84                     self.logger.critical("kafka is working again")
  85                     kafka_working = True
  86                 if not self.aiomain_task_admin:
  87                     await asyncio.sleep(10, loop=self.loop)
  88                     self.logger.debug("Starting admin subscription task")
  89                     self.aiomain_task_admin = asyncio.ensure_future(self.msg.aioread(("admin",), loop=self.loop,
  90                                                                                      group_id=False,
  91                                                                                      aiocallback=self._msg_callback),
  92                                                                     loop=self.loop)
  93                 if not self.aiomain_task:
  94                     await asyncio.sleep(10, loop=self.loop)
  95                     self.logger.debug("Starting non-admin subscription task")
  96                     self.aiomain_task = asyncio.ensure_future(self.msg.aioread(("ns", "nsi"), loop=self.loop,
  97                                                                                aiocallback=self._msg_callback),
  98                                                               loop=self.loop)
  99                 done, _ = await asyncio.wait([self.aiomain_task, self.aiomain_task_admin],
 100                                              timeout=None, loop=self.loop, return_when=asyncio.FIRST_COMPLETED)
 101                 try:
 102                     if self.aiomain_task_admin in done:
 103                         exc = self.aiomain_task_admin.exception()
 104                         self.logger.error("admin subscription task exception: {}".format(exc))
 105                         self.aiomain_task_admin = None
 106                     if self.aiomain_task in done:
 107                         exc = self.aiomain_task.exception()
 108                         self.logger.error("non-admin subscription task exception: {}".format(exc))
 109                         self.aiomain_task = None
 110                 except asyncio.CancelledError:
 111                     pass
 112             except Exception as e:
 113                 if self.to_terminate:
 114                     return
 115                 if kafka_working:
 116                     # logging only first time
 117                     self.logger.critical("Error accessing kafka '{}'. Retrying ...".format(e))
 118                     kafka_working = False
 119             await asyncio.sleep(10, loop=self.loop)
 120
 121     def run(self):
 122         """
 123         Start of the thread
 124         :return: None
 125         """
 126         self.loop = asyncio.new_event_loop()
 127         try:
 128             if not self.db:
 129                 if self.config["database"]["driver"] == "mongo":
 130                     self.db = dbmongo.DbMongo()
 131                     self.db.db_connect(self.config["database"])
 132                 elif self.config["database"]["driver"] == "memory":
 133                     self.db = dbmemory.DbMemory()
 134                     self.db.db_connect(self.config["database"])
 135                 else:
 136                     raise SubscriptionException("Invalid configuration param '{}' at '[database]':'driver'".format(
 137                         self.config["database"]["driver"]))
 138             if not self.msg:
 139                 config_msg = self.config["message"].copy()
 140                 config_msg["loop"] = self.loop
 141                 if config_msg["driver"] == "local":
 142                     self.msg = msglocal.MsgLocal()
 143                     self.msg.connect(config_msg)
 144                 elif config_msg["driver"] == "kafka":
 145                     self.msg = msgkafka.MsgKafka()
 146                     self.msg.connect(config_msg)
 147                 else:
 148                     raise SubscriptionException("Invalid configuration param '{}' at '[message]':'driver'".format(
 149                         config_msg["driver"]))
 150             self.nslcm = NsLcmNotification(self.db)
 151         except (DbException, MsgException) as e:
 152             raise SubscriptionException(str(e), http_code=e.http_code)
 153
 154         self.logger.debug("Starting")
 155         while not self.to_terminate:
 156             try:
 157
 158                 self.loop.run_until_complete(asyncio.ensure_future(self.start_kafka(), loop=self.loop))
 159             # except asyncio.CancelledError:
 160             #     break  # if cancelled it should end, breaking loop
 161             except Exception as e:
 162                 if not self.to_terminate:
 163                     self.logger.exception("Exception '{}' at messaging read loop".format(e), exc_info=True)
 164
 165         self.logger.debug("Finishing")
 166         self._stop()
 167         self.loop.close()
 168
 169     async def _msg_callback(self, topic, command, params):
 170         """
 171         Callback to process a received message from kafka
 172         :param topic:  topic received
 173         :param command:  command received
 174         :param params: rest of parameters
 175         :return: None
 176         """
 177         msg_to_send = []
 178         try:
 179             if topic == "ns":
 180                 if command == "terminated" and params["operationState"] in ("COMPLETED", "PARTIALLY_COMPLETED"):
 181                     self.logger.debug("received ns terminated {}".format(params))
 182                     if params.get("autoremove"):
 183                         self.engine.del_item(self.internal_session, "nsrs", _id=params["nsr_id"],
 184                                              not_send_msg=msg_to_send)
 185                         self.logger.debug("ns={} deleted from database".format(params["nsr_id"]))
 186                 # Check for nslcm notification
 187                 if isinstance(params, dict):
 188                     # Check availability of operationState and command
 189                     if (not params.get("operationState")) or (not command) or (not params.get("operationParams")):
 190                         self.logger.debug("Message can not be used for notification of nslcm")
 191                     else:
 192                         nsd_id = params["operationParams"].get("nsdId")
 193                         ns_instance_id = params["operationParams"].get("nsInstanceId")
 194                         # Any one among nsd_id, ns_instance_id should be present.
 195                         if not (nsd_id or ns_instance_id):
 196                             self.logger.debug("Message can not be used for notification of nslcm")
 197                         else:
 198                             op_state = params["operationState"]
 199                             event_details = {"topic": topic, "command": command.upper(), "params": params}
 200                             subscribers = self.nslcm.get_subscribers(nsd_id, ns_instance_id, command.upper(), op_state,
 201                                                                      event_details)
 202                             # self.logger.debug("subscribers list: ")
 203                             # self.logger.debug(subscribers)
 204                             if subscribers:
 205                                 asyncio.ensure_future(self.nslcm.send_notifications(subscribers, loop=self.loop),
 206                                                       loop=self.loop)
 207                 else:
 208                     self.logger.debug("Message can not be used for notification of nslcm")
 209             elif topic == "nsi":
 210                 if command == "terminated" and params["operationState"] in ("COMPLETED", "PARTIALLY_COMPLETED"):
 211                     self.logger.debug("received nsi terminated {}".format(params))
 212                     if params.get("autoremove"):
 213                         self.engine.del_item(self.internal_session, "nsis", _id=params["nsir_id"],
 214                                              not_send_msg=msg_to_send)
 215                         self.logger.debug("nsis={} deleted from database".format(params["nsir_id"]))
 216             elif topic == "admin":
 217                 self.logger.debug("received {} {} {}".format(topic, command, params))
 218                 if command in ["echo", "ping"]:   # ignored commands
 219                     pass
 220                 elif command == "revoke_token":
 221                     if params:
 222                         if isinstance(params, dict) and "_id" in params:
 223                             tid = params.get("_id")
 224                             self.engine.authenticator.tokens_cache.pop(tid, None)
 225                             self.logger.debug("token '{}' removed from token_cache".format(tid))
 226                         else:
 227                             self.logger.debug("unrecognized params in command '{} {}': {}"
 228                                               .format(topic, command, params))
 229                     else:
 230                         self.engine.authenticator.tokens_cache.clear()
 231                         self.logger.debug("token_cache cleared")
 232                 else:
 233                     self.logger.debug("unrecognized command '{} {}'".format(topic, command))
 234             # writing to kafka must be done with our own loop. For this reason it is not allowed Engine to do that,
 235             # but content to be written is stored at msg_to_send
 236             for msg in msg_to_send:
 237                 await self.msg.aiowrite(*msg, loop=self.loop)
 238         except (EngineException, DbException, MsgException) as e:
 239             self.logger.error("Error while processing topic={} command={}: {}".format(topic, command, e))
 240         except Exception as e:
 241             self.logger.exception("Exception while processing topic={} command={}: {}".format(topic, command, e),
 242                                   exc_info=True)
 243
 244     def _stop(self):
 245         """
 246         Close all connections
 247         :return: None
 248         """
 249         try:
 250             if self.db:
 251                 self.db.db_disconnect()
 252             if self.msg:
 253                 self.msg.disconnect()
 254         except (DbException, MsgException) as e:
 255             raise SubscriptionException(str(e), http_code=e.http_code)
 256
 257     def terminate(self):
 258         """
 259         This is a threading safe method to terminate this thread. Termination is done asynchronous afterwards,
 260         but not immediately.
 261         :return: None
 262         """
 263         self.to_terminate = True
 264         if self.aiomain_task:
 265             self.loop.call_soon_threadsafe(self.aiomain_task.cancel)
 266         if self.aiomain_task_admin:
 267             self.loop.call_soon_threadsafe(self.aiomain_task_admin.cancel)