osm_mon/dashboarder/dashboarder.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright 2018 Whitestack, LLC
   4 # *************************************************************
   5
   6 # This file is part of OSM Monitoring module
   7 # All Rights Reserved to Whitestack, LLC
   8
   9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  10 # not use this file except in compliance with the License. You may obtain
  11 # a copy of the License at
  12
  13 #         http://www.apache.org/licenses/LICENSE-2.0
  14
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  18 # License for the specific language governing permissions and limitations
  19 # under the License.
  20 # For those usages not covered by the Apache License, Version 2.0 please
  21 # contact: bdiaz@whitestack.com or glavado@whitestack.com
  22 ##
  23 import logging
  24 import time
  25 import socket
  26 import asyncio
  27
  28 from osm_mon.dashboarder.service import DashboarderService
  29 from osm_mon.core.config import Config
  30 from osm_mon.core.message_bus_client import MessageBusClient
  31
  32 log = logging.getLogger(__name__)
  33
  34
  35 class Dashboarder:
  36     def __init__(self, config: Config, loop=None):
  37         self.conf = config
  38         self.service = DashboarderService(config)
  39         if not loop:
  40             loop = asyncio.get_event_loop()
  41         self.loop = loop
  42         self.msg_bus = MessageBusClient(config)
  43
  44     # run consumer for grafana user management
  45     def run(self):
  46         self.loop.run_until_complete(self.start())
  47
  48     async def start(self):
  49         topics = ["users", "project"]
  50         await self.msg_bus.aioread(topics, self._user_msg)
  51
  52     async def _user_msg(self, topic, key, values):
  53         log.debug("Message from kafka bus received: topic: %s and values: %s and key: %s", topic, values, key)
  54         try:
  55             if topic == "users" and key == "created":
  56                 log.debug("Received message from kafka for creating user")
  57                 user = values['username']
  58                 self.service.create_grafana_user(user)
  59             elif topic == "users" and key == "deleted":
  60                 log.debug("Received message from kafka for deleting user")
  61                 user = values['username']
  62                 self.service.delete_grafana_user(user)
  63             elif topic == "users" and key == "edited":
  64                 log.debug("Received message from kafka for associating user to team")
  65                 user_id = values["_id"]
  66                 project_data = values["changes"]["project_role_mappings"]
  67                 self.service.create_grafana_team_member(project_data, user_id)
  68             elif topic == "project" and key == "created":
  69                 log.debug("Received message from kafka for creating team")
  70                 team_name = values["name"]
  71                 self.service.create_grafana_team(team_name)
  72             elif topic == "project" and key == "deleted":
  73                 log.debug("Received message from kafka for deleting team")
  74                 project_name = values["name"]
  75                 self.service.delete_grafana_team(project_name)
  76             elif topic == "project" and key == "edited":
  77                 log.debug("Received message from kafka for team name update")
  78                 project_old_name = values["original"]["name"]
  79                 project_new_name = values["changes"]["name"]
  80                 self.service.update_grafana_team(project_new_name, project_old_name)
  81         except Exception:
  82             log.exception("Exception processing message: ")
  83
  84     def dashboard_forever(self):
  85         log.debug('dashboard_forever')
  86         while True:
  87             try:
  88                 socket.gethostbyname("grafana")
  89                 log.debug("Dashboard backend is running")
  90             except socket.error:
  91                 log.debug("Dashboard backend is not available")
  92                 time.sleep(int(self.conf.get('dashboarder', 'interval')))
  93                 continue
  94             try:
  95                 self.create_dashboards()
  96                 time.sleep(int(self.conf.get('dashboarder', 'interval')))
  97             except Exception:
  98                 log.exception("Error creating dashboards")
  99
 100     def create_dashboards(self):
 101         self.service.create_dashboards()
 102         log.debug('I just called the dashboarder service!')