Handle worker process start up failures (#4877)

Author: pragnagopa

src/WebJobs.Script.WebHost/Standby/StandbyManager.cs

@@ -6,6 +6,7 @@
 using System.Reactive.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Hosting;
 using Microsoft.Azure.WebJobs.Script.Rpc;
 using Microsoft.Azure.WebJobs.Script.WebHost.Properties;
 using Microsoft.Extensions.Configuration;
@@ -32,20 +33,21 @@ public class StandbyManager : IStandbyManager, IDisposable
         private readonly HostNameProvider _hostNameProvider;
         private readonly IDisposable _changeTokenCallbackSubscription;
         private readonly TimeSpan _specializationTimerInterval;
+        private readonly IApplicationLifetime _applicationLifetime;
 
         private Timer _specializationTimer;
         private static CancellationTokenSource _standbyCancellationTokenSource = new CancellationTokenSource();
         private static IChangeToken _standbyChangeToken = new CancellationChangeToken(_standbyCancellationTokenSource.Token);
         private static SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
 
         public StandbyManager(IScriptHostManager scriptHostManager, IWebHostLanguageWorkerChannelManager languageWorkerChannelManager, IConfiguration configuration, IScriptWebHostEnvironment webHostEnvironment,
-            IEnvironment environment, IOptionsMonitor<ScriptApplicationHostOptions> options, ILogger<StandbyManager> logger, HostNameProvider hostNameProvider)
-            : this(scriptHostManager, languageWorkerChannelManager, configuration, webHostEnvironment, environment, options, logger, hostNameProvider, TimeSpan.FromMilliseconds(500))
+            IEnvironment environment, IOptionsMonitor<ScriptApplicationHostOptions> options, ILogger<StandbyManager> logger, HostNameProvider hostNameProvider, IApplicationLifetime applicationLifetime)
+            : this(scriptHostManager, languageWorkerChannelManager, configuration, webHostEnvironment, environment, options, logger, hostNameProvider, applicationLifetime, TimeSpan.FromMilliseconds(500))
         {
         }
 
         public StandbyManager(IScriptHostManager scriptHostManager, IWebHostLanguageWorkerChannelManager languageWorkerChannelManager, IConfiguration configuration, IScriptWebHostEnvironment webHostEnvironment,
-            IEnvironment environment, IOptionsMonitor<ScriptApplicationHostOptions> options, ILogger<StandbyManager> logger, HostNameProvider hostNameProvider, TimeSpan specializationTimerInterval)
+            IEnvironment environment, IOptionsMonitor<ScriptApplicationHostOptions> options, ILogger<StandbyManager> logger, HostNameProvider hostNameProvider, IApplicationLifetime applicationLifetime, TimeSpan specializationTimerInterval)
         {
             _scriptHostManager = scriptHostManager ?? throw new ArgumentNullException(nameof(scriptHostManager));
             _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -58,13 +60,23 @@ public StandbyManager(IScriptHostManager scriptHostManager, IWebHostLanguageWork
             _hostNameProvider = hostNameProvider ?? throw new ArgumentNullException(nameof(hostNameProvider));
             _changeTokenCallbackSubscription = ChangeToken.RegisterChangeCallback(_ => _logger.LogDebug($"{nameof(StandbyManager)}.{nameof(ChangeToken)} callback has fired."), null);
             _specializationTimerInterval = specializationTimerInterval;
+            _applicationLifetime = applicationLifetime;
         }
 
         public static IChangeToken ChangeToken => _standbyChangeToken;
 
         public Task SpecializeHostAsync()
         {
-            return _specializationTask.Value;
+            return _specializationTask.Value.ContinueWith(t =>
+            {
+                if (t.IsFaulted)
+                {
+                    // if we fail during specialization for whatever reason
+                    // this is fatal, so we shutdown
+                    _logger.LogError(t.Exception, $"Specialization failed. Shutting down.");
+                    _applicationLifetime.StopApplication();
+                }
+            });
         }
 
         public async Task SpecializeHostCoreAsync()
@@ -86,6 +98,7 @@ public async Task SpecializeHostCoreAsync()
             _hostNameProvider.Reset();
 
             await _languageWorkerChannelManager.SpecializeAsync();
+
             NotifyChange();
             await _scriptHostManager.RestartHostAsync();
             await _scriptHostManager.DelayUntilHostReady();
@@ -165,15 +178,14 @@ private async Task CreateStandbyWarmupFunctions()
             _logger.LogInformation($"StandbyMode placeholder function directory created");
         }
 
