2019-10-15 05:08:06 +00:00
|
|
|
// Copyright 2016-2018, 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.
|
|
|
|
|
|
|
|
package main
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"encoding/binary"
|
|
|
|
"io"
|
|
|
|
|
2024-01-17 09:35:20 +00:00
|
|
|
"google.golang.org/protobuf/types/known/emptypb"
|
2019-10-15 05:08:06 +00:00
|
|
|
|
|
|
|
"google.golang.org/grpc/encoding"
|
|
|
|
"google.golang.org/grpc/encoding/proto"
|
Maintain alias compat for older Node.js SDKs on new CLIs
This change updates the engine to detect if a `RegisterResource` request
is coming from an older Node.js SDK that is using incorrect alias specs
and, if so, transforms the aliases to be correct. This allows us to
maintain compatibility for users who have upgraded their CLI but are
still using an older version of the Node.js SDK with incorrect alias
specs.
We detect if the request is from a Node.js SDK by looking at the gRPC
request's metadata headers, specifically looking at the "pulumi-runtime"
and "user-agent" headers.
First, if the request has a "pulumi-runtime" header with a value of
"nodejs", we know it's coming from the Node.js plugin. The Node.js
language plugin proxies gRPC calls from the Node.js SDK to the resource
monitor and the proxy now sets the "pulumi-runtime" header to "nodejs"
for `RegisterResource` calls.
Second, if the request has a "user-agent" header that starts with
"grpc-node-js/", we know it's coming from the Node.js SDK. This is the
case for inline programs in the automation API, which connects directly
to the resource monitor, rather than going through the language plugin's
proxy.
We can't just look at "user-agent", because in the proxy case it will
have a Go-specific "user-agent".
Updated Node.js SDKs set a new `aliasSpecs` field on the
`RegisterResource` request, which indicates that the alias specs are
correct, and no transforms are needed.
2023-05-31 22:37:59 +00:00
|
|
|
"google.golang.org/grpc/metadata"
|
2019-10-15 05:08:06 +00:00
|
|
|
|
2021-03-17 13:20:05 +00:00
|
|
|
"github.com/pulumi/pulumi/sdk/v3/go/common/util/logging"
|
|
|
|
pulumirpc "github.com/pulumi/pulumi/sdk/v3/proto/go"
|
2019-10-15 05:08:06 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
// pipes is the platform agnostic abstraction over a pair of channels we can read and write binary
|
|
|
|
// data over. It is provided through the `createPipes` functions provided in `proxy_unix.go` (where
|
|
|
|
// it is implemented on top of fifo files), and in `proxy_windows.go` (where it is implemented on
|
|
|
|
// top of named pipes).
|
|
|
|
type pipes interface {
|
|
|
|
// The directory containing the two streams to read and write from. This will be passed to the
|
|
|
|
// nodejs process so it can connect to our read and writes streams for communication.
|
|
|
|
directory() string
|
|
|
|
|
|
|
|
// Attempt to create and connect to the read and write streams.
|
|
|
|
connect() error
|
|
|
|
|
|
|
|
// The stream that we will use to read in requests send to us by the nodejs process.
|
|
|
|
reader() io.Reader
|
|
|
|
|
|
|
|
// The stream we will write responses back to the nodejs process with.
|
|
|
|
writer() io.Writer
|
|
|
|
|
|
|
|
// called when we're done with the pipes and want to clean up any os resources we may have
|
|
|
|
// allocated (for example, actual files and directories on disk).
|
|
|
|
shutdown()
|
|
|
|
}
|
|
|
|
|
|
|
|
func createAndServePipes(ctx context.Context, target pulumirpc.ResourceMonitorClient) (pipes, chan error, error) {
|
|
|
|
pipes, err := createPipes()
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
pipesDone := servePipes(ctx, pipes, target)
|
|
|
|
return pipes, pipesDone, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func servePipes(ctx context.Context, pipes pipes, target pulumirpc.ResourceMonitorClient) chan error {
|
|
|
|
done := make(chan error)
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
// Keep reading and writing from the pipes until we run into an error or are canceled.
|
|
|
|
err := func() error {
|
|
|
|
pbcodec := encoding.GetCodec(proto.Name)
|
|
|
|
|
|
|
|
err := pipes.connect()
|
|
|
|
if err != nil {
|
|
|
|
logging.V(10).Infof("Sync invoke: Error connecting to pipes: %s\n", err)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
// read a 4-byte request length
|
|
|
|
logging.V(10).Infoln("Sync invoke: Reading length from request pipe")
|
|
|
|
var reqLen uint32
|
|
|
|
if err := binary.Read(pipes.reader(), binary.BigEndian, &reqLen); err != nil {
|
|
|
|
// This is benign on shutdown.
|
|
|
|
if err == io.EOF {
|
|
|
|
// We were asked to gracefully cancel. Just exit now.
|
|
|
|
logging.V(10).Infof("Sync invoke: Gracefully shutting down")
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
logging.V(10).Infof("Sync invoke: Received error reading length from pipe: %s\n", err)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// read the request in full
|
|
|
|
logging.V(10).Infoln("Sync invoke: Reading message from request pipe")
|
|
|
|
reqBytes := make([]byte, reqLen)
|
|
|
|
if _, err := io.ReadFull(pipes.reader(), reqBytes); err != nil {
|
|
|
|
logging.V(10).Infof("Sync invoke: Received error reading message from pipe: %s\n", err)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// decode and dispatch the request
|
|
|
|
logging.V(10).Infof("Sync invoke: Unmarshalling request")
|
2022-04-14 09:59:46 +00:00
|
|
|
var req pulumirpc.ResourceInvokeRequest
|
2019-10-15 05:08:06 +00:00
|
|
|
if err := pbcodec.Unmarshal(reqBytes, &req); err != nil {
|
|
|
|
logging.V(10).Infof("Sync invoke: Received error reading full from pipe: %s\n", err)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
logging.V(10).Infof("Sync invoke: Invoking: %s", req.GetTok())
|
|
|
|
res, err := target.Invoke(ctx, &req)
|
2019-10-22 21:13:07 +00:00
|
|
|
// Unfortunately, `monitor.Invoke` can return errors for non-exceptional cases. This
|
|
|
|
// can happen, for example, if the underlying provider call ends up returning an
|
|
|
|
// error itself. A common case of this is the aws-provider which will often return
|
|
|
|
// errors like:
|
|
|
|
//
|
|
|
|
// aws:ec2/getRouteTable:getRouteTable: Your query returned no results.
|
|
|
|
//
|
|
|
|
// This is not an error in the sense that something has gone terribly wrong and we
|
|
|
|
// need to shutdown. Rather, it's just a validation issue that needs to be passed
|
|
|
|
// back to the user program to handle.
|
|
|
|
//
|
|
|
|
// In the async codepath, this just returns as a grpc error which is thrown in the
|
|
|
|
// sdk code (and can be handled by the user). Unfortunately, our sync-rpc pipes have
|
|
|
|
// no way to send either an err or a success result. They can only send success,
|
|
|
|
// and only *terminate* on error.
|
|
|
|
//
|
|
|
|
// So, to workaround this issue, we abuse the 'success result' slightly.
|
|
|
|
// Specifically, we take advantage of the fact that the success-result contains a
|
|
|
|
// list of 'failures' to indicate if 'Check' reported any problems. We just stash
|
|
|
|
// this error into that array, knowing that the SDK will see it and immediately
|
|
|
|
// throw, just like it would have done for an RPC error in the normal async case.
|
2019-10-15 05:08:06 +00:00
|
|
|
if err != nil {
|
|
|
|
logging.V(10).Infof("Sync invoke: Received error invoking: %s\n", err)
|
2019-10-22 21:13:07 +00:00
|
|
|
logging.V(10).Infof("Sync invoke: Converting error to response.\n")
|
|
|
|
if res == nil {
|
|
|
|
res = &pulumirpc.InvokeResponse{}
|
|
|
|
}
|
|
|
|
|
|
|
|
res.Failures = append(res.Failures, &pulumirpc.CheckFailure{
|
|
|
|
Property: "",
|
|
|
|
Reason: err.Error(),
|
|
|
|
})
|
2019-10-15 05:08:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// encode the response
|
|
|
|
logging.V(10).Infof("Sync invoke: Marshalling response")
|
|
|
|
resBytes, err := pbcodec.Marshal(res)
|
|
|
|
if err != nil {
|
|
|
|
logging.V(10).Infof("Sync invoke: Received error marshalling: %s\n", err)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// write the 4-byte response length
|
|
|
|
logging.V(10).Infoln("Sync invoke: Writing length to request pipe")
|
|
|
|
if err := binary.Write(pipes.writer(), binary.BigEndian, uint32(len(resBytes))); err != nil {
|
|
|
|
logging.V(10).Infof("Sync invoke: Error writing length to pipe: %s\n", err)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// write the response in full
|
|
|
|
logging.V(10).Infoln("Sync invoke: Writing message to request pipe")
|
|
|
|
if _, err := pipes.writer().Write(resBytes); err != nil {
|
|
|
|
logging.V(10).Infof("Sync invoke: Error writing message to pipe: %s\n", err)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
// Signal our caller that we're done.
|
|
|
|
done <- err
|
|
|
|
close(done)
|
|
|
|
|
|
|
|
// cleanup any resources the pipes were holding onto.
|
|
|
|
pipes.shutdown()
|
|
|
|
}()
|
|
|
|
|
|
|
|
return done
|
|
|
|
}
|
|
|
|
|
|
|
|
// Forward all resource monitor calls that we're serving to nodejs back to the engine to actually
|
|
|
|
// perform.
|
|
|
|
|
|
|
|
type monitorProxy struct {
|
2024-03-07 08:52:34 +00:00
|
|
|
pulumirpc.UnsafeResourceMonitorServer
|
2022-12-14 19:20:26 +00:00
|
|
|
|
2019-10-15 05:08:06 +00:00
|
|
|
target pulumirpc.ResourceMonitorClient
|
|
|
|
}
|
|
|
|
|
|
|
|
func (p *monitorProxy) Invoke(
|
2023-03-03 16:36:39 +00:00
|
|
|
ctx context.Context, req *pulumirpc.ResourceInvokeRequest,
|
|
|
|
) (*pulumirpc.InvokeResponse, error) {
|
2019-10-15 05:08:06 +00:00
|
|
|
return p.target.Invoke(ctx, req)
|
|
|
|
}
|
|
|
|
|
2019-10-22 07:20:26 +00:00
|
|
|
func (p *monitorProxy) StreamInvoke(
|
2023-03-03 16:36:39 +00:00
|
|
|
req *pulumirpc.ResourceInvokeRequest, server pulumirpc.ResourceMonitor_StreamInvokeServer,
|
|
|
|
) error {
|
2019-10-22 07:20:26 +00:00
|
|
|
client, err := p.target.StreamInvoke(context.Background(), req)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
in, err := client.Recv()
|
|
|
|
if err == io.EOF {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := server.Send(in); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-06-30 14:48:56 +00:00
|
|
|
func (p *monitorProxy) Call(
|
2024-02-08 13:16:23 +00:00
|
|
|
ctx context.Context, req *pulumirpc.ResourceCallRequest,
|
2023-03-03 16:36:39 +00:00
|
|
|
) (*pulumirpc.CallResponse, error) {
|
2021-06-30 14:48:56 +00:00
|
|
|
return p.target.Call(ctx, req)
|
|
|
|
}
|
|
|
|
|
2019-10-15 05:08:06 +00:00
|
|
|
func (p *monitorProxy) ReadResource(
|
2023-03-03 16:36:39 +00:00
|
|
|
ctx context.Context, req *pulumirpc.ReadResourceRequest,
|
|
|
|
) (*pulumirpc.ReadResourceResponse, error) {
|
2019-10-15 05:08:06 +00:00
|
|
|
return p.target.ReadResource(ctx, req)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (p *monitorProxy) RegisterResource(
|
2023-03-03 16:36:39 +00:00
|
|
|
ctx context.Context, req *pulumirpc.RegisterResourceRequest,
|
|
|
|
) (*pulumirpc.RegisterResourceResponse, error) {
|
Maintain alias compat for older Node.js SDKs on new CLIs
This change updates the engine to detect if a `RegisterResource` request
is coming from an older Node.js SDK that is using incorrect alias specs
and, if so, transforms the aliases to be correct. This allows us to
maintain compatibility for users who have upgraded their CLI but are
still using an older version of the Node.js SDK with incorrect alias
specs.
We detect if the request is from a Node.js SDK by looking at the gRPC
request's metadata headers, specifically looking at the "pulumi-runtime"
and "user-agent" headers.
First, if the request has a "pulumi-runtime" header with a value of
"nodejs", we know it's coming from the Node.js plugin. The Node.js
language plugin proxies gRPC calls from the Node.js SDK to the resource
monitor and the proxy now sets the "pulumi-runtime" header to "nodejs"
for `RegisterResource` calls.
Second, if the request has a "user-agent" header that starts with
"grpc-node-js/", we know it's coming from the Node.js SDK. This is the
case for inline programs in the automation API, which connects directly
to the resource monitor, rather than going through the language plugin's
proxy.
We can't just look at "user-agent", because in the proxy case it will
have a Go-specific "user-agent".
Updated Node.js SDKs set a new `aliasSpecs` field on the
`RegisterResource` request, which indicates that the alias specs are
correct, and no transforms are needed.
2023-05-31 22:37:59 +00:00
|
|
|
// Add the "pulumi-runtime" header to the context so the engine can detect this request
|
|
|
|
// is coming from the nodejs language plugin.
|
|
|
|
// Setting this header is not required for other SDKs. We do it for nodejs so the engine
|
|
|
|
// can workaround a bug in older Node.js SDKs where alias specs weren't specified correctly.
|
|
|
|
ctx = metadata.AppendToOutgoingContext(ctx, "pulumi-runtime", "nodejs")
|
2019-10-15 05:08:06 +00:00
|
|
|
return p.target.RegisterResource(ctx, req)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (p *monitorProxy) RegisterResourceOutputs(
|
2023-03-03 16:36:39 +00:00
|
|
|
ctx context.Context, req *pulumirpc.RegisterResourceOutputsRequest,
|
2024-01-17 09:35:20 +00:00
|
|
|
) (*emptypb.Empty, error) {
|
2019-10-15 05:08:06 +00:00
|
|
|
return p.target.RegisterResourceOutputs(ctx, req)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (p *monitorProxy) SupportsFeature(
|
2023-03-03 16:36:39 +00:00
|
|
|
ctx context.Context, req *pulumirpc.SupportsFeatureRequest,
|
|
|
|
) (*pulumirpc.SupportsFeatureResponse, error) {
|
2019-10-15 05:08:06 +00:00
|
|
|
return p.target.SupportsFeature(ctx, req)
|
|
|
|
}
|
2024-03-07 08:52:34 +00:00
|
|
|
|
|
|
|
func (p *monitorProxy) RegisterStackTransform(
|
|
|
|
ctx context.Context, req *pulumirpc.Callback,
|
|
|
|
) (*emptypb.Empty, error) {
|
|
|
|
return p.target.RegisterStackTransform(ctx, req)
|
|
|
|
}
|
RegisterProvider engine work (#16241)
<!---
Thanks so much for your contribution! If this is your first time
contributing, please ensure that you have read the
[CONTRIBUTING](https://github.com/pulumi/pulumi/blob/master/CONTRIBUTING.md)
documentation.
-->
# Description
<!--- Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context. -->
This adds support for a `RegisterProvider` method to the engine. This
allows an SDK process to send the information for a package (name,
version, url, etc, and parameter in the future) and get back a UUID for
that run of the engine that can be used to re-lookup that information.
That allows the SDK to just send the `provider` field in
`RegisterResourceRequest` instead of filling in `version`,
`pluginDownloadURL` etc (and importantly not having to fill in
`parameter` for parameterised providers, which could be a large amount
of data).
This doesn't update any of the SDKs to yet use this method. We can do
that piecemeal, but it will require core sdk and codegen changes for
each language.
## Checklist
- [x] I have run `make tidy` to update any new dependencies
- [x] I have run `make lint` to verify my code passes the lint check
- [x] I have formatted my code using `gofumpt`
<!--- Please provide details if the checkbox below is to be left
unchecked. -->
- [x] I have added tests that prove my fix is effective or that my
feature works
<!---
User-facing changes require a CHANGELOG entry.
-->
- [ ] I have run `make changelog` and committed the
`changelog/pending/<file>` documenting my change
<!--
If the change(s) in this PR is a modification of an existing call to the
Pulumi Cloud,
then the service should honor older versions of the CLI where this
change would not exist.
You must then bump the API version in
/pkg/backend/httpstate/client/api.go, as well as add
it to the service.
-->
- [ ] Yes, there are changes in this PR that warrants bumping the Pulumi
Cloud API version
<!-- @Pulumi employees: If yes, you must submit corresponding changes in
the service repo. -->
2024-05-23 06:16:59 +00:00
|
|
|
|
|
|
|
func (p *monitorProxy) RegisterProvider(
|
|
|
|
ctx context.Context, req *pulumirpc.RegisterProviderRequest,
|
|
|
|
) (*pulumirpc.RegisterProviderResponse, error) {
|
|
|
|
return p.target.RegisterProvider(ctx, req)
|
|
|
|
}
|