Fix failing test, lint, and update changelog

- set_attribute and also set_status are not called if span was already ended. Simple reorder of function calls helped
- I also realized that accessing end_time is dangerous, in SDK it is protected by lock. I solved this by remembering if we already called end. Calling multiple times end on span is not problematic, but it generates warning logs and is probably not good practice
- updated CHANGELOG.md
- fix lint

Author: UOndro

CHANGELOG.md

@@ -51,6 +51,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2461](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2461))
 - Remove SDK dependency from opentelemetry-instrumentation-grpc
   ([#2474](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2474))
+- `opentelemetry-instrumentation-grpc` User should be able to cancel grpc stream
+  ([#2093](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2093))
 
 ## Version 1.24.0/0.45b0 (2024-03-28)
 

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_client.py

@@ -74,20 +74,25 @@ def _safe_invoke(function: Callable, *args):
         )
 
 
+# pylint:disable=abstract-method
 class OpenTelemetryStreamWrapper(wrapt.ObjectProxy):
     def __init__(self, wrapped, span: trace.Span):
         super().__init__(wrapped)
         self._self_span = span
+        self._span_ended = False
 
     def _end_span_if_not_already_ended(self, status_code=None, status=None):
-        if self._self_span.end_time is None:
-            self._self_span.end()
-            if status_code is not None:
-                self._self_span.set_attribute(
-                    SpanAttributes.RPC_GRPC_STATUS_CODE, status_code
-                )
-            if status is not None:
-                self._self_span.set_status(status)
+        if self._span_ended:
+            return
+
+        if status_code is not None:
+            self._self_span.set_attribute(
+                SpanAttributes.RPC_GRPC_STATUS_CODE, status_code
+            )
+        if status is not None:
+            self._self_span.set_status(status)
+        self._span_ended = True
+        self._self_span.end()
 
     def __del__(self):
         self._end_span_if_not_already_ended()

instrumentation/opentelemetry-instrumentation-grpc/tests/test_client_interceptor.py

@@ -169,8 +169,8 @@ def test_unary_stream(self):
 
     def test_unary_stream_can_be_cancel(self):
         responses = server_streaming_method(self._stub, serialize=False)
-        for i, _ in enumerate(responses):
-            if i == 1:
+        for response_num, _ in enumerate(responses):
+            if response_num == 1:
                 responses.cancel()
                 break
         spans = self.memory_exporter.get_finished_spans()
@@ -280,8 +280,8 @@ def test_stream_stream(self):
 
     def test_stream_stream_can_be_cancel(self):
         responses = bidirectional_streaming_method(self._stub, serialize=False)
-        for i, _ in enumerate(responses):
-            if i == 1:
+        for response_num, _ in enumerate(responses):
+            if response_num == 1:
                 responses.cancel()
                 break
         spans = self.memory_exporter.get_finished_spans()