// Copyright 2016-2024, Pulumi Corporation. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. import * as grpc from "@grpc/grpc-js"; import { randomUUID } from "crypto"; import * as jspb from "google-protobuf"; import * as gstruct from "google-protobuf/google/protobuf/struct_pb"; import * as log from "../log"; import { output } from "../output"; import * as callrpc from "../proto/callback_grpc_pb"; import * as callproto from "../proto/callback_pb"; import { Callback, CallbackInvokeRequest, CallbackInvokeResponse } from "../proto/callback_pb"; import * as resrpc from "../proto/resource_grpc_pb"; import * as resproto from "../proto/resource_pb"; import { Alias, ComponentResourceOptions, CustomResourceOptions, DependencyProviderResource, DependencyResource, ProviderResource, Resource, ResourceOptions, ResourceTransform, ResourceTransformArgs, URN, rootStackResource, } from "../resource"; import { InvokeOptions, InvokeTransform, InvokeTransformArgs } from "../invoke"; import { mapAliasesForRequest } from "./resource"; import { deserializeProperties, serializeProperties, unknownValue } from "./rpc"; /** * Raises the gRPC Max Message size from `4194304` (4mb) to `419430400` (400mb) * * @internal */ const maxRPCMessageSize: number = 1024 * 1024 * 400; type CallbackFunction = (args: Uint8Array) => Promise<jspb.Message>; export interface ICallbackServer { registerTransform(callback: ResourceTransform): Promise<callproto.Callback>; registerStackTransform(callback: ResourceTransform): void; registerStackInvokeTransform(callback: InvokeTransform): void; registerStackInvokeTransformAsync(callback: InvokeTransform): Promise<callproto.Callback>; shutdown(): void; // Wait for any pendind registerStackTransform calls to complete. awaitStackRegistrations(): Promise<void>; } export class CallbackServer implements ICallbackServer { private readonly _callbacks = new Map<string, CallbackFunction>(); private readonly _monitor: resrpc.IResourceMonitorClient; private readonly _server: grpc.Server; private readonly _target: Promise<string>; private _pendingRegistrations: number = 0; private _awaitQueue: ((reason?: any) => void)[] = []; constructor(monitor: resrpc.IResourceMonitorClient) { this._monitor = monitor; this._server = new grpc.Server({ "grpc.max_receive_message_length": maxRPCMessageSize, }); const implementation: callrpc.ICallbacksServer = { invoke: this.invoke.bind(this), }; this._server.addService(callrpc.CallbacksService, implementation); const self = this; this._target = new Promise<string>((resolve, reject) => { self._server.bindAsync(`127.0.0.1:0`, grpc.ServerCredentials.createInsecure(), (err, port) => { if (err !== null) { reject(err); return; } // The server takes a while to _actually_ startup so we need to keep trying to send an invoke // to ourselves before we resolve the address to tell the engine about it. const target = `127.0.0.1:${port}`; const client = new callrpc.CallbacksClient(target, grpc.credentials.createInsecure()); const connect = () => { client.invoke(new CallbackInvokeRequest(), (error, _) => { if (error?.code === grpc.status.UNAVAILABLE) { setTimeout(connect, 1000); return; } // The expected error given we didn't give a token to the invoke. if (error?.details === "callback not found: ") { resolve(target); return; } reject(error); }); }; connect(); }); }); } awaitStackRegistrations(): Promise<void> { if (this._pendingRegistrations === 0) { return Promise.resolve(); } return new Promise<void>((resolve, reject) => { this._awaitQueue.push((reason?: any) => { if (reason !== undefined) { reject(reason); } else { resolve(); } }); }); } shutdown(): void { this._server.forceShutdown(); } private async invoke( call: grpc.ServerUnaryCall<CallbackInvokeRequest, CallbackInvokeResponse>, callback: grpc.sendUnaryData<CallbackInvokeResponse>, ) { const req = call.request; const cb = this._callbacks.get(req.getToken()); if (cb === undefined) { const err = new grpc.StatusBuilder(); err.withCode(grpc.status.INVALID_ARGUMENT); err.withDetails("callback not found: " + req.getToken()); callback(err.build()); return; } try { const response = await cb(req.getRequest_asU8()); const resp = new CallbackInvokeResponse(); resp.setResponse(response.serializeBinary()); callback(null, resp); } catch (e) { const err = new grpc.StatusBuilder(); err.withCode(grpc.status.UNKNOWN); if (e instanceof Error) { err.withDetails(e.message); } else { err.withDetails(JSON.stringify(e)); } callback(err.build()); } } async registerTransform(transform: ResourceTransform): Promise<callproto.Callback> { const cb = async (bytes: Uint8Array): Promise<jspb.Message> => { const request = resproto.TransformRequest.deserializeBinary(bytes); let opts = request.getOptions() || new resproto.TransformResourceOptions(); let ropts: ResourceOptions; if (request.getCustom()) { ropts = { deleteBeforeReplace: opts.getDeleteBeforeReplace(), additionalSecretOutputs: opts.getAdditionalSecretOutputsList(), } as CustomResourceOptions; } else { const providers: Record<string, ProviderResource> = {}; for (const [key, value] of opts.getProvidersMap().entries()) { providers[key] = new DependencyProviderResource(value); } ropts = { providers: providers, } as ComponentResourceOptions; } ropts.aliases = opts.getAliasesList().map((alias): string | Alias => { if (alias.hasUrn()) { return alias.getUrn(); } else { const spec = alias.getSpec(); if (spec === undefined) { throw new Error("alias must have either a urn or a spec"); } const nodeAlias: Alias = { name: spec.getName(), type: spec.getType(), project: spec.getProject(), stack: spec.getStack(), parent: spec.getParenturn() !== "" ? new DependencyResource(spec.getParenturn()) : undefined, }; if (spec.getNoparent()) { nodeAlias.parent = rootStackResource; } return nodeAlias; } }); const timeouts = opts.getCustomTimeouts(); if (timeouts !== undefined) { ropts.customTimeouts = { create: timeouts.getCreate(), update: timeouts.getUpdate(), delete: timeouts.getDelete(), }; } ropts.deletedWith = opts.getDeletedWith() !== "" ? new DependencyResource(opts.getDeletedWith()) : undefined; ropts.dependsOn = opts.getDependsOnList().map((dep) => new DependencyResource(dep)); ropts.ignoreChanges = opts.getIgnoreChangesList(); ropts.parent = request.getParent() !== "" ? new DependencyResource(request.getParent()) : undefined; ropts.pluginDownloadURL = opts.getPluginDownloadUrl() !== "" ? opts.getPluginDownloadUrl() : undefined; ropts.protect = opts.getProtect(); ropts.provider = opts.getProvider() !== "" ? new DependencyProviderResource(opts.getProvider()) : undefined; ropts.replaceOnChanges = opts.getReplaceOnChangesList(); ropts.retainOnDelete = opts.getRetainOnDelete(); ropts.version = opts.getVersion() !== "" ? opts.getVersion() : undefined; const props = request.getProperties(); const args: ResourceTransformArgs = { custom: request.getCustom(), type: request.getType(), name: request.getName(), props: props === undefined ? {} : deserializeProperties(props), opts: ropts, }; const result = await transform(args); const response = new resproto.TransformResponse(); if (result === undefined) { response.setProperties(request.getProperties()); response.setOptions(request.getOptions()); } else { const mprops = await serializeProperties("props", result.props); response.setProperties(gstruct.Struct.fromJavaScript(mprops)); // Copy the options over. if (result.opts !== undefined) { opts = new resproto.TransformResourceOptions(); if (result.opts.aliases !== undefined) { const aliases = []; const uniqueAliases = new Set<Alias | URN>(); for (const alias of result.opts.aliases || []) { const aliasVal = await output(alias).promise(); if (!uniqueAliases.has(aliasVal)) { uniqueAliases.add(aliasVal); aliases.push(aliasVal); } } opts.setAliasesList(await mapAliasesForRequest(aliases, request.getParent())); } if (result.opts.customTimeouts !== undefined) { const customTimeouts = new resproto.RegisterResourceRequest.CustomTimeouts(); if (result.opts.customTimeouts.create !== undefined) { customTimeouts.setCreate(result.opts.customTimeouts.create); } if (result.opts.customTimeouts.update !== undefined) { customTimeouts.setUpdate(result.opts.customTimeouts.update); } if (result.opts.customTimeouts.delete !== undefined) { customTimeouts.setDelete(result.opts.customTimeouts.delete); } opts.setCustomTimeouts(customTimeouts); } if (result.opts.deletedWith !== undefined) { opts.setDeletedWith(await result.opts.deletedWith.urn.promise()); } if (result.opts.dependsOn !== undefined) { const resolvedDeps = await output(result.opts.dependsOn).promise(); const deps = []; if (Resource.isInstance(resolvedDeps)) { deps.push(await resolvedDeps.urn.promise()); } else { for (const dep of resolvedDeps) { deps.push(await dep.urn.promise()); } } opts.setDependsOnList(deps); } if (result.opts.ignoreChanges !== undefined) { opts.setIgnoreChangesList(result.opts.ignoreChanges); } if (result.opts.pluginDownloadURL !== undefined) { opts.setPluginDownloadUrl(result.opts.pluginDownloadURL); } if (result.opts.protect !== undefined) { opts.setProtect(result.opts.protect); } if (result.opts.provider !== undefined) { const providerURN = await result.opts.provider.urn.promise(); const providerID = (await result.opts.provider.id.promise()) || unknownValue; opts.setProvider(`${providerURN}::${providerID}`); } if (result.opts.replaceOnChanges !== undefined) { opts.setReplaceOnChangesList(result.opts.replaceOnChanges); } if (result.opts.retainOnDelete !== undefined) { opts.setRetainOnDelete(result.opts.retainOnDelete); } if (result.opts.version !== undefined) { opts.setVersion(result.opts.version); } if (request.getCustom()) { const copts = result.opts as CustomResourceOptions; if (copts.deleteBeforeReplace !== undefined) { opts.setDeleteBeforeReplace(copts.deleteBeforeReplace); } if (copts.additionalSecretOutputs !== undefined) { opts.setAdditionalSecretOutputsList(copts.additionalSecretOutputs); } } else { const copts = result.opts as ComponentResourceOptions; if (copts.providers !== undefined) { const providers = opts.getProvidersMap(); if (copts.providers && !Array.isArray(copts.providers)) { for (const k in copts.providers) { if (Object.prototype.hasOwnProperty.call(copts.providers, k)) { const v = copts.providers[k]; if (k !== v.getPackage()) { const message = `provider resource map where key ${k} doesn't match provider ${v.getPackage()}`; log.warn(message); } } } } const provs = Object.values(copts.providers); for (const prov of provs) { const providerURN = await prov.urn.promise(); const providerID = (await prov.id.promise()) || unknownValue; providers.set(prov.getPackage(), `${providerURN}::${providerID}`); } opts.clearProvidersMap(); } } } response.setOptions(opts); } return response; }; const tryCb = async (bytes: Uint8Array): Promise<jspb.Message> => { try { return await cb(bytes); } catch (e) { throw new Error(`transform failed: ${e}`); } }; const uuid = randomUUID(); this._callbacks.set(uuid, tryCb); const req = new Callback(); req.setToken(uuid); req.setTarget(await this._target); return req; } registerStackTransform(transform: ResourceTransform): void { this._pendingRegistrations++; this.registerTransform(transform) .then( (req) => { return new Promise((resolve, reject) => { this._monitor.registerStackTransform(req, (err, _) => { if (err !== null) { // Remove this from the list of callbacks given we didn't manage to actually register it. this._callbacks.delete(req.getToken()); reject(); } else { resolve(); } }); }); }, (err) => log.error(`failed to register stack transform: ${err}`), ) .finally(() => { this._pendingRegistrations--; if (this._pendingRegistrations === 0) { const queue = this._awaitQueue; this._awaitQueue = []; for (const waiter of queue) { waiter(); } } }); } async registerStackInvokeTransformAsync(transform: InvokeTransform): Promise<callproto.Callback> { const cb = async (bytes: Uint8Array): Promise<jspb.Message> => { const request = resproto.TransformInvokeRequest.deserializeBinary(bytes); let opts = request.getOptions() || new resproto.TransformInvokeOptions(); const ropts: InvokeOptions = {}; ropts.pluginDownloadURL = opts.getPluginDownloadUrl() !== "" ? opts.getPluginDownloadUrl() : undefined; ropts.provider = opts.getProvider() !== "" ? new DependencyProviderResource(opts.getProvider()) : undefined; ropts.version = opts.getVersion() !== "" ? opts.getVersion() : undefined; const invokeArgs = request.getArgs(); const args: InvokeTransformArgs = { token: request.getToken(), args: invokeArgs === undefined ? {} : deserializeProperties(invokeArgs), opts: ropts, }; const result = await transform(args); const response = new resproto.TransformInvokeResponse(); if (result === undefined) { response.setArgs(request.getArgs()); response.setOptions(request.getOptions()); } else { const margs = await serializeProperties("args", result.args); response.setArgs(gstruct.Struct.fromJavaScript(margs)); // Copy the options over. if (result.opts !== undefined) { opts = new resproto.TransformInvokeOptions(); if (result.opts.pluginDownloadURL !== undefined) { opts.setPluginDownloadUrl(result.opts.pluginDownloadURL); } if (result.opts.provider !== undefined) { const providerURN = await result.opts.provider.urn.promise(); const providerID = (await result.opts.provider.id.promise()) || unknownValue; opts.setProvider(`${providerURN}::${providerID}`); } if (result.opts.version !== undefined) { opts.setVersion(result.opts.version); } response.setOptions(opts); } } return response; }; const tryCb = async (bytes: Uint8Array): Promise<jspb.Message> => { try { return await cb(bytes); } catch (e) { throw new Error(`transform failed: ${e}`); } }; const uuid = randomUUID(); this._callbacks.set(uuid, tryCb); const req = new Callback(); req.setToken(uuid); req.setTarget(await this._target); return req; } registerStackInvokeTransform(transform: InvokeTransform): void { this._pendingRegistrations++; this.registerStackInvokeTransformAsync(transform) .then( (req) => { return new Promise((resolve, reject) => { this._monitor.registerStackInvokeTransform(req, (err, _) => { if (err !== null) { // Remove this from the list of callbacks given we didn't manage to actually register it. this._callbacks.delete(req.getToken()); reject(); } else { resolve(); } }); }); }, (err) => log.error(`failed to register stack transform: ${err}`), ) .finally(() => { this._pendingRegistrations--; if (this._pendingRegistrations === 0) { const queue = this._awaitQueue; this._awaitQueue = []; for (const waiter of queue) { waiter(); } } }); } }