fix(sdk): replaces nri/net/multiplex with yamux

Author: joe0BAB

go.mod

@@ -6,6 +6,7 @@ require (
 	connectrpc.com/connect v1.18.1
 	github.com/alecthomas/kong v1.11.0
 	github.com/containerd/nri v0.9.0
+	github.com/hashicorp/yamux v0.1.2
 	github.com/keybase/go-keychain v0.0.1
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.10.0

go.sum

@@ -30,6 +30,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/pprof v0.0.0-20240424215950-a892ee059fd6 h1:k7nVchz72niMH6YLQNvHSdIE7iqsQxK1P41mySCvssg=
 github.com/google/pprof v0.0.0-20240424215950-a892ee059fd6/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw=
+github.com/hashicorp/yamux v0.1.2 h1:XtB8kyFOyHXYVFnwT5C3+Bdo8gArse7j2AQ0DA0Uey8=
+github.com/hashicorp/yamux v0.1.2/go.mod h1:C+zze2n6e/7wshOZep2A70/aQU6QBRWJO/G6FT1wIns=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
 github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg=
 github.com/keybase/go-keychain v0.0.1 h1:way+bWYa6lDppZoZcgMbYsvC7GxljxrskdNInRtuthU=

internal/ipc/design.md

@@ -0,0 +1,55 @@
+# IPC and multiplexing a socket
+The plugin system has two parts:
+- the runtime: launches plugins or allows manual launched plugins to connect
+- the plugin(s): registers to the runtime and is domain expert for a specific secret provider
+
+From the plugin system perspective, the runtime is a lifecycle management server for plugins.
+However, from the secrets engine perspective, the runtime is a client that can request secrets from a plugin.
+
+To avoid having a socket per plugin, we multiplex the socket.
+```mermaid
+flowchart TD
+    Socket[(Multiplexed<br>Socket)]
+
+    subgraph PluginBlock [Plugin]
+        direction TB
+        PSrv["Provider (Server)"]
+        PC["Plugin (Client)"]
+    end
+
+    subgraph ResolverBlock [Resolver]
+        direction TB
+        RPC["Resolver (Provider Client)"]
+        PServ[Plugin Server/Runtime]
+    end
+
+    Socket -->|register plugin| PServ
+    Socket -->|get secret| PSrv
+    PC -->|register plugin| Socket
+    RPC -->|get secret| Socket
+
+```
+
+## Choosing a multiplexer
+The multiplexer adds a custom layer on top of the socket that allows running servers on both ends of the socket.
+
+### nri/net/multiplex - a minimal multiplexer
+The plugin system in [containerd/nr](https://github.com/containerd/nri) implements its own simple frame-based multiplexer [nri/net/multiplex](https://github.com/containerd/nri/tree/main/pkg/net/multiplex). 
+It provides two streams on each side that need to be re-used for all communication.
+This works well for [ttrpc](https://github.com/containerd/ttrpc) which uses its own length-prefixed framing.
+However, the standard Go HTTP server inside `Server(net.Listener)` does one `Accept()`, gets one `net.Conn`, and then loops inside serveConn to decode requests. 
+But because that sub-connection isn’t a real socket—and because the mux delivers no further “new connections” and any pipelined data after the first request can get lost or stuck in the mux’s framing, and the server never sees it.
+
+Alternatively, HTTP/2 without TLS could be used as it has gives control over the framing.
+Unfortunately, Go's `net/http` package does not easily support HTTP/2 without TLS and getting it work comes with its own set of challenges, such as requiring a custom `net.Listener` implementation that handles the HTTP/2 framing.
+
+TLDR: [nri/net/multiplex](https://github.com/containerd/nri/tree/main/pkg/net/multiplex) is not ideal for general HTTP servers.
+
+### Yamux
+
+Yamux is a full-featured, multiplexing protocol that allows multiple streams to be sent over a single TCP connection, actively maintained by Hashicorp and e.g. used in Hashicorp's Nomad.
+I.e., using Yamux, Go's `net/http` works out-of-the-box.
+
+
+
+

internal/ipc/ipc.go

@@ -8,39 +8,33 @@ import (
 	"net/http"
 	"sync"
 
-	"github.com/containerd/nri/pkg/net/multiplex"
+	"github.com/hashicorp/yamux"
 )
 
-type PluginIPC interface {
-	// Conn returns a connection that can be used to reach the runtime server on the other end of
-	// the multiplexed connection (this is not the original net.Conn, but a multiplexed connection!).
-	Conn() net.Conn
-	// Wait blocks forever until the server is closed or an error occurs.
-	// Cancelling the context will not close the server, but will return nil.
-	Wait(ctx context.Context) error
-	// Close shuts down the server and closes the multiplexer, and its connection/listener.
-	Close() error
-}
-
-func NewPluginIPC(sockConn net.Conn, handler http.Handler) (PluginIPC, error) {
-	return newMuxedIPC(sockConn, handler, multiplex.PluginServiceConn, multiplex.RuntimeServiceConn)
+func NewPluginIPC(sockConn net.Conn, handler http.Handler) (IPC, *http.Client, error) {
+	session, err := yamux.Client(sockConn, nil)
+	if err != nil {
+		return nil, nil, fmt.Errorf("creating yamux client: %w", err)
+	}
+	i, c := newMuxedIPC(session, handler)
+	return i, c, nil
 }
 
-type RuntimeIPC interface {
-	// Conn returns a connection that can be used to reach the server running in the plugin on the other end of
-	// the multiplexed connection (this is not the original net.Conn, but a multiplexed connection!).
-	Conn() net.Conn
+type IPC interface {
 	// Wait blocks forever until the server is closed or an error occurs.
 	// Cancelling the context will not close the server, but will return nil.
 	Wait(ctx context.Context) error
 	// Close shuts down the server and closes the multiplexer, and its connection/listener.
 	Close() error
-	// Unblock unblocks the multiplexed connection, allowing it to read from the socket.
-	Unblock()
 }
 
-func NewRuntimeIPC(sockConn net.Conn, handler http.Handler) (RuntimeIPC, error) {
-	return newMuxedIPC(sockConn, handler, multiplex.RuntimeServiceConn, multiplex.PluginServiceConn, multiplex.WithBlockedRead())
+func NewRuntimeIPC(sockConn net.Conn, handler http.Handler) (IPC, *http.Client, error) {
+	session, err := yamux.Server(sockConn, nil)
+	if err != nil {
+		return nil, nil, fmt.Errorf("creating yamux server: %w", err)
+	}
+	i, c := newMuxedIPC(session, handler)
+	return i, c, nil
 }
 
 type ipcServer struct {
@@ -49,7 +43,7 @@ type ipcServer struct {
 	err    error
 }
 
-func newIpcServer(l net.Listener, handler http.Handler, onError func()) *ipcServer {
+func newIpcServer(l net.Listener, handler http.Handler, onError func() error) *ipcServer {
 	result := &ipcServer{
 		done: make(chan struct{}),
 		server: &http.Server{
@@ -59,48 +53,28 @@ func newIpcServer(l net.Listener, handler http.Handler, onError func()) *ipcServ
 	go func() {
 		err := result.server.Serve(l)
 		if !errors.Is(err, http.ErrServerClosed) {
-			onError()
-			result.err = err
+			result.err = errors.Join(err, onError())
 		}
 		close(result.done)
 	}()
 	return result
 }
 
 type ipcImpl struct {
-	mConn    net.Conn
 	server   *ipcServer
 	teardown func() error
-	unblock  func()
 }
 
-func newMuxedIPC(sockConn net.Conn, handler http.Handler, listenerID, connID multiplex.ConnID, options ...multiplex.Option) (*ipcImpl, error) {
-	mux := multiplex.Multiplex(sockConn, options...)
-	listener, err := mux.Listen(listenerID)
-	if err != nil {
-		mux.Close()
-		return nil, err
-	}
-	conn, err := mux.Open(connID)
-	if err != nil {
-		mux.Close()
-		return nil, fmt.Errorf("failed to multiplex grcp client connection: %w", err)
-	}
-	server := newIpcServer(listener, handler, func() { mux.Close() })
+func newMuxedIPC(session *yamux.Session, handler http.Handler) (*ipcImpl, *http.Client) {
+	server := newIpcServer(session, handler, session.Close)
 	return &ipcImpl{
-		mConn:  conn,
 		server: server,
 		teardown: sync.OnceValue(func() error {
-			err := errors.Join(server.server.Close(), mux.Close())
+			err := errors.Join(server.server.Close(), session.Close())
 			<-server.done
 			return err
 		}),
-		unblock: mux.Unblock,
-	}, nil
-}
-
-func (i *ipcImpl) Conn() net.Conn {
-	return i.mConn
+	}, createYamuxedClient(session)
 }
 
 func (i *ipcImpl) Wait(ctx context.Context) error {
@@ -116,6 +90,11 @@ func (i *ipcImpl) Close() error {
 	return i.teardown()
 }
 
-func (i *ipcImpl) Unblock() {
-	i.unblock()
+func createYamuxedClient(session *yamux.Session) *http.Client {
+	transport := &http.Transport{
+		DialContext: func(context.Context, string, string) (net.Conn, error) {
+			return session.Open()
+		},
+	}
+	return &http.Client{Transport: transport}
 }

internal/ipc/ipc_test.go

@@ -0,0 +1,98 @@
+package ipc
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	mockPingPath = "/ping"
+)
+
+func Test_newExternalPlugin(t *testing.T) {
+	tests := []struct {
+		name string
+		test func(t *testing.T)
+	}{
+		{
+			name: "ping pong on both sides",
+			test: func(t *testing.T) {
+				socketPath := filepath.Join(os.TempDir(), "secrets-engine-plugin.sock")
+				os.Remove(socketPath)
+				require.NoError(t, os.MkdirAll(filepath.Dir(socketPath), 0755))
+				l, err := net.ListenUnix("unix", &net.UnixAddr{
+					Name: socketPath,
+					Net:  "unix",
+				})
+				require.NoError(t, err)
+				doneRuntime := make(chan struct{})
+				donePlugin := make(chan struct{})
+				go func() {
+					sock, err := l.Accept()
+					if err != nil {
+						log.Fatalf("plugin accept: %v", err)
+					}
+					defer sock.Close()
+
+					pluginHandler := http.NewServeMux()
+					pluginHandler.HandleFunc(mockPingPath, func(w http.ResponseWriter, _ *http.Request) {
+						fmt.Fprint(w, "pong-runtime")
+					})
+					i, c, err := NewRuntimeIPC(sock, pluginHandler)
+					if err != nil {
+						log.Fatalf("plugin IPC setup error: %v", err)
+					}
+					defer i.Close()
+					assertCommunicationToServer(t, c, "pong-plugin")
+					assertCommunicationToServer(t, c, "pong-plugin") // run at least twice to ensure re-opening connection works
+
+					close(doneRuntime)
+					<-donePlugin
+				}()
+
+				sock, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: socketPath, Net: "unix"})
+				if err != nil {
+					log.Fatalf("dial error: %v", err)
+				}
+				t.Cleanup(func() { sock.Close() })
+
+				runtimeHandler := http.NewServeMux()
+				runtimeHandler.HandleFunc(mockPingPath, func(w http.ResponseWriter, _ *http.Request) {
+					fmt.Fprint(w, "pong-plugin")
+				})
+				i, c, err := NewPluginIPC(sock, runtimeHandler)
+				require.NoError(t, err)
+				t.Cleanup(func() { i.Close() })
+
+				assertCommunicationToServer(t, c, "pong-runtime")
+				assertCommunicationToServer(t, c, "pong-runtime") // run at least twice to ensure re-opening connection works
+				close(donePlugin)
+				<-doneRuntime
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.test(t)
+		})
+	}
+}
+
+func assertCommunicationToServer(t *testing.T, c *http.Client, response string) {
+	req, _ := http.NewRequest("GET", "http://unused"+mockPingPath, nil)
+	resp, err := c.Do(req)
+	require.NoError(t, err)
+	body, _ := io.ReadAll(resp.Body)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Equal(t, response, string(body))
+}

pkg/adaptation/setup.go

@@ -14,9 +14,9 @@ import (
 )
 
 type SetupResult struct {
-	conn  net.Conn
-	cfg   pluginCfgIn
-	close func() error
+	client *http.Client
+	cfg    pluginCfgIn
+	close  func() error
 }
 
 var _ pluginCfgInValidator = &setupValidator{}
@@ -36,11 +36,10 @@ func Setup(conn net.Conn, v setupValidator) (*SetupResult, error) {
 	})
 	registrator := newRegistrationLogic(v, chRegistrationResult)
 	httpMux.Handle(resolverv1connect.NewEngineServiceHandler(&RegisterService{registrator}))
-	i, err := ipc.NewRuntimeIPC(conn, httpMux)
+	i, c, err := ipc.NewRuntimeIPC(conn, httpMux)
 	if err != nil {
 		return nil, err
 	}
-	i.Unblock()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	chIpcErr := make(chan error, 1)
@@ -63,9 +62,9 @@ func Setup(conn net.Conn, v setupValidator) (*SetupResult, error) {
 		return nil, errors.New("plugin registration timed out")
 	}
 	return &SetupResult{
-		conn:  conn,
-		cfg:   out,
-		close: i.Close,
+		client: c,
+		cfg:    out,
+		close:  i.Close,
 	}, nil
 }
 

plugin/go.mod

@@ -14,17 +14,12 @@ require (
 )
 
 require (
-	github.com/containerd/log v0.1.0 // indirect
-	github.com/containerd/ttrpc v1.2.6-0.20240827082320-b5cd6e4b3287 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/hashicorp/yamux v0.1.2 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/sys v0.29.0 // indirect
-	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20230731190214-cbb8c96f2d6d // indirect
-	google.golang.org/grpc v1.57.1 // indirect
 	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )