This tool optimizes the kubernetes workloads resource requsts and limits.
# This file is execfile()d with the current directory set to its containing dir.
#
# This file only contains a selection of the most common options. For a full
# list see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
#
# All configuration values have a default; values that are commented out
# serve to show the default.

import os
import sys
import shutil

# -- Path setup --------------------------------------------------------------

__location__ = os.path.dirname(__file__)

# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
sys.path.insert(0, os.path.join(__location__, "../src"))

# -- Run sphinx-apidoc -------------------------------------------------------
# This hack is necessary since RTD does not issue `sphinx-apidoc` before running
# `sphinx-build -b html . _build/html`. See Issue:
# https://github.com/readthedocs/readthedocs.org/issues/1139
# DON'T FORGET: Check the box "Install your project inside a virtualenv using
# setup.py install" in the RTD Advanced Settings.
# Additionally it helps us to avoid running apidoc manually

try:  # for Sphinx >= 1.7
    from sphinx.ext import apidoc
except ImportError:
    from sphinx import apidoc

output_dir = os.path.join(__location__, "api")
module_dir = os.path.join(__location__, "../src/k8soptimizer")
try:
    shutil.rmtree(output_dir)
except FileNotFoundError:
    pass

try:
    import sphinx

    cmd_line = f"sphinx-apidoc --implicit-namespaces -f -o {output_dir} {module_dir}"

    args = cmd_line.split(" ")
    if tuple(sphinx.__version__.split(".")) >= ("1", "7"):
        # This is a rudimentary parse_version to avoid external dependencies
        args = args[1:]

    apidoc.main(args)
except Exception as e:
    print("Running `sphinx-apidoc` failed!\n{}".format(e))

# -- General configuration ---------------------------------------------------

# If your documentation needs a minimal Sphinx version, state it here.
# needs_sphinx = '1.0'

# Add any Sphinx extension module names here, as strings. They can be extensions
# coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
extensions = [
    "sphinx.ext.autodoc",
    "sphinx.ext.intersphinx",
    "sphinx.ext.todo",
    "sphinx.ext.autosummary",
    "sphinx.ext.viewcode",
    "sphinx.ext.coverage",
    "sphinx.ext.doctest",
    "sphinx.ext.ifconfig",
    "sphinx.ext.mathjax",
    "sphinx.ext.napoleon",
]

# Add any paths that contain templates here, relative to this directory.
templates_path = ["_templates"]

# The suffix of source filenames.
source_suffix = ".rst"

# The encoding of source files.
# source_encoding = 'utf-8-sig'

# The master toctree document.
master_doc = "index"

# General information about the project.
project = "k8soptimizer"
copyright = "2023, Philipp Hellmich"

# The version info for the project you're documenting, acts as replacement for
# |version| and |release|, also used in various other places throughout the
# built documents.
#
# version: The short X.Y version.
# release: The full version, including alpha/beta/rc tags.
# If you don’t need the separation provided between version and release,
# just set them both to the same value.
try:
    from k8soptimizer import __version__ as version
except ImportError:
    version = ""

if not version or version.lower() == "unknown":
    version = os.getenv("READTHEDOCS_VERSION", "unknown")  # automatically set by RTD

release = version

# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
# language = None

# There are two options for replacing |today|: either, you set today to some
# non-false value, then it is used:
# today = ''
# Else, today_fmt is used as the format for a strftime call.
# today_fmt = '%B %d, %Y'

# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store", ".venv"]

# The reST default role (used for this markup: `text`) to use for all documents.
# default_role = None

# If true, '()' will be appended to :func: etc. cross-reference text.
# add_function_parentheses = True

# If true, the current module name will be prepended to all description
# unit titles (such as .. function::).
# add_module_names = True

# If true, sectionauthor and moduleauthor directives will be shown in the
# output. They are ignored by default.
# show_authors = False

# The name of the Pygments (syntax highlighting) style to use.
pygments_style = "sphinx"

# A list of ignored prefixes for module index sorting.
# modindex_common_prefix = []

# If true, keep warnings as "system message" paragraphs in the built documents.
# keep_warnings = False

# If this is True, todo emits a warning for each TODO entries. The default is False.
todo_emit_warnings = True


# -- Options for HTML output -------------------------------------------------

# The theme to use for HTML and HTML Help pages.  See the documentation for
# a list of builtin themes.
html_theme = "alabaster"

# Theme options are theme-specific and customize the look and feel of a theme
# further.  For a list of options available for each theme, see the
# documentation.
html_theme_options = {
    "sidebar_width": "300px",
    "page_width": "1200px"
}

# Add any paths that contain custom themes here, relative to this directory.
# html_theme_path = []

# The name for this set of Sphinx documents.  If None, it defaults to
# "<project> v<release> documentation".
# html_title = None

# A shorter title for the navigation bar.  Default is the same as html_title.
# html_short_title = None

# The name of an image file (relative to this directory) to place at the top
# of the sidebar.
# html_logo = ""

# The name of an image file (within the static path) to use as favicon of the
# docs.  This file should be a Windows icon file (.ico) being 16x16 or 32x32
# pixels large.
# html_favicon = None

# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ["_static"]

# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
# using the given strftime format.
# html_last_updated_fmt = '%b %d, %Y'

# If true, SmartyPants will be used to convert quotes and dashes to
# typographically correct entities.
# html_use_smartypants = True

# Custom sidebar templates, maps document names to template names.
# html_sidebars = {}

# Additional templates that should be rendered to pages, maps page names to
# template names.
# html_additional_pages = {}

# If false, no module index is generated.
# html_domain_indices = True

# If false, no index is generated.
# html_use_index = True

# If true, the index is split into individual pages for each letter.
# html_split_index = False

# If true, links to the reST sources are added to the pages.
# html_show_sourcelink = True

# If true, "Created using Sphinx" is shown in the HTML footer. Default is True.
# html_show_sphinx = True

# If true, "(C) Copyright ..." is shown in the HTML footer. Default is True.
# html_show_copyright = True

# If true, an OpenSearch description file will be output, and all pages will
# contain a <link> tag referring to it.  The value of this option must be the
# base URL from which the finished HTML is served.
# html_use_opensearch = ''

# This is the file name suffix for HTML files (e.g. ".xhtml").
# html_file_suffix = None

# Output file base name for HTML help builder.
htmlhelp_basename = "k8soptimizer-doc"


# -- Options for LaTeX output ------------------------------------------------

latex_elements = {
    # The paper size ("letterpaper" or "a4paper").
    # "papersize": "letterpaper",
    # The font size ("10pt", "11pt" or "12pt").
    # "pointsize": "10pt",
    # Additional stuff for the LaTeX preamble.
    # "preamble": "",
}

# Grouping the document tree into LaTeX files. List of tuples
# (source start file, target name, title, author, documentclass [howto/manual]).
latex_documents = [
    ("index", "user_guide.tex", "k8soptimizer Documentation", "Philipp Hellmich", "manual")
]

# The name of an image file (relative to this directory) to place at the top of
# the title page.
# latex_logo = ""

# For "manual" documents, if this is true, then toplevel headings are parts,
# not chapters.
# latex_use_parts = False

# If true, show page references after internal links.
# latex_show_pagerefs = False

# If true, show URL addresses after external links.
# latex_show_urls = False

# Documents to append as an appendix to all manuals.
# latex_appendices = []

# If false, no module index is generated.
# latex_domain_indices = True

# -- External mapping --------------------------------------------------------
python_version = ".".join(map(str, sys.version_info[0:2]))
intersphinx_mapping = {
    "sphinx": ("https://www.sphinx-doc.org/en/master", None),
    "python": ("https://docs.python.org/" + python_version, None),
    "matplotlib": ("https://matplotlib.org", None),
    "numpy": ("https://numpy.org/doc/stable", None),
    "sklearn": ("https://scikit-learn.org/stable", None),
    "pandas": ("https://pandas.pydata.org/pandas-docs/stable", None),
    "scipy": ("https://docs.scipy.org/doc/scipy/reference", None),
    "setuptools": ("https://setuptools.pypa.io/en/stable/", None),
    "pyscaffold": ("https://pyscaffold.org/en/stable", None),
}

print(f"loading configurations for {project} {version} ...", file=sys.stderr)

docs/conf.py
arvatoaws-labs/k8soptimizer
"""
    Setup file for k8soptimizer.
    Use setup.cfg to configure your project.

    This file was generated with PyScaffold 4.5.
    PyScaffold helps you to put up the scaffold of your new Python project.
    Learn more under: https://pyscaffold.org/
"""
from setuptools import setup

if __name__ == "__main__":
    try:
        setup(use_scm_version={"version_scheme": "no-guess-dev"})
    except:  # noqa
        print(
            "\n\nAn error occurred while building the project, "
            "please ensure you have the most updated version of setuptools, "
            "setuptools_scm and wheel with:\n"
            "   pip install -U setuptools setuptools_scm wheel\n\n"
        )
        raise

setup.py
arvatoaws-labs/k8soptimizer
import argparse
import re
from datetime import datetime, timezone

from beartype import beartype

__author__ = "Philipp Hellmich"
__copyright__ = "Arvato Systems GmbH"
__license__ = "MIT"


@beartype
def convert_memory_request_to_bytes(size: str) -> int:
    """
    Convert a memory size string to bytes.

    Args:
        size (str): The memory size string (e.g., '1Gi', '512M', '2.5Ki').

    Returns:
        int: The size in bytes.

    Raises:
        ValueError: If the input size has an invalid format or unit.

    Example:
        size_in_bytes = convert_memory_request_to_bytes('2Gi')
    """
    units = {
        "B": 1,
        "K": 1024,
        "M": 1024**2,
        "G": 1024**3,
        "T": 1024**4,
        "Ki": 1024,
        "Mi": 1024**2,
        "Gi": 1024**3,
        "Ti": 1024**4,
    }

    pattern = r"(\d+(\.\d+)?)\s*([a-zA-Z]+)"
    match = re.match(pattern, size)

    if match:
        value = float(match.group(1))
        unit = match.group(3)
    else:
        raise ValueError("Invalid format: {}".format(size))

    if unit not in units:
        raise ValueError("Invalid unit: {}".format(unit))

    bytes_value = value * units[unit]
    return int(bytes_value)


@beartype
def convert_cpu_request_to_cores(size: str) -> float:
    """
    Convert a CPU request size string to cores.

    Args:
        size (str): The CPU request size string (e.g., '500m', '1k', '0.5').

    Returns:
        float: The size in cores.

    Raises:
        ValueError: If the input size has an invalid format or unit.

    Example:
        cores = convert_cpu_request_to_cores('2.5k')
    """
    units = {"m": 1 / 1000, "k": 1000, "": 1}

    pattern = r"(\d+\.?\d?)(.*)?"
    match = re.match(pattern, size)

    if match:
        value = float(match.group(1))
        unit = match.group(2)
    else:
        raise ValueError("Invalid format: {}".format(size))

    if unit not in units:
        raise ValueError("Invalid unit: {}".format(unit))

    bytes_value = value * units[unit]
    return float(bytes_value)


@beartype
def format_pairs(values: dict) -> str:
    """
    Format a dictionary of key-value pairs into a comma-separated string.

    Args:
        values (dict): A dictionary containing key-value pairs.

    Returns:
        str: A string containing formatted key-value pairs, separated by commas.

    Example:
        formatted_string = format_pairs({"key1": "value1", "key2": "value2"})
    """
    formatted_pairs = []
    for key, value in values.items():
        formatted_pairs.append(f"{key}={value}")
    return ", ".join(formatted_pairs)


def create_timestamp():
    """
    Create a timestamp representing the current date and time in UTC.

    Returns:
        datetime: A datetime object representing the current UTC date and time.

    Example:
        timestamp = create_timestamp()
    """
    return datetime.now(timezone.utc)


@beartype
def calculate_minutes_ago_from_timestamp(datetime_object: datetime) -> int:
    """
    Calculate the number of minutes ago a given timestamp is from the current time.

    Args:
        datetime_object (datetime): The datetime object to calculate the minutes ago.

    Returns:
        int: The number of minutes ago.

    Example:
        timestamp = datetime(2023, 9, 7, 12, 0, tzinfo=timezone.utc)
        minutes_ago = calculate_minutes_ago_from_timestamp(timestamp)
    """
    current_time = datetime.now(timezone.utc)
    time_difference = current_time - datetime_object
    return int(time_difference.total_seconds() / 60)


@beartype
def calculate_minutes_ago_from_timestamp_str(timestamp: str) -> int:
    """
    Calculate the number of minutes ago a given timestamp is from the current time.

    Args:
        datetime_object (datetime): The datetime object to calculate the minutes ago.

    Returns:
        int: The number of minutes ago.

    Example:
        timestamp = '2023-09-11T09:04:50.539072+00:00'
        minutes_ago = calculate_minutes_ago_from_timestamp(timestamp)
    """
    datetime_object = datetime.fromisoformat(timestamp)
    return calculate_minutes_ago_from_timestamp(datetime_object)


