composite filter init

Author: AgraVator

core/src/main/java/io/grpc/internal/ManagedChannelImpl.java

@@ -1115,6 +1115,7 @@ public void start(Listener<RespT> observer, Metadata headers) {
         callOptions = callOptions.withOption(MethodInfo.KEY, methodInfo);
       }
       if (interceptor != null) {
+        // todo: AgraVator this is where the interceptors are applied
         delegate = interceptor.interceptCall(method, callOptions, channel);
       } else {
         delegate = channel.newCall(method, callOptions);

opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java

@@ -101,7 +101,7 @@ private GrpcOpenTelemetry(Builder builder) {
     this.optionalLabels = ImmutableList.copyOf(builder.optionalLabels);
     this.openTelemetryMetricsModule = new OpenTelemetryMetricsModule(
         STOPWATCH_SUPPLIER, resource, optionalLabels, builder.plugins,
-        builder.targetFilter);
+        openTelemetrySdk.getPropagators(), builder.targetFilter);
     this.openTelemetryTracingModule = new OpenTelemetryTracingModule(openTelemetrySdk);
     this.sink = new OpenTelemetryMetricSink(meter, enableMetrics, disableDefault, optionalLabels);
   }

opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryMetricsModule.java

@@ -49,6 +49,7 @@
 import io.opentelemetry.api.baggage.Baggage;
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.ContextPropagators;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -100,24 +101,28 @@ final class OpenTelemetryMetricsModule {
   private final boolean localityEnabled;
   private final boolean backendServiceEnabled;
   private final ImmutableList<OpenTelemetryPlugin> plugins;
+  private final ContextPropagators contextPropagators;
   @Nullable
   private final TargetFilter targetAttributeFilter;
 
   OpenTelemetryMetricsModule(Supplier<Stopwatch> stopwatchSupplier,
                              OpenTelemetryMetricsResource resource,
-                             Collection<String> optionalLabels, List<OpenTelemetryPlugin> plugins) {
-    this(stopwatchSupplier, resource, optionalLabels, plugins, null);
+      Collection<String> optionalLabels, List<OpenTelemetryPlugin> plugins,
+      ContextPropagators contextPropagators) {
+    this(stopwatchSupplier, resource, optionalLabels, plugins, contextPropagators, null);
   }
 
   OpenTelemetryMetricsModule(Supplier<Stopwatch> stopwatchSupplier,
       OpenTelemetryMetricsResource resource,
       Collection<String> optionalLabels, List<OpenTelemetryPlugin> plugins,
-      @Nullable TargetFilter targetAttributeFilter) {
+      ContextPropagators contextPropagators,
+          @Nullable TargetFilter targetAttributeFilter) {
     this.resource = checkNotNull(resource, "resource");
     this.stopwatchSupplier = checkNotNull(stopwatchSupplier, "stopwatchSupplier");
     this.localityEnabled = optionalLabels.contains(LOCALITY_KEY.getKey());
     this.backendServiceEnabled = optionalLabels.contains(BACKEND_SERVICE_KEY.getKey());
     this.plugins = ImmutableList.copyOf(plugins);
+    this.contextPropagators = checkNotNull(contextPropagators, "contextPropagators");
     this.targetAttributeFilter = targetAttributeFilter;
   }
 
@@ -159,10 +164,13 @@ static String recordMethodName(String fullMethodName, boolean isGeneratedMethod)
     return isGeneratedMethod ? fullMethodName : "other";
   }
 
-  private static Context otelContextWithBaggage() {
-    Baggage baggage = BAGGAGE_KEY.get();
+  private static Context otelContextWithBaggage(@Nullable Baggage baggage) {
     if (baggage == null) {
-      return Context.current();
+      Baggage currentBaggage = BAGGAGE_KEY.get();
+      if (currentBaggage == null) {
+        return Context.current();
+      }
+      return Context.current().with(currentBaggage);
     }
     return Context.current().with(baggage);
   }
@@ -282,7 +290,7 @@ public void streamClosed(Status status) {
     }
 
     void recordFinishedAttempt() {
-      Context otelContext = otelContextWithBaggage();
+      Context otelContext = otelContextWithBaggage(null);
       AttributesBuilder builder = io.opentelemetry.api.common.Attributes.builder()
           .put(METHOD_KEY, fullMethodName)
           .put(TARGET_KEY, target)
@@ -448,7 +456,7 @@ void callEnded(Status status) {
     }
 
     void recordFinishedCall() {
-      Context otelContext = otelContextWithBaggage();
+      Context otelContext = otelContextWithBaggage(null);
       if (attemptsPerCall.get() == 0) {
         ClientTracer tracer = newClientTracer(null);
         tracer.attemptNanos = attemptDelayStopwatch.elapsed(TimeUnit.NANOSECONDS);
@@ -553,13 +561,15 @@ private static final class ServerTracer extends ServerStreamTracer {
     private final Stopwatch stopwatch;
     private volatile long outboundWireSize;
     private volatile long inboundWireSize;
+    private final Baggage baggage;
 
     ServerTracer(OpenTelemetryMetricsModule module, String fullMethodName,
-        List<OpenTelemetryPlugin.ServerStreamPlugin> streamPlugins) {
+        List<OpenTelemetryPlugin.ServerStreamPlugin> streamPlugins, @Nullable Baggage baggage) {
       this.module = checkNotNull(module, "module");
       this.fullMethodName = fullMethodName;
       this.streamPlugins = checkNotNull(streamPlugins, "streamPlugins");
       this.stopwatch = module.stopwatchSupplier.get().start();
+      this.baggage = baggage;
     }
 
     @Override
@@ -606,7 +616,7 @@ public void inboundWireSize(long bytes) {
      */
     @Override
     public void streamClosed(Status status) {
-      Context otelContext = otelContextWithBaggage();
+      Context otelContext = otelContextWithBaggage(baggage);
       if (streamClosedUpdater != null) {
         if (streamClosedUpdater.getAndSet(this, 1) != 0) {
           return;
@@ -657,7 +667,10 @@ public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata
         }
         streamPlugins = Collections.unmodifiableList(streamPluginsMutable);
       }
-      return new ServerTracer(OpenTelemetryMetricsModule.this, fullMethodName, streamPlugins);
+      Context context = contextPropagators.getTextMapPropagator().extract(
+          Context.current(), headers, MetadataGetter.getInstance());
+      Baggage baggage = Baggage.fromContext(context);
+      return new ServerTracer(OpenTelemetryMetricsModule.this, fullMethodName, streamPlugins, baggage);
     }
   }