Fixing MetricsEventManager memory leak (#6818)

Author: mathewc

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

@@ -7,7 +7,6 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Threading;
-using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs.Script.Configuration;
 using Microsoft.Azure.WebJobs.Script.Description;
 using Microsoft.Azure.WebJobs.Script.Diagnostics;
@@ -45,7 +44,7 @@ public MetricsEventManager(IOptionsMonitor<AppServiceOptions> appServiceOptions,
             // Initialize the periodic log flush timer
             _metricsFlushTimer = new Timer(TimerFlush, null, metricsFlushIntervalMS, metricsFlushIntervalMS);
 
-            _functionActivityTracker = new FunctionActivityTracker(_appServiceOptions, _eventGenerator, metricsPublisher, linuxContainerActivityPublisher, _functionActivityFlushIntervalSeconds);
+            _functionActivityTracker = new FunctionActivityTracker(_appServiceOptions, _eventGenerator, metricsPublisher, linuxContainerActivityPublisher, _functionActivityFlushIntervalSeconds, _logger);
         }
 
         /// <summary>
@@ -316,7 +315,7 @@ protected virtual void Dispose(bool disposing)
 
                     if (_functionActivityTracker != null)
                     {
-                        _functionActivityTracker.StopEtwTaskAndRaiseFinishedEvent();
+                        _functionActivityTracker.StopTimerAndRaiseFinishedEvent();
                         _functionActivityTracker.Dispose();
                     }
                 }
@@ -334,24 +333,30 @@ public void Dispose()
 
         private class FunctionActivityTracker : IDisposable
         {
+            // this interval should stay at 1 second because the timer is also
+            // used to emit events every Nth second
+            private const int _activityTimerIntervalMS = 1000;
+
             private readonly IMetricsPublisher _metricsPublisher;
             private readonly ILinuxContainerActivityPublisher _linuxContainerActivityPublisher;
             private readonly object _runningFunctionsSyncLock = new object();
+            private readonly Timer _activityTimer;
+            private readonly ILogger<MetricsEventManager> _logger;
 
             private ulong _totalExecutionCount = 0;
             private int _activeFunctionCount = 0;
             private int _functionActivityFlushInterval;
-            private CancellationTokenSource _etwTaskCancellationSource = new CancellationTokenSource();
             private ConcurrentQueue<FunctionMetrics> _functionMetricsQueue = new ConcurrentQueue<FunctionMetrics>();
             private List<FunctionStartedEvent> _runningFunctions = new List<FunctionStartedEvent>();
             private bool _disposed = false;
             private IOptionsMonitor<AppServiceOptions> _appServiceOptions;
+            private int _activityFlushCounter;
 
             // This ID is just an event grouping mechanism that can be used by event consumers
             // to group events coming from the same app host.
             private string _executionId = Guid.NewGuid().ToString();
 
-            internal FunctionActivityTracker(IOptionsMonitor<AppServiceOptions> appServiceOptions, IEventGenerator generator, IMetricsPublisher metricsPublisher, ILinuxContainerActivityPublisher linuxContainerActivityPublisher, int functionActivityFlushInterval)
+            internal FunctionActivityTracker(IOptionsMonitor<AppServiceOptions> appServiceOptions, IEventGenerator generator, IMetricsPublisher metricsPublisher, ILinuxContainerActivityPublisher linuxContainerActivityPublisher, int functionActivityFlushInterval, ILogger<MetricsEventManager> logger)
             {
                 MetricsEventGenerator = generator;
                 _appServiceOptions = appServiceOptions;
@@ -367,43 +372,37 @@ internal FunctionActivityTracker(IOptionsMonitor<AppServiceOptions> appServiceOp
                     _metricsPublisher = metricsPublisher;
                 }
 
-                StartActivityTimer();
+                _activityFlushCounter = _functionActivityFlushInterval;
+                _activityTimer = new Timer(TimerFlush, null, _activityTimerIntervalMS, _activityTimerIntervalMS);
+
+                _logger = logger;
             }
 
             internal IEventGenerator MetricsEventGenerator { get; private set; }
 
-            private void StartActivityTimer()
+            private void TimerFlush(object state)
             {
-                Task.Run(
-                    async () =>
+                try
+                {
+                    // we raise these events every interval as needed
+                    RaiseMetricsPerFunctionEvent();
+
+                    // only raise these events every Nth interval
+                    if (_activityFlushCounter >= _functionActivityFlushInterval)
                     {
-                        try
-                        {
-                            int currentSecond = _functionActivityFlushInterval;
-                            while (!_etwTaskCancellationSource.Token.IsCancellationRequested)
-                            {
-                                RaiseMetricsPerFunctionEvent();
-
-                                if (currentSecond >= _functionActivityFlushInterval)
-                                {
-                                    RaiseFunctionMetricEvents();
-                                    currentSecond = 0;
-                                }
-                                else
-                                {
-                                    currentSecond = currentSecond + 1;
-                                }
-
-                                await Task.Delay(TimeSpan.FromSeconds(1), _etwTaskCancellationSource.Token);
-                            }
-                        }
-                        catch (TaskCanceledException)
-                        {
-                            // This exception gets throws when cancellation request is raised via cancellation token.
-                            // Let's eat this exception and continue
-                        }
-                    },
-                    _etwTaskCancellationSource.Token);
+                        RaiseFunctionMetricEvents();
+                        _activityFlushCounter = 0;
+                    }
+                    else
+                    {
+                        _activityFlushCounter += 1;
+                    }
+                }
+                catch (Exception ex)
+                {
+                    // log error and continue
+                    _logger.LogError(ex, "Error occurred when logging function activity");
+                }
             }
 
             protected virtual void Dispose(bool disposing)
