update to use async

Signed-off-by: Zen <[email protected]>

Author: desultory

src/json_exporter/json_endpoint.py

@@ -49,13 +49,13 @@ def populate_metrics(self):
                                            **metric_args, **values,
                                            _log_init=False, logger=self.logger))
 
-    def get_data(self, label_filter={}):
+    async def get_data(self, label_filter={}):
         """
         Gets the data from the endpoint.
         Updates the JSON labels.
         Populates the metrics using the new labels and data.
         """
-        from requests import get, post
+        from aiohttp import ClientSession
         from json import loads
         from json.decoder import JSONDecodeError
         from time import time
@@ -71,20 +71,21 @@ def get_data(self, label_filter={}):
             kwargs['headers'] = self.headers
 
         start_time = time()
-        if self.post_data:
-            self.logger.debug("POST data: %s", self.post_data)
-            request = post(self.endpoint, json=self.post_data, **kwargs)
-        else:
-            request = get(self.endpoint, **kwargs)
+        async with ClientSession() as session:
+            if self.post_data:
+                async with session.post(self.endpoint, json=self.post_data, **kwargs) as request:
+                    self.logger.debug("POST data: %s", self.post_data)
+                    request_data = await request.text()
+            else:
+                async with session.get(self.endpoint, **kwargs) as request:
+                    request_data = await request.text()
         self._request_time = time() - start_time
-        self.logger.debug("Request time: %s", self._request_time)
+        self.logger.info("[%s] Request time: %s" % (self.endpoint, self._request_time))
 
-        if request.status_code != 200:
-            raise ValueError("[%s] Request failed: %s" % (request.status_code, request.text))
-        self.logger.debug("Got data: %s", request.text)
+        self.logger.debug("Got data: %s", request_data)
 
         try:
-            self.data = loads(request.text)
+            self.data = loads(request_data)
         except JSONDecodeError as error:
             raise ValueError("Failed to decode JSON: %s" % error)
 

src/json_exporter/json_exporter.py

@@ -1,4 +1,5 @@
 from prometheus_exporter import Exporter, cached_exporter
+from asyncio import TaskGroup
 
 
 @cached_exporter
@@ -27,11 +28,13 @@ def get_labels(self):
             labels |= endpoint.labels
         return labels
 
-    def get_metrics(self, label_filter={}):
+    async def get_metrics(self, label_filter={}):
         """ Get metrics list from each endpoint, add them together """
         metric_list = []
+        async with TaskGroup() as tg:
+            for endpoint in self.endpoints:
+                self.logger.debug("Creating data task for: %s", endpoint.name)
+                tg.create_task(endpoint.get_data(label_filter=label_filter))
         for endpoint in self.endpoints:
-            self.logger.debug("Getting metrics from endpoint: %s", endpoint.name)
-            endpoint.get_data(label_filter=label_filter)
-            metric_list.extend(endpoint.metrics)
-        return metric_list + super().get_metrics(label_filter=label_filter)
+            metric_list += endpoint.metrics
+        return metric_list + await super().get_metrics(label_filter=label_filter)