Merge pull request #96 from robusta-dev/fix-linting-errors

Fix linting errors

pyproject.toml

@@ -17,7 +17,7 @@ multi_line_output = 3
 include_trailing_comma = true
 
 [tool.mypy]
-plugins = "numpy.typing.mypy_plugin"
+plugins = "numpy.typing.mypy_plugin,pydantic.mypy"
 
 [tool.poetry.scripts]
 krr = "robusta_krr.main:run"

robusta_krr/core/integrations/kubernetes.py

@@ -1,6 +1,6 @@
 import asyncio
-from concurrent.futures import ThreadPoolExecutor
 import itertools
+from concurrent.futures import ThreadPoolExecutor
 from typing import Optional, Union
 
 from kubernetes import client, config  # type: ignore
@@ -10,22 +10,21 @@
     V1DaemonSetList,
     V1Deployment,
     V1DeploymentList,
+    V1Job,
     V1JobList,
     V1LabelSelector,
-    V1PodList,
     V1Pod,
-    V1Job,
+    V1PodList,
     V1StatefulSet,
     V1StatefulSetList,
     V2HorizontalPodAutoscaler,
     V2HorizontalPodAutoscalerList,
 )
 
-from robusta_krr.core.models.objects import K8sObjectData, PodData, HPAData
+from robusta_krr.core.models.objects import HPAData, K8sObjectData, PodData
 from robusta_krr.core.models.result import ResourceAllocations
 from robusta_krr.utils.configurable import Configurable
 
-
 AnyKubernetesAPIObject = Union[V1Deployment, V1DaemonSet, V1StatefulSet, V1Pod, V1Job]
 
 

robusta_krr/core/integrations/prometheus/loader.py

@@ -1,8 +1,6 @@
-import asyncio
 import datetime
-from typing import Optional, no_type_check
-
 from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
 
 from kubernetes import config as k8s_config
 from kubernetes.client.api_client import ApiClient
@@ -49,10 +47,12 @@ def __init__(
             if cluster is not None
             else None
         )
-        self.loader = self.get_metrics_service(config, api_client=self.api_client, cluster=cluster)
-        if not self.loader:
+        loader = self.get_metrics_service(config, api_client=self.api_client, cluster=cluster)
+        if loader is None:
             raise PrometheusNotFound("No Prometheus or metrics service found")
 
