Node.js Worker 2.0.0 and Capability Flag
Update to Node.js Worker 2.0.0 and send appropriate capability flag
MODIFY
sample/Node/HttpTrigger-CustomRoute-Get/index.js
sample/Node/HttpTrigger-CustomRoute-Get/index.js
@@ -28,5 +28,5 @@ module.exports = function (context, req) {
}
});
- context.done(null, results);
+ context.done(null, { body: results });
}

MODIFY
src/WebJobs.Script/WebJobs.Script.csproj
src/WebJobs.Script/WebJobs.Script.csproj
<PackageReference Include="Grpc.Core" Version="1.20.1" />
<PackageReference Include="Microsoft.Azure.AppService.Proxy.Client" Version="2.0.10120001-f33e57bd" />
<PackageReference Include="Microsoft.Azure.Functions.JavaWorker" Version="1.5.2-SNAPSHOT" />
- <PackageReference Include="Microsoft.Azure.Functions.NodeJsWorker" Version="1.2.0" />
+ <PackageReference Include="Microsoft.Azure.Functions.NodeJsWorker" Version="2.0.0" />
<PackageReference Include="Microsoft.Azure.Functions.PowerShellWorker" Version="1.0.197" />
<PackageReference Include="Microsoft.Azure.WebJobs" Version="3.0.15" />
<PackageReference Include="Microsoft.Azure.WebJobs.Extensions" Version="3.0.5" />

MODIFY
src/WebJobs.Script/Workers/Rpc/RpcWorkerChannel.cs
src/WebJobs.Script/Workers/Rpc/RpcWorkerChannel.cs
@@ -30,6 +30,7 @@ internal class RpcWorkerChannel : IRpcWorkerChannel, IDisposable
private readonly IScriptEventManager _eventManager;
private readonly RpcWorkerConfig _workerConfig;
private readonly string _runtime;
+ private readonly IEnvironment _environment;
private readonly IOptionsMonitor<ScriptApplicationHostOptions> _applicationHostOptions;
private IDisposable _functionLoadRequestResponseEvent;
@@ -63,6 +64,7 @@ internal class RpcWorkerChannel : IRpcWorkerChannel, IDisposable
ILogger logger,
IMetricsLogger metricsLogger,
int attemptCount,
+ IEnvironment environment,
IOptionsMonitor<ScriptApplicationHostOptions> applicationHostOptions)
{
_workerId = workerId;
@@ -72,6 +74,7 @@ internal class RpcWorkerChannel : IRpcWorkerChannel, IDisposable
_rpcWorkerProcess = rpcWorkerProcess;
_workerChannelLogger = logger;
_metricsLogger = metricsLogger;
+ _environment = environment;
_applicationHostOptions = applicationHostOptions;
_workerCapabilities = new Capabilities(_workerChannelLogger);
@@ -133,12 +136,21 @@ internal void SendWorkerInitRequest(RpcEvent startEvent)
.Take(1)
.Subscribe(WorkerInitResponse, HandleWorkerInitError);
- SendStreamingMessage(new StreamingMessage
+ var initRequest = new WorkerInitRequest()
{
- WorkerInitRequest = new WorkerInitRequest()
+ HostVersion = ScriptHost.Version,
+ };
+
+ // Run as Functions Host V2 compatible
+ if (_environment.IsV2CompatibilityMode())
{
- HostVersion = ScriptHost.Version
+ _workerChannelLogger.LogDebug("Worker and host running in V2 compatibility mode");
+ initRequest.Capabilities.Add(RpcWorkerConstants.V2Compatable, "true");
}
+
+ SendStreamingMessage(new StreamingMessage
+ {
+ WorkerInitRequest = initRequest
});
}
@@ -161,6 +173,7 @@ internal void WorkerInitResponse(RpcEvent initEvent)
_workerChannelLogger.LogDebug("Received WorkerInitResponse. Worker process initialized");
_initMessage = initEvent.Message.WorkerInitResponse;
+ _workerChannelLogger.LogDebug($"Worker capabilities: {_initMessage.Capabilities}");
if (_initMessage.Result.IsFailure(out Exception exc))
{
HandleWorkerInitError(exc);

MODIFY
src/WebJobs.Script/Workers/Rpc/RpcWorkerChannelFactory.cs
src/WebJobs.Script/Workers/Rpc/RpcWorkerChannelFactory.cs
@@ -20,6 +20,7 @@ public class RpcWorkerChannelFactory : IRpcWorkerChannelFactory
private readonly IRpcWorkerProcessFactory _rpcWorkerProcessFactory = null;
private readonly IScriptEventManager _eventManager = null;
private readonly IEnumerable<RpcWorkerConfig> _workerConfigs = null;
+ private readonly IEnvironment _environment = null;
private readonly IOptionsMonitor<ScriptApplicationHostOptions> _applicationHostOptions = null;
public RpcWorkerChannelFactory(IScriptEventManager eventManager, IEnvironment environment, IRpcServer rpcServer, ILoggerFactory loggerFactory, IOptions<LanguageWorkerOptions> languageWorkerOptions,
@@ -29,6 +30,7 @@ public class RpcWorkerChannelFactory : IRpcWorkerChannelFactory
_loggerFactory = loggerFactory;
_workerConfigs = languageWorkerOptions.Value.WorkerConfigs;
_rpcWorkerProcessFactory = rpcWorkerProcessManager;
+ _environment = environment;
_applicationHostOptions = applicationHostOptions;
}
@@ -50,6 +52,7 @@ public IRpcWorkerChannel Create(string scriptRootPath, string runtime, IMetricsL
workerLogger,
metricsLogger,
attemptCount,
+ _environment,
_applicationHostOptions);
}
}

