Prometheus workload loader

README.md

@@ -430,7 +430,7 @@ If your Prometheus monitors multiple clusters we require the label you defined f
 For example, if your cluster has the Prometheus label `cluster: "my-cluster-name"`, then run this command:
 
 ```sh
-krr.py simple --prometheus-label cluster -l my-cluster-name
+krr.py simple --prometheus-cluster-key cluster -l my-cluster-name
 ```
 
 You may also need the `-p` flag to explicitly give Prometheus' URL.

examples/custom_strategy.py

@@ -3,7 +3,7 @@
 import pydantic as pd
 
 import robusta_krr
-from robusta_krr.api.models import K8sObjectData, MetricsPodData, ResourceRecommendation, ResourceType, RunResult
+from robusta_krr.api.models import K8sWorkload, MetricsPodData, ResourceRecommendation, ResourceType, RunResult
 from robusta_krr.api.strategies import BaseStrategy, StrategySettings
 from robusta_krr.core.integrations.prometheus.metrics import MaxMemoryLoader, PercentileCPULoader
 
@@ -24,7 +24,7 @@ class CustomStrategy(BaseStrategy[CustomStrategySettings]):
     rich_console = True  # Whether to use rich console for the CLI
     metrics = [PercentileCPULoader(90), MaxMemoryLoader]  # The metrics to use for the strategy
 
-    def run(self, history_data: MetricsPodData, object_data: K8sObjectData) -> RunResult:
+    def run(self, history_data: MetricsPodData, object_data: K8sWorkload) -> RunResult:
         return {
             ResourceType.CPU: ResourceRecommendation(request=self.settings.param_1, limit=None),
             ResourceType.Memory: ResourceRecommendation(request=self.settings.param_2, limit=self.settings.param_2),

robusta_krr/api/models.py

@@ -1,14 +1,14 @@
 from robusta_krr.core.abstract.strategies import MetricsPodData, PodsTimeData, ResourceRecommendation, RunResult
 from robusta_krr.core.models.allocations import RecommendationValue, ResourceAllocations, ResourceType
-from robusta_krr.core.models.objects import K8sObjectData, PodData
+from robusta_krr.core.models.objects import K8sWorkload, PodData
 from robusta_krr.core.models.result import ResourceScan, Result
 from robusta_krr.core.models.severity import Severity, register_severity_calculator
 
 __all__ = [
     "ResourceType",
     "ResourceAllocations",
     "RecommendationValue",
-    "K8sObjectData",
+    "K8sWorkload",
     "PodData",
     "Result",
     "Severity",

robusta_krr/core/abstract/cluster_loader.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import abc
+import logging
+from typing import Optional, TYPE_CHECKING
+
+from .workload_loader import BaseWorkloadLoader
+
+if TYPE_CHECKING:
+    from robusta_krr.core.integrations.prometheus.connector import PrometheusConnector
+
+
+logger = logging.getLogger("krr")
+
+
+class BaseClusterLoader(abc.ABC):
+    """
+    A class that wraps loading data from multiple clusters.
+    For example, a centralized prometheus server that can query multiple clusters.
+    Or one kubeconfig can define connections to multiple clusters.
+    """
+
+    @abc.abstractmethod
+    async def list_clusters(self) -> Optional[list[str]]:
+        pass
+
+    @abc.abstractmethod
+    def get_workload_loader(self, cluster: Optional[str]) -> BaseWorkloadLoader:
+        pass
+
+    def try_get_workload_loader(self, cluster: Optional[str]) -> Optional[BaseWorkloadLoader]:
+        try:
+            return self.get_workload_loader(cluster)
+        except Exception as e:
+            logger.error(f"Could not connect to cluster {cluster} and will skip it: {e}")
+            return None
+
+    @abc.abstractmethod
+    def get_prometheus(self, cluster: Optional[str]) -> PrometheusConnector:
+        """
+        Connect to a Prometheus server and return a PrometheusConnector instance.
+        Cluster = None means that prometheus is the only one: either centralized or in-cluster.
+        raise prometrix.PrometheusNotFound if Prometheus is not available.
+        """
+
+        pass

robusta_krr/core/abstract/metrics.py

@@ -2,7 +2,7 @@
 from abc import ABC, abstractmethod
 
 from robusta_krr.core.abstract.strategies import PodsTimeData
-from robusta_krr.core.models.objects import K8sObjectData
+from robusta_krr.core.models.objects import K8sWorkload
 
 
 class BaseMetric(ABC):
@@ -16,6 +16,6 @@ class BaseMetric(ABC):
 
     @abstractmethod
     async def load_data(
-        self, object: K8sObjectData, period: datetime.timedelta, step: datetime.timedelta
+        self, object: K8sWorkload, period: datetime.timedelta, step: datetime.timedelta
     ) -> PodsTimeData:
         ...

robusta_krr/core/abstract/strategies.py

@@ -9,15 +9,15 @@
 import pydantic as pd
 from numpy.typing import NDArray
 
-from robusta_krr.core.models.result import K8sObjectData, ResourceType
+from robusta_krr.core.models.result import K8sWorkload, ResourceType
 
 if TYPE_CHECKING:
-    from robusta_krr.core.abstract.metrics import BaseMetric  # noqa: F401
     from robusta_krr.core.integrations.prometheus.metrics import PrometheusMetric
 
 SelfRR = TypeVar("SelfRR", bound="ResourceRecommendation")
 
 
+# TODO: rename so it isn't the same name as ResourceRecommendation in result.py
 class ResourceRecommendation(pd.BaseModel):
     """A class to represent resource recommendation with optional request and limit values.
 
@@ -133,7 +133,7 @@ def description(self) -> Optional[str]:
     # Abstract method that needs to be implemented by subclass.
     # This method is intended to calculate resource recommendation based on history data and kubernetes object data.
     @abc.abstractmethod
-    def run(self, history_data: MetricsPodData, object_data: K8sObjectData) -> RunResult:
+    def run(self, history_data: MetricsPodData, object_data: K8sWorkload) -> RunResult:
         pass
 
     # This method is intended to return a strategy by its name.
@@ -167,6 +167,6 @@ def get_settings_type(cls) -> type[StrategySettings]:
     "StrategySettings",
     "PodsTimeData",
     "MetricsPodData",
-    "K8sObjectData",
+    "K8sWorkload",
     "ResourceType",
 ]

robusta_krr/core/abstract/workload_loader.py

@@ -0,0 +1,23 @@
+import abc
+import logging
+
+from robusta_krr.core.models.objects import K8sWorkload, PodData
+
+
+logger = logging.getLogger("krr")
+
+
+class BaseWorkloadLoader(abc.ABC):
+    """A base class for single cluster workload loaders."""
+
+    @abc.abstractmethod
+    async def list_workloads(self) -> list[K8sWorkload]:
+        pass
+
+
+class IListPodsFallback(abc.ABC):
+    """This is an interface that a workload loader can implement to have a fallback method to list pods."""
+
+    @abc.abstractmethod
+    async def load_pods(self, object: K8sWorkload) -> list[PodData]:
+        pass