Merge pull request #254 from Helene/topo_refresher

Add monitor observing if MetaData refresh is required

Author: Helene

source/collector.py

@@ -30,6 +30,7 @@
 from threading import Thread
 from metadata import MetadataHandler
 from bridgeLogger import getBridgeLogger
+from refresher import TopoRefreshManager
 from utils import classattributes, cond_execution_time, get_runtime_statistics
 
 
@@ -73,15 +74,8 @@ def parse_tags(self, filtersMap, defaultLabels):
                     break
             # detected zimon key, do we need refresh local TOPO?
             if not found:
-                cache_size = len(local_cache)
-                local_cache.union(ident)
-                updated_size = len(local_cache)
-                if updated_size > cache_size:
-                    logger.trace(MSG['NewKeyDetected'].format('|'.join(ident)))
-                    md = MetadataHandler()
-                    Thread(name='AdHocMetaDataUpdate', target=md.update).start()
-                else:
-                    logger.trace(MSG['NewKeyAlreadyReported'].format('|'.join(ident)))
+                topoRefresher = TopoRefreshManager()
+                topoRefresher.update_local_cache(ident)
 
                 constructedTags = dict(zip(defaultLabels, ident))
                 if len(self.columnInfo.keys) == 1:

source/messages.py

@@ -35,6 +35,8 @@
        'CollectorErr': 'Failed to initialize connection to pmcollector: {}, quitting',
        'MetaError': 'Metadata could not be retrieved. Check log file for more details, quitting',
        'MetaSuccess': 'Successfully retrieved MetaData',
+       'RequestMetaUpdate': 'MetaData update requested',
+       'ElapsedTimeSinceLastUpdate': 'Elapsed time:{} since last MetaData update',
        'QueryError': 'Query request could not be proceed. Reason: {}',
        'SearchErr': 'Search for {} did cause exception: {}',
        'LookupErr': 'Lookup for metric {} did not return any results',
@@ -66,6 +68,8 @@
        'FileAddedToWatch': 'The file {} added to watching files.',
        'StartWatchingFiles': 'Start watching file changes in the path {}.',
        'StopWatchingFiles': 'Stop watching file changes in the path {}.',
+       'StartMonitoringTopoChanges': 'Start monitoring if a Metadata refresh needed.',
+       'StopMonitoringTopoChanges': 'Stop monitoring Metadata refresh requests.',
        'PathNoCfgFiles': 'The path does not contain any configuration files.',
        'NewKeyDetected': 'Detected not known single ts identifiers {}.',
        'NewKeyAlreadyReported': 'Single ts identifiers {} already reported as not known.',

source/metadata.py

@@ -25,14 +25,15 @@
 from queryHandler.QueryHandler import QueryHandler2 as QueryHandler
 from queryHandler.Topo import Topo
 from queryHandler import SensorConfig
-from utils import execution_time
+from utils import execution_time, synchronized
 from messages import ERR, MSG
 from metaclasses import Singleton
 from time import time, sleep
 from datetime import datetime
+from threading import Lock
 
 
-local_cache = []
+topoUpdateLock = Lock()
 
 
 class MetadataHandler(metaclass=Singleton):
@@ -189,6 +190,7 @@ def GET(self, **params):
         return resp
 
     @execution_time()
