Repository: meltano/sdk
Pull request #2486 (2024-06-18 15:13:45)

feat: Emit target metrics
Closes https://github.com/meltano/sdk/issues/1573

<!-- readthedocs-preview meltano-sdk start -->
----
📚 Documentation preview 📚: https://meltano-sdk--2486.org.readthedocs.build/en/2486/

<!-- readthedocs-preview meltano-sdk end -->

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/meltano/sdk/pull/2486?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
Attention: Patch coverage is `94.73684%` with `2 lines` in your changes missing coverage. Please review.
> Project coverage is 89.22%. Comparing base [(`6334091`)](https://app.codecov.io/gh/meltano/sdk/commit/63340915751b798bade84572a6d6bda2da9e2d1b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) to head [(`94a642b`)](https://app.codecov.io/gh/meltano/sdk/commit/94a642beb9c7e5d3942c92f148c1ad4aeec3798c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).

| [Files](https://app.codecov.io/gh/meltano/sdk/pull/2486?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) | Patch % | Lines |
|---|---|---|
| [singer\_sdk/io\_base.py](https://app.codecov.io/gh/meltano/sdk/pull/2486?src=pr&el=tree&filepath=singer_sdk%2Fio_base.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9pb19iYXNlLnB5) | 92.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/meltano/sdk/pull/2486?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #2486      +/-   ##
==========================================
+ Coverage   89.18%   89.22%   +0.04%     
==========================================
  Files          54       54              
  Lines        4790     4809      +19     
  Branches      937      940       +3     
==========================================
+ Hits         4272     4291      +19     
  Misses        361      361              
  Partials      157      157              
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/meltano/sdk/pull/2486?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).

codspeed-hq[bot]: ## [CodSpeed Performance Report](https://codspeed.io/meltano/sdk/branches/1573-feat-compute-target-metrics)

### Merging #2486 will **not alter performance**

<sub>Comparing <code>1573-feat-compute-target-metrics</code> (94a642b) with <code>main</code> (6334091)</sub>



### Summary

`✅ 6` untouched benchmarks






Patch:
diff --git a/docs/implementation/metrics.md b/docs/implementation/metrics.md
index 87678cbe..3b2a2a6a 100644
--- a/docs/implementation/metrics.md
+++ b/docs/implementation/metrics.md
@@ -1,8 +1,14 @@
-# Tap Metrics
+# Tap and Target Metrics
 
 Metrics logging is specified in the
-[Singer Spec](https://hub.meltano.com/singer/spec#metrics). The SDK will automatically
-emit metrics for `record_count`, `http_request_duration` and `sync_duration`.
+[Singer Spec](https://hub.meltano.com/singer/spec#metrics).
+
+The SDK will automatically emit the following metrics:
+
+- `record_count`: The number of records processed by the tap or target.
+- `http_request_duration`: The duration of HTTP requests made by the tap.
+- `sync_duration`: The duration of the sync operation.
+- `batch_processing_time`: The duration of processing a batch of records.
 
 ## Customization options
 
diff --git a/singer_sdk/metrics.py b/singer_sdk/metrics.py
index 50d7d392..fb97e693 100644
--- a/singer_sdk/metrics.py
+++ b/singer_sdk/metrics.py
@@ -47,6 +47,7 @@ class Tag(str, enum.Enum):
     JOB_TYPE = "job_type"
     HTTP_STATUS_CODE = "http_status_code"
     STATUS = "status"
+    PID = "pid"
 
 
 class Metric(str, enum.Enum):
@@ -58,6 +59,7 @@ class Metric(str, enum.Enum):
     HTTP_REQUEST_COUNT = "http_request_count"
     JOB_DURATION = "job_duration"
     SYNC_DURATION = "sync_duration"
+    BATCH_PROCESSING_TIME = "batch_processing_time"
 
 
 @dataclass
@@ -116,6 +118,7 @@ class Meter(metaclass=abc.ABCMeta):
         """
         self.metric = metric
         self.tags = tags or {}
+        self.tags[Tag.PID] = os.getpid()
         self.logger = get_metrics_logger()
 
     @property
@@ -182,6 +185,10 @@ class Counter(Meter):
         self.log_interval = log_interval
         self.last_log_time = time()
 
+    def exit(self) -> None:
+        """Exit the counter context."""
+        self._pop()
+
     def __enter__(self) -> Counter:
         """Enter the counter context.
 
@@ -204,7 +211,7 @@ class Counter(Meter):
             exc_val: The exception value.
             exc_tb: The exception traceback.
         """
-        self._pop()
+        self.exit()
 
     def _pop(self) -> None:
         """Log and reset the counter."""
diff --git a/singer_sdk/sinks/core.py b/singer_sdk/sinks/core.py
index e3c1ef56..5a936a63 100644
--- a/singer_sdk/sinks/core.py
+++ b/singer_sdk/sinks/core.py
@@ -17,6 +17,7 @@ import jsonschema
 import jsonschema.validators
 from typing_extensions import override
 
+from singer_sdk import metrics
 from singer_sdk._singerlib.json import deserialize_json
 from singer_sdk.exceptions import (
     InvalidJSONSchema,
@@ -193,6 +194,31 @@ class Sink(metaclass=abc.ABCMeta):  # noqa: PLR0904
         )
 
         self._validator: BaseJSONSchemaValidator | None = self.get_validator()
+        self._record_counter: metrics.Counter = metrics.record_counter(self.stream_name)
+        self._batch_timer = metrics.Timer(
+            metrics.Metric.BATCH_PROCESSING_TIME,
+            tags={
+                metrics.Tag.STREAM: self.stream_name,
+            },
+        )
+
+    @property
+    def record_counter_metric(self) -> metrics.Counter:
+        """Get the record counter for this sink.
+
+        Returns:
+            The Meter instance for the record counter.
+        """
+        return self._record_counter
+
+    @property
+    def batch_processing_timer(self) -> metrics.Timer:
+        """Get the batch processing timer for this sink.
+
+        Returns:
+            The Meter instance for the batch processing timer.
+        """
+        return self._batch_timer
 
     @cached_property
     def validate_schema(self) -> bool:
@@ -685,6 +711,7 @@ class Sink(metaclass=abc.ABCMeta):  # noqa: PLR0904
         should not be relied on, it's recommended to use a uuid as well.
         """
         self.logger.info("Cleaning up %s", self.stream_name)
+        self.record_counter_metric.exit()
 
     def process_batch_files(
         self,
diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index 81c991a0..22ad2817 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -359,6 +359,7 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
 
             sink.tally_record_read()
             sink.process_record(transformed_record, context)
+            sink.record_counter_metric.increment()
             sink._after_process_record(context)  # noqa: SLF001
 
             if sink.is_full:
@@ -510,7 +511,8 @@ class Target(PluginBase, SingerReader, metaclass=abc.ABCMeta):
             return
 
         draining_status = sink.start_drain()
-        sink.process_batch(draining_status)
+        with sink.batch_processing_timer:
+            sink.process_batch(draining_status)
         sink.mark_drained()
 
     def _drain_all(self, sink_list: list[Sink], parallelism: int) -> None:


Test patch:
diff --git a/tests/core/test_metrics.py b/tests/core/test_metrics.py
index c78969f8..b0de4c9b 100644
--- a/tests/core/test_metrics.py
+++ b/tests/core/test_metrics.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import os
 
 import pytest
 import time_machine
@@ -18,6 +19,8 @@ class CustomObject:
 
 
 def test_meter():
+    pid = os.getpid()
+
     class _MyMeter(metrics.Meter):
         def __enter__(self):
             return self
@@ -27,11 +30,14 @@ def test_meter():
 
     meter = _MyMeter(metrics.Metric.RECORD_COUNT)
 
-    assert meter.tags == {}
+    assert meter.tags == {metrics.Tag.PID: pid}
 
     stream_context = {"parent_id": 1}
     meter.context = stream_context
-    assert meter.tags == {metrics.Tag.CONTEXT: stream_context}
+    assert meter.tags == {
+        metrics.Tag.CONTEXT: stream_context,
+        metrics.Tag.PID: pid,
+    }
 
     meter.context = None
     assert metrics.Tag.CONTEXT not in meter.tags
@@ -39,6 +45,7 @@ def test_meter():
 
 def test_record_counter(caplog: pytest.LogCaptureFixture):
     caplog.set_level(logging.INFO, logger=metrics.METRICS_LOGGER_NAME)
+    pid = os.getpid()
     custom_object = CustomObject("test", 1)
 
     with metrics.record_counter(
@@ -68,6 +75,7 @@ def test_record_counter(caplog: pytest.LogCaptureFixture):
         assert point.tags == {
             metrics.Tag.STREAM: "test_stream",
             metrics.Tag.ENDPOINT: "test_endpoint",
+            metrics.Tag.PID: pid,
             "custom_tag": "pytest",
             "custom_obj": custom_object,
         }
@@ -79,6 +87,7 @@ def test_record_counter(caplog: pytest.LogCaptureFixture):
 
 def test_sync_timer(caplog: pytest.LogCaptureFixture):
     caplog.set_level(logging.INFO, logger=metrics.METRICS_LOGGER_NAME)
+    pid = os.getpid()
     traveler = time_machine.travel(0, tick=False)
     traveler.start()
 
@@ -100,6 +109,7 @@ def test_sync_timer(caplog: pytest.LogCaptureFixture):
     assert point.tags == {
         metrics.Tag.STREAM: "test_stream",
         metrics.Tag.STATUS: "succeeded",
+        metrics.Tag.PID: pid,
         "custom_tag": "pytest",
     }