addressing concurrency issue in Span API (open-telemetry#1259)

alrex · web-flow · commit 27dc6916fc5c · 2020-10-26T11:35:16.000-07:00
Before this change, the following would cause unintended behaviour in the Span API's add_event, update_name and set_status methods:

- thread A calls set_status, locks to check if a span has ended, and releases the lock
- thread B obtains the lock and ends a span
- thread A continues its call of set_status

This change ensures that the update operations are done with the lock being held. It's done in a decorator, but that is completely optional.
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py
@@ -371,6 +371,17 @@ def _create_immutable_attributes(attributes):
     return MappingProxyType(attributes.copy() if attributes else {})
 
 
+def _check_span_ended(func):
+    def wrapper(self, *args, **kwargs):
+        with self._lock:  # pylint: disable=protected-access
+            if self.end_time is not None:
+                logger.warning("Calling %s on an ended span.", func.__name__)
+                return
+            func(self, *args, **kwargs)
+
+    return wrapper
+
+
 class Span(trace_api.Span):
     """See `opentelemetry.trace.Span`.
 
@@ -560,19 +571,18 @@ def get_span_context(self):
         return self.context
 
     def set_attribute(self, key: str, value: types.AttributeValue) -> None:
-        with self._lock:
-            if not self.is_recording():
-                return
-            has_ended = self.end_time is not None
-        if has_ended:
-            logger.warning("Setting attribute on ended span.")
+        if not _is_valid_attribute_value(value):
             return
 
         if not key:
             logger.warning("invalid key (empty or null)")
             return
 
-        if _is_valid_attribute_value(value):
+        with self._lock:
+            if self.end_time is not None:
+                logger.warning("Setting attribute on ended span.")
+                return
+
             # Freeze mutable sequences defensively
             if isinstance(value, MutableSequence):
                 value = tuple(value)
@@ -582,18 +592,10 @@ def set_attribute(self, key: str, value: types.AttributeValue) -> None:
                 except ValueError:
                     logger.warning("Byte attribute could not be decoded.")
                     return
-            with self._lock:
-                self.attributes[key] = value
+            self.attributes[key] = value
 
+    @_check_span_ended
     def _add_event(self, event: EventBase) -> None:
-        with self._lock:
-            if not self.is_recording():
-                return
-            has_ended = self.end_time is not None
-
-        if has_ended:
-            logger.warning("Calling add_event() on an ended span.")
-            return
         self.events.append(event)
 
     def add_event(
@@ -618,56 +620,39 @@ def start(
         parent_context: Optional[context_api.Context] = None,
     ) -> None:
         with self._lock:
-            if not self.is_recording():
+            if self.start_time is not None:
+                logger.warning("Calling start() on a started span.")
                 return
-            has_started = self.start_time is not None
-            if not has_started:
-                self._start_time = (
-                    start_time if start_time is not None else time_ns()
-                )
-        if has_started:
-            logger.warning("Calling start() on a started span.")
-            return
+            self._start_time = (
+                start_time if start_time is not None else time_ns()
+            )
+
         self.span_processor.on_start(self, parent_context=parent_context)
 
     def end(self, end_time: Optional[int] = None) -> None:
         with self._lock:
-            if not self.is_recording():
-                return
             if self.start_time is None:
                 raise RuntimeError("Calling end() on a not started span.")
-            has_ended = self.end_time is not None
-            if not has_ended:
-                if self.status is None:
-                    self.status = Status(canonical_code=StatusCanonicalCode.OK)
+            if self.end_time is not None:
+                logger.warning("Calling end() on an ended span.")
+                return
 
-                self._end_time = (
-                    end_time if end_time is not None else time_ns()
-                )
+            if self.status is None:
+                self.status = Status(canonical_code=StatusCanonicalCode.OK)
 
-        if has_ended:
-            logger.warning("Calling end() on an ended span.")
-            return
+            self._end_time = end_time if end_time is not None else time_ns()
 
         self.span_processor.on_end(self)
 
+    @_check_span_ended
     def update_name(self, name: str) -> None:
-        with self._lock:
-            has_ended = self.end_time is not None
-        if has_ended:
-            logger.warning("Calling update_name() on an ended span.")
-            return
         self.name = name
 
     def is_recording(self) -> bool:
         return True
 
+    @_check_span_ended
     def set_status(self, status: trace_api.Status) -> None:
-        with self._lock:
-            has_ended = self.end_time is not None
-        if has_ended:
-            logger.warning("Calling set_status() on an ended span.")
-            return
         self.status = status
 
     def __exit__(
