Merge pull request #199 from appsignal/implement-minutely-probes

Implement minutely probes

.changesets/add-minutely-probes-system.md

@@ -0,0 +1,22 @@
+---
+bump: "minor"
+type: "add"
+---
+
+Add a minutely probes system. This can be used, alongside our metric helpers, to report metrics to AppSignal once per minute.
+
+```python
+from appsignal import probes, set_gauge
+
+def new_carts(previous_carts=None):
+    current_carts = Cart.objects.all().count()
+
+    if previous_carts is not None:
+      set_gauge("new_carts", current_carts - previous_carts)
+
+    return current_carts
+
+probes.register("new_carts", new_carts)
+```
+
+The minutely probes system starts by default, but no probes are automatically registered. You can use the `enable_minutely_probes` configuration option to disable it.

conftest.py

@@ -15,14 +15,15 @@
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 from opentelemetry.trace import set_tracer_provider
 
+from appsignal import probes
 from appsignal.agent import agent
 from appsignal.opentelemetry import METRICS_PREFERRED_TEMPORALITY
 
 
 @pytest.fixture(scope="function", autouse=True)
 def disable_start_opentelemetry(mocker):
-    mocker.patch("appsignal.opentelemetry._start_opentelemetry_tracer")
-    mocker.patch("appsignal.opentelemetry._start_opentelemetry_metrics")
+    mocker.patch("appsignal.opentelemetry._start_tracer")
+    mocker.patch("appsignal.opentelemetry._start_metrics")
 
 
 @pytest.fixture(scope="session", autouse=True)
@@ -86,6 +87,14 @@ def remove_logging_handlers_after_tests():
         logger.removeHandler(handler)
 
 
+@pytest.fixture(scope="function", autouse=True)
+def stop_and_clear_probes_after_tests():
+    yield
+
+    probes.stop()
+    probes.clear()
+
+
 @pytest.fixture(scope="function", autouse=True)
 def reset_agent_active_state():
     agent.active = False

src/appsignal/client.py

@@ -7,7 +7,8 @@
 
 from .agent import agent
 from .config import Config, Options
-from .opentelemetry import start_opentelemetry
+from .opentelemetry import start as start_opentelemetry
+from .probes import start as start_probes
 
 
 if TYPE_CHECKING:
@@ -28,7 +29,7 @@ class Client:
 
     def __init__(self, **options: Unpack[Options]) -> None:
         self._config = Config(options)
-        self.start_logger()
+        self._start_logger()
 
         if not self._config.is_active():
             self._logger.info("AppSignal not starting: no active config found")
@@ -38,8 +39,13 @@ def start(self) -> None:
             self._logger.info("Starting AppSignal")
             agent.start(self._config)
             start_opentelemetry(self._config)
+            self._start_probes()
 
-    def start_logger(self) -> None:
+    def _start_probes(self) -> None:
+        if self._config.option("enable_minutely_probes"):
+            start_probes()
+
+    def _start_logger(self) -> None:
         self._logger = logging.getLogger("appsignal")
         self._logger.setLevel(self.LOG_LEVELS[self._config.option("log_level")])
 

src/appsignal/config.py

@@ -32,6 +32,7 @@ class Options(TypedDict, total=False):
     )
     dns_servers: list[str] | None
     enable_host_metrics: bool | None
+    enable_minutely_probes: bool | None
     enable_nginx_metrics: bool | None
     enable_statsd: bool | None
     endpoint: str | None
@@ -80,6 +81,7 @@ class Config:
         ca_file_path=CA_FILE_PATH,
         diagnose_endpoint="https://appsignal.com/diag",
         enable_host_metrics=True,
+        enable_minutely_probes=True,
         enable_nginx_metrics=False,
         enable_statsd=False,
         environment=DEFAULT_ENVIRONMENT,
@@ -156,6 +158,9 @@ def load_from_environment() -> Options:
             enable_host_metrics=parse_bool(
                 os.environ.get("APPSIGNAL_ENABLE_HOST_METRICS")
             ),
+            enable_minutely_probes=parse_bool(
+                os.environ.get("APPSIGNAL_ENABLE_MINUTELY_PROBES")
+            ),
             enable_nginx_metrics=parse_bool(
                 os.environ.get("APPSIGNAL_ENABLE_NGINX_METRICS")
             ),