@beartype
def is_valid_regex(value: str) -> bool:
    try:
        re.compile(value)
        return True
    except re.error:
        return False


@beartype
def valid_regex_arg(value: str) -> str:
    if is_valid_regex(value):
        return value
    raise argparse.ArgumentTypeError(f"'{value}' is not a valid regular expression")


@beartype
def is_valid_k8s_name(name: str) -> bool:
    # Check length
    if len(name) > 253:
        return False

    # Check pattern
    pattern = r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$"
    if bool(re.match(pattern, name)):
        return True
    return False


@beartype
def valid_k8s_name_arg(name: str) -> str:
    if is_valid_k8s_name(name):
        return name
    raise argparse.ArgumentTypeError(f"'{name}' is not a valid k8s object name")

src/k8soptimizer/helpers.py
arvatoaws-labs/k8soptimizer
"""
This is a skeleton file that can serve as a starting point for a Python
console script. To run this script uncomment the following lines in the
``[options.entry_points]`` section in ``setup.cfg``::

    console_scripts =
         k8soptimizer = k8soptimizer.main:run

Then run ``pip install .`` (or ``pip install -e .`` for editable mode)
which will install the command ``k8soptimizer`` inside your current environment.

Besides console scripts, the header (i.e. until ``_logger``...) of this file can
also be used as template for Python modules.

Note:
    This file can be renamed depending on your needs or safely removed if not needed.

References:
    - https://setuptools.pypa.io/en/latest/userguide/entry_point.html
    - https://pip.pypa.io/en/stable/reference/pip_install
"""

import argparse
import json
import logging
import os
import re
import sys

import requests
from beartype import beartype
from beartype.typing import Optional, Tuple
from kubernetes import client, config
from kubernetes.client.models import (
    V1Container,
    V1Deployment,
    V1DeploymentList,
    V1NamespaceList,
    V2HorizontalPodAutoscaler,
)
from pythonjsonlogger import jsonlogger

from . import __version__, helpers

__author__ = "Philipp Hellmich"
__copyright__ = "Arvato Systems GmbH"
__license__ = "MIT"

__domain__ = "arvato-aws.io"

_logger = logging.getLogger(__name__)


PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://localhost:9090")

NAMESPACE_PATTERN = os.getenv("NAMESPACE_PATTERN", ".*")
DEPLOYMENT_PATTERN = os.getenv("DEPLOYMENT_PATTERN", ".*")
CONTAINER_PATTERN = os.getenv("CONTAINER_PATTERN", ".*")

# in minutes
CREATE_AGE_THRESHOLD = int(os.getenv("CREATE_AGE_THRESHOLD", 60))
# cannot not be less than 5 minutes)
UPDATE_AGE_THRESHOLD = int(os.getenv("UPDATE_AGE_THRESHOLD", 60))
MIN_LOOKBACK_MINUTES = int(os.getenv("MIN_LOOKBACK_MINUTES", 30))
MAX_LOOKBACK_MINUTES = int(os.getenv("MIN_LOOKBACK_MINUTES", 3600 * 24 * 30))
OFFSET_LOOKBACK_MINUTES = int(os.getenv("OFFSET_LOOKBACK_MINUTES", 5))
DEFAULT_LOOKBACK_MINUTES = int(os.getenv("DEFAULT_LOOKBACK_MINUTES", 3600 * 24 * 7))
DEFAULT_QUANTILE_OVER_TIME = float(os.getenv("DEFAULT_QUANTILE_OVER_TIME", 0.95))

# operating mode
DRY_RUN_MODE = bool(os.getenv("DRY_RUN_MODE", False))

MIN_CPU_REQUEST = float(
    os.getenv("MIN_CPU_REQUEST", 0.010)
)  # below 10m will not work reliable with hpa
MAX_CPU_REQUEST = float(os.getenv("MAX_CPU_REQUEST", 16))
MAX_CPU_REQUEST_NODEJS = 1.0
CPU_REQUEST_RATIO = float(os.getenv("CPU_REQUEST_RATIO", 1.0))
MIN_MEMORY_REQUEST = int(os.getenv("MIN_MEMORY_REQUEST", 1024**2 * 16))
MAX_MEMORY_REQUEST = int(os.getenv("MAX_MEMORY_REQUEST", 1024**3 * 16))
MEMORY_REQUEST_RATIO = float(os.getenv("MEMORY_REQUEST_RATIO", 1.0))
MEMORY_LIMIT_RATIO = float(os.getenv("MEMORY_LIMIT_RATIO", 1.5))
MIN_MEMORY_LIMIT = int(
    os.getenv("MIN_MEMORY_LIMIT", MIN_MEMORY_REQUEST * MEMORY_LIMIT_RATIO)
)
MAX_MEMORY_LIMIT = int(
    os.getenv("MAX_MEMORY_LIMIT", MAX_MEMORY_REQUEST * MEMORY_LIMIT_RATIO)
)
CHANGE_THRESHOLD = os.getenv("CHANGE_THRESHOLD", 0.1)
HPA_THRESHOLD = os.getenv("HPA_THRESHOLD", 0.5)

LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
LOG_FORMAT = os.getenv("LOG_FORMAT", "json").lower()

stats = {}
stats["old_cpu_sum"] = 0
stats["new_cpu_sum"] = 0
stats["old_memory_sum"] = 0
stats["new_memory_sum"] = 0

# ---- Python API ----
# The functions defined in this section can be imported by users in their
# Python scripts/interactive interpreter, e.g. via
# `from k8soptimizer.skeleton import fib`,
# when using this Python module as a library.


class AppFilter(logging.Filter):
    extra = {}

    def __init__(self, extra={}):
        self.extra = extra
        super(AppFilter, self).__init__()

    def filter(self, record):
        for key, value in self.extra.items():
            record.__setattr__(key, value)
        for key in list(record.__dict__.keys()):
            if key not in self.extra and key not in [
                "name",
                "msg",
                "args",
                "levelname",
                "levelno",
                "pathname",
                "filename",
                "module",
                "exc_info",
                "exc_text",
                "stack_info",
                "lineno",
                "funcName",
                "created",
                "msecs",
                "relativeCreated",
                "thread",
                "threadName",
                "processName",
                "process",
            ]:
                del record.__dict__[key]
        return True


@beartype
def query_prometheus(query: str) -> dict:
    """
    Query Prometheus API with the specified query string.

    Args:
        query (str): The Prometheus query string.

    Returns:
        dict: The JSON response from the Prometheus API.

    Raises:
        RuntimeError: If the response is missing expected data fields.

    Example:
        response = query_prometheus('sum(rate(http_requests_total{job="api"}[5m]))')
    """
    _logger.debug(query)
    response = requests.get(PROMETHEUS_URL + "/api/v1/query", params={"query": query})
    j = json.loads(response.text)
    _logger.debug(j)
    if "data" not in j:
        raise RuntimeError("Got invalid results from query: {}".format(query))
    if "result" not in j["data"]:
        raise RuntimeError("Got invalid results from query: {}".format(query))
    return j


@beartype
def verify_prometheus_connection() -> bool:
    """
    Verify connection to the Prometheus API.

    Returns:
        bool: True if the connection is successful, False otherwise.

    Raises:
        RuntimeError: If the response is missing expected data fields or the connection fails.

    Example:
        connection_successful = verify_prometheus_connection()
    """
    response = requests.get(PROMETHEUS_URL + "/api/v1/status/buildinfo")
    j = json.loads(response.text)
    _logger.debug(j)
    if "status" not in j:
        raise RuntimeError("Got invalid results request: {}".format(response.text))
    if j["status"] == "success":
        return True
    raise RuntimeError("Connection to prometheus api failed")


@beartype
def verify_kubernetes_connection() -> bool:
    """
    Verify connection to the Kubernetes API.

    Returns:
        bool: True if the connection is successful, False otherwise.

    Raises:
        RuntimeError: If the connection to the Kubernetes API fails or if there is a configuration error.

    Example:
        connection_successful = verify_kubernetes_connection()
    """
    config.load_kube_config()
    client.ApisApi().get_api_versions_with_http_info()
    return True


@beartype
def get_max_cpu_cores_per_runtime(runtime: str) -> int:
    """
    Get the maximum number of CPU cores allowed per runtime.

    Args:
        runtime (str): The name of the runtime.

    Returns:
        int: The maximum number of CPU cores.

    Example:
        max_cores = get_max_cpu_cores_per_runtime("nodejs")
    """
    if runtime == "nodejs":
        return 1
    return 100


@beartype
def get_number_of_samples_from_history(
    namespace: str,
    workload: str,
    workload_type: str = "deployment",
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
) -> int:
    """
    Get the CPU cores usage history for a specific container.

    Args:
        namespace (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        workload_type (str, optional): The type of workload. Default is "deployment".
        lookback_minutes (int, optional): The number of minutes to look back in time for the query. Default is DEFAULT_LOOKBACK_MINUTES.

    Returns:
        float: The number of samples from prometheus.

    Raises:
        RuntimeError: If no data is found for the Prometheus query.

    Example:
        samples = get_number_of_samples_from_history("my-namespace", "my-deployment")
    """
    query = 'max by (namespace,workload,workload_type) (count_over_time(kube_workload_container_resource_usage_cpu_cores_avg{{namespace="{namespace}", workload="{workload}", workload_type="{workload_type}"}}[{lookback_minutes}m]))'.format(
        namespace=namespace,
        workload=workload,
        workload_type=workload_type,
        lookback_minutes=lookback_minutes,
    )
    j = query_prometheus(query)

    if j["data"]["result"] == []:
        raise RuntimeError("No data found for prometheus query: {}".format(query))
    return int(j["data"]["result"][0]["value"][1])


# FIXME make avg
@beartype
def get_max_pods_per_deployment_history(
    namespace_name: str,
    deployment_name: str,
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
    quantile_over_time: float = DEFAULT_QUANTILE_OVER_TIME,
) -> int:
    """
    Get the maximum number of pods for a deployment based on historical data.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        deployment_name (str): The name of the deployment.
        lookback_minutes (int, optional): The number of minutes to look back in time for the query. Default is DEFAULT_LOOKBACK_MINUTES.
        quantile_over_time (float, optional): The quantile value for the query. Default is DEFAULT_QUANTILE_OVER_TIME.

    Returns:
        int: The maximum number of pods.

    Raises:
        RuntimeError: If no data is found for the Prometheus query.

    Example:
        max_pods = get_max_pods_per_deployment_history("my-namespace", "my-deployment")
    """
    query = 'max(quantile_over_time({quantile_over_time}, kube_deployment_spec_replicas{{job="kube-state-metrics", namespace="{namespace_name}", deployment="{deployment_name}"}}[{lookback_minutes}m]))'.format(
        quantile_over_time=quantile_over_time,
        namespace_name=namespace_name,
        deployment_name=deployment_name,
        lookback_minutes=lookback_minutes,
    )
    j = query_prometheus(query)

    if j["data"]["result"] == []:
        raise RuntimeError("No data found for prometheus query: {}".format(query))
    return int(j["data"]["result"][0]["value"][1])


@beartype
def get_cpu_cores_usage_history(
    namespace: str,
    workload: str,
    container: str,
    workload_type: str = "deployment",
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
    quantile_over_time: float = DEFAULT_QUANTILE_OVER_TIME,
) -> float:
    """
    Get the CPU cores usage history for a specific container.

    Args:
        namespace (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        container (str): The name of the container.
        workload_type (str, optional): The type of workload. Default is "deployment".
        lookback_minutes (int, optional): The number of minutes to look back in time for the query. Default is DEFAULT_LOOKBACK_MINUTES.
        quantile_over_time (float, optional): The quantile value for the query. Default is DEFAULT_QUANTILE_OVER_TIME.

    Returns:
        float: The CPU cores usage value.

    Raises:
        RuntimeError: If no data is found for the Prometheus query.

    Example:
        cpu_usage = get_cpu_cores_usage_history("my-namespace", "my-deployment", "my-container")
    """
    query = 'quantile_over_time({quantile_over_time}, kube_workload_container_resource_usage_cpu_cores_avg{{namespace="{namespace}", workload="{workload}", workload_type="{workload_type}", container="{container}"}}[{lookback_minutes}m])'.format(
        quantile_over_time=quantile_over_time,
        namespace=namespace,
        workload=workload,
        workload_type=workload_type,
        container=container,
        lookback_minutes=lookback_minutes,
    )
    j = query_prometheus(query)

    if j["data"]["result"] == []:
        raise RuntimeError("No data found for prometheus query: {}".format(query))
    return float(j["data"]["result"][0]["value"][1])


