Added integration with Sentry

pghoard/__init__.py

@@ -4,4 +4,3 @@
 Copyright (c) 2016 Ohmu Ltd
 See LICENSE for details
 """
-from . import mapping, monitoring

pghoard/basebackup/delta.py

@@ -154,7 +154,7 @@ def _delta_upload_hexdigest(
         result_hash = hashlib.blake2s()
 
         def progress_callback(n_bytes: int = 1) -> None:
-            self.metrics.increase("pghoard.basebackup_bytes_uploaded", inc_value=n_bytes, tags={"delta": True})
+            self.metrics.increase("pghoard.basebackup_bytes_uploaded", inc_value=n_bytes, tags={"delta": "True"})
 
         with NamedTemporaryFile(dir=temp_dir, prefix=os.path.basename(chunk_path), suffix=".tmp") as raw_output_obj:
             raw_output_file = cast(FileLikeWithName, raw_output_obj)

pghoard/metrics.py

@@ -2,37 +2,47 @@
 Interface for monitoring clients
 
 """
-import pghoard
+import logging
+from dataclasses import dataclass
+from typing import Dict, Optional, Type
 
+from pghoard.monitoring import (PrometheusClient, PushgatewayClient, SentryClient, StatsClient)
+from pghoard.monitoring.base import MetricsClient
+
+LOG = logging.getLogger(__name__)
 
-class Metrics:
-    def __init__(self, **configs):
-        self.clients = self._init_clients(configs)
 
-    def _init_clients(self, configs):
-        clients = {}
+@dataclass()
+class AvailableClient:
+    client_class: Type[MetricsClient]
+    config_key: str
 
-        if not isinstance(configs, dict):
-            return clients
 
-        map_client = pghoard.mapping.clients
-        for k, config in configs.items():
-            if isinstance(config, dict) and k in map_client:
-                path, classname = map_client[k]
-                mod = __import__(path, fromlist=[classname])
-                klass = getattr(mod, classname)
-                clients[k] = klass(config)
+class Metrics:
+    available_clients = [
+        AvailableClient(StatsClient, "statsd"),
+        AvailableClient(PrometheusClient, "prometheus"),
+        AvailableClient(PushgatewayClient, "pushgateway"),
+        AvailableClient(SentryClient, "sentry"),
+    ]
+
+    def __init__(self, **configs):
+        self.clients = {}
 
-        return clients
+        for client_info in self.available_clients:
+            client_config = configs.get(client_info.config_key)
+            if isinstance(client_config, dict):
+                LOG.info("Initializing monitoring client %s", client_info.config_key)
+                self.clients[client_info.config_key] = client_info.client_class(client_config)
 
-    def gauge(self, metric, value, tags=None):
+    def gauge(self, metric: str, value: float, tags: Optional[Dict[str, str]] = None) -> None:
         for client in self.clients.values():
             client.gauge(metric, value, tags)
 
-    def increase(self, metric, inc_value=1, tags=None):
+    def increase(self, metric: str, inc_value: int = 1, tags: Optional[Dict[str, str]] = None) -> None:
         for client in self.clients.values():
             client.increase(metric, inc_value, tags)
 
-    def unexpected_exception(self, ex, where, tags=None):
+    def unexpected_exception(self, ex: Exception, where: str, tags: Optional[Dict[str, str]] = None) -> None:
         for client in self.clients.values():
             client.unexpected_exception(ex, where, tags)

pghoard/monitoring/__init__.py

@@ -1,5 +1,5 @@
-import pkgutil
-
-__path__ = pkgutil.extend_path(__path__, __name__)  # type: ignore
-for importer, modname, ispkg in pkgutil.walk_packages(path=__path__, prefix=__name__ + "."):
-    __import__(modname)
+# Copyright (c) 2024 Aiven, Helsinki, Finland. https://aiven.io/
+from .prometheus import PrometheusClient
+from .pushgateway import PushgatewayClient
+from .sentry import SentryClient
+from .statsd import StatsClient

pghoard/monitoring/base.py

@@ -0,0 +1,16 @@
+# Copyright (c) 2024 Aiven, Helsinki, Finland. https://aiven.io/
+from typing import Any, Dict, Optional
+
+
+class MetricsClient:
+    def __init__(self, config: Dict[str, Any]):
+        self.config = config
+
+    def gauge(self, metric: str, value: float, tags: Optional[Dict[str, str]] = None) -> None:
+        pass
+
+    def increase(self, metric: str, inc_value: int = 1, tags: Optional[Dict[str, str]] = None) -> None:
+        pass
+
+    def unexpected_exception(self, ex: Exception, where: str, tags: Optional[Dict[str, str]] = None) -> None:
+        pass

pghoard/monitoring/prometheus.py

@@ -5,9 +5,12 @@
 
 import time
 
+from pghoard.monitoring.base import MetricsClient
 
-class PrometheusClient:
+
+class PrometheusClient(MetricsClient):
     def __init__(self, config):
