[Porting Fix] Do not use cached worker option when creating worker channel (#5973)

pragnagopa · web-flow · commit 89ca803de548 · 2020-05-05T11:05:12.000-07:00
diff --git a/src/WebJobs.Script/Host/ScriptHost.cs b/src/WebJobs.Script/Host/ScriptHost.cs
@@ -52,7 +52,6 @@ public class ScriptHost : JobHost, IScriptJobHost
         private readonly IFileLoggingStatusManager _fileLoggingStatusManager;
         private readonly IHostIdProvider _hostIdProvider;
         private readonly IHttpRoutesManager _httpRoutesManager;
-        private readonly IEnumerable<RpcWorkerConfig> _workerConfigs;
         private readonly IMetricsLogger _metricsLogger = null;
         private readonly string _hostLogPath;
         private readonly Stopwatch _stopwatch = new Stopwatch();
@@ -83,7 +82,6 @@ public class ScriptHost : JobHost, IScriptJobHost
         // Map from BindingType to the Assembly Qualified Type name for its IExtensionConfigProvider object.
 
         public ScriptHost(IOptions<JobHostOptions> options,
-            IOptions<LanguageWorkerOptions> languageWorkerOptions,
             IOptions<HttpWorkerOptions> httpWorkerOptions,
             IEnvironment environment,
             IJobHostContextFactory jobHostContextFactory,
@@ -123,7 +121,6 @@ public ScriptHost(IOptions<JobHostOptions> options,
             _applicationLifetime = applicationLifetime;
             _hostIdProvider = hostIdProvider;
             _httpRoutesManager = httpRoutesManager;
-            _workerConfigs = languageWorkerOptions.Value.WorkerConfigs;
             _isHttpWorker = httpWorkerOptions.Value.Description != null;
             ScriptOptions = scriptHostOptions.Value;
             _scriptHostManager = scriptHostManager;
diff --git a/src/WebJobs.Script/ScriptHostBuilderExtensions.cs b/src/WebJobs.Script/ScriptHostBuilderExtensions.cs
@@ -161,7 +161,7 @@ public static IHostBuilder AddScriptHostCore(this IHostBuilder builder, ScriptAp
                 services.AddSingleton<IOptionsMonitor<ScriptApplicationHostOptions>>(new ScriptApplicationHostOptionsMonitor(applicationHostOptions));
                 services.ConfigureOptions<ScriptHostOptionsSetup>();
                 services.ConfigureOptions<JobHostFunctionTimeoutOptionsSetup>();
-                // TODO: pgopa only add this to WebHostServiceCollection
+                // LanguageWorkerOptionsSetup should be registered in WebHostServiceCollection as well to enable starting worker processing in placeholder mode.
                 services.ConfigureOptions<LanguageWorkerOptionsSetup>();
                 services.ConfigureOptions<HttpWorkerOptionsSetup>();
                 services.ConfigureOptions<ManagedDependencyOptionsSetup>();
diff --git a/src/WebJobs.Script/Workers/FunctionInvocationDispatcherFactory.cs b/src/WebJobs.Script/Workers/FunctionInvocationDispatcherFactory.cs
@@ -25,7 +25,7 @@ public FunctionInvocationDispatcherFactory(IOptions<ScriptJobHostOptions> script
             IHttpWorkerChannelFactory httpWorkerChannelFactory,
             IRpcWorkerChannelFactory rpcWorkerChannelFactory,
             IOptions<HttpWorkerOptions> httpWorkerOptions,
-            IOptions<LanguageWorkerOptions> rpcWorkerOptions,
+            IOptionsMonitor<LanguageWorkerOptions> rpcWorkerOptions,
             IEnvironment environment,
             IWebHostRpcWorkerChannelManager webHostLanguageWorkerChannelManager,
             IJobHostRpcWorkerChannelManager jobHostLanguageWorkerChannelManager,
diff --git a/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs b/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs
@@ -58,7 +58,7 @@ public RpcFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHost
             IScriptEventManager eventManager,
             ILoggerFactory loggerFactory,
             IRpcWorkerChannelFactory rpcWorkerChannelFactory,
-            IOptions<LanguageWorkerOptions> languageWorkerOptions,
+            IOptionsMonitor<LanguageWorkerOptions> languageWorkerOptions,
             IWebHostRpcWorkerChannelManager webHostLanguageWorkerChannelManager,
             IJobHostRpcWorkerChannelManager jobHostLanguageWorkerChannelManager,
             IOptions<ManagedDependencyOptions> managedDependencyOptions,
@@ -71,7 +71,7 @@ public RpcFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHost
             _webHostLanguageWorkerChannelManager = webHostLanguageWorkerChannelManager;
             _jobHostLanguageWorkerChannelManager = jobHostLanguageWorkerChannelManager;
             _eventManager = eventManager;
-            _workerConfigs = languageWorkerOptions.Value.WorkerConfigs;
+            _workerConfigs = languageWorkerOptions.CurrentValue.WorkerConfigs;
             _managedDependencyOptions = managedDependencyOptions ?? throw new ArgumentNullException(nameof(managedDependencyOptions));
             _logger = loggerFactory.CreateLogger<RpcFunctionInvocationDispatcher>();
             _rpcWorkerChannelFactory = rpcWorkerChannelFactory;
@@ -113,7 +113,7 @@ internal async void InitializeJobhostLanguageWorkerChannelAsync()
 
         internal Task InitializeJobhostLanguageWorkerChannelAsync(int attemptCount)
         {
-            var rpcWorkerChannel = _rpcWorkerChannelFactory.Create(_scriptOptions.RootScriptPath, _workerRuntime, _metricsLogger, attemptCount);
+            var rpcWorkerChannel = _rpcWorkerChannelFactory.Create(_scriptOptions.RootScriptPath, _workerRuntime, _metricsLogger, attemptCount, _workerConfigs);
             rpcWorkerChannel.SetupFunctionInvocationBuffers(_functions);
             _jobHostLanguageWorkerChannelManager.AddChannel(rpcWorkerChannel);
             rpcWorkerChannel.StartWorkerProcessAsync().ContinueWith(workerInitTask =>
diff --git a/src/WebJobs.Script/Workers/Rpc/IRpcWorkerChannelFactory.cs b/src/WebJobs.Script/Workers/Rpc/IRpcWorkerChannelFactory.cs
@@ -1,12 +1,13 @@
 ﻿// Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
+using System.Collections.Generic;
 using Microsoft.Azure.WebJobs.Script.Diagnostics;
 
 namespace Microsoft.Azure.WebJobs.Script.Workers.Rpc
 {
     public interface IRpcWorkerChannelFactory
     {
-        IRpcWorkerChannel Create(string scriptRootPath, string language, IMetricsLogger metricsLogger, int attemptCount);
+        IRpcWorkerChannel Create(string scriptRootPath, string language, IMetricsLogger metricsLogger, int attemptCount, IEnumerable<RpcWorkerConfig> workerConfigs);
     }
 }
diff --git a/src/WebJobs.Script/Workers/Rpc/RpcWorkerChannel.cs b/src/WebJobs.Script/Workers/Rpc/RpcWorkerChannel.cs
@@ -109,6 +109,8 @@ internal RpcWorkerChannel(
 
         internal IWorkerProcess WorkerProcess => _rpcWorkerProcess;
 
+        internal RpcWorkerConfig Config => _workerConfig;
+
         public bool IsChannelReadyForInvocations()
         {
             return !_disposing && !_disposed && _state.HasFlag(RpcWorkerChannelState.InvocationBuffersInitialized | RpcWorkerChannelState.Initialized);
diff --git a/src/WebJobs.Script/Workers/Rpc/RpcWorkerChannelFactory.cs b/src/WebJobs.Script/Workers/Rpc/RpcWorkerChannelFactory.cs
@@ -8,7 +8,6 @@
 using Microsoft.Azure.WebJobs.Script.Abstractions;
 using Microsoft.Azure.WebJobs.Script.Diagnostics;
 using Microsoft.Azure.WebJobs.Script.Eventing;
-using Microsoft.Azure.WebJobs.Script.ManagedDependencies;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
@@ -19,22 +18,20 @@ public class RpcWorkerChannelFactory : IRpcWorkerChannelFactory
         private readonly ILoggerFactory _loggerFactory = null;
         private readonly IRpcWorkerProcessFactory _rpcWorkerProcessFactory = null;
         private readonly IScriptEventManager _eventManager = null;
-        private readonly IEnumerable<RpcWorkerConfig> _workerConfigs = null;
         private readonly IOptionsMonitor<ScriptApplicationHostOptions> _applicationHostOptions = null;
 
-        public RpcWorkerChannelFactory(IScriptEventManager eventManager, IEnvironment environment, IRpcServer rpcServer, ILoggerFactory loggerFactory, IOptions<LanguageWorkerOptions> languageWorkerOptions,
+        public RpcWorkerChannelFactory(IScriptEventManager eventManager, IEnvironment environment, IRpcServer rpcServer, ILoggerFactory loggerFactory, IOptionsMonitor<LanguageWorkerOptions> languageWorkerOptions,
             IOptionsMonitor<ScriptApplicationHostOptions> applicationHostOptions, IRpcWorkerProcessFactory rpcWorkerProcessManager)
         {
             _eventManager = eventManager;
             _loggerFactory = loggerFactory;
-            _workerConfigs = languageWorkerOptions.Value.WorkerConfigs;
             _rpcWorkerProcessFactory = rpcWorkerProcessManager;
             _applicationHostOptions = applicationHostOptions;
         }
 
-        public IRpcWorkerChannel Create(string scriptRootPath, string runtime, IMetricsLogger metricsLogger, int attemptCount)
+        public IRpcWorkerChannel Create(string scriptRootPath, string runtime, IMetricsLogger metricsLogger, int attemptCount, IEnumerable<RpcWorkerConfig> workerConfigs)
         {
-            var languageWorkerConfig = _workerConfigs.Where(c => c.Description.Language.Equals(runtime, StringComparison.OrdinalIgnoreCase)).FirstOrDefault();
+            var languageWorkerConfig = workerConfigs.Where(c => c.Description.Language.Equals(runtime, StringComparison.OrdinalIgnoreCase)).FirstOrDefault();
             if (languageWorkerConfig == null)
             {
                 throw new InvalidOperationException($"WorkerCofig for runtime: {runtime} not found");
diff --git a/src/WebJobs.Script/Workers/Rpc/WebHostRpcWorkerChannelManager.cs b/src/WebJobs.Script/Workers/Rpc/WebHostRpcWorkerChannelManager.cs
@@ -19,6 +19,7 @@ public class WebHostRpcWorkerChannelManager : IWebHostRpcWorkerChannelManager
         private readonly ILogger _logger = null;
         private readonly TimeSpan workerInitTimeout = TimeSpan.FromSeconds(30);
         private readonly IOptionsMonitor<ScriptApplicationHostOptions> _applicationHostOptions = null;
+        private readonly IOptionsMonitor<LanguageWorkerOptions> _lanuageworkerOptions = null;
         private readonly IScriptEventManager _eventManager = null;
         private readonly IEnvironment _environment;
         private readonly ILoggerFactory _loggerFactory = null;
@@ -29,7 +30,7 @@ public class WebHostRpcWorkerChannelManager : IWebHostRpcWorkerChannelManager
 
         private ConcurrentDictionary<string, Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>>> _workerChannels = new ConcurrentDictionary<string, Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>>>(StringComparer.OrdinalIgnoreCase);
 
-        public WebHostRpcWorkerChannelManager(IScriptEventManager eventManager, IEnvironment environment, ILoggerFactory loggerFactory, IRpcWorkerChannelFactory rpcWorkerChannelFactory, IOptionsMonitor<ScriptApplicationHostOptions> applicationHostOptions, IMetricsLogger metricsLogger)
+        public WebHostRpcWorkerChannelManager(IScriptEventManager eventManager, IEnvironment environment, ILoggerFactory loggerFactory, IRpcWorkerChannelFactory rpcWorkerChannelFactory, IOptionsMonitor<ScriptApplicationHostOptions> applicationHostOptions, IMetricsLogger metricsLogger, IOptionsMonitor<LanguageWorkerOptions> languageWorkerOptions)
         {
             _environment = environment ?? throw new ArgumentNullException(nameof(environment));
             _eventManager = eventManager;
@@ -38,6 +39,7 @@ public WebHostRpcWorkerChannelManager(IScriptEventManager eventManager, IEnviron
             _rpcWorkerChannelFactory = rpcWorkerChannelFactory;
             _logger = loggerFactory.CreateLogger<WebHostRpcWorkerChannelManager>();
             _applicationHostOptions = applicationHostOptions;
+            _lanuageworkerOptions = languageWorkerOptions;
 
             _shutdownStandbyWorkerChannels = ScheduleShutdownStandbyChannels;
             _shutdownStandbyWorkerChannels = _shutdownStandbyWorkerChannels.Debounce(milliseconds: 5000);
@@ -56,7 +58,7 @@ internal async Task<IRpcWorkerChannel> InitializeLanguageWorkerChannel(string ru
             _logger.LogDebug("Creating language worker channel for runtime:{runtime}", runtime);
             try
             {
-                rpcWorkerChannel = _rpcWorkerChannelFactory.Create(scriptRootPath, runtime, _metricsLogger, 0);
+                rpcWorkerChannel = _rpcWorkerChannelFactory.Create(scriptRootPath, runtime, _metricsLogger, 0, _lanuageworkerOptions.CurrentValue.WorkerConfigs);
                 AddOrUpdateWorkerChannels(runtime, rpcWorkerChannel);
                 await rpcWorkerChannel.StartWorkerProcessAsync().ContinueWith(processStartTask =>
                 {
diff --git a/test/WebJobs.Script.Tests.Integration/Host/StandbyManager/StandbyManagerE2ETests_Windows.cs b/test/WebJobs.Script.Tests.Integration/Host/StandbyManager/StandbyManagerE2ETests_Windows.cs
@@ -236,4 +236,4 @@ await TestHelpers.Await(() =>
             Assert.Equal(0, result.Count());
         }
     }
-}
+}
diff --git a/test/WebJobs.Script.Tests.Integration/WebHostEndToEnd/SpecializationE2ETests.cs b/test/WebJobs.Script.Tests.Integration/WebHostEndToEnd/SpecializationE2ETests.cs
@@ -219,7 +219,7 @@ public async Task Specialization_ResetsSharedLoadContext()
         }
 
         [Fact]
-        public async Task StartAsync_SetsCorrectActiveHost()
+        public async Task StartAsync_SetsCorrectActiveHost_RefreshesLanguageWorkerOptions()
         {
             var builder = CreateStandbyHostBuilder();
 
@@ -229,27 +229,40 @@ public async Task StartAsync_SetsCorrectActiveHost()
             Task ignore = Task.Delay(3000).ContinueWith(_ => _pauseAfterStandbyHostBuild.Release());
 
             IWebHost host = builder.Build();
-            var manager = host.Services.GetService<WebJobsScriptHostService>();
+            var scriptHostService = host.Services.GetService<WebJobsScriptHostService>();
+            var channelFactory = host.Services.GetService<IRpcWorkerChannelFactory>();
+            var workerOptionsPlaceholderMode = host.Services.GetService<IOptions<LanguageWorkerOptions>>();
+            Assert.Equal(4, workerOptionsPlaceholderMode.Value.WorkerConfigs.Count);
+            var rpcChannelInPlaceholderMode = (RpcWorkerChannel)channelFactory.Create("/", "python", null, 0, workerOptionsPlaceholderMode.Value.WorkerConfigs);
+            Assert.Equal("3.6", rpcChannelInPlaceholderMode.Config.Description.DefaultRuntimeVersion);
+
 
             // TestServer will block in the constructor so pull out the StandbyManager and use it
             // directly for this test.
             var standbyManager = host.Services.GetService<IStandbyManager>();
 
-            var standbyStart = Task.Run(async () => await manager.StartAsync(CancellationToken.None));
+            var standbyStart = Task.Run(async () => await scriptHostService.StartAsync(CancellationToken.None));
 
             // Wait until we've completed the build once. The standby host is built and now waiting for
             // _pauseAfterHostBuild to release it.
             await TestHelpers.Await(() => _buildCount.CurrentCount == 1);
 
             _environment.SetEnvironmentVariable(EnvironmentSettingNames.AzureWebsiteContainerReady, "1");
-            _environment.SetEnvironmentVariable(EnvironmentSettingNames.AzureWebsitePlaceholderMode, "0");
+            _environment.SetEnvironmentVariable(EnvironmentSettingNames.AzureWebsitePlaceholderMode, "0"); 
+            _environment.SetEnvironmentVariable(RpcWorkerConstants.FunctionWorkerRuntimeSettingName, "python");
+            _environment.SetEnvironmentVariable(RpcWorkerConstants.FunctionWorkerRuntimeVersionSettingName, "3.7");
 
             var specializeTask = Task.Run(async () => await standbyManager.SpecializeHostAsync());
 
             await Task.WhenAll(standbyStart, specializeTask);
 
-            var options = manager.Services.GetService<IOptions<ScriptJobHostOptions>>();
+            var options = scriptHostService.Services.GetService<IOptions<ScriptJobHostOptions>>();
             Assert.Equal(_specializedScriptRoot, options.Value.RootScriptPath);
+
+            var workerOptionsAtJobhostLevel = scriptHostService.Services.GetService<IOptions<LanguageWorkerOptions>>();
+            Assert.Equal(1, workerOptionsAtJobhostLevel.Value.WorkerConfigs.Count);
+            var rpcChannelAfterSpecialization = (RpcWorkerChannel)channelFactory.Create("/", "python", null, 0, workerOptionsAtJobhostLevel.Value.WorkerConfigs);
+            Assert.Equal("3.7", rpcChannelAfterSpecialization.Config.Description.DefaultRuntimeVersion);
         }
 
         private IWebHostBuilder CreateStandbyHostBuilder(params string[] functions)
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs b/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs
@@ -489,15 +489,15 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(string
             var mockFunctionDispatcherLoadBalancer = new Mock<IRpcFunctionInvocationDispatcherLoadBalancer>();
 
             _javaTestChannel = new TestRpcWorkerChannel(Guid.NewGuid().ToString(), "java", eventManager, _testLogger, false);
-
+            var optionsMonitor = TestHelpers.CreateOptionsMonitor(workerConfigOptions);
             return new RpcFunctionInvocationDispatcher(scriptOptions,
                 metricsLogger.Object,
                 testEnv,
                 mockApplicationLifetime.Object,
                 eventManager,
                 loggerFactory,
                 testLanguageWorkerChannelFactory,
-                new OptionsWrapper<LanguageWorkerOptions>(workerConfigOptions),
+                optionsMonitor,
                 testWebHostLanguageWorkerChannelManager,
                 jobHostLanguageWorkerChannelManager,
                 new OptionsWrapper<ManagedDependencyOptions>(new ManagedDependencyOptions()),
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/TesRpcWorkerChannelManager.cs b/test/WebJobs.Script.Tests/Workers/Rpc/TesRpcWorkerChannelManager.cs
@@ -48,7 +48,7 @@ public Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>> GetChannels(s
         public async Task<IRpcWorkerChannel> InitializeChannelAsync(string language)
         {
             var metricsLogger = new Mock<IMetricsLogger>();
-            IRpcWorkerChannel workerChannel = _testLanguageWorkerChannelFactory.Create(_scriptRootPath, language, metricsLogger.Object, 0);
+            IRpcWorkerChannel workerChannel = _testLanguageWorkerChannelFactory.Create(_scriptRootPath, language, metricsLogger.Object, 0, TestHelpers.GetTestWorkerConfigs());
             if (_workerChannels.TryGetValue(language, out Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>> workerChannels))
             {
                 workerChannels.Add(workerChannel.Id, new TaskCompletionSource<IRpcWorkerChannel>());
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/TestRpcWorkerChannelFactory.cs b/test/WebJobs.Script.Tests/Workers/Rpc/TestRpcWorkerChannelFactory.cs
@@ -29,7 +29,7 @@ public TestRpcWorkerChannelFactory(IScriptEventManager eventManager, ILogger tes
             _throwOnProcessStartUp = throwOnProcessStartUp;
     }
 
-        public IRpcWorkerChannel Create(string scriptRootPath, string language, IMetricsLogger metricsLogger, int attemptCount)
+        public IRpcWorkerChannel Create(string scriptRootPath, string language, IMetricsLogger metricsLogger, int attemptCount, IEnumerable<RpcWorkerConfig> workerConfigs)
         {
             return new TestRpcWorkerChannel(Guid.NewGuid().ToString(), language, _eventManager, _testLogger, throwOnProcessStartUp: _throwOnProcessStartUp);
         }
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/WebHostRpcWorkerChannelManagerTests.cs b/test/WebJobs.Script.Tests/Workers/Rpc/WebHostRpcWorkerChannelManagerTests.cs

Original file line number	Diff line number	Diff line change
`@@ -1,12 +1,13 @@`
`1`	`1`	`// Copyright (c) .NET Foundation. All rights reserved.`
`2`	`2`	`// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.`
`3`	`3`
	`4`	`+using System.Collections.Generic;`
`4`	`5`	`using Microsoft.Azure.WebJobs.Script.Diagnostics;`
`5`	`6`
`6`	`7`	`namespace Microsoft.Azure.WebJobs.Script.Workers.Rpc`
`7`	`8`	`{`
`8`	`9`	`public interface IRpcWorkerChannelFactory`
`9`	`10`	`{`
`10`		`- IRpcWorkerChannel Create(string scriptRootPath, string language, IMetricsLogger metricsLogger, int attemptCount);`
	`11`	`+ IRpcWorkerChannel Create(string scriptRootPath, string language, IMetricsLogger metricsLogger, int attemptCount, IEnumerable<RpcWorkerConfig> workerConfigs);`
`11`	`12`	`}`
`12`	`13`	`}`
Original file line number	Diff line number	Diff line change
`@@ -109,6 +109,8 @@ internal RpcWorkerChannel(`
`109`	`109`
`110`	`110`	`internal IWorkerProcess WorkerProcess => _rpcWorkerProcess;`
`111`	`111`
	`112`	`+ internal RpcWorkerConfig Config => _workerConfig;`
	`113`	`+`
`112`	`114`	`public bool IsChannelReadyForInvocations()`
`113`	`115`	`{`
`114`	`116`	`return !_disposing && !_disposed && _state.HasFlag(RpcWorkerChannelState.InvocationBuffersInitialized \| RpcWorkerChannelState.Initialized);`
Original file line number	Diff line number	Diff line change
`@@ -236,4 +236,4 @@ await TestHelpers.Await(() =>`
`236`	`236`	`Assert.Equal(0, result.Count());`
`237`	`237`	`}`
`238`	`238`	`}`
`239`		`-}`
	`239`	`+}`