Merge pull request #458 from powerapi-ng/feat/report-processor-tags

feat(report): Allow metadata to be sanitized and flattened

.flake8

@@ -1,4 +1,4 @@
 [flake8]
 max-line-length = 160
-ignore = W504, F401, F811
+ignore = W504, F401, F811, E501
 exclude = powerapi/test_utils

src/powerapi/report/power_report.py

@@ -151,45 +151,44 @@ def to_virtiofs_db(report: PowerReport) -> Tuple[str, str]:
         power = report.power
         return filename, power
 
-    def gen_tag(self, metadata_kept):
+    def generate_tags(self, selected_tags: None | list[str] = None) -> dict[str, Any]:
         """
-        Generate the tags list of the report.
-        :param metadata_kept: The metadata to keep
+        Generate the report tags from its metadata.
+        :param selected_tags: List of tags to be included (in flattened/sanitized form), None to include everything
+        :return: a single level dictionary containing the tags of the report
         """
-        # Always sensor and target are kept
-        tags = {'sensor': self.sensor, 'target': self.target}
+        flattened_tags = self.flatten_tags(self.metadata)
+        sanitized_tags_name = self.sanitize_tags_name(flattened_tags)
+        sanitized_tags = {sanitized_tags_name[k]: v for k, v in flattened_tags.items()}
 
-        if metadata_kept:
-            for metadata_name in metadata_kept:
-                if metadata_name not in self.metadata:
-                    raise BadInputData(f'No tag "{metadata_name}" found in power report', self)
-                tags[metadata_name] = self.metadata[metadata_name]
+        if selected_tags:
+            tags = {k: v for k, v in sanitized_tags.items() if k in selected_tags}
         else:
-            tags.update(self.metadata)
+            tags = sanitized_tags
 
-        return tags
+        return {'sensor': self.sensor, 'target': self.target} | tags
 
     @staticmethod
-    def to_influxdb(report: PowerReport, tags: List[str]) -> Dict:
+    def to_influxdb(report: PowerReport, tags: None | list[str]) -> dict[str, Any]:
         """
         :return: a dictionary, that can be stored into an influxdb, from a given PowerReport
         """
         return {
             'measurement': 'power_consumption',
-            'tags': report.gen_tag(tags),
+            'tags': report.generate_tags(tags),
             'time': str(report.timestamp),
             'fields': {
                 'power': report.power
             }
         }
 
     @staticmethod
-    def to_prometheus(report: PowerReport, tags: List[str]) -> Dict:
+    def to_prometheus(report: PowerReport, tags: None | list[str]) -> dict[str, Any]:
         """
         :return: a dictionary, that can be stored into a prometheus instance, from a given PowerReport
         """
         return {
-            'tags': report.gen_tag(tags),
+            'tags': report.generate_tags(tags),
             'time': int(report.timestamp.timestamp()),
             'value': report.power
         }

src/powerapi/report/report.py

@@ -29,8 +29,11 @@
 
 from __future__ import annotations
 
+from collections import Counter
 from datetime import datetime
-from typing import Dict, NewType, Tuple, List, Any
+from typing import Dict, NewType, Tuple, List, Any, Iterable
+from zlib import crc32
+
 from powerapi.exception import PowerAPIExceptionWithMessage, PowerAPIException
 from powerapi.message import Message
 
@@ -43,6 +46,8 @@
 CSV_HEADER_COMMON = [TIMESTAMP_KEY, SENSOR_KEY, TARGET_KEY]
 CsvLines = NewType('CsvLines', Tuple[List[str], Dict[str, str]])
 
+TAGS_NAME_TRANSLATION_TABLE = str.maketrans('.-/', '___')
+
 
 class BadInputData(PowerAPIExceptionWithMessage):
     """
@@ -133,3 +138,40 @@ def create_empty_report():
         Creates an empty report
         """
         return Report(None, None, None)
