fix: correct variable references in workflow (#56)

---------

Signed-off-by: Gerard Vanloo <gerard.vanloo@ibm.com>

Author: Red-GV

.github/workflows/sre-build-push-images.yaml

@@ -26,13 +26,13 @@ jobs:
         uses: docker/build-push-action@v6
         with:
           context: sre/tools/kubernetes-topology-mapper
-          platforms:
-            - linux/amd64
-            - linux/arm64
+          platforms: |
+            linux/amd64
+            linux/arm64
           push: true
-          tags:
-            - it-bench/topology-monitor:0.0.1
-            - it-bench/topology-monitor:latest
+          tags: |
+            it-bench/topology-monitor:0.0.1
+            it-bench/topology-monitor:latest
   unsupported-checkout-service:
     runs-on: ubuntu-latest
     steps:
@@ -50,19 +50,19 @@ jobs:
         uses: docker/build-push-action@v6
         with:
           context: sre/tools/astronomy-shop-checkout-service
-          platforms:
-            - linux/amd64
+          platforms: |
+            linux/amd64
           push: true
-          tags:
-            - it-bench/unsupported-checkout-service-amd64:0.0.1
-            - it-bench/unsupported-checkout-service-amd64:latest
+          tags: |
+            it-bench/unsupported-checkout-service-amd64:0.0.1
+            it-bench/unsupported-checkout-service-amd64:latest
       - name: Build and push Unsupported Astronomy Shop Checkout Service image (arm)
         uses: docker/build-push-action@v6
         with:
           context: sre/tools/astronomy-shop-checkout-service
-          platforms:
-            - linux/arm64
+          platforms: |
+            linux/arm64
           push: true
-          tags:
-            - it-bench/unsupported-checkout-service-arm64:0.0.1
-            - it-bench/unsupported-checkout-service-arm64:latest
+          tags: |
+            it-bench/unsupported-checkout-service-arm64:0.0.1
+            it-bench/unsupported-checkout-service-arm64:latest

sre/tools/kubernetes-topology-mapper/resource_watcher.py

@@ -1,5 +1,3 @@
-# resource_watcher.py
-
 import logging
 import threading
 import queue
@@ -33,44 +31,44 @@ def __init__(self, k8s_client, topology_manager, event_logger):
         self.k8s_client = k8s_client
         self.topology = topology_manager
         self.event_logger = event_logger
-        
+
         self.watch_threads: Dict[str, threading.Thread] = {}
         self.resource_versions: Dict[str, str] = {}
 
         self.stop_event = threading.Event()
-        
+
         # Bounded queue to prevent unbounded growth
         self.event_queue = queue.Queue(maxsize=10000)
-        
+
         self.processor_thread = None
-        
+
         # Tuning parameters for batch refresh:
         self.REFRESH_EVENT_COUNT = 100   # e.g., refresh after 100 relevant events
         self.REFRESH_TIME_LIMIT = 30.0   # or 30 seconds
 
     def start(self):
         """Start the watch threads and event processor."""
         self.stop_event.clear()
-        
+
         # Start consumer thread to process queued events
         self.processor_thread = threading.Thread(
-            target=self._process_events, 
-            daemon=True, 
+            target=self._process_events,
+            daemon=True,
             name="event-processor"
         )
         self.processor_thread.start()
-        
+
         # Start watches for each resource type
         self._start_resource_watches()
 
     def stop(self):
         """Stop all watch threads and gracefully shut down the processor."""
         self.logger.info("Stopping resource watcher...")
         self.stop_event.set()
-        
+
         # Put sentinel in the queue so _process_events will exit
         self.event_queue.put(None)
-        
+
         # Join the processor thread
         if self.processor_thread:
             self.processor_thread.join(timeout=5)
@@ -94,12 +92,12 @@ def _start_resource_watches(self):
             'Pod', 'Service', 'ConfigMap', 'Secret', 'PersistentVolumeClaim',
             'PersistentVolume', 'Node', 'Namespace', 'ServiceAccount', 'Endpoints'
         }
-        
+
         # Start core resources first
         for kind in core_resources:
             self._start_watch("v1", kind)
             time.sleep(0.5)  # small delay to avoid spamming the API
-            
+
         # Then watch API group resources
         for api_version, kind in self.k8s_client.get_api_resources():
             if kind not in core_resources:
@@ -109,14 +107,14 @@ def _start_resource_watches(self):
     def _start_watch(self, api_version: str, kind: str):
         """Start a watch thread for a specific resource type."""
         watch_key = f"{api_version}/{kind}"
-        
+
         # Avoid duplicates
         if watch_key in self.watch_threads:
             if self.watch_threads[watch_key].is_alive():
                 return
             else:
                 self.logger.info(f"Restarting dead watch thread for {kind}")
-                
+
         self.logger.info(f"Starting watch for {kind}")
         thread = threading.Thread(
             target=self._watch_resource,
@@ -142,16 +140,16 @@ def _watch_resource(self, api_version: str, kind: str):
                 try:
                     list_response = resource.get()
                     resource_version = list_response.metadata.resourceVersion
-                    
+
                     watch_iter = resource.watch(resource_version=resource_version)
                     start_time = time.time()
                     for event in watch_iter:
                         if self.stop_event.is_set() or (time.time() - start_time > 3600):
                             break
-                        
+
                         event_type = event['type']
                         obj = event['object']
-                        
+
                         # If queue is full, this blocks
                         self.event_queue.put({
                             'type': event_type,
@@ -188,10 +186,10 @@ def _get_resource_info(self, obj) -> Dict[str, Any]:
 
         namespace = getattr(obj.metadata, 'namespace', '')
         name = getattr(obj.metadata, 'name', '')
-        
+
         # Use the TopologyManager to get a stable ID (pre-existing function)
         stable_id = self.topology._get_stable_node_id(group, version, obj.kind, namespace, name)
-        
+
         # Extract owners
         owners = []
         if getattr(obj.metadata, 'ownerReferences', None):
@@ -201,7 +199,7 @@ def _get_resource_info(self, obj) -> Dict[str, Any]:
                     'name': ref.name,
                     'uid': ref.uid
                 })
-        
+
         return {
             'kind': obj.kind,
             'group': group,
@@ -216,7 +214,7 @@ def _get_resource_info(self, obj) -> Dict[str, Any]:
     def _process_events(self):
         """
         Main loop: pop events from the queue, log them (with ID/UID/owners),
-        and occasionally refresh topology. 
+        and occasionally refresh topology.
         """
         self.logger.info("Event processing thread started.")
 
@@ -232,7 +230,7 @@ def _process_events(self):
                 event = self.event_queue.get(timeout=1.0)
             except queue.Empty:
                 continue
-            
+
             if event is None:
                 # Sentinel for shutdown
                 break
@@ -264,7 +262,7 @@ def _process_events(self):
                         self.logger.debug(f"Triggering refresh_topology() after "
                                           f"{events_since_last_refresh} events or {int(elapsed)}s elapsed.")
                         self.topology.refresh_topology()
-                        
+
                         # Reset counters
                         last_refresh_time = now
                         events_since_last_refresh = 0
@@ -274,4 +272,4 @@ def _process_events(self):
             finally:
                 self.event_queue.task_done()
 
-        self.logger.info("Event processing thread exiting.")
+        self.logger.info("Event processing thread exiting.")