Simplify code for main and strategies

README.md

@@ -219,8 +219,10 @@ Most helpful flags:
 - `--cpu-min` Sets the minimum recommended cpu value in millicores
 - `--mem-min` Sets the minimum recommended memory value in MB
 - `--history_duration` The duration of the Prometheus history data to use (in hours)
+- `--cpu-percentile` How to calculate the CPU recommendation (90 means 90th percentile of historical CPU)
+- `--memory-buffer-percentage` How to calculate the memory recommendation ("5" means add a 5% buffer to max historical memory)
 
-More specific information on Strategy Settings can be found using
+More information on strategy settings can be found using
 
 ```sh
 krr simple --help

krr.py

@@ -1,4 +1,4 @@
-from robusta_krr import run
+from robusta_krr import app
 
 if __name__ == "__main__":
-    run()
+    app()

robusta_krr/__init__.py

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

robusta_krr/core/abstract/strategies.py

@@ -69,20 +69,13 @@ def timeframe_timedelta(self) -> datetime.timedelta:
 
 RunResult = dict[ResourceType, ResourceRecommendation]
 
-SelfBS = TypeVar("SelfBS", bound="BaseStrategy")
-_StrategySettings = TypeVar("_StrategySettings", bound=StrategySettings)
-
 
 # An abstract base class for strategy implementation.
 # This class requires implementation of a 'run' method for calculating recommendation.
 # Make a subclass if you want to create a concrete strategy.
-class BaseStrategy(abc.ABC, Generic[_StrategySettings]):
+class BaseStrategy(abc.ABC):
     """An abstract base class for strategy implementation.
 
-    This class is generic, and requires a type for the settings.
-    This settings type will be used for the settings property of the strategy.
-    It will be used to generate CLI parameters for this strategy, validated by pydantic.
-
     This class requires implementation of a 'run' method for calculating recommendation.
     Additionally, it provides a 'description' property for generating a description for the strategy.
     Description property uses the docstring of the strategy class and the settings of the strategy.
@@ -102,62 +95,29 @@ class BaseStrategy(abc.ABC, Generic[_StrategySettings]):
     def metrics(self) -> Sequence[type[PrometheusMetric]]:
         pass
 
-    def __init__(self, settings: _StrategySettings):
+    def __init__(self, settings: StrategySettings):
         self.settings = settings
 
     def __str__(self) -> str:
-        return self._display_name.title()
-
-    @property
-    def _display_name(self) -> str:
-        return getattr(self, "display_name", self.__class__.__name__.lower().removeprefix("strategy"))
+        return self.display_name.title()
 
     @property
+    @abc.abstractmethod
     def description(self) -> Optional[str]:
         """
         Generate a description for the strategy.
-        You can use the settings in the description by using the format syntax.
-        Also you can use Rich's markdown syntax to format the description.
+        You can use Rich's markdown syntax to format the description.
         """
-
-        if self.__doc__ is None:
-            return None
-
-        return f"[b]{self} Strategy[/b]\n\n" + dedent(self.__doc__.format_map(self.settings.dict())).strip()
+        pass
 
     # 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:
         pass
 
-    # This method is intended to return a strategy by its name.
-    @classmethod
-    def find(cls: type[SelfBS], name: str) -> type[SelfBS]:
-        strategies = cls.get_all()
-        if name.lower() in strategies:
-            return strategies[name.lower()]
-
-        raise ValueError(f"Unknown strategy name: {name}. Available strategies: {', '.join(strategies)}")
-
-    # This method is intended to return all the available strategies.
-    @classmethod
-    def get_all(cls: type[SelfBS]) -> dict[str, type[SelfBS]]:
-        from robusta_krr import strategies as _  # noqa: F401
-
-        return {sub_cls.display_name.lower(): sub_cls for sub_cls in cls.__subclasses__()}
-
-    # This method is intended to return the type of settings used in strategy.
-    @classmethod
-    def get_settings_type(cls) -> type[StrategySettings]:
-        return get_args(cls.__orig_bases__[0])[0]  # type: ignore
-
-
-AnyStrategy = BaseStrategy[StrategySettings]
-
 
 __all__ = [
-    "AnyStrategy",
     "BaseStrategy",
     "StrategySettings",
     "PodsTimeData",

robusta_krr/core/models/config.py

@@ -11,7 +11,7 @@
 from rich.logging import RichHandler
 
 from robusta_krr.core.abstract import formatters
-from robusta_krr.core.abstract.strategies import AnyStrategy, BaseStrategy
+from robusta_krr.core.abstract.strategies import BaseStrategy
 from robusta_krr.core.models.objects import KindLiteral
 
 logger = logging.getLogger("krr")
@@ -28,8 +28,8 @@ class Config(pd.BaseSettings):
     selector: Optional[str] = None
 
     # Value settings
-    cpu_min_value: int = pd.Field(10, ge=0)  # in millicores
-    memory_min_value: int = pd.Field(100, ge=0)  # in megabytes
+    cpu_min_value: int = pd.Field(10, ge=0, description="Sets the minimum recommended cpu value in millicores.")  # in millicores
+    memory_min_value: int = pd.Field(100, ge=0, description="Sets the minimum recommended memory value in MB.")  # in megabytes
 
     # Prometheus Settings
     prometheus_url: Optional[str] = pd.Field(None)
@@ -60,8 +60,6 @@ class Config(pd.BaseSettings):
     file_output: Optional[str] = pd.Field(None)
     slack_output: Optional[str] = pd.Field(None)
 
-    other_args: dict[str, Any]
-
     # Internal
     inside_cluster: bool = False
     _logging_console: Optional[Console] = pd.PrivateAttr(None)
@@ -94,20 +92,11 @@ def validate_resources(cls, v: Union[list[str], Literal["*"]]) -> Union[list[str
             return "*"
 
         return [val.capitalize() for val in v]
-
-    def create_strategy(self) -> AnyStrategy:
-        StrategyType = AnyStrategy.find(self.strategy)
-        StrategySettingsType = StrategyType.get_settings_type()
-        return StrategyType(StrategySettingsType(**self.other_args))  # type: ignore
-
-    @pd.validator("strategy")
-    def validate_strategy(cls, v: str) -> str:
-        BaseStrategy.find(v)  # NOTE: raises if strategy is not found
-        return v
+
 
     @pd.validator("format")
     def validate_format(cls, v: str) -> str:
-        formatters.find(v)  # NOTE: raises if strategy is not found
+        formatters.find(v)  # NOTE: raises if formatter is not found
         return v
 
     @property
@@ -150,7 +139,7 @@ def set_config(config: Config) -> None:
         logger.setLevel(logging.DEBUG if config.verbose else logging.CRITICAL if config.quiet else logging.INFO)
 
 
-# NOTE: This class is just a proxy for _config.
+# NOTE: This class is just a proxy for _config so you can access it globally without passing it around everywhere
 # Import settings from this module and use it like it is just a config object.
 class _Settings(Config):  # Config here is used for type checking
     def __init__(self) -> None:

robusta_krr/core/runner.py

@@ -10,7 +10,7 @@
 from prometrix import PrometheusNotFound
 from slack_sdk import WebClient
 
-from robusta_krr.core.abstract.strategies import ResourceRecommendation, RunResult
+from robusta_krr.core.abstract.strategies import ResourceRecommendation, RunResult, BaseStrategy, StrategySettings
 from robusta_krr.core.integrations.kubernetes import KubernetesLoader
 from robusta_krr.core.integrations.prometheus import ClusterNotSpecifiedException, PrometheusMetricsLoader
 from robusta_krr.core.models.config import settings
@@ -27,11 +27,11 @@
 class Runner:
     EXPECTED_EXCEPTIONS = (KeyboardInterrupt, PrometheusNotFound)
 
-    def __init__(self) -> None:
+    def __init__(self, strategy) -> None:
         self._k8s_loader = KubernetesLoader()
         self._metrics_service_loaders: dict[Optional[str], Union[PrometheusMetricsLoader, Exception]] = {}
         self._metrics_service_loaders_error_logged: set[Exception] = set()
-        self._strategy = settings.create_strategy()
+        self._strategy = strategy
 
         # This executor will be running calculations for recommendations
         self._executor = ThreadPoolExecutor(settings.max_workers)