diff --git a/release_notes.md b/release_notes.md index bdeb5c55d9..18172ac547 100644 --- a/release_notes.md +++ b/release_notes.md @@ -4,5 +4,6 @@ - My change description (#PR) --> +- Fixed worker startup errors during metadata indexing immediately exhausting the language worker restart budget instead of retrying. (#11955) - Fixed Linux language worker SIGTERM exits being reported as worker failures. (#11944) - Prevent extension system keys from being regenerated and overwritten when the startup context cache is stale, which previously could invalidate already-published extension webhook URLs (e.g. Event Grid, Durable Task). (#11936) diff --git a/src/Functions.Rpc.Server/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs b/src/Functions.Rpc.Server/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs index 048e12e964..535b080dc7 100644 --- a/src/Functions.Rpc.Server/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs +++ b/src/Functions.Rpc.Server/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs @@ -26,6 +26,7 @@ namespace Microsoft.Azure.WebJobs.Script.Workers.Rpc { internal class RpcFunctionInvocationDispatcher : IFunctionInvocationDispatcher { + private const int ErrorEventsThresholdMultiplier = 3; private static readonly int MultiLanguageDefaultProcessCount = 1; private readonly IMetricsLogger _metricsLogger; @@ -102,21 +103,20 @@ public RpcFunctionInvocationDispatcher(IOptions scriptHost _hostingConfigOptions = hostingConfigOptions; _hostMetrics = hostMetrics ?? throw new ArgumentNullException(nameof(hostMetrics)); State = FunctionInvocationDispatcherState.Default; + _maxProcessCount = new Lazy>(GetMaxProcessCount); + InitializeErrorEventsThreshold(_workerRuntime); _workerErrorSubscription = _eventManager.OfType().Subscribe(WorkerError); _workerRestartSubscription = _eventManager.OfType().Subscribe(WorkerRestart); - _shutdownStandbyWorkerChannels = ShutdownWebhostLanguageWorkerChannels; _shutdownStandbyWorkerChannels = _shutdownStandbyWorkerChannels.Debounce(milliseconds: 5000); - - _maxProcessCount = new Lazy>(GetMaxProcessCount); } internal Task MaxProcessCount => _maxProcessCount.Value; public FunctionInvocationDispatcherState State { get; private set; } - public int ErrorEventsThreshold { get; private set; } + public int ErrorEventsThreshold { get; private set; } = ErrorEventsThresholdMultiplier; public IJobHostRpcWorkerChannelManager JobHostLanguageWorkerChannelManager => _jobHostLanguageWorkerChannelManager; @@ -149,6 +149,30 @@ private async Task GetMaxProcessCount() return (await GetAllWorkerChannelsAsync()).Count(); } + private void InitializeErrorEventsThreshold(string workerRuntime) + { + if (_environment.IsMultiLanguageRuntimeEnvironment()) + { + SetErrorEventsThreshold(MultiLanguageDefaultProcessCount); + return; + } + + if (!string.IsNullOrEmpty(workerRuntime)) + { + var workerConfig = _workerConfigs + .FirstOrDefault(c => string.Equals(c.Description?.Language, workerRuntime, StringComparison.InvariantCultureIgnoreCase)); + if (workerConfig?.CountOptions is not null) + { + SetErrorEventsThreshold(workerConfig.CountOptions.ProcessCount); + } + } + } + + private void SetErrorEventsThreshold(int maxProcessCount) + { + ErrorEventsThreshold = ErrorEventsThresholdMultiplier * Math.Max(MultiLanguageDefaultProcessCount, maxProcessCount); + } + internal async Task InitializeJobhostLanguageWorkerChannelAsync(IEnumerable languages = null) { if (languages == null) @@ -174,8 +198,11 @@ internal async Task InitializeJobhostLanguageWorkerChannelAsync(int attemptCount _logger.LogDebug("Adding jobhost language worker channel for runtime: {language}. workerId:{id}", language, rpcWorkerChannel.Id); // if the worker is indexing, we will not have function metadata yet. So, we cannot set up invocation buffers or send load requests - rpcWorkerChannel.SetupFunctionInvocationBuffers(_functions); - rpcWorkerChannel.SendFunctionLoadRequests(_managedDependencyOptions.Value, _scriptOptions.FunctionTimeout); + if (_functions is not null) + { + rpcWorkerChannel.SetupFunctionInvocationBuffers(_functions); + rpcWorkerChannel.SendFunctionLoadRequests(_managedDependencyOptions.Value, _scriptOptions.FunctionTimeout); + } } SetFunctionDispatcherStateToInitializedAndLog(); } @@ -325,7 +352,9 @@ public async Task InitializeAsync(IEnumerable functions, Cance _restartWait = workerConfig?.CountOptions.ProcessRestartInterval ?? _defaultProcessRestartInterval; _shutdownTimeout = workerConfig?.CountOptions.ProcessShutdownTimeout ?? _defaultProcessShutdownInterval; } - ErrorEventsThreshold = 3 * await _maxProcessCount.Value; + + // This was initialized to a default, but it's possible that the worker name was not known at that time. Re-initialize now just in case it's changed. + SetErrorEventsThreshold(await _maxProcessCount.Value); if (Utility.IsSupportedRuntime(_workerRuntime, _workerConfigs) || _environment.IsMultiLanguageRuntimeEnvironment()) { diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs b/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs index 3cb4547998..564483c209 100644 --- a/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs +++ b/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs @@ -44,6 +44,53 @@ public RpcFunctionInvocationDispatcherTests() EnvironmentExtensions.ClearCache(); } + private enum WorkerRecoveryAction + { + RestartWorker, + StopApplication + } + + [Fact] + public void FunctionDispatcher_ErrorEventsThreshold_DefaultsToThree() + { + RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(runtime: null); + + Assert.Equal(3, functionDispatcher.ErrorEventsThreshold); + } + + [Fact] + public void FunctionDispatcher_ErrorEventsThreshold_DefaultsToThree_WhenWorkerConfigIsIncomplete() + { + IList workerConfigs = + [ + new(), + new() + { + Description = new RpcWorkerDescription() + }, + new() + { + Description = TestHelpers.GetTestWorkerDescription(RpcWorkerConstants.NodeLanguageWorkerName, ".js"), + CountOptions = null + } + ]; + + RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher( + runtime: RpcWorkerConstants.NodeLanguageWorkerName, workerConfigs: workerConfigs); + + Assert.Equal(3, functionDispatcher.ErrorEventsThreshold); + } + + [Fact] + public async Task FunctionDispatcher_ErrorEventsThreshold_RemainsThree_WhenNoWorkerConfigMatches() + { + RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(runtime: RpcWorkerConstants.PythonLanguageWorkerName); + + await functionDispatcher.InitializeAsync(GetTestFunctionsList(RpcWorkerConstants.PythonLanguageWorkerName)); + + Assert.Equal(3, functionDispatcher.ErrorEventsThreshold); + } + [Fact] public async Task GetWorkerStatusesAsync_ReturnsExpectedResult() { @@ -109,6 +156,80 @@ public async Task WorkerIndexing_Starting_WebhostChannel_Succeeds() Assert.Equal(0, finalJobhostChannelCount); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task FunctionDispatcher_WorkerIndexing_FirstWorkerError_RestartsWorker(bool initializeWithEmptyMetadata) + { + const int processCount = 2; + const string workerId = "metadata-worker"; + IList workerConfigs = + [ + new() + { + Description = TestHelpers.GetTestWorkerDescription( + RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, ".dll", workerIndexing: true), + CountOptions = new WorkerProcessCountOptions + { + ProcessCount = processCount + } + } + ]; + var recoveryActionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var eventManager = new ScriptEventManager(); + var applicationLifetime = new Mock(); + applicationLifetime.SetupGet(m => m.ApplicationStopping).Returns(CancellationToken.None); + applicationLifetime.Setup(m => m.StopApplication()) + .Callback(() => recoveryActionSource.TrySetResult(WorkerRecoveryAction.StopApplication)); + + var webHostChannelManager = new Mock(); + webHostChannelManager.Setup(m => m.GetChannels(RpcWorkerConstants.DotNetIsolatedLanguageWorkerName)) + .Returns((IDictionary>)null); + webHostChannelManager.Setup(m => m.ShutdownChannelIfExistsAsync( + RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, workerId, It.IsAny())) + .ReturnsAsync(true); + + var retryChannel = new Mock(); + retryChannel.SetupGet(m => m.Id).Returns("retry-worker"); + retryChannel.Setup(m => m.StartWorkerProcessAsync(CancellationToken.None)) + .Callback(() => recoveryActionSource.TrySetResult(WorkerRecoveryAction.RestartWorker)) + .Returns(Task.CompletedTask); + + var channelFactory = new Mock(); + channelFactory.Setup(m => m.Create( + It.IsAny(), RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, It.IsAny(), 1, + It.IsAny>())) + .Returns(retryChannel.Object); + + RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher( + maxProcessCountValue: processCount, + runtime: RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, + workerIndexing: true, + channelFactory: channelFactory.Object, + mockwebHostLanguageWorkerChannelManager: webHostChannelManager, + eventManager: eventManager, + applicationLifetime: applicationLifetime.Object, + workerConfigs: workerConfigs); + + if (initializeWithEmptyMetadata) + { + await functionDispatcher.InitializeAsync(Array.Empty()); + } + + eventManager.Publish(new WorkerErrorEvent( + RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, workerId, new TimeoutException("Worker did not send StartStream."))); + + WorkerRecoveryAction recoveryAction = await recoveryActionSource.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + Assert.Equal(WorkerRecoveryAction.RestartWorker, recoveryAction); + await TestHelpers.Await(() => functionDispatcher.State == FunctionInvocationDispatcherState.Initialized); + applicationLifetime.Verify(m => m.StopApplication(), Times.Never); + retryChannel.Verify(m => m.SetupFunctionInvocationBuffers(It.IsAny>()), Times.Never); + retryChannel.Verify(m => m.SendFunctionLoadRequests(It.IsAny(), It.IsAny()), Times.Never); + Assert.Equal(3 * processCount, functionDispatcher.ErrorEventsThreshold); + Assert.Single(functionDispatcher.LanguageWorkerErrors); + } + [Fact] public async Task Starting_MultipleJobhostChannels_Failed() { @@ -765,13 +886,20 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher( bool workerIndexing = false, bool placeholder = false, IRpcWorkerChannelFactory channelFactory = null, - CancellationTokenSource applicationStoppingSource = null) + CancellationTokenSource applicationStoppingSource = null, + IScriptEventManager eventManager = null, + IHostApplicationLifetime applicationLifetime = null, + IList workerConfigs = null) { - var eventManager = new ScriptEventManager(); + eventManager ??= new ScriptEventManager(); var metricsLogger = new Mock(); - var mockApplicationLifetime = new Mock(); var stoppingSource = applicationStoppingSource ?? new CancellationTokenSource(); - mockApplicationLifetime.Setup(m => m.ApplicationStopping).Returns(stoppingSource.Token); + if (applicationLifetime is null) + { + var mockApplicationLifetime = new Mock(); + mockApplicationLifetime.Setup(m => m.ApplicationStopping).Returns(stoppingSource.Token); + applicationLifetime = mockApplicationLifetime.Object; + } var testEnv = new TestEnvironment(); TimeSpan intervals = startupIntervals ?? TimeSpan.FromMilliseconds(100); @@ -795,8 +923,8 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher( var workerConfigOptions = new LanguageWorkerOptions { - WorkerConfigs = TestHelpers.GetTestWorkerConfigs(processCountValue: maxProcessCountValue, processStartupInterval: intervals, - processRestartInterval: intervals, processShutdownTimeout: TimeSpan.FromSeconds(1), workerIndexing: workerIndexing) + WorkerConfigs = workerConfigs ?? TestHelpers.GetTestWorkerConfigs(processCountValue: maxProcessCountValue, processStartupInterval: intervals, + processRestartInterval: intervals, processShutdownTimeout: TimeSpan.FromSeconds(1), workerIndexing: workerIndexing) }; channelFactory ??= new TestRpcWorkerChannelFactory(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, throwOnProcessStartUp); @@ -828,7 +956,7 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher( return new RpcFunctionInvocationDispatcher(scriptOptions, metricsLogger.Object, testEnv, - mockApplicationLifetime.Object, + applicationLifetime, eventManager, _testLoggerFactory, channelFactory,