Merge branch 'main' into fix-linting-errors

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "robusta-krr"
-version = "1.3.0-dev"
+version = "1.3.2-dev"
 description = "Robusta's Resource Recommendation engine for Kubernetes"
 authors = ["Pavel Zhukov <[email protected]>"]
 license = "MIT"

robusta_krr/__init__.py

@@ -1,4 +1,4 @@
 from .main import run
 
-__version__ = "1.3.0-dev"
+__version__ = "1.3.2-dev"
 __all__ = ["run", "__version__"]

robusta_krr/core/integrations/prometheus/__init__.py

@@ -1,7 +1,3 @@
 from .loader import MetricsLoader
-from .metrics_service.prometheus_metrics_service import (
-    ClusterNotSpecifiedException,
-    CustomPrometheusConnect,
-    PrometheusDiscovery,
-    PrometheusNotFound,
-)
+from .metrics_service.prometheus_metrics_service import PrometheusDiscovery, PrometheusNotFound
+from .prometheus_client import CustomPrometheusConnect, ClusterNotSpecifiedException

robusta_krr/core/integrations/prometheus/metrics/memory_metric.py

@@ -37,13 +37,13 @@ def get_query(self, object: K8sObjectData, resolution: Optional[str]) -> str:
         cluster_label = self.get_prometheus_cluster_label()
         resolution_formatted = f"[{resolution}]" if resolution else ""
         return (
-            f"max(max_over_time(container_memory_working_set_bytes{{"
+            f"max_over_time(container_memory_working_set_bytes{{"
             f'namespace="{object.namespace}", '
             f'pod=~"{pods_selector}", '
             f'container="{object.container}"'
             f"{cluster_label}}}"
             f"{resolution_formatted}"
-            f")) by (container, pod, job, id)"
+            f")"
         )
 
     def get_query_type(self) -> QueryType:

robusta_krr/core/integrations/prometheus/metrics_service/prometheus_metrics_service.py

@@ -1,12 +1,10 @@
 import asyncio
 import datetime
+from typing import List, Optional, Type
 from concurrent.futures import ThreadPoolExecutor
-from typing import Optional, Type, no_type_check
 
-import requests
 from kubernetes.client import ApiClient
-from prometheus_api_client import PrometheusConnect, Retry
-from requests.adapters import HTTPAdapter
+from prometheus_api_client import PrometheusApiClientException
 from requests.exceptions import ConnectionError, HTTPError
 
 from robusta_krr.core.abstract.strategies import ResourceHistoryData
@@ -16,6 +14,7 @@
 from robusta_krr.utils.service_discovery import MetricsServiceDiscovery
 
 from ..metrics import BaseMetricLoader
+from ..prometheus_client import ClusterNotSpecifiedException, CustomPrometheusConnect
 from .base_metric_service import MetricsNotFound, MetricsService
 
 
@@ -50,33 +49,6 @@ class PrometheusNotFound(MetricsNotFound):
     pass
 
 
-class ClusterNotSpecifiedException(Exception):
-    """
-    An exception raised when a prometheus requires a cluster label but an invalid one is provided.
-    """
-
-    pass
-
-
-class CustomPrometheusConnect(PrometheusConnect):
-    """
-    Custom PrometheusConnect class to handle retries.
-    """
-
-    @no_type_check
-    def __init__(
-        self,
-        url: str = "http://127.0.0.1:9090",
-        headers: dict = None,
-        disable_ssl: bool = False,
-        retry: Retry = None,
-        auth: tuple = None,
-    ):
-        super().__init__(url, headers, disable_ssl, retry, auth)
-        self._session = requests.Session()
-        self._session.mount(self.url, HTTPAdapter(max_retries=retry, pool_maxsize=10, pool_block=True))
-
-
 class PrometheusMetricsService(MetricsService):
     """
     A class for fetching metrics from Prometheus.
@@ -161,10 +133,12 @@ def validate_cluster_name(self):
                 f"Label {cluster_label} does not exist, Rerun krr with the flag `-l <cluster>` where <cluster> is one of {cluster_names}"
             )
 
-    # Superclass method returns Optional[list[str]], but here we return list[str]
-    # NOTE that this does not break Liskov Substitution Principle
-    def get_cluster_names(self) -> list[str]:
-        return self.prometheus.get_label_values(label_name=self.config.prometheus_label)
+    def get_cluster_names(self) -> Optional[List[str]]:
+        try:
+            return self.prometheus.get_label_values(label_name=self.config.prometheus_label)
+        except PrometheusApiClientException:
+            self.error("Labels api not present on prometheus client")
+            return []
 
     async def gather_data(
         self,

robusta_krr/core/integrations/prometheus/prometheus_client.py

@@ -0,0 +1,88 @@
+from typing import no_type_check
+
+import requests
+from datetime import datetime
+from prometheus_api_client import PrometheusConnect, Retry, PrometheusApiClientException
+from requests.adapters import HTTPAdapter
+
+
+class ClusterNotSpecifiedException(Exception):
+    """
+    An exception raised when a prometheus requires a cluster label but an invalid one is provided.
+    """
+
+    pass
+
+
+class CustomPrometheusConnect(PrometheusConnect):
+    """
+    Custom PrometheusConnect class to handle retries.
+    """
+
+    def __init__(
+        self,
+        url: str = "http://127.0.0.1:9090",
+        headers: dict = None,
+        disable_ssl: bool = False,
+        retry: Retry = None,
+        auth: tuple = None,
+    ):
+        super().__init__(url, headers, disable_ssl, retry, auth)
+        self._session = requests.Session()
+        self._session.mount(self.url, HTTPAdapter(max_retries=retry, pool_maxsize=10, pool_block=True))
+
+    def custom_query(self, query: str, params: dict = None):
+        params = params or {}
+        data = None
+        query = str(query)
+        # using the query API to get raw data
+        response = self._session.post(
+            "{0}/api/v1/query".format(self.url),
+            data={"query": query, **params},
+            verify=self.ssl_verification,
+            headers=self.headers,
+            auth=self.auth,
+        )
+        if response.status_code == 200:
+            data = response.json()["data"]["result"]
+        else:
+            raise PrometheusApiClientException(
+                "HTTP Status Code {} ({!r})".format(response.status_code, response.content)
+            )
+
+        return data
+
+    def custom_query_range(
+        self,
+        query: str,
+        start_time: datetime,
+        end_time: datetime,
+        step: str,
+        params: dict = None,
+    ):
+        start = round(start_time.timestamp())
+        end = round(end_time.timestamp())
+        params = params or {}
+        data = None
+        query = str(query)
+        # using the query_range API to get raw data
+        response = self._session.post(
+            "{0}/api/v1/query_range".format(self.url),
+            data={
+                "query": query,
+                "start": start,
+                "end": end,
+                "step": step,
+                **params,
+            },
+            verify=self.ssl_verification,
+            headers=self.headers,
+            auth=self.auth,
+        )
+        if response.status_code == 200:
+            data = response.json()["data"]["result"]
+        else:
+            raise PrometheusApiClientException(
+                "HTTP Status Code {} ({!r})".format(response.status_code, response.content)
+            )
+        return data