@beartype
def get_memory_bytes_usage_history(
    namespace: str,
    workload: str,
    container: str,
    workload_type: str = "deployment",
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
    quantile_over_time: float = DEFAULT_QUANTILE_OVER_TIME,
) -> float:
    """
    Get the memory usage history (in bytes) for a specific container.

    Args:
        namespace (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        container (str): The name of the container.
        workload_type (str, optional): The type of workload. Default is "deployment".
        lookback_minutes (int, optional): The number of minutes to look back in time for the query. Default is DEFAULT_LOOKBACK_MINUTES.
        quantile_over_time (float, optional): The quantile value for the query. Default is DEFAULT_QUANTILE_OVER_TIME.

    Returns:
        float: The memory usage value in bytes.

    Raises:
        RuntimeError: If no data is found for the Prometheus query.

    Example:
        memory_usage = get_memory_bytes_usage_history("my-namespace", "my-deployment", "my-container")
    """
    query = 'quantile_over_time({quantile_over_time}, kube_workload_container_resource_usage_memory_bytes_max{{namespace="{namespace}", workload="{workload}", workload_type="{workload_type}", container="{container}"}}[{lookback_minutes}m])'.format(
        quantile_over_time=quantile_over_time,
        namespace=namespace,
        workload=workload,
        workload_type=workload_type,
        container=container,
        lookback_minutes=lookback_minutes,
    )
    j = query_prometheus(query)

    if j["data"]["result"] == []:
        raise RuntimeError("No data found for prometheus query: {}".format(query))
    return float(j["data"]["result"][0]["value"][1])


@beartype
def discover_container_runtime(
    namespace: str, workload: str, container: str, workload_type: str = "deployment"
) -> Optional[str]:
    """
    Discover the container runtime for a specific container.

    Args:
        namespace (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        container (str): The name of the container.
        workload_type (str, optional): The type of workload. Default is "deployment".

    Returns:
        Optional[str]: The name of the container runtime, or None if it couldn't be determined.

    Example:
        runtime = discover_container_runtime("my-namespace", "my-deployment", "my-container")
    """
    if is_nodejs_container(namespace, workload, container, workload_type):
        return "nodejs"
    return None


@beartype
def get_hpa_for_deployment(
    namespace_name: str, deployment_name: str
) -> Optional[V2HorizontalPodAutoscaler]:
    """
    Get the Horizontal Pod Autoscaler (HPA) associated with a specific deployment.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        deployment_name (str): The name of the deployment.

    Returns:
        Optional[V2HorizontalPodAutoscaler]: The HPA object if found, or None if not found.

    Example:
        hpa = get_hpa_for_deployment("my-namespace", "my-deployment")
    """
    autoscaling_api = client.AutoscalingV2Api()
    for hpa in autoscaling_api.list_namespaced_horizontal_pod_autoscaler(
        namespace=namespace_name
    ).items:
        if hpa.spec.scale_target_ref.kind != "Deployment":
            continue
        if hpa.spec.scale_target_ref.name != deployment_name:
            continue
        return hpa
    return None


@beartype
def is_hpa_enabled_for_deployment(namespace_name: str, deployment_name: str) -> bool:
    """
    Check if Horizontal Pod Autoscaling (HPA) is enabled for a specific deployment.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        deployment_name (str): The name of the deployment.

    Returns:
        bool: True if HPA is enabled, False otherwise.

    Example:
        hpa_enabled = is_hpa_enabled_for_deployment("my-namespace", "my-deployment")
    """
    return get_hpa_for_deployment(namespace_name, deployment_name) is not None


@beartype
def calculate_hpa_target_ratio(
    namespace_name: str,
    deployment_name: str,
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
) -> dict:
    """
    Calculate the target ratio for Horizontal Pod Autoscaling (HPA) based on historical data.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        deployment_name (str): The name of the deployment.
        lookback_minutes (int, optional): The number of minutes to look back in time for historical data. Default is DEFAULT_LOOKBACK_MINUTES.

    Returns:
        dict: A dictionary with target ratios for CPU and memory.

    Example:
        target_ratios = calculate_hpa_target_ratio("my-namespace", "my-deployment")
    """
    hpa_ratio_addon = 0
    hpa_min_replica = 0
    hpa_max_replica = 0
    hpa_range = 0
    target_ratio_cpu = CPU_REQUEST_RATIO
    target_ratio_memory = MEMORY_REQUEST_RATIO

    hpa = get_hpa_for_deployment(namespace_name, deployment_name)
    if hpa is None:
        _logger.info("Hpa not found for: %s" % deployment_name)
        return {"cpu": float(target_ratio_cpu), "memory": float(target_ratio_memory)}

    hpa_min_replica = hpa.spec.min_replicas
    hpa_max_replica = hpa.spec.max_replicas
    hpa_range = hpa_max_replica - hpa_min_replica
    for metric in hpa.spec.metrics:
        if metric.type != "Resource":
            continue
        if metric.resource.name == "cpu":
            hpa_target_cpu = metric.resource.target.average_utilization / 100
            target_ratio_cpu = round(1 / hpa_target_cpu, 2)
            _logger.info(
                "Changing target ratio cpu to match hpa utilization: %s"
                % target_ratio_cpu
            )
        if metric.resource.name == "memory":
            hpa_target_memory = metric.resource.target.average_utilization / 100
            target_ratio_memory = round(1 / hpa_target_memory, 2)
            _logger.info(
                "Changing target ratio memory to match hpa utilization: %s"
                % target_ratio_memory
            )

    replica_count_history = round(
        get_max_pods_per_deployment_history(
            namespace_name, deployment_name, lookback_minutes
        )
    )

    _logger.debug("Hpa min repliacs: %s" % hpa_min_replica)
    _logger.debug("Hpa max replicas: %s" % hpa_max_replica)
    _logger.debug("Hpa avg count history: %s" % replica_count_history)

    # increase cpu request if current replica count is higher than min replica count
    if replica_count_history > hpa_min_replica and hpa_range > 0:
        _logger.debug("Hpa avg range: %s" % hpa_range)
        hpa_range_position = replica_count_history - hpa_min_replica
        _logger.debug("Hpa avg range position: %s" % hpa_range_position)
        hpa_ratio_addon = round((hpa_range_position) / hpa_range, 2)

    if hpa_ratio_addon > HPA_THRESHOLD:
        _logger.debug(
            "Increasing target ratio cpu due to hpa near limit: %s" % hpa_ratio_addon
        )
        target_ratio_cpu = round(target_ratio_cpu + hpa_ratio_addon, 3)

    return {"cpu": float(target_ratio_cpu), "memory": float(target_ratio_memory)}


@beartype
def calculate_cpu_requests(
    namespace_name: str,
    workload: str,
    workload_type: str,
    container_name: str,
    target_ratio_cpu: float,
    lookback_minutes: int,
) -> float:
    """
    Calculate the CPU requests for a specific container based on historical data and target ratio.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        workload_type (str): The type of workload (e.g., deployment,daemonset,statefulset).
        container_name (str): The name of the container.
        target_ratio_cpu (float): The target ratio for CPU.
        lookback_minutes (int): The number of minutes to look back in time for historical data.

    Returns:
        float: The calculated CPU requests.

    Example:
        cpu_requests = calculate_cpu_requests("my-namespace", "my-workload", "deployment", "my-container", 1.5, 60)
    """
    new_cpu = round(
        max(
            MIN_CPU_REQUEST,
            min(
                MAX_CPU_REQUEST,
                get_cpu_cores_usage_history(
                    namespace_name,
                    workload,
                    container_name,
                    workload_type,
                    lookback_minutes,
                )
                * target_ratio_cpu,
            ),
        ),
        3,
    )
    runtime = discover_container_runtime(
        namespace_name, workload, container_name, workload_type
    )
    _logger.debug("Runtime: %s" % runtime)
    if runtime == "nodejs":
        new_cpu = min(MAX_CPU_REQUEST_NODEJS, new_cpu)

    return float(new_cpu)


@beartype
def calculate_memory_requests(
    namespace_name: str,
    workload: str,
    workload_type: str,
    container_name: str,
    target_ratio_memory: float,
    lookback_minutes: int,
):
    """
    Calculate the memory requests for a specific container based on historical data, target ratio, and OOM history.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        workload_type (str): The type of workload (e.g., deployment,daemonset,statefulset).
        container_name (str): The name of the container.
        target_ratio_memory (float): The target ratio for memory.
        lookback_minutes (int): The number of minutes to look back in time for historical data.

    Returns:
        int: The calculated memory requests in bytes.

    Example:
        memory_requests = calculate_memory_requests("my-namespace", "my-workload", "deployment", "my-container", 1.5, 60)
    """
    if (
        get_oom_killed_history(
            namespace_name, workload, container_name, workload_type, lookback_minutes
        )
        > 0
    ):
        target_ratio_memory = target_ratio_memory * 2

    new_memory = round(
        max(
            MIN_MEMORY_REQUEST,
            min(
                MAX_MEMORY_REQUEST,
                get_memory_bytes_usage_history(
                    namespace_name,
                    workload,
                    container_name,
                    workload_type,
                )
                * target_ratio_memory,
            ),
        )
    )

    return int(new_memory)


@beartype
def calculate_memory_limits(
    namespace_name: str,
    workload: str,
    workload_type: str,
    container_name: str,
    memory_requests: int,
) -> int:
    """
    Calculate the memory limits for a specific container based on memory requests.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        workload_type (str): The type of workload (e.g., deployment,daemonset,statefulset).
        container_name (str): The name of the container.
        memory_requests (int): The memory requests in bytes.

    Returns:
        int: The calculated memory limits in bytes.

    Example:
        memory_limits = calculate_memory_limits("my-namespace", "my-workload", "deployment", "my-container", 2048)
    """
    new_memory_limit = max(
        MIN_MEMORY_LIMIT,
        min(MAX_MEMORY_LIMIT, memory_requests * MEMORY_LIMIT_RATIO),
    )
    # new_memory_limit = memory_requests * MEMORY_LIMIT_RATIO
    return round(new_memory_limit)


@beartype
def get_namespaces(namespace_pattern: str = ".*") -> V1NamespaceList:
    """
    Get a list of Kubernetes namespaces that match the specified pattern.

    Args:
        namespace_pattern (str, optional): A regular expression pattern to filter namespaces. Default is ".*".

    Returns:
        V1NamespaceList: A list of namespaces.

    Example:
        namespaces = get_namespaces("my-namespace.*")
    """
    core_api = client.CoreV1Api()
    resp_ns = core_api.list_namespace(watch=False)
    items = []

    for namespace in resp_ns.items:
        _logger.debug(namespace)
        namespace_name = namespace.metadata.name

        x = re.search(namespace_pattern, namespace_name)
        if x is None:
            _logger.debug(
                "Skipping namespace due to NAMESPACE_PATTERN: %s" % namespace_name
            )
            continue

        items.append(namespace)

    return V1NamespaceList(items=items)


@beartype
def get_deployments(
    namespace_name: str, deplopyment_pattern: str = ".*", only_running: bool = True
) -> V1DeploymentList:
    """
    Get a list of Kubernetes deployments in a specific namespace that match the specified pattern.

    Args:
        namespace_name (str): The name of the Kubernetes namespace.
        deplopyment_pattern (str, optional): A regular expression pattern to filter deployments. Default is ".*".
        only_running (bool, optional): Flag to include only running deployments. Default is True.

    Returns:
        V1DeploymentList: A list of deployments.

    Example:
        deployments = get_deployments("my-namespace", "my-deployment.*", only_running=True)
    """
    apis_api = client.AppsV1Api()
    resp_deploy = apis_api.list_namespaced_deployment(namespace=namespace_name)
    items = []
    for deployment in resp_deploy.items:
        _logger.debug(deployment)
        deployment_name = deployment.metadata.name

        x = re.search(deplopyment_pattern, deployment_name)
        if x is None:
            _logger.debug(
                "Skipping deployment due to DEPLOYMENT_PATTERN: %s" % deployment_name
            )
            continue

        if only_running and deployment.spec.replicas == 0:
            _logger.debug(
                "Skipping deployment due to zero replicas: %s" % deployment_name
            )
            continue

        items.append(deployment)

    return V1DeploymentList(items=items)


@beartype
def get_oom_killed_history(
    namespace: str,
    workload: str,
    container: str,
    workload_type: str = "deployment",
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
) -> int:
    """
    Get the count of out-of-memory (OOM) events for a specific container based on historical data.

    Args:
        namespace (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        container (str): The name of the container.
        workload_type (str, optional): The type of workload. Default is "deployment".
        lookback_minutes (int, optional): The number of minutes to look back in time for historical data.
                                         Default is the value of DEFAULT_LOOKBACK_MINUTES.

    Returns:
        int: The count of OOM events.

    Example:
        oom_count = get_oom_killed_history("my-namespace", "my-workload", "my-container", "deployment", 60)
    """
    query = 'sum_over_time(kube_workload_container_resource_usage_memory_oom_killed{{namespace="{namespace}", workload="{workload}", workload_type="{workload_type}", container="{container}"}}[{lookback_minutes}m])'.format(
        namespace=namespace,
        workload=workload,
        workload_type=workload_type,
        container=container,
        lookback_minutes=lookback_minutes,
    )
    j = query_prometheus(query)

    if j["data"]["result"] == []:
        return 0

    if float(j["data"]["result"][0]["value"][1]) > 0:
        return round(float(j["data"]["result"][0]["value"][1]))

    return 0


