Language Extensibility

Language Extensibility

In order to bring Language Extensibility to functions, we've split the functions runtime into the host (which manages function events), and language worker processes (which runs user functions for a given language). These two pieces communicate using gRPC as a messaging layer.

Language Worker Implementations

• https://github.com/Azure/azure-functions-nodejs-worker
• https://github.com/Azure/azure-functions-java-worker

IWorkerProvider

We require a small amount of C# code in order to create a language worker. In particular, the IWorkerProvider interface must be implemented by a language worker. This provides the host with:

1. A static description of the worker (what language is supported, what file extensions are supported)

        public WorkerDescription GetDescription() => new WorkerDescription
        {
            Language = "Node",
            Extension = ".js",
            DefaultExecutablePath = "node",
            DefaultWorkerPath = Path.Combine("dist", "src", "nodejsWorker.js"),
        };

2. A hook to configure arguments which will be used by the host in order to start the worker. Arguments are set to sensible defaults based on DefaultExecutablePath and DefaultWorkerPath before being passed to TryConfigureArguments.

        public bool TryConfigureArguments(ArgumentsDescription args, IConfiguration config, ILogger logger)
        {
            var options = new DefaultWorkerOptions();
            config.GetSection("workers:node").Bind(options);
            if (options.TryGetDebugPort(out int debugPort))
            {
                args.ExecutableArguments.Add($"--inspect={debugPort}");
            }
            return true;
        }

The IWorkerProvider should be distributed via nuget, along with any artifacts that the runtime requires. By convention, these artifacts are placed in <functions-host-dir>/<output-dir>/workers/<language>

Currently, these IWorkerProviders are built and deployed with the functions host. However, we would like to use the BYOB extensions mechanism to load custom language workers in order to decouple from the host.

gRPC Contract

For gRPC, we currently have one call which creates a duplex stream of StreamingMessages, which are the protobuf messages defined below. All messages between host and worker flow across this bi-directional stream. The host runs the gRPC server, and the language worker processes act as clients.

service FunctionRpc {
 rpc EventStream (stream StreamingMessage) returns (stream StreamingMessage) {}
}

Protobuf Contract

Every message that crosses the channel is a StreamingMessage, which contains a request id and a content union that contains one sub-message.

This contract lives here: https://github.com/Azure/azure-webjobs-sdk-script/blob/dev/src/WebJobs.Script.Grpc/Proto/FunctionRpc.proto

message StreamingMessage {
  string request_id = 1;
  oneof content {
    // Worker signals to host that it has been started
    StartStream start_stream = 20; 

    // Host sends capabilities/init data to worker
    WorkerInitRequest worker_init_request = 17;

    // Worker responds after initializing with its capabilities & status
    WorkerInitResponse worker_init_response = 16;

    // Host sends required metadata to worker to load function
    FunctionLoadRequest function_load_request = 8;

    // Worker responds after loading with the load result
    FunctionLoadResponse function_load_response = 9;
    
    // Host sends invocation information (function id, binding data, parameters) to worker
    InvocationRequest invocation_request = 4;

    // Worker sends response to host
    InvocationResponse invocation_response = 5;

    // Structured log from the worker based off of the ILogger interface
    RpcLog rpc_log = 2;
  }
}

Implementing a Language Worker

At it's core, the language worker needs to provide event handlers for streaming messages. Here's an example in node:

eventStream.on('data', (msg) => {
      let event = <string>msg.content;       // find the type of streaming message
      let eventHandler = (<any>this)[event]; // find the handler on the class for the streaming message type
      if (eventHandler) {
        eventHandler.apply(this, [msg.requestId, msg[event]]);  // invoke the handler with the id and message
      } else {
        console.error(`Worker ${workerId} had no handler for message '${event}'`)
      }
    });
...
  // example of a streaming message handler for workerInitRequest
  public workerInitRequest(requestId: string, msg: rpc.WorkerInitRequest) {
    // nothing to be done, so write the workerInitResponse back to the stream
    this._eventStream.write({
      requestId: requestId,
      workerInitResponse: {
        result: {
          status: Status.Success
        }
      }
    });
  }

More message handler implementations here: https://github.com/Azure/azure-functions-nodejs-worker/blob/dev/src/WorkerChannel.ts

Host/Worker Lifecycle

1. Start Host
2. Start gRPC server
3. For each function, create a language worker process (via IWorkerProvider) if necessary. One process per language.
4. StartStream - Worker connects to Host gRPC server
5. WorkerInitRequest, WorkerInitResponse - Host and Worker exchange version info & capabilities
6. FunctionLoadRequest, FunctionLoadResponse - Host sends Worker function metadata, Worker loads function
7. InvocationRequest, InvocationResponse - Host sends function id, binding data, parameters, Worker response with function result