+        super().__init__(config)
         self._tags = config.get("tags", {})
         self.metrics = {}
 

pghoard/monitoring/pushgateway.py

@@ -6,9 +6,12 @@
 
 import requests
 
+from pghoard.monitoring.base import MetricsClient
 
-class PushgatewayClient:
+
+class PushgatewayClient(MetricsClient):
     def __init__(self, config):
+        super().__init__(config)
         self._endpoint = config.get("endpoint", "")
         self._job = config.get("job", "pghoard")
         self._instance = config.get("instance", "")

pghoard/monitoring/sentry.py

@@ -0,0 +1,46 @@
+# Copyright (c) 2024 Aiven, Helsinki, Finland. https://aiven.io/
+import logging
+from typing import Any, Dict, Optional
+
+from pghoard.monitoring.base import MetricsClient
+
+LOG = logging.getLogger(__name__)
+
+
+class SentryClient(MetricsClient):
+    def __init__(self, config: Dict[str, Any]):
+        super().__init__(config)
+        self.sentry = None
+        if config is None:
+            LOG.info("Sentry configuration not found, skipping setup")
+            return
+        dsn = config.get("dsn")
+        if dsn is None:
+            LOG.info("Sentry DSN not found, skipping setup")
+            return
+        try:
+            import sentry_sdk
+            from sentry_sdk.integrations.logging import LoggingIntegration
+        except ImportError:
+            LOG.info("Sentry SDK not found, skipping setup")
+            return
+        self.sentry = sentry_sdk
+        sentry_logging = LoggingIntegration(
+            level=logging.INFO,
+            event_level=logging.CRITICAL,
+        )
+        tags = config.pop("tags", {})
+        sentry_sdk.init(**config, integrations=[sentry_logging])
+        for key, value in tags.items():
+            sentry_sdk.set_tag(key, value)
+
+    def unexpected_exception(self, ex: Exception, where: str, tags: Optional[Dict[str, str]] = None) -> None:
+        if not self.sentry:
+            return
+
+        with self.sentry.push_scope() as scope:
+            scope.set_tag("where", where)
+            if tags and isinstance(tags, dict):
+                for key, value in tags.items():
+                    scope.set_tag(key, value)
+            self.sentry.capture_exception(ex)

pghoard/monitoring/statsd.py

@@ -8,9 +8,12 @@
 """
 import socket
 
+from pghoard.monitoring.base import MetricsClient
 
-class StatsClient:
+
+class StatsClient(MetricsClient):
     def __init__(self, config):
+        super().__init__(config)
         self._dest_addr = (config.get("host", "127.0.0.1"), config.get("port", 8125))
         self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self._tags = config.get("tags", {})

pghoard/pghoard.py

@@ -976,7 +976,8 @@ def load_config(self, _signal=None, _frame=None):  # pylint: disable=unused-argu
         self.metrics = metrics.Metrics(
             statsd=self.config.get("statsd", None),
             pushgateway=self.config.get("pushgateway", None),
-            prometheus=self.config.get("prometheus", None)
+            prometheus=self.config.get("prometheus", None),
+            sentry=self.config.get("sentry", None)
         )
 
         # need to refresh the web server config too

pyproject.toml

@@ -64,6 +64,7 @@ dev = [
     "types-python-dateutil",
     "types-requests",
     "types-six",
+    "sentry-sdk",
 ]
 constraints = [
     "astroid==2.5.8",

test/monitoring/test_sentry.py

@@ -0,0 +1,50 @@
+import contextlib
+import logging
+
+import sentry_sdk
+
+from pghoard.monitoring.sentry import SentryClient
+
+
+@contextlib.contextmanager
+def patch_sentry_init(events):
+    original_init = sentry_sdk.init
+
+    def patched_init(*args, **kwargs):
+        kwargs["transport"] = events.append
+        kwargs.pop("dsn", None)
+        return original_init(*args, **kwargs)
+
+    sentry_sdk.init = patched_init
+    try:
+        yield
+    finally:
+        sentry_sdk.init = original_init
+
+
+def test_missing_config():
+    client = SentryClient(config=None)
+    client.unexpected_exception(ValueError("hello !"), where="tests")
+    client.gauge("something", 123.456)
+    client.increase("something")
+
+
+def test_exception_send():
+    events = []
+    with patch_sentry_init(events):
+        client = SentryClient(config={"dsn": "http://localhost:9000", "tags": {"foo": "bar"}})
+    client.unexpected_exception(ValueError("hello !"), where="tests")
+    assert len(events) == 1
+
+
+def test_logging_integration():
+    events = []
+    with patch_sentry_init(events):
+        SentryClient(config={"dsn": "http://localhost:9000", "tags": {"foo": "bar"}})
+
+    logging.warning("Info")
+    assert len(events) == 0
+    logging.error("Error")
+    assert len(events) == 0
+    logging.critical("Critical")
+    assert len(events) == 1