X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Fdashboarder%2Fdashboarder.py;h=b25a8d9d3f308816f55941d0e6c0bfda878cbcee;hb=a353daa53d1b4d8da6fcbfefd40f309a5cf33688;hp=22aa1bfb26657c0a6f9095813ba609292a97725c;hpb=bc79632385cfd40077c89fa3493db646c0f2bb34;p=osm%2FMON.git diff --git a/osm_mon/dashboarder/dashboarder.py b/osm_mon/dashboarder/dashboarder.py index 22aa1bf..b25a8d9 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,78 @@ 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) + log.info("Sucessfully subscribed to kafka topic(s) %s", str(topics)) + break + except Exception as e: + # Failed to subscribe to kafka topics + log.error("Error when subscribing to topic(s) %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 +136,22 @@ 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.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!")