Repository: jina-ai/serve (language: python)

Issue:
Missing gateway metric recording logic
**Describe the bug**
<!-- A clear and concise description of what the bug is. -->
There is a branch in the metric code that doesnt record bytes metrics.

The bug was brought up in a discussion in a PR:
https://github.com/jina-ai/jina/pull/5344#discussion_r1015224194

**Way to reproduce**
<!-- copy past your code/pull request link -->
Run the following script
```python
from jina import Flow

f = Flow(
    tracing=False,
    metrics=True,
    metrics_exporter_host='localhost',
    metrics_exporter_port=4317,
    port=12345,
).add(name='name1').add(name='name2', needs=['gateway']).add(name='name3', needs=['name1', 'name2'], disable_reduce=True)
with f:
    f.post('/')
```
Then check the exported metrics in Grafana (in the screenshots section)

---

<!-- Optional, but really help us locate the problem faster -->

**Environment**
<!-- Run `jina --version-full` and copy paste the output here -->
- jina 3.11.1
- docarray 0.18.1
- jcloud 0.0.36
- jina-hubble-sdk 0.23.3
- jina-proto 0.1.13
- protobuf 3.20.3
- proto-backend python
- grpcio 1.47.2
- pyyaml 6.0
- python 3.10.6
- platform Darwin
- platform-release 21.6.0
- platform-version Darwin Kernel Version 21.6.0: Wed Aug 10 14:28:23 PDT 2022;
  root:xnu-8020.141.5~2/RELEASE_ARM64_T6000
- architecture arm64
- processor arm
- uid 55173526391343
- session-id 400383ae-5e92-11ed-a048-322e16710a2f
- uptime 2022-11-07T19:49:34.885055
- ci-vendor (unset)
- internal False
* JINA_DEFAULT_HOST (unset)
* JINA_DEFAULT_TIMEOUT_CTRL (unset)
* JINA_DEPLOYMENT_NAME (unset)
* JINA_DISABLE_UVLOOP (unset)
* JINA_EARLY_STOP (unset)
* JINA_FULL_CLI (unset)
* JINA_GATEWAY_IMAGE (unset)
* JINA_GRPC_RECV_BYTES (unset)
* JINA_GRPC_SEND_BYTES (unset)
* JINA_HUB_NO_IMAGE_REBUILD (unset)
* JINA_LOG_CONFIG (unset)
* JINA_LOG_LEVEL (unset)
* JINA_LOG_NO_COLOR (unset)
* JINA_MP_START_METHOD (unset)
* JINA_OPTOUT_TELEMETRY (unset)
* JINA_RANDOM_PORT_MAX (unset)
* JINA_RANDOM_PORT_MIN (unset)
* JINA_LOCKS_ROOT (unset)
* JINA_K8S_ACCESS_MODES (unset)
* JINA_K8S_STORAGE_CLASS_NAME (unset)
* JINA_K8S_STORAGE_CAPACITY (unset)