@@ -412,7 +411,7 @@ protected virtual void Dispose(bool disposing)
                 {
                     if (disposing)
                     {
-                        _etwTaskCancellationSource.Dispose();
+                        _activityTimer?.Dispose();
                     }
                     _disposed = true;
                 }
@@ -450,9 +449,11 @@ internal void FunctionCompleted(FunctionStartedEvent startedEvent)
                 RaiseFunctionMetricEvent(startedEvent, _activeFunctionCount, DateTime.UtcNow);
             }
 
-            internal void StopEtwTaskAndRaiseFinishedEvent()
+            internal void StopTimerAndRaiseFinishedEvent()
             {
-                _etwTaskCancellationSource.Cancel();
+                // stop the timer if it has been started
+                _activityTimer?.Change(Timeout.Infinite, Timeout.Infinite);
+
                 RaiseMetricsPerFunctionEvent();
             }
 
@@ -474,20 +475,23 @@ private void RaiseFunctionMetricEvents()
 
                 // We only need to raise events here for functions that aren't completed.
                 // Events are raised immediately for completed functions elsewhere.
-                var runningFunctions = new List<FunctionStartedEvent>();
+                FunctionStartedEvent[] runningFunctionsSnapshot = null;
                 lock (_runningFunctionsSyncLock)
                 {
                     // effectively we're pruning all the completed invocations here
-                    runningFunctions = _runningFunctions = _runningFunctions.Where(p => !p.Completed).ToList();
+                    _runningFunctions = _runningFunctions.Where(p => !p.Completed).ToList();
+
+                    // create a snapshot within the lock so we can enumerate below
+                    runningFunctionsSnapshot = _runningFunctions.ToArray();
                 }
 
                 // we calculate concurrency here based on count, since these events are raised
                 // on a background thread, so we want the actual count for this interval, not
                 // the current count.
                 var currentTime = DateTime.UtcNow;
-                foreach (var runningFunction in runningFunctions)
+                foreach (var runningFunction in runningFunctionsSnapshot)
                 {
-                    RaiseFunctionMetricEvent(runningFunction, runningFunctions.Count, currentTime);
+                    RaiseFunctionMetricEvent(runningFunction, runningFunctionsSnapshot.Length, currentTime);
                 }
             }
 

