add parallel processing

Author: frct1

requirements-dev.txt

@@ -1,4 +1,5 @@
 openstacksdk==4.2.0
+httpx==0.28.1
 prometheus_client==0.21.1
 pytest<9.0.0,>=8.0.0
 pytest-cov==6.0.0

requirements.txt

@@ -1,2 +1,3 @@
 openstacksdk==4.2.0
+httpx==0.28.1
 prometheus_client==0.21.1

src/collectors/instances_per_hypervisor.py

@@ -3,43 +3,49 @@
 import time
 import logging
 from util import instances_per_hypervisor
+import asyncio
 
 logger = logging.getLogger(__name__)
 
 
 class InstancesPerHypervisorCollector(Collector):
-    def __init__(self, cloud_name):
+    def __init__(self, cloud_name, cache_time):
         super().__init__()
         self.cloud_name = cloud_name
         self.metrics = []
         self.last_update = 0
-        self.cache_time = 60
+        self.cache_time = cache_time
+        self.gauge = None
 
     def _fetch_metrics(self):
+        """Fetch metrics from OpenStack and update the gauge."""
         try:
-            new_metrics = instances_per_hypervisor.export_metrics(self.cloud_name)
-            self.metrics = new_metrics
-            self.last_update = time.time()
+            # Clear previous metrics
+            self.gauge = GaugeMetricFamily(
+                "openstack_peepo_exporter_instances_per_hypervisor",
+                "Number of instances per hypervisor",
+                labels=["hypervisor_name", "hypervisor_id"],
+            )
+
+            new_metrics = asyncio.run(
+                instances_per_hypervisor.export_metrics(self.cloud_name)
+            )
             logger.info(f"Updated metrics with {len(new_metrics)} hypervisors")
 
+            # Update the gauge with new values
+            for metric in new_metrics:
+                self.gauge.add_metric(
+                    [metric["hypervisor_name"], metric["hypervisor_id"]],
+                    metric["instance_count"],
+                )
+
+            # Update last_update timestamp
+            self.last_update = time.time()
         except Exception as e:
-            logger.error(f"Error updating metrics: {str(e)}")
-            raise
+            logger.error(f"Error updating metrics: {e}")
 
     def collect(self):
         if time.time() - self.last_update > self.cache_time:
             self._fetch_metrics()
 
-        gauge = GaugeMetricFamily(
-            "openstack_peepo_exporter_instances_per_hypervisor",
-            "Number of instances per hypervisor",
-            labels=["hypervisor_name", "hypervisor_id"],
-        )
-
-        for metric in self.metrics:
-            gauge.add_metric(
-                [metric["hypervisor_name"], metric["hypervisor_id"]],
-                metric["instance_count"],
-            )
-
-        yield gauge
+        yield self.gauge

src/main.py

@@ -31,11 +31,17 @@
         parser.add_argument(
             "--cloud", default=os.environ.get("OS_CLOUD"), help="The name of the cloud."
         )
+        parser.add_argument(
+            "--cache-time",
+            default=os.environ.get("CACHE_TIME", 60),
+            type=int,
+            help="The time to cache the metrics in seconds.",
+        )
         args = parser.parse_args()
 
         # Register the collector
         registry = CollectorRegistry()
-        registry.register(InstancesPerHypervisorCollector(args.cloud))
+        registry.register(InstancesPerHypervisorCollector(args.cloud, args.cache_time))
 
         # Start the metrics server
         start_http_server(args.port, args.addr, registry)

src/util/instances_per_hypervisor.py

@@ -1,28 +1,49 @@
 import openstack
+import httpx
+import asyncio
+import logging
+import time
 
+logger = logging.getLogger(__name__)
 
-def export_metrics(cloud_name: str):
+
+async def fetch_allocations(client, rp_id):
+    response = await client.get(f"/resource_providers/{rp_id}/allocations")
+    return rp_id, response.json()
+
+
+async def export_metrics(cloud_name: str):
     print(f"exporting metrics for {cloud_name}")
     conn = openstack.connect(cloud=cloud_name)
+    headers = conn.session.get_auth_headers()
+
+    placement_endpoint = conn.session.get_endpoint(
+        service_type="placement", interface="public"
+    )
 
     instance_counts = []
+    async with httpx.AsyncClient(
+        base_url=placement_endpoint, headers=headers
+    ) as client:
+        rps = list(conn.placement.resource_providers())
+        tasks = [fetch_allocations(client, rp.id) for rp in rps]
+        start_time = time.time()
+        results = await asyncio.gather(*tasks)
+        end_time = time.time()
+        logger.info(
+            f"Time taken to fetch {len(rps)} allocations: {end_time - start_time} seconds"
+        )
+
+        for rp in rps:
+            allocations = next((r[1] for r in results if r[0] == rp.id), {})
+            instance_count = len(allocations.get("allocations", {}))
 
-    for rp in conn.placement.resource_providers():
-        print(f"rp: {rp}")
-        hypervisor_id = rp.id
-        hypervisor_name = rp.name
-
-        allocations = conn.placement.get(
-            f"/resource_providers/{rp.id}/allocations"
-        ).json()
-        instance_count = len(allocations.get("allocations", {}))
-        data = {
-            "hypervisor_id": hypervisor_id,
-            "hypervisor_name": hypervisor_name,
-            "instance_count": instance_count,
-        }
-        instance_counts.append(data)
-
-        print(f"Hypervisor: {hypervisor_name}, Instances: {instance_count}")
+            data = {
+                "hypervisor_id": rp.id,
+                "hypervisor_name": rp.name,
+                "instance_count": instance_count,
+            }
+            instance_counts.append(data)
+            logger.debug(f"Hypervisor: {rp.name}, Instances: {instance_count}")
 
     return instance_counts