Add support for process.disk.io metric in system-metrics instrumentation (#4397)

Author: Manvi2402

CHANGELOG.md

@@ -15,6 +15,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add `BaggageLogProcessor` to `opentelemetry-processor-baggage`
   ([#4371](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4371))
 
+- `opentelemetry-instrumentation-system-metrics`: Add support for `process.disk.io` metric in system-metrics instrumentation
+  ([#4397](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/4397))
 - Switch to SPDX license headers and add CI enforcement
   ([#4533](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4533))
 - Bump `pylint` to `4.0.5`

instrumentation/opentelemetry-instrumentation-system-metrics/src/opentelemetry/instrumentation/system_metrics/__init__.py

@@ -30,6 +30,7 @@
         "process.cpu.utilization": None,
         "process.memory.usage": None,
         "process.memory.virtual": None,
+        "process.disk.io": ["read", "write"],
         "process.open_file_descriptor.count": None,
         "process.thread.count": None,
         "process.runtime.memory": ["rss", "vms"],
@@ -134,6 +135,7 @@
     "process.memory.virtual": None,
     "process.open_file_descriptor.count": None,
     "process.thread.count": None,
+    "process.disk.io": ["read", "write"],
     "process.runtime.memory": ["rss", "vms"],
     "process.runtime.cpu.time": ["user", "system"],
     "process.runtime.gc_count": None,
@@ -208,6 +210,7 @@ def __init__(
         self._cpu_utilization_labels = self._labels.copy()
         self._memory_usage_labels = self._labels.copy()
         self._memory_virtual_labels = self._labels.copy()
+        self._process_disk_io_labels = self._labels.copy()
         self._open_file_descriptor_count_labels = self._labels.copy()
         self._thread_count_labels = self._labels.copy()
 
@@ -459,6 +462,14 @@ def _instrument(self, **kwargs: Any):
                 description="Process threads count.",
             )
 
+        if "process.disk.io" in self._config:
+            self._meter.create_observable_counter(
+                name="process.disk.io",
+                callbacks=[self._get_process_disk_io],
+                description="Disk bytes transferred for the process.",
+                unit="By",
+            )
+
         # FIXME: process.runtime keys are deprecated and will be removed in subsequent releases.
         # When removing them, remember to clean also the callbacks and labels
 
@@ -909,6 +920,27 @@ def _get_thread_count(
             self._proc.num_threads(), self._thread_count_labels.copy()
         )
 
+    def _get_process_disk_io(
+        self, options: CallbackOptions
+    ) -> Iterable[Observation]:
+        """Observer callback for process disk IO"""
+        try:
+            proc_disk = self._proc.io_counters()
+        except (AttributeError, NotImplementedError, PermissionError):
+            # io_counters() is not available on all platforms (e.g. macOS)
+            return
+        for metric in self._config["process.disk.io"]:
+            if metric == "read":
+                self._process_disk_io_labels["direction"] = "read"
+                yield Observation(
+                    proc_disk.read_bytes, self._process_disk_io_labels.copy()
+                )
+            elif metric == "write":
+                self._process_disk_io_labels["direction"] = "write"
+                yield Observation(
+                    proc_disk.write_bytes, self._process_disk_io_labels.copy()
+                )
+
     # runtime callbacks
 
     def _get_runtime_memory(

instrumentation/opentelemetry-instrumentation-system-metrics/tests/test_system_metrics.py

@@ -110,6 +110,7 @@ def test_system_metrics_instrument(self):
             "process.cpu.time",
             "process.cpu.utilization",
             "process.memory.usage",
+            "process.disk.io",
             "process.memory.virtual",
             "process.thread.count",
             f"process.runtime.{self.implementation}.memory",
@@ -151,6 +152,7 @@ def test_process_metrics_instrument(self):
             "process.memory.virtual": None,
             "process.open_file_descriptor.count": None,
             "process.thread.count": None,
+            "process.disk.io": ["read", "write"],
         }
 
         reader = InMemoryMetricReader()
@@ -169,6 +171,7 @@ def test_process_metrics_instrument(self):
             "process.memory.virtual",
             "process.cpu.time",
             "process.thread.count",
+            "process.disk.io",
             "process.context_switches",
             "process.cpu.utilization",
         ]
@@ -887,6 +890,28 @@ def test_thread_count(self, mock_process_thread_num):
         expected = [_SystemMetricsResult({}, 42)]
         self._test_metrics("process.thread.count", expected)
 
+    @mock.patch("psutil.Process.io_counters")
+    def test_process_disk_io(self, mock_process_io_counters):
+        PIOCounters = namedtuple("PIOCounters", ["read_bytes", "write_bytes"])
+
+        mock_process_io_counters.configure_mock(
+            **{"return_value": PIOCounters(read_bytes=1024, write_bytes=2048)}
+        )
+
+        expected = [
+            _SystemMetricsResult({"direction": "read"}, 1024),
+            _SystemMetricsResult({"direction": "write"}, 2048),
+        ]
+        self._test_metrics("process.disk.io", expected)
+
+    @mock.patch("psutil.Process.io_counters")
+    def test_process_disk_io_not_implemented_error(
+        self, mock_process_io_counters
+    ):
+        mock_process_io_counters.side_effect = NotImplementedError
+
+        self._assert_metrics_not_found("process.disk.io")
+
     @mock.patch("psutil.Process.cpu_percent")
     @mock.patch("psutil.cpu_count")
     def test_cpu_utilization(self, mock_cpu_count, mock_process_cpu_percent):