added prometheus exporter

backend/prometheus_metrics.py

@@ -0,0 +1,96 @@
+from prometheus_client import Counter, generate_latest, CONTENT_TYPE_LATEST, CollectorRegistry
+from typing import Dict, Tuple
+import json
+import os
+from models import DrinkType, MateType
+
+
+class DrinkSalesMetrics:
+    def __init__(self, stats_file: str = "drink_stats.json"):
+        self.stats_file = stats_file
+        self.stats: Dict[Tuple[str, str], int] = self._load_stats()
+        
+        # Erstelle eine eigene Registry ohne Standard-Metriken
+        self.registry = CollectorRegistry()
+        
+        # Prometheus Counter für verkaufte Getränke
+        self.drinks_sold_counter = Counter(
+            'tschunk_drinks_sold_total',
+            'Total number of drinks sold',
+            ['drink_type', 'mate_type'],
+            registry=self.registry
+        )
+        
+        # Initialisiere Counter mit gespeicherten Werten
+        self._initialize_counters()
+    
+    def _load_stats(self) -> Dict[Tuple[str, str], int]:
+        """Lädt die gespeicherten Statistiken aus der JSON-Datei."""
+        if os.path.exists(self.stats_file):
+            try:
+                with open(self.stats_file, 'r', encoding='utf-8') as f:
+                    data = json.load(f)
+                    # Konvertiere String-Tupel zurück zu Tuple
+                    return {tuple(k.split('|')): v for k, v in data.items()}
+            except (json.JSONDecodeError, KeyError, ValueError) as e:
+                print(f"Fehler beim Laden der Statistiken: {e}")
+                return {}
+        return {}
+    
+    def _save_stats(self):
+        """Speichert die Statistiken in eine JSON-Datei."""
+        try:
+            # Konvertiere Tuple zu String für JSON-Serialisierung
+            data = {f"{k[0]}|{k[1]}": v for k, v in self.stats.items()}
+            with open(self.stats_file, 'w', encoding='utf-8') as f:
+                json.dump(data, f, indent=2, ensure_ascii=False)
+        except Exception as e:
+            print(f"Fehler beim Speichern der Statistiken: {e}")
+    
+    def _initialize_counters(self):
+        """Initialisiert die Prometheus-Counter mit den gespeicherten Werten."""
+        for (drink_type, mate_type), count in self.stats.items():
+            # Setze den Counter auf den gespeicherten Wert
+            # Da Counter nur inkrementiert werden können, müssen wir das manuell machen
+            for _ in range(count):
+                self.drinks_sold_counter.labels(
+                    drink_type=drink_type,
+                    mate_type=mate_type
+                ).inc()
+    
+    def record_drinks(self, drinks):
+        """Zeichnet verkaufte Getränke auf und aktualisiert die Statistiken."""
+        for drink in drinks:
+            drink_type = drink.drink_type.value
+            mate_type = drink.mate_type.value
+            quantity = drink.quantity
+            
+            # Aktualisiere lokale Statistiken
+            key = (drink_type, mate_type)
+            self.stats[key] = self.stats.get(key, 0) + quantity
+            
+            # Aktualisiere Prometheus-Counter
+            self.drinks_sold_counter.labels(
+                drink_type=drink_type,
+                mate_type=mate_type
+            ).inc(quantity)
+        
+        # Speichere Statistiken persistent
+        self._save_stats()
+    
+    def get_stats(self) -> Dict[str, Dict[str, int]]:
+        """Gibt die aktuellen Statistiken zurück."""
+        result = {}
+        for (drink_type, mate_type), count in self.stats.items():
+            if drink_type not in result:
+                result[drink_type] = {}
+            result[drink_type][mate_type] = count
+        return result
+    
+    def get_metrics(self):
+        """Gibt die Prometheus-Metriken zurück."""
+        return generate_latest(self.registry)
+
+
+# Globale Instanz der Metriken
+drink_metrics = DrinkSalesMetrics()