@beartype
def is_nodejs_container(
    namespace: str, workload: str, container: str, workload_type: str = "deployment"
) -> bool:
    """
    Check if a container in a specific workload and namespace is a Node.js container.

    Args:
        namespace (str): The name of the Kubernetes namespace.
        workload (str): The name of the workload (e.g., myapp).
        container (str): The name of the container.
        workload_type (str, optional): The type of workload. Default is "deployment".

    Returns:
        bool: True if the container is a Node.js container, False otherwise.

    Example:
        is_nodejs = is_nodejs_container("my-namespace", "my-workload", "my-container", "deployment")
    """
    query = 'count(nodejs_version_info{{container="{container}"}} * on(namespace,pod) group_left(workload, workload_type) namespace_workload_pod:kube_pod_owner:relabel{{workload="{workload}", workload_type="{workload_type}", namespace="{namespace}"}}) by (namespace, workload, workload_type, container)'.format(
        namespace=namespace,
        workload=workload,
        workload_type=workload_type,
        container=container,
    )
    j = query_prometheus(query)

    if j["data"]["result"] == []:
        return False

    if float(j["data"]["result"][0]["value"][1]) > 0:
        return True

    return False


@beartype
def get_resources_from_deployment(deployment: V1Deployment) -> dict:
    """
    Get resource specifications (requests and limits) for containers in a deployment.

    Args:
        deployment (V1Deployment): The Kubernetes deployment object.

    Returns:
        dict: A dictionary containing resource specifications for each container.

    Example:
        deployment = get_deployment_by_name("my-namespace", "my-deployment")
        resources = get_resources_from_deployment(deployment)
    """
    res = {}
    for container in deployment.spec.template.spec.containers:
        res[container.name] = {}
        try:
            res[container.name]["requests"] = container.resources.requests
        except AttributeError:
            res[container.name]["requests"] = {}
        try:
            res[container.name]["limits"] = container.resources.limits
        except AttributeError:
            res[container.name]["limits"] = {}
    return res


@beartype
def calculate_lookback_minutes_from_deployment(deployment: V1Deployment) -> int:
    """
    Calculate the lookback minutes based on a deployment's creation and last update timestamps.

    Args:
        deployment (V1Deployment): The Kubernetes deployment object.

    Returns:
        int: The calculated lookback minutes.

    Raises:
        RuntimeError: If the deployment is too young or was modified too recently.

    Example:
        deployment = get_deployment_by_name("my-namespace", "my-deployment")
        lookback = calculate_lookback_minutes_from_deployment(deployment)
    """

    lookback_minutes = DEFAULT_LOOKBACK_MINUTES
    _logger.debug("DEFAULT_LOOKBACK_MINUTES is %s" % DEFAULT_LOOKBACK_MINUTES)
    if DEFAULT_LOOKBACK_MINUTES < MIN_LOOKBACK_MINUTES:
        raise RuntimeError(
            f"The specified lookback period ({lookback_minutes} minutes) is below the minimum required ({MIN_LOOKBACK_MINUTES} minutes). Please provide a longer lookback period to ensure sufficient historical data is available."
        )

    if deployment.metadata.creation_timestamp is None:
        raise RuntimeError("The deployment has no creation timestamp.")

    creation_minutes_ago = helpers.calculate_minutes_ago_from_timestamp(
        deployment.metadata.creation_timestamp
    )

    if creation_minutes_ago < CREATE_AGE_THRESHOLD:
        raise RuntimeError(
            f"The deployment is too young. It was created {creation_minutes_ago} minutes ago, which is below the minimum threshold of {CREATE_AGE_THRESHOLD} minutes."
        )
    lookback_minutes = min(MAX_LOOKBACK_MINUTES, lookback_minutes, creation_minutes_ago)
    _logger.debug("creation_minutes_ago is %s" % creation_minutes_ago)

    if (
        deployment.metadata.annotations is not None
        and "k8soptimizer.arvato-aws.io/last-update" in deployment.metadata.annotations
    ):
        update_minutes_ago = helpers.calculate_minutes_ago_from_timestamp_str(
            deployment.metadata.annotations["k8soptimizer.arvato-aws.io/last-update"]
        )
        _logger.debug("update_minutes_ago is %s" % update_minutes_ago)
        update_minutes_ago_with_offset = update_minutes_ago - OFFSET_LOOKBACK_MINUTES
        _logger.debug(
            "update_minutes_ago_with_offset is %s" % update_minutes_ago_with_offset
        )

        if update_minutes_ago < OFFSET_LOOKBACK_MINUTES:
            raise RuntimeError(
                f"The deployment was optimized too recently. It was updated {update_minutes_ago} minutes ago, which is even below the lookback offset of {OFFSET_LOOKBACK_MINUTES} minutes since last update."
            )

        if update_minutes_ago_with_offset < UPDATE_AGE_THRESHOLD:
            raise RuntimeError(
                f"The deployment was optimized too recently. It was updated {update_minutes_ago_with_offset} minutes ago, which is below the minimum threshold of {UPDATE_AGE_THRESHOLD} minutes since last update."
            )

        lookback_minutes = min(
            MAX_LOOKBACK_MINUTES, lookback_minutes, update_minutes_ago_with_offset
        )

    history_samples = get_number_of_samples_from_history(
        deployment.metadata.namespace,
        deployment.metadata.name,
        "deployment",
        lookback_minutes,
    )

    if history_samples < MIN_LOOKBACK_MINUTES:
        raise RuntimeError(
            f"The provided history samples ({history_samples} samples) are below the minimum required ({MIN_LOOKBACK_MINUTES}). Please ensure an adequate amount of historical data is available."
        )

    _logger.debug("history_samples is %s" % history_samples)

    lookback_minutes = min(MAX_LOOKBACK_MINUTES, lookback_minutes, history_samples)

    if lookback_minutes < MIN_LOOKBACK_MINUTES:
        raise RuntimeError(
            f"The specified lookback period ({lookback_minutes} minutes) is below the minimum required ({MIN_LOOKBACK_MINUTES} minutes). Please provide a longer lookback period to ensure sufficient historical data is available."
        )

    _logger.debug("final lookback_minutes is %s" % lookback_minutes)

    return int(lookback_minutes)


@beartype
def optimize_deployment(
    deployment: V1Deployment, container_pattern=CONTAINER_PATTERN, dry_run=True
) -> V1Deployment:
    """
    Optimize the resources (CPU and memory) for containers in a deployment.

    Args:
        deployment (V1Deployment): The Kubernetes deployment object to be optimized.
        dry_run (bool, optional): If True, the optimization changes will be simulated (dry-run mode).
                                 If False, the changes will be applied. Default is True.

    Returns:
        V1Deployment: The optimized Kubernetes deployment object.

    Example:
        deployment = get_deployment_by_name("my-namespace", "my-deployment")
        optimized_deployment = optimize_deployment(deployment, dry_run=True)
    """
    apis_api = client.AppsV1Api()
    namespace_name = deployment.metadata.namespace
    deployment_name = deployment.metadata.name

    extra = {"namespace": namespace_name, "deployment": deployment_name}
    _logger.addFilter(AppFilter(extra))
    # _logger = logging.LoggerAdapter(_logger, extra)

    _logger.info("Optimizing deployment: %s" % deployment_name)

    old_resources = get_resources_from_deployment(deployment)
    lookback_minutes = calculate_lookback_minutes_from_deployment(deployment)
    _logger.info("Lookback minutes: %s" % lookback_minutes)
    target_ratio = calculate_hpa_target_ratio(
        namespace_name, deployment_name, lookback_minutes
    )
    _logger.info("Target ratio cpu: %s" % target_ratio["cpu"])
    _logger.info("Target ratio memory: %s" % target_ratio["memory"])
    changed = False
    for i, container in enumerate(deployment.spec.template.spec.containers):
        container_name = container.name
        extra = {
            "namespace": namespace_name,
            "deployment": deployment_name,
            "container": container_name,
        }
        _logger.addFilter(AppFilter(extra))
        x = re.search(container_pattern, container_name)
        if x is None:
            _logger.debug(
                "Skipping container due to CONTAINER_PATTERN: %s" % container_name
            )
            continue

        container_new, changed_container = optimize_container(
            namespace_name,
            deployment_name,
            container,
            "deployment",
            target_ratio["cpu"],
            target_ratio["memory"],
            lookback_minutes,
            deployment.spec.replicas,
        )
        if changed_container:
            changed = True
        deployment.spec.template.spec.containers[i] = container_new

    extra = {"namespace": namespace_name, "deployment": deployment_name}
    _logger.addFilter(AppFilter(extra))

    if changed:
        deployment.metadata.annotations[
            "k8soptimizer.{}/old-resources".format(__domain__)
        ] = json.dumps(old_resources)
        deployment.metadata.annotations[
            "k8soptimizer.{}/last-update".format(__domain__)
        ] = helpers.create_timestamp()

        # Apply the changes
        if dry_run is True:
            _logger.info("Updating (dry-run) deployment: %s" % deployment_name)
            apis_api.patch_namespaced_deployment(
                deployment_name,
                namespace_name,
                deployment,
                pretty=True,
                dry_run="All",
            )
        else:
            _logger.info("Updating deployment: %s" % deployment_name)
            apis_api.patch_namespaced_deployment(
                deployment_name, namespace_name, deployment, pretty=True
            )
    else:
        _logger.info("Nothing changed deployment: %s" % deployment_name)
    return deployment


@beartype
def optimize_container(
    namespace_name: str,
    workload: str,
    container: V1Container,
    workload_type: str = "deployment",
    target_ratio_cpu: float = 1,
    target_ratio_memory: float = 1,
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
    current_replicas: int = 1,
) -> Tuple[V1Container, bool]:
    """
    Optimize resources (CPU and memory) for a container.

    Args:
        namespace_name (str): The namespace of the container.
        workload (str): The name of the workload associated with the container.
        container (V1Container): The Kubernetes container object to be optimized.
        workload_type (str, optional): The type of workload (e.g., "deployment"). Default is "deployment".
        target_ratio_cpu (float, optional): The target ratio for CPU optimization. Default is 1.
        target_ratio_memory (float, optional): The target ratio for memory optimization. Default is 1.
        lookback_minutes (int, optional): The lookback minutes for historical data. Default is DEFAULT_LOOKBACK_MINUTES.
        current_replicas (int, optional): The current number of replicas. Default is 1.

    Returns:
        V1Container: The optimized Kubernetes container object.
        changed (bool): True if the container was changed, False otherwise.

    Example:
        namespace = "my-namespace"
        workload = "my-workload"
        container = get_container_by_name(namespace, workload, "my-container")
        optimized_container = optimize_container(namespace, workload, container, target_ratio_cpu=0.8)
    """
    container_name = container.name

    _logger.info("Processing container: %s" % container_name)

    old_cpu = get_cpu_requests_from_container(container)
    new_cpu, changed_cpu = optimize_container_cpu_requests(
        namespace_name,
        workload,
        container,
        workload_type,
        target_ratio_cpu,
        lookback_minutes,
    )
    old_memory = get_memory_requests_from_container(container)
    new_memory, changed_memory = optimize_container_memory_requests(
        namespace_name,
        workload,
        container,
        workload_type,
        target_ratio_memory,
        lookback_minutes,
    )
    new_memory_limit, changed_memory_limit = optimize_container_memory_limits(
        namespace_name, workload, container, workload_type, new_memory
    )

    stats["old_cpu_sum"] += old_cpu * current_replicas
    stats["new_cpu_sum"] += new_cpu * current_replicas
    stats["old_memory_sum"] += old_memory * current_replicas
    stats["new_memory_sum"] += new_memory * current_replicas

    container.resources.requests["cpu"] = str(round(new_cpu * 1000)) + "m"
    if "cpu" in container.resources.limits:
        del container.resources.limits["cpu"]
        changed_cpu_limit = True
    else:
        changed_cpu_limit = False
    container.resources.requests["memory"] = str(round(new_memory / 1024 / 1024)) + "Mi"
    container.resources.limits["memory"] = (
        str(round(new_memory_limit / 1024 / 1024)) + "Mi"
    )

    return container, any(
        [changed_cpu, changed_cpu_limit, changed_memory, changed_memory_limit]
    )


@beartype
def get_cpu_requests_from_container(container: V1Container) -> float:
    """
    Get the CPU requests from a Kubernetes container.

    Args:
        container (V1Container): The Kubernetes container object.

    Returns:
        float: The CPU requests in cores.

    Example:
        container = V1Container(name="my-container", resources=V1ResourceRequirements(requests={"cpu": "100m"}))
        cpu_requests = get_cpu_requests_from_container(container)
    """
    try:
        old_cpu = helpers.convert_cpu_request_to_cores(
            container.resources.requests["cpu"]
        )
    except (KeyError, AttributeError):
        old_cpu = 0.001
    return float(old_cpu)


