interop: fix C-core metadata publication and Python OTel tracing context

Author: AgraVator

src/core/call/metadata_batch.h

@@ -317,7 +317,7 @@ struct GrpcServerStatsBinMetadata : public SimpleSliceBasedMetadata {
 
 // grpc-trace-bin metadata trait.
 struct GrpcTraceBinMetadata : public SimpleSliceBasedMetadata {
-  static constexpr bool kPublishToApp = false;
+  static constexpr bool kPublishToApp = true;
   static constexpr bool kRepeatable = false;
   static constexpr bool kTransferOnTrailersOnly = false;
   using CompressionTraits = FrequentKeyWithNoValueCompressionCompressor;
@@ -326,7 +326,7 @@ struct GrpcTraceBinMetadata : public SimpleSliceBasedMetadata {
 
 // grpc-tags-bin metadata trait.
 struct GrpcTagsBinMetadata : public SimpleSliceBasedMetadata {
-  static constexpr bool kPublishToApp = false;
+  static constexpr bool kPublishToApp = true;
   static constexpr bool kRepeatable = false;
   static constexpr bool kTransferOnTrailersOnly = false;
   using CompressionTraits = FrequentKeyWithNoValueCompressionCompressor;

src/core/lib/surface/call_utils.h

@@ -112,6 +112,12 @@ class PublishToAppEncoder {
       if constexpr (std::is_same<XForwardedHostMetadata, Which>::value) {
         Append(Which::key(), value);
       }
+      if constexpr (std::is_same<GrpcTraceBinMetadata, Which>::value) {
+        Append(Which::key(), value);
+      }
+      if constexpr (std::is_same<GrpcTagsBinMetadata, Which>::value) {
+        Append(Which::key(), value);
+      }
     }
   }
 

src/python/grpcio_tests/tests/interop/client.py

@@ -285,22 +285,14 @@ def _test_case_from_arg(test_case_arg):
 
 
 def test_interoperability(args):
+    channel = _create_channel(args)
+    stub = create_stub(channel, args)
+    test_case = _test_case_from_arg(args.test_case)
+    test_case.test_interoperability(stub, args)
     if args.enable_opentelemetry:
-        import grpc_observability
-
         from tests.interop import otel_interop_helper
 
-        with grpc_observability.OpenTelemetryPlugin():
-            channel = _create_channel(args)
-            stub = create_stub(channel, args)
-            test_case = _test_case_from_arg(args.test_case)
-            test_case.test_interoperability(stub, args)
-            otel_interop_helper.flush_tracer_provider()
-    else:
-        channel = _create_channel(args)
-        stub = create_stub(channel, args)
-        test_case = _test_case_from_arg(args.test_case)
-        test_case.test_interoperability(stub, args)
+        otel_interop_helper.flush_tracer_provider()
 
 
 if __name__ == "__main__":

src/python/grpcio_tests/tests/interop/otel_interop_helper.py

@@ -106,8 +106,8 @@ def export(self, spans: Tuple[ReadableSpan, ...]) -> None:
 
         try:
             self._stub.Export(request, timeout=5)
-        except Exception:
-            pass
+        except Exception as e:
+            print(f"OTLPSpanExporter Export exception: {e}", flush=True)
 
     def shutdown(self) -> None:
         self._channel.close()

src/python/grpcio_tests/tests/interop/server.py

@@ -121,13 +121,7 @@ def serve(args):
     else:
         server.add_insecure_port("[::]:{}".format(args.port))
 
-    if args.enable_opentelemetry:
-        import grpc_observability
-
-        with grpc_observability.OpenTelemetryPlugin():
-            _serve_internal(server, enable_otel=True)
-    else:
-        _serve_internal(server)
+    _serve_internal(server, enable_otel=args.enable_opentelemetry)
 
 
 if __name__ == "__main__":

test/cpp/interop/run_otel_interop_test.py

@@ -35,6 +35,10 @@ def get_free_port():
 def run_cmd(args, desc, env=None, cwd=None):
     print(f"Executing: {' '.join(args)} ({desc})")
     proc_env = os.environ.copy()
+    if "CC" not in proc_env and os.path.exists("/usr/bin/gcc"):
+        proc_env["CC"] = "/usr/bin/gcc"
+    if "CXX" not in proc_env and os.path.exists("/usr/bin/g++"):
+        proc_env["CXX"] = "/usr/bin/g++"
     if env:
         proc_env.update(env)
     res = subprocess.run(
@@ -52,6 +56,10 @@ def start_proc(args, env, desc, cwd=None):
     print(f"Starting in background: {' '.join(args)} ({desc})")
     # Inherit system environment and merge with custom variables
     proc_env = os.environ.copy()
+    if "CC" not in proc_env and os.path.exists("/usr/bin/gcc"):
+        proc_env["CC"] = "/usr/bin/gcc"
+    if "CXX" not in proc_env and os.path.exists("/usr/bin/g++"):
+        proc_env["CXX"] = "/usr/bin/g++"
     proc_env.update(env)
     return subprocess.Popen(
         args,
@@ -294,6 +302,7 @@ def main():
                     ":grpc-interop-testing:installDist",
                     "-x",
                     "test",
+                    "-PskipCodegen=true",
                 ],
                 "Building Java interop targets",
                 cwd="../grpc-java",
@@ -331,24 +340,28 @@ def main():
         )
         time.sleep(1)  # wait for collector to start listening
 
-        # Start Server
-        env = {
-            "GRPC_EXPERIMENTAL_ENABLE_OTEL_TRACING": "true",
+        # Base env for OTLP exporter
+        base_env = {
             "OTEL_EXPORTER_OTLP_ENDPOINT": f"http://localhost:{collector_port}",
             "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc",
             "OTEL_TRACES_EXPORTER": "otlp",
             "OTEL_METRICS_EXPORTER": "none",
             "OTEL_LOGS_EXPORTER": "none",
             "GRPC_BAZEL_RUNTIME": "1",
         }
+
+        server_env = base_env.copy()
+        if args.server in ("c++", "java"):
+            server_env["GRPC_EXPERIMENTAL_ENABLE_OTEL_TRACING"] = "true"
+
         if args.server == "c++":
             server_proc = start_proc(
                 [
                     "./bazel-bin/test/cpp/interop/interop_server",
                     f"--port={server_port}",
                     "--enable_opentelemetry=true",
                 ],
-                env,
+                server_env,
                 "C++ Interop Server",
             )
         elif args.server == "java":
@@ -359,7 +372,7 @@ def main():
                     "--use_tls=false",
                     "--enable_opentelemetry=true",
                 ],
-                env,
+                server_env,
                 "Java Interop Server",
             )
         elif args.server == "python":
@@ -370,13 +383,17 @@ def main():
                     "--use_tls=false",
                     "--enable_opentelemetry=true",
                 ],
-                env,
+                server_env,
                 "Python Interop Server",
             )
         time.sleep(3)  # wait for server to bind and start
 
         # Run Client
         print(f"Running {args.client.upper()} Client...")
+        client_env = base_env.copy()
+        if args.client in ("c++", "java"):
+            client_env["GRPC_EXPERIMENTAL_ENABLE_OTEL_TRACING"] = "true"
+
         if args.client == "c++":
             client_res = run_cmd(
                 [
@@ -387,7 +404,7 @@ def main():
                     "--enable_opentelemetry=true",
                 ],
                 "Running C++ Interop Client",
-                env=env,
+                env=client_env,
             )
         elif args.client == "java":
             client_res = run_cmd(
@@ -400,7 +417,7 @@ def main():
                     "--enable_opentelemetry=true",
                 ],
                 "Running Java Interop Client",
-                env=env,
+                env=client_env,
             )
         elif args.client == "python":
             client_res = run_cmd(
@@ -413,7 +430,7 @@ def main():
                     "--enable_opentelemetry=true",
                 ],
                 "Running Python Interop Client",
-                env=env,
+                env=client_env,
             )
 
         print("Client finished. Waiting for spans to flush...")