-        private void OnSpecializationTimerTick(object state)
+        private async void OnSpecializationTimerTick(object state)
         {
             if (!_webHostEnvironment.InStandbyMode && _environment.IsContainerReady())
             {
                 _specializationTimer?.Dispose();
                 _specializationTimer = null;
 
-                SpecializeHostAsync().ContinueWith(t => _logger.LogError(t.Exception, "Error specializing host."),
-                    TaskContinuationOptions.OnlyOnFaulted);
+                await SpecializeHostAsync();
             }
         }
 

src/WebJobs.Script/Rpc/FunctionRegistration/FunctionDispatcher.cs

@@ -111,8 +111,7 @@ internal Task InitializeJobhostLanguageWorkerChannelAsync(int attemptCount)
             var languageWorkerChannel = _languageWorkerChannelFactory.CreateLanguageWorkerChannel(_scriptOptions.RootScriptPath, _workerRuntime, _metricsLogger, attemptCount, _managedDependencyOptions);
             languageWorkerChannel.SetupFunctionInvocationBuffers(_functions);
             _jobHostLanguageWorkerChannelManager.AddChannel(languageWorkerChannel);
-            languageWorkerChannel.StartWorkerProcessAsync()
-                 .ContinueWith(workerInitTask =>
+            languageWorkerChannel.StartWorkerProcessAsync().ContinueWith(workerInitTask =>
                  {
                      if (workerInitTask.IsCompleted)
                      {
@@ -194,14 +193,23 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions)
                 Dictionary<string, TaskCompletionSource<ILanguageWorkerChannel>> webhostLanguageWorkerChannels = _webHostLanguageWorkerChannelManager.GetChannels(_workerRuntime);
                 if (webhostLanguageWorkerChannels != null)
                 {
-                    foreach (string workerId in webhostLanguageWorkerChannels.Keys)
+                    foreach (string workerId in webhostLanguageWorkerChannels.Keys.ToList())
                     {
                         if (webhostLanguageWorkerChannels.TryGetValue(workerId, out TaskCompletionSource<ILanguageWorkerChannel> initializedLanguageWorkerChannelTask))
                         {
                             _logger.LogDebug("Found initialized language worker channel for runtime: {workerRuntime} workerId:{workerId}", _workerRuntime, workerId);
-                            ILanguageWorkerChannel initializedLanguageWorkerChannel = await initializedLanguageWorkerChannelTask.Task;
-                            initializedLanguageWorkerChannel.SetupFunctionInvocationBuffers(_functions);
-                            initializedLanguageWorkerChannel.SendFunctionLoadRequests();
+                            try
+                            {
+                                ILanguageWorkerChannel initializedLanguageWorkerChannel = await initializedLanguageWorkerChannelTask.Task;
+                                initializedLanguageWorkerChannel.SetupFunctionInvocationBuffers(_functions);
+                                initializedLanguageWorkerChannel.SendFunctionLoadRequests();
+                            }
+                            catch (Exception ex)
+                            {
+                                _logger.LogWarning(ex, "Removing errored webhost language worker channel for runtime: {workerRuntime} workerId:{workerId}", _workerRuntime, workerId);
+                                await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(_workerRuntime, workerId);
+                                InitializeWebhostLanguageWorkerChannel();
+                            }
                         }
                     }
                     StartWorkerProcesses(webhostLanguageWorkerChannels.Count(), InitializeWebhostLanguageWorkerChannel);

src/WebJobs.Script/Rpc/ILanguageWorkerProcess.cs

@@ -1,12 +1,14 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
+using System.Threading.Tasks;
+
 namespace Microsoft.Azure.WebJobs.Script.Rpc
 {
     public interface ILanguageWorkerProcess
     {
         int Id { get; }
 
-        void StartProcess();
+        Task StartProcessAsync();
     }
 }

src/WebJobs.Script/Rpc/LanguageWorkerChannel.cs

@@ -8,6 +8,7 @@
 using System.IO;
 using System.Linq;
 using System.Reactive.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;
 using Microsoft.AspNetCore.Http;
@@ -108,27 +109,27 @@ internal LanguageWorkerChannel(
 
         internal ILanguageWorkerProcess WorkerProcess => _languageWorkerProcess;
 
-        public Task StartWorkerProcessAsync()
+        public async Task StartWorkerProcessAsync()
         {
             _startSubscription = _inboundWorkerEvents.Where(msg => msg.MessageType == MsgType.StartStream)
                 .Timeout(TimeSpan.FromSeconds(LanguageWorkerConstants.ProcessStartTimeoutSeconds))
                 .Take(1)
-                .Subscribe(SendWorkerInitRequest, HandleWorkerChannelError);
-
-            _languageWorkerProcess.StartProcess();
+                .Subscribe(SendWorkerInitRequest, HandleWorkerStartStreamError);
 
+            _workerChannelLogger.LogDebug("Initiating Worker Process start up");
+            await _languageWorkerProcess.StartProcessAsync();
             _state = LanguageWorkerChannelState.Initializing;
-
-            return _workerInitTask.Task;
+            await _workerInitTask.Task;
         }
 
         // send capabilities to worker, wait for WorkerInitResponse
         internal void SendWorkerInitRequest(RpcEvent startEvent)
         {
+            _workerChannelLogger.LogDebug("Worker Process started. Received StartStream message");
             _inboundWorkerEvents.Where(msg => msg.MessageType == MsgType.WorkerInitResponse)
                 .Timeout(workerInitTimeout)
                 .Take(1)
-                .Subscribe(WorkerInitResponse, HandleWorkerChannelError);
+                .Subscribe(WorkerInitResponse, HandleWorkerInitError);
 
             SendStreamingMessage(new StreamingMessage
             {
@@ -155,11 +156,11 @@ internal void WorkerInitResponse(RpcEvent initEvent)
             _startLatencyMetric?.Dispose();
             _startLatencyMetric = null;
 
-            _workerChannelLogger.LogDebug("Received WorkerInitResponse");
+            _workerChannelLogger.LogDebug("Received WorkerInitResponse. Worker process initialized");
             _initMessage = initEvent.Message.WorkerInitResponse;
             if (_initMessage.Result.IsFailure(out Exception exc))
             {
-                HandleWorkerChannelError(exc);
+                HandleWorkerInitError(exc);
                 _workerInitTask.SetResult(false);
                 return;
             }
@@ -196,7 +197,7 @@ public Task SendFunctionEnvironmentReloadRequest()
                 .Add(_inboundWorkerEvents.Where(msg => msg.MessageType == MsgType.FunctionEnvironmentReloadResponse)
                 .Timeout(workerInitTimeout)
                 .Take(1)
-                .Subscribe((msg) => FunctionEnvironmentReloadResponse(msg.Message.FunctionEnvironmentReloadResponse)));
+                .Subscribe((msg) => FunctionEnvironmentReloadResponse(msg.Message.FunctionEnvironmentReloadResponse), HandleWorkerEnvReloadError));
 
             IDictionary processEnv = Environment.GetEnvironmentVariables();
 
@@ -388,8 +389,27 @@ internal void Log(RpcEvent msg)
             }
         }
 
-        internal void HandleWorkerChannelError(Exception exc)
+        internal void HandleWorkerStartStreamError(Exception exc)
+        {
+            _workerChannelLogger.LogError(exc, "Starting worker process failed");
+            PublishWorkerErrorEvent(exc);
+        }
+
+        internal void HandleWorkerEnvReloadError(Exception exc)
+        {
+            _workerChannelLogger.LogError(exc, "Reloading environment variables failed");
+            _reloadTask.SetException(exc);
+        }
+
+        internal void HandleWorkerInitError(Exception exc)
+        {
+            _workerChannelLogger.LogError(exc, "Initializing worker process failed");
+            PublishWorkerErrorEvent(exc);
+        }
+
+        private void PublishWorkerErrorEvent(Exception exc)
         {
+            _workerInitTask.SetException(exc);
             if (_disposing)
             {
                 return;

src/WebJobs.Script/Rpc/LanguageWorkerProcess.cs

@@ -5,6 +5,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs.Logging;
 using Microsoft.Azure.WebJobs.Script.Abstractions;
 using Microsoft.Azure.WebJobs.Script.Eventing;
@@ -61,7 +62,7 @@ internal LanguageWorkerProcess(string runtime,
 
         internal Queue<string> ProcessStdErrDataQueue => _processStdErrDataQueue;
 
-        public void StartProcess()
+        public Task StartProcessAsync()
         {
             try
             {
@@ -79,10 +80,12 @@ public void StartProcess()
 
                 // Register process only after it starts
                 _processRegistry?.Register(_process);
+                return Task.CompletedTask;
             }
             catch (Exception ex)
             {
-                throw new HostInitializationException($"Failed to start Language Worker Channel for language :{_runtime}", ex);
+                _workerProcessLogger.LogError(ex, "Failed to start Language Worker Channel for language :{_runtime}", _runtime);
+                return Task.FromException(ex);
             }
         }