fix: Prometheus client missing time argument

slo_generator/api/main.py

@@ -211,7 +211,7 @@ def process_batch_req(request, data, config):
     for url in urls:
         if "pubsub_batch_handler" in config:
             LOGGER.info(f"Sending {url} to pubsub batch handler.")
-            from google.cloud import pubsub_v1
+            from google.cloud import pubsub_v1  # noqa: PLC0415
 
             # pytype: disable=attribute-error
 

slo_generator/backends/cloud_monitoring_mql.py

@@ -15,6 +15,7 @@
 `cloud_monitoring_mql.py`
 Cloud Monitoring backend implementation with MQL (Monitoring Query Language).
 """
+# ruff: noqa: UP045
 
 import logging
 import pprint

slo_generator/backends/cloud_service_monitoring.py

@@ -15,6 +15,7 @@
 `cloud_service_monitoring.py`
 Cloud Service Monitoring exporter class.
 """
+# ruff: noqa: UP045,UP007
 
 import difflib
 import json

slo_generator/backends/prometheus.py

@@ -15,6 +15,7 @@
 `prometheus.py`
 Prometheus backend implementation.
 """
+# ruff: noqa: UP045,UP007
 
 import json
 import logging
@@ -117,6 +118,7 @@ def distribution_cut(
         res_good = self.query(
             expr,
             window,
+            timestamp=timestamp,
             operators=["increase", "sum"],
             labels=labels,
         )
@@ -130,6 +132,7 @@ def distribution_cut(
         res_valid = self.query(
             expr_count,
             window,
+            timestamp=timestamp,
             operators=["increase", "sum"],
         )
         valid_count = PrometheusBackend.count(res_valid)
@@ -163,7 +166,7 @@ def query(  # noqa: PLR0913
             labels = {}
         filter = PrometheusBackend._fmt_query(filter, window, operators, labels)
         LOGGER.debug(f"Query: {filter}")
-        response = self.client.query(metric=filter)
+        response = self.client.query(metric=filter, time=timestamp)
         response = json.loads(response)
         LOGGER.debug(pprint.pformat(response))
         return response

slo_generator/cli.py

@@ -166,7 +166,7 @@ def compute(slo_config, config, export, delete, timestamp):
 def api(ctx, config, exporters, signature_type, target, port):  # noqa: PLR0913
     """Run an API that can receive requests (supports both 'http' and
     'cloudevents' signature types)."""
-    from functions_framework._cli import _cli
+    from functions_framework._cli import _cli  # noqa: PLC0415
 
     os.environ["EXPORTERS"] = exporters
     os.environ["CONFIG_PATH"] = config

slo_generator/compute.py

@@ -15,6 +15,7 @@
 `compute.py`
 Compute utilities.
 """
+# ruff: noqa: UP045
 
 import logging
 import pprint

slo_generator/utils.py

@@ -15,6 +15,7 @@
 `utils.py`
 Utility functions.
 """
+# ruff: noqa: UP045
 
 import argparse
 import errno
@@ -188,7 +189,7 @@ def setup_logging():
 
     # Ignore Cloud SDK warning when using a user instead of service account
     try:
-        from google.auth._default import _CLOUD_SDK_CREDENTIALS_WARNING
+        from google.auth._default import _CLOUD_SDK_CREDENTIALS_WARNING  # noqa: PLC0415
 
         warnings.filterwarnings("ignore", message=_CLOUD_SDK_CREDENTIALS_WARNING)
     except ImportError: