Merge pull request #306 from Sanketika-Obsrv/dataset-metrics-fix

fix: #OBS-I354 dataset level alert metric caching issue fix

Author: ravismula

api-service/src/configs/IngestionConfig.ts

@@ -84,12 +84,12 @@ export const rawIngestionSpecDefaults = {
     "flattenSpec": [
         {
             "type": "path",
-            "expr": "$.obsrv_meta.source.connector",
+            "expr": "$.obsrv_meta.source.entry_source",
             "name": "obsrv.meta.source.connector"
         },
         {
             "type": "path",
-            "expr": "$.obsrv_meta.source.connectorInstance",
+            "expr": "$.obsrv_meta.source.id",
             "name": "obsrv.meta.source.id"
         }
     ],

api-service/src/services/TableGenerator.ts

@@ -66,7 +66,7 @@ class BaseTableGenerator {
         }
         if (!_.isEmpty(transformations_config)) {
             const transformationFields = _.map(transformations_config, (tf) => ({
-                expr: "$." + `['${tf.field_key}']`,
+                expr: "$." + tf.field_key.split('.').map((fieldpart: string) => `['${fieldpart}']`).join('.'),
                 name: tf.field_key,
                 data_type: tf.transformation_function.datatype,
                 arrival_format: tf.transformation_function.datatype,

command-service/src/command/alert_manager_command.py

@@ -5,7 +5,7 @@
 from model.data_models import Action, ActionResponse, CommandPayload
 from service.db_service import DatabaseService
 from service.http_service import HttpService
-
+from copy import deepcopy
 
 class AlertManagerService(ICommand):
 
@@ -95,17 +95,18 @@ def get_dataset_source_config(self, dataset_id: str) -> str:
     def get_modified_metric(
         self, service: str, metric: dict, payload: CommandPayload
     ) -> dict:
+        metric_data = deepcopy(metric)
         if service == "flink":
             substring = f"{payload.dataset_id}"
             modified_substring = substring.replace("-", "_")
-            modified_metric = metric["metric"].replace("dataset_id", modified_substring)
-            metric["metric"] = modified_metric
-            return metric
+            modified_metric = metric_data["metric"].replace("dataset_id", modified_substring)
+            metric_data["metric"] = modified_metric
+            return metric_data
         else:
-            metric["metric"] = metric["metric"].replace(
+            metric_data["metric"] = metric_data["metric"].replace(
                 "dataset_id", payload.dataset_id
             )
-            return metric
+            return metric_data
 
     def create_alert_metric(
         self, payload: CommandPayload, service: str, metric: dict, dataset_name: str