+
+    @staticmethod
+    def sanitize_tags_name(tags: Iterable[str]) -> dict[str, str]:
+        """
+        Generate a dict containing the tags name and theirs corresponding sanitized version.
+
+        The tags name are sanitized according to InfluxDB and Prometheus restrictions.
+        If a sanitized tag have conflicts (`tag-name` and `tag.name` -> `tag_name`) a hash of the input tag will be
+        appended at the end of the sanitized tag name. This allows to have stable tags name in the destination database.
+        :param tags: Iterable object containing the tags name
+        :return: Dictionary containing the input tag name as key and its sanitized version as value
+        """
+        sanitized_tags = {tag: tag.translate(TAGS_NAME_TRANSLATION_TABLE) for tag in tags}
+        conflict_count = Counter(sanitized_tags.values())
+        return {
+            tag_orig: (tag_new if conflict_count[tag_new] == 1 else f'{tag_new}_{crc32(tag_orig.encode()):x}')
+            for tag_orig, tag_new in sanitized_tags.items()
+        }
+
+    @staticmethod
+    def flatten_tags(tags: dict[str, Any], separator: str = '_') -> dict[str, Any]:
+        """
+        Flatten nested dictionaries within a tags dictionary.
+
+        This method takes a dictionary of tags, which may contain nested dictionaries as values, and flattens them into
+        a single-level dictionary. Each key in the flattened dictionary is constructed by concatenating the keys from
+        the nested dictionaries with their parent keys, separated by the specified separator.
+
+        This is particularly useful for databases that only support canonical (non-nested) types as values.
+        :param tags: Input tags dict
+        :param separator: Separator to use for the flattened tags name
+        :return: Flattened tags dict
+        """
+        return {
+            f"{pkey}{separator}{ckey}" if isinstance(pvalue, dict) else pkey: cvalue for pkey, pvalue in tags.items()
+            for ckey, cvalue in (pvalue.items() if isinstance(pvalue, dict) else {pkey: pvalue}.items())
+        }

tests/unit/report/conftest.py

@@ -26,47 +26,68 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+from datetime import datetime
+
 import pytest
 
 from powerapi.report import PowerReport
-from tests.utils.report.power import gen_json_power_report
 
 
 @pytest.fixture
 def power_report_without_metadata() -> PowerReport:
     """
     Generates a power_power
     """
-    json_input = gen_json_power_report(1)[0]
-    report = PowerReport.from_json(json_input)
-
-    return report
+    ts = datetime(2020, 1, 1, 0, 0, 0)
+    sensor = 'pytest'
+    target = 'test'
+    power = 42
+    metadata = {}
+    return PowerReport(ts, sensor, target, power, metadata)
 
 
 @pytest.fixture
 def power_report_with_metadata(power_report_without_metadata) -> PowerReport:
     """
-    Generates a power_power
+    Generates a power report with single-level metadata.
     """
-    power_report_without_metadata.metadata = {'k1': 'v1',
-                                              'k2': 'v2',
-                                              'k3': 333,
-                                              'k4': 'vv4'}
-
+    power_report_without_metadata.metadata = {
+        'scope': 'cpu',
+        'socket': 0,
+        'formula': '0000000000000000000000000000000000000000'
+    }
     return power_report_without_metadata
 
 
 @pytest.fixture
-def power_report_with_nested_metadata(power_report_without_metadata) -> PowerReport:
+def power_report_with_metadata_expected_tags(power_report_with_metadata) -> set[str]:
     """
-    Generates a power_power
+    Returns the expected tags for the power report with single-level metadata.
     """
-    power_report_without_metadata.metadata = {'k1': {'k1_k1': 1},
-                                              'k2': 'v2',
-                                              'k3': 333,
-                                              'k4': {'k4_k1': 'v1',
-                                                     'k4_k2': {'k4_k2_k1': 'v2'}
-                                                     }
-                                              }
+    return {'sensor', 'target', 'scope', 'socket', 'formula'}
 
+
+@pytest.fixture
+def power_report_with_nested_metadata(power_report_without_metadata) -> PowerReport:
+    """
+    Generates a power report with nested metadata.
+    """
+    power_report_without_metadata.metadata = {
+        'scope': 'cpu',
+        'socket': 0,
+        'formula': '0000000000000000000000000000000000000000',
+        'k8s': {
+            'app.kubernetes.io/name': 'test',
+            'app.kubernetes.io/managed-by': 'pytest',
+            'helm.sh/chart': 'powerapi-pytest-1.0.0'
+        }
+    }
     return power_report_without_metadata
+
+
+@pytest.fixture
+def power_report_with_nested_metadata_expected_tags(power_report_with_nested_metadata) -> set[str]:
+    """
+    Returns the expected tags for the power report with nested metadata.
+    """
+    return {'sensor', 'target', 'scope', 'socket', 'formula', 'k8s_app_kubernetes_io_name', 'k8s_app_kubernetes_io_managed_by', 'k8s_helm_sh_chart'}