Implement ResourceProvider#StreamInvoke

[hausdorff]

No description provided.

[hausdorff]

@pgavlin I'm going to mark this as ready for review because I think it's directionally correct. LMK what you think.

[pgavlin]

Seems like streamInvoke only supports async mode?

[hausdorff]

Yes! On purpose. By definition it's an AsyncIterable.

[pgavlin]

Perfect. I don't think we need to throw in this case, then--opts.async is essentially meaningless, so we might as well ignore it entirely.

[pgavlin]

I still think we shouldn't throw here.

[hausdorff]

oops missed this. will fix tonight.

[hausdorff]

Ok I just made streamInvoke async natively.

[hausdorff]

Just an update on this change—we've decided that we need to be able to cancel a StreamInvoke from the language provider. For example:

const deployments = pulumi.runtime.streamInvoke("kubernetes:kubernetes:watch", {
    group: "apps",
    version: "v1",
    kind: "Deployment",
});

for await (const { object: deployment, type } of deployments) {
    console.log(`${deployment.metadata!.name} was ${type}`);
    break;
}

deployments.close(); // <-- Signals to the resource provider to stop the invoke.

The code will be done today, but it's probably best, given that people are trying to leave so they can go to the end-of-sprint unwinder, to check this in on Monday.

[hausdorff]

I've just added the ability to "cancel" a streamInvoke from the language provider, so this is close to being ready to go. The commit messages contain more context, but in short, it's now possible to write something like:

// `streamInvoke` to retrieve all updates to any `Deployment`, enumerate 1
// update from the stream, then `cancel` giving the Kubernetes provider to
// clean up and close gracefully.
const deployments = await streamInvoke("kubernetes:kubernetes:watch", {
    group: "apps", version: "v1", kind: "Deployment",
    break;
});
for await (const d of deployments) {
    break;
}
deployments.cancel();

The corresponding "graceful shutdown" code in the Kubernetes provider (which you can see in pulumi/pulumi-kubernetes#858) looks like the following snippet, and specifically in the last case. Something ~like this would have to be implemented in any provider that chose to support ResourceProvider#StreamInvoke.

for {
	select {
	case <-k.canceler.context.Done():
		//
		// `kubeProvider#Cancel` was called. Terminate the `StreamInvoke` RPC, free all
		// resources, and exit without error.
		//

		watch.Stop()
		return nil
	case event := <-watch.ResultChan():
		//
		// Kubernetes resource was updated. Publish resource update back to user.
		//

		resp, err := plugin.MarshalProperties(
			resource.NewPropertyMapFromMap(
				map[string]interface{}{
					"type":   event.Type,
					"object": event.Object.(*unstructured.Unstructured).Object,
				}),
			plugin.MarshalOptions{})
		if err != nil {
			return err
		}

		err = server.Send(&pulumirpc.InvokeResponse{Return: resp})
		if err != nil {
			return err
		}
	case <-server.Context().Done():
		//
		// gRPC stream was cancelled. Usually this happens in the language provider, e.g.,
		// in the call to `cancel` below. Terminate the `StreamInvoke` RPC, free all
		// resources, and exit without error.
		//
		//     const deployments = await streamInvoke("kubernetes:kubernetes:watch", {
		//         group: "apps", version: "v1", kind: "Deployment",
		//     });
		//     deployments.cancel();
		//

		watch.Stop()
		return nil
	}
}

[pgavlin]

This is a bit weird--do we expect to only get these on the first message?

[hausdorff]

Yeah, it's weird. We're streaming InvokeResponse back, but because we don't have a separate, first-class structured error channel it seems like you have to cram the error conditions into the message type—which we do both here and in Invoke. So you end up with this weird thing.

We could just take this out. It's not clear to me which is better.

[pgavlin]

Eh, I think we can leave it for now. Maybe add a note with what you described above.