Merge pull request #59 from docker/feat/adaptation-external-plugins

feat(sdk): plugin runtime to handle externally launched plugins

Author: joe0BAB

Dockerfile

@@ -18,7 +18,7 @@ FROM base AS lint
 COPY --from=lint-base /usr/bin/golangci-lint /usr/bin/golangci-lint
 ARG TARGETOS
 ARG TARGETARCH
-RUN --mount=target=. \
+RUN --mount=type=bind,target=. \
     --mount=type=cache,target=/go/pkg/mod \
     --mount=type=cache,target=/root/.cache/go-build \
     --mount=type=cache,target=/root/.cache/golangci-lint <<EOD

go.mod

@@ -2,10 +2,13 @@ module github.com/docker/secrets-engine
 
 go 1.24.3
 
+replace github.com/docker/secrets-engine/plugin => ./plugin/
+
 require (
 	connectrpc.com/connect v1.18.1
 	github.com/alecthomas/kong v1.11.0
 	github.com/containerd/nri v0.9.0
+	github.com/docker/secrets-engine/plugin v0.0.0-00010101000000-000000000000
 	github.com/hashicorp/yamux v0.1.2
 	github.com/keybase/go-keychain v0.0.1
 	github.com/sirupsen/logrus v1.9.3

internal/ipc/design.md

@@ -51,5 +51,32 @@ Yamux is a full-featured, multiplexing protocol that allows multiple streams to
 Using Yamux we get Go's `net/http` out-of-the-box.
 
 
+## Decisions
 
+---
+
+2025-07-02 IPC stack
+
+The IPC stack consists of multiple parts that need to play well together:
+- socket multiplexing
+- API format (includes networking protocol + serialization format)
+
+At this point in time we have decided to go with yamux + connect rpc. 
+Connect rpc in itself uses protobuf for data serialization combined with gRPC over http for networking.
+A main advantage is that we can keep using Go's standard library's `net/http` stack for server and client.
+See [Connect: A better gRPC](https://buf.build/blog/connect-a-better-grpc) for a detailed comparison against e.g.`grpc-go`.
+Also connect rpc is part of CNCF ([source](https://www.cncf.io/projects/connect-rpc/)).
+
+Potential drawbacks: Performance
+
+Using [nri/net/multiplex](https://github.com/containerd/nri/tree/main/pkg/net/multiplex) with [ttrpc](https://github.com/containerd/ttrpc) probably would be the most performant solution.
+It re-uses one stream over the multiplexed socket per direction and does not have the overhead of the HTTP protocol as Protobuf gets streamed directly over the multiplexer.
+Although lightweight, it has stopped evolving and has not caught up to the latest improvements on Protobuf.
+Another major downside is that it's mainly Go only. 
+Plugins written in a different language would come at a high cost.
+
+We argue that in our use case since the networking only happens locally the overhead of GRPC over HTTP and the cost of opening a new yamux stream per API request are negligible. 
+In addition, the main performance bottleneck will be within the actual plugins due to IO operations, additional upstream network requests and potentially authentication.
+
+---
 

internal/ipc/ipc.go

@@ -8,9 +8,11 @@ import (
 	"net"
 	"net/http"
 	"sync"
+	"syscall"
 	"time"
 
 	"github.com/hashicorp/yamux"
+	"github.com/sirupsen/logrus"
 )
 
 const (
@@ -56,7 +58,7 @@ type ipcServer struct {
 	err    error
 }
 
-func newIpcServer(l net.Listener, handler http.Handler, onClose func(error) error) *ipcServer {
+func newIpcServer(l net.Listener, handler http.Handler, afterClose func(error) error) *ipcServer {
 	result := &ipcServer{
 		done: make(chan struct{}),
 		server: &http.Server{
@@ -68,7 +70,7 @@ func newIpcServer(l net.Listener, handler http.Handler, onClose func(error) erro
 		if errors.Is(err, http.ErrServerClosed) { // not an error, client closed the connection
 			err = nil
 		}
-		result.err = errors.Join(filterEOF(err), onClose(err)) // EOF: only forward to the onClose handler, but filter out internal forwarding
+		result.err = errors.Join(filterBrokenPipe(filterEOF(err)), afterClose(err)) // EOF: only forward to the afterClose handler, but filter out internal forwarding
 		close(result.done)
 	}()
 	return result
@@ -81,6 +83,13 @@ func filterEOF(err error) error {
 	return err
 }
 
+func filterBrokenPipe(err error) error {
+	if errors.Is(err, syscall.EPIPE) {
+		return nil
+	}
+	return err
+}
+
 type ipcImpl struct {
 	server   *ipcServer
 	teardown func() error
@@ -99,16 +108,34 @@ func newMuxedIPC(session *yamux.Session, handler http.Handler, onClose func(erro
 		}
 		return session.Close()
 	})
+	c := createYamuxedClient(session)
 	return &ipcImpl{
 		server: server,
 		teardown: sync.OnceValue(func() error {
-			ctx, cancel := context.WithTimeout(context.Background(), cfg.shutdownTimeout)
-			defer cancel()
-			err := server.server.Shutdown(ctx)
+			_ = session.GoAway()
+			c.CloseIdleConnections()
+			waitForClientToDisconnect(session, cfg.shutdownTimeout)
+			err := server.server.Close()
 			<-server.done
-			return errors.Join(err, session.Close(), server.err)
+			return errors.Join(err, server.err)
 		}),
-	}, createYamuxedClient(session)
+	}, c
+}
+
+func waitForClientToDisconnect(s *yamux.Session, t time.Duration) {
+	timeout := time.After(t)
+	for {
+		select {
+		case <-time.After(50 * time.Millisecond):
+		case <-timeout:
+			logrus.Debugf("Timeout expired but %d streams still open, shutting down server...", s.NumStreams())
+			return
+		}
+		streams := s.NumStreams()
+		if streams <= 0 {
+			return
+		}
+	}
 }
 
 func (i *ipcImpl) Close() error {
@@ -120,6 +147,11 @@ func createYamuxedClient(session *yamux.Session) *http.Client {
 		DialContext: func(context.Context, string, string) (net.Conn, error) {
 			return session.Open()
 		},
+		// We don't want to use http keepalive because
+		// - we keep re-using the underlying socket anyway
+		// - it allows clean bookkeeping of yamux session / any yamux session means IPC happening and not stale keepalive connections
+		// -> we can check for session.NumStreams() on shutdown
+		DisableKeepAlives: true,
 	}
 	return &http.Client{Transport: transport}
 }

pkg/adaptation/plugin.go

@@ -1,10 +1,18 @@
 package adaptation
 
 import (
+	"context"
+	"net"
 	"sync"
 	"time"
 
+	"connectrpc.com/connect"
+	"google.golang.org/protobuf/proto"
+
 	"github.com/docker/secrets-engine/pkg/api"
+	v1 "github.com/docker/secrets-engine/pkg/api/resolver/v1"
+	"github.com/docker/secrets-engine/pkg/api/resolver/v1/resolverv1connect"
+	"github.com/docker/secrets-engine/pkg/secrets"
 )
 
 var (
@@ -24,3 +32,59 @@ func getPluginRegistrationTimeout() time.Duration {
 	defer timeoutCfgLock.RUnlock()
 	return pluginRegistrationTimeout
 }
+
+var (
+	_ secrets.Resolver = &plugin{}
+)
+
+type plugin struct {
+	sync.Mutex
+	base           string
+	pattern        secrets.Pattern
+	version        string
+	pluginClient   resolverv1connect.PluginServiceClient
+	resolverClient resolverv1connect.ResolverServiceClient
+	close          func() error
+}
+
+// newExternalPlugin creates a plugin (stub) for an accepted external plugin connection.
+func newExternalPlugin(conn net.Conn, v setupValidator) (*plugin, error) {
+	r, err := setup(conn, v)
+	if err != nil {
+		return nil, err
+	}
+	return &plugin{
+		base:           r.cfg.name,
+		pattern:        r.cfg.pattern,
+		version:        r.cfg.version,
+		pluginClient:   resolverv1connect.NewPluginServiceClient(r.client, "http://unix"),
+		resolverClient: resolverv1connect.NewResolverServiceClient(r.client, "http://unix"),
+		close:          r.close,
+	}, nil
+}
+
+func (p *plugin) GetSecret(ctx context.Context, request secrets.Request) (secrets.Envelope, error) {
+	req := connect.NewRequest(v1.GetSecretRequest_builder{
+		SecretId: proto.String(request.ID.String()),
+	}.Build())
+	resp, err := p.resolverClient.GetSecret(ctx, req)
+	if err != nil {
+		return envelopeErr(request, err), err
+	}
+	id, err := secrets.ParseID(resp.Msg.GetSecretId())
+	if err != nil {
+		return envelopeErr(request, err), err
+	}
+	if id != request.ID {
+		return envelopeErr(request, secrets.ErrIDMismatch), secrets.ErrIDMismatch
+	}
+	return secrets.Envelope{
+		ID:       id,
+		Value:    []byte(resp.Msg.GetSecretValue()),
+		Provider: p.base,
+	}, nil
+}
+
+func envelopeErr(req secrets.Request, err error) secrets.Envelope {
+	return secrets.Envelope{ID: req.ID, ResolvedAt: time.Now(), Error: err.Error()}
+}