+    @synchronized(topoUpdateLock)
     def update(self, refresh_all=False):
         '''Read the topology from ZIMon and update
         the tables for metrics, keys, key elements (tag keys)

source/refresher.py

@@ -0,0 +1,107 @@
+'''
+##############################################################################
+# Copyright 2023 IBM Corp.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+##############################################################################
+
+Created on Dec 9, 2024
+
+@author: HWASSMAN
+'''
+
+import cherrypy
+import time
+from bridgeLogger import getBridgeLogger
+from utils import synchronized
+from metaclasses import Singleton
+from metadata import MetadataHandler
+from messages import MSG
+from threading import Thread, Lock
+
+LOCK = Lock()
+
+
+class TopoRefreshManager(object, metaclass=Singleton):
+    running = False
+    thread = None
+    refresh_delay_secs = 30
+
+    def __init__(self, call_func_on_change=None, *args, **kwargs):
+        self._cached_stamp = {}
+        self.logger = getBridgeLogger()
+        self.update_required = False
+        self.new_keys = set()
+        self.call_func_on_change = call_func_on_change
+        self.args = args
+        self.kwargs = kwargs
+
+    def start_monitor(self):
+        """ Function to start monitor in a thread"""
+        self.running = True
+        if not self.thread:
+            self.thread = Thread(name='TopoRefreshManager', target=self.monitor)
+            self.thread.start()
+            cherrypy.engine.log('Started custom thread %r.' % self.thread.name)
+            self.logger.debug(MSG['StartMonitoringTopoChanges'])
+
+    @synchronized(LOCK)
+    def update_local_cache(self, key):
+        """ Function to cache unknown key locally"""
+        cache_size = len(self.new_keys)
+        self.new_keys = self.new_keys.union(key)
+        updated_size = len(self.new_keys)
+        if updated_size > cache_size:
+            self.logger.trace(MSG['NewKeyDetected'].format('|'.join(key)))
+            self.update_required = True
+        else:
+            self.logger.trace(MSG['NewKeyAlreadyReported'].format('|'.join(key)))
+
+    @synchronized(LOCK)
+    def clear_local_cache(self):
+        """ Function to clear local cache after topo refresh"""
+        self.new_keys.clear()
+        self.update_required = False
+
+    def monitor(self):
+        """ Function to keep watching in a loop """
+        while self.running:
+            try:
+                time.sleep(self.refresh_delay_secs)
+                if self.update_required:
+                    md = MetadataHandler()
+                    elapsed_time = time.time() - md.getUpdateTime
+                    self.logger.trace(MSG['ElapsedTimeSinceLastUpdate'].format(elapsed_time))
+                    if elapsed_time > 300 or self.new_keys > 100:
+                        self.logger.info(MSG['RequestMetaUpdate'])
+                        if self.call_func_on_change is not None:
+                            self.call_func_on_change(*self.args, **self.kwargs)
+                            self.clear_local_cache()
+            except KeyboardInterrupt:
+                self.logger.details(MSG['StopMonitoringTopoChanges'])
+                break
+            except Exception as e:
+                self.logger.warning(
+                    MSG['UnexpecterError'].format(type(e).__name__))
+                pass
+
+    def stop_monitor(self):
+        """ Function to break monitoring """
+        try:
+            self.running = False
+            if self.thread:
+                self.thread.join()
+                cherrypy.engine.log('Stopped custom thread %r.' % self.thread.name)
+                self.thread = None
+        except KeyboardInterrupt:
+            print(f"Recived KeyboardInterrupt during stopping the thread {self.thread.name}")

source/utils.py

@@ -22,6 +22,7 @@
 
 import time
 import copy
+from threading import Lock
 from typing import Callable, TypeVar, Any
 from functools import wraps
 from messages import MSG
@@ -88,6 +89,18 @@ def no_outer(f: Callable[..., T]) -> Callable[..., T]:
     return outer if enabled else no_outer
 
 
+def synchronized(lock: Lock):
+    """Decorator which takes a Lock and runs the function under that Lock"""
+    def funcWrapper(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            with lock:
+                ret = func(*args, **kwargs)
+            return ret
+        return wrapper
+    return funcWrapper
+
+
 def get_runtime_statistics(enabled: bool = False) -> Callable[[Callable[..., T]], Callable[..., T]]:
     """ Conditionally executes the passed through function f with profiling."""
 

source/zimonGrafanaIntf.py

@@ -40,6 +40,7 @@
 from opentsdb import OpenTsdbApi
 from prometheus import PrometheusExporter
 from profiler import Profiler
+from refresher import TopoRefreshManager
 from watcher import ConfigWatcher
 from cherrypy import _cperror
 from cherrypy.lib.cpstats import StatsPage
@@ -392,6 +393,9 @@ def main(argv):
         watcher = ConfigWatcher(files_to_watch, refresh_metadata, refresh_all=True)
         cherrypy.engine.subscribe('start', watcher.start_watch)
         cherrypy.engine.subscribe('stop', watcher.stop_watch)
+        refresher = TopoRefreshManager(refresh_metadata, refresh_all=False)
+        cherrypy.engine.subscribe('start', refresher.start_monitor)
+        cherrypy.engine.subscribe('stop', refresher.stop_monitor)
         cherrypy.engine.start()
         cherrypy.engine.log('test')
         logger.info("%s", MSG['ConnApplications'].format(",\n ".join(registered_apps)))