+        self.loader = loader
+
         self.info(f"{self.loader.name()} connected successfully for {cluster or 'default'} cluster")
 
     def get_metrics_service(
@@ -68,9 +68,11 @@ def get_metrics_service(
                 self.echo(f"{service_name} found")
                 loader.validate_cluster_name()
                 return loader
-            except MetricsNotFound as e:
+            except MetricsNotFound:
                 self.debug(f"{service_name} not found")
 
+        return None
+
     async def gather_data(
         self,
         object: K8sObjectData,

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

@@ -1,4 +1,5 @@
 from typing import Any, Optional
+
 from robusta_krr.core.abstract.strategies import Metric
 
 from .base_metric import BaseMetricLoader, QueryType

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

@@ -2,11 +2,13 @@
 
 import abc
 import asyncio
-from concurrent.futures import ThreadPoolExecutor
 import datetime
-from typing import TYPE_CHECKING, Callable, Optional, TypeVar
 import enum
+from concurrent.futures import ThreadPoolExecutor
+from typing import TYPE_CHECKING, Callable, Optional, TypeVar
+
 import numpy as np
+
 from robusta_krr.core.abstract.strategies import Metric, ResourceHistoryData
 from robusta_krr.core.models.config import Config
 from robusta_krr.core.models.objects import K8sObjectData
@@ -15,7 +17,8 @@
 if TYPE_CHECKING:
     from .. import CustomPrometheusConnect
 
-    MetricsDictionary = dict[str, type[BaseMetricLoader]]
+
+MetricsDictionary = dict[str, type["BaseMetricLoader"]]
 
 
 class QueryType(str, enum.Enum):
@@ -72,6 +75,9 @@ def get_query(self, object: K8sObjectData, resolution: Optional[str]) -> str:
 
         pass
 
+    def get_query_type(self) -> QueryType:
+        return QueryType.QueryRange
+
     def get_graph_query(self, object: K8sObjectData, resolution: Optional[str]) -> str:
         """
         This method should be implemented by all subclasses to provide a query string in the metadata to produce relevant graphs.
@@ -158,7 +164,7 @@ async def load_data(
             step=self._step_to_string(step),
         )
         result = await self.query_prometheus(metric=metric, query_type=query_type)
-        # adding the query in the results for a graph 
+        # adding the query in the results for a graph
         metric.query = self.get_graph_query(object, resolution)
 
         if result == []:
@@ -173,7 +179,7 @@ async def load_data(
         )
 
     @staticmethod
-    def get_by_resource(resource: str, strategy: Optional[str]) -> type[BaseMetricLoader]:
+    def get_by_resource(resource: str, strategy: str) -> type[BaseMetricLoader]:
         """
         Fetches the metric loader corresponding to the specified resource.
 

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

@@ -1,9 +1,10 @@
 from typing import Optional
+
 from robusta_krr.core.models.allocations import ResourceType
 from robusta_krr.core.models.objects import K8sObjectData
 
 from .base_filtered_metric import BaseFilteredMetricLoader
-from .base_metric import bind_metric, QueryType
+from .base_metric import QueryType, bind_metric
 
 
 @bind_metric(ResourceType.CPU)

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

@@ -1,9 +1,10 @@
 from typing import Optional
+
 from robusta_krr.core.models.allocations import ResourceType
 from robusta_krr.core.models.objects import K8sObjectData
 
 from .base_filtered_metric import BaseFilteredMetricLoader
-from .base_metric import bind_metric, QueryType, override_metric
+from .base_metric import QueryType, bind_metric, override_metric
 
 
 @bind_metric(ResourceType.Memory)
@@ -23,9 +24,10 @@ def get_query(self, object: K8sObjectData, resolution: Optional[str]) -> str:
     def get_query_type(self) -> QueryType:
         return QueryType.QueryRange
 
+
 # This is a temporary solutions, metric loaders will be moved to strategy in the future
 @override_metric("simple", ResourceType.Memory)
-class MemoryMetricLoader(MemoryMetricLoader):
+class SimpleMemoryMetricLoader(MemoryMetricLoader):
     """
     A class that overrides the memory metric on the simple strategy.
     """
@@ -47,5 +49,5 @@ def get_query(self, object: K8sObjectData, resolution: Optional[str]) -> str:
     def get_query_type(self) -> QueryType:
         return QueryType.Query
 
-    def get_graph_query(self, object: K8sObjectData, resolution: Optional[str]) -> str: 
+    def get_graph_query(self, object: K8sObjectData, resolution: Optional[str]) -> str:
         return super().get_query(object, resolution)

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

@@ -1,6 +1,6 @@
 import abc
-from concurrent.futures import ThreadPoolExecutor
 import datetime
+from concurrent.futures import ThreadPoolExecutor
 from typing import List, Optional
 
 from kubernetes.client.api_client import ApiClient
@@ -42,7 +42,7 @@ def name(self) -> str:
         return classname.replace("MetricsService", "") if classname != MetricsService.__name__ else classname
 
     @abc.abstractmethod
-    async def get_cluster_names(self) -> Optional[List[str]]:
+    def get_cluster_names(self) -> Optional[List[str]]:
         ...
 
     @abc.abstractmethod
@@ -51,7 +51,6 @@ async def gather_data(
         object: K8sObjectData,
         resource: ResourceType,
         period: datetime.timedelta,
-        *,
         step: datetime.timedelta = datetime.timedelta(minutes=30),
     ) -> ResourceHistoryData:
         ...

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

@@ -1,7 +1,7 @@
 import asyncio
-from concurrent.futures import ThreadPoolExecutor
 import datetime
 from typing import List, Optional, Type
+from concurrent.futures import ThreadPoolExecutor
 
 from kubernetes.client import ApiClient
 from prometheus_api_client import PrometheusApiClientException
@@ -11,14 +11,14 @@
 from robusta_krr.core.models.config import Config
 from robusta_krr.core.models.objects import K8sObjectData, PodData
 from robusta_krr.core.models.result import ResourceType
-from robusta_krr.utils.service_discovery import ServiceDiscovery
+from robusta_krr.utils.service_discovery import MetricsServiceDiscovery
 
 from ..metrics import BaseMetricLoader
-from ..prometheus_client import CustomPrometheusConnect
+from ..prometheus_client import ClusterNotSpecifiedException, CustomPrometheusConnect
 from .base_metric_service import MetricsNotFound, MetricsService
 
 
-class PrometheusDiscovery(ServiceDiscovery):
+class PrometheusDiscovery(MetricsServiceDiscovery):
     def find_metrics_url(self, *, api_client: Optional[ApiClient] = None) -> Optional[str]:
         """
         Finds the Prometheus URL using selectors.
@@ -60,7 +60,7 @@ def __init__(
         *,
         cluster: Optional[str] = None,
         api_client: Optional[ApiClient] = None,
-        service_discovery: Type[ServiceDiscovery] = PrometheusDiscovery,
+        service_discovery: Type[MetricsServiceDiscovery] = PrometheusDiscovery,
         executor: Optional[ThreadPoolExecutor] = None,
     ) -> None:
         super().__init__(config=config, api_client=api_client, cluster=cluster, executor=executor)
@@ -87,7 +87,7 @@ def __init__(
 
         if self.auth_header:
             headers = {"Authorization": self.auth_header}
-        elif not self.config.inside_cluster:
+        elif not self.config.inside_cluster and self.api_client is not None:
             self.api_client.update_params_for_auth(headers, {}, ["BearerToken"])
 
         self.prometheus = CustomPrometheusConnect(url=self.url, disable_ssl=not self.ssl_enabled, headers=headers)

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

@@ -1,15 +1,15 @@
+from concurrent.futures import ThreadPoolExecutor
 from typing import Optional
 
 from kubernetes.client import ApiClient
-from requests.exceptions import ConnectionError, HTTPError
-from concurrent.futures import ThreadPoolExecutor
+
 from robusta_krr.core.models.config import Config
-from robusta_krr.utils.service_discovery import ServiceDiscovery
+from robusta_krr.utils.service_discovery import MetricsServiceDiscovery
 
 from .prometheus_metrics_service import MetricsNotFound, PrometheusMetricsService
 
 
-class ThanosMetricsDiscovery(ServiceDiscovery):
+class ThanosMetricsDiscovery(MetricsServiceDiscovery):
     def find_metrics_url(self, *, api_client: Optional[ApiClient] = None) -> Optional[str]:
         """
         Finds the Thanos URL using selectors.

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

@@ -1,15 +1,15 @@
-from typing import Optional
 from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
+
 from kubernetes.client import ApiClient
-from requests.exceptions import ConnectionError, HTTPError
 
 from robusta_krr.core.models.config import Config
-from robusta_krr.utils.service_discovery import ServiceDiscovery
+from robusta_krr.utils.service_discovery import MetricsServiceDiscovery
 
 from .prometheus_metrics_service import MetricsNotFound, PrometheusMetricsService
 
 
-class VictoriaMetricsDiscovery(ServiceDiscovery):
+class VictoriaMetricsDiscovery(MetricsServiceDiscovery):
     def find_metrics_url(self, *, api_client: Optional[ApiClient] = None) -> Optional[str]:
         """
         Finds the Victoria Metrics URL using selectors.

robusta_krr/core/runner.py

@@ -1,8 +1,7 @@
 import asyncio
 import math
-from typing import Optional, Union
-
 from concurrent.futures import ThreadPoolExecutor
+from typing import Optional, Union
 
 from robusta_krr.core.abstract.strategies import ResourceRecommendation, RunResult
 from robusta_krr.core.integrations.kubernetes import KubernetesLoader
@@ -65,7 +64,7 @@ def _process_result(self, result: Result) -> None:
         Formatter = self.config.Formatter
         formatted = result.format(Formatter)
         self.echo("\n", no_prefix=True)
-        self.print_result(formatted, rich=Formatter.__rich_console__)
+        self.print_result(formatted, rich=getattr(Formatter, "__rich_console__", False))
 
     def __get_resource_minimal(self, resource: ResourceType) -> float:
         if resource == ResourceType.CPU:
@@ -204,5 +203,5 @@ async def run(self) -> None:
             self._process_result(result)
         except ClusterNotSpecifiedException as e:
             self.error(e)
-        except Exception as e:
+        except Exception:
             self.console.print_exception(extra_lines=1, max_frames=10)

robusta_krr/formatters/table.py

@@ -1,5 +1,5 @@
 import itertools
-from typing import Any, Optional
+from typing import Any
 
 from rich.table import Table
 

robusta_krr/main.py

@@ -6,7 +6,6 @@
 from typing import List, Literal, Optional, Union
 from uuid import UUID
 
-
 import typer
 import urllib3
 

robusta_krr/utils/configurable.py

@@ -1,6 +1,6 @@
 import abc
 from inspect import getframeinfo, stack
-from typing import Literal
+from typing import Literal, Union
 
 from rich.console import Console
 
@@ -93,9 +93,9 @@ def warning(self, message: str = "") -> None:
 
         self.echo(message, type="WARNING")
 
-    def error(self, message: str = "") -> None:
+    def error(self, message: Union[str, Exception] = "") -> None:
         """
         Echoes an error message to the user
         """
 
-        self.echo(message, type="ERROR")
+        self.echo(str(message), type="ERROR")

robusta_krr/utils/service_discovery.py

@@ -1,3 +1,4 @@
+from abc import ABC, abstractmethod
 from typing import Optional
 
 from cachetools import TTLCache
@@ -82,3 +83,9 @@ def find_url(self, selectors: list[str]) -> Optional[str]:
                 return ingress_url
 
         return None
+
+
+class MetricsServiceDiscovery(ServiceDiscovery, ABC):
+    @abstractmethod
+    def find_metrics_url(self, *, api_client: Optional[ApiClient] = None) -> Optional[str]:
+        pass

tests/conftest.py

@@ -1,6 +1,6 @@
 import random
 from datetime import datetime, timedelta
-from unittest.mock import AsyncMock, PropertyMock, patch
+from unittest.mock import AsyncMock, patch
 
 import numpy as np
 import pytest