Adds timing metrics for Protocol compute and Event response

canvas_sdk/utils/stats.py

@@ -0,0 +1,27 @@
+from time import time
+from typing import Any
+
+
+def get_duration_ms(start_time: time) -> int:
+    return int((time() - start_time) * 1000)
+
+
+LINE_PROTOCOL_TRANSLATION = str.maketrans(
+    {
+        ",": r"\,",
+        "=": r"\=",
+        " ": r"\ ",
+        ":": r"__",
+    }
+)
+
+
+def tags_to_line_protocol(tags: dict[str, Any]) -> str:
+    """Generate a tags string compatible with the InfluxDB line protocol.
+
+    See: https://docs.influxdata.com/influxdb/v1.1/write_protocols/line_protocol_tutorial/
+    """
+    return ",".join(
+        f"{tag_name}={str(tag_value).translate(LINE_PROTOCOL_TRANSLATION)}"
+        for tag_name, tag_value in tags.items()
+    )

plugin_runner/plugin_runner.py

@@ -5,12 +5,14 @@
 import pathlib
 import signal
 import sys
+import time
 import traceback
 from collections import defaultdict
 from types import FrameType
 from typing import Any, Optional
 
 import grpc
+import statsd
 from plugin_synchronizer import publish_message
 from sandbox import Sandbox
 
@@ -23,6 +25,7 @@
     add_PluginRunnerServicer_to_server,
 )
 from canvas_sdk.events import Event, EventResponse, EventType
+from canvas_sdk.utils.stats import get_duration_ms, tags_to_line_protocol
 from logger import log
 
 ENV = os.getenv("ENV", "development")
@@ -52,12 +55,14 @@ class PluginRunner(PluginRunnerServicer):
     """This process runs provided plugins that register interest in incoming events."""
 
     def __init__(self) -> None:
+        self.statsd_client = statsd.StatsClient()
         super().__init__()
 
     sandbox: Sandbox
 
     async def HandleEvent(self, request: Event, context: Any) -> EventResponse:
         """This is invoked when an event comes in."""
+        event_start_time = time.time()
         event_name = EventType.Name(request.type)
         relevant_plugins = EVENT_PROTOCOL_MAP.get(event_name, [])
 
@@ -69,12 +74,29 @@ async def HandleEvent(self, request: Event, context: Any) -> EventResponse:
 
             try:
                 protocol = protocol_class(request, plugin.get("secrets", {}))
+                compute_start_time = time.time()
                 effects = await asyncio.get_running_loop().run_in_executor(None, protocol.compute)
+                compute_duration = get_duration_ms(compute_start_time)
+                log.info(f"{plugin_name}.compute() completed ({compute_duration} ms)")
+                statsd_tags = tags_to_line_protocol({"plugin": plugin_name})
+                self.statsd_client.timing(
+                    f"plugins.protocol_duration_ms,{statsd_tags}",
+                    delta=compute_duration,
+                )
             except Exception as e:
                 log.error(traceback.format_exception(e))
                 continue
             effect_list += effects
 
+        event_duration = get_duration_ms(event_start_time)
+        # Don't log anything if a protocol didn't actually run.
+        if relevant_plugins:
+            log.info(f"Responded to Event {event_name} ({event_duration} ms)")
+            statsd_tags = tags_to_line_protocol({"event": event_name})
+            self.statsd_client.timing(
+                f"plugins.event_duration_ms,{statsd_tags}",
+                delta=event_duration,
+            )
         yield EventResponse(success=True, effects=effect_list)
 
     async def ReloadPlugins(