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.

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:

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"),
        };

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) {}
}

At it's core, the language worker needs to handle this event stream

eventStream.on('data', (msg) => {
      let event = <string>msg.content;
      let eventHandler = (<any>this)[event];
      if (eventHandler) {
        eventHandler.apply(this, [msg.requestId, msg[event]]);
      } else {
        console.error(`Worker ${workerId} had no handler for message '${event}'`)
      }
    });
...
  public workerInitRequest(requestId: string, msg: rpc.WorkerInitRequest) {
    this._eventStream.write({
      requestId: requestId,
      workerInitResponse: {
        result: {
          status: Status.Success
        }
      }
    });
  }

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.

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;
  }
}

Language Extensibility