**Screenshots**
<!-- If applicable, add screenshots to help explain your problem. -->
sending_request_seconds is recorded by MetricsTimer and records all 3 deployments
![image](https://user-images.githubusercontent.com/56836461/200302521-90410c3a-274d-45e1-91f2-fc27ed26f2e3.png)

sent_request_bytes is not recorded for 'name3' deployment
![image](https://user-images.githubusercontent.com/56836461/200302350-57b37d87-bf7f-4a99-8817-3435fe2e8cc6.png)




Discussion / hints:

Fix patch:
diff --git a/jina/serve/networking.py b/jina/serve/networking.py
index deea639aff299..8887de5aaf78b 100644
--- a/jina/serve/networking.py
+++ b/jina/serve/networking.py
@@ -435,14 +435,15 @@ async def send_requests(
 
             timer = self._get_metric_timer()
             if request_type == DataRequest and len(requests) == 1:
+                request = requests[0]
                 if self.single_data_stub:
+                    self._record_request_bytes_metric(request.nbytes)
                     call_result = self.single_data_stub.process_single_data(
-                        requests[0],
+                        request,
                         metadata=metadata,
                         compression=compression,
                         timeout=timeout,
                     )
-                    self._record_request_bytes_metric(requests[0].nbytes)
                     with timer:
                         metadata, response = (
                             await call_result.trailing_metadata(),
@@ -452,8 +453,7 @@ async def send_requests(
                     return response, metadata
 
                 elif self.stream_stub:
-                    for response in requests:
-                        self._record_request_bytes_metric(response.nbytes)
+                    self._record_request_bytes_metric(request.nbytes)
 
                     with timer:
                         async for response in self.stream_stub.Call(
@@ -466,9 +466,9 @@ async def send_requests(
                             return response, None
 
             if request_type == DataRequest and len(requests) > 1:
-                # TODO: This section of the code has no observability
-                # TODO: Does this section not need to deal with streaming stubs?
                 if self.data_list_stub:
+                    for request in requests:
+                        self._record_request_bytes_metric(request.nbytes)
                     call_result = self.data_list_stub.process_data(
                         requests,
                         metadata=metadata,
@@ -480,6 +480,7 @@ async def send_requests(
                             await call_result.trailing_metadata(),
                             await call_result,
                         )
+                        self._record_received_bytes_metric(response.nbytes)
                     return response, metadata
                 else:
                     raise ValueError(


Test patch:
diff --git a/tests/unit/serve/instrumentation/conftest.py b/tests/unit/serve/instrumentation/conftest.py
index c06261eabb134..2dbccf5053ff8 100644
--- a/tests/unit/serve/instrumentation/conftest.py
+++ b/tests/unit/serve/instrumentation/conftest.py
@@ -9,6 +9,7 @@
     MetricExporter,
     MetricExportResult,
     MetricsData,
+    PeriodicExportingMetricReader,
 )
 
 
@@ -54,12 +55,11 @@ def __del__(self):
         self.f.close()
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope='function')
 def monkeypatch_metric_exporter(
     tmpdir_factory: pytest.TempdirFactory,
 ) -> Tuple[Callable, Callable]:
     import opentelemetry.sdk.metrics.export
-    from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
     from pathlib import Path
     import time
     import os
@@ -78,7 +78,7 @@ def collect_metrics():
             tick_counter = int(f.read())
         with open(tick_counter_filename, 'w') as f:
             f.write(str(tick_counter + 1))
-        time.sleep(1)
+        time.sleep(2)
 
     def _get_service_name(otel_measurement):
         return otel_measurement[0]['resource_metrics'][0]['resource']['attributes'][
@@ -102,7 +102,7 @@ def __init__(self, *args, **kwargs) -> None:
 
             super().__init__(
                 exporter=self.exporter,
-                export_interval_millis=1_000,
+                export_interval_millis=500,
             )
 
         def _ticker(self) -> None:
@@ -115,5 +115,7 @@ def _ticker(self) -> None:
                     self.collect(timeout_millis=self._export_timeout_millis)
             self.collect(timeout_millis=self._export_interval_millis)
 
+    real_reader = opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader
     opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader = PatchedTextReader
-    return collect_metrics, read_metrics
+    yield collect_metrics, read_metrics
+    opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader = real_reader
diff --git a/tests/unit/serve/instrumentation/test_gateway_metric_labels.py b/tests/unit/serve/instrumentation/test_gateway_metric_labels.py
index 054d9b1b13a0d..068927f655a66 100644
--- a/tests/unit/serve/instrumentation/test_gateway_metric_labels.py
+++ b/tests/unit/serve/instrumentation/test_gateway_metric_labels.py
@@ -34,81 +34,122 @@ def meow(self, docs, **kwargs):
             i['name']: i['data']['data_points'] for i in gateway_metrics
         }
 
-        assert (
-            'address'
-            in gateway_metric_data_point['jina_sending_request_seconds'][0][
-                'attributes'
-            ]
-        )
-        assert (
-            'address'
-            in gateway_metric_data_point['jina_sent_request_bytes'][0]['attributes']
-        )
-        assert (
-            'address'
-            in gateway_metric_data_point['jina_received_response_bytes'][0][
-                'attributes'
-            ]
-        )
-        assert (
-            'address'
-            in gateway_metric_data_point['jina_sending_request_seconds'][1][
-                'attributes'
-            ]
-        )
-        assert (
-            'address'
-            in gateway_metric_data_point['jina_sent_request_bytes'][1]['attributes']
-        )
-        assert (
-            'address'
-            in gateway_metric_data_point['jina_received_response_bytes'][1][
-                'attributes'
-            ]
-        )
+    assert (
+        'address'
+        in gateway_metric_data_point['jina_sending_request_seconds'][0][
+            'attributes'
+        ]
+    )
+    assert (
+        'address'
+        in gateway_metric_data_point['jina_sent_request_bytes'][0]['attributes']
+    )
+    assert (
+        'address'
+        in gateway_metric_data_point['jina_received_response_bytes'][0][
+            'attributes'
+        ]
+    )
+    assert (
+        'address'
+        in gateway_metric_data_point['jina_sending_request_seconds'][1][
+            'attributes'
+        ]
+    )
+    assert (
+        'address'
+        in gateway_metric_data_point['jina_sent_request_bytes'][1]['attributes']
+    )
+    assert (
+        'address'
+        in gateway_metric_data_point['jina_received_response_bytes'][1][
+            'attributes'
+        ]
+    )
 
-        assert (
-            'deployment'
-            in gateway_metric_data_point['jina_sending_request_seconds'][0][
-                'attributes'
-            ]
-        )
-        assert (
-            'deployment'
-            in gateway_metric_data_point['jina_sent_request_bytes'][0]['attributes']
-        )
-        assert (
-            'deployment'
-            in gateway_metric_data_point['jina_received_response_bytes'][0][
-                'attributes'
-            ]
-        )
-        assert (
-            'deployment'
-            in gateway_metric_data_point['jina_sending_request_seconds'][1][
-                'attributes'
-            ]
-        )
-        assert (
-            'deployment'
-            in gateway_metric_data_point['jina_sent_request_bytes'][1]['attributes']
-        )
-        assert (
-            'deployment'
-            in gateway_metric_data_point['jina_received_response_bytes'][1][
-                'attributes'
-            ]
-        )
+    assert (
+        'deployment'
+        in gateway_metric_data_point['jina_sending_request_seconds'][0][
+            'attributes'
+        ]
+    )
+    assert (
+        'deployment'
+        in gateway_metric_data_point['jina_sent_request_bytes'][0]['attributes']
+    )
+    assert (
+        'deployment'
+        in gateway_metric_data_point['jina_received_response_bytes'][0][
+            'attributes'
+        ]
+    )
+    assert (
+        'deployment'
+        in gateway_metric_data_point['jina_sending_request_seconds'][1][
+            'attributes'
+        ]
+    )
+    assert (
+        'deployment'
+        in gateway_metric_data_point['jina_sent_request_bytes'][1]['attributes']
+    )
+    assert (
+        'deployment'
+        in gateway_metric_data_point['jina_received_response_bytes'][1][
+            'attributes'
+        ]
+    )
 
-        assert (
-            gateway_metric_data_point['jina_received_response_bytes'][0]['attributes'][
-                'deployment'
-            ]
-            == 'first_exec'
-        )
-        assert (
-            gateway_metric_data_point['jina_received_response_bytes'][1]['attributes'][
-                'deployment'
-            ]
-            == 'second_exec'
+    assert {'first_exec', 'second_exec'} == {
+        i['attributes']['deployment']
+        for i in gateway_metric_data_point['jina_received_response_bytes']
+    }
+    assert {'first_exec', 'second_exec'} == {
+        i['attributes']['deployment']
+        for i in gateway_metric_data_point['jina_sent_request_bytes']
+    }
+    assert {'first_exec', 'second_exec'} == {
+        i['attributes']['deployment']
+        for i in gateway_metric_data_point['jina_sending_request_seconds']
+    }
+
+
+def test_merge_with_no_reduce(monkeypatch_metric_exporter):
+    collect_metrics, read_metrics = monkeypatch_metric_exporter
+
+    f = (
+        Flow(
+            tracing=False,
+            metrics=True,
+            metrics_exporter_host='localhost',
+            metrics_exporter_port=4317,
+            port=12345,
         )
+        .add(name='name1')
+        .add(name='name2', needs=['gateway'])
+        .add(name='name3', needs=['name1', 'name2'], disable_reduce=True)
+    )
+    with f:
+        f.post('/')
+        collect_metrics()
+        metrics = read_metrics()
+
+        gateway_metrics = metrics['gateway/rep-0'][0]['resource_metrics'][0][
+            'scope_metrics'
+        ][0]['metrics']
+        gateway_metric_data_point = {
+            i['name']: i['data']['data_points'] for i in gateway_metrics
+        }
+
+    assert {'name1', 'name2', 'name3'} == {
+        i['attributes']['deployment']
+        for i in gateway_metric_data_point['jina_received_response_bytes']
+    }
+    assert {'name1', 'name2', 'name3'} == {
+        i['attributes']['deployment']
+        for i in gateway_metric_data_point['jina_sent_request_bytes']
+    }
+    assert {'name1', 'name2', 'name3'} == {
+        i['attributes']['deployment']
+        for i in gateway_metric_data_point['jina_sending_request_seconds']
+    }
diff --git a/tests/unit/serve/instrumentation/test_instrumentation.py b/tests/unit/serve/instrumentation/test_instrumentation.py
index 6ed329f58b9ff..c1d1d8228cba8 100644
--- a/tests/unit/serve/instrumentation/test_instrumentation.py
+++ b/tests/unit/serve/instrumentation/test_instrumentation.py
@@ -86,6 +86,7 @@ def _sleep():
         'cat': 'meow',
         'dog': 'woof',
     }
+
     @MetricsTimer(summary, histogram, labels)
     def _sleep():
         time.sleep(0.1)