adambernier commented on code in PR #170:
URL: https://github.com/apache/otava/pull/170#discussion_r3840626964
##########
otava/importer.py:
##########
@@ -827,6 +830,90 @@ def fetch_all_metric_names(self, test_conf:
BigQueryTestConfig) -> List[str]:
return [m for m in test_conf.metrics.keys()]
+class InfluxDBImporter(Importer):
+ def __init__(self, influxdb: InfluxDB):
+ self.__influxdb = influxdb
+
+ @staticmethod
+ def __selected_metrics(
+ defined_metrics: Dict[str, InfluxDBMetric], selected_metrics:
Optional[List[str]]
+ ) -> Dict[str, InfluxDBMetric]:
+ if selected_metrics is not None:
+ return {name: defined_metrics[name] for name in selected_metrics}
+ return defined_metrics
+
+ def fetch_data(self, test_conf: TestConfig, selector: DataSelector =
DataSelector()) -> Series:
+ if not isinstance(test_conf, InfluxDBTestConfig):
+ raise ValueError("Expected InfluxDBTestConfig")
+
+ since_time = selector.since_time
+ until_time = selector.until_time
+ if since_time.timestamp() > until_time.timestamp():
+ raise DataImportError(
+ f"Invalid time range:
[{format_timestamp(int(since_time.timestamp()))}, "
+ f"{format_timestamp(int(until_time.timestamp()))}]"
+ )
+
+ metrics = self.__selected_metrics(test_conf.metrics, selector.metrics)
+ query = test_conf.query
+ if "%{BRANCH}" in query:
+ if not selector.branch:
+ raise DataImportError(
+ f"Test {test_conf.name} uses %{{BRANCH}} in query but
--branch was not specified"
+ )
+ branch_literal = "'" + selector.branch.replace("'", "''") + "'"
+ query = query.replace("%{BRANCH}", branch_literal)
+
+ try:
+ columns, rows = self.__influxdb.fetch_data(query,
test_conf.query_language)
+ except Exception as err:
+ raise DataImportError(f"Failed to import test {test_conf.name}:
{err}") from err
+
+ try:
+ time_index = columns.index(test_conf.time_column)
+ attr_indexes = [columns.index(column) for column in
test_conf.attributes]
+ metric_names = [metric.name for metric in metrics.values()]
+ metric_indexes = [columns.index(metric.column) for metric in
metrics.values()]
+ except ValueError as err:
+ raise DataImportError(f"Column not found {err.args[0]}")
+
+ time = []
+ data = {name: [] for name in metric_names}
+ attributes = {columns[index]: [] for index in attr_indexes}
+ for row in rows:
+ timestamp = row[time_index]
+ if timestamp < since_time or timestamp >= until_time:
Review Comment:
Fixed in 0e5b49f. Timezone-less Arrow timestamps are now interpreted as UTC
before range filtering and epoch conversion. The Arrow-backed test uses an
explicit timestamp(ns) column without a timezone and asserts both filtering and
UTC epoch values; timezone-aware rows remain covered by the query-language
tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
