Internal PR for #271 (#272)

Add option to specify data set under which metrics timeseries will be
created under when pushing metrics to cognite.

Co-authored-by: Ahmed AlSahaf <[email protected]>
Co-authored-by: Babatunde Aromire <[email protected]>

CHANGELOG.md

@@ -12,6 +12,13 @@ Changes are grouped as follows
 - `Fixed` for any bug fixes.
 - `Security` in case of vulnerabilities.
 
+
+## [6.1.0]
+
+### Added
+
+ * Added ability to specify dataset under which metrics timeseries are created
+
 ## [6.0.2]
 
 ### Fixed

cognite/extractorutils/__init__.py

@@ -16,5 +16,5 @@
 Cognite extractor utils is a Python package that simplifies the development of new extractors.
 """
 
-__version__ = "6.0.2"
+__version__ = "6.1.0"
 from .base import Extractor

cognite/extractorutils/configtools/elements.py

@@ -436,6 +436,7 @@ class _CogniteMetricsConfig:
     external_id_prefix: str
     asset_name: Optional[str]
     asset_external_id: Optional[str]
+    data_set: Optional[EitherIdConfig] = None
 
     push_interval: TimeIntervalConfig = TimeIntervalConfig("30s")
 
@@ -485,6 +486,7 @@ def start_pushers(self, cdf_client: CogniteClient, cancellation_token: Event = E
                 external_id_prefix=self.cognite.external_id_prefix,
                 push_interval=self.cognite.push_interval.seconds,
                 asset=asset,
+                data_set=self.cognite.data_set.either_id if self.cognite.data_set else None,
                 thread_name="CogniteMetricsPusher",  # There is only one Cognite project as a target
                 cancellation_token=cancellation_token,
             )

cognite/extractorutils/metrics.py

@@ -55,6 +55,7 @@ def __init__(self):
 from cognite.client import CogniteClient
 from cognite.client.data_classes import Asset, Datapoints, DatapointsArray, TimeSeries
 from cognite.client.exceptions import CogniteDuplicatedError
+from cognite.extractorutils.util import EitherId
 
 from .util import ensure_time_series
 
@@ -330,6 +331,7 @@ class CognitePusher(AbstractMetricsPusher):
         external_id_prefix: Unique external ID prefix for this pusher.
         push_interval: Seconds between each upload call
         asset: Optional contextualization.
+        data_set: Data set the metrics timeseries created under.
         thread_name: Name of thread to start. If omitted, a standard name such as Thread-4 will be generated.
         cancellation_token: Event object to be used as a thread cancelation event
     """
@@ -340,6 +342,7 @@ def __init__(
         external_id_prefix: str,
         push_interval: int,
         asset: Optional[Asset] = None,
+        data_set: Optional[EitherId] = None,
         thread_name: Optional[str] = None,
         cancellation_token: Event = Event(),
     ):
@@ -348,6 +351,7 @@ def __init__(
         self.cdf_client = cdf_client
         self.asset = asset
         self.external_id_prefix = external_id_prefix
+        self.data_set = data_set
 
         self._init_cdf()
 
@@ -372,6 +376,14 @@ def _init_cdf(self) -> None:
         else:
             asset_id = None
 
+        data_set_id = None
+        if self.data_set:
+            dataset = self.cdf_client.data_sets.retrieve(
+                id=self.data_set.internal_id, external_id=self.data_set.external_id
+            )
+            if dataset:
+                data_set_id = dataset.id
+
         for metric in REGISTRY.collect():
             if type(metric) == Metric and metric.type in ["gauge", "counter"]:
                 external_id = self.external_id_prefix + metric.name
@@ -383,6 +395,7 @@ def _init_cdf(self) -> None:
                         legacy_name=external_id,
                         description=metric.documentation,
                         asset_id=asset_id,  # type: ignore  # this is optional. Type hint in SDK is wrong
+                        data_set_id=data_set_id,  # type: ignore  # this is optional. Type hint in SDK is wrong
                     )
                 )
 

cognite/extractorutils/util.py

@@ -121,7 +121,7 @@ def __init__(self, **kwargs: Union[int, str, None]):
             raise TypeError("Internal IDs must be integers")
 
         if external_id is not None and not isinstance(external_id, str):
-            raise TypeError("Internal IDs must be integers")
+            raise TypeError("External IDs must be strings")
 
         self.internal_id: Optional[int] = internal_id
         self.external_id: Optional[str] = external_id

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "cognite-extractor-utils"
-version = "6.0.2"
+version = "6.1.0"
 description = "Utilities for easier development of extractors for CDF"
 authors = ["Mathias Lohne <[email protected]>"]
 license = "Apache-2.0"

schema/metrics_config.schema.json

@@ -29,7 +29,10 @@
                         "description": "PushGateway password"
                     },
                     "clear-after": {
-                        "type": ["null", "integer"],
+                        "type": [
+                            "null",
+                            "integer"
+                        ],
                         "description": "Clear metrics after this many seconds when the extractor stops. Default is disabled"
                     },
                     "push-interval": {
@@ -44,7 +47,9 @@
             "type": "object",
             "description": "Push metrics to CDF timeseries. Requires CDF credentials to be configured",
             "unevaluatedProperties": false,
-            "required": ["external-id-prefix"],
+            "required": [
+                "external-id-prefix"
+            ],
             "properties": {
                 "external-id-prefix": {
                     "type": "string",
@@ -62,6 +67,10 @@
                     "type": "integer",
                     "description": "Interval in seconds between each push to CDF",
                     "default": 30
+                },
+                "data-set": {
+                    "type": "string",
+                    "description": "Data set the metrics timeseries created under"
                 }
             }
         },