re-enabling FunctionResultAggregator

Author: brettsam

src/WebJobs.Script.WebHost/Diagnostics/FunctionInstanceLogCollectorProvider.cs

@@ -0,0 +1,38 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using Microsoft.Azure.WebJobs.Host.Executors;
+using Microsoft.Azure.WebJobs.Host.Loggers;
+using Microsoft.Azure.WebJobs.Script.Diagnostics;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics
+{
+    internal class FunctionInstanceLogCollectorProvider : IEventCollectorProvider
+    {
+        private readonly IFunctionMetadataManager _metadataManager;
+        private readonly IProxyMetadataManager _proxyMetadataManager;
+        private readonly IMetricsLogger _metrics;
+        private readonly IHostIdProvider _hostIdProvider;
+        private readonly IConfiguration _configuration;
+        private readonly ILoggerFactory _loggerFactory;
+
+        public FunctionInstanceLogCollectorProvider(IFunctionMetadataManager metadataManager, IProxyMetadataManager proxyMetadataManager,
+            IMetricsLogger metrics, IHostIdProvider hostIdProvider, IConfiguration configuration, ILoggerFactory loggerFactory)
+        {
+            _metadataManager = metadataManager ?? throw new ArgumentNullException(nameof(metadataManager));
+            _proxyMetadataManager = proxyMetadataManager ?? throw new ArgumentNullException(nameof(proxyMetadataManager));
+            _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
+            _hostIdProvider = hostIdProvider ?? throw new ArgumentNullException(nameof(hostIdProvider));
+            _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
+            _loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
+        }
+
+        public IAsyncCollector<FunctionInstanceLogEntry> Create()
+        {
+            return new FunctionInstanceLogger(_metadataManager, _proxyMetadataManager, _metrics, _hostIdProvider, _configuration, _loggerFactory);
+        }
+    }
+}

src/WebJobs.Script.WebHost/WebScriptHostBuilderExtension.cs

@@ -4,6 +4,7 @@
 using System;
 using Microsoft.Azure.WebJobs.Host.Config;
 using Microsoft.Azure.WebJobs.Host.Executors;
+using Microsoft.Azure.WebJobs.Host.Loggers;
 using Microsoft.Azure.WebJobs.Host.Timers;
 using Microsoft.Azure.WebJobs.Script.Diagnostics;
 using Microsoft.Azure.WebJobs.Script.WebHost.Configuration;
@@ -72,7 +73,7 @@ public static IHostBuilder AddWebScriptHost(this IHostBuilder builder, IServiceP
 
                     // Logging and diagnostics
                     services.TryAddSingleton<IMetricsLogger, WebHostMetricsLogger>();
-                    services.AddSingleton<IAsyncCollector<Host.Loggers.FunctionInstanceLogEntry>, FunctionInstanceLogger>();
+                    services.AddSingleton<IEventCollectorProvider, FunctionInstanceLogCollectorProvider>();
 
                     // Hosted services
                     services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService, HttpInitializationService>());

test/WebJobs.Script.Tests.Integration/WebHostEndToEnd/CSharpEndToEndTests.cs

@@ -10,6 +10,7 @@
 using System.Net.Http.Headers;
 using System.Text;
 using System.Threading.Tasks;
+using Microsoft.Azure.WebJobs.Logging;
 using Microsoft.Azure.WebJobs.Script.Diagnostics;
 using Microsoft.Azure.WebJobs.Script.Rpc;
 using Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics;
@@ -111,6 +112,14 @@ await TestHelpers.Await(() =>
             // Make sure we get a metric logged from both ILogger and TraceWriter
             var key = MetricsEventManager.GetAggregateKey(MetricEventNames.FunctionUserLog, "Scenarios");
             Assert.Equal(2, Fixture.MetricsLogger.LoggedEvents.Where(p => p == key).Count());
+
+            // Make sure we've gotten a log from the aggregator
+            IEnumerable<LogMessage> getAggregatorLogs() => Fixture.Host.GetLogMessages().Where(p => p.Category == LogCategories.Aggregator);
+
+            await TestHelpers.Await(() => getAggregatorLogs().Any());
+
+            var aggregatorLogs = getAggregatorLogs();
+            Assert.Equal(1, aggregatorLogs.Count());
         }
 
         [Fact]
@@ -405,6 +414,11 @@ public override void ConfigureJobHost(IWebJobsBuilder webJobsBuilder)
                         "Scenarios"
                     };
                 });
+
+                webJobsBuilder.Services.Configure<FunctionResultAggregatorOptions>(o =>
+                {
+                    o.BatchSize = 1;
+                });
             }
         }
 

test/WebJobs.Script.Tests/Configuration/AggregatorConfigurationTests.cs

@@ -3,8 +3,12 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using Microsoft.Azure.WebJobs.Host.Loggers;
 using Microsoft.Azure.WebJobs.Logging;
 using Microsoft.Azure.WebJobs.Script.Configuration;
+using Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -16,17 +20,42 @@ namespace Microsoft.Azure.WebJobs.Script.Tests.Configuration
 {
     public class AggregatorConfigurationTests
     {
+        private static readonly string AggregatorPath = ConfigurationPath.Combine(ConfigurationSectionNames.JobHost, ConfigurationSectionNames.Aggregator);
+
+        [Fact]
+        public void Aggregator_Registered_ByDefault()
+        {
+            IHost host = new HostBuilder()
+                .ConfigureDefaultTestWebScriptHost()
+                .Build();
+
+            // Make sure there are two providers registered, and that one has a type with
+            // "FunctionResultAggregatorProvider" (since it is internal to WebJobs)
+            var eventCollectorProviders = host.Services.GetServices<IEventCollectorProvider>();
+            Assert.Equal(2, eventCollectorProviders.Count());
+            Assert.Single(eventCollectorProviders.OfType<FunctionInstanceLogCollectorProvider>());
+            Assert.Single(eventCollectorProviders.Where(p => p.GetType().Name.Contains("FunctionResultAggregatorProvider")));
+
+            // Also make sure that when requesting the collectors, we end up with a composite that
+            // includes both of the collectors above. This prevents any overriding of the IEventCollectorFactory
+            var eventLogger = host.Services.GetServices<IAsyncCollector<FunctionInstanceLogEntry>>().Single();
+            var field = eventLogger.GetType().GetField("_collectors", BindingFlags.NonPublic | BindingFlags.Instance);
+            var collectors = (IEnumerable<IAsyncCollector<FunctionInstanceLogEntry>>)field.GetValue(eventLogger);
+            Assert.Equal(2, collectors.Count());
+            Assert.Single(collectors.OfType<FunctionInstanceLogger>());
+            Assert.Single(collectors.Where(p => p.GetType().Name.Contains("FunctionResultAggregator")));
+        }
+
         [Fact]
         public void Configuration_BindsTo_AggregatorOptions()
         {
-            string aggregatorPath = ConfigurationPath.Combine(ConfigurationSectionNames.JobHost, ConfigurationSectionNames.Aggregator);
             IHost host = new HostBuilder()
                 .ConfigureAppConfiguration(c =>
                 {
                     c.AddInMemoryCollection(new Dictionary<string, string>
                     {
-                        { ConfigurationPath.Combine(aggregatorPath, "BatchSize"), "33" },
-                        { ConfigurationPath.Combine(aggregatorPath, "FlushTimeout"), "00:00:33" }
+                        { ConfigurationPath.Combine(AggregatorPath, "BatchSize"), "33" },
+                        { ConfigurationPath.Combine(AggregatorPath, "FlushTimeout"), "00:00:33" }
                     });
                 })
                 .ConfigureDefaultTestWebScriptHost()