Added multi-site API metrics calls

e2clab/constants/default.py

@@ -74,6 +74,7 @@ WORKFLOW_VALIDATE_FILE = load_def("WORKFLOW_VALIDATE_FILE", "workflow-validate.o
 PROVENANCE_DATA = load_def("PROVENANCE_DATA", "provenance-data")
 MONITORING_DATA = load_def("MONITORING_DATA", "monitoring-data")
 MONITORING_IOT_DATA = load_def("MONITORING_IOT_DATA", "iotlab-data")
+MONITORING_KWOLLECT_DATA = load_def("MONITORING_KWOLLECT_DATA", "kwollect-data")
 # Workflow_env
 WORKFLOW_ENV_PREFIX = load_def("WORKFLOW_ENV_PREFIX", "env_")
 # Monitoring IOT

e2clab/managers/monitoring_kwollect.py

@@ -3,11 +3,14 @@ Kwollect monitoring manager
 """
 
 import csv
+from pathlib import Path
 from typing import Optional
 
 import pytz
 from enoslib.infra.enos_g5k.g5k_api_utils import get_api_client
+from grid5000 import Grid5000Error
 
+import e2clab.constants.default as default
 from e2clab.constants import WORKFLOW_TASKS, Environment, WorkflowTasks
 from e2clab.log import get_logger
 from e2clab.managers.manager import Manager
@@ -16,6 +19,8 @@ from e2clab.probe import TaskProbe
 METRICS = "metrics"
 STEP = "step"
 
+ALL = "all"
+
 API_TZ = pytz.timezone("Europe/Paris")
 
 
@@ -32,7 +37,7 @@ class MonitoringKwollectManager(Manager):
         "properties": {
             METRICS: {
                 "description": "Metrics to pull from job, '[all]' to pull all metrics",
-                "default": ["all"],
+                "default": [ALL],
                 "type": "array",
                 "items": {"type": "string"},
             },
@@ -66,7 +71,10 @@ class MonitoringKwollectManager(Manager):
                 f"Access kwollect metric dashboard for job {id}: {dash_addr}"
             )
 
-    def _backup(self, output_dir):
+    def _backup(self, output_dir: Path):
+        kwollect_output_dir = output_dir / default.MONITORING_KWOLLECT_DATA
+        kwollect_output_dir.mkdir(exist_ok=True)
+
         task_probe = TaskProbe.get_probe()
         records = task_probe.get_records()
 
@@ -88,7 +96,7 @@ class MonitoringKwollectManager(Manager):
                 "start_time": start_str,
                 "end_time": end_str,
             }
-            if "all" in metrics:
+            if metrics == ALL:
                 kwargs.pop("metrics")
             site = job.site
             nodes_str = self._filter_site_nodes(site)
@@ -97,9 +105,27 @@ class MonitoringKwollectManager(Manager):
             else:
                 oarjobid = job.uid
                 kwargs["job_id"] = oarjobid
+
+            self.logger.debug(
+                f"Pulling kwollect data from {site} API with kwargs: {kwargs}"
+            )
+
             # API call
-            metrics = self.g5k_client.sites[site].metrics.list(**kwargs)
-            self._dump_metrics(metrics, output_dir=output_dir)
+            try:
+                metrics_list = self.g5k_client.sites[site].metrics.list(**kwargs)
+            except Grid5000Error as e:
+                self.logger.error(f"Failed API call to {site} site")
+                self.logger.error(e)
+                continue
+            if len(metrics_list) > 0:
+                # TODO: add more comprehensive naming
+                self._dump_metrics(
+                    metrics_list=metrics_list,
+                    output_dir=kwollect_output_dir,
+                    filename=site,
+                )
+            else:
+                self.logger.info(f"No metrics data found for Grid'5000 '{site}' site")
 
     def _destroy(self):
         # Nothing to do
@@ -109,11 +135,14 @@ class MonitoringKwollectManager(Manager):
         """This manager only works for Grid5000"""
         return Environment.G5K
 
-    def _get_metrics_str(self):
-        m_list = self.config[METRICS]
-        return ",".join(m_list)
+    def _get_metrics_str(self) -> str:
+        m_list = self.config.get(METRICS, [ALL])
+        if ALL in m_list:
+            return ALL
+        else:
+            return ",".join(m_list)
 
-    def _filter_site_nodes(self, site: str) -> Optional[None]:
+    def _filter_site_nodes(self, site: str) -> Optional[str]:
         nodes = []
         for agent in self.agent:
             addr = agent.address
@@ -125,20 +154,23 @@ class MonitoringKwollectManager(Manager):
         else:
             return None
 
-    @classmethod
-    def get_metrics_str(cls, config):
-        return ",".join(config[METRICS])
-
     def _get_viz_address(self, site: str) -> str:
         """Returns address to dashboard"""
         addr = f"https://api.grid5000.fr/stable/sites/{site}/metrics/dashboard"
         return addr
 
     @staticmethod
-    def _dump_metrics(metrics, output_dir):
-        metrics_csv = list(map(lambda m: m.to_dict(), metrics))
+    def _dump_metrics(metrics_list, output_dir: Path, filename: str) -> None:
+        """Dump Kwollect API metrics to CSV
+
+        Args:
+            metrics (_type_): List of metrics records
+            output_dir (Path): Dir to output
+            filename (str): name of the file
+        """
+        metrics_csv = list(map(lambda m: m.to_dict(), metrics_list))
         keys = metrics_csv[0].keys()
-        out_file = output_dir / "test.csv"
+        out_file = output_dir / f"{filename}.csv"
         with open(out_file, "w") as out:
             dict_writer = csv.DictWriter(out, keys)
             dict_writer.writeheader()

e2clab/tests/unit/test_managers.py

@@ -383,6 +383,10 @@ class TestKwollectManager(TestE2cLab):
         excepted = "wattmetre_power_watt,prom_default_metrics,pdu_outlet_power_watt"
         self.assertEqual(metrics_str, excepted)
 
+        test_config = {METRICS: ["test", "testtest", "all"]}
+        manager = MonitoringKwollectManager(test_config)
+        self.assertEqual("all", manager._get_metrics_str())
+
     def test_filter_site_nodes(self):
 
         self.kwo_manager.agent = [