From ecdb7d7ddb2b0f124e3e50bea7bd10e87836b4be Mon Sep 17 00:00:00 2001 From: Tamal Saha Date: Tue, 31 Mar 2026 18:04:09 +0600 Subject: [PATCH] feat: add remote gRPC plugin support with keepalive Add support for reliable gRPC communication over the network, enabling plugins to run on different machines and communicate over TCP/IP. Key changes: - Add GRPCRemoteClientConfig and NewGRPCRemoteClient for remote connections - Add DefaultGRPCRemoteKeepaliveParams for reliable network transport - Add health check utilities for server availability verification - Add complete working example in examples/remote/ - Update README.md with remote plugin documentation The implementation uses gRPC keepalive pings (10s interval, 20s timeout) to maintain reliable connections over potentially unreliable networks. TLS is supported for secure communication over untrusted networks. Signed-off-by: Tamal Saha --- AGENTS.md | 124 +++++++++++++++++ README.md | 76 +++++++++- examples/remote/README.md | 98 +++++++++++++ examples/remote/client/main.go | 105 ++++++++++++++ examples/remote/server/main.go | 82 +++++++++++ examples/remote/shared/interface.go | 110 +++++++++++++++ grpc_remote.go | 208 ++++++++++++++++++++++++++++ grpc_remote_test.go | 154 ++++++++++++++++++++ 8 files changed, 954 insertions(+), 3 deletions(-) create mode 100644 AGENTS.md create mode 100644 examples/remote/README.md create mode 100644 examples/remote/client/main.go create mode 100644 examples/remote/server/main.go create mode 100644 examples/remote/shared/interface.go create mode 100644 grpc_remote.go create mode 100644 grpc_remote_test.go diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 00000000..5792a540 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,124 @@ +# AGENTS.md - Coding Guidelines for go-plugin + +## Project Overview + +`github.com/hashicorp/go-plugin` is a Go library for building plugin systems using +process isolation with net/rpc or gRPC transport. Plugins are standalone binaries +that communicate with a host process. Go 1.24, MPL-2.0 license. + +## Build / Lint / Test Commands + +```bash +# Build all packages +go build ./... + +# Run all tests with race detection +go test -race ./... + +# Run all tests with verbose output and coverage +go test -race ./... -v -coverprofile=coverage.out + +# Run a single test +go test -race -run TestClient -v . +go test -race -run TestServer_testMode -v ./... + +# Run tests in a specific package +go test -race -v ./internal/cmdrunner + +# Lint +golangci-lint run + +# Format check (CI fails if files are changed) +go fmt ./... + +# Regenerate protobuf code (requires buf) +buf generate --path test/grpc/test.proto +``` + +## Code Style + +### File Headers + +Every `.go` file must start with the copyright header: + +```go +// Copyright IBM Corp. 2016, 2025 +// SPDX-License-Identifier: MPL-2.0 +``` + +### Package Declarations + +- Root package is `plugin`. +- Internal packages live under `internal/` (e.g., `internal/cmdrunner`, + `internal/grpcmux`, `internal/plugin`). +- Public sub-packages: `runner/`, `test/grpc/`. + +### Imports + +Group imports in three blocks separated by blank lines: +1. Standard library +2. Third-party packages +3. Internal packages (`github.com/hashicorp/go-plugin/...`) + +Use named imports sparingly; only when disambiguation is needed: + +```go +import ( + "context" + "fmt" + + hclog "github.com/hashicorp/go-hclog" + "github.com/hashicorp/go-plugin/internal/grpcmux" + "google.golang.org/grpc" +) +``` + +### Naming Conventions + +- Exported types and functions use PascalCase: `Client`, `ServeConfig`, `HandshakeConfig`. +- Unexported uses camelCase: `managedClients`, `defaultPluginLogBufferSize`. +- Interfaces define behavior; use `-er` suffix where appropriate (`Runner`). +- Test helpers accept `testing.TB` (not `*testing.T`) to support both tests and benchmarks. +- Prefix test functions with `Test`: `TestClient`, `TestServer_testMode`. +- Use underscores in test names for readability: `TestClient_killStart`. + +### Error Handling + +- Return errors; do not panic in library code. +- Define sentinel errors as package-level `var` using `errors.New`: + ```go + var ErrProcessNotFound = cmdrunner.ErrProcessNotFound + var ErrChecksumsDoNotMatch = errors.New("checksums did not match") + ``` +- Wrap errors with `fmt.Errorf("context: %w", err)` when adding context. +- Discard cleanup errors with `_ =`: `_ = l.Close()`. + +### Types and Interfaces + +- Struct fields are documented with comments above each field. +- Use interface compliance checks in test files: + ```go + var _ Plugin = (*testInterfacePlugin)(nil) + var _ Plugin = new(NetRPCUnsupportedPlugin) + ``` +- Prefer embedding for composition: `NetRPCUnsupportedPlugin` is designed for embedding. + +### Testing Patterns + +- Tests are in `*_test.go` files in the same package (not `_test` package). +- Use `t.Fatal` / `t.Fatalf` for unexpected errors, not `t.Error`. +- Use `t.TempDir()` for temporary directories; clean up with `defer`. +- Use `context.WithCancel` with `defer cancel()` for context-based tests. +- Helper processes are spawned via `helperProcess()` in `plugin_test.go`. + +### Logging + +- Use `github.com/hashicorp/go-hclog` for all logging. +- Prefer `hclog.New()` with explicit `LoggerOptions` in non-test code. +- Logger is passed via `ClientConfig.Logger` or created with sensible defaults. + +### gRPC / Protobuf + +- Proto definitions in `test/grpc/test.proto`. +- Generated code uses `buf` with settings in `buf.yaml` / `buf.gen.yaml`. +- Do not edit generated `*.pb.go` files directly. diff --git a/README.md b/README.md index 50baee06..236f2bbb 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,15 @@ created for [Packer](https://www.packer.io), it is additionally in use by [Boundary](https://www.boundaryproject.io), and [Waypoint](https://www.waypointproject.io). -While the plugin system is over RPC, it is currently only designed to work -over a local [reliable] network. Plugins over a real network are not supported -and will lead to unexpected behavior. +## Remote Plugin Support + +**New!** The plugin system now supports reliable gRPC communication over the +network. Plugins can run on different machines and communicate over TCP/IP +with built-in keepalive for reliable connections. + +For local plugins (subprocess-based), see the traditional usage below. +For remote plugins (network-based), see the [Remote Plugin Support](#remote-plugin-support-1) +section. This plugin system has been used on millions of machines across many different projects and has proven to be battle hardened and ready for production use. @@ -75,6 +81,70 @@ reattach. checksum and RPC communications can be configured to use TLS. The host process must be properly secured to protect this configuration. +## Remote Plugin Support + +The plugin system now supports reliable gRPC communication over the network. +Unlike the traditional local subprocess-based approach, remote plugins can +run on different machines and communicate over TCP/IP. + +### Key Features + +- **Network Transport**: Plugins communicate over TCP/IP instead of local sockets +- **Keepalive**: Automatic keepalive pings maintain reliable connections over + potentially unreliable networks +- **Health Checks**: Clients can verify server availability before connecting +- **TLS Support**: Can use TLS for secure connections over untrusted networks +- **Graceful Shutdown**: Server handles signals for clean shutdown + +### Server Usage + +```go +import ( + "net" + "google.golang.org/grpc" + "google.golang.org/grpc/health" + "github.com/hashicorp/go-plugin" +) + +// Create a TCP listener +lis, _ := net.Listen("tcp", ":50051") + +// Create server with keepalive enforcement +grpcServer := grpc.NewServer( + grpc.KeepaliveEnforcementPolicy( + plugin.DefaultGRPCServerKeepaliveEnforcementPolicy(), + ), +) + +// Register health check +healthCheck := health.NewServer() +healthCheck.SetServingStatus(plugin.GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING) +grpc_health_v1.RegisterHealthServer(grpcServer, healthCheck) + +// Register plugins and serve +grpcServer.Serve(lis) +``` + +### Client Usage + +```go +import "github.com/hashicorp/go-plugin" + +// Connect to remote server with keepalive +client, err := plugin.NewGRPCRemoteClient(&plugin.GRPCRemoteClientConfig{ + Addr: "localhost:50051", + Plugins: pluginMap, +}) +defer client.Close() + +// Dispense and use the plugin +raw, _ := client.Dispense("kv") +kv := raw.(MyPlugin) +kv.Put("key", []byte("value")) +``` + +See the `examples/remote` directory for a complete working example. + ## Architecture The HashiCorp plugin system works by launching subprocesses and communicating diff --git a/examples/remote/README.md b/examples/remote/README.md new file mode 100644 index 00000000..5886b373 --- /dev/null +++ b/examples/remote/README.md @@ -0,0 +1,98 @@ +# Remote Plugin Example + +This example demonstrates how to use go-plugin for remote gRPC plugin +communication over the network. Unlike the local subprocess-based examples, +this shows how to connect to a plugin server that may be running on a +different machine. + +## Features + +- **Network Transport**: Plugins communicate over TCP/IP instead of local sockets +- **Keepalive**: Automatic keepalive pings maintain reliable connections +- **Health Checks**: Clients can verify server availability before connecting +- **Graceful Shutdown**: Server handles SIGTERM for clean shutdown + +## Running the Example + +### 1. Start the Server + +```bash +cd server +go run . -addr :50051 +``` + +The server will listen on port 50051 and accept connections from anywhere. + +### 2. Run the Client + +In another terminal: + +```bash +cd client + +# Put a value +go run . -addr localhost:50051 put mykey "hello world" + +# Get the value +go run . -addr localhost:50051 get mykey + +# Health check +go run . -addr localhost:50051 -health +``` + +## Code Structure + +- `shared/` - Shared interface and gRPC implementation +- `server/` - Remote plugin server +- `client/` - Remote plugin client + +## Key Differences from Local Plugins + +1. **No Subprocess Management**: The server runs as a standalone process +2. **Network Addresses**: Uses `host:port` addresses instead of local sockets +3. **Keepalive**: Built-in keepalive for reliable network transport +4. **TLS Support**: Can use TLS for secure connections over untrusted networks + +## API Usage + +### Server Side + +```go +import ( + "google.golang.org/grpc" + "google.golang.org/grpc/health" + "github.com/hashicorp/go-plugin" +) + +// Create server with keepalive +grpcServer := grpc.NewServer( + grpc.KeepaliveEnforcementPolicy( + plugin.DefaultGRPCServerKeepaliveEnforcementPolicy(), + ), +) + +// Register health check +healthCheck := health.NewServer() +healthCheck.SetServingStatus(plugin.GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING) +grpc_health_v1.RegisterHealthServer(grpcServer, healthCheck) + +// Register plugins and serve +grpcServer.Serve(listener) +``` + +### Client Side + +```go +import "github.com/hashicorp/go-plugin" + +// Connect to remote server with keepalive +client, err := plugin.NewGRPCRemoteClient(&plugin.GRPCRemoteClientConfig{ + Addr: "localhost:50051", + Plugins: pluginMap, +}) +defer client.Close() + +// Use the plugin +raw, err := client.Dispense("kv") +kv := raw.(KV) +``` diff --git a/examples/remote/client/main.go b/examples/remote/client/main.go new file mode 100644 index 00000000..710a6c42 --- /dev/null +++ b/examples/remote/client/main.go @@ -0,0 +1,105 @@ +// Copyright IBM Corp. 2016, 2025 +// SPDX-License-Identifier: MPL-2.0 + +// This is an example of a remote plugin client that connects to a gRPC +// plugin server over the network. Unlike local plugins which are subprocess-based, +// this client connects to a remote server that may be on a different machine. +// +// The client uses keepalive to maintain reliable connections over potentially +// unreliable network connections. +// +// Usage: +// +// go run . -addr localhost:50051 [get|put] [args...] +// +// Example: +// +// # Start the server in one terminal: +// cd server && go run . -addr :50051 +// +// # In another terminal, put a value: +// cd client && go run . -addr localhost:50051 put mykey "hello world" +// +// # Get the value: +// cd client && go run . -addr localhost:50051 get mykey +package main + +import ( + "context" + "flag" + "fmt" + "log" + "os" + + "github.com/hashicorp/go-plugin" + "github.com/hashicorp/go-plugin/examples/remote/shared" +) + +func main() { + addr := flag.String("addr", "localhost:50051", "Address of the remote plugin server") + healthCheckOnly := flag.Bool("health", false, "Only perform health check and exit") + flag.Parse() + + args := flag.Args() + if len(args) < 1 && !*healthCheckOnly { + fmt.Println("Usage: client -addr
[get|put] [args...]") + fmt.Println(" client -addr
-health") + os.Exit(1) + } + + // Optionally perform a health check first + if *healthCheckOnly { + fmt.Printf("Checking health of %s...\n", *addr) + err := plugin.GRPCRemoteClientHealthCheck(context.Background(), *addr) + if err != nil { + log.Fatalf("Health check failed: %v", err) + } + fmt.Println("Server is healthy!") + return + } + + // Create a remote gRPC client. This connects to the server over the + // network with automatic keepalive for reliable connections. + client, err := plugin.NewGRPCRemoteClient(&plugin.GRPCRemoteClientConfig{ + Addr: *addr, + Plugins: shared.PluginMap, + }) + if err != nil { + log.Fatalf("Failed to connect: %v", err) + } + defer client.Close() + + // Dispense the KV plugin from the remote server + raw, err := client.Dispense("kv") + if err != nil { + log.Fatalf("Failed to dispense plugin: %v", err) + } + + kv := raw.(shared.KV) + + // Execute the requested command + switch args[0] { + case "get": + if len(args) < 2 { + log.Fatal("Usage: client get ") + } + value, err := kv.Get(args[1]) + if err != nil { + log.Fatalf("Get failed: %v", err) + } + fmt.Println(string(value)) + + case "put": + if len(args) < 3 { + log.Fatal("Usage: client put ") + } + err := kv.Put(args[1], []byte(args[2])) + if err != nil { + log.Fatalf("Put failed: %v", err) + } + fmt.Println("OK") + + default: + log.Fatalf("Unknown command: %s (use 'get' or 'put')", args[0]) + } +} diff --git a/examples/remote/server/main.go b/examples/remote/server/main.go new file mode 100644 index 00000000..f83271a6 --- /dev/null +++ b/examples/remote/server/main.go @@ -0,0 +1,82 @@ +// Copyright IBM Corp. 2016, 2025 +// SPDX-License-Identifier: MPL-2.0 + +// This is an example of a remote plugin server that serves a gRPC plugin +// over the network. Unlike local plugins which are subprocess-based, +// this server can be accessed from anywhere on the network. +// +// Usage: +// +// go run . -addr :50051 +// +// This server uses keepalive to maintain reliable connections over the +// network, making it suitable for cross-machine plugin communication. +package main + +import ( + "flag" + "fmt" + "log" + "net" + "os" + "os/signal" + "syscall" + + "github.com/hashicorp/go-plugin" + "github.com/hashicorp/go-plugin/examples/remote/shared" + "google.golang.org/grpc" + "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" +) + +func main() { + addr := flag.String("addr", ":50051", "Address to listen on (e.g. :50051 or 0.0.0.0:50051)") + flag.Parse() + + // Create a TCP listener for the network + lis, err := net.Listen("tcp", *addr) + if err != nil { + log.Fatalf("Failed to listen: %v", err) + } + fmt.Printf("Remote plugin server listening on %s\n", lis.Addr().String()) + + // Create a gRPC server with keepalive enforcement for reliable + // network connections. This ensures the server accepts keepalive + // pings from remote clients. + grpcServer := grpc.NewServer( + grpc.KeepaliveEnforcementPolicy( + plugin.DefaultGRPCServerKeepaliveEnforcementPolicy(), + ), + ) + + // Register health check service - this allows clients to verify + // the server is available before attempting to use plugins. + healthCheck := health.NewServer() + healthCheck.SetServingStatus( + plugin.GRPCServiceName, + grpc_health_v1.HealthCheckResponse_SERVING, + ) + grpc_health_v1.RegisterHealthServer(grpcServer, healthCheck) + + // Register our plugin + kvPlugin := &shared.KVGRPCPlugin{Impl: shared.NewInMemoryKV()} + if err := kvPlugin.GRPCServer(nil, grpcServer); err != nil { + log.Fatalf("Failed to register plugin: %v", err) + } + + // Handle graceful shutdown + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) + go func() { + <-sigCh + fmt.Println("\nShutting down server...") + grpcServer.GracefulStop() + }() + + fmt.Println("Server ready. Clients can connect to:", lis.Addr().String()) + fmt.Println("Press Ctrl+C to stop.") + + if err := grpcServer.Serve(lis); err != nil { + log.Fatalf("Failed to serve: %v", err) + } +} diff --git a/examples/remote/shared/interface.go b/examples/remote/shared/interface.go new file mode 100644 index 00000000..53b7d674 --- /dev/null +++ b/examples/remote/shared/interface.go @@ -0,0 +1,110 @@ +// Copyright IBM Corp. 2016, 2025 +// SPDX-License-Identifier: MPL-2.0 + +// Package shared contains shared data between the host and plugins. +// This is used by both the remote plugin server and client. +package shared + +import ( + "context" + "fmt" + + "github.com/hashicorp/go-plugin" + "github.com/hashicorp/go-plugin/examples/grpc/proto" + "google.golang.org/grpc" +) + +// Handshake is a common handshake that is shared by plugin and host. +var Handshake = plugin.HandshakeConfig{ + ProtocolVersion: 1, + MagicCookieKey: "BASIC_PLUGIN", + MagicCookieValue: "hello", +} + +// PluginMap is the map of plugins we can dispense. +var PluginMap = map[string]plugin.Plugin{ + "kv": &KVGRPCPlugin{}, +} + +// KV is the interface that we're exposing as a plugin. +type KV interface { + Put(key string, value []byte) error + Get(key string) ([]byte, error) +} + +// KVGRPCPlugin is the implementation of plugin.GRPCPlugin for remote serving. +type KVGRPCPlugin struct { + plugin.NetRPCUnsupportedPlugin + Impl KV +} + +func (p *KVGRPCPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error { + proto.RegisterKVServer(s, &GRPCServer{Impl: p.Impl}) + return nil +} + +func (p *KVGRPCPlugin) GRPCClient(ctx context.Context, broker *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error) { + return &GRPCClient{client: proto.NewKVClient(c)}, nil +} + +// GRPCClient is an implementation of KV that talks over gRPC. +type GRPCClient struct { + client proto.KVClient +} + +func (c *GRPCClient) Put(key string, value []byte) error { + _, err := c.client.Put(context.Background(), &proto.PutRequest{ + Key: key, + Value: value, + }) + return err +} + +func (c *GRPCClient) Get(key string) ([]byte, error) { + resp, err := c.client.Get(context.Background(), &proto.GetRequest{ + Key: key, + }) + if err != nil { + return nil, err + } + return resp.Value, nil +} + +// GRPCServer is the gRPC server that GRPCClient talks to. +type GRPCServer struct { + proto.UnimplementedKVServer + Impl KV +} + +func (s *GRPCServer) Put(ctx context.Context, req *proto.PutRequest) (*proto.Empty, error) { + return &proto.Empty{}, s.Impl.Put(req.Key, req.Value) +} + +func (s *GRPCServer) Get(ctx context.Context, req *proto.GetRequest) (*proto.GetResponse, error) { + v, err := s.Impl.Get(req.Key) + return &proto.GetResponse{Value: v}, err +} + +// InMemoryKV is a simple in-memory KV store for demo purposes. +type InMemoryKV struct { + store map[string][]byte +} + +func NewInMemoryKV() *InMemoryKV { + return &InMemoryKV{store: make(map[string][]byte)} +} + +func (kv *InMemoryKV) Put(key string, value []byte) error { + kv.store[key] = value + fmt.Printf("PUT %s = %s\n", key, string(value)) + return nil +} + +func (kv *InMemoryKV) Get(key string) ([]byte, error) { + value, ok := kv.store[key] + if !ok { + return nil, fmt.Errorf("key not found: %s", key) + } + fmt.Printf("GET %s = %s\n", key, string(value)) + return value, nil +} diff --git a/grpc_remote.go b/grpc_remote.go new file mode 100644 index 00000000..66a500b5 --- /dev/null +++ b/grpc_remote.go @@ -0,0 +1,208 @@ +// Copyright IBM Corp. 2016, 2025 +// SPDX-License-Identifier: MPL-2.0 + +package plugin + +import ( + "context" + "crypto/tls" + "math" + "time" + + "github.com/hashicorp/go-hclog" + "github.com/hashicorp/go-plugin/internal/plugin" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/keepalive" +) + +// GRPCRemoteClientConfig configures a remote gRPC plugin client that connects +// to a plugin server running on a network address, rather than a local +// subprocess. +// +// Remote gRPC clients are designed to work reliably over network connections +// with automatic keepalive, reconnection, and long-lived streaming support. +type GRPCRemoteClientConfig struct { + // Addr is the network address of the remote plugin server. + // This should be in "host:port" format for TCP connections. + Addr string + + // TLSConfig, if set, enables TLS for the connection. + TLSConfig *tls.Config + + // Plugins are the plugins that can be consumed. + Plugins PluginSet + + // Logger is used for logging. If none is provided, a default logger + // will be created. + Logger hclog.Logger + + // GRPCDialOptions allows passing custom gRPC dial options. + // Keepalive options will be merged with these defaults if + // KeepaliveParams are not set. + GRPCDialOptions []grpc.DialOption + + // KeepaliveParams configures client-side keepalive for reliable + // network transport. If nil, sensible defaults are used suitable + // for remote connections. + KeepaliveParams *keepalive.ClientParameters + + // KeepaliveEnforcementPolicy configures server-side keepalive + // enforcement. This is used only when the server configures + // keepalive with this policy. + KeepaliveEnforcementPolicy *keepalive.EnforcementPolicy +} + +// DefaultGRPCRemoteKeepaliveParams returns keepalive parameters suitable +// for reliable operation over network connections. These defaults send +// keepalive pings every 10 seconds and wait 20 seconds for a response +// before considering the connection dead. +func DefaultGRPCRemoteKeepaliveParams() keepalive.ClientParameters { + return keepalive.ClientParameters{ + // Send keepalive pings every 10 seconds when there is no activity. + Time: 10 * time.Second, + // Wait 20 seconds for a keepalive ping response before considering + // the connection dead. + Timeout: 20 * time.Second, + // Send keepalive pings even if there are no active streams. + // This ensures the connection stays alive during idle periods. + PermitWithoutStream: true, + } +} + +// DefaultGRPCServerKeepaliveEnforcementPolicy returns an enforcement policy +// suitable for remote gRPC servers. Use this when creating the gRPC server +// to ensure it accepts keepalive pings from remote clients. +func DefaultGRPCServerKeepaliveEnforcementPolicy() keepalive.EnforcementPolicy { + return keepalive.EnforcementPolicy{ + // Allow clients to send keepalive pings without active streams. + MinTime: 10 * time.Second, + PermitWithoutStream: true, + } +} + +// NewGRPCRemoteClient creates a new GRPCClient connected to a remote plugin +// server at the given address. Unlike the subprocess-based Client, this +// connects directly to a running gRPC server over the network. +// +// The returned GRPCClient uses keepalive to maintain reliable connections +// over potentially unreliable networks. Callers should call Close() when +// done to release resources. +func NewGRPCRemoteClient(config *GRPCRemoteClientConfig) (*GRPCClient, error) { + if config.Logger == nil { + config.Logger = hclog.New(&hclog.LoggerOptions{ + Output: hclog.DefaultOutput, + Level: hclog.Trace, + Name: "plugin", + }) + } + + conn, err := dialRemoteGRPCConn(config) + if err != nil { + return nil, err + } + + doneCtx, cancel := context.WithCancel(context.Background()) + _ = cancel // cancel is called by GRPCClient.Close via its Close method + + // Start the broker for auxiliary gRPC connections + brokerGRPCClient := newGRPCBrokerClient(conn) + broker := newGRPCBroker(brokerGRPCClient, config.TLSConfig, UnixSocketConfig{}, nil, nil) + go broker.Run() + go func() { _ = brokerGRPCClient.StartStream() }() + + // Start stdio client - this may fail gracefully if the server doesn't + // support it (e.g. for non-Go plugins). + stdioClient, err := newGRPCStdioClient(doneCtx, config.Logger.Named("stdio"), conn) + if err != nil { + config.Logger.Warn("failed to create stdio client", "err", err) + } + if stdioClient != nil { + go stdioClient.Run(nil, nil) + } + + cl := &GRPCClient{ + Conn: conn, + Plugins: config.Plugins, + doneCtx: doneCtx, + broker: broker, + controller: plugin.NewGRPCControllerClient(conn), + } + + return cl, nil +} + +// dialRemoteGRPCConn creates a gRPC connection to a remote server with +// keepalive and appropriate settings for reliable network transport. +func dialRemoteGRPCConn(config *GRPCRemoteClientConfig) (*grpc.ClientConn, error) { + opts := make([]grpc.DialOption, 0) + + // Configure TLS + if config.TLSConfig == nil { + opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) + } else { + opts = append(opts, grpc.WithTransportCredentials( + credentials.NewTLS(config.TLSConfig))) + } + + // Configure keepalive for reliable network transport + keepaliveParams := DefaultGRPCRemoteKeepaliveParams() + if config.KeepaliveParams != nil { + keepaliveParams = *config.KeepaliveParams + } + opts = append(opts, grpc.WithKeepaliveParams(keepaliveParams)) + + // Set large message size limits + opts = append(opts, + grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(math.MaxInt32)), + grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(math.MaxInt32))) + + // Add any custom dial options (merging with defaults) + opts = append(opts, config.GRPCDialOptions...) + + conn, err := grpc.Dial(config.Addr, opts...) + if err != nil { + return nil, err + } + + return conn, nil +} + +// NewGRPCRemoteServer creates a gRPC server suitable for remote plugin hosting. +// It configures keepalive enforcement to work reliably with remote clients +// and returns a server that can be registered with plugin services. +// +// This is a convenience function; you can also create the server directly +// using grpc.NewServer with the appropriate options. +func NewGRPCRemoteServer(opts ...grpc.ServerOption) *grpc.Server { + // Add keepalive enforcement policy for remote clients + opts = append(opts, + grpc.KeepaliveEnforcementPolicy(DefaultGRPCServerKeepaliveEnforcementPolicy())) + + return grpc.NewServer(opts...) +} + +// GRPCRemoteClientHealthCheck performs a health check on a remote gRPC plugin +// server. This can be used to verify that a remote server is available and +// healthy before attempting to use it. +func GRPCRemoteClientHealthCheck(ctx context.Context, addr string, opts ...grpc.DialOption) error { + // Add keepalive for reliability + keepaliveParams := DefaultGRPCRemoteKeepaliveParams() + opts = append(opts, grpc.WithKeepaliveParams(keepaliveParams)) + opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) + + conn, err := grpc.Dial(addr, opts...) + if err != nil { + return err + } + defer conn.Close() + + client := grpc_health_v1.NewHealthClient(conn) + _, err = client.Check(ctx, &grpc_health_v1.HealthCheckRequest{ + Service: GRPCServiceName, + }) + + return err +} diff --git a/grpc_remote_test.go b/grpc_remote_test.go new file mode 100644 index 00000000..f9ce5976 --- /dev/null +++ b/grpc_remote_test.go @@ -0,0 +1,154 @@ +// Copyright IBM Corp. 2016, 2025 +// SPDX-License-Identifier: MPL-2.0 + +package plugin + +import ( + "context" + "net" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/keepalive" +) + +func TestGRPCRemoteClient(t *testing.T) { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer lis.Close() + + s := grpc.NewServer() + registerTestGRPCServices(t, s) + go s.Serve(lis) + defer s.Stop() + + client, err := NewGRPCRemoteClient(&GRPCRemoteClientConfig{ + Addr: lis.Addr().String(), + Plugins: testGRPCPluginMap, + }) + if err != nil { + t.Fatal(err) + } + defer client.Close() + + if err := client.Ping(); err != nil { + t.Fatal(err) + } + + raw, err := client.Dispense("test") + if err != nil { + t.Fatal(err) + } + + impl, ok := raw.(testInterface) + if !ok { + t.Fatalf("expected testInterface, got %T", raw) + } + + result := impl.Double(21) + if result != 42 { + t.Fatalf("expected 42, got %d", result) + } +} + +func TestGRPCRemoteClient_WithKeepalive(t *testing.T) { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer lis.Close() + + s := grpc.NewServer( + grpc.KeepaliveEnforcementPolicy(DefaultGRPCServerKeepaliveEnforcementPolicy()), + ) + registerTestGRPCServices(t, s) + go s.Serve(lis) + defer s.Stop() + + client, err := NewGRPCRemoteClient(&GRPCRemoteClientConfig{ + Addr: lis.Addr().String(), + Plugins: testGRPCPluginMap, + KeepaliveParams: &keepalive.ClientParameters{ + Time: 1 * time.Second, + Timeout: 5 * time.Second, + PermitWithoutStream: true, + }, + }) + if err != nil { + t.Fatal(err) + } + defer client.Close() + + if err := client.Ping(); err != nil { + t.Fatal(err) + } + + time.Sleep(100 * time.Millisecond) + + if err := client.Ping(); err != nil { + t.Fatal(err) + } +} + +func TestGRPCRemoteClient_HealthCheck(t *testing.T) { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer lis.Close() + + s := grpc.NewServer() + healthCheck := health.NewServer() + healthCheck.SetServingStatus(GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING) + grpc_health_v1.RegisterHealthServer(s, healthCheck) + go s.Serve(lis) + defer s.Stop() + + err = GRPCRemoteClientHealthCheck(context.Background(), lis.Addr().String()) + if err != nil { + t.Fatal(err) + } +} + +func TestGRPCRemoteClient_HealthCheckFails(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + + err := GRPCRemoteClientHealthCheck(ctx, "127.0.0.1:1") + if err == nil { + t.Fatal("expected health check to fail for non-existent server") + } +} + +func TestGRPCRemoteClient_ServerNotRunning(t *testing.T) { + client, err := NewGRPCRemoteClient(&GRPCRemoteClientConfig{ + Addr: "127.0.0.1:1", + Plugins: testGRPCPluginMap, + }) + if err != nil { + t.Fatal(err) + } + defer client.Close() + + if err := client.Ping(); err == nil { + t.Fatal("expected ping to fail when server is not running") + } +} + +func registerTestGRPCServices(t *testing.T, s *grpc.Server) { + t.Helper() + + healthCheck := health.NewServer() + healthCheck.SetServingStatus(GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING) + grpc_health_v1.RegisterHealthServer(s, healthCheck) + + testPlugin := &testGRPCInterfacePlugin{} + if err := testPlugin.GRPCServer(nil, s); err != nil { + t.Fatal(err) + } +}