[RPC]Lock process restarts on WorkerError (#6930)

pragnagopa · web-flow · commit 713942cbc75d · 2021-01-21T14:02:14.000-08:00
[RPC]Lock process restarts on WorkerError
diff --git a/src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs b/src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs
@@ -524,7 +524,7 @@ internal void HandleWorkerInitError(Exception exc)
         internal void HandleWorkerFunctionLoadError(Exception exc)
         {
             _workerChannelLogger.LogError(exc, "Loading function failed.");
-            if (_disposing)
+            if (_disposing || _disposed)
             {
                 return;
             }
@@ -534,7 +534,7 @@ internal void HandleWorkerFunctionLoadError(Exception exc)
         private void PublishWorkerErrorEvent(Exception exc)
         {
             _workerInitTask.SetException(exc);
-            if (_disposing)
+            if (_disposing || _disposed)
             {
                 return;
             }
diff --git a/src/WebJobs.Script/Workers/ProcessManagement/WorkerProcess.cs b/src/WebJobs.Script/Workers/ProcessManagement/WorkerProcess.cs
@@ -26,7 +26,6 @@ internal abstract class WorkerProcess : IWorkerProcess, IDisposable
         private Process _process;
         private bool _useStdErrorStreamForErrorsOnly;
         private ProcessMonitor _processMonitor;
-        private bool _disposing;
         private Queue<string> _processStdErrDataQueue = new Queue<string>(3);
 
         internal WorkerProcess(IScriptEventManager eventManager, IProcessRegistry processRegistry, ILogger workerProcessLogger, IWorkerConsoleLogSource consoleLogSource, IMetricsLogger metricsLogger, bool useStdErrStreamForErrorsOnly = false)
@@ -39,6 +38,8 @@ internal WorkerProcess(IScriptEventManager eventManager, IProcessRegistry proces
             _useStdErrorStreamForErrorsOnly = useStdErrStreamForErrorsOnly;
         }
 
+        protected bool Disposing { get; private set; }
+
         public int Id => _process.Id;
 
         internal Queue<string> ProcessStdErrDataQueue => _processStdErrDataQueue;
@@ -129,7 +130,7 @@ private void LogError(string msg)
 
         private void OnProcessExited(object sender, EventArgs e)
         {
-            if (_disposing)
+            if (Disposing)
             {
                 // No action needed
                 return;
@@ -192,7 +193,7 @@ internal void BuildAndLogConsoleLog(string msg, LogLevel level)
 
         public void Dispose()
         {
-            _disposing = true;
+            Disposing = true;
             // best effort process disposal
             try
             {
diff --git a/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs b/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs
@@ -29,7 +29,9 @@ internal class RpcFunctionInvocationDispatcher : IFunctionInvocationDispatcher
         private readonly IEnvironment _environment;
         private readonly IApplicationLifetime _applicationLifetime;
         private readonly TimeSpan _shutdownTimeout = TimeSpan.FromSeconds(10);
-        private readonly TimeSpan thresholdBetweenRestarts = TimeSpan.FromMinutes(WorkerConstants.WorkerRestartErrorIntervalThresholdInMinutes);
+        private readonly TimeSpan _restartWait = TimeSpan.FromSeconds(10);
+        private readonly SemaphoreSlim _restartWorkerProcessSLock = new SemaphoreSlim(1, 1);
+        private readonly TimeSpan _thresholdBetweenRestarts = TimeSpan.FromMinutes(WorkerConstants.WorkerRestartErrorIntervalThresholdInMinutes);
 
         private IScriptEventManager _eventManager;
         private IEnumerable<RpcWorkerConfig> _workerConfigs;
@@ -48,6 +50,7 @@ internal class RpcFunctionInvocationDispatcher : IFunctionInvocationDispatcher
         private IEnumerable<FunctionMetadata> _functions;
         private ConcurrentStack<WorkerErrorEvent> _languageWorkerErrors = new ConcurrentStack<WorkerErrorEvent>();
         private CancellationTokenSource _processStartCancellationToken = new CancellationTokenSource();
+        private CancellationTokenSource _disposeToken = new CancellationTokenSource();
         private int _debounceMilliSeconds = (int)TimeSpan.FromSeconds(10).TotalMilliseconds;
 
         public RpcFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHostOptions,
@@ -310,35 +313,43 @@ internal async Task<IEnumerable<IRpcWorkerChannel>> GetInitializedWorkerChannels
 
         public async void WorkerError(WorkerErrorEvent workerError)
         {
-            if (!_disposing)
+            if (_disposing || _disposed)
             {
-                if (string.Equals(_workerRuntime, workerError.Language))
-                {
-                    _logger.LogDebug("Handling WorkerErrorEvent for runtime:{runtime}, workerId:{workerId}. Failed with: {exception}", workerError.Language, _workerRuntime, workerError.Exception);
-                    AddOrUpdateErrorBucket(workerError);
-                    await DisposeAndRestartWorkerChannel(workerError.Language, workerError.WorkerId, workerError.Exception);
-                }
-                else
-                {
-                    _logger.LogDebug("Received WorkerErrorEvent for runtime:{runtime}, workerId:{workerId}", workerError.Language, workerError.WorkerId);
-                    _logger.LogDebug("WorkerErrorEvent runtime:{runtime} does not match current runtime:{currentRuntime}. Failed with: {exception}", workerError.Language, _workerRuntime, workerError.Exception);
-                }
+                return;
+            }
+
+            if (string.Equals(_workerRuntime, workerError.Language))
+            {
+                _logger.LogDebug("Handling WorkerErrorEvent for runtime:{runtime}, workerId:{workerId}. Failed with: {exception}", workerError.Language, _workerRuntime, workerError.Exception);
+                AddOrUpdateErrorBucket(workerError);
+                await DisposeAndRestartWorkerChannel(workerError.Language, workerError.WorkerId, workerError.Exception);
+            }
+            else
+            {
+                _logger.LogDebug("Received WorkerErrorEvent for runtime:{runtime}, workerId:{workerId}", workerError.Language, workerError.WorkerId);
+                _logger.LogDebug("WorkerErrorEvent runtime:{runtime} does not match current runtime:{currentRuntime}. Failed with: {exception}", workerError.Language, _workerRuntime, workerError.Exception);
             }
         }
 
         public async void WorkerRestart(WorkerRestartEvent workerRestart)
         {
-            if (!_disposing)
+            if (_disposing || _disposed)
             {
-                _logger.LogDebug("Handling WorkerRestartEvent for runtime:{runtime}, workerId:{workerId}", workerRestart.Language, workerRestart.WorkerId);
-                await DisposeAndRestartWorkerChannel(workerRestart.Language, workerRestart.WorkerId);
+                return;
             }
+
+            _logger.LogDebug("Handling WorkerRestartEvent for runtime:{runtime}, workerId:{workerId}", workerRestart.Language, workerRestart.WorkerId);
+            await DisposeAndRestartWorkerChannel(workerRestart.Language, workerRestart.WorkerId);
         }
 
         private async Task DisposeAndRestartWorkerChannel(string runtime, string workerId, Exception workerException = null)
         {
-            _logger.LogDebug("Attempting to dispose webhost or jobhost channel for workerId: {channelId}, runtime:{language}", workerId, runtime);
+            if (_disposing || _disposed)
+            {
+                return;
+            }
 
+            _logger.LogDebug("Attempting to dispose webhost or jobhost channel for workerId: '{channelId}', runtime: '{language}'", workerId, runtime);
             bool isWebHostChannelDisposed = await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(runtime, workerId, workerException);
             bool isJobHostChannelDisposed = false;
             if (!isWebHostChannelDisposed)
@@ -348,7 +359,7 @@ private async Task DisposeAndRestartWorkerChannel(string runtime, string workerI
 
             if (!isWebHostChannelDisposed && !isJobHostChannelDisposed)
             {
-                _logger.LogDebug("Did not find WebHost or JobHost channel to dispose for workerId: {channelId}, runtime:{language}", workerId, runtime);
+                _logger.LogDebug("Did not find WebHost or JobHost channel to dispose for workerId: '{channelId}', runtime: '{language}'", workerId, runtime);
             }
 
             if (ShouldRestartWorkerChannel(runtime, isWebHostChannelDisposed, isJobHostChannelDisposed))
@@ -360,13 +371,13 @@ private async Task DisposeAndRestartWorkerChannel(string runtime, string workerI
                     _logger.LogDebug("No initialized worker channels for runtime '{runtime}'. Delaying future invocations", runtime);
                 }
                 // Restart worker channel
-                _logger.LogDebug("Restarting worker channel for runtime:{runtime}", runtime);
-                await RestartWorkerChannel(runtime, workerId);
+                _logger.LogDebug("Restarting worker channel for runtime: '{runtime}'", runtime);
+                await RestartWorkerChannel(runtime);
                 // State is set back to "Initialized" when worker channel is up again
             }
             else
             {
-                _logger.LogDebug("Skipping worker channel restart for errored worker runtime:{runtime}, current runtime:{currentRuntime}, isWebHostChannel:{isWebHostChannel}, isJobHostChannel:{isJobHostChannel}",
+                _logger.LogDebug("Skipping worker channel restart for errored worker runtime: '{runtime}', current runtime: '{currentRuntime}', isWebHostChannel: '{isWebHostChannel}', isJobHostChannel: '{isJobHostChannel}'",
                     runtime, _workerRuntime, isWebHostChannelDisposed, isJobHostChannelDisposed);
             }
         }
@@ -376,11 +387,33 @@ internal bool ShouldRestartWorkerChannel(string runtime, bool isWebHostChannel,
             return string.Equals(_workerRuntime, runtime, StringComparison.InvariantCultureIgnoreCase) && (isWebHostChannel || isJobHostChannel);
         }
 
-        private async Task RestartWorkerChannel(string runtime, string workerId)
+        private async Task RestartWorkerChannel(string runtime)
         {
+            if (_disposing || _disposed)
+            {
+                return;
+            }
+
             if (_languageWorkerErrors.Count < ErrorEventsThreshold)
             {
-                await InitializeJobhostLanguageWorkerChannelAsync(_languageWorkerErrors.Count);
+                try
+                {
+                    // Issue only one restart at a time.
+                    await _restartWorkerProcessSLock.WaitAsync();
+                    await InitializeJobhostLanguageWorkerChannelAsync(_languageWorkerErrors.Count);
+                }
+                finally
+                {
+                    // Wait before releasing the lock to give time for the process to startup and initialize.
+                    try
+                    {
+                        await Task.Delay(_restartWait, _disposeToken.Token);
+                        _restartWorkerProcessSLock.Release();
+                    }
+                    catch (TaskCanceledException)
+                    {
+                    }
+                }
             }
             else if (_jobHostLanguageWorkerChannelManager.GetChannels().Count() == 0)
             {
@@ -393,12 +426,12 @@ private void AddOrUpdateErrorBucket(WorkerErrorEvent currentErrorEvent)
         {
             if (_languageWorkerErrors.TryPeek(out WorkerErrorEvent top))
             {
-                if ((currentErrorEvent.CreatedAt - top.CreatedAt) > thresholdBetweenRestarts)
+                if ((currentErrorEvent.CreatedAt - top.CreatedAt) > _thresholdBetweenRestarts)
                 {
                     while (!_languageWorkerErrors.IsEmpty)
                     {
                         _languageWorkerErrors.TryPop(out WorkerErrorEvent popped);
-                        _logger.LogDebug($"Popping out errorEvent createdAt:{popped.CreatedAt} workerId:{popped.WorkerId}");
+                        _logger.LogDebug($"Popping out errorEvent createdAt: '{popped.CreatedAt}' workerId: '{popped.WorkerId}'");
                     }
                 }
             }
@@ -407,14 +440,21 @@ private void AddOrUpdateErrorBucket(WorkerErrorEvent currentErrorEvent)
 
         protected virtual void Dispose(bool disposing)
         {
-            if (!_disposed && disposing)
+            if (!_disposed)
             {
-                _logger.LogDebug("Disposing FunctionDispatcher");
-                _workerErrorSubscription.Dispose();
-                _workerRestartSubscription.Dispose();
-                _processStartCancellationToken.Cancel();
-                _processStartCancellationToken.Dispose();
-                _jobHostLanguageWorkerChannelManager.ShutdownChannels();
+                if (_disposing)
+                {
+                    _logger.LogDebug("Disposing FunctionDispatcher");
+                    _disposeToken.Cancel();
+                    _disposeToken.Dispose();
+                    _restartWorkerProcessSLock.Dispose();
+                    _workerErrorSubscription.Dispose();
+                    _workerRestartSubscription.Dispose();
+                    _processStartCancellationToken.Cancel();
+                    _processStartCancellationToken.Dispose();
+                    _jobHostLanguageWorkerChannelManager.ShutdownChannels();
+                }
+
                 State = FunctionInvocationDispatcherState.Disposed;
                 _disposed = true;
             }
@@ -435,7 +475,7 @@ public async Task<bool> RestartWorkerWithInvocationIdAsync(string invocationId)
             {
                 if (channel.IsExecutingInvocation(invocationId))
                 {
-                    _logger.LogInformation($"Restarting channel '{channel.Id}' that is executing invocation '{invocationId}' and timed out.");
+                    _logger.LogDebug($"Restarting channel with workerId: '{channel.Id}' that is executing invocation: '{invocationId}' and timed out.");
                     await DisposeAndRestartWorkerChannel(_workerRuntime, channel.Id);
                     return true;
                 }
diff --git a/src/WebJobs.Script/Workers/Rpc/RpcWorkerProcess.cs b/src/WebJobs.Script/Workers/Rpc/RpcWorkerProcess.cs
@@ -54,6 +54,10 @@ internal override Process CreateWorkerProcess()
 
         internal override void HandleWorkerProcessExitError(WorkerProcessExitException rpcWorkerProcessExitException)
         {
+            if (Disposing)
+            {
+                return;
+            }
             if (rpcWorkerProcessExitException == null)
             {
                 throw new ArgumentNullException(nameof(rpcWorkerProcessExitException));
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs b/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs
@@ -384,6 +384,11 @@ public async Task FunctionDispatcher_Error_WithinThreshold_BucketFills()
                 {
                     TestRpcWorkerChannel testWorkerChannel = channel as TestRpcWorkerChannel;
                     testWorkerChannel.RaiseWorkerError();
+                    if (i < 2)
+                    {
+                        // wait for restart to complete before raising another error
+                        await WaitForJobhostWorkerChannelsToStartup(functionDispatcher, 1);
+                    }
                 }
             }
 

Original file line number	Diff line number	Diff line change
`@@ -524,7 +524,7 @@ internal void HandleWorkerInitError(Exception exc)`
`524`	`524`	`internal void HandleWorkerFunctionLoadError(Exception exc)`
`525`	`525`	`{`
`526`	`526`	`_workerChannelLogger.LogError(exc, "Loading function failed.");`
`527`		`- if (_disposing)`
	`527`	`+ if (_disposing \|\| _disposed)`
`528`	`528`	`{`
`529`	`529`	`return;`
`530`	`530`	`}`
`@@ -534,7 +534,7 @@ internal void HandleWorkerFunctionLoadError(Exception exc)`
`534`	`534`	`private void PublishWorkerErrorEvent(Exception exc)`
`535`	`535`	`{`
`536`	`536`	`_workerInitTask.SetException(exc);`
`537`		`- if (_disposing)`
	`537`	`+ if (_disposing \|\| _disposed)`
`538`	`538`	`{`
`539`	`539`	`return;`
`540`	`540`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,6 @@ internal abstract class WorkerProcess : IWorkerProcess, IDisposable`
`26`	`26`	`private Process _process;`
`27`	`27`	`private bool _useStdErrorStreamForErrorsOnly;`
`28`	`28`	`private ProcessMonitor _processMonitor;`
`29`		`- private bool _disposing;`
`30`	`29`	`private Queue<string> _processStdErrDataQueue = new Queue<string>(3);`
`31`	`30`
`32`	`31`	`internal WorkerProcess(IScriptEventManager eventManager, IProcessRegistry processRegistry, ILogger workerProcessLogger, IWorkerConsoleLogSource consoleLogSource, IMetricsLogger metricsLogger, bool useStdErrStreamForErrorsOnly = false)`
`@@ -39,6 +38,8 @@ internal WorkerProcess(IScriptEventManager eventManager, IProcessRegistry proces`
`39`	`38`	`_useStdErrorStreamForErrorsOnly = useStdErrStreamForErrorsOnly;`
`40`	`39`	`}`
`41`	`40`
	`41`	`+ protected bool Disposing { get; private set; }`
	`42`	`+`
`42`	`43`	`public int Id => _process.Id;`
`43`	`44`
`44`	`45`	`internal Queue<string> ProcessStdErrDataQueue => _processStdErrDataQueue;`
`@@ -129,7 +130,7 @@ private void LogError(string msg)`
`129`	`130`
`130`	`131`	`private void OnProcessExited(object sender, EventArgs e)`
`131`	`132`	`{`
`132`		`- if (_disposing)`
	`133`	`+ if (Disposing)`
`133`	`134`	`{`
`134`	`135`	`// No action needed`
`135`	`136`	`return;`
`@@ -192,7 +193,7 @@ internal void BuildAndLogConsoleLog(string msg, LogLevel level)`
`192`	`193`
`193`	`194`	`public void Dispose()`
`194`	`195`	`{`
`195`		`- _disposing = true;`
	`196`	`+ Disposing = true;`
`196`	`197`	`// best effort process disposal`
`197`	`198`	`try`
`198`	`199`	`{`
Original file line number	Diff line number	Diff line change
`@@ -54,6 +54,10 @@ internal override Process CreateWorkerProcess()`
`54`	`54`
`55`	`55`	`internal override void HandleWorkerProcessExitError(WorkerProcessExitException rpcWorkerProcessExitException)`
`56`	`56`	`{`
	`57`	`+ if (Disposing)`
	`58`	`+ {`
	`59`	`+ return;`
	`60`	`+ }`
`57`	`61`	`if (rpcWorkerProcessExitException == null)`
`58`	`62`	`{`
`59`	`63`	`throw new ArgumentNullException(nameof(rpcWorkerProcessExitException));`
Original file line number	Diff line number	Diff line change
`@@ -384,6 +384,11 @@ public async Task FunctionDispatcher_Error_WithinThreshold_BucketFills()`
`384`	`384`	`{`
`385`	`385`	`TestRpcWorkerChannel testWorkerChannel = channel as TestRpcWorkerChannel;`
`386`	`386`	`testWorkerChannel.RaiseWorkerError();`
	`387`	`+ if (i < 2)`
	`388`	`+ {`
	`389`	`+ // wait for restart to complete before raising another error`
	`390`	`+ await WaitForJobhostWorkerChannelsToStartup(functionDispatcher, 1);`
	`391`	`+ }`
`387`	`392`	`}`
`388`	`393`	`}`
`389`	`394`