frostfs-testlib/src/frostfs_testlib/load/load_metrics.py

from abc import ABC
from typing import Any, Optional

from frostfs_testlib.load.load_config import LoadScenario


class OperationMetric(ABC):
    _NAME = ""
    _SUCCESS = ""
    _ERRORS = ""
    _THROUGHPUT = ""
    _LATENCY = ""

    def __init__(self, summary) -> None:
        self.summary = summary
        self.metrics = summary["metrics"]

    @property
    def total_iterations(self) -> int:
        return self._get_metric(self._SUCCESS) + self._get_metric(self._ERRORS)

    @property
    def success_iterations(self) -> int:
        return self._get_metric(self._SUCCESS)

    @property
    def latency(self) -> dict:
        return self._get_metric(self._LATENCY)

    @property
    def rate(self) -> float:
        return self._get_metric_rate(self._SUCCESS)

    @property
    def failed_iterations(self) -> int:
        return self._get_metric(self._ERRORS)

    @property
    def throughput(self) -> float:
        return self._get_metric_rate(self._THROUGHPUT)

    def _get_metric(self, metric: str) -> int:
        metrics_method_map = {
            "counter": self._get_counter_metric,
            "gauge": self._get_gauge_metric,
            "trend": self._get_trend_metrics,
        }

        if metric not in self.metrics:
            return 0

        metric = self.metrics[metric]
        metric_type = metric["type"]
        if metric_type not in metrics_method_map:
            raise Exception(f"Unsupported metric type: {metric_type}, supported: {metrics_method_map.keys()}")

        return metrics_method_map[metric_type](metric)

    def _get_metric_rate(self, metric: str) -> int:
        metrics_method_map = {"counter": self._get_counter_metric_rate}

        if metric not in self.metrics:
            return 0

        metric = self.metrics[metric]
        metric_type = metric["type"]
        if metric_type not in metrics_method_map:
            raise Exception(f"Unsupported rate metric type: {metric_type}, supported: {metrics_method_map.keys()}")

        return metrics_method_map[metric_type](metric)

    def _get_counter_metric_rate(self, metric: str) -> int:
        return metric["values"]["rate"]

    def _get_counter_metric(self, metric: str) -> int:
        return metric["values"]["count"]

    def _get_gauge_metric(self, metric: str) -> int:
        return metric["values"]["value"]

    def _get_trend_metrics(self, metric: str) -> int:
        return metric["values"]


class WriteOperationMetric(OperationMetric):
    _NAME = "Write"
    _SUCCESS = ""
    _ERRORS = ""
    _THROUGHPUT = "data_sent"
    _LATENCY = ""


class ReadOperationMetric(OperationMetric):
    _NAME = "Read"
    _SUCCESS = ""
    _ERRORS = ""
    _THROUGHPUT = "data_received"
    _LATENCY = ""


class DeleteOperationMetric(OperationMetric):
    _NAME = "Delete"
    _SUCCESS = ""
    _ERRORS = ""
    _THROUGHPUT = ""
    _LATENCY = ""


class GrpcWriteOperationMetric(WriteOperationMetric):
    _SUCCESS = "frostfs_obj_put_total"
    _ERRORS = "frostfs_obj_put_fails"
    _LATENCY = "frostfs_obj_put_duration"


class GrpcReadOperationMetric(ReadOperationMetric):
    _SUCCESS = "frostfs_obj_get_total"
    _ERRORS = "frostfs_obj_get_fails"
    _LATENCY = "frostfs_obj_get_duration"


class GrpcDeleteOperationMetric(DeleteOperationMetric):
    _SUCCESS = "frostfs_obj_delete_total"
    _ERRORS = "frostfs_obj_delete_fails"
    _LATENCY = "frostfs_obj_delete_duration"


class S3WriteOperationMetric(WriteOperationMetric):
    _SUCCESS = "aws_obj_put_total"
    _ERRORS = "aws_obj_put_fails"
    _LATENCY = "aws_obj_put_duration"


