feat(app): long term metric storage (#173)

Co-authored-by: Ralf Grubenmann <[email protected]>

Pipfile

@@ -13,6 +13,8 @@ expiringdict = "*"
 jinja2 = "*"
 pytz = "*"
 prometheus_client = "*"
+dataconf = "*"
+boto3 = "*"
 
 [dev-packages]
 black = "*"

controller/config.py

@@ -2,7 +2,7 @@
 import os
 import yaml
 
-from controller.utils import sanitize_prometheus_metric_label_name
+from controller.config_types import AuditlogConfig, PrometheusMetricsConfig
 
 api_group = os.getenv("CRD_API_GROUP", "amalthea.dev")
 api_version = os.getenv("CRD_API_VERSION", "v1alpha1")
@@ -72,9 +72,5 @@
 CHILD_KEY_LABEL_KEY = f"{api_group}/child-key"
 MAIN_POD_LABEL_KEY = f"{api_group}/main-pod"
 
-METRICS_ENABLED = os.environ.get("METRICS_ENABLED", "false").lower() == "true"
-METRICS_EXTRA_LABELS = json.loads(os.environ.get("METRICS_EXTRA_LABELS", "[]"))
-METRICS_EXTRA_LABELS_SANITIZED = tuple([
-    sanitize_prometheus_metric_label_name(i) for i in METRICS_EXTRA_LABELS
-])
-METRICS_PORT = int(os.environ.get("METRICS_PORT", 8765))
+METRICS = PrometheusMetricsConfig.dataconf_from_env()
+AUDITLOG = AuditlogConfig.dataconf_from_env()

controller/config_types.py

@@ -0,0 +1,62 @@
+from dataclasses import dataclass, field
+import dataconf
+import json
+from typing import Optional, Union, List
+
+
+@dataclass
+class S3Config:
+    """The configuration needed to upload metrics to S3."""
+    endpoint: str
+    bucket: str
+    path_prefix: str
+    access_key_id: str
+    secret_access_key: str
+    rotation_period_seconds: Union[str, int] = 86400
+
+    def __post_init__(self):
+        if type(self.rotation_period_seconds) is str:
+            self.rotation_period_seconds = int(self.rotation_period_seconds)
+
+
+@dataclass
+class MetricsBaseConfig:
+    """Base metrics/auditlog configuration."""
+    enabled: Union[str, bool] = False
+    extra_labels: Union[str, List[str]] = field(default_factory=list)
+
+    def __post_init__(self):
+        if type(self.enabled) is str:
+            self.enabled = self.enabled.lower() == "true"
+        if type(self.extra_labels) is str:
+            self.extra_labels = json.loads(self.extra_labels)
+
+
+@dataclass
+class AuditlogConfig(MetricsBaseConfig):
+    """The configuration used for the auditlogs."""
+    s3: Optional[S3Config] = None
+
+    def __post_init__(self):
+        super().__post_init__()
+        if self.enabled and not self.s3:
+            raise ValueError("If auditlog is enabled then the S3 configuration has to be provided.")
+
+    @classmethod
+    def dataconf_from_env(cls, prefix="AUDITLOG_"):
+        return dataconf.env(prefix, cls)
+
+
+@dataclass
+class PrometheusMetricsConfig(MetricsBaseConfig):
+    """The configuration for prometheus metrics"""
+    port: Union[str, int] = 8765
+
+    def __post_init__(self):
+        super().__post_init__()
+        if type(self.port) is str:
+            self.port = int(self.port)
+
+    @classmethod
+    def dataconf_from_env(cls, prefix="METRICS_"):
+        return dataconf.env(prefix, cls)

controller/metrics/events.py

@@ -0,0 +1,45 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from datetime import datetime
+from dateutil import parser
+from typing import Any, Dict, Optional
+
+from controller.server_status_enum import ServerStatusEnum
+
+
+@dataclass
+class MetricEvent:
+    """Every element in the metrics queue that is created by
+    amalthea and consumed by the metrics handlers conforms to this
+    structure."""
+    event_timestamp: datetime
+    session: Dict[str, Any]
+    sessionCreationTimestamp: Optional[datetime] = None
+    old_status: Optional[ServerStatusEnum] = None
+    status: Optional[ServerStatusEnum] = None
+
+    def __post_init__(self):
+        if self.status and type(self.status) is str:
+            self.status = ServerStatusEnum(self.status)
+        if self.old_status and type(self.old_status) is str:
+            self.old_status = ServerStatusEnum(self.old_status)
+        if self.session.get("metadata", {}).get("creationTimestamp"):
+            self.sessionCreationTimestamp = parser.isoparse(
+                self.session.get("metadata", {}).get("creationTimestamp")
+            )
+
+    def __repr__(self) -> str:
+        return (
+            f"MetricEvent(event_timestamp={self.event_timestamp}, old_status={self.old_status}, "
+            f"status={self.status}, sessionCreationTimestamp={self.sessionCreationTimestamp}, "
+            "session=<redacted>)"
+        )
+
+
+class MetricEventHandler(ABC):
+    """Abstract class for the queue workers that will
+    be doing the final publishing or persisting of any metrics
+    that are generated by amalthea in the metrics queue."""
+    @abstractmethod
+    def publish(self, metric_event: MetricEvent):
+        pass