@beartype
def get_memory_requests_from_container(container: V1Container) -> int:
    """
    Get the memory requests from a Kubernetes container.

    Args:
        container (V1Container): The Kubernetes container object.

    Returns:
        int: The memory requests in bytes.

    Example:
        container = V1Container(name="my-container", resources=V1ResourceRequirements(requests={"memory": "2Gi"}))
        memory_requests = get_memory_requests_from_container(container)
    """
    try:
        old_memory = helpers.convert_memory_request_to_bytes(
            container.resources.requests["memory"]
        )
    except (KeyError, AttributeError):
        old_memory = 1024**2 * 1
    return int(old_memory)


@beartype
def get_memory_limits_from_container(container: V1Container) -> int:
    """
    Get the memory limits from a Kubernetes container.

    Args:
        container (V1Container): The Kubernetes container object.

    Returns:
        int: The memory limits in bytes.

    Example:
        container = V1Container(name="my-container", resources=V1ResourceRequirements(limits={"memory": "4Gi"}))
        memory_limits = get_memory_limits_from_container(container)
    """
    try:
        old_memory = helpers.convert_memory_request_to_bytes(
            container.resources.limits["memory"]
        )
    except (KeyError, AttributeError):
        old_memory = 1024**2 * 1
    return int(old_memory)


@beartype
def optimize_container_cpu_requests(
    namespace_name: str,
    workload: str,
    container: V1Container,
    workload_type: str = "deployment",
    target_ratio: float = 1,
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
) -> Tuple[float, bool]:
    """
    Optimize CPU requests for a Kubernetes container.

    Args:
        namespace_name (str): The namespace of the workload.
        workload (str): The name of the workload.
        container (V1Container): The Kubernetes container object.
        workload_type (str, optional): The type of workload. Defaults to "deployment".
        target_ratio (float, optional): The target ratio for CPU optimization. Defaults to 1.
        lookback_minutes (int, optional): The number of minutes to look back for resource usage data. Defaults to DEFAULT_LOOKBACK_MINUTES.

    Returns:
        float: The new CPU request in cores.
        bool: If something was changed

    Example:
        namespace_name = "my-namespace"
        workload = "my-workload"
        container = V1Container(name="my-container", resources=V1ResourceRequirements(requests={"cpu": "100m"}))
        new_cpu = optimize_container_cpu_requests(namespace_name, workload, container)
    """
    container_name = container.name

    try:
        _logger.debug(container.resources.requests["cpu"])
        old_cpu = helpers.convert_cpu_request_to_cores(
            container.resources.requests["cpu"]
        )
    except (KeyError, AttributeError):
        _logger.info("Could not read old CPU requests aassuming it is 0.001")
        old_cpu = 0.001

    new_cpu = calculate_cpu_requests(
        namespace_name,
        workload,
        workload_type,
        container_name,
        target_ratio,
        lookback_minutes,
    )

    diff_cpu = round(((new_cpu / old_cpu) - 1) * 100)
    change_too_small = abs(diff_cpu) < CHANGE_THRESHOLD * 100

    if change_too_small:
        _logger.info("CPU requests change is too small: {}%".format(diff_cpu))
        new_cpu = old_cpu
    else:
        _logger.info(
            "CPU requests change: {} -> {} ({}%)".format(
                str(round(old_cpu * 1000)) + "m",
                str(round(new_cpu * 1000)) + "m",
                diff_cpu,
            )
        )

    return float(new_cpu), not change_too_small


@beartype
def optimize_container_memory_requests(
    namespace_name: str,
    workload: str,
    container: V1Container,
    workload_type: str = "deployment",
    target_ratio: float = 1,
    lookback_minutes: int = DEFAULT_LOOKBACK_MINUTES,
) -> Tuple[int, bool]:
    """
    Optimize memory requests for a Kubernetes container.

    Args:
        namespace_name (str): The namespace of the workload.
        workload (str): The name of the workload.
        container (V1Container): The Kubernetes container object.
        workload_type (str, optional): The type of workload. Defaults to "deployment".
        target_ratio (float, optional): The target ratio for memory optimization. Defaults to 1.
        lookback_minutes (int, optional): The number of minutes to look back for resource usage data. Defaults to DEFAULT_LOOKBACK_MINUTES.

    Returns:
        int: The new memory request in bytes.
        bool: If something was changed

    Example:
        namespace_name = "my-namespace"
        workload = "my-workload"
        container = V1Container(name="my-container", resources=V1ResourceRequirements(requests={"memory": "1Gi"}))
        new_memory = optimize_container_memory_requests(namespace_name, workload, container)
    """
    container_name = container.name

    try:
        _logger.debug(container.resources.requests["memory"])
        old_memory = helpers.convert_memory_request_to_bytes(
            container.resources.requests["memory"]
        )
    except (KeyError, AttributeError):
        _logger.info("Could not read old meory requests aassuming it is 1")
        old_memory = 1

    new_memory = calculate_memory_requests(
        namespace_name,
        workload,
        workload_type,
        container_name,
        target_ratio,
        lookback_minutes,
    )
    diff_memory = round(((new_memory / old_memory) - 1) * 100)
    change_too_small = abs(diff_memory) < CHANGE_THRESHOLD * 100

    if change_too_small:
        _logger.info("Memory request change is too small: {}%".format(diff_memory))
        new_memory = old_memory
    else:
        _logger.info(
            "Memory requests change: {} -> {} ({}%)".format(
                str(round(old_memory / 1024 / 1024)) + "Mi",
                str(round(new_memory / 1024 / 1024)) + "Mi",
                diff_memory,
            )
        )
    return int(new_memory), not change_too_small


@beartype
def optimize_container_memory_limits(
    namespace_name: str,
    workload: str,
    container: V1Container,
    workload_type: str = "deployment",
    new_memory: int = MIN_MEMORY_REQUEST,
) -> Tuple[int, bool]:
    """
    Optimize memory limits for a Kubernetes container.

    Args:
        namespace_name (str): The namespace of the workload.
        workload (str): The name of the workload.
        container (V1Container): The Kubernetes container object.
        workload_type (str, optional): The type of workload. Defaults to "deployment".
        new_memory (int, optional): The new memory request in bytes. Defaults to MIN_MEMORY_REQUEST.

    Returns:
        int: The new memory limit in bytes.
        bool: If something was changed

    Example:
        namespace_name = "my-namespace"
        workload = "my-workload"
        container = V1Container(name="my-container", resources=V1ResourceRequirements(limits={"memory": "2Gi"}))
        new_memory_limit = optimize_container_memory_limits(namespace_name, workload, container)
    """
    container_name = container.name

    try:
        old_memory_limit = helpers.convert_memory_request_to_bytes(
            container.resources.limits["memory"]
        )
    except (KeyError, AttributeError):
        _logger.info("Could not read old meory limits aassuming it is 1")
        old_memory_limit = 1

    new_memory_limit = calculate_memory_limits(
        namespace_name, workload, workload_type, container_name, new_memory
    )
    diff_memory_limit = round(((new_memory_limit / old_memory_limit) - 1) * 100)

    change_too_small = abs(diff_memory_limit) < CHANGE_THRESHOLD * 100

    if change_too_small:
        _logger.info("Memory limit change is too small: {}%".format(diff_memory_limit))
        new_memory_limit = old_memory_limit
    else:
        _logger.info(
            "Memory limits change: {} -> {} ({}%)".format(
                str(round(old_memory_limit / 1024 / 1024)) + "Mi",
                str(round(new_memory_limit / 1024 / 1024)) + "Mi",
                diff_memory_limit,
            )
        )

    return int(new_memory_limit), change_too_small


def print_stats():
    if stats["old_cpu_sum"] > 0:
        diff_cpu_sum = round(((stats["new_cpu_sum"] / stats["old_cpu_sum"]) - 1) * 100)
        diff_memory_sum = round(
            ((stats["new_memory_sum"] / stats["old_memory_sum"]) - 1) * 100
        )

        _logger.info(
            "Summary cpu requests change: {} -> {} ({}%)".format(
                str(round(stats["old_cpu_sum"] * 1000)) + "m",
                str(round(stats["new_cpu_sum"] * 1000)) + "m",
                diff_cpu_sum,
            )
        )

        _logger.info(
            "Summary memory requests change: {} -> {} ({}%)".format(
                str(round(stats["old_memory_sum"] / 1024 / 1024)) + "Mi",
                str(round(stats["new_memory_sum"] / 1024 / 1024)) + "Mi",
                diff_memory_sum,
            )
        )


# ---- CLI ----
# The functions defined in this section are wrappers around the main Python
# API allowing them to be called directly from the terminal as a CLI
# executable/script.


def parse_args(args):
    """Parse command line parameters

    Args:
      args (List[str]): command line parameters as list of strings
          (for example  ``["--help"]``).

    Returns:
      :obj:`argparse.Namespace`: command line parameters namespace
    """
    parser = argparse.ArgumentParser(description="k8soptimizer")
    parser.add_argument(
        "--version",
        action="version",
        version=f"k8soptimizer {__version__}",
    )

    parser.add_argument(
        "--log",
        "--log-level",
        action="store",
        default=LOG_LEVEL,
        help="Set loglevel.",
        dest="loglevel",
    )

    parser.add_argument(
        "--log-format",
        action="store",
        default=LOG_FORMAT,
        help="Set logformat (txt, json).",
        dest="logformat",
    )

    parser.add_argument(
        "--dry-run",
        action="store_true",
        default=DRY_RUN_MODE,
        help="Perform a dry run without making any actual changes.",
        dest="dry_run",
    )

    group_ns = parser.add_mutually_exclusive_group()
    group_ns.add_argument(
        "-n",
        "--namespace",
        action="store",
        help="Set namespace",
        dest="namespace",
        type=helpers.valid_k8s_name_arg,
    )
    group_ns.add_argument(
        "--namespace-pattern",
        action="store",
        default=NAMESPACE_PATTERN,
        help="Set namespace pattern (regex)",
        dest="namespace_pattern",
        type=helpers.valid_regex_arg,
    )

    group_ds = parser.add_mutually_exclusive_group()
    group_ds.add_argument(
        "-d",
        "--deployment",
        action="store",
        help="Set deployment",
        dest="deployment",
        type=helpers.valid_k8s_name_arg,
    )
    parser.add_argument(
        "--deployment-pattern",
        action="store",
        default=DEPLOYMENT_PATTERN,
        help="Set deployment pattern (regex)",
        dest="deplopyment_pattern",
        type=helpers.valid_regex_arg,
    )

    group_cs = parser.add_mutually_exclusive_group()
    group_cs.add_argument(
        "-c",
        "--container",
        action="store",
        help="Set container",
        dest="container",
        type=helpers.is_valid_k8s_name,
    )
    group_cs.add_argument(
        "--container-pattern",
        action="store",
        default=DEPLOYMENT_PATTERN,
        help="Set container pattern (regex)",
        dest="container_pattern",
        type=helpers.valid_regex_arg,
    )

    return parser.parse_args(args)


def setup_logging(loglevel: str = "info", logformat: str = "json"):
    """Setup basic logging

    Args:
        loglevel (int): minimum loglevel for emitting messages
    """

    # logformat = "[%(asctime)s] %(levelname)s:%(name)s:%(message)s"
    # logging.basicConfig(
    #     level=loglevel, stream=sys.stdout, format=logformat, datefmt="%Y-%m-%d %H:%M:%S"
    # )

    logger = logging.getLogger()

    if logformat == "txt":
        log_handler = logging.StreamHandler(sys.stdout)
        formatter = logging.Formatter(
            "[%(asctime)s] %(levelname)s: %(name)s:%(message)s",
            datefmt="%Y-%m-%d %H:%M:%S",
        )
    elif logformat == "json":
        log_handler = logging.StreamHandler(sys.stdout)
        formatter = jsonlogger.JsonFormatter(
            "%(timestamp)s %(levelname)s %(message)s ", timestamp=True
        )
    else:
        raise ValueError("Invalid logformat. Use 'txt' or 'json'.")

    log_handler.setFormatter(formatter)
    logger.addHandler(log_handler)
    logger.setLevel(loglevel.upper())