MODIFY
src/WebJobs.Script/Workers/Rpc/RpcWorkerConstants.cs
src/WebJobs.Script/Workers/Rpc/RpcWorkerConstants.cs
@@ -38,12 +38,15 @@ public static class RpcWorkerConstants
public const string RpcHttpStatusCode = "statusCode";
public const string RpcHttpStatus = "status";
- // Capabilites
+ // Worker Capabilites
public const string RawHttpBodyBytes = "RawHttpBodyBytes";
public const string TypedDataCollection = "TypedDataCollection";
public const string RpcHttpBodyOnly = "RpcHttpBodyOnly";
public const string RpcHttpTriggerMetadataRemoved = "RpcHttpTriggerMetadataRemoved";
+ // Host Capabilites
+ public const string V2Compatable = "V2Compatable";
+
// dotnet executable file path components
public const string DotNetExecutableName = "dotnet";
public const string DotNetExecutableNameWithExtension = DotNetExecutableName + ".exe";

MODIFY
test/WebJobs.Script.Tests.Integration/WebJobs.Script.Tests.Integration.csproj
test/WebJobs.Script.Tests.Integration/WebJobs.Script.Tests.Integration.csproj
<PackageReference Include="Microsoft.AspNetCore.TestHost" Version="3.0.0" />
<PackageReference Include="Microsoft.Azure.DocumentDB.Core" Version="2.3.0" />
<PackageReference Include="Microsoft.Azure.EventHubs" Version="2.1.0" />
- <PackageReference Include="Microsoft.Azure.Functions.NodeJsWorker" Version="1.2.0" />
+ <PackageReference Include="Microsoft.Azure.Functions.NodeJsWorker" Version="2.0.0" />
<PackageReference Include="Microsoft.Azure.WebJobs.Extensions.Storage" Version="3.0.10-11660" />
<PackageReference Include="Microsoft.Azure.Functions.JavaWorker" Version="1.5.2-SNAPSHOT" />
<PackageReference Include="Microsoft.Azure.Mobile.Client" Version="4.0.2" />

MODIFY
test/WebJobs.Script.Tests/Workers/Rpc/RpcWorkerChannelTests.cs
test/WebJobs.Script.Tests/Workers/Rpc/RpcWorkerChannelTests.cs
@@ -39,6 +39,7 @@ public class RpcWorkerChannelTests
private readonly TestLogger _logger;
private readonly IEnumerable<FunctionMetadata> _functions = new List<FunctionMetadata>();
private readonly RpcWorkerConfig _testWorkerConfig;
+ private readonly TestEnvironment _testEnvironment;
private readonly IOptionsMonitor<ScriptApplicationHostOptions> _hostOptionsMonitor;
private RpcWorkerChannel _workerChannel;
@@ -48,6 +49,7 @@ public RpcWorkerChannelTests()
_testFunctionRpcService = new TestFunctionRpcService(_eventManager, _workerId, _logger, _expectedLogMsg);
_testWorkerConfig = TestHelpers.GetTestWorkerConfigs().FirstOrDefault();
_mockrpcWorkerProcess.Setup(m => m.StartProcessAsync()).Returns(Task.CompletedTask);
+ _testEnvironment = new TestEnvironment();
var hostOptions = new ScriptApplicationHostOptions
{
@@ -67,6 +69,7 @@ public RpcWorkerChannelTests()
_logger,
_metricsLogger,
0,
+ _testEnvironment,
_hostOptionsMonitor);
}
@@ -116,6 +119,7 @@ public async Task StartWorkerProcessAsync_WorkerProcess_Throws()
_logger,
_metricsLogger,
0,
+ _testEnvironment,
_hostOptionsMonitor);
await Assert.ThrowsAsync<FileNotFoundException>(async () => await _workerChannel.StartWorkerProcessAsync());
}
@@ -138,6 +142,26 @@ public void SendWorkerInitRequest_PublishesOutboundEvent()
Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, _expectedLogMsg)));
}
+ [Fact]
+ public void SendWorkerInitRequest_PublishesOutboundEvent_V2Compatable()
+ {
+ _testEnvironment.SetEnvironmentVariable(EnvironmentSettingNames.FunctionsV2CompatibilityModeKey, "true");
+ StartStream startStream = new StartStream()
+ {
+ WorkerId = _workerId
+ };
+ StreamingMessage startStreamMessage = new StreamingMessage()
+ {
+ StartStream = startStream
+ };
+ RpcEvent rpcEvent = new RpcEvent(_workerId, startStreamMessage);
+ _workerChannel.SendWorkerInitRequest(rpcEvent);
+ _testFunctionRpcService.PublishWorkerInitResponseEvent();
+ var traces = _logger.GetLogMessages();
+ Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, _expectedLogMsg)));
+ Assert.True(traces.Any(m => string.Equals(m.FormattedMessage, "Worker and host running in V2 compatibility mode")));
+ }
+
[Theory]
[InlineData(RpcLog.Types.Level.Information, RpcLog.Types.Level.Information)]
[InlineData(RpcLog.Types.Level.Error, RpcLog.Types.Level.Error)]
@@ -172,6 +196,7 @@ public async Task Drain_Verify()
_logger,
_metricsLogger,
0,
+ _testEnvironment,
_hostOptionsMonitor);
channel.SetupFunctionInvocationBuffers(GetTestFunctionsList("node"));
ScriptInvocationContext scriptInvocationContext = GetTestScriptInvocationContext(invocationId, resultSource);

Node.js Worker 2.0.0 and Capability Flag (#5321)

Update to Node.js Worker 2.0.0 and send appropriate capability flag