test/WebJobs.Script.Tests.Integration/Diagnostics/MetricsEventManagerTests.cs

@@ -15,11 +15,14 @@
 using Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics;
 using Microsoft.Azure.WebJobs.Script.WebHost.Metrics;
 using Microsoft.Azure.WebJobs.Script.WebHost.Models;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Options;
+using Microsoft.WebJobs.Script.Tests;
 using Moq;
 using Moq.Protected;
 using Xunit;
+using Xunit.Sdk;
 
 namespace Microsoft.Azure.WebJobs.Script.Tests
 {
@@ -33,14 +36,25 @@ public class MetricsEventManagerTests
         private readonly List<FunctionExecutionEventArguments> _functionExecutionEventArguments;
         private readonly List<SystemMetricEvent> _events;
         private readonly Mock<ILinuxContainerActivityPublisher> _linuxFunctionExecutionActivityPublisher;
+        private readonly Mock<IMetricsPublisher> _mockEventPublisher;
+        private readonly Mock<IEventGenerator> _mockEventGenerator;
+        private readonly Mock<IOptionsMonitor<AppServiceOptions>> _mockAppServiceOptions;
+        private readonly TestLoggerProvider _testLoggerProvider;
+        private readonly LoggerFactory _loggerFactory;
+
+        private bool _throwOnExecutionEvent;
         private readonly object _syncLock = new object();
 
         public MetricsEventManagerTests()
         {
+            _loggerFactory = new LoggerFactory();
+            _testLoggerProvider = new TestLoggerProvider();
+            _loggerFactory.AddProvider(_testLoggerProvider);
+
             _functionExecutionEventArguments = new List<FunctionExecutionEventArguments>();
 
-            var mockEventGenerator = new Mock<IEventGenerator>();
-            mockEventGenerator.Setup(e => e.LogFunctionExecutionEvent(
+            _mockEventGenerator = new Mock<IEventGenerator>();
+            _mockEventGenerator.Setup(e => e.LogFunctionExecutionEvent(
                     It.IsAny<string>(),
                     It.IsAny<string>(),
                     It.IsAny<int>(),
@@ -51,14 +65,20 @@ public MetricsEventManagerTests()
                     It.IsAny<bool>()))
                 .Callback((string executionId, string siteName, int concurrency, string functionName, string invocationId, string executionStage, long executionTimeSpan, bool success) =>
                 {
+                    if (_throwOnExecutionEvent && executionStage == ExecutionStage.InProgress.ToString())
+                    {
+                        _throwOnExecutionEvent = false;
+                        throw new Exception("Kaboom!");
+                    }
+
                     lock (_syncLock)
                     {
                         _functionExecutionEventArguments.Add(new FunctionExecutionEventArguments(executionId, siteName, concurrency, functionName, invocationId, executionStage, executionTimeSpan, success));
                     }      
                 });
 
             _events = new List<SystemMetricEvent>();
-            mockEventGenerator.Setup(p => p.LogFunctionMetricEvent(
+            _mockEventGenerator.Setup(p => p.LogFunctionMetricEvent(
                     It.IsAny<string>(),
                     It.IsAny<string>(),
                     It.IsAny<string>(),
@@ -91,11 +111,12 @@ public MetricsEventManagerTests()
                     }
                 });
 
-            var mockMetricsPublisher = new Mock<IMetricsPublisher>();
-            var testAppServiceOptions = new Mock<IOptionsMonitor<AppServiceOptions>>();
-            testAppServiceOptions.Setup(a => a.CurrentValue).Returns(new AppServiceOptions { AppName = "RandomAppName", SubscriptionId = Guid.NewGuid().ToString() });
+            _mockEventPublisher = new Mock<IMetricsPublisher>();
+            _mockAppServiceOptions = new Mock<IOptionsMonitor<AppServiceOptions>>();
+            _mockAppServiceOptions.Setup(a => a.CurrentValue).Returns(new AppServiceOptions { AppName = "RandomAppName", SubscriptionId = Guid.NewGuid().ToString() });
             _linuxFunctionExecutionActivityPublisher = new Mock<ILinuxContainerActivityPublisher>();
-            _metricsEventManager = new MetricsEventManager(testAppServiceOptions.Object, mockEventGenerator.Object, MinimumLongRunningDurationInMs / 1000, mockMetricsPublisher.Object, _linuxFunctionExecutionActivityPublisher.Object, NullLogger<MetricsEventManager>.Instance);
+            var logger = _loggerFactory.CreateLogger<MetricsEventManager>();
+            _metricsEventManager = new MetricsEventManager(_mockAppServiceOptions.Object, _mockEventGenerator.Object, MinimumLongRunningDurationInMs / 1000, _mockEventPublisher.Object, _linuxFunctionExecutionActivityPublisher.Object, logger);
             _metricsLogger = new WebHostMetricsLogger(_metricsEventManager);
         }
 
@@ -554,6 +575,69 @@ public async Task MetricsEventManager_MultipleConcurrentLongFunctionExecutions()
                     SerializeFunctionExecutionEventArguments(_functionExecutionEventArguments)));
         }
 
+        [Fact]
+        public async Task MetricsEventManager_ActivityTimer_HandlesExceptions()
+        {
+            // create a local event manager for this test, so we can override the flush interval
+            var logger = _loggerFactory.CreateLogger<MetricsEventManager>();
+            var metricsEventManager = new MetricsEventManager(_mockAppServiceOptions.Object, _mockEventGenerator.Object, 1, _mockEventPublisher.Object, _linuxFunctionExecutionActivityPublisher.Object, logger);
+            var metricsLogger = new WebHostMetricsLogger(metricsEventManager);
+
+            // execute some functions
+            var taskList = new List<Task>();
+            for (int currentIndex = 0; currentIndex < 10; currentIndex++)
+            {
+                taskList.Add(ShortTestFunction(metricsLogger));
+            }
+            await Task.WhenAll(taskList);
+
+            // wait for a flush to occur
+            await Task.Delay(1000);
+
+            // verify events
+            Assert.Equal(10, _functionExecutionEventArguments.Count);
+            Assert.True(_functionExecutionEventArguments.All(p => p.ExecutionStage == ExecutionStage.Finished.ToString()));
+
+            // now force a logging error for an in progress function
+            // on the background timer
+            _throwOnExecutionEvent = true;
+            var id = Guid.NewGuid();
+            var functionMetadata = new FunctionMetadata
+            {
+                Name = "Test"
+            };
+            var functionEvent = new FunctionStartedEvent(id, functionMetadata);
+            metricsLogger.BeginEvent(functionEvent);
+
+            // wait for the error to be logged
+            LogMessage errorLog = null;
+            await TestHelpers.Await(() =>
+            {
+                errorLog = _testLoggerProvider.GetAllLogMessages().SingleOrDefault();
+                return errorLog != null;
+            }, timeout: 5000);
+
+            // verify error was logged
+            Assert.Equal(LogLevel.Error, errorLog.Level);
+            Assert.Equal("Error occurred when logging function activity", errorLog.FormattedMessage);
+
+            // execute some more functions, verifying that the timer is
+            // still running
+            taskList = new List<Task>();
+            for (int currentIndex = 0; currentIndex < 10; currentIndex++)
+            {
+                taskList.Add(ShortTestFunction(metricsLogger));
+            }
+            await Task.WhenAll(taskList);
+
+            await Task.Delay(1000);
+
+            // verify events
+            Assert.Equal(20, _functionExecutionEventArguments.Count(p => p.ExecutionStage == ExecutionStage.Finished.ToString()));
+            int inProgressCount = _functionExecutionEventArguments.Count(p => p.InvocationId == id.ToString() && p.ExecutionStage == ExecutionStage.InProgress.ToString());
+            Assert.True(inProgressCount > 0);
+        }
+
         [Fact]
         public async Task MetricsEventManager_MultipleConcurrentFunctions()
         {