druid: don't put quotes arount timestamp if not __time (#501)

* druid: don't put quotes arount timestamp if not __time * add timestamp alias option * fix condition * fix condition
anodot · Jan 10, 2024 · f68f146 · f68f146
1 parent 389f83e
commit f68f146
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 4 deletions.
diff --git a/agent/src/agent/pipeline/config/stages/js_convert_metrics.py b/agent/src/agent/pipeline/config/stages/js_convert_metrics.py
@@ -39,7 +39,7 @@ class JSConvertMetrics30(JSConvertMetrics):
 
     def get_js_vars(self):
         return f"""
-state['TIMESTAMP_COLUMN'] = '{self.pipeline.timestamp_path}';
+state['TIMESTAMP_COLUMN'] = '{self.pipeline.timestamp_alias or self.pipeline.timestamp_path}';
 state['DIMENSIONS'] = {self.pipeline.dimension_paths_with_names};
 state['MEASUREMENTS'] = {self.pipeline.value_paths_with_names};
 state['COUNT_RECORDS'] = {int(self.pipeline.count_records)};

diff --git a/agent/src/agent/pipeline/jdbc/query.py b/agent/src/agent/pipeline/jdbc/query.py
@@ -49,14 +49,16 @@ def _get_timestamp_condition(self) -> str:
 
     def _supports_indexed_timestamp_condition(self) -> bool:
         return self.pipeline.timestamp_type in [pipeline.TimestampType.DATETIME, pipeline.TimestampType.STRING] and \
-               self.pipeline.source.type in [source.TYPE_MSSQL, source.TYPE_IMPALA, source.TYPE_DRUID]
+            self.pipeline.source.type in [source.TYPE_MSSQL, source.TYPE_IMPALA, source.TYPE_DRUID]
 
     def _get_indexed_query(self) -> str:
         if self.pipeline.source.type == source.TYPE_DRUID:
-            return f'TIME_IN_INTERVAL("{self.pipeline.timestamp_path}", ' \
+            # if timestamp alias is set then we assume an aggregation function is used and we don't need quotes
+            ts = self.pipeline.timestamp_path if self.pipeline.timestamp_alias else f'"{self.pipeline.timestamp_path}"'
+            return f'TIME_IN_INTERVAL({ts}, ' \
                    f'\'{self.TIMESTAMP_VALUE_ISO}/PT{self.pipeline.interval}S\')'
         if self.pipeline.source.type == source.TYPE_MSSQL:
-            return f"{self.pipeline.timestamp_path} BETWEEN DATEADD(second, {self.TIMESTAMP_VALUE}, '1970-01-01') AND "\
+            return f"{self.pipeline.timestamp_path} BETWEEN DATEADD(second, {self.TIMESTAMP_VALUE}, '1970-01-01') AND " \
                    f"DATEADD(second, {self.TIMESTAMP_VALUE} + {self.pipeline.interval}, '1970-01-01')"
         if self.pipeline.source.type == source.TYPE_IMPALA:
             return f"{self.pipeline.timestamp_path} BETWEEN CAST(FROM_UNIXTIME({self.TIMESTAMP_VALUE}) as TIMESTAMP) AND " \

diff --git a/agent/src/agent/pipeline/pipeline.py b/agent/src/agent/pipeline/pipeline.py
@@ -212,6 +212,10 @@ def tag_configurations(self) -> Optional[dict]:
     def timestamp_path(self) -> str:
         return self._get_property_path(self.config['timestamp']['name'])
 
+    @property
+    def timestamp_alias(self) -> str:
+        return self._get_property_path(self.config['timestamp'].get('alias'))
+
     @property
     def timezone(self) -> str:
         return self.config.get('timezone', 'UTC')

diff --git a/test-datasets/test_victoria.jsonl b/test-datasets/test_victoria.jsonl
@@ -1,4 +1,5 @@
 {"metric":{"__name__":"log_messages_total","job":" victoriametrics ","instance":"victoriametrics:8428","app_version":"victoria-metrics-20200701-115333-tags-v1.37.4-0-g8da3f773a","level":"инфо","location":"VictoriaMetrics/app/vmstorage/main.go:86"},"values":[5,7,1,1,3,9,2,4,1,2],"timestamps":[1644113932648,1644113942648,1644113952648,1644113962648,1644113972648,1644113982648,1644113992648,1644114002648,1644114012648,1644114022648]}
+{"metric":{"__name__":"kube_pod_container_info","job":" victoriametrics ","instance":"victoriametrics:8428","app_version":"victoria-metrics-20200701-115333-tags-v1.37.4-0-g8da3f773a","level":"инфо","location":"VictoriaMetrics/app/vmstorage/main.go:86"},"values":[5,7,1,1,3,9,2,4,1,2],"timestamps":[1644113932648,1644113942648,1644113952648,1644113962648,1644113972648,1644113982648,1644113992648,1644114002648,1644114012648,1644114022648]}
 {"metric":{"__name__":"log_messages_total2","job":"victoriametrics","instance":"victoriametrics:8428","app_version":"victoria-metrics-20200701-115333-tags-v1.37.4-0-g8da3f773a","level":"info","location":"VictoriaMetrics/app/vmstorage/main.go:86"},"values":[2,4,6,8,10,12,14,16,18,20],"timestamps":[1644113942000,1644113952000,1644113962000,1644113972000,1644113982000,1644113992000,1644114002000,1644114012000,1644114022000,1644114032000]}
 {"metric":{"__name__":"diff_dims", "not_needed": "value", "job":"victoriametrics","location":"VictoriaMetrics/app/vmstorage/main.go:86","instance":"victoriametrics:8428"},"values":[1,3],"timestamps":[1644113932648,1644113942648]}
 {"metric":{"__name__":"diff_dims", "not_needed": "value", "job":"victoriametrics","location":"VictoriaMetrics/app/vmstorage/main.go:86"},"values":[2,4],"timestamps":[1644113942000,1644113952000]}