Merge main into stable/4.0.x. Update 20231101

* commit '3d217923ab5ed5a4a4bae8f334f6e8404ab884be':
  I260 Write OpenTelemetry span into redis (#263)

datastore/shared/util/otel.py

@@ -1,4 +1,5 @@
 from contextlib import nullcontext
+from typing import Any, Dict
 
 from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
@@ -12,6 +13,7 @@
 
 
 OTEL_DATA_FIELD_KEY = "__otel_data"
+otel_initialized = False
 
 
 def is_otel_enabled():
@@ -28,7 +30,6 @@ def init(service_name):
     """
     if not is_otel_enabled():
         return
-
     span_exporter = OTLPSpanExporter(
         endpoint="http://collector:4317",
         insecure=True
@@ -42,6 +43,8 @@ def init(service_name):
     trace.set_tracer_provider(tracer_provider)
     span_processor = BatchSpanProcessor(span_exporter)
     tracer_provider.add_span_processor(span_processor)
+    global otel_initialized
+    otel_initialized = True
 
 
 def instrument_flask(app):
@@ -66,7 +69,23 @@ def make_span(name, attributes=None):
     if not is_otel_enabled():
         return nullcontext()
 
+    global otel_initialized
+    assert (
+        otel_initialized
+    ), "datastore:Opentelemetry span to be set before having set a TRACER_PROVIDER"
+
     tracer = trace.get_tracer_provider().get_tracer(__name__)
     span = tracer.start_as_current_span(name, attributes=attributes)
 
     return span
+
+
+def inject_otel_data(fields: Dict[str, Any]) -> None:
+    if not is_otel_enabled():
+        return
+
+    span_context = trace.get_current_span().get_span_context()
+    trace_id = span_context.trace_id
+    span_id = span_context.span_id
+    span_data = f"{trace_id}:{span_id}:{span_context.trace_flags}"
+    fields[OTEL_DATA_FIELD_KEY] = span_data

datastore/writer/redis_backend/redis_messaging_backend_service.py

@@ -12,6 +12,7 @@
     fqfield_from_fqid_and_field,
     logger,
 )
+from datastore.shared.util.otel import inject_otel_data
 from datastore.writer.core import Messaging
 
 from .connection_handler import ConnectionHandler
@@ -30,6 +31,7 @@ def handle_events(
         log_all_modified_fields: bool = True,
     ) -> None:
         modified_fqfields = self.get_modified_fqfields(events_per_position)
+        inject_otel_data(modified_fqfields)
         if log_all_modified_fields:
             logger.debug(
                 f"written fqfields into {MODIFIED_FIELDS_TOPIC}: "