[pack] Initialize Java language worker at webhost level (#3824)

Author: pragnagopa

schemas/json/host.json

@@ -126,13 +126,7 @@
             "description": "Configuration settings for Language Workers.",
             "type": "object",
 
-            "properties": {
-                "maxMessageLength": {
-                    "description": "Defines maximum message size in MegaBytes (MB) for the streaming messages between the functions host and a language worker",
-                    "type": "integer",
-                    "minimum": 4,
-                    "default": 32
-                },
+            "properties": {               
                 "workersDirectory": {
                     "description": "Specifies full path of the directory for language workers",
                     "type": "string"

src/WebJobs.Script.Grpc/Abstractions/IRpcServer.cs

@@ -21,6 +21,12 @@ public interface IRpcServer
         /// <returns>A task that completes when the server is ready</returns>
         Task StartAsync();
 
+        /// <summary>
+        /// Forces a server shutdown
+        /// </summary>
+        /// <returns>A task that completes when the server is killed</returns>
+        Task KillAsync();
+
         /// <summary>
         /// Gracefully shuts down the rpc server, allowing existing calls to finish
         /// </summary>

src/WebJobs.Script.Grpc/Server/GrpcServer.cs

@@ -14,11 +14,12 @@ public class GrpcServer : IRpcServer, IDisposable
     {
         private Server _server;
         private bool _disposed = false;
+        public const int MaxMessageLengthBytes = 128 * 1024 * 1024;
 
-        public GrpcServer(FunctionRpc.FunctionRpcBase serviceImpl, int grpcMaxMessageLength)
+        public GrpcServer(FunctionRpc.FunctionRpcBase serviceImpl)
         {
-            ChannelOption maxReceiveMessageLength = new ChannelOption(ChannelOptions.MaxReceiveMessageLength, grpcMaxMessageLength);
-            ChannelOption maxSendMessageLength = new ChannelOption(ChannelOptions.MaxSendMessageLength, grpcMaxMessageLength);
+            ChannelOption maxReceiveMessageLength = new ChannelOption(ChannelOptions.MaxReceiveMessageLength, MaxMessageLengthBytes);
+            ChannelOption maxSendMessageLength = new ChannelOption(ChannelOptions.MaxSendMessageLength, MaxMessageLengthBytes);
             ChannelOption[] grpcChannelOptions = { maxReceiveMessageLength, maxSendMessageLength };
             _server = new Server(grpcChannelOptions)
             {
@@ -37,6 +38,8 @@ public Task StartAsync()
 
         public Task ShutdownAsync() => _server.ShutdownAsync();
 
+        public Task KillAsync() => _server.KillAsync();
+
         protected virtual void Dispose(bool disposing)
         {
             if (!_disposed)

src/WebJobs.Script.Grpc/WebJobs.Script.Grpc.csproj

@@ -20,10 +20,10 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Google.Protobuf" Version="3.5.1" />
-    <PackageReference Include="Google.Protobuf.Tools" Version="3.5.1" />
-    <PackageReference Include="Grpc.Core" Version="1.12.0" />
-    <PackageReference Include="Grpc.Tools" Version="1.12.0">
+    <PackageReference Include="Google.Protobuf" Version="3.6.1" />
+    <PackageReference Include="Google.Protobuf.Tools" Version="3.6.1" />
+    <PackageReference Include="Grpc.Core" Version="1.16.0" />
+    <PackageReference Include="Grpc.Tools" Version="1.16.0">
       <PrivateAssets>All</PrivateAssets>
     </PackageReference>
     <PackageReference Include="Microsoft.Extensions.Logging" Version="2.1.1" />

src/WebJobs.Script.Grpc/azure-functions-language-worker-protobuf/src/proto/FunctionRpc.proto

@@ -5,6 +5,7 @@ option java_multiple_files = true;
 option java_package = "com.microsoft.azure.functions.rpc.messages";
 option java_outer_classname = "FunctionProto";
 option csharp_namespace = "Microsoft.Azure.WebJobs.Script.Grpc.Messages";
+option go_package ="github.com/Azure/azure-functions-go-worker/internal/rpc";
 
 package AzureFunctionsRpcMessages;
 
@@ -64,6 +65,10 @@ message StreamingMessage {
 
     // Worker logs a message back to the host
     RpcLog rpc_log = 2;
+
+    FunctionEnvironmentReloadRequest function_environment_reload_request = 25;
+
+    FunctionEnvironmentReloadResponse function_environment_reload_response = 26;
   }
 }
 
@@ -179,6 +184,16 @@ message WorkerStatusRequest{
 message WorkerStatusResponse {
 }
 
+message FunctionEnvironmentReloadRequest {
+  // Environment variables from the current process
+  map<string, string> environment_variables = 1;
+}
+
+message FunctionEnvironmentReloadResponse {
+  // Status of the response
+  StatusResult result = 3;
+}
+
 // Host tells the worker to load a Function
 message FunctionLoadRequest {
   // unique function identifier (avoid name collisions, facilitate reload case)
@@ -354,4 +369,4 @@ message RpcHttp {
   map<string,string> query = 15;
   bool enable_content_negotiation= 16;
   TypedData rawBody = 17;
-}
+}

src/WebJobs.Script.Grpc/generate_protos.bat

@@ -35,10 +35,10 @@ setlocal
 cd /d %~dp0
 
 set NUGET_PATH=%UserProfile%\.nuget\packages
-set GRPC_TOOLS_PATH=%NUGET_PATH%\grpc.tools\1.12.0\tools\windows_x86
+set GRPC_TOOLS_PATH=%NUGET_PATH%\grpc.tools\1.16.0\tools\windows_x86
 set PROTO_PATH=.\azure-functions-language-worker-protobuf\src\proto
 set PROTO=.\azure-functions-language-worker-protobuf\src\proto\FunctionRpc.proto
-set PROTOBUF_TOOLS=%NUGET_PATH%\google.protobuf.tools\3.5.1\tools
+set PROTOBUF_TOOLS=%NUGET_PATH%\google.protobuf.tools\3.6.1\tools
 set MSGDIR=.\Messages
 
 if exist %MSGDIR% rmdir /s /q %MSGDIR%

src/WebJobs.Script.Grpc/generate_protos.sh

@@ -49,9 +49,9 @@ else
 	NUGET_PATH=$NUGET_ROOT/packages
 fi
 
-GRPC_TOOLS_PATH=$NUGET_PATH/grpc.tools/1.12.0/tools/$PLATFORM
+GRPC_TOOLS_PATH=$NUGET_PATH/grpc.tools/1.16.0/tools/$PLATFORM
 PROTO_PATH=./azure-functions-language-worker-protobuf/src/proto
-PROTOBUF_TOOLS=$NUGET_PATH/google.protobuf.tools/3.5.1/tools
+PROTOBUF_TOOLS=$NUGET_PATH/google.protobuf.tools/3.6.1/tools
 PROTO=./azure-functions-language-worker-protobuf/src/proto/FunctionRpc.proto
 MSGDIR=./Messages
 

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

@@ -2,9 +2,13 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.IO;
+using System.Reactive.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Azure.WebJobs.Script.Eventing;
+using Microsoft.Azure.WebJobs.Script.Rpc;
 using Microsoft.Azure.WebJobs.Script.WebHost.Properties;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.FileProviders;
@@ -24,17 +28,17 @@ public class StandbyManager : IStandbyManager
         private readonly Lazy<Task> _specializationTask;
         private readonly IScriptWebHostEnvironment _webHostEnvironment;
         private readonly IEnvironment _environment;
+        private readonly ILanguageWorkerChannelManager _languageWorkerChannelManager;
         private readonly IConfigurationRoot _configuration;
         private readonly ILogger _logger;
-
         private readonly TimeSpan _specializationTimerInterval = TimeSpan.FromMilliseconds(500);
-        private Timer _specializationTimer;
 
+        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, IConfiguration configuration, IScriptWebHostEnvironment webHostEnvironment,
+        public StandbyManager(IScriptHostManager scriptHostManager, ILanguageWorkerChannelManager languageWorkerChannelManager, IConfiguration configuration, IScriptWebHostEnvironment webHostEnvironment,
             IEnvironment environment, IOptionsMonitor<ScriptApplicationHostOptions> options, ILogger<StandbyManager> logger)
         {
             _scriptHostManager = scriptHostManager ?? throw new ArgumentNullException(nameof(scriptHostManager));
@@ -43,8 +47,8 @@ public StandbyManager(IScriptHostManager scriptHostManager, IConfiguration confi
             _specializationTask = new Lazy<Task>(SpecializeHostCoreAsync, LazyThreadSafetyMode.ExecutionAndPublication);
             _webHostEnvironment = webHostEnvironment ?? throw new ArgumentNullException(nameof(webHostEnvironment));
             _environment = environment ?? throw new ArgumentNullException(nameof(environment));
-
             _configuration = configuration as IConfigurationRoot ?? throw new ArgumentNullException(nameof(configuration));
+            _languageWorkerChannelManager = languageWorkerChannelManager ?? throw new ArgumentNullException(nameof(languageWorkerChannelManager));
         }
 
         public static IChangeToken ChangeToken => _standbyChangeToken;
@@ -66,8 +70,8 @@ public async Task SpecializeHostCoreAsync()
             // Trigger a configuration reload to pick up all current settings
             _configuration?.Reload();
 
+            await _languageWorkerChannelManager.SpecializeAsync();
             NotifyChange();
-
             await _scriptHostManager.RestartHostAsync();
             await _scriptHostManager.DelayUntilHostReady();
         }

src/WebJobs.Script.WebHost/WebHostServiceCollectionExtensions.cs

@@ -65,9 +65,6 @@ public static void AddWebJobsScriptHost(this IServiceCollection services, IConfi
             // Standby services
             services.AddStandbyServices();
 
-            // Core script host services
-            services.AddSingleton<WebJobsScriptHostService>();
-            services.AddSingleton<IHostedService>(s => s.GetRequiredService<WebJobsScriptHostService>());
             services.AddSingleton<IScriptHostManager>(s => s.GetRequiredService<WebJobsScriptHostService>());
             services.AddSingleton<IScriptWebHostEnvironment, ScriptWebHostEnvironment>();
             services.AddSingleton<IStandbyManager, StandbyManager>();
@@ -99,7 +96,6 @@ public static void AddWebJobsScriptHost(this IServiceCollection services, IConfi
             // Management services
             services.AddSingleton<IWebFunctionsManager, WebFunctionsManager>();
             services.AddSingleton<IInstanceManager, InstanceManager>();
-
             services.AddSingleton(_ => new HttpClient());
             services.AddSingleton<IFileSystem>(_ => FileUtility.Instance);
             services.AddTransient<VirtualFileSystem>();
@@ -109,8 +105,13 @@ public static void AddWebJobsScriptHost(this IServiceCollection services, IConfi
             services.TryAddSingleton<ISecretManagerProvider, DefaultSecretManagerProvider>();
 
             // Register common services with the WebHost
+            // Language Worker Hosted Services need to be intialized before WebJobsScriptHostService
             ScriptHostBuilderExtensions.AddCommonServices(services);
 
+            // Core script host services
+            services.AddSingleton<WebJobsScriptHostService>();
+            services.AddSingleton<IHostedService>(s => s.GetRequiredService<WebJobsScriptHostService>());
+
             // Configuration
             services.TryAddEnumerable(ServiceDescriptor.Singleton<IConfigureOptions<ScriptApplicationHostOptions>, ScriptApplicationHostOptionsSetup>());
             services.ConfigureOptions<LanguageWorkerOptionsSetup>();

src/WebJobs.Script.WebHost/WebJobsScriptHostService.cs

@@ -125,7 +125,8 @@ private async Task StartHostAsync(CancellationToken cancellationToken, int attem
                     State = ScriptHostState.Default;
                 }
 
-                bool isOffline = CheckAppOffline();
+                bool isOffline = Utility.CheckAppOffline(_applicationHostOptions.CurrentValue.ScriptPath);
+                State = isOffline ? ScriptHostState.Offline : State;
                 bool hasNonTransientErrors = startupMode.HasFlag(JobHostStartupMode.HandlingNonTransientError);
 
                 // If we're in a non-transient error state or offline, skip host initialization
@@ -290,19 +291,6 @@ private void LogInitialization(bool isOffline, int attemptCount, int startCount)
             }
         }
 
-        private bool CheckAppOffline()
-        {
-            // check if we should be in an offline state
-            string offlineFilePath = Path.Combine(_applicationHostOptions.CurrentValue.ScriptPath, ScriptConstants.AppOfflineFileName);
-            if (File.Exists(offlineFilePath))
-            {
-                State = ScriptHostState.Offline;
-                return true;
-            }
-
-            return false;
-        }
-
         private void OnHostHealthCheckTimer(object state)
         {
             bool isHealthy = IsHostHealthy();