Improve metrics helper with gauges and timings

src/lambda_obs/metrics.py

@@ -32,6 +32,15 @@ def add_metric(self, name: str, value: float, unit: str = "Count") -> None:
         self._values[name] = self._values.get(name, 0.0) + float(value)
         self._units[name] = unit
 
+    def set_metric(self, name: str, value: float, unit: str = "None") -> None:
+        """Set an absolute gauge value (does not accumulate)."""
+        self._values[name] = float(value)
+        self._units[name] = unit
+
+    def timing(self, name: str, milliseconds: float) -> None:
+        """Record a latency sample in milliseconds."""
+        self.add_metric(name, milliseconds, unit="Milliseconds")
+
     def clear(self) -> None:
         self._values.clear()
         self._units.clear()

tests/test_metrics.py

@@ -4,18 +4,45 @@
 from lambda_obs.metrics import Metrics
 
 
-def test_metrics_flush_emits_emf_like_payload():
+def test_metrics_flush_emits_emf_line():
     buf = io.StringIO()
     m = Metrics("DemoNS", "demo", stream=buf)
-    m.add_dimension("stage", "test")
     m.add_metric("Invocations", 1)
-    m.add_metric("Invocations", 2)
     payload = m.flush()
-    assert payload["Invocations"] == 3.0
+    assert payload["Invocations"] == 1.0
     assert payload["service"] == "demo"
-    assert payload["stage"] == "test"
-    assert payload["_aws"]["CloudWatchMetrics"][0]["Namespace"] == "DemoNS"
-    # buffer cleared after flush
-    assert m.serialize()["Invocations"] if False else True
     line = json.loads(buf.getvalue().strip())
-    assert line["Invocations"] == 3.0
+    assert line["_aws"]["CloudWatchMetrics"][0]["Namespace"] == "DemoNS"
+    assert m._values == {}
+
+
+def test_metrics_add_dimension():
+    buf = io.StringIO()
+    m = Metrics("DemoNS", "demo", stream=buf)
+    m.add_dimension("stage", "dev")
+    m.add_metric("Errors", 2)
+    payload = m.flush()
+    assert payload["stage"] == "dev"
+    assert payload["Errors"] == 2.0
+
+
+def test_metrics_set_metric_is_absolute():
+    buf = io.StringIO()
+    m = Metrics("DemoNS", "demo", stream=buf)
+    m.set_metric("QueueDepth", 5, unit="Count")
+    m.set_metric("QueueDepth", 3, unit="Count")
+    payload = m.flush()
+    assert payload["QueueDepth"] == 3.0
+
+
+def test_metrics_timing():
+    buf = io.StringIO()
+    m = Metrics("DemoNS", "demo", stream=buf)
+    m.timing("HandlerDuration", 12.5)
+    payload = m.flush()
+    assert payload["HandlerDuration"] == 12.5
+    units = {
+        item["Name"]: item["Unit"]
+        for item in payload["_aws"]["CloudWatchMetrics"][0]["Metrics"]
+    }
+    assert units["HandlerDuration"] == "Milliseconds"