[sdk-logs+traces] Add internal processor weight api (#5424)

Author: CodeBlanch

src/OpenTelemetry/BaseProcessor.cs

@@ -27,6 +27,18 @@ public BaseProcessor()
     /// </summary>
     public BaseProvider? ParentProvider { get; private set; }
 
+    /// <summary>
+    /// Gets or sets the weight of the processor when added to the provider
+    /// pipeline. Default value: <c>0</c>.
+    /// </summary>
+    /// <remarks>
+    /// Note: Weight is used to order processors when building a provider
+    /// pipeline. Lower weighted processors come before higher weighted
+    /// processors. Changing the weight after a pipeline has been constructed
+    /// has no effect.
+    /// </remarks>
+    internal int PipelineWeight { get; set; }
+
     /// <summary>
     /// Called synchronously when a telemetry object is started.
     /// </summary>

src/OpenTelemetry/Logs/LoggerProviderSdk.cs

@@ -54,7 +54,8 @@ public LoggerProviderSdk(
         resourceBuilder.ServiceProvider = serviceProvider;
         this.Resource = resourceBuilder.Build();
 
-        foreach (var processor in state.Processors)
+        // Note: Linq OrderBy performs a stable sort, which is a requirement here
+        foreach (var processor in state.Processors.OrderBy(p => p.PipelineWeight))
         {
             this.AddProcessor(processor);
         }

src/OpenTelemetry/Trace/Builder/TracerProviderBuilderSdk.cs

@@ -162,13 +162,13 @@ public TracerProviderBuilder ConfigureServices(Action<IServiceCollection> config
         throw new NotSupportedException("Services cannot be configured after ServiceProvider has been created.");
     }
 
-    public void AddExceptionProcessorIfEnabled()
+    public void AddExceptionProcessorIfEnabled(ref IEnumerable<BaseProcessor<Activity>> processors)
     {
         if (this.ExceptionProcessorEnabled)
         {
             try
             {
-                this.Processors.Insert(0, new ExceptionProcessor());
+                processors = new BaseProcessor<Activity>[] { new ExceptionProcessor() }.Concat(processors);
             }
             catch (Exception ex)
             {

src/OpenTelemetry/Trace/TracerProviderSdk.cs

@@ -53,8 +53,6 @@ internal TracerProviderSdk(
         StringBuilder processorsAdded = new StringBuilder();
         StringBuilder instrumentationFactoriesAdded = new StringBuilder();
 
-        state.AddExceptionProcessorIfEnabled();
-
         var resourceBuilder = state.ResourceBuilder ?? ResourceBuilder.CreateDefault();
         resourceBuilder.ServiceProvider = serviceProvider;
         this.Resource = resourceBuilder.Build();
@@ -74,7 +72,12 @@ internal TracerProviderSdk(
             }
         }
 
-        foreach (var processor in state.Processors)
+        // Note: Linq OrderBy performs a stable sort, which is a requirement here
+        IEnumerable<BaseProcessor<Activity>> processors = state.Processors.OrderBy(p => p.PipelineWeight);
+
+        state.AddExceptionProcessorIfEnabled(ref processors);
+
+        foreach (var processor in processors)
         {
             this.AddProcessor(processor);
             processorsAdded.Append(processor.GetType());

test/OpenTelemetry.Tests/Logs/LoggerProviderBuilderExtensionsTests.cs

@@ -127,16 +127,65 @@ public void LoggerProviderBuilderConfigureResourceBuilderTests()
         Assert.Contains(provider.Resource.Attributes, value => value.Key == "key1" && (string)value.Value == "value1");
     }
 
+    [Fact]
+    public void LoggerProviderBuilderUsingDependencyInjectionTest()
+    {
+        using var provider = Sdk.CreateLoggerProviderBuilder()
+            .AddProcessor<CustomProcessor>()
+            .AddProcessor<CustomProcessor>()
+            .Build() as LoggerProviderSdk;
+
+        Assert.NotNull(provider);
+
+        var processors = ((IServiceProvider)provider.OwnedServiceProvider!).GetServices<CustomProcessor>();
+
+        // Note: Two "Add" calls but it is a singleton so only a single registration is produced
+        Assert.Single(processors);
+
+        var processor = provider.Processor as CompositeProcessor<LogRecord>;
+
+        Assert.NotNull(processor);
+
+        // Note: Two "Add" calls due yield two processors added to provider, even though they are the same
+        Assert.True(processor.Head.Value is CustomProcessor);
+        Assert.True(processor.Head.Next?.Value is CustomProcessor);
+    }
+
     [Fact]
     public void LoggerProviderBuilderAddProcessorTest()
     {
-        List<CustomProcessor> processors = new();
+        List<CustomProcessor> processorsToAdd = new()
+        {
+            new CustomProcessor()
+            {
+                Name = "A",
+            },
+            new CustomProcessor()
+            {
+                Name = "B",
+            },
+            new CustomProcessor()
+            {
+                Name = "C",
+            },
+        };
 
-        using (var provider = Sdk.CreateLoggerProviderBuilder()
-            .AddProcessor<CustomProcessor>()
-            .AddProcessor(sp => new CustomProcessor())
-            .AddProcessor(new CustomProcessor())
-            .Build() as LoggerProviderSdk)
+        var builder = Sdk.CreateLoggerProviderBuilder();
+        foreach (var processor in processorsToAdd)
+        {
+            builder.AddProcessor(processor);
+        }
+
+        List<CustomProcessor> expectedProcessors = new()
+        {
+            processorsToAdd.First(p => p.Name == "A"),
+            processorsToAdd.First(p => p.Name == "B"),
+            processorsToAdd.First(p => p.Name == "C"),
+        };
+
+        List<CustomProcessor> actualProcessors = new();
+
+        using (var provider = builder.Build() as LoggerProviderSdk)
         {
             Assert.NotNull(provider);
             Assert.NotNull(provider.Processor);
@@ -151,16 +200,106 @@ public void LoggerProviderBuilderAddProcessorTest()
                 var processor = current.Value as CustomProcessor;
                 Assert.NotNull(processor);
 
-                processors.Add(processor);
+                actualProcessors.Add(processor);
                 Assert.False(processor.Disposed);
 
                 current = current.Next;
             }
+
+            Assert.Equal(expectedProcessors, actualProcessors);
+        }
+
+        foreach (var processor in actualProcessors)
+        {
+            Assert.True(processor.Disposed);
+        }
+    }
+
+    [Fact]
+    public void LoggerProviderBuilderAddProcessorWithWeightTest()
+    {
+        List<CustomProcessor> processorsToAdd = new()
+        {
+            new CustomProcessor()
+            {
+                Name = "C",
+                PipelineWeight = 0,
+            },
+            new CustomProcessor()
+            {
+                Name = "E",
+                PipelineWeight = 10_000,
+            },
+            new CustomProcessor()
+            {
+                Name = "B",
+                PipelineWeight = -10_000,
+            },
+            new CustomProcessor()
+            {
+                Name = "F",
+                PipelineWeight = int.MaxValue,
+            },
+            new CustomProcessor()
+            {
+                Name = "A",
+                PipelineWeight = int.MinValue,
+            },
+            new CustomProcessor()
+            {
+                Name = "D",
+                PipelineWeight = 0,
+            },
+        };
+
+        var builder = Sdk.CreateLoggerProviderBuilder();
+        foreach (var processor in processorsToAdd)
+        {
+            builder.AddProcessor(processor);
         }
 
-        Assert.Equal(3, processors.Count);
+        List<CustomProcessor> expectedProcessors = new()
+        {
+            processorsToAdd.First(p => p.Name == "A"),
+            processorsToAdd.First(p => p.Name == "B"),
+            processorsToAdd.First(p => p.Name == "C"),
+            processorsToAdd.First(p => p.Name == "D"),
+            processorsToAdd.First(p => p.Name == "E"),
+            processorsToAdd.First(p => p.Name == "F"),
+        };
+
+        List<CustomProcessor> actualProcessors = new();
+
+        using (var provider = builder.Build() as LoggerProviderSdk)
+        {
+            Assert.NotNull(provider);
+            Assert.NotNull(provider.Processor);
+
+            var compositeProcessor = provider.Processor as CompositeProcessor<LogRecord>;
+
+            Assert.NotNull(compositeProcessor);
+
+            var lastWeight = int.MinValue;
+            var current = compositeProcessor.Head;
+            while (current != null)
+            {
+                var processor = current.Value as CustomProcessor;
+                Assert.NotNull(processor);
+
+                actualProcessors.Add(processor);
+                Assert.False(processor.Disposed);
+
+                Assert.True(processor.PipelineWeight >= lastWeight);
+
+                lastWeight = processor.PipelineWeight;
+
+                current = current.Next;
+            }
+
+            Assert.Equal(expectedProcessors, actualProcessors);
+        }
 
-        foreach (var processor in processors)
+        foreach (var processor in actualProcessors)
         {
             Assert.True(processor.Disposed);
         }
@@ -187,6 +326,7 @@ public void Dispose()
 
     private sealed class CustomProcessor : BaseProcessor<LogRecord>
     {
+        public string? Name;
         public bool Disposed;
 
         protected override void Dispose(bool disposing)