class S3ReadOperationMetric(ReadOperationMetric):
    _SUCCESS = "aws_obj_get_total"
    _ERRORS = "aws_obj_get_fails"
    _LATENCY = "aws_obj_get_duration"


class S3DeleteOperationMetric(DeleteOperationMetric):
    _SUCCESS = "aws_obj_delete_total"
    _ERRORS = "aws_obj_delete_fails"
    _LATENCY = "aws_obj_delete_duration"


class S3LocalWriteOperationMetric(WriteOperationMetric):
    _SUCCESS = "s3local_obj_put_total"
    _ERRORS = "s3local_obj_put_fails"
    _LATENCY = "s3local_obj_put_duration"


class S3LocalReadOperationMetric(ReadOperationMetric):
    _SUCCESS = "s3local_obj_get_total"
    _ERRORS = "s3local_obj_get_fails"
    _LATENCY = "s3local_obj_get_duration"


class LocalWriteOperationMetric(WriteOperationMetric):
    _SUCCESS = "local_obj_put_total"
    _ERRORS = "local_obj_put_fails"
    _LATENCY = "local_obj_put_duration"


class LocalReadOperationMetric(ReadOperationMetric):
    _SUCCESS = "local_obj_get_total"
    _ERRORS = "local_obj_get_fails"


class LocalDeleteOperationMetric(DeleteOperationMetric):
    _SUCCESS = "local_obj_delete_total"
    _ERRORS = "local_obj_delete_fails"


class VerifyReadOperationMetric(ReadOperationMetric):
    _SUCCESS = "verified_obj"
    _ERRORS = "invalid_obj"


class MetricsBase(ABC):
    def __init__(self) -> None:
        self.write: Optional[WriteOperationMetric] = None
        self.read: Optional[ReadOperationMetric] = None
        self.delete: Optional[DeleteOperationMetric] = None

    @property
    def operations(self) -> list[OperationMetric]:
        return [metric for metric in [self.write, self.read, self.delete] if metric is not None]


class GrpcMetrics(MetricsBase):
    def __init__(self, summary) -> None:
        super().__init__()
        self.write = GrpcWriteOperationMetric(summary)
        self.read = GrpcReadOperationMetric(summary)
        self.delete = GrpcDeleteOperationMetric(summary)


class S3Metrics(MetricsBase):
    def __init__(self, summary) -> None:
        super().__init__()
        self.write = S3WriteOperationMetric(summary)
        self.read = S3ReadOperationMetric(summary)
        self.delete = S3DeleteOperationMetric(summary)


class S3LocalMetrics(MetricsBase):
    def __init__(self, summary) -> None:
        super().__init__()
        self.write = S3LocalWriteOperationMetric(summary)
        self.read = S3LocalReadOperationMetric(summary)


class LocalMetrics(MetricsBase):
    def __init__(self, summary) -> None:
        super().__init__()
        self.write = LocalWriteOperationMetric(summary)
        self.read = LocalReadOperationMetric(summary)
        self.delete = LocalDeleteOperationMetric(summary)


class VerifyMetrics(MetricsBase):
    def __init__(self, summary) -> None:
        super().__init__()
        self.read = VerifyReadOperationMetric(summary)


def get_metrics_object(load_type: LoadScenario, summary: dict[str, Any]) -> MetricsBase:
    class_map = {
        LoadScenario.gRPC: GrpcMetrics,
        LoadScenario.gRPC_CAR: GrpcMetrics,
        LoadScenario.HTTP: GrpcMetrics,
        LoadScenario.S3: S3Metrics,
        LoadScenario.S3_CAR: S3Metrics,
        LoadScenario.S3_MULTIPART: S3Metrics,
        LoadScenario.S3_LOCAL: S3LocalMetrics,
        LoadScenario.VERIFY: VerifyMetrics,
        LoadScenario.LOCAL: LocalMetrics,
    }

    return class_map[load_type](summary)