Gerrrr commented on code in PR #170:
URL: https://github.com/apache/otava/pull/170#discussion_r3840550724
##########
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:
InfluxDB returns time as Arrow `Timestamp(Nanosecond, None)`. Converting a
real InfluxDB result to Python produces a timezone-naive datetime, while
Otava's time bounds are timezone-aware:
```python
from datetime import datetime
import pyarrow
from otava.data_selector import DataSelector
table = pyarrow.table(
{
"time": pyarrow.array(
[datetime(2024, 1, 2)],
type=pyarrow.timestamp("ns"),
)
}
)
influxdb_time = table.to_pylist()[0]["time"]
selector = DataSelector()
influxdb_time < selector.until_time
```
This throws:
```python
TypeError: can't compare offset-naive and offset-aware datetimes
```
This means the importer fails while filtering normal InfluxDB results.
We should normalize InfluxDB timestamps to UTC before comparing them and
calling .timestamp(). Please add a test with an Arrow timestamp[ns] column
without a timezone. The current test data explicitly uses `timezone.utc`, which
hides this problem.
##########
examples/influxdb/otava.yaml:
##########
@@ -0,0 +1,23 @@
+# InfluxDB 3 connection settings can also be supplied with INFLUXDB_HOST,
Review Comment:
Please make this a reproducible example. PostgreSQL and Graphite examples
include Docker Compose configuration, seeded data, and an Otava configuration.
This directory only contains otava.yaml, so there is no way to run the example
as documented.
##########
tests/influxdb_test.py:
##########
@@ -0,0 +1,168 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+
+import os
+from datetime import datetime, timezone
+from unittest.mock import Mock
+
+import pyarrow as pa
+import pytest
+
+from otava.config import load_config_from_file
+from otava.data_selector import DataSelector
+from otava.importer import DataImportError, InfluxDBImporter
+from otava.influxdb import InfluxDB, InfluxDBConfig
+from otava.main import create_otava_cli_parser
+from otava.test_config import (
+ InfluxDBMetric,
+ InfluxDBTestConfig,
+ TestConfigError,
+ create_test_config,
+)
+
+
+def selector():
+ result = DataSelector()
+ result.since_time = datetime(2024, 1, 1, tzinfo=timezone.utc)
+ result.until_time = datetime(2024, 1, 5, tzinfo=timezone.utc)
+ return result
+
+
+def test_influxdb_connection_config_precedence(tmp_path, monkeypatch):
+ config_file = tmp_path / "otava.yaml"
+ config_file.write_text(
+ "influxdb:\n host: yaml-host\n database: yaml-db\n token:
yaml-token\n"
+ )
+ monkeypatch.setenv("INFLUXDB_HOST", "env-host")
+ monkeypatch.setenv("INFLUXDB_DATABASE", "env-db")
+ monkeypatch.setenv("INFLUXDB_TOKEN", "env-token")
+
+ config = load_config_from_file(
+ str(config_file),
+ arg_overrides=["--influxdb-host", "cli-host", "--influxdb-token",
"cli-token"],
+ )
+ assert config.influxdb.host == "cli-host"
+ assert config.influxdb.database == "env-db"
+ assert config.influxdb.token == "cli-token"
+ assert os.environ["INFLUXDB_HOST"] == "env-host"
+
+
+def test_cli_help_includes_influxdb_options():
+ help_text = create_otava_cli_parser().format_help()
+ assert "InfluxDB Options:" in help_text
+ assert "--influxdb-host" in help_text
+ assert "--influxdb-database" in help_text
+ assert "--influxdb-token" in help_text
+
+
+def test_influxdb_test_config_defaults_to_sql_and_parses_metrics():
+ test = create_test_config(
+ "latency",
+ {
+ "type": "influxdb",
+ "query": "SELECT * FROM latency",
+ "attributes": ["branch"],
+ "metrics": {"p95": {"column": "p95_ms", "direction": -1, "scale":
0.001}},
+ },
+ )
+ assert isinstance(test, InfluxDBTestConfig)
+ assert test.query_language == "sql"
+ assert test.metrics["p95"] == InfluxDBMetric("p95", -1, 0.001, "p95_ms")
+
+
+def test_influxdb_test_config_supports_influxql_and_rejects_unknown_language():
+ test = create_test_config(
+ "latency",
+ {"type": "influxdb", "query": "SELECT * FROM latency", "metrics":
["p95_ms"], "query_language": "influxql"},
+ )
+ assert test.query_language == "influxql"
+ with pytest.raises(TestConfigError):
+ create_test_config(
+ "latency",
+ {"type": "influxdb", "query": "SELECT * FROM latency", "metrics":
["p95_ms"], "query_language": "flux"},
+ )
+
+
+def test_influxdb_importer_reads_arrow_table_and_applies_selection():
Review Comment:
We should have an e2e test with a Dockerized InfluxDB instance, similar to
`postgres_e2e_test.py` and `graphite_e2e_test.py`.
Mocking the returned Arrow table does not test the actual client/server
contract, authentication, query language, or returned schema. In particular, an
e2e test would have caught the timezone issue above.
--
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]
