X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Fdashboarder%2Fdashboarder.py;h=8b16988a40d77a61ed68c91a95e60afe4df18dbf;hb=refs%2Ftags%2Fbranch-bug1511-start;hp=dc21e3349ad49c73038cb0af1b826586a1efff7e;hpb=456d0f323cfcb5fe6b8cad5a6c3e6633875633cd;p=osm%2FMON.git diff --git a/osm_mon/dashboarder/dashboarder.py b/osm_mon/dashboarder/dashboarder.py index dc21e33..8b16988 100644 --- a/osm_mon/dashboarder/dashboarder.py +++ b/osm_mon/dashboarder/dashboarder.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- -# Copyright 2018 Whitestack, LLC +# Copyright 2021 Whitestack, LLC # ************************************************************* # This file is part of OSM Monitoring module @@ -18,36 +18,121 @@ # License for the specific language governing permissions and limitations # under the License. # For those usages not covered by the Apache License, Version 2.0 please -# contact: bdiaz@whitestack.com or glavado@whitestack.com +# contact: fbravo@whitestack.com or glavado@whitestack.com ## + import logging import time - -import peewee +import socket +import asyncio from osm_mon.dashboarder.service import DashboarderService from osm_mon.core.config import Config +from osm_mon.core.message_bus_client import MessageBusClient log = logging.getLogger(__name__) class Dashboarder: - def __init__(self, config: Config): + def __init__(self, config: Config, loop=None): self.conf = config self.service = DashboarderService(config) + if not loop: + loop = asyncio.get_event_loop() + self.loop = loop + self.msg_bus = MessageBusClient(config) + + # run consumer for grafana user management + def run(self): + self.loop.run_until_complete(self.start()) + + async def start(self): + topics = ["users", "project"] + try: + await self.msg_bus.aioread(topics, self._user_msg) + except Exception as e: + # Failed to subscribe to kafka topics + log.error("Error when subscribing to topics %s", str(topics)) + log.exception("Exception %s", str(e)) + + async def _user_msg(self, topic, key, values): + log.debug("Message from kafka bus received: topic: %s and values: %s and key: %s", topic, values, key) + try: + if topic == "users" and key == "created": + log.debug("Received message from kafka for creating user") + if values.get('username'): + user = values['username'] + else: + user = values['changes']['username'] + self.service.create_grafana_user(user) + # user-created and mapping is done with osm cli + if values.get('changes'): + # user-project-role mapping is included in change + if values['changes'].get('project_role_mappings'): + user_id = values["_id"] + project_data = values["changes"]["project_role_mappings"] + project_list = values["changes"].get("projects") + self.service.create_grafana_team_member(project_data, user_id, project_list) + elif values.get('project_role_mappings'): + # for fresh project-role-mapping + user_id = values.get("_id") + project_data = values["project_role_mappings"] + if user_id: + self.service.create_grafana_team_member(project_data, user_id) + else: + # for keystone we will get username + self.service.create_grafana_team_member(project_data, user=values['username']) + elif topic == "users" and key == "deleted": + log.debug("Received message from kafka for deleting user") + user = values['username'] + self.service.delete_grafana_user(user) + elif topic == "users" and key == "edited": + log.debug("Received message from kafka for associating user to team") + user_id = values["_id"] + if values["changes"].get("remove_project_role_mappings") and not \ + values["changes"].get("add_project_role_mappings"): + # Removing user-project role mapping + self.service.remove_grafana_team_members(user_id, + values["changes"].get("remove_project_role_mappings")) + else: + # Changing user project role mapping + if values["changes"].get("project_role_mappings"): + project_data = values["changes"]["project_role_mappings"] + else: + project_data = values["changes"]["add_project_role_mappings"] + self.service.create_grafana_team_member(project_data, user_id) + elif topic == "project" and key == "created": + log.debug("Received message from kafka for creating team") + team_name = values["name"] + self.service.create_grafana_team(team_name) + elif topic == "project" and key == "deleted": + log.debug("Received message from kafka for deleting team") + project_name = values["name"] + self.service.delete_grafana_team(project_name) + elif topic == "project" and key == "edited": + log.debug("Received message from kafka for team name update") + project_old_name = values["original"]["name"] + project_new_name = values["changes"]["name"] + self.service.update_grafana_team(project_new_name, project_old_name) + except Exception: + log.exception("Exception processing message: ") def dashboard_forever(self): log.debug('dashboard_forever') while True: + try: + socket.gethostbyname("grafana") + log.debug("Dashboard backend is running") + except socket.error: + log.debug("Dashboard backend is not available") + time.sleep(int(self.conf.get('dashboarder', 'interval'))) + continue try: self.create_dashboards() time.sleep(int(self.conf.get('dashboarder', 'interval'))) - except peewee.PeeweeException: - log.exception("Database error consuming message: ") - raise except Exception: log.exception("Error creating dashboards") def create_dashboards(self): self.service.create_dashboards() - log.debug('I just called the dashboarder service!') + log.debug('Dashboarder Service > create_dashboards called!')