Merge pull request #227 from appsignal/implement-scheduler-and-heartb…

…eat-checkins

Implement scheduler and heartbeat checkins

.changesets/implement-heartbeat-checkins.md

@@ -0,0 +1,26 @@
+---
+bump: minor
+type: add
+---
+
+Add support for heartbeat check-ins.
+
+Use the `appsignal.check_in.heartbeat` function to send a single heartbeat check-in event from your application. This can be used, for example, in your application's main loop:
+
+```python
+from appsignal.check_in import heartbeat
+
+while True:
+  heartbeat("job_processor")
+  process_job()
+```
+
+Heartbeats are deduplicated and sent asynchronously, without blocking the current thread. Regardless of how often the `.heartbeat` function is called, at most one heartbeat with the same identifier will be sent every ten seconds.
+
+Pass `continuous=True` as the second argument to send heartbeats continuously during the entire lifetime of the current process. This can be used, for example, after your application has finished its boot process:
+
+```python
+def main():
+  start_app()
+  heartbeat("my_app", continuous=True)
+```

.changesets/send-checkins-concurrently.md

@@ -0,0 +1,6 @@
+---
+bump: patch
+type: change
+---
+
+Send check-ins concurrently. When calling `appsignal.check_in.cron`, instead of blocking the current thread while the check-in events are sent, schedule them to be sent in a separate thread.

conftest.py

@@ -3,33 +3,39 @@
 import os
 import platform
 import tempfile
+from typing import Any, Callable, Generator
 
 import pytest
 from opentelemetry.metrics import set_meter_provider
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.metrics.export import InMemoryMetricReader
 from opentelemetry.sdk.resources import Resource
-from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
 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.check_in.heartbeat import (
+    _kill_continuous_heartbeats,
+    _reset_heartbeat_continuous_interval_seconds,
+)
+from appsignal.check_in.scheduler import _reset_scheduler
 from appsignal.client import _reset_client
 from appsignal.heartbeat import _heartbeat_class_warning, _heartbeat_helper_warning
 from appsignal.internal_logger import _reset_logger
 from appsignal.opentelemetry import METRICS_PREFERRED_TEMPORALITY
 
 
 @pytest.fixture(scope="function", autouse=True)
-def disable_start_opentelemetry(mocker):
+def disable_start_opentelemetry(mocker: Any) -> Any:
     mocker.patch("appsignal.opentelemetry._start_tracer")
     mocker.patch("appsignal.opentelemetry._start_metrics")
 
 
 @pytest.fixture(scope="session", autouse=True)
-def start_in_memory_metric_reader():
+def start_in_memory_metric_reader() -> Generator[InMemoryMetricReader, None, None]:
     metric_reader = InMemoryMetricReader(
         preferred_temporality=METRICS_PREFERRED_TEMPORALITY
     )
@@ -40,7 +46,7 @@ def start_in_memory_metric_reader():
 
 
 @pytest.fixture(scope="session", autouse=True)
-def start_in_memory_span_exporter():
+def start_in_memory_span_exporter() -> Generator[InMemorySpanExporter, None, None]:
     span_exporter = InMemorySpanExporter()
     exporter_processor = SimpleSpanProcessor(span_exporter)
     provider = TracerProvider()
@@ -51,18 +57,22 @@ def start_in_memory_span_exporter():
 
 
 @pytest.fixture(scope="function")
-def metrics(start_in_memory_metric_reader):
+def metrics(
+    start_in_memory_metric_reader: InMemoryMetricReader,
+) -> Generator[Callable[[], Any], None, None]:
     # Getting the metrics data implicitly wipes its state
     start_in_memory_metric_reader.get_metrics_data()
 
     yield start_in_memory_metric_reader.get_metrics_data
 
 
 @pytest.fixture(scope="function")
-def spans(start_in_memory_span_exporter):
+def spans(
+    start_in_memory_span_exporter: InMemorySpanExporter,
+) -> Generator[Callable[[], tuple[ReadableSpan, ...]], None, None]:
     start_in_memory_span_exporter.clear()
 