src/appsignal/opentelemetry.py

@@ -108,7 +108,7 @@ def add_requests_instrumentation() -> None:
 }
 
 
-def start_opentelemetry(config: Config) -> None:
+def start(config: Config) -> None:
     # Configure OpenTelemetry request headers config
     request_headers = list_to_env_str(config.option("request_headers"))
     if request_headers:
@@ -117,13 +117,13 @@ def start_opentelemetry(config: Config) -> None:
         )
 
     opentelemetry_port = config.option("opentelemetry_port")
-    _start_opentelemetry_tracer(opentelemetry_port)
-    _start_opentelemetry_metrics(opentelemetry_port)
+    _start_tracer(opentelemetry_port)
+    _start_metrics(opentelemetry_port)
 
     add_instrumentations(config)
 
 
-def _start_opentelemetry_tracer(opentelemetry_port: str | int) -> None:
+def _start_tracer(opentelemetry_port: str | int) -> None:
     otlp_exporter = OTLPSpanExporter(
         endpoint=f"http://localhost:{opentelemetry_port}/v1/traces"
     )
@@ -143,7 +143,7 @@ def _start_opentelemetry_tracer(opentelemetry_port: str | int) -> None:
 }
 
 
-def _start_opentelemetry_metrics(opentelemetry_port: str | int) -> None:
+def _start_metrics(opentelemetry_port: str | int) -> None:
     metric_exporter = OTLPMetricExporter(
         endpoint=f"http://localhost:{opentelemetry_port}/v1/metrics",
         preferred_temporality=METRICS_PREFERRED_TEMPORALITY,

src/appsignal/probes.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import logging
+from inspect import signature
+from threading import Event, Lock, Thread
+from time import gmtime
+from typing import Any, Callable, Optional, TypeVar, Union, cast
+
+
+T = TypeVar("T")
+
+Probe = Union[Callable[[], None], Callable[[Optional[T]], Optional[T]]]
+
+_probes: dict[str, Probe] = {}
+_probe_states: dict[str, Any] = {}
+_lock: Lock = Lock()
+_thread: Thread | None = None
+_stop_event: Event = Event()
+
+
+def start() -> None:
+    global _thread
+    if _thread is None:
+        _thread = Thread(target=_minutely_loop, daemon=True)
+        _thread.start()
+
+
+def _minutely_loop() -> None:
+    wait_time = _initial_wait_time()
+
+    while True:
+        if _stop_event.wait(timeout=wait_time):
+            break
+
+        _run_probes()
+        wait_time = _wait_time()
+
+
+def _run_probes() -> None:
+    with _lock:
+        for name in _probes:
+            _run_probe(name)
+
+
+def _run_probe(name: str) -> None:
+    logger = logging.getLogger("appsignal")
+    logger.debug(f"Gathering minutely metrics with `{name}` probe")
+
+    try:
+        probe = _probes[name]
+
+        if len(signature(probe).parameters) > 0:
+            probe = cast(Callable[[Any], Any], probe)
+            state = _probe_states.get(name)
+            result = probe(state)
+            _probe_states[name] = result
+        else:
+            probe = cast(Callable[[], None], probe)
+            probe()
+
+    except Exception as e:
+        logger.debug(f"Error in minutely probe `{name}`: {e}")
+
+
+def _wait_time() -> int:
+    return 60 - gmtime().tm_sec
+
+
+def _initial_wait_time() -> int:
+    remaining_seconds = _wait_time()
+    if remaining_seconds > 30:
+        return remaining_seconds
+
+    return remaining_seconds + 60
+
+
+def register(name: str, probe: Probe) -> None:
+    with _lock:
+        if name in _probes:
+            logger = logging.getLogger("appsignal")
+            logger.debug(
+                f"A probe with the name `{name}` is already "
+                "registered. Overwriting the entry with the new probe."
+            )
+
+        _probes[name] = probe
+
+
+def unregister(name: str) -> None:
+    with _lock:
+        if name in _probes:
+            del _probes[name]
+        if name in _probe_states:
+            del _probe_states[name]
+
+
+def stop() -> None:
+    global _thread
+    if _thread is not None:
+        _stop_event.set()
+        _thread.join()
+        _thread = None
+        _stop_event.clear()
+
+
+def clear() -> None:
+    with _lock:
+        _probes.clear()
+        _probe_states.clear()