X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Fdashboarder%2Fdashboarder.py;h=c3f77b29b07c7e0fc3d8d7dc682f33610f2c46e0;hb=98a6cc5f8f38a950bbf7c27b41f5422badb382fa;hp=22aa1bfb26657c0a6f9095813ba609292a97725c;hpb=bc79632385cfd40077c89fa3493db646c0f2bb34;p=osm%2FMON.git diff --git a/osm_mon/dashboarder/dashboarder.py b/osm_mon/dashboarder/dashboarder.py index 22aa1bf..c3f77b2 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,12 +18,14 @@ # 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 socket import asyncio +from urllib.parse import urlparse from osm_mon.dashboarder.service import DashboarderService from osm_mon.core.config import Config @@ -45,26 +47,77 @@ class Dashboarder: def run(self): self.loop.run_until_complete(self.start()) - async def start(self): + async def start(self, wait_time=5): topics = ["users", "project"] - await self.msg_bus.aioread(topics, self._user_msg) + while True: + try: + await self.msg_bus.aioread(topics, self._user_msg) + break + 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)) + # Wait for some time for kaka to stabilize and then reattempt to subscribe again + time.sleep(wait_time) + log.info("Retrying to subscribe the kafka topic(s) %s", str(topics)) 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) + 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") - user = values['username'] + 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'] + 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"] - project_data = values["changes"]["project_role_mappings"] - self.service.create_grafana_team_member(project_data, user_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"] @@ -82,21 +135,30 @@ class Dashboarder: log.exception("Exception processing message: ") def dashboard_forever(self): - log.debug('dashboard_forever') + log.debug("dashboard_forever") + grafana_parsed_uri = urlparse(self.conf.get("grafana", "url")) while True: try: - socket.gethostbyname("grafana") + socket.gethostbyname(grafana_parsed_uri.hostname) log.debug("Dashboard backend is running") except socket.error: log.debug("Dashboard backend is not available") - time.sleep(int(self.conf.get('dashboarder', 'interval'))) + time.sleep(int(self.conf.get("dashboarder", "interval"))) continue try: + self.grafana_cleanup() self.create_dashboards() - time.sleep(int(self.conf.get('dashboarder', 'interval'))) + time.sleep(int(self.conf.get("dashboarder", "interval"))) 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!") + + def grafana_cleanup(self): + # Cleaning up non existing users from grafana + self.service.delete_non_existing_users() + # TODO + # Cleanup of teams from grafana + log.debug("Deleted non existing users from dashbaorder service")