Re-raise redis.WatchErrors when they occur

In #2668, we started to avoid having the `redis.WatchError` mark the span as
having failed, but we were inadvertently suppressing that exception from the
calling code.  `redis.WatchError` is used for flow-of-control concurrency in
many applications, and applications depend on catching the error to handle
concurrent changes to keys during redis pipelines.

This re-raises the `WatchError` to keep the instrumentation transparent to the
application.

Fixes #2639

instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py

@@ -203,6 +203,8 @@ def _traced_execute_pipeline(func, instance, args, kwargs):
             span_name,
         ) = _build_span_meta_data_for_pipeline(instance)
 
+        exception = None
+
         with tracer.start_as_current_span(
             span_name, kind=trace.SpanKind.CLIENT
         ) as span:
@@ -216,13 +218,17 @@ def _traced_execute_pipeline(func, instance, args, kwargs):
             response = None
             try:
                 response = func(*args, **kwargs)
-            except redis.WatchError:
+            except redis.WatchError as watch_exception:
                 span.set_status(StatusCode.UNSET)
+                exception = watch_exception
 
             if callable(response_hook):
                 response_hook(span, instance, response)
 
-            return response
+        if exception:
+            raise exception
+
+        return response
 
     pipeline_class = (
         "BasePipeline" if redis.VERSION < (3, 0, 0) else "Pipeline"
@@ -279,6 +285,8 @@ async def _async_traced_execute_pipeline(func, instance, args, kwargs):
             span_name,
         ) = _build_span_meta_data_for_pipeline(instance)
 
+        exception = None
+
         with tracer.start_as_current_span(
             span_name, kind=trace.SpanKind.CLIENT
         ) as span:
@@ -292,12 +300,17 @@ async def _async_traced_execute_pipeline(func, instance, args, kwargs):
             response = None
             try:
                 response = await func(*args, **kwargs)
-            except redis.WatchError:
+            except redis.WatchError as watch_exception:
                 span.set_status(StatusCode.UNSET)
+                exception = watch_exception
 
             if callable(response_hook):
                 response_hook(span, instance, response)
-            return response
+
+        if exception:
+            raise exception
+
+        return response
 
     if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
         wrap_function_wrapper(

instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py

@@ -359,16 +359,14 @@ def test_response_error(self):
 
     def test_watch_error_sync(self):
         def redis_operations():
-            try:
+            with pytest.raises(WatchError):
                 redis_client = fakeredis.FakeStrictRedis()
                 pipe = redis_client.pipeline(transaction=True)
                 pipe.watch("a")
                 redis_client.set("a", "bad")  # This will cause the WatchError
                 pipe.multi()
                 pipe.set("a", "1")
                 pipe.execute()
-            except WatchError:
-                pass
 
         redis_operations()
 
@@ -400,16 +398,14 @@ def tearDown(self):
     @pytest.mark.asyncio
     async def test_watch_error_async(self):
         async def redis_operations():
-            try:
+            with pytest.raises(WatchError):
                 redis_client = FakeRedis()
                 async with redis_client.pipeline(transaction=False) as pipe:
                     await pipe.watch("a")
                     await redis_client.set("a", "bad")
                     pipe.multi()
                     await pipe.set("a", "1")
                     await pipe.execute()
-            except WatchError:
-                pass
 
         await redis_operations()