Adding ability to allow worker to send Function load responses in batch (#8107)

* Adding FunctionLoadResponses to send single load response in case of multiple functions

* Subscribed to LoadResponseCollection

* Updated subtree from https://github.com/azure/azure-functions-language-worker-protobuf. Tag: v1.5.4-protofile. Commit: 576c9de

* Removing redundant line

* Added failure testcase for LoadRepsonseCollection

Author: surgupta-msft

src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs

@@ -294,11 +294,19 @@ public void SendFunctionLoadRequests(ManagedDependencyOptions managedDependencyO
                         .Timeout(_functionLoadTimeout)
                         .Take(_functions.Count())
                         .Subscribe((msg) => LoadResponse(msg.Message.FunctionLoadResponse), HandleWorkerFunctionLoadError));
+
+                    _eventSubscriptions.Add(_inboundWorkerEvents.Where(msg => msg.MessageType == MsgType.FunctionLoadResponseCollection)
+                        .Timeout(_functionLoadTimeout)
+                        .Take(_functions.Count())
+                        .Subscribe((msg) => LoadResponse(msg.Message.FunctionLoadResponseCollection), HandleWorkerFunctionLoadError));
                 }
                 else
                 {
                     _eventSubscriptions.Add(_inboundWorkerEvents.Where(msg => msg.MessageType == MsgType.FunctionLoadResponse)
                         .Subscribe((msg) => LoadResponse(msg.Message.FunctionLoadResponse), HandleWorkerFunctionLoadError));
+
+                    _eventSubscriptions.Add(_inboundWorkerEvents.Where(msg => msg.MessageType == MsgType.FunctionLoadResponseCollection)
+                        .Subscribe((msg) => LoadResponse(msg.Message.FunctionLoadResponseCollection), HandleWorkerFunctionLoadError));
                 }
 
                 // Load Request is also sent for disabled function as it is invocable using the portal and admin endpoints
@@ -462,6 +470,16 @@ internal void LoadResponse(FunctionLoadResponse loadResponse)
             _inputLinks.Add(disposableLink);
         }
 
+        internal void LoadResponse(FunctionLoadResponseCollection loadResponseCollection)
+        {
+            _workerChannelLogger.LogDebug("Received FunctionLoadResponseCollection with number of functions: '{count}'.", loadResponseCollection.FunctionLoadResponses.Count);
+
+            foreach (FunctionLoadResponse loadResponse in loadResponseCollection.FunctionLoadResponses)
+            {
+                LoadResponse(loadResponse);
+            }
+        }
+
         internal async Task SendInvocationRequest(ScriptInvocationContext context)
         {
             try

test/WebJobs.Script.Tests/Workers/Rpc/GrpcWorkerChannelTests.cs

@@ -489,6 +489,39 @@ public void ReceivesInboundEvent_FunctionLoadResponse()
             Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Received FunctionLoadResponse for function: 'js1' with functionId: 'TestFunctionId1'.")));
         }
 
+        [Fact]
+        public void ReceivesInboundEvent_Failed_FunctionLoadResponses()
+        {
+            var functionMetadatas = GetTestFunctionsList("node");
+            _workerChannel.SetupFunctionInvocationBuffers(functionMetadatas);
+            _workerChannel.SendFunctionLoadRequests(null, TimeSpan.FromMinutes(1));
+            _testFunctionRpcService.PublishFunctionLoadResponsesEvent(
+                            new List<string>() { "TestFunctionId1", "TestFunctionId2" },
+                            new StatusResult() { Status = StatusResult.Types.Status.Failure });
+            var traces = _logger.GetLogMessages();
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Setting up FunctionInvocationBuffer for function: 'js1' with functionId: 'TestFunctionId1'")));
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Setting up FunctionInvocationBuffer for function: 'js2' with functionId: 'TestFunctionId2'")));
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Worker failed to load function: 'js1' with function id: 'TestFunctionId1'.")));
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Worker failed to load function: 'js2' with function id: 'TestFunctionId2'.")));
+        }
+
+        [Fact]
+        public void ReceivesInboundEvent_FunctionLoadResponses()
+        {
+            var functionMetadatas = GetTestFunctionsList("node");
+            _workerChannel.SetupFunctionInvocationBuffers(functionMetadatas);
+            _workerChannel.SendFunctionLoadRequests(null, TimeSpan.FromMinutes(1));
+            _testFunctionRpcService.PublishFunctionLoadResponsesEvent(
+                            new List<string>() { "TestFunctionId1", "TestFunctionId2" },
+                            new StatusResult() { Status = StatusResult.Types.Status.Success });
+            var traces = _logger.GetLogMessages();
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Setting up FunctionInvocationBuffer for function: 'js1' with functionId: 'TestFunctionId1'")));
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Setting up FunctionInvocationBuffer for function: 'js2' with functionId: 'TestFunctionId2'")));
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, string.Format("Received FunctionLoadResponseCollection with number of functions: '{0}'.", functionMetadatas.ToList().Count))));
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Received FunctionLoadResponse for function: 'js1' with functionId: 'TestFunctionId1'.")));
+            Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Received FunctionLoadResponse for function: 'js2' with functionId: 'TestFunctionId2'.")));
+        }
+
         [Fact]
         public void ReceivesInboundEvent_Successful_FunctionMetadataResponse()
         {

test/WebJobs.Script.Tests/Workers/Rpc/TestFunctionRpcService.cs

@@ -62,6 +62,28 @@ public void PublishFunctionLoadResponseEvent(string functionId)
             _eventManager.Publish(new InboundGrpcEvent(_workerId, responseMessage));
         }
 
+        public void PublishFunctionLoadResponsesEvent(List<string> functionIds, StatusResult statusResult)
+        {
+            FunctionLoadResponseCollection functionLoadResponseCollection = new FunctionLoadResponseCollection();
+
+            foreach (string functionId in functionIds)
+            {
+                FunctionLoadResponse functionLoadResponse = new FunctionLoadResponse()
+                {
+                    FunctionId = functionId,
+                    Result = statusResult
+                };
+
+                functionLoadResponseCollection.FunctionLoadResponses.Add(functionLoadResponse);
+            }
+
+            StreamingMessage responseMessage = new StreamingMessage()
+            {
+                FunctionLoadResponseCollection = functionLoadResponseCollection
+            };
+            _eventManager.Publish(new InboundGrpcEvent(_workerId, responseMessage));
+        }
+
         public void PublishFunctionEnvironmentReloadResponseEvent()
         {
             FunctionEnvironmentReloadResponse relaodEnvResponse = GetTestFunctionEnvReloadResponse();