def main(args):
    """Wrapper allowing :func:`fib` to be called with string arguments in a CLI fashion

    Instead of returning the value from :func:`fib`, it prints the result to the
    ``stdout`` in a nicely formatted message.

    Args:
        args (List[str]): command line parameters as list of strings
            (for example  ``["--verbose", "42"]``).
    """
    args = parse_args(args)
    setup_logging(args.loglevel, args.logformat)
    _logger.info("Starting k8soptimizer...")

    verify_kubernetes_connection()
    verify_prometheus_connection()

    namespace_pattern = args.namespace_pattern
    if args.namespace is not None:
        namespace_pattern = "^{}$".format(args.namespace)
    deplopyment_pattern = args.deplopyment_pattern
    if args.deployment is not None:
        deplopyment_pattern = "^{}$".format(args.deployment)
    container_pattern = args.container_pattern
    if args.container is not None:
        container_pattern = "^{}$".format(args.container)
    _logger.info("Using namespace_pattern: {}".format(namespace_pattern))
    _logger.info("Using deplopyment_pattern: {}".format(deplopyment_pattern))
    _logger.info("Using container_pattern: {}".format(container_pattern))

    for namespace in get_namespaces(namespace_pattern).items:
        extra = {"namespace": namespace.metadata.name}
        _logger.addFilter(AppFilter(extra))
        for deployment in get_deployments(
            namespace.metadata.name, deplopyment_pattern
        ).items:
            try:
                optimize_deployment(deployment, container_pattern, args.dry_run)
            except Exception as e:
                _logger.warning(
                    "An error occurred while optimizing the deployment: %s", str(e)
                )

    extra = {}
    _logger.addFilter(AppFilter(extra))

    print_stats()

    _logger.info("Finished k8soptimizer")


def run():
    """Calls :func:`main` passing the CLI arguments extracted from :obj:`sys.argv`

    This function can be used as entry point to create console scripts with setuptools.
    """
    main(sys.argv[1:])


if __name__ == "__main__":
    # ^  This is a guard statement that will prevent the following code from
    #    being executed in the case someone imports this file instead of
    #    executing it as a script.
    #    https://docs.python.org/3/library/__main__.html

    # After installing your project with pip, users can also run your Python
    # modules as scripts via the ``-m`` flag, as defined in PEP 338::
    #
    #     python -m k8soptimizer.skeleton 42
    #
    run()

src/k8soptimizer/main.py
arvatoaws-labs/k8soptimizer
"""
    Dummy conftest.py for k8soptimizer.

    If you don't know what this is for, just leave it empty.
    Read more about conftest.py under:
    - https://docs.pytest.org/en/stable/fixture.html
    - https://docs.pytest.org/en/stable/writing_plugins.html
"""

# import pytest

tests/conftest.py
arvatoaws-labs/k8soptimizer
import argparse
from datetime import datetime, timedelta, timezone

import pytest

import k8soptimizer.helpers as helpers

__author__ = "Philipp Hellmich"
__copyright__ = "Arvato Systems GmbH"
__license__ = "MIT"


def test_convert_memory_request_to_bytes():
    assert helpers.convert_memory_request_to_bytes("1B") == 1
    assert helpers.convert_memory_request_to_bytes("1K") == 1024
    assert helpers.convert_memory_request_to_bytes("1M") == 1024**2
    assert helpers.convert_memory_request_to_bytes("1G") == 1024**3
    assert helpers.convert_memory_request_to_bytes("1T") == 1024**4
    assert helpers.convert_memory_request_to_bytes("1Ki") == 1024
    assert helpers.convert_memory_request_to_bytes("1Mi") == 1024**2
    assert helpers.convert_memory_request_to_bytes("1Gi") == 1024**3
    assert helpers.convert_memory_request_to_bytes("1Ti") == 1024**4

    with pytest.raises(ValueError) as exc_info:
        helpers.convert_memory_request_to_bytes("hallo")
    assert str(exc_info.value).startswith("Invalid format")

    with pytest.raises(ValueError) as exc_info:
        helpers.convert_memory_request_to_bytes("1Jon")
    assert str(exc_info.value).startswith("Invalid unit")


def test_convert_cpu_request_to_cores():
    assert helpers.convert_cpu_request_to_cores("1000m") == 1
    assert helpers.convert_cpu_request_to_cores("1000k") == 1000000
    assert helpers.convert_cpu_request_to_cores("1") == 1
    assert helpers.convert_cpu_request_to_cores("0.5") == 0.5

    with pytest.raises(ValueError) as exc_info:
        helpers.convert_cpu_request_to_cores("1z")
    assert str(exc_info.value).startswith("Invalid unit")

    with pytest.raises(ValueError) as exc_info:
        helpers.convert_cpu_request_to_cores("OTTO")
    assert str(exc_info.value).startswith("Invalid format")


def test_format_pairs():
    assert helpers.format_pairs({"name": "otto"}) == "name=otto"
    assert helpers.format_pairs({"name": "otto", "age": 6}) == "name=otto, age=6"


def test_calculate_minutes_ago():
    # Create datetime objects from timestamp strings
    timestamp1 = datetime.now(timezone.utc) - timedelta(days=1)
    timestamp2 = datetime.now(timezone.utc) - timedelta(minutes=15)
    timestamp3 = datetime.now(timezone.utc)

    # Use the calculate_minutes_ago_from_timestamp function
    assert helpers.calculate_minutes_ago_from_timestamp(timestamp1) == 60 * 24
    assert helpers.calculate_minutes_ago_from_timestamp(timestamp2) == 15
    assert helpers.calculate_minutes_ago_from_timestamp(timestamp3) == 0


def test_calculate_minutes_ago_str():
    # Create datetime objects from timestamp strings
    timestamp1 = datetime.now(timezone.utc) - timedelta(days=1)
    timestamp2 = datetime.now(timezone.utc) - timedelta(minutes=15)
    timestamp3 = datetime.now(timezone.utc)

    # Use the calculate_minutes_ago_from_timestamp function
    assert (
        helpers.calculate_minutes_ago_from_timestamp_str(timestamp1.isoformat())
        == 60 * 24
    )
    assert (
        helpers.calculate_minutes_ago_from_timestamp_str(timestamp2.isoformat()) == 15
    )
    assert helpers.calculate_minutes_ago_from_timestamp_str(timestamp3.isoformat()) == 0


def test_create_timestamp_str():
    assert helpers.calculate_minutes_ago_from_timestamp(helpers.create_timestamp()) == 0


def test_is_valid_regex():
    assert helpers.is_valid_regex("(") is False
    assert helpers.is_valid_regex("^test$") is True


def test_valid_regex_arg():
    with pytest.raises(argparse.ArgumentTypeError) as exc_info:
        helpers.valid_regex_arg("(")
    assert str(exc_info.value).rfind("is not a valid regular expression") > 0
    assert helpers.valid_regex_arg("^test$") == "^test$"


def test_is_valid_k8s_name():
    assert helpers.is_valid_k8s_name("teststring" * 300) is False
    assert helpers.is_valid_k8s_name("(") is False
    assert helpers.is_valid_k8s_name("hallo") is True


def test_valid_k8s_name_arg():
    with pytest.raises(argparse.ArgumentTypeError) as exc_info:
        helpers.valid_k8s_name_arg("(")
    assert str(exc_info.value).rfind("is not a valid k8s object name") > 0
    assert helpers.valid_k8s_name_arg("hallo") == "hallo"

tests/test_helpers.py
arvatoaws-labs/k8soptimizer
import json
import unittest
from datetime import datetime, timedelta, timezone

# Standard library imports...
from unittest.mock import patch

import pytest
from kubernetes.client.models import (
    V1Container,
    V1Deployment,
    V1DeploymentList,
    V1DeploymentSpec,
    V1LabelSelector,
    V1Namespace,
    V1NamespaceList,
    V1ObjectMeta,
    V1PodSpec,
    V1PodTemplateSpec,
    V1ResourceRequirements,
    V2CrossVersionObjectReference,
    V2HorizontalPodAutoscaler,
    V2HorizontalPodAutoscalerList,
    V2HorizontalPodAutoscalerSpec,
    V2MetricSpec,
    V2MetricTarget,
    V2ResourceMetricSource,
)

import k8soptimizer.main as main

__author__ = "Philipp Hellmich"
__copyright__ = "Arvato Systems GmbH"
__license__ = "MIT"


@patch("requests.get")  # Mock the requests.get function
def test_query_prometheus(mock_requests_get):
    # Define your test data and expected response
    expected_result = {"data": {"result": [{"value": [0, 42]}]}}

    # Mock the response from requests.get
    mock_response = unittest.mock.Mock()
    mock_response.text = json.dumps(expected_result)
    mock_requests_get.return_value = mock_response

    # Call the function under test
    result = main.query_prometheus("node_load1")

    # Verify that the function behaves as expected
    assert result == expected_result  # Check if the result is as expected


@patch("requests.get")  # Mock the requests.get function
def test_verify_prometheus_connection(mock_requests_get):
    # Define your test data and expected response
    expected_result = {"status": "success"}

    # Mock the response from requests.get
    mock_response = unittest.mock.Mock()
    mock_response.text = json.dumps(expected_result)
    mock_requests_get.return_value = mock_response

    # Call the function under test
    result = main.verify_prometheus_connection()

    # Verify that the function behaves as expected
    assert result is True  # Check if the result is as expected


@patch("k8soptimizer.main.query_prometheus")
def test_get_max_pods_per_deployment_history(mock_func1):
    # Define your test data and expected response
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"

    mock_func1.return_value = {"data": {"result": [{"value": [0, 42]}]}}

    # Call the function under test
    result = main.get_max_pods_per_deployment_history(
        namespace_name,
        deployment_name,
        lookback_minutes=3600 * 7 * 24,
        quantile_over_time=0.95,
    )

    # Verify that the function behaves as expected
    assert (
        result == mock_func1.return_value["data"]["result"][0]["value"][1]
    )  # Check if the result is as expected

    mock_func1.return_value = {"data": {"result": []}}

    with pytest.raises(RuntimeError) as exc_info:
        main.get_max_pods_per_deployment_history(
            namespace_name,
            deployment_name,
            lookback_minutes=3600 * 24 * 7,
            quantile_over_time=0.95,
        )

    # Check the exception message or other attributes if needed
    assert "No data found" in str(exc_info.value)


@patch("k8soptimizer.main.query_prometheus")
def test_get_cpu_cores_usage_history(mock_func1):
    # Define your test data and expected response
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"

    mock_func1.return_value = {"data": {"result": [{"value": [0, 8]}]}}

    # Call the function under test
    result = main.get_cpu_cores_usage_history(
        namespace_name,
        deployment_name,
        "nginx",
        lookback_minutes=3600 * 7 * 24,
        quantile_over_time=0.95,
    )

    # Verify that the function behaves as expected
    assert (
        result == mock_func1.return_value["data"]["result"][0]["value"][1]
    )  # Check if the result is as expected

    mock_func1.return_value = {"data": {"result": []}}

    with pytest.raises(RuntimeError) as exc_info:
        main.get_cpu_cores_usage_history(
            namespace_name,
            deployment_name,
            "nginx",
            lookback_minutes=3600 * 24 * 7,
            quantile_over_time=0.95,
        )

    # Check the exception message or other attributes if needed
    assert "No data found" in str(exc_info.value)


@patch("k8soptimizer.main.query_prometheus")
def test_get_memory_bytes_usage_history(mock_func1):
    # Define your test data and expected response
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"

    mock_func1.return_value = {"data": {"result": [{"value": [0, 1024 * 8]}]}}

    # Call the function under test
    result = main.get_memory_bytes_usage_history(
        namespace_name,
        deployment_name,
        "nginx",
        lookback_minutes=3600 * 7 * 24,
        quantile_over_time=0.95,
    )

    # Verify that the function behaves as expected
    assert (
        result == mock_func1.return_value["data"]["result"][0]["value"][1]
    )  # Check if the result is as expected

    mock_func1.return_value = {"data": {"result": []}}

    with pytest.raises(RuntimeError) as exc_info:
        main.get_memory_bytes_usage_history(
            namespace_name,
            deployment_name,
            "nginx",
            lookback_minutes=3600 * 24 * 7,
            quantile_over_time=0.95,
        )

    # Check the exception message or other attributes if needed
    assert "No data found" in str(exc_info.value)


@patch("k8soptimizer.main.query_prometheus")
def test_discover_container_runtime(mock_func1):
    # Define your test data and expected response
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"

    mock_func1.return_value = {"data": {"result": [{"value": [0, 1]}]}}

    # Call the function under test
    result = main.discover_container_runtime(namespace_name, deployment_name, "nodejs")

    # Verify that the function behaves as expected
    assert result == "nodejs"  # Check if the result is as expected

    mock_func1.return_value = {"data": {"result": [{"value": [0, 0]}]}}

    # Call the function under test
    result = main.discover_container_runtime(namespace_name, deployment_name, "nginx")

    # Verify that the function behaves as expected
    assert result is None  # Check if the result is as expected


@patch(
    "k8soptimizer.main.client.CoreV1Api.list_namespace"
)  # Mock the requests.get function
def test_get_namespaces(mock_requests_get):
    # Define a list of V1Namespace objects
    namespace1 = V1Namespace(metadata=V1ObjectMeta(name="namespace1"))
    namespace2 = V1Namespace(metadata=V1ObjectMeta(name="namespace2"))
    namespace_list = V1NamespaceList(items=[namespace1, namespace2])

    mock_requests_get.return_value = namespace_list

    # Call the function under test
    result = main.get_namespaces(".*")

    # Verify that the function behaves as expected
    assert len(result.items) == 2  # Check if the result is as expected

    # Define a list of V1Namespace objects
    namespace_list = V1NamespaceList(items=[namespace1])

    mock_requests_get.return_value = namespace_list

    # Call the function under test
    result = main.get_namespaces("namespace1")

    assert len(result.items) == 1  # Check if the result is as expected
    assert (
        result.items[0].metadata.name == "namespace1"
    )  # Check if the result is as expected

    # Call the function under test
    result = main.get_namespaces("namespaceX")

    assert len(result.items) == 0  # Check if the result is as expected


