diff --git a/client.go b/client.go index cd89985a..2beea5f2 100644 --- a/client.go +++ b/client.go @@ -27,6 +27,7 @@ import ( "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-plugin/internal/cmdrunner" + "github.com/hashicorp/go-plugin/internal/grpcmux" "github.com/hashicorp/go-plugin/runner" "google.golang.org/grpc" ) @@ -63,6 +64,13 @@ var ( // ErrSecureConfigAndReattach is returned when both Reattach and // SecureConfig are set. ErrSecureConfigAndReattach = errors.New("only one of Reattach or SecureConfig can be set") + + // ErrGRPCBrokerMuxNotSupported is returned when the client requests + // multiplexing over the gRPC broker, but the plugin does not support the + // feature. In most cases, this should be resolvable by updating and + // rebuilding the plugin, or restarting the plugin with + // ClientConfig.GRPCBrokerMultiplex set to false. + ErrGRPCBrokerMuxNotSupported = errors.New("client requested gRPC broker multiplexing but plugin does not support the feature") ) // Client handles the lifecycle of a plugin application. It launches @@ -102,6 +110,9 @@ type Client struct { processKilled bool unixSocketCfg UnixSocketConfig + + grpcMuxerOnce sync.Once + grpcMuxer *grpcmux.GRPCClientMuxer } // NegotiatedVersion returns the protocol version negotiated with the server. @@ -237,6 +248,19 @@ type ClientConfig struct { // protocol. GRPCDialOptions []grpc.DialOption + // GRPCBrokerMultiplex turns on multiplexing for the gRPC broker. The gRPC + // broker will multiplex all brokered gRPC servers over the plugin's original + // listener socket instead of making a new listener for each server. The + // go-plugin library currently only includes a Go implementation for the + // server (i.e. plugin) side of gRPC broker multiplexing. + // + // Does not support reattaching. + // + // Multiplexed gRPC streams MUST be established sequentially, i.e. after + // calling AcceptAndServe from one side, wait for the other side to Dial + // before calling AcceptAndServe again. + GRPCBrokerMultiplex bool + // SkipHostEnv allows plugins to run without inheriting the parent process' // environment variables. SkipHostEnv bool @@ -352,7 +376,7 @@ func CleanupClients() { wg.Wait() } -// Creates a new plugin client which manages the lifecycle of an external +// NewClient creates a new plugin client which manages the lifecycle of an external // plugin and gets the address for the RPC connection. // // The client must be cleaned up at some point by calling Kill(). If @@ -374,10 +398,10 @@ func NewClient(config *ClientConfig) (c *Client) { } if config.SyncStdout == nil { - config.SyncStdout = ioutil.Discard + config.SyncStdout = io.Discard } if config.SyncStderr == nil { - config.SyncStderr = ioutil.Discard + config.SyncStderr = io.Discard } if config.AllowedProtocols == nil { @@ -572,6 +596,10 @@ func (c *Client) Start() (addr net.Addr, err error) { if c.config.SecureConfig != nil && c.config.Reattach != nil { return nil, ErrSecureConfigAndReattach } + + if c.config.GRPCBrokerMultiplex && c.config.Reattach != nil { + return nil, fmt.Errorf("gRPC broker multiplexing is not supported with Reattach config") + } } if c.config.Reattach != nil { @@ -603,6 +631,9 @@ func (c *Client) Start() (addr net.Addr, err error) { fmt.Sprintf("PLUGIN_MAX_PORT=%d", c.config.MaxPort), fmt.Sprintf("PLUGIN_PROTOCOL_VERSIONS=%s", strings.Join(versionStrings, ",")), } + if c.config.GRPCBrokerMultiplex { + env = append(env, fmt.Sprintf("%s=true", envMultiplexGRPC)) + } cmd := c.config.Cmd if cmd == nil { @@ -790,7 +821,7 @@ func (c *Client) Start() (addr net.Addr, err error) { // Trim the line and split by "|" in order to get the parts of // the output. line = strings.TrimSpace(line) - parts := strings.SplitN(line, "|", 6) + parts := strings.Split(line, "|") if len(parts) < 4 { errText := fmt.Sprintf("Unrecognized remote plugin message: %s", line) if !ok { @@ -878,6 +909,18 @@ func (c *Client) Start() (addr net.Addr, err error) { return nil, fmt.Errorf("error parsing server cert: %s", err) } } + + if c.config.GRPCBrokerMultiplex && c.protocol == ProtocolGRPC { + if len(parts) <= 6 { + return nil, fmt.Errorf("%w; for Go plugins, you will need to update the "+ + "github.com/hashicorp/go-plugin dependency and recompile", ErrGRPCBrokerMuxNotSupported) + } + if muxSupported, err := strconv.ParseBool(parts[6]); err != nil { + return nil, fmt.Errorf("error parsing %q as a boolean for gRPC broker multiplexing support", parts[6]) + } else if !muxSupported { + return nil, ErrGRPCBrokerMuxNotSupported + } + } } c.address = addr @@ -951,12 +994,11 @@ func (c *Client) reattach() (net.Addr, error) { if c.config.Reattach.Test { c.negotiatedVersion = c.config.Reattach.ProtocolVersion - } - - // If we're in test mode, we do NOT set the process. This avoids the - // process being killed (the only purpose we have for c.process), since - // in test mode the process is responsible for exiting on its own. - if !c.config.Reattach.Test { + } else { + // If we're in test mode, we do NOT set the runner. This avoids the + // runner being killed (the only purpose we have for setting c.runner + // when reattaching), since in test mode the process is responsible for + // exiting on its own. c.runner = r } @@ -1061,11 +1103,24 @@ func netAddrDialer(addr net.Addr) func(string, time.Duration) (net.Conn, error) // dialer is compatible with grpc.WithDialer and creates the connection // to the plugin. func (c *Client) dialer(_ string, timeout time.Duration) (net.Conn, error) { - conn, err := netAddrDialer(c.address)("", timeout) + muxer, err := c.getGRPCMuxer(c.address) if err != nil { return nil, err } + var conn net.Conn + if muxer.Enabled() { + conn, err = muxer.Dial() + if err != nil { + return nil, err + } + } else { + conn, err = netAddrDialer(c.address)("", timeout) + if err != nil { + return nil, err + } + } + // If we have a TLS config we wrap our connection. We only do this // for net/rpc since gRPC uses its own mechanism for TLS. if c.protocol == ProtocolNetRPC && c.config.TLSConfig != nil { @@ -1075,6 +1130,22 @@ func (c *Client) dialer(_ string, timeout time.Duration) (net.Conn, error) { return conn, nil } +func (c *Client) getGRPCMuxer(addr net.Addr) (*grpcmux.GRPCClientMuxer, error) { + if c.protocol != ProtocolGRPC || !c.config.GRPCBrokerMultiplex { + return nil, nil + } + + var err error + c.grpcMuxerOnce.Do(func() { + c.grpcMuxer, err = grpcmux.NewGRPCClientMuxer(c.logger, addr) + }) + if err != nil { + return nil, err + } + + return c.grpcMuxer, nil +} + var stdErrBufferSize = 64 * 1024 func (c *Client) logStderr(name string, r io.Reader) { diff --git a/client_test.go b/client_test.go index 682b31e5..11a4b0d9 100644 --- a/client_test.go +++ b/client_test.go @@ -7,9 +7,9 @@ import ( "bytes" "context" "crypto/sha256" + "errors" "fmt" "io" - "io/ioutil" "log" "net" "os" @@ -65,10 +65,7 @@ func TestClient(t *testing.T) { // This tests a bug where Kill would start func TestClient_killStart(t *testing.T) { // Create a temporary dir to store the result file - td, err := ioutil.TempDir("", "plugin") - if err != nil { - t.Fatalf("err: %s", err) - } + td := t.TempDir() defer os.RemoveAll(td) // Start the client @@ -115,10 +112,7 @@ func TestClient_killStart(t *testing.T) { func TestClient_testCleanup(t *testing.T) { // Create a temporary dir to store the result file - td, err := ioutil.TempDir("", "plugin") - if err != nil { - t.Fatalf("err: %s", err) - } + td := t.TempDir() defer os.RemoveAll(td) // Create a path that the helper process will write on cleanup @@ -825,7 +819,7 @@ func TestClient_textLogLevel(t *testing.T) { func TestClient_Stdin(t *testing.T) { // Overwrite stdin for this test with a temporary file - tf, err := ioutil.TempFile("", "terraform") + tf, err := os.CreateTemp("", "terraform") if err != nil { t.Fatalf("err: %s", err) } @@ -914,6 +908,34 @@ func TestClient_SkipHostEnv(t *testing.T) { } } +func TestClient_RequestGRPCMultiplexing_UnsupportedByPlugin(t *testing.T) { + for _, name := range []string{ + "mux-grpc-with-old-plugin", + "mux-grpc-with-unsupported-plugin", + } { + t.Run(name, func(t *testing.T) { + process := helperProcess(name) + c := NewClient(&ClientConfig{ + Cmd: process, + HandshakeConfig: testHandshake, + Plugins: testGRPCPluginMap, + AllowedProtocols: []Protocol{ProtocolGRPC}, + GRPCBrokerMultiplex: true, + }) + defer c.Kill() + + _, err := c.Start() + if err == nil { + t.Fatal("expected error") + } + + if !errors.Is(err, ErrGRPCBrokerMuxNotSupported) { + t.Fatalf("expected %s, but got %s", ErrGRPCBrokerMuxNotSupported, err) + } + }) + } +} + func TestClient_SecureConfig(t *testing.T) { // Test failure case secureConfig := &SecureConfig{ diff --git a/constants.go b/constants.go index 32e58602..e7f5bbe5 100644 --- a/constants.go +++ b/constants.go @@ -11,4 +11,6 @@ const ( // EnvUnixSocketGroup specifies the owning, writable group to set for Unix // sockets created by _plugins_. Does not affect client behavior. EnvUnixSocketGroup = "PLUGIN_UNIX_SOCKET_GROUP" + + envMultiplexGRPC = "PLUGIN_MULTIPLEX_GRPC" ) diff --git a/go.mod b/go.mod index 8939f2b4..02c74a00 100644 --- a/go.mod +++ b/go.mod @@ -5,11 +5,12 @@ go 1.17 require ( github.com/golang/protobuf v1.5.0 github.com/hashicorp/go-hclog v0.14.1 - github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb + github.com/hashicorp/yamux v0.1.1 github.com/jhump/protoreflect v1.15.1 github.com/mitchellh/go-testing-interface v0.0.0-20171004221916-a61a99592b77 github.com/oklog/run v1.0.0 google.golang.org/grpc v1.38.0 + google.golang.org/protobuf v1.28.2-0.20230222093303-bc1253ad3743 ) require ( @@ -22,5 +23,4 @@ require ( golang.org/x/sys v0.13.0 // indirect golang.org/x/text v0.13.0 // indirect google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect - google.golang.org/protobuf v1.28.2-0.20230222093303-bc1253ad3743 // indirect ) diff --git a/go.sum b/go.sum index 8c306a5f..93f8021e 100644 --- a/go.sum +++ b/go.sum @@ -34,21 +34,13 @@ github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= -github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hashicorp/go-hclog v0.14.1 h1:nQcJDQwIAGnmoUWp8ubocEX40cCml/17YkF6csQLReU= github.com/hashicorp/go-hclog v0.14.1/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39EeqMAyrmvFQ= -github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb h1:b5rjCoWHc7eqmAS4/qyk21ZsHyb6Mxv/jykxvNTkU4M= -github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM= -github.com/jhump/gopoet v0.0.0-20190322174617-17282ff210b3/go.mod h1:me9yfT6IJSlOL3FCfrg+L6yzUEZ+5jW6WHt4Sk+UPUI= -github.com/jhump/gopoet v0.1.0/go.mod h1:me9yfT6IJSlOL3FCfrg+L6yzUEZ+5jW6WHt4Sk+UPUI= -github.com/jhump/goprotoc v0.5.0/go.mod h1:VrbvcYrQOrTi3i0Vf+m+oqQWk9l72mjkJCYo7UvLHRQ= -github.com/jhump/protoreflect v1.11.0/go.mod h1:U7aMIjN0NWq9swDP7xDdoMfRHb35uiuTd3Z9nFXJf5E= +github.com/hashicorp/yamux v0.1.1 h1:yrQxtgseBDrq9Y652vSRDvsKCJKOUD+GzTS4Y0Y8pvE= +github.com/hashicorp/yamux v0.1.1/go.mod h1:CtWFDAQgb7dxtzFs4tWbplKIe2jSi3+5vKbgIO0SLnQ= github.com/jhump/protoreflect v1.15.1 h1:HUMERORf3I3ZdX05WaQ6MIpd/NJ434hTp5YiKgfCL6c= github.com/jhump/protoreflect v1.15.1/go.mod h1:jD/2GMKKE6OqX8qTjhADU1e6DShO+gavG9e0Q693nKo= -github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= -github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= -github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/mattn/go-colorable v0.1.4 h1:snbPLB8fVfU9iwbbo30TPtbLRzwWu6aJS6Xh4eaaviA= github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE= github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s= @@ -62,85 +54,39 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= -github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= -github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= -github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= -github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= -github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= -golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= -golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= -golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= -golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= -golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= -golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= -golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= -golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM= golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4 h1:uVc8UZUe6tr40fFVnUP5Oj+veunVezqYl9z7DYw9xzw= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= -golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= -golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE= golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= -golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= -golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo= -golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= -golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= -golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= -golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= -golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k= golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= -golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= -golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= -golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= -golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= -golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM= google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= @@ -164,15 +110,10 @@ google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= -google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.28.2-0.20230222093303-bc1253ad3743 h1:yqElulDvOF26oZ2O+2/aoX7mQ8DY/6+p39neytrycd8= google.golang.org/protobuf v1.28.2-0.20230222093303-bc1253ad3743/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= diff --git a/grpc_broker.go b/grpc_broker.go index b86561a0..5b17e37f 100644 --- a/grpc_broker.go +++ b/grpc_broker.go @@ -14,6 +14,7 @@ import ( "sync/atomic" "time" + "github.com/hashicorp/go-plugin/internal/grpcmux" "github.com/hashicorp/go-plugin/internal/plugin" "github.com/hashicorp/go-plugin/runner" @@ -40,6 +41,8 @@ type sendErr struct { // connection information to/from the plugin. Implements GRPCBrokerServer and // streamer interfaces. type gRPCBrokerServer struct { + plugin.UnimplementedGRPCBrokerServer + // send is used to send connection info to the gRPC stream. send chan *sendErr @@ -263,29 +266,39 @@ func (s *gRPCBrokerClientImpl) Close() { type GRPCBroker struct { nextId uint32 streamer streamer - streams map[uint32]*gRPCBrokerPending tls *tls.Config doneCh chan struct{} o sync.Once + clientStreams map[uint32]*gRPCBrokerPending + serverStreams map[uint32]*gRPCBrokerPending + unixSocketCfg UnixSocketConfig addrTranslator runner.AddrTranslator + dialMutex sync.Mutex + + muxer grpcmux.GRPCMuxer + sync.Mutex } type gRPCBrokerPending struct { ch chan *plugin.ConnInfo doneCh chan struct{} + once sync.Once } -func newGRPCBroker(s streamer, tls *tls.Config, unixSocketCfg UnixSocketConfig, addrTranslator runner.AddrTranslator) *GRPCBroker { +func newGRPCBroker(s streamer, tls *tls.Config, unixSocketCfg UnixSocketConfig, addrTranslator runner.AddrTranslator, muxer grpcmux.GRPCMuxer) *GRPCBroker { return &GRPCBroker{ streamer: s, - streams: make(map[uint32]*gRPCBrokerPending), tls: tls, doneCh: make(chan struct{}), + clientStreams: make(map[uint32]*gRPCBrokerPending), + serverStreams: make(map[uint32]*gRPCBrokerPending), + muxer: muxer, + unixSocketCfg: unixSocketCfg, addrTranslator: addrTranslator, } @@ -295,6 +308,42 @@ func newGRPCBroker(s streamer, tls *tls.Config, unixSocketCfg UnixSocketConfig, // // This should not be called multiple times with the same ID at one time. func (b *GRPCBroker) Accept(id uint32) (net.Listener, error) { + if b.muxer.Enabled() { + p := b.getServerStream(id) + go func() { + err := b.listenForKnocks(id) + if err != nil { + log.Printf("[ERR]: error listening for knocks, id: %d, error: %s", id, err) + } + }() + + ln, err := b.muxer.Listener(id, p.doneCh) + if err != nil { + return nil, err + } + + ln = &rmListener{ + Listener: ln, + close: func() error { + // We could have multiple listeners on the same ID, so use sync.Once + // for closing doneCh to ensure we don't get a panic. + p.once.Do(func() { + close(p.doneCh) + }) + + b.Lock() + defer b.Unlock() + + // No longer need to listen for knocks once the listener is closed. + delete(b.serverStreams, id) + + return nil + }, + } + + return ln, nil + } + listener, err := serverListener(b.unixSocketCfg) if err != nil { return nil, err @@ -327,20 +376,20 @@ func (b *GRPCBroker) Accept(id uint32) (net.Listener, error) { // connection is opened every call, these calls should be used sparingly. // Multiple gRPC server implementations can be registered to a single // AcceptAndServe call. -func (b *GRPCBroker) AcceptAndServe(id uint32, s func([]grpc.ServerOption) *grpc.Server) { - listener, err := b.Accept(id) +func (b *GRPCBroker) AcceptAndServe(id uint32, newGRPCServer func([]grpc.ServerOption) *grpc.Server) { + ln, err := b.Accept(id) if err != nil { log.Printf("[ERR] plugin: plugin acceptAndServe error: %s", err) return } - defer listener.Close() + defer ln.Close() var opts []grpc.ServerOption if b.tls != nil { opts = []grpc.ServerOption{grpc.Creds(credentials.NewTLS(b.tls))} } - server := s(opts) + server := newGRPCServer(opts) // Here we use a run group to close this goroutine if the server is shutdown // or the broker is shutdown. @@ -348,7 +397,7 @@ func (b *GRPCBroker) AcceptAndServe(id uint32, s func([]grpc.ServerOption) *grpc { // Serve on the listener, if shutting down call GracefulStop. g.Add(func() error { - return server.Serve(listener) + return server.Serve(ln) }, func(err error) { server.GracefulStop() }) @@ -381,12 +430,108 @@ func (b *GRPCBroker) Close() error { return nil } +func (b *GRPCBroker) listenForKnocks(id uint32) error { + p := b.getServerStream(id) + for { + select { + case msg := <-p.ch: + // Shouldn't be possible. + if msg.ServiceId != id { + return fmt.Errorf("knock received with wrong service ID; expected %d but got %d", id, msg.ServiceId) + } + + // Also shouldn't be possible. + if msg.Knock == nil || !msg.Knock.Knock || msg.Knock.Ack { + return fmt.Errorf("knock received for service ID %d with incorrect values; knock=%+v", id, msg.Knock) + } + + // Successful knock, open the door for the given ID. + var ackError string + err := b.muxer.AcceptKnock(id) + if err != nil { + ackError = err.Error() + } + + // Send back an acknowledgement to allow the client to start dialling. + err = b.streamer.Send(&plugin.ConnInfo{ + ServiceId: id, + Knock: &plugin.ConnInfo_Knock{ + Knock: true, + Ack: true, + Error: ackError, + }, + }) + if err != nil { + return fmt.Errorf("error sending back knock acknowledgement: %w", err) + } + case <-p.doneCh: + return nil + } + } +} + +func (b *GRPCBroker) knock(id uint32) error { + // Send a knock. + err := b.streamer.Send(&plugin.ConnInfo{ + ServiceId: id, + Knock: &plugin.ConnInfo_Knock{ + Knock: true, + }, + }) + if err != nil { + return err + } + + // Wait for the ack. + p := b.getClientStream(id) + select { + case msg := <-p.ch: + if msg.ServiceId != id { + return fmt.Errorf("handshake failed for multiplexing on id %d; got response for %d", id, msg.ServiceId) + } + if msg.Knock == nil || !msg.Knock.Knock || !msg.Knock.Ack { + return fmt.Errorf("handshake failed for multiplexing on id %d; expected knock and ack, but got %+v", id, msg.Knock) + } + if msg.Knock.Error != "" { + return fmt.Errorf("failed to knock for id %d: %s", id, msg.Knock.Error) + } + case <-time.After(5 * time.Second): + return fmt.Errorf("timeout waiting for multiplexing knock handshake on id %d", id) + } + + return nil +} + +func (b *GRPCBroker) muxDial(id uint32) func(string, time.Duration) (net.Conn, error) { + return func(string, time.Duration) (net.Conn, error) { + b.dialMutex.Lock() + defer b.dialMutex.Unlock() + + // Tell the other side the listener ID it should give the next stream to. + err := b.knock(id) + if err != nil { + return nil, fmt.Errorf("failed to knock before dialling client: %w", err) + } + + conn, err := b.muxer.Dial() + if err != nil { + return nil, err + } + + return conn, nil + } +} + // Dial opens a connection by ID. func (b *GRPCBroker) Dial(id uint32) (conn *grpc.ClientConn, err error) { + if b.muxer.Enabled() { + return dialGRPCConn(b.tls, b.muxDial(id)) + } + var c *plugin.ConnInfo // Open the stream - p := b.getStream(id) + p := b.getClientStream(id) select { case c = <-p.ch: close(p.doneCh) @@ -434,37 +579,63 @@ func (m *GRPCBroker) NextId() uint32 { // the plugin host/client. func (m *GRPCBroker) Run() { for { - stream, err := m.streamer.Recv() + msg, err := m.streamer.Recv() if err != nil { // Once we receive an error, just exit break } // Initialize the waiter - p := m.getStream(stream.ServiceId) + var p *gRPCBrokerPending + if msg.Knock != nil && msg.Knock.Knock && !msg.Knock.Ack { + p = m.getServerStream(msg.ServiceId) + // The server side doesn't close the channel immediately as it needs + // to continuously listen for knocks. + } else { + p = m.getClientStream(msg.ServiceId) + go m.timeoutWait(msg.ServiceId, p) + } select { - case p.ch <- stream: + case p.ch <- msg: default: } + } +} - go m.timeoutWait(stream.ServiceId, p) +// getClientStream is a buffer to receive new connection info and knock acks +// by stream ID. +func (m *GRPCBroker) getClientStream(id uint32) *gRPCBrokerPending { + m.Lock() + defer m.Unlock() + + p, ok := m.clientStreams[id] + if ok { + return p + } + + m.clientStreams[id] = &gRPCBrokerPending{ + ch: make(chan *plugin.ConnInfo, 1), + doneCh: make(chan struct{}), } + return m.clientStreams[id] } -func (m *GRPCBroker) getStream(id uint32) *gRPCBrokerPending { +// getServerStream is a buffer to receive knocks to a multiplexed stream ID +// that its side is listening on. Not used unless multiplexing is enabled. +func (m *GRPCBroker) getServerStream(id uint32) *gRPCBrokerPending { m.Lock() defer m.Unlock() - p, ok := m.streams[id] + p, ok := m.serverStreams[id] if ok { return p } - m.streams[id] = &gRPCBrokerPending{ + m.serverStreams[id] = &gRPCBrokerPending{ ch: make(chan *plugin.ConnInfo, 1), doneCh: make(chan struct{}), } - return m.streams[id] + return m.serverStreams[id] } func (m *GRPCBroker) timeoutWait(id uint32, p *gRPCBrokerPending) { @@ -479,5 +650,5 @@ func (m *GRPCBroker) timeoutWait(id uint32, p *gRPCBrokerPending) { defer m.Unlock() // Delete the stream so no one else can grab it - delete(m.streams, id) + delete(m.clientStreams, id) } diff --git a/grpc_client.go b/grpc_client.go index 583e4250..627649d8 100644 --- a/grpc_client.go +++ b/grpc_client.go @@ -61,9 +61,14 @@ func newGRPCClient(doneCtx context.Context, c *Client) (*GRPCClient, error) { return nil, err } + muxer, err := c.getGRPCMuxer(c.address) + if err != nil { + return nil, err + } + // Start the broker. brokerGRPCClient := newGRPCBrokerClient(conn) - broker := newGRPCBroker(brokerGRPCClient, c.config.TLSConfig, c.unixSocketCfg, c.runner) + broker := newGRPCBroker(brokerGRPCClient, c.config.TLSConfig, c.unixSocketCfg, c.runner, muxer) go broker.Run() go brokerGRPCClient.StartStream() diff --git a/grpc_client_test.go b/grpc_client_test.go index b58969e8..a7800a5a 100644 --- a/grpc_client_test.go +++ b/grpc_client_test.go @@ -14,8 +14,17 @@ import ( reflectpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha" ) -func TestGRPCClient_App(t *testing.T) { - client, server := TestPluginGRPCConn(t, map[string]Plugin{ +func TestGRPC_App(t *testing.T) { + t.Run("default", func(t *testing.T) { + testGRPCClientApp(t, false) + }) + t.Run("mux", func(t *testing.T) { + testGRPCClientApp(t, true) + }) +} + +func testGRPCClientApp(t *testing.T, multiplex bool) { + client, server := TestPluginGRPCConn(t, multiplex, map[string]Plugin{ "test": new(testGRPCInterfacePlugin), }) defer client.Close() @@ -59,7 +68,16 @@ func TestGRPCConn_BidirectionalPing(t *testing.T) { } func TestGRPCC_Stream(t *testing.T) { - client, server := TestPluginGRPCConn(t, map[string]Plugin{ + t.Run("default", func(t *testing.T) { + testGRPCStream(t, false) + }) + t.Run("mux", func(t *testing.T) { + testGRPCStream(t, true) + }) +} + +func testGRPCStream(t *testing.T, multiplex bool) { + client, server := TestPluginGRPCConn(t, multiplex, map[string]Plugin{ "test": new(testGRPCInterfacePlugin), }) defer client.Close() @@ -86,8 +104,17 @@ func TestGRPCC_Stream(t *testing.T) { } } -func TestGRPCClient_Ping(t *testing.T) { - client, server := TestPluginGRPCConn(t, map[string]Plugin{ +func TestGRPC_Ping(t *testing.T) { + t.Run("default", func(t *testing.T) { + testGRPCClientPing(t, false) + }) + t.Run("mux", func(t *testing.T) { + testGRPCClientPing(t, true) + }) +} + +func testGRPCClientPing(t *testing.T, multiplex bool) { + client, server := TestPluginGRPCConn(t, multiplex, map[string]Plugin{ "test": new(testGRPCInterfacePlugin), }) defer client.Close() @@ -110,10 +137,19 @@ func TestGRPCClient_Ping(t *testing.T) { } } -func TestGRPCClient_Reflection(t *testing.T) { +func TestGRPC_Reflection(t *testing.T) { + t.Run("default", func(t *testing.T) { + testGRPCClientReflection(t, false) + }) + t.Run("mux", func(t *testing.T) { + testGRPCClientReflection(t, true) + }) +} + +func testGRPCClientReflection(t *testing.T, multiplex bool) { ctx := context.Background() - client, server := TestPluginGRPCConn(t, map[string]Plugin{ + client, server := TestPluginGRPCConn(t, multiplex, map[string]Plugin{ "test": new(testGRPCInterfacePlugin), }) defer client.Close() diff --git a/grpc_server.go b/grpc_server.go index 369f958a..a5f40c7f 100644 --- a/grpc_server.go +++ b/grpc_server.go @@ -12,6 +12,7 @@ import ( "net" hclog "github.com/hashicorp/go-hclog" + "github.com/hashicorp/go-plugin/internal/grpcmux" "github.com/hashicorp/go-plugin/internal/plugin" "google.golang.org/grpc" "google.golang.org/grpc/credentials" @@ -61,6 +62,8 @@ type GRPCServer struct { stdioServer *grpcStdioServer logger hclog.Logger + + muxer *grpcmux.GRPCServerMuxer } // ServerProtocol impl. @@ -84,7 +87,7 @@ func (s *GRPCServer) Init() error { // Register the broker service brokerServer := newGRPCBrokerServer() plugin.RegisterGRPCBrokerServer(s.server, brokerServer) - s.broker = newGRPCBroker(brokerServer, s.TLS, unixSocketConfigFromEnv(), nil) + s.broker = newGRPCBroker(brokerServer, s.TLS, unixSocketConfigFromEnv(), nil, s.muxer) go s.broker.Run() // Register the controller diff --git a/internal/cmdrunner/testdata/go.mod b/internal/cmdrunner/testdata/go.mod index 6d622066..0312062f 100644 --- a/internal/cmdrunner/testdata/go.mod +++ b/internal/cmdrunner/testdata/go.mod @@ -10,14 +10,14 @@ require ( github.com/fatih/color v1.7.0 // indirect github.com/golang/protobuf v1.5.0 // indirect github.com/hashicorp/go-hclog v0.14.1 // indirect - github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb // indirect + github.com/hashicorp/yamux v0.1.1 // indirect github.com/mattn/go-colorable v0.1.4 // indirect github.com/mattn/go-isatty v0.0.10 // indirect github.com/mitchellh/go-testing-interface v0.0.0-20171004221916-a61a99592b77 // indirect github.com/oklog/run v1.0.0 // indirect - golang.org/x/net v0.7.0 // indirect - golang.org/x/sys v0.5.0 // indirect - golang.org/x/text v0.7.0 // indirect + golang.org/x/net v0.17.0 // indirect + golang.org/x/sys v0.13.0 // indirect + golang.org/x/text v0.13.0 // indirect google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect google.golang.org/grpc v1.38.0 // indirect google.golang.org/protobuf v1.28.2-0.20230222093303-bc1253ad3743 // indirect diff --git a/internal/cmdrunner/testdata/go.sum b/internal/cmdrunner/testdata/go.sum index 5561c146..d9a3dfe1 100644 --- a/internal/cmdrunner/testdata/go.sum +++ b/internal/cmdrunner/testdata/go.sum @@ -36,8 +36,8 @@ github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hashicorp/go-hclog v0.14.1 h1:nQcJDQwIAGnmoUWp8ubocEX40cCml/17YkF6csQLReU= github.com/hashicorp/go-hclog v0.14.1/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39EeqMAyrmvFQ= -github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb h1:b5rjCoWHc7eqmAS4/qyk21ZsHyb6Mxv/jykxvNTkU4M= -github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM= +github.com/hashicorp/yamux v0.1.1 h1:yrQxtgseBDrq9Y652vSRDvsKCJKOUD+GzTS4Y0Y8pvE= +github.com/hashicorp/yamux v0.1.1/go.mod h1:CtWFDAQgb7dxtzFs4tWbplKIe2jSi3+5vKbgIO0SLnQ= github.com/jhump/protoreflect v1.15.1 h1:HUMERORf3I3ZdX05WaQ6MIpd/NJ434hTp5YiKgfCL6c= github.com/mattn/go-colorable v0.1.4 h1:snbPLB8fVfU9iwbbo30TPtbLRzwWu6aJS6Xh4eaaviA= github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE= @@ -64,8 +64,8 @@ golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73r golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g= -golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM= +golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -74,11 +74,11 @@ golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5h golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU= -golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE= +golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= -golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo= -golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k= +golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= diff --git a/internal/grpcmux/blocked_client_listener.go b/internal/grpcmux/blocked_client_listener.go new file mode 100644 index 00000000..e8a3a152 --- /dev/null +++ b/internal/grpcmux/blocked_client_listener.go @@ -0,0 +1,51 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package grpcmux + +import ( + "io" + "net" + + "github.com/hashicorp/yamux" +) + +var _ net.Listener = (*blockedClientListener)(nil) + +// blockedClientListener accepts connections for a specific gRPC broker stream +// ID on the client (host) side of the connection. +type blockedClientListener struct { + session *yamux.Session + waitCh chan struct{} + doneCh <-chan struct{} +} + +func newBlockedClientListener(session *yamux.Session, doneCh <-chan struct{}) *blockedClientListener { + return &blockedClientListener{ + waitCh: make(chan struct{}, 1), + doneCh: doneCh, + session: session, + } +} + +func (b *blockedClientListener) Accept() (net.Conn, error) { + select { + case <-b.waitCh: + return b.session.Accept() + case <-b.doneCh: + return nil, io.EOF + } +} + +func (b *blockedClientListener) Addr() net.Addr { + return b.session.Addr() +} + +func (b *blockedClientListener) Close() error { + // We don't close the session, the client muxer is responsible for that. + return nil +} + +func (b *blockedClientListener) unblock() { + b.waitCh <- struct{}{} +} diff --git a/internal/grpcmux/blocked_server_listener.go b/internal/grpcmux/blocked_server_listener.go new file mode 100644 index 00000000..0edb2c05 --- /dev/null +++ b/internal/grpcmux/blocked_server_listener.go @@ -0,0 +1,49 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package grpcmux + +import ( + "io" + "net" +) + +var _ net.Listener = (*blockedServerListener)(nil) + +// blockedServerListener accepts connections for a specific gRPC broker stream +// ID on the server (plugin) side of the connection. +type blockedServerListener struct { + addr net.Addr + acceptCh chan acceptResult + doneCh <-chan struct{} +} + +type acceptResult struct { + conn net.Conn + err error +} + +func newBlockedServerListener(addr net.Addr, doneCh <-chan struct{}) *blockedServerListener { + return &blockedServerListener{ + addr: addr, + acceptCh: make(chan acceptResult), + doneCh: doneCh, + } +} + +func (b *blockedServerListener) Accept() (net.Conn, error) { + select { + case accept := <-b.acceptCh: + return accept.conn, accept.err + case <-b.doneCh: + return nil, io.EOF + } +} + +func (b *blockedServerListener) Addr() net.Addr { + return b.addr +} + +func (b *blockedServerListener) Close() error { + return nil +} diff --git a/internal/grpcmux/grpc_client_muxer.go b/internal/grpcmux/grpc_client_muxer.go new file mode 100644 index 00000000..b203ba46 --- /dev/null +++ b/internal/grpcmux/grpc_client_muxer.go @@ -0,0 +1,105 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package grpcmux + +import ( + "fmt" + "net" + "sync" + + "github.com/hashicorp/go-hclog" + "github.com/hashicorp/yamux" +) + +var _ GRPCMuxer = (*GRPCClientMuxer)(nil) + +// GRPCClientMuxer implements the client (host) side of the gRPC broker's +// GRPCMuxer interface for multiplexing multiple gRPC broker connections over +// a single net.Conn. +// +// The client dials the initial net.Conn eagerly, and creates a yamux.Session +// as the implementation for multiplexing any additional connections. +// +// Each net.Listener returned from Listener will block until the client receives +// a knock that matches its gRPC broker stream ID. There is no default listener +// on the client, as it is a client for the gRPC broker's control services. (See +// GRPCServerMuxer for more details). +type GRPCClientMuxer struct { + logger hclog.Logger + session *yamux.Session + + acceptMutex sync.Mutex + acceptListeners map[uint32]*blockedClientListener +} + +func NewGRPCClientMuxer(logger hclog.Logger, addr net.Addr) (*GRPCClientMuxer, error) { + // Eagerly establish the underlying connection as early as possible. + logger.Debug("making new client mux initial connection", "addr", addr) + conn, err := net.Dial(addr.Network(), addr.String()) + if err != nil { + return nil, err + } + if tcpConn, ok := conn.(*net.TCPConn); ok { + // Make sure to set keep alive so that the connection doesn't die + _ = tcpConn.SetKeepAlive(true) + } + + cfg := yamux.DefaultConfig() + cfg.Logger = logger.Named("yamux").StandardLogger(&hclog.StandardLoggerOptions{ + InferLevels: true, + }) + cfg.LogOutput = nil + sess, err := yamux.Client(conn, cfg) + if err != nil { + return nil, err + } + + logger.Debug("client muxer connected", "addr", addr) + m := &GRPCClientMuxer{ + logger: logger, + session: sess, + acceptListeners: make(map[uint32]*blockedClientListener), + } + + return m, nil +} + +func (m *GRPCClientMuxer) Enabled() bool { + return m != nil +} + +func (m *GRPCClientMuxer) Listener(id uint32, doneCh <-chan struct{}) (net.Listener, error) { + ln := newBlockedClientListener(m.session, doneCh) + + m.acceptMutex.Lock() + m.acceptListeners[id] = ln + m.acceptMutex.Unlock() + + return ln, nil +} + +func (m *GRPCClientMuxer) AcceptKnock(id uint32) error { + m.acceptMutex.Lock() + defer m.acceptMutex.Unlock() + + ln, ok := m.acceptListeners[id] + if !ok { + return fmt.Errorf("no listener for id %d", id) + } + ln.unblock() + return nil +} + +func (m *GRPCClientMuxer) Dial() (net.Conn, error) { + stream, err := m.session.Open() + if err != nil { + return nil, fmt.Errorf("error dialling new client stream: %w", err) + } + + return stream, nil +} + +func (m *GRPCClientMuxer) Close() error { + return m.session.Close() +} diff --git a/internal/grpcmux/grpc_muxer.go b/internal/grpcmux/grpc_muxer.go new file mode 100644 index 00000000..c52aaf55 --- /dev/null +++ b/internal/grpcmux/grpc_muxer.go @@ -0,0 +1,41 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package grpcmux + +import ( + "net" +) + +// GRPCMuxer enables multiple implementations of net.Listener to accept +// connections over a single "main" multiplexed net.Conn, and dial multiple +// client connections over the same multiplexed net.Conn. +// +// The first multiplexed connection is used to serve the gRPC broker's own +// control services: plugin.GRPCBroker, plugin.GRPCController, plugin.GRPCStdio. +// +// Clients must "knock" before dialling, to tell the server side that the +// next net.Conn should be accepted onto a specific stream ID. The knock is a +// bidirectional streaming message on the plugin.GRPCBroker service. +type GRPCMuxer interface { + // Enabled determines whether multiplexing should be used. It saves users + // of the interface from having to compare an interface with nil, which + // is a bit awkward to do correctly. + Enabled() bool + + // Listener returns a multiplexed listener that will wait until AcceptKnock + // is called with a matching ID before its Accept function returns. + Listener(id uint32, doneCh <-chan struct{}) (net.Listener, error) + + // AcceptKnock unblocks the listener with the matching ID, and returns an + // error if it hasn't been created yet. + AcceptKnock(id uint32) error + + // Dial makes a new multiplexed client connection. To dial a specific ID, + // a knock must be sent first. + Dial() (net.Conn, error) + + // Close closes connections and releases any resources associated with the + // muxer. + Close() error +} diff --git a/internal/grpcmux/grpc_server_muxer.go b/internal/grpcmux/grpc_server_muxer.go new file mode 100644 index 00000000..27696ee7 --- /dev/null +++ b/internal/grpcmux/grpc_server_muxer.go @@ -0,0 +1,190 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package grpcmux + +import ( + "errors" + "fmt" + "net" + "sync" + "time" + + "github.com/hashicorp/go-hclog" + "github.com/hashicorp/yamux" +) + +var _ GRPCMuxer = (*GRPCServerMuxer)(nil) +var _ net.Listener = (*GRPCServerMuxer)(nil) + +// GRPCServerMuxer implements the server (plugin) side of the gRPC broker's +// GRPCMuxer interface for multiplexing multiple gRPC broker connections over +// a single net.Conn. +// +// The server side needs a listener to serve the gRPC broker's control services, +// which includes the service we will receive knocks on. That means we always +// accept the first connection onto a "default" main listener, and if we accept +// any further connections without receiving a knock first, they are also given +// to the default listener. +// +// When creating additional multiplexed listeners for specific stream IDs, we +// can't control the order in which gRPC servers will call Accept() on each +// listener, but we do need to control which gRPC server accepts which connection. +// As such, each multiplexed listener blocks waiting on a channel. It will be +// unblocked when a knock is received for the matching stream ID. +type GRPCServerMuxer struct { + addr net.Addr + logger hclog.Logger + + sessionErrCh chan error + sess *yamux.Session + + knockCh chan uint32 + + acceptMutex sync.Mutex + acceptChannels map[uint32]chan acceptResult +} + +func NewGRPCServerMuxer(logger hclog.Logger, ln net.Listener) *GRPCServerMuxer { + m := &GRPCServerMuxer{ + addr: ln.Addr(), + logger: logger, + + sessionErrCh: make(chan error), + + knockCh: make(chan uint32, 1), + acceptChannels: make(map[uint32]chan acceptResult), + } + + go m.acceptSession(ln) + + return m +} + +// acceptSessionAndMuxAccept is responsible for establishing the yamux session, +// and then kicking off the acceptLoop function. +func (m *GRPCServerMuxer) acceptSession(ln net.Listener) { + defer close(m.sessionErrCh) + + m.logger.Debug("accepting initial connection", "addr", m.addr) + conn, err := ln.Accept() + if err != nil { + m.sessionErrCh <- err + return + } + + m.logger.Debug("initial server connection accepted", "addr", m.addr) + cfg := yamux.DefaultConfig() + cfg.Logger = m.logger.Named("yamux").StandardLogger(&hclog.StandardLoggerOptions{ + InferLevels: true, + }) + cfg.LogOutput = nil + m.sess, err = yamux.Server(conn, cfg) + if err != nil { + m.sessionErrCh <- err + return + } +} + +func (m *GRPCServerMuxer) session() (*yamux.Session, error) { + select { + case err := <-m.sessionErrCh: + if err != nil { + return nil, err + } + case <-time.After(5 * time.Second): + return nil, errors.New("timed out waiting for connection to be established") + } + + // Should never happen. + if m.sess == nil { + return nil, errors.New("no connection established and no error received") + } + + return m.sess, nil +} + +// Accept accepts all incoming connections and routes them to the correct +// stream ID based on the most recent knock received. +func (m *GRPCServerMuxer) Accept() (net.Conn, error) { + session, err := m.session() + if err != nil { + return nil, fmt.Errorf("error establishing yamux session: %w", err) + } + + for { + conn, acceptErr := session.Accept() + + select { + case id := <-m.knockCh: + m.acceptMutex.Lock() + acceptCh, ok := m.acceptChannels[id] + m.acceptMutex.Unlock() + + if !ok { + if conn != nil { + _ = conn.Close() + } + return nil, fmt.Errorf("received knock on ID %d that doesn't have a listener", id) + } + m.logger.Debug("sending conn to brokered listener", "id", id) + acceptCh <- acceptResult{ + conn: conn, + err: acceptErr, + } + default: + m.logger.Debug("sending conn to default listener") + return conn, acceptErr + } + } +} + +func (m *GRPCServerMuxer) Addr() net.Addr { + return m.addr +} + +func (m *GRPCServerMuxer) Close() error { + session, err := m.session() + if err != nil { + return err + } + + return session.Close() +} + +func (m *GRPCServerMuxer) Enabled() bool { + return m != nil +} + +func (m *GRPCServerMuxer) Listener(id uint32, doneCh <-chan struct{}) (net.Listener, error) { + sess, err := m.session() + if err != nil { + return nil, err + } + + ln := newBlockedServerListener(sess.Addr(), doneCh) + m.acceptMutex.Lock() + m.acceptChannels[id] = ln.acceptCh + m.acceptMutex.Unlock() + + return ln, nil +} + +func (m *GRPCServerMuxer) Dial() (net.Conn, error) { + sess, err := m.session() + if err != nil { + return nil, err + } + + stream, err := sess.OpenStream() + if err != nil { + return nil, fmt.Errorf("error dialling new server stream: %w", err) + } + + return stream, nil +} + +func (m *GRPCServerMuxer) AcceptKnock(id uint32) error { + m.knockCh <- id + return nil +} diff --git a/internal/plugin/grpc_broker.pb.go b/internal/plugin/grpc_broker.pb.go index 0514012d..acc6dc9c 100644 --- a/internal/plugin/grpc_broker.pb.go +++ b/internal/plugin/grpc_broker.pb.go @@ -28,9 +28,10 @@ type ConnInfo struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - ServiceId uint32 `protobuf:"varint,1,opt,name=service_id,json=serviceId,proto3" json:"service_id,omitempty"` - Network string `protobuf:"bytes,2,opt,name=network,proto3" json:"network,omitempty"` - Address string `protobuf:"bytes,3,opt,name=address,proto3" json:"address,omitempty"` + ServiceId uint32 `protobuf:"varint,1,opt,name=service_id,json=serviceId,proto3" json:"service_id,omitempty"` + Network string `protobuf:"bytes,2,opt,name=network,proto3" json:"network,omitempty"` + Address string `protobuf:"bytes,3,opt,name=address,proto3" json:"address,omitempty"` + Knock *ConnInfo_Knock `protobuf:"bytes,4,opt,name=knock,proto3" json:"knock,omitempty"` } func (x *ConnInfo) Reset() { @@ -86,24 +87,101 @@ func (x *ConnInfo) GetAddress() string { return "" } +func (x *ConnInfo) GetKnock() *ConnInfo_Knock { + if x != nil { + return x.Knock + } + return nil +} + +type ConnInfo_Knock struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Knock bool `protobuf:"varint,1,opt,name=knock,proto3" json:"knock,omitempty"` + Ack bool `protobuf:"varint,2,opt,name=ack,proto3" json:"ack,omitempty"` + Error string `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"` +} + +func (x *ConnInfo_Knock) Reset() { + *x = ConnInfo_Knock{} + if protoimpl.UnsafeEnabled { + mi := &file_internal_plugin_grpc_broker_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *ConnInfo_Knock) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ConnInfo_Knock) ProtoMessage() {} + +func (x *ConnInfo_Knock) ProtoReflect() protoreflect.Message { + mi := &file_internal_plugin_grpc_broker_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ConnInfo_Knock.ProtoReflect.Descriptor instead. +func (*ConnInfo_Knock) Descriptor() ([]byte, []int) { + return file_internal_plugin_grpc_broker_proto_rawDescGZIP(), []int{0, 0} +} + +func (x *ConnInfo_Knock) GetKnock() bool { + if x != nil { + return x.Knock + } + return false +} + +func (x *ConnInfo_Knock) GetAck() bool { + if x != nil { + return x.Ack + } + return false +} + +func (x *ConnInfo_Knock) GetError() string { + if x != nil { + return x.Error + } + return "" +} + var File_internal_plugin_grpc_broker_proto protoreflect.FileDescriptor var file_internal_plugin_grpc_broker_proto_rawDesc = []byte{ 0x0a, 0x21, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x5f, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x2e, 0x70, 0x72, - 0x6f, 0x74, 0x6f, 0x12, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x22, 0x5d, 0x0a, 0x08, 0x43, - 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x73, 0x65, 0x72, 0x76, 0x69, - 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x09, 0x73, 0x65, 0x72, - 0x76, 0x69, 0x63, 0x65, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, - 0x6b, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, - 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, - 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x32, 0x43, 0x0a, 0x0a, 0x47, 0x52, - 0x50, 0x43, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x12, 0x35, 0x0a, 0x0b, 0x53, 0x74, 0x61, 0x72, - 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x10, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, - 0x2e, 0x43, 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x66, 0x6f, 0x1a, 0x10, 0x2e, 0x70, 0x6c, 0x75, 0x67, - 0x69, 0x6e, 0x2e, 0x43, 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x66, 0x6f, 0x28, 0x01, 0x30, 0x01, 0x42, - 0x0a, 0x5a, 0x08, 0x2e, 0x2f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, - 0x74, 0x6f, 0x33, + 0x6f, 0x74, 0x6f, 0x12, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x22, 0xd2, 0x01, 0x0a, 0x08, + 0x43, 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x73, 0x65, 0x72, 0x76, + 0x69, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x09, 0x73, 0x65, + 0x72, 0x76, 0x69, 0x63, 0x65, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x6e, 0x65, 0x74, 0x77, 0x6f, + 0x72, 0x6b, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, + 0x6b, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x03, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x2c, 0x0a, 0x05, 0x6b, + 0x6e, 0x6f, 0x63, 0x6b, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x70, 0x6c, 0x75, + 0x67, 0x69, 0x6e, 0x2e, 0x43, 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x66, 0x6f, 0x2e, 0x4b, 0x6e, 0x6f, + 0x63, 0x6b, 0x52, 0x05, 0x6b, 0x6e, 0x6f, 0x63, 0x6b, 0x1a, 0x45, 0x0a, 0x05, 0x4b, 0x6e, 0x6f, + 0x63, 0x6b, 0x12, 0x14, 0x0a, 0x05, 0x6b, 0x6e, 0x6f, 0x63, 0x6b, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x08, 0x52, 0x05, 0x6b, 0x6e, 0x6f, 0x63, 0x6b, 0x12, 0x10, 0x0a, 0x03, 0x61, 0x63, 0x6b, 0x18, + 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x03, 0x61, 0x63, 0x6b, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, + 0x72, 0x6f, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, + 0x32, 0x43, 0x0a, 0x0a, 0x47, 0x52, 0x50, 0x43, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x12, 0x35, + 0x0a, 0x0b, 0x53, 0x74, 0x61, 0x72, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x10, 0x2e, + 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x43, 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x66, 0x6f, 0x1a, + 0x10, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x43, 0x6f, 0x6e, 0x6e, 0x49, 0x6e, 0x66, + 0x6f, 0x28, 0x01, 0x30, 0x01, 0x42, 0x0a, 0x5a, 0x08, 0x2e, 0x2f, 0x70, 0x6c, 0x75, 0x67, 0x69, + 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -118,18 +196,20 @@ func file_internal_plugin_grpc_broker_proto_rawDescGZIP() []byte { return file_internal_plugin_grpc_broker_proto_rawDescData } -var file_internal_plugin_grpc_broker_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_internal_plugin_grpc_broker_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_internal_plugin_grpc_broker_proto_goTypes = []interface{}{ - (*ConnInfo)(nil), // 0: plugin.ConnInfo + (*ConnInfo)(nil), // 0: plugin.ConnInfo + (*ConnInfo_Knock)(nil), // 1: plugin.ConnInfo.Knock } var file_internal_plugin_grpc_broker_proto_depIdxs = []int32{ - 0, // 0: plugin.GRPCBroker.StartStream:input_type -> plugin.ConnInfo - 0, // 1: plugin.GRPCBroker.StartStream:output_type -> plugin.ConnInfo - 1, // [1:2] is the sub-list for method output_type - 0, // [0:1] is the sub-list for method input_type - 0, // [0:0] is the sub-list for extension type_name - 0, // [0:0] is the sub-list for extension extendee - 0, // [0:0] is the sub-list for field type_name + 1, // 0: plugin.ConnInfo.knock:type_name -> plugin.ConnInfo.Knock + 0, // 1: plugin.GRPCBroker.StartStream:input_type -> plugin.ConnInfo + 0, // 2: plugin.GRPCBroker.StartStream:output_type -> plugin.ConnInfo + 2, // [2:3] is the sub-list for method output_type + 1, // [1:2] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name } func init() { file_internal_plugin_grpc_broker_proto_init() } @@ -150,6 +230,18 @@ func file_internal_plugin_grpc_broker_proto_init() { return nil } } + file_internal_plugin_grpc_broker_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*ConnInfo_Knock); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } } type x struct{} out := protoimpl.TypeBuilder{ @@ -157,7 +249,7 @@ func file_internal_plugin_grpc_broker_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_internal_plugin_grpc_broker_proto_rawDesc, NumEnums: 0, - NumMessages: 1, + NumMessages: 2, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/plugin/grpc_broker.proto b/internal/plugin/grpc_broker.proto index 195ca16e..c92cd645 100644 --- a/internal/plugin/grpc_broker.proto +++ b/internal/plugin/grpc_broker.proto @@ -9,6 +9,12 @@ message ConnInfo { uint32 service_id = 1; string network = 2; string address = 3; + message Knock { + bool knock = 1; + bool ack = 2; + string error = 3; + } + Knock knock = 4; } service GRPCBroker { diff --git a/plugin_test.go b/plugin_test.go index 45fdf23c..66608f4e 100644 --- a/plugin_test.go +++ b/plugin_test.go @@ -51,6 +51,9 @@ type testStreamer interface { Stream(int32, int32) ([]int32, error) } +// testInterfacePlugin implements the Plugin interface +var _ Plugin = (*testInterfacePlugin)(nil) + // testInterfacePlugin is the implementation of Plugin to create // RPC client/server implementations for testInterface. type testInterfacePlugin struct { @@ -79,6 +82,10 @@ func (p *testInterfacePlugin) impl() testInterface { } } +// testGRPCInterfacePlugin implements both Plugin and GRPCPlugin interfaces +var _ Plugin = (*testGRPCInterfacePlugin)(nil) +var _ GRPCPlugin = (*testGRPCInterfacePlugin)(nil) + // testGRPCInterfacePlugin is a test implementation of the GRPCPlugin interface type testGRPCInterfacePlugin struct { Impl testInterface @@ -116,6 +123,8 @@ func (p *testGRPCInterfacePlugin) impl() testInterface { } // testInterfaceImpl implements testInterface concretely +var _ testInterface = (*testInterfaceImpl)(nil) + type testInterfaceImpl struct { logger hclog.Logger } @@ -241,16 +250,16 @@ func (s *testGRPCServer) PrintKV( func (s *testGRPCServer) Bidirectional(ctx context.Context, req *grpctest.BidirectionalRequest) (*grpctest.BidirectionalResponse, error) { conn, err := s.broker.Dial(req.Id) if err != nil { - return nil, err + return nil, fmt.Errorf("server dial error: %w", err) } pingPongClient := grpctest.NewPingPongClient(conn) resp, err := pingPongClient.Ping(ctx, &grpctest.PingRequest{}) if err != nil { - return nil, err + return nil, fmt.Errorf("server Ping() error: %w", err) } if resp.Msg != "pong" { - return nil, errors.New("Bad PingPong") + return nil, errors.New("server Bad PingPong") } nextID := s.broker.NextId() @@ -352,21 +361,21 @@ func (c *testGRPCClient) Bidirectional() error { Id: nextID, }) if err != nil { - return err + return fmt.Errorf("client Bidirectional() error: %w", err) } conn, err := c.broker.Dial(resp.Id) if err != nil { - return err + return fmt.Errorf("client dial error: %w", err) } pingPongClient := grpctest.NewPingPongClient(conn) pResp, err := pingPongClient.Ping(context.Background(), &grpctest.PingRequest{}) if err != nil { - return err + return fmt.Errorf("client Ping() error: %w", err) } if pResp.Msg != "pong" { - return errors.New("Bad PingPong") + return errors.New("client Bad PingPong") } return nil } @@ -679,6 +688,25 @@ func TestHelperProcess(*testing.T) { } os.Exit(1) + case "mux-grpc-with-old-plugin": + // gRPC broker multiplexing requested, but plugin has no awareness of the + // feature, so just prints 6 segments in the protocol negotiation line. + // Client should fail with a helpful error. + if os.Getenv(envMultiplexGRPC) == "" { + fmt.Println("failed precondition for mux test") + os.Exit(1) + } + fmt.Printf("%d|%d|tcp|:1234|%s|\n", CoreProtocolVersion, testHandshake.ProtocolVersion, ProtocolGRPC) + <-make(chan int) + case "mux-grpc-with-unsupported-plugin": + // gRPC broker multiplexing requested, but plugin explicitly does not + // support it. Client should fail with a helpful error. + if os.Getenv(envMultiplexGRPC) == "" { + fmt.Println("failed precondition for mux test") + os.Exit(1) + } + fmt.Printf("%d|%d|tcp|:1234|%s||false\n", CoreProtocolVersion, testHandshake.ProtocolVersion, ProtocolGRPC) + <-make(chan int) default: fmt.Fprintf(os.Stderr, "Unknown command: %q\n", cmd) os.Exit(2) diff --git a/server.go b/server.go index 1ba5f231..e741bc7f 100644 --- a/server.go +++ b/server.go @@ -21,6 +21,7 @@ import ( "strings" hclog "github.com/hashicorp/go-hclog" + "github.com/hashicorp/go-plugin/internal/grpcmux" "google.golang.org/grpc" ) @@ -387,6 +388,12 @@ func Serve(opts *ServeConfig) { } case ProtocolGRPC: + var muxer *grpcmux.GRPCServerMuxer + if multiplex, _ := strconv.ParseBool(os.Getenv(envMultiplexGRPC)); multiplex { + muxer = grpcmux.NewGRPCServerMuxer(logger, listener) + listener = muxer + } + // Create the gRPC server server = &GRPCServer{ Plugins: pluginSet, @@ -396,6 +403,7 @@ func Serve(opts *ServeConfig) { Stderr: stderr_r, DoneCh: doneCh, logger: logger, + muxer: muxer, } default: @@ -414,13 +422,27 @@ func Serve(opts *ServeConfig) { // bring it up. In test mode, we don't do this because clients will // attach via a reattach config. if opts.Test == nil { - fmt.Printf("%d|%d|%s|%s|%s|%s\n", + const grpcBrokerMultiplexingSupported = true + protocolLine := fmt.Sprintf("%d|%d|%s|%s|%s|%s", CoreProtocolVersion, protoVersion, listener.Addr().Network(), listener.Addr().String(), protoType, serverCert) + + // Old clients will error with new plugins if we blindly append the + // seventh segment for gRPC broker multiplexing support, because old + // client code uses strings.SplitN(line, "|", 6), which means a seventh + // segment will get appended to the sixth segment as "sixthpart|true". + // + // If the environment variable is set, we assume the client is new enough + // to handle a seventh segment, as it should now use + // strings.Split(line, "|") and always handle each segment individually. + if os.Getenv(envMultiplexGRPC) != "" { + protocolLine += fmt.Sprintf("|%v", grpcBrokerMultiplexingSupported) + } + fmt.Printf("%s\n", protocolLine) os.Stdout.Sync() } else if ch := opts.Test.ReattachConfigCh; ch != nil { // Send back the reattach config that can be used. This isn't @@ -585,10 +607,7 @@ func serverListener_unix(unixSocketCfg UnixSocketConfig) (net.Listener, error) { // Wrap the listener in rmListener so that the Unix domain socket file // is removed on close. - return &rmListener{ - Listener: l, - Path: path, - }, nil + return newDeleteFileListener(l, path), nil } func setGroupWritable(path, groupString string, mode os.FileMode) error { @@ -618,11 +637,21 @@ func setGroupWritable(path, groupString string, mode os.FileMode) error { } // rmListener is an implementation of net.Listener that forwards most -// calls to the listener but also removes a file as part of the close. We -// use this to cleanup the unix domain socket on close. +// calls to the listener but also calls an additional close function. We +// use this to cleanup the unix domain socket on close, as well as clean +// up multiplexed listeners. type rmListener struct { net.Listener - Path string + close func() error +} + +func newDeleteFileListener(ln net.Listener, path string) *rmListener { + return &rmListener{ + Listener: ln, + close: func() error { + return os.Remove(path) + }, + } } func (l *rmListener) Close() error { @@ -632,5 +661,5 @@ func (l *rmListener) Close() error { } // Remove the file - return os.Remove(l.Path) + return l.close() } diff --git a/server_test.go b/server_test.go index 68161ecc..24bb3a35 100644 --- a/server_test.go +++ b/server_test.go @@ -208,10 +208,7 @@ func TestRmListener(t *testing.T) { } // Create the listener and test close - rmL := &rmListener{ - Listener: l, - Path: path, - } + rmL := newDeleteFileListener(l, path) if err := rmL.Close(); err != nil { t.Fatalf("err: %s", err) } diff --git a/testing.go b/testing.go index ae48b7a3..a8735dfc 100644 --- a/testing.go +++ b/testing.go @@ -11,7 +11,7 @@ import ( "net/rpc" hclog "github.com/hashicorp/go-hclog" - "github.com/hashicorp/go-plugin/internal/plugin" + "github.com/hashicorp/go-plugin/internal/grpcmux" "github.com/mitchellh/go-testing-interface" "google.golang.org/grpc" ) @@ -135,49 +135,51 @@ func TestGRPCConn(t testing.T, register func(*grpc.Server)) (*grpc.ClientConn, * // TestPluginGRPCConn returns a plugin gRPC client and server that are connected // together and configured. This is used to test gRPC connections. -func TestPluginGRPCConn(t testing.T, ps map[string]Plugin) (*GRPCClient, *GRPCServer) { +func TestPluginGRPCConn(t testing.T, multiplex bool, ps map[string]Plugin) (*GRPCClient, *GRPCServer) { // Create a listener - l, err := net.Listen("tcp", "127.0.0.1:0") + ln, err := serverListener(UnixSocketConfig{}) if err != nil { - t.Fatalf("err: %s", err) + t.Fatal(err) } + logger := hclog.New(&hclog.LoggerOptions{ + Level: hclog.Debug, + }) + // Start up the server + var muxer *grpcmux.GRPCServerMuxer + if multiplex { + muxer = grpcmux.NewGRPCServerMuxer(logger, ln) + ln = muxer + } server := &GRPCServer{ Plugins: ps, DoneCh: make(chan struct{}), Server: DefaultGRPCServer, Stdout: new(bytes.Buffer), Stderr: new(bytes.Buffer), - logger: hclog.Default(), + logger: logger, + muxer: muxer, } if err := server.Init(); err != nil { t.Fatalf("err: %s", err) } - go server.Serve(l) - - // Connect to the server - conn, err := grpc.Dial( - l.Addr().String(), - grpc.WithBlock(), - grpc.WithInsecure()) - if err != nil { - t.Fatalf("err: %s", err) + go server.Serve(ln) + + client := &Client{ + address: ln.Addr(), + protocol: ProtocolGRPC, + config: &ClientConfig{ + Plugins: ps, + GRPCBrokerMultiplex: multiplex, + }, + logger: logger, } - brokerGRPCClient := newGRPCBrokerClient(conn) - broker := newGRPCBroker(brokerGRPCClient, nil, UnixSocketConfig{}, nil) - go broker.Run() - go brokerGRPCClient.StartStream() - - // Create the client - client := &GRPCClient{ - Conn: conn, - Plugins: ps, - broker: broker, - doneCtx: context.Background(), - controller: plugin.NewGRPCControllerClient(conn), + grpcClient, err := newGRPCClient(context.Background(), client) + if err != nil { + t.Fatal(err) } - return client, server + return grpcClient, server }