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 from osm_common import dbmongo, dbmemory, msglocal, msgkafka
  28 from osm_common.dbbase import DbException
  29 from osm_common.msgbase import MsgException
  30 from osm_nbi.engine import EngineException
  31
  32 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  33
  34
  35 class SubscriptionException(Exception):
  36
  37     def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
  38         self.http_code = http_code
  39         Exception.__init__(self, message)
  40
  41
  42 class SubscriptionThread(threading.Thread):
  43
  44     def __init__(self, config, engine):
  45         """
  46         Constructor of class
  47         :param config: configuration parameters of database and messaging
  48         :param engine: an instance of Engine class, used for deleting instances
  49         """
  50         threading.Thread.__init__(self)
  51         self.to_terminate = False
  52         self.config = config
  53         self.db = None
  54         self.msg = None
  55         self.engine = engine
  56         self.loop = None
  57         self.logger = logging.getLogger("nbi.subscriptions")
  58         self.aiomain_task = None  # asyncio task for receiving kafka bus
  59         self.internal_session = {  # used for a session to the engine methods
  60             "project_id": (),
  61             "set_project": (),
  62             "admin": True,
  63             "force": False,
  64             "public": None,
  65             "method": "delete",
  66         }
  67
  68     async def start_kafka(self):
  69         # timeout_wait_for_kafka = 3*60
  70         kafka_working = True
  71         while not self.to_terminate:
  72             try:
  73                 # bug 710 635. The library aiokafka does not recieve anything when the topci at kafka has not been
  74                 # created.
  75                 # Before subscribe, send dummy messages
  76                 await self.msg.aiowrite("admin", "echo", "dummy message", loop=self.loop)
  77                 await self.msg.aiowrite("ns", "echo", "dummy message", loop=self.loop)
  78                 await self.msg.aiowrite("nsi", "echo", "dummy message", loop=self.loop)
  79                 if not kafka_working:
  80                     self.logger.critical("kafka is working again")
  81                     kafka_working = True
  82                 await asyncio.sleep(10, loop=self.loop)
  83                 self.aiomain_task = asyncio.ensure_future(self.msg.aioread(("ns", "nsi"), loop=self.loop,
  84                                                                            callback=self._msg_callback),
  85                                                           loop=self.loop)
  86                 await asyncio.wait_for(self.aiomain_task, timeout=None, loop=self.loop)
  87             except Exception as e:
  88                 if self.to_terminate:
  89                     return
  90                 if kafka_working:
  91                     # logging only first time
  92                     self.logger.critical("Error accessing kafka '{}'. Retrying ...".format(e))
  93                     kafka_working = False
  94             await asyncio.sleep(10, loop=self.loop)
  95
  96     def run(self):
  97         """
  98         Start of the thread
  99         :return: None
 100         """
 101         self.loop = asyncio.new_event_loop()
 102         try:
 103             if not self.db:
 104                 if self.config["database"]["driver"] == "mongo":
 105                     self.db = dbmongo.DbMongo()
 106                     self.db.db_connect(self.config["database"])
 107                 elif self.config["database"]["driver"] == "memory":
 108                     self.db = dbmemory.DbMemory()
 109                     self.db.db_connect(self.config["database"])
 110                 else:
 111                     raise SubscriptionException("Invalid configuration param '{}' at '[database]':'driver'".format(
 112                         self.config["database"]["driver"]))
 113             if not self.msg:
 114                 config_msg = self.config["message"].copy()
 115                 config_msg["loop"] = self.loop
 116                 if config_msg["driver"] == "local":
 117                     self.msg = msglocal.MsgLocal()
 118                     self.msg.connect(config_msg)
 119                 elif config_msg["driver"] == "kafka":
 120                     self.msg = msgkafka.MsgKafka()
 121                     self.msg.connect(config_msg)
 122                 else:
 123                     raise SubscriptionException("Invalid configuration param '{}' at '[message]':'driver'".format(
 124                         config_msg["driver"]))
 125
 126         except (DbException, MsgException) as e:
 127             raise SubscriptionException(str(e), http_code=e.http_code)
 128
 129         self.logger.debug("Starting")
 130         while not self.to_terminate:
 131             try:
 132
 133                 self.loop.run_until_complete(asyncio.ensure_future(self.start_kafka(), loop=self.loop))
 134             # except asyncio.CancelledError:
 135             #     break  # if cancelled it should end, breaking loop
 136             except Exception as e:
 137                 if not self.to_terminate:
 138                     self.logger.exception("Exception '{}' at messaging read loop".format(e), exc_info=True)
 139
 140         self.logger.debug("Finishing")
 141         self._stop()
 142         self.loop.close()
 143
 144     def _msg_callback(self, topic, command, params):
 145         """
 146         Callback to process a received message from kafka
 147         :param topic:  topic received
 148         :param command:  command received
 149         :param params: rest of parameters
 150         :return: None
 151         """
 152         try:
 153             if topic == "ns":
 154                 if command == "terminated" and params["operationState"] in ("COMPLETED", "PARTIALLY_COMPLETED"):
 155                     self.logger.debug("received ns terminated {}".format(params))
 156                     if params.get("autoremove"):
 157                         self.engine.del_item(self.internal_session, "nsrs", _id=params["nsr_id"])
 158                         self.logger.debug("ns={} deleted from database".format(params["nsr_id"]))
 159                     return
 160             if topic == "nsi":
 161                 if command == "terminated" and params["operationState"] in ("COMPLETED", "PARTIALLY_COMPLETED"):
 162                     self.logger.debug("received nsi terminated {}".format(params))
 163                     if params.get("autoremove"):
 164                         self.engine.del_item(self.internal_session, "nsis", _id=params["nsir_id"])
 165                         self.logger.debug("nsis={} deleted from database".format(params["nsir_id"]))
 166                     return
 167         except (EngineException, DbException, MsgException) as e:
 168             self.logger.error("Error while processing topic={} command={}: {}".format(topic, command, e))
 169         except Exception as e:
 170             self.logger.exception("Exception while processing topic={} command={}: {}".format(topic, command, e),
 171                                   exc_info=True)
 172
 173     def _stop(self):
 174         """
 175         Close all connections
 176         :return: None
 177         """
 178         try:
 179             if self.db:
 180                 self.db.db_disconnect()
 181             if self.msg:
 182                 self.msg.disconnect()
 183         except (DbException, MsgException) as e:
 184             raise SubscriptionException(str(e), http_code=e.http_code)
 185
 186     def terminate(self):
 187         """
 188         This is a threading safe method to terminate this thread. Termination is done asynchronous afterwards,
 189         but not immediately.
 190         :return: None
 191         """
 192         self.to_terminate = True
 193         if self.aiomain_task:
 194             self.loop.call_soon_threadsafe(self.aiomain_task.cancel)