-    def get_and_clear_spans():
+    def get_and_clear_spans() -> tuple[ReadableSpan, ...]:
         spans = start_in_memory_span_exporter.get_finished_spans()
         start_in_memory_span_exporter.clear()
         return spans
@@ -71,7 +81,7 @@ def get_and_clear_spans():
 
 
 @pytest.fixture(scope="function", autouse=True)
-def reset_environment_between_tests():
+def reset_environment_between_tests() -> Any:
     old_environ = dict(os.environ)
 
     yield
@@ -81,40 +91,49 @@ def reset_environment_between_tests():
 
 
 @pytest.fixture(scope="function", autouse=True)
-def reset_internal_logger_after_tests():
+def reset_internal_logger_after_tests() -> Any:
     yield
 
     _reset_logger()
 
 
 @pytest.fixture(scope="function", autouse=True)
-def stop_and_clear_probes_after_tests():
+def stop_and_clear_probes_after_tests() -> Any:
     yield
 
     probes.stop()
     probes.clear()
 
 
 @pytest.fixture(scope="function", autouse=True)
-def reset_agent_active_state():
+def reset_agent_active_state() -> Any:
     agent.active = False
 
 
 @pytest.fixture(scope="function", autouse=True)
-def reset_global_client():
+def reset_global_client() -> Any:
     _reset_client()
 
 
 @pytest.fixture(scope="function", autouse=True)
-def stop_agent():
+def reset_checkins() -> Any:
+    yield
+
+    _reset_heartbeat_continuous_interval_seconds()
+    _kill_continuous_heartbeats()
+    _reset_scheduler()
+
+
+@pytest.fixture(scope="function", autouse=True)
+def stop_agent() -> Any:
     tmp_path = "/tmp" if platform.system() == "Darwin" else tempfile.gettempdir()
     working_dir = os.path.join(tmp_path, "appsignal")
     if os.path.isdir(working_dir):
         os.system(f"rm -rf {working_dir}")
 
 
 @pytest.fixture(scope="function")
-def reset_heartbeat_warnings():
+def reset_heartbeat_warnings() -> Any:
     _heartbeat_class_warning.reset()
     _heartbeat_helper_warning.reset()
 

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
   "opentelemetry-api>=1.26.0",
   "opentelemetry-sdk>=1.26.0",
   "opentelemetry-exporter-otlp-proto-http",
+  "typing-extensions"
 ]
 dynamic = ["version"]
 

src/appsignal/check_in/__init__.py

@@ -0,0 +1,5 @@
+from .cron import Cron, cron
+from .heartbeat import heartbeat
+
+
+__all__ = ["Cron", "cron", "heartbeat"]

src/appsignal/check_in/cron.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from binascii import hexlify
+from os import urandom
+from typing import Any, Callable, Literal, TypeVar
+
+from .event import cron as cron_event
+from .scheduler import scheduler
+
+
+T = TypeVar("T")
+
+
+class Cron:
+    identifier: str
+    digest: str
+
+    def __init__(self, identifier: str) -> None:
+        self.identifier = identifier
+        self.digest = hexlify(urandom(8)).decode("utf-8")
+
+    def start(self) -> None:
+        scheduler().schedule(cron_event(self.identifier, self.digest, "start"))
+
+    def finish(self) -> None:
+        scheduler().schedule(cron_event(self.identifier, self.digest, "finish"))
+
+    def __enter__(self) -> None:
+        self.start()
+
+    def __exit__(
+        self, exc_type: Any = None, exc_value: Any = None, traceback: Any = None
+    ) -> Literal[False]:
+        if exc_type is None:
+            self.finish()
+
+        return False
+
+
+def cron(identifier: str, fn: Callable[[], T] | None = None) -> None | T:
+    cron = Cron(identifier)
+    output = None
+
+    if fn is not None:
+        cron.start()
+        output = fn()
+
+    cron.finish()
+    return output