@patch(
    "k8soptimizer.main.client.AppsV1Api.list_namespaced_deployment"
)  # Mock the requests.get function
def test_get_deployments(mock_requests_get):
    # Define a list of V1Namespace objects
    deployment1 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment1",
            namespace="default",
        ),
        spec=V1DeploymentSpec(
            replicas=1,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "1"}, limits={"cpu": "1"}
                            ),
                        )
                    ]
                )
            ),
        ),
    )
    deployment2 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment2",
            namespace="default",
        ),
        spec=V1DeploymentSpec(
            replicas=2,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(limits={"cpu": "2"}),
                        )
                    ]
                )
            ),
        ),
    )
    deployment3 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment3",
        ),
        spec=V1DeploymentSpec(
            replicas=0,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(limits={"cpu": "1"}),
                        )
                    ]
                )
            ),
        ),
    )
    deployment_list = V1DeploymentList(items=[deployment1, deployment2, deployment3])

    mock_requests_get.return_value = deployment_list

    # Call the function under test
    result = main.get_deployments("default", ".*")

    # Verify that the function behaves as expected
    assert len(result.items) == 2  # Check if the result is as expected

    # Define a list of V1Namespace objects
    deployment_list = V1DeploymentList(items=[deployment1, deployment2])

    mock_requests_get.return_value = deployment_list

    # Call the function under test
    result = main.get_deployments("default", "^deployment1$")

    assert len(result.items) == 1  # Check if the result is as expected
    assert (
        result.items[0].metadata.name == "deployment1"
    )  # Check if the result is as expected

    # Call the function under test
    result = main.get_deployments("default", "deploymentX")

    assert len(result.items) == 0  # Check if the result is as expected


@patch(
    "k8soptimizer.main.client.AutoscalingV2Api.list_namespaced_horizontal_pod_autoscaler"
)  # Mock the requests.get function
def test_get_hpa_for_deployment(mock_requests_get):
    hpa1 = V2HorizontalPodAutoscaler(
        metadata=V1ObjectMeta(name="deployment1"),
        spec=V2HorizontalPodAutoscalerSpec(
            max_replicas=10,
            min_replicas=1,
            metrics=[
                V2MetricSpec(
                    type="Resource",
                    resource=V2ResourceMetricSource(
                        name="cpu",
                        target=V2MetricTarget(
                            average_utilization=80, type="Utilization"
                        ),
                    ),
                )
            ],
            scale_target_ref=V2CrossVersionObjectReference(
                kind="Deployment", name="deployment1"
            ),
        ),
    )

    hpa2 = V2HorizontalPodAutoscaler(
        metadata=V1ObjectMeta(name="deployment2"),
        spec=V2HorizontalPodAutoscalerSpec(
            max_replicas=10,
            min_replicas=1,
            metrics=[
                V2MetricSpec(
                    type="Resource",
                    resource=V2ResourceMetricSource(
                        name="cpu",
                        target=V2MetricTarget(
                            average_utilization=80, type="Utilization"
                        ),
                    ),
                )
            ],
            scale_target_ref=V2CrossVersionObjectReference(
                kind="Deployment", name="deployment2"
            ),
        ),
    )
    hpa_list = V2HorizontalPodAutoscalerList(items=[hpa1, hpa2])

    mock_requests_get.return_value = hpa_list

    # Call the function under test
    result = main.get_hpa_for_deployment("default", "deployment1")

    assert result is not None

    assert result.metadata.name == "deployment1"


test_data_hpa = [
    # Test case 0: Normal case
    {
        "input_params": {
            "avg_cpu": 100,
            "avg_memory": 100,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 3,
        },
        "expected_output": {"ratio_cpu": 1.0, "ratio_memory": 1.0},
    },
    # Test case 1: Expected to raise CPU ratio to match HPA
    {
        "input_params": {
            "avg_cpu": 80,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 3,
        },
        "expected_output": {"ratio_cpu": 1.25, "ratio_memory": 1},
    },
    # Test case 2: Expected to raise memory ratio to match HPA
    {
        "input_params": {
            "avg_memory": 80,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 3,
        },
        "expected_output": {"ratio_cpu": 1.0, "ratio_memory": 1.25},
    },
    # Test case 3: Expected to raise both CPU and memory ratio to match HPA
    {
        "input_params": {
            "avg_cpu": 80,
            "avg_memory": 80,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 3,
        },
        "expected_output": {"ratio_cpu": 1.25, "ratio_memory": 1.25},
    },
    # Test case 4: Expected to raise CPU because of max_replicas reached
    {
        "input_params": {
            "avg_cpu": 100,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 10,
        },
        "expected_output": {"ratio_cpu": 2.0, "ratio_memory": 1.0},
    },
    # Test case 5: Expected to raise CPU because is almost max_replicas reached
    {
        "input_params": {
            "avg_cpu": 100,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 9,
        },
        "expected_output": {"ratio_cpu": 1.89, "ratio_memory": 1.0},
    },
    # Test case 6: Expected to raise CPU ratio because max_replicas is reached
    {
        "input_params": {
            "avg_cpu": 100,
            "avg_memory": 100,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 3,
            "pod_oom_history": 5,
        },
        "expected_output": {"ratio_cpu": 1.0, "ratio_memory": 1.0},
    },
    # Test case 7: Expected to raise CPU and memory ratio because almost max_replicas is reached and ooms werde found
    {
        "input_params": {
            "avg_cpu": 100,
            "avg_memory": 100,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 3,
            "pod_oom_history": 5,
        },
        "expected_output": {"ratio_cpu": 1.0, "ratio_memory": 1.0},
    },
    # Test case 8: Expected to raise both CPU and memory ratio to match HPA even if history is 0
    {
        "input_params": {
            "avg_cpu": 80,
            "avg_memory": 80,
            "min_replicas": 1,
            "max_replicas": 10,
            "pod_replica_history": 0,
            "pod_oom_history": 0,
        },
        "expected_output": {"ratio_cpu": 1.25, "ratio_memory": 1.25},
    },
]


@pytest.mark.parametrize("test_case", test_data_hpa)
@patch("k8soptimizer.main.get_oom_killed_history")
@patch("k8soptimizer.main.get_max_pods_per_deployment_history")
@patch("k8soptimizer.main.get_hpa_for_deployment")
def test_calculate_hpa_target_ratio(mock_func1, mock_func2, mock_func3, test_case):
    # Extract input parameters and expected output from the test case
    input_params = test_case["input_params"]
    expected_output = test_case["expected_output"]

    metrics = []
    if "avg_memory" in input_params:
        metrics.append(
            V2MetricSpec(
                type="Resource",
                resource=V2ResourceMetricSource(
                    name="memory",
                    target=V2MetricTarget(
                        average_utilization=input_params["avg_memory"],
                        type="Utilization",
                    ),
                ),
            )
        )
    if "avg_cpu" in input_params:
        metrics.append(
            V2MetricSpec(
                type="Resource",
                resource=V2ResourceMetricSource(
                    name="cpu",
                    target=V2MetricTarget(
                        average_utilization=input_params["avg_cpu"], type="Utilization"
                    ),
                ),
            )
        )

    hpa1 = V2HorizontalPodAutoscaler(
        metadata=V1ObjectMeta(name="deployment1"),
        spec=V2HorizontalPodAutoscalerSpec(
            max_replicas=input_params["max_replicas"],
            min_replicas=input_params["min_replicas"],
            metrics=metrics,
            scale_target_ref=V2CrossVersionObjectReference(
                kind="Deployment", name="deployment1"
            ),
        ),
    )

    mock_func1.return_value = hpa1
    mock_func2.return_value = input_params["pod_replica_history"]
    mock_func3.return_value = 0
    if "pod_oom_history" in input_params:
        mock_func3.return_value = input_params["pod_oom_history"]

    # Call the function under test
    result = main.calculate_hpa_target_ratio("default", "deployment1")

    assert result["cpu"] == pytest.approx(expected_output["ratio_cpu"], rel=1e-2)
    assert result["memory"] == pytest.approx(expected_output["ratio_memory"], rel=1e-2)


test_data_cpu = [
    # Test case 0: Normal case
    {
        "input_params": {
            "cpu_history": main.MIN_CPU_REQUEST + 1,
            "cpu_ratio": 1.0,
            "runtime": None,
        },
        "expected_output": main.MIN_CPU_REQUEST + 1,
    },
    # Test case 1: Below min cpu
    {
        "input_params": {"cpu_history": 0.000001, "cpu_ratio": 1.0, "runtime": None},
        "expected_output": main.MIN_CPU_REQUEST,
    },
    # Test case 2: Below min cpu
    {
        "input_params": {"cpu_history": 1, "cpu_ratio": 0.000001, "runtime": None},
        "expected_output": main.MIN_CPU_REQUEST,
    },
    # Test case 3: Higher max cpu
    {
        "input_params": {"cpu_history": 9999999999, "cpu_ratio": 1.0, "runtime": None},
        "expected_output": main.MAX_CPU_REQUEST,
    },
    # Test case 4: Higher max cpu
    {
        "input_params": {"cpu_history": 1, "cpu_ratio": 9999999999.0, "runtime": None},
        "expected_output": main.MAX_CPU_REQUEST,
    },
    # Test case 5: nodejs
    {
        "input_params": {
            "cpu_history": 10,
            "cpu_ratio": 9999999999.0,
            "runtime": "nodejs",
        },
        "expected_output": main.MAX_CPU_REQUEST_NODEJS,
    },
]


@pytest.mark.parametrize("test_case", test_data_cpu)
@patch("k8soptimizer.main.discover_container_runtime")
@patch("k8soptimizer.main.get_cpu_cores_usage_history")
def test_calculate_cpu_requests(mock_func1, mock_func2, test_case):
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"
    input_params = test_case["input_params"]
    expected_output = test_case["expected_output"]

    mock_func1.return_value = input_params["cpu_history"]
    mock_func2.return_value = None

    input_params = test_case["input_params"]
    expected_output = test_case["expected_output"]

    if "runtime" in input_params:
        mock_func2.return_value = input_params["runtime"]

    # Call the function under test
    result = main.calculate_cpu_requests(
        namespace_name,
        deployment_name,
        "development",
        "nginx",
        input_params["cpu_ratio"],
        main.DEFAULT_LOOKBACK_MINUTES,
    )

    assert result == pytest.approx(expected_output, rel=1e-2)


test_data_memory = [
    # Test case 0: Normal case
    {
        "input_params": {
            "memory_history": main.MIN_MEMORY_REQUEST + 1024,
            "memory_ratio": 1.0,
            "runtime": None,
        },
        "expected_output": main.MIN_MEMORY_REQUEST + 1024,
    },
    # Test case 1: Below min memory
    {
        "input_params": {
            "memory_history": 0.000001,
            "memory_ratio": 1.0,
            "runtime": None,
        },
        "expected_output": main.MIN_MEMORY_REQUEST,
    },
    # Test case 2: Below min memory
    {
        "input_params": {
            "memory_history": 1,
            "memory_ratio": 000000.1,
            "runtime": None,
        },
        "expected_output": main.MIN_MEMORY_REQUEST,
    },
    # Test case 3: Higher max memory
    {
        "input_params": {
            "memory_history": 999999999999999,
            "memory_ratio": 1.0,
            "runtime": None,
        },
        "expected_output": main.MAX_MEMORY_REQUEST,
    },
    # Test case 4: Higher max memory
    {
        "input_params": {
            "memory_history": 1,
            "memory_ratio": 999999999999999.0,
            "runtime": None,
        },
        "expected_output": main.MAX_MEMORY_REQUEST,
    },
    # Test case 5: nodejs
    {
        "input_params": {
            "memory_history": 10,
            "memory_ratio": 999999999999999.0,
            "runtime": "nodejs",
        },
        "expected_output": main.MAX_MEMORY_REQUEST,
    },
    # Test case 6: OOM killed
    {
        "input_params": {
            "memory_history": main.MIN_MEMORY_REQUEST + 1024,
            "memory_ratio": 1.0,
            "runtime": None,
            "oom_killed": 11,
        },
        "expected_output": (main.MIN_MEMORY_REQUEST + 1024) * 2,
    },
]


@pytest.mark.parametrize("test_case", test_data_memory)
@patch("k8soptimizer.main.get_oom_killed_history")
@patch("k8soptimizer.main.discover_container_runtime")
@patch("k8soptimizer.main.get_memory_bytes_usage_history")
def test_calculate_memory_requests(mock_func1, mock_func2, mock_func3, test_case):
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"
    input_params = test_case["input_params"]
    expected_output = test_case["expected_output"]

    mock_func1.return_value = input_params["memory_history"]
    mock_func2.return_value = None
    mock_func3.return_value = 0

    if "runtime" in input_params:
        mock_func2.return_value = input_params["runtime"]
    if "oom_killed" in input_params:
        mock_func3.return_value = input_params["oom_killed"]

    # Call the function under test
    result = main.calculate_memory_requests(
        namespace_name,
        deployment_name,
        "development",
        "nginx",
        input_params["memory_ratio"],
        main.DEFAULT_LOOKBACK_MINUTES,
    )

    assert result == pytest.approx(expected_output, rel=1e-2)


test_data_optimize_container = [
    # Test case 0: Normal case with cpu and memory limits
    {
        "input_params": {
            "old_requests": {"cpu": "6", "memory": "1G"},
            "old_limits": {"cpu": "1", "memory": "10G"},
            "new_requests": {"cpu": 3, "memory": 1024**3 * 2},
            "new_limits": {"cpu": 3, "memory": 1024**3 * 4},
        },
        "expected_output": {
            "requests": {"cpu": "3000m", "memory": "2048Mi"},
            "limits": {"memory": "4096Mi"},
        },
    },
    # Test case 1: Normal case with no limits
    {
        "input_params": {
            "old_requests": {"cpu": "6", "memory": "1G"},
            "old_limits": {},
            "new_requests": {"cpu": 3, "memory": 1024**3 * 2},
            "new_limits": {"cpu": 3, "memory": 1024**3 * 4},
        },
        "expected_output": {
            "requests": {"cpu": "3000m", "memory": "2048Mi"},
            "limits": {"memory": "4096Mi"},
        },
    },
    # Test case 2: Normal case with no limits
    {
        "input_params": {
            "old_requests": {"cpu": "6", "memory": "1G"},
            "old_limits": {},
            "new_requests": {"cpu": 1.0, "memory": 1024**3 * 2},
            "new_limits": {"memory": 1024**3 * 4},
        },
        "expected_output": {
            "requests": {"cpu": "1000m", "memory": "2048Mi"},
            "limits": {"memory": "4096Mi"},
        },
    },
]


@pytest.mark.parametrize("test_case", test_data_optimize_container)
@patch("k8soptimizer.main.calculate_memory_limits")
@patch("k8soptimizer.main.calculate_memory_requests")
@patch("k8soptimizer.main.calculate_cpu_requests")
def test_optimize_container(mock_func1, mock_func2, mock_func3, test_case):
    expected_output = test_case["expected_output"]

    mock_func1.return_value = test_case["input_params"]["new_requests"]["cpu"]
    mock_func2.return_value = test_case["input_params"]["new_requests"]["memory"]
    mock_func3.return_value = test_case["input_params"]["new_limits"]["memory"]

    container = V1Container(
        name="nginx",
        resources=V1ResourceRequirements(
            requests=test_case["input_params"]["old_requests"],
            limits=test_case["input_params"]["old_limits"],
        ),
    )

    container, changed = main.optimize_container(
        "default",
        "deployment1",
        container,
        "deployment",
        1.0,
        1.0,
        main.DEFAULT_LOOKBACK_MINUTES,
    )

    assert container.resources.requests == expected_output["requests"]
    assert container.resources.limits == expected_output["limits"]
    assert changed


def test_get_resources_from_deployment():
    # Define a list of V1Namespace objects
    deployment1 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment1",
        ),
        spec=V1DeploymentSpec(
            replicas=1,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "1"}, limits={"cpu": "1"}
                            ),
                        ),
                        V1Container(
                            name="php",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "8", "memory": "6Gi"}, limits={}
                            ),
                        ),
                        V1Container(name="php-monitor"),
                    ]
                )
            ),
        ),
    )

    expected_result = {
        "nginx": {
            "requests": deployment1.spec.template.spec.containers[0].resources.requests,
            "limits": deployment1.spec.template.spec.containers[0].resources.limits,
        },
        "php": {
            "requests": deployment1.spec.template.spec.containers[1].resources.requests,
            "limits": {},
        },
        "php-monitor": {"requests": {}, "limits": {}},
    }
    result = main.get_resources_from_deployment(deployment1)

    for key, value in expected_result.items():
        assert result[key] == value

    # assert result == expected_result


@patch("k8soptimizer.main.client.AppsV1Api.patch_namespaced_deployment")
@patch("k8soptimizer.main.calculate_lookback_minutes_from_deployment")
@patch("k8soptimizer.main.calculate_hpa_target_ratio")
@patch("k8soptimizer.main.optimize_container")
def test_optimize_deployment(mock_func1, mock_func2, mock_func3, mock_func4):
    deployment1_input = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment1",
            namespace="default",
            creation_timestamp=datetime.now(timezone.utc) - timedelta(days=1),
            annotations={},
        ),
        spec=V1DeploymentSpec(
            replicas=1,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "1"}, limits={"cpu": "1"}
                            ),
                        ),
                        V1Container(
                            name="php",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "2", "memory": "4Gi"}, limits={}
                            ),
                        ),
                        V1Container(name="php-monitor"),
                    ]
                )
            ),
        ),
    )

    deployment1_output = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment1",
            namespace="default",
            creation_timestamp=datetime.now(timezone.utc) - timedelta(days=1),
            annotations={},
        ),
        spec=V1DeploymentSpec(
            replicas=1,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "1", "memory": "2Gi"},
                                limits={"memory": "4Gi"},
                            ),
                        ),
                        V1Container(
                            name="php",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "4", "memory": "8Gi"}, limits={}
                            ),
                        ),
                        V1Container(name="php-monitor"),
                    ]
                )
            ),
        ),
    )

    mock_func1.return_value = deployment1_output, True
    mock_func2.return_value = {"cpu": 2, "memory": 2}
    mock_func3.return_value = 60
    mock_func4.return_value = True

    result = main.optimize_deployment(deployment1_input)

    assert (
        "k8soptimizer.arvato-aws.io/old-resources" in result.metadata.annotations.keys()
    )
    assert (
        "k8soptimizer.arvato-aws.io/last-update" in result.metadata.annotations.keys()
    )


def test_parse_args_version(capsys):
    args = ["--version"]
    with pytest.raises(SystemExit) as excinfo:
        main.parse_args(args)

    captured = capsys.readouterr()
    assert "k8soptimizer" in captured.out
    assert excinfo.value.code == 0


@patch("k8soptimizer.main.optimize_deployment")
@patch("k8soptimizer.main.verify_kubernetes_connection")
@patch("k8soptimizer.main.verify_prometheus_connection")
@patch("k8soptimizer.main.get_deployments")
@patch("k8soptimizer.main.get_namespaces")
def test_main(mock_func1, mock_func2, mock_func3, mock_func4, mock_func5):
    # Define a list of V1Namespace objects
    namespace1 = V1Namespace(metadata=V1ObjectMeta(name="namespace1"))
    namespace2 = V1Namespace(metadata=V1ObjectMeta(name="namespace2"))
    namespace_list = V1NamespaceList(items=[namespace1, namespace2])

    # Define a list of V1Namespace objects
    deployment1 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment1",
            namespace="default",
        ),
        spec=V1DeploymentSpec(
            replicas=1,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "1"}, limits={"cpu": "1"}
                            ),
                        )
                    ]
                )
            ),
        ),
    )
    deployment2 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment2",
            namespace="default",
        ),
        spec=V1DeploymentSpec(
            replicas=2,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(limits={"cpu": "2"}),
                        )
                    ]
                )
            ),
        ),
    )
    deployment3 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment3",
        ),
        spec=V1DeploymentSpec(
            replicas=0,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(limits={"cpu": "1"}),
                        )
                    ]
                )
            ),
        ),
    )
    deployment_list = V1DeploymentList(items=[deployment1, deployment2, deployment3])

    mock_func1.return_value = namespace_list
    mock_func2.return_value = deployment_list
    mock_func3.return_value = True
    mock_func4.return_value = True
    mock_func5.return_value = True

    main.stats["old_cpu_sum"] = 100
    main.stats["new_cpu_sum"] = 150
    main.stats["old_memory_sum"] = 1000
    main.stats["new_memory_sum"] = 2000

    main.main([])


@patch("k8soptimizer.main.query_prometheus")
def test_get_oom_killed_history(mock_func1):
    # Define your test data and expected response
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"
    container_name = "test_container"

    mock_func1.return_value = {"data": {"result": [{"value": [0, 1]}]}}

    # Call the function under test
    result = main.get_oom_killed_history(
        namespace_name, deployment_name, container_name
    )

    # Verify that the function behaves as expected
    assert result == 1  # Check if the result is as expected

    mock_func1.return_value = {"data": {"result": [{"value": [0, 0]}]}}

    # Call the function under test
    result = main.get_oom_killed_history(
        namespace_name, deployment_name, container_name
    )

    # Verify that the function behaves as expected
    assert result == 0  # Check if the result is as expected


# calculate_lookback_minutes_from_deployment


@patch("k8soptimizer.main.get_number_of_samples_from_history")
def test_calculate_lookback_minutes_from_deployment(mock_func1):
    # Define a list of V1Namespace objects
    deployment1 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment1",
            namespace="default",
            creation_timestamp=datetime.now(timezone.utc) - timedelta(days=1),
        ),
        spec=V1DeploymentSpec(
            replicas=1,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(
                                requests={"cpu": "1"}, limits={"cpu": "1"}
                            ),
                        )
                    ]
                )
            ),
        ),
    )

    deployment2 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment2",
            namespace="default",
            creation_timestamp=datetime.now(timezone.utc),
        ),
        spec=V1DeploymentSpec(
            replicas=2,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(limits={"cpu": "2"}),
                        )
                    ]
                )
            ),
        ),
    )

    deployment3 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment3",
            annotations={
                "k8soptimizer.arvato-aws.io/last-update": (
                    datetime.now(timezone.utc) - timedelta(days=1)
                ).isoformat()
            },
            creation_timestamp=datetime.now(timezone.utc) - timedelta(days=1),
        ),
        spec=V1DeploymentSpec(
            replicas=0,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(limits={"cpu": "1"}),
                        )
                    ]
                )
            ),
        ),
    )

    deployment4 = V1Deployment(
        metadata=V1ObjectMeta(
            name="deployment4",
            annotations={
                "k8soptimizer.arvato-aws.io/last-update": (
                    datetime.now(timezone.utc)
                ).isoformat()
            },
            creation_timestamp=datetime.now(timezone.utc) - timedelta(days=1),
        ),
        spec=V1DeploymentSpec(
            replicas=0,
            selector=V1LabelSelector(match_labels={"app": "nginx"}),
            template=V1PodTemplateSpec(
                spec=V1PodSpec(
                    containers=[
                        V1Container(
                            name="nginx",
                            resources=V1ResourceRequirements(limits={"cpu": "1"}),
                        )
                    ]
                )
            ),
        ),
    )

    mock_func1.return_value = 60

    assert main.calculate_lookback_minutes_from_deployment(deployment1) == 60

    with pytest.raises(RuntimeError) as exc_info:
        main.calculate_lookback_minutes_from_deployment(deployment2)
    assert str(exc_info.value).startswith("The deployment is too young")

    assert main.calculate_lookback_minutes_from_deployment(deployment3) == 60

    with pytest.raises(RuntimeError) as exc_info:
        main.calculate_lookback_minutes_from_deployment(deployment4)
    assert str(exc_info.value).startswith("The deployment was optimized too recently")

    mock_func1.return_value = 0

    with pytest.raises(RuntimeError) as exc_info:
        main.calculate_lookback_minutes_from_deployment(deployment3)
    assert str(exc_info.value).rfind("history samples") > 0

    mock_func1.return_value = 999999

    main.calculate_lookback_minutes_from_deployment(
        deployment3
    ) == main.MAX_LOOKBACK_MINUTES


test_data_optimize_container = [
    # Test case 0: Normal case
    {
        "request": main.MIN_MEMORY_REQUEST + 1,
        "limit": round((main.MIN_MEMORY_REQUEST + 1) * main.MEMORY_LIMIT_RATIO),
    },
    # Test case 1: Too low values
    {"request": 1024**2 * 1, "limit": main.MIN_MEMORY_LIMIT},
    # Test case 2: Too high values
    {"request": 1024**3 * 999999, "limit": main.MAX_MEMORY_LIMIT},
    # Test case 3: Too high values
    {"request": main.MAX_MEMORY_LIMIT + 1024, "limit": main.MAX_MEMORY_LIMIT},
]


@pytest.mark.parametrize("test_case", test_data_optimize_container)
def test_calculate_memory_limits(test_case):
    # Define your test data and expected response
    namespace_name = "test_namespace"
    deployment_name = "test_deployment"

    # Call the function under test
    result = main.calculate_memory_limits(
        namespace_name, deployment_name, "development", "nginx", test_case["request"]
    )

    # Verify that the function behaves as expected
    assert result == test_case["limit"]

tests/test_main.py
arvatoaws-labs/k8soptimizer