Merge pull request #54039 from vvoland/work-ext-client
extpoints: Let extensions use a regular Engine API client
This commit is contained in:
19 files changed
+1120
-2
No files matched your search
@@ -361,7 +361,9 @@ func (cli *daemonCLI) start(ctx context.Context) (retErr error) {
|
||||
return err
|
||||
}
|
||||
httpServer.Protocols = &p
|
||||
httpServer.Handler = newHTTPHandler(ctx, gs, extProxy, apiServer.CreateMux(ctx, routers...))
|
||||
handler := newHTTPHandler(ctx, gs, extProxy, apiServer.CreateMux(ctx, routers...))
|
||||
httpServer.Handler = handler
|
||||
d.StartAPISockets(handler)
|
||||
|
||||
go d.ProcessClusterNotifications(ctx, c.GetWatchStream())
|
||||
|
||||
|
||||
@@ -38,6 +38,7 @@ import (
|
||||
networktypes "github.com/moby/moby/api/types/network"
|
||||
registrytypes "github.com/moby/moby/api/types/registry"
|
||||
"github.com/moby/moby/api/types/swarm"
|
||||
"github.com/moby/moby/v2/daemon/internal/apisocket"
|
||||
"github.com/moby/moby/v2/daemon/internal/nri"
|
||||
"github.com/moby/sys/user"
|
||||
"github.com/moby/sys/userns"
|
||||
@@ -136,6 +137,7 @@ type Daemon struct {
|
||||
nri *nri.NRI
|
||||
pluginManager *plugin.Manager
|
||||
extensionHost *host.Host
|
||||
apiSockets *apisocket.Manager
|
||||
linkIndex *linkIndex
|
||||
containerdClient *containerd.Client
|
||||
containerd libcontainerdtypes.Client
|
||||
@@ -933,6 +935,7 @@ func NewDaemon(ctx context.Context, config *config.Config, pluginStore *plugin.S
|
||||
d := &Daemon{
|
||||
PluginStore: pluginStore,
|
||||
startupDone: make(chan struct{}),
|
||||
apiSockets: apisocket.NewManager(),
|
||||
}
|
||||
cfgStore := &configStore{
|
||||
Config: *config,
|
||||
@@ -1556,6 +1559,11 @@ func (daemon *Daemon) Shutdown(ctx context.Context) error {
|
||||
log.G(ctx).WithError(err).Error("failed to shut down extensions")
|
||||
}
|
||||
}
|
||||
// Close the API sockets only after extensions have shut down, so they can
|
||||
// still use the API while shutting down.
|
||||
if err := daemon.apiSockets.Close(); err != nil {
|
||||
log.G(ctx).WithError(err).Error("failed to close API sockets")
|
||||
}
|
||||
|
||||
// Keep mounts and networking running on daemon shutdown if
|
||||
// we are to keep containers running and restore them.
|
||||
|
||||
+41
-1
@@ -2,7 +2,10 @@ package daemon
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
@@ -21,13 +24,19 @@ import (
|
||||
// newExtensionHost builds the daemon's extension host.
|
||||
func (daemon *Daemon) newExtensionHost(ctx context.Context) (*host.Host, error) {
|
||||
cfg := daemon.Config()
|
||||
runtimeDir := filepath.Join(cfg.ExecRoot, "extensions")
|
||||
runtimeDir := extensionRuntimeDir(&cfg)
|
||||
if err := os.MkdirAll(runtimeDir, 0o700); err != nil {
|
||||
return nil, fmt.Errorf("create extension runtime directory: %w", err)
|
||||
}
|
||||
if err := os.Chmod(runtimeDir, 0o700); err != nil {
|
||||
return nil, fmt.Errorf("set permissions on extension runtime directory: %w", err)
|
||||
}
|
||||
// A crashed daemon leaves its API socket behind. Opening the socket
|
||||
// replaces it, but a daemon whose extensions never use the API would
|
||||
// otherwise keep the dead socket around.
|
||||
if err := os.Remove(extensionAPISocket(&cfg)); err != nil && !errors.Is(err, fs.ErrNotExist) {
|
||||
log.G(ctx).WithError(err).Warn("failed to remove stale extension API socket")
|
||||
}
|
||||
defaults := slices.Concat(builtinExtensions, daemon.daemonExtensions())
|
||||
exts, err := selectBuiltinExtensions(cfg.Extensions, defaults, daemon.optionalExtensions())
|
||||
if err != nil {
|
||||
@@ -46,6 +55,37 @@ func (daemon *Daemon) newExtensionHost(ctx context.Context) (*host.Host, error)
|
||||
)
|
||||
}
|
||||
|
||||
// extensionRuntimeDir holds the extension sockets.
|
||||
// It is private to the daemon user.
|
||||
func extensionRuntimeDir(cfg *config.Config) string {
|
||||
return filepath.Join(cfg.ExecRoot, "extensions")
|
||||
}
|
||||
|
||||
// extensionAPISocket is the path of the API socket for extensions.
|
||||
func extensionAPISocket(cfg *config.Config) string {
|
||||
return filepath.Join(extensionRuntimeDir(cfg), "api.sock")
|
||||
}
|
||||
|
||||
// extensionAPIEndpoint provides the daemon.api point.
|
||||
// It opens the API socket for extensions when an extension first asks for the
|
||||
// endpoint, so daemons without such extensions expose no extra API endpoint.
|
||||
// The socket is separate from the configured hosts, which may be TLS-only TCP
|
||||
// or socket-activated and so unusable by extensions.
|
||||
func (daemon *Daemon) extensionAPIEndpoint(ctx context.Context) (string, error) {
|
||||
cfg := daemon.Config()
|
||||
path := extensionAPISocket(&cfg)
|
||||
if err := daemon.apiSockets.Open(ctx, path); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "unix://" + path, nil
|
||||
}
|
||||
|
||||
// StartAPISockets serves h on the API sockets the daemon creates for its own
|
||||
// consumers, such as extensions.
|
||||
func (daemon *Daemon) StartAPISockets(h http.Handler) {
|
||||
daemon.apiSockets.Start(h)
|
||||
}
|
||||
|
||||
// extensionConfig converts daemon.json's extensions.configs entries to host form.
|
||||
func extensionConfig(cfg *config.Config) map[extensions.ExtensionID]extensions.Config {
|
||||
if len(cfg.Extensions.Configs) == 0 {
|
||||
|
||||
@@ -10,11 +10,13 @@ import (
|
||||
"github.com/moby/extensions/clientpoint"
|
||||
"github.com/moby/extensions/serverpoint"
|
||||
"github.com/moby/moby/v2/daemon/config"
|
||||
"github.com/moby/moby/v2/daemon/internal/extensionapi"
|
||||
"github.com/moby/moby/v2/daemon/internal/extensionconfig"
|
||||
"github.com/moby/moby/v2/daemon/internal/extensionstorage/bbolt"
|
||||
|
||||
containernamegeneratorpb "github.com/moby/moby/v2/extpoints/containernamegenerator/v0/protogen"
|
||||
containerstatepb "github.com/moby/moby/v2/extpoints/containerstate/v0/protogen"
|
||||
daemonapipb "github.com/moby/moby/v2/extpoints/daemonapi/v0/protogen"
|
||||
daemonconfigpb "github.com/moby/moby/v2/extpoints/daemonconfig/v0/protogen"
|
||||
servicenamegeneratorpb "github.com/moby/moby/v2/extpoints/servicenamegenerator/v0/protogen"
|
||||
storagepb "github.com/moby/moby/v2/extpoints/storage/kv/v0/protogen"
|
||||
@@ -41,6 +43,7 @@ func (daemon *Daemon) daemonExtensions() []extensions.Extension {
|
||||
return []extensions.Extension{
|
||||
extensionconfig.New(&cfg),
|
||||
newMonolithExtension(daemon),
|
||||
extensionapi.New(daemon.extensionAPIEndpoint),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -96,5 +99,6 @@ func dependencyProviders() []serverpoint.Registration {
|
||||
storagepb.ServerPoint,
|
||||
daemonconfigpb.ServerPoint,
|
||||
containerstatepb.ServerPoint,
|
||||
daemonapipb.ServerPoint,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
//go:build !windows
|
||||
|
||||
package apisocket
|
||||
|
||||
import (
|
||||
"net"
|
||||
|
||||
"github.com/docker/go-connections/sockets"
|
||||
)
|
||||
|
||||
func listen(path string) (net.Listener, error) {
|
||||
return sockets.NewUnixSocketWithOpts(path, sockets.WithChmod(0o600))
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package apisocket
|
||||
|
||||
import (
|
||||
"net"
|
||||
|
||||
"github.com/docker/go-connections/sockets"
|
||||
)
|
||||
|
||||
func listen(path string) (net.Listener, error) {
|
||||
return sockets.NewUnixSocketWithOpts(path, sockets.WithBasePermissions())
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
// Package apisocket serves the Engine API on unix sockets that the daemon
|
||||
// creates for its own consumers, separately from the configured hosts.
|
||||
package apisocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/containerd/log"
|
||||
)
|
||||
|
||||
// Manager owns the daemon-created API sockets and serves them all with one
|
||||
// server.
|
||||
type Manager struct {
|
||||
handler atomic.Pointer[http.Handler]
|
||||
srv *http.Server
|
||||
|
||||
mu sync.Mutex
|
||||
sockets map[string]*listener
|
||||
closed bool
|
||||
}
|
||||
|
||||
// NewManager returns a Manager with no sockets.
|
||||
func NewManager() *Manager {
|
||||
m := &Manager{sockets: make(map[string]*listener)}
|
||||
|
||||
var p http.Protocols
|
||||
p.SetHTTP1(true)
|
||||
p.SetUnencryptedHTTP2(true)
|
||||
m.srv = &http.Server{
|
||||
Handler: http.HandlerFunc(m.serveHTTP),
|
||||
Protocols: &p,
|
||||
ReadHeaderTimeout: 5 * time.Minute, // Same as the configured hosts.
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// Open creates a unix socket at path and serves the API on it.
|
||||
// On Unix, the socket is accessible only to its owner.
|
||||
// Until [Manager.Start], requests fail with 503 Service Unavailable, so a
|
||||
// socket can be handed out before the API is ready.
|
||||
// Opening a path that is already open does nothing.
|
||||
func (m *Manager) Open(ctx context.Context, path string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.closed {
|
||||
return errors.New("API socket manager is closed")
|
||||
}
|
||||
if _, ok := m.sockets[path]; ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
ls, err := listen(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listen on API socket %q: %w", path, err)
|
||||
}
|
||||
ul, ok := ls.(*net.UnixListener)
|
||||
if !ok {
|
||||
_ = ls.Close()
|
||||
return fmt.Errorf("listen on API socket %q: unexpected listener type %T", path, ls)
|
||||
}
|
||||
|
||||
l := &listener{UnixListener: ul, conns: make(map[*conn]struct{})}
|
||||
m.sockets[path] = l
|
||||
|
||||
log.G(ctx).Infof("API listen on %s", path)
|
||||
go func() {
|
||||
if err := m.srv.Serve(l); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
log.G(ctx).WithFields(log.Fields{
|
||||
"error": err,
|
||||
"listener": path,
|
||||
}).Error("ServeAPI error")
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Start serves h on all sockets, including ones opened later.
|
||||
func (m *Manager) Start(h http.Handler) {
|
||||
m.handler.Store(&h)
|
||||
}
|
||||
|
||||
func (m *Manager) serveHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
h := m.handler.Load()
|
||||
if h == nil {
|
||||
http.Error(w, "the daemon is not ready to serve API requests", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
(*h).ServeHTTP(w, r)
|
||||
}
|
||||
|
||||
// Close stops serving all sockets, closes all their connections, including
|
||||
// hijacked ones, and removes the socket files.
|
||||
// Later calls to [Manager.Open] fail.
|
||||
func (m *Manager) Close() error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.closed = true
|
||||
err := m.srv.Close()
|
||||
for _, l := range m.sockets {
|
||||
// Serve may not track the listener yet; closing it here removes the
|
||||
// socket file either way.
|
||||
_ = l.Close()
|
||||
l.closeConns()
|
||||
}
|
||||
clear(m.sockets)
|
||||
return err
|
||||
}
|
||||
|
||||
// listener tracks its connections, because the server does not track
|
||||
// hijacked connections such as attach and exec streams.
|
||||
type listener struct {
|
||||
*net.UnixListener
|
||||
|
||||
mu sync.Mutex
|
||||
conns map[*conn]struct{}
|
||||
}
|
||||
|
||||
func (l *listener) Accept() (net.Conn, error) {
|
||||
uc, err := l.AcceptUnix()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c := &conn{UnixConn: uc, l: l}
|
||||
l.mu.Lock()
|
||||
l.conns[c] = struct{}{}
|
||||
l.mu.Unlock()
|
||||
return c, nil
|
||||
}
|
||||
|
||||
func (l *listener) closeConns() {
|
||||
l.mu.Lock()
|
||||
conns := make([]*conn, 0, len(l.conns))
|
||||
for c := range l.conns {
|
||||
conns = append(conns, c)
|
||||
}
|
||||
l.mu.Unlock()
|
||||
for _, c := range conns {
|
||||
_ = c.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// conn embeds *net.UnixConn so that CloseWrite stays available for
|
||||
// half-closing hijacked streams.
|
||||
type conn struct {
|
||||
*net.UnixConn
|
||||
l *listener
|
||||
}
|
||||
|
||||
func (c *conn) Close() error {
|
||||
c.l.mu.Lock()
|
||||
delete(c.l.conns, c)
|
||||
c.l.mu.Unlock()
|
||||
return c.UnixConn.Close()
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package apisocket
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/moby/sys/reexec"
|
||||
"gotest.tools/v3/assert"
|
||||
"gotest.tools/v3/skip"
|
||||
)
|
||||
|
||||
const (
|
||||
socketConnectCommand = "apisocket-test-connect"
|
||||
unprivilegedUID = 65534
|
||||
)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
reexec.Register(socketConnectCommand, func() {
|
||||
if uid := os.Geteuid(); uid != unprivilegedUID {
|
||||
fmt.Fprintf(os.Stderr, "connection probe has unexpected UID %d\n", uid)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
c, err := net.DialTimeout("unix", os.Args[1], 5*time.Second) // #nosec G704 -- reexec dials a test-owned Unix socket.
|
||||
if c != nil {
|
||||
defer c.Close()
|
||||
}
|
||||
if os.Args[2] == "true" && err == nil {
|
||||
return
|
||||
}
|
||||
if os.Args[2] == "false" && errors.Is(err, os.ErrPermission) {
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "unexpected connection result (allow %s): %v\n", os.Args[2], err)
|
||||
os.Exit(1)
|
||||
})
|
||||
if reexec.Init() {
|
||||
return
|
||||
}
|
||||
os.Exit(m.Run())
|
||||
}
|
||||
|
||||
func TestManagerRejectsOtherUsers(t *testing.T) {
|
||||
t.Parallel()
|
||||
skip.If(t, os.Geteuid() != 0, "requires root to connect as another user")
|
||||
|
||||
// Keep every parent traversable so only the socket can deny access.
|
||||
dir, err := os.MkdirTemp("/tmp", "apisocket")
|
||||
assert.NilError(t, err)
|
||||
t.Cleanup(func() { _ = os.RemoveAll(dir) })
|
||||
assert.NilError(t, os.Chmod(dir, 0o755))
|
||||
|
||||
path := filepath.Join(dir, "api.sock")
|
||||
|
||||
m := NewManager()
|
||||
|
||||
t.Cleanup(func() { assert.NilError(t, m.Close()) })
|
||||
assert.NilError(t, m.Open(t.Context(), path))
|
||||
|
||||
checkConnection := func(gid uint32, allow bool) {
|
||||
t.Helper()
|
||||
|
||||
// Keep the creating thread alive for reexec's parent-death signal.
|
||||
runtime.LockOSThread()
|
||||
defer runtime.UnlockOSThread()
|
||||
|
||||
cmd := reexec.Command(socketConnectCommand, path, strconv.FormatBool(allow))
|
||||
cmd.SysProcAttr.Credential = &syscall.Credential{Uid: unprivilegedUID, Gid: gid}
|
||||
out, err := cmd.CombinedOutput()
|
||||
assert.NilError(t, err, "connection probe with GID %d (allow %t): %s", gid, allow, out)
|
||||
}
|
||||
|
||||
sameGroup := uint32(os.Getegid())
|
||||
checkConnection(sameGroup, false)
|
||||
checkConnection(sameGroup+1, false)
|
||||
|
||||
// Prove both probes can reach the listener when socket access is allowed.
|
||||
assert.NilError(t, os.Chmod(path, 0o666))
|
||||
checkConnection(sameGroup, true)
|
||||
checkConnection(sameGroup+1, true)
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package apisocket
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"io/fs"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gotest.tools/v3/assert"
|
||||
is "gotest.tools/v3/assert/cmp"
|
||||
)
|
||||
|
||||
func TestManagerLifecycle(t *testing.T) {
|
||||
// t.TempDir paths can exceed the unix socket path limit on macOS.
|
||||
dir, err := os.MkdirTemp("", "apisocket")
|
||||
assert.NilError(t, err)
|
||||
t.Cleanup(func() { _ = os.RemoveAll(dir) })
|
||||
path := filepath.Join(dir, "api.sock")
|
||||
ctx := t.Context()
|
||||
|
||||
m := NewManager()
|
||||
assert.NilError(t, m.Open(ctx, path))
|
||||
assert.NilError(t, m.Open(ctx, path), "opening an open socket again must be a no-op")
|
||||
|
||||
c := &http.Client{Transport: &http.Transport{
|
||||
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
return (&net.Dialer{}).DialContext(ctx, "unix", path)
|
||||
},
|
||||
}}
|
||||
t.Cleanup(c.CloseIdleConnections)
|
||||
status := func() int {
|
||||
t.Helper()
|
||||
resp, err := c.Get("http://api/_ping")
|
||||
assert.NilError(t, err)
|
||||
_ = resp.Body.Close()
|
||||
return resp.StatusCode
|
||||
}
|
||||
|
||||
assert.Check(t, is.Equal(status(), http.StatusServiceUnavailable), "requests before Start must fail")
|
||||
|
||||
m.Start(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/hijack" {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
// Keep the hijacked connection open, like an attach stream.
|
||||
hc, _, err := http.NewResponseController(w).Hijack()
|
||||
if err == nil {
|
||||
_, _ = hc.Write([]byte("hijacked\n"))
|
||||
}
|
||||
}))
|
||||
assert.Check(t, is.Equal(status(), http.StatusNoContent))
|
||||
|
||||
hijacked, err := net.Dial("unix", path)
|
||||
assert.NilError(t, err)
|
||||
defer hijacked.Close()
|
||||
_, err = hijacked.Write([]byte("GET /hijack HTTP/1.1\r\nHost: api\r\n\r\n"))
|
||||
assert.NilError(t, err)
|
||||
br := bufio.NewReader(hijacked)
|
||||
line, err := br.ReadString('\n')
|
||||
assert.NilError(t, err)
|
||||
assert.Check(t, is.Equal(line, "hijacked\n"))
|
||||
|
||||
assert.NilError(t, m.Close())
|
||||
_, err = os.Stat(path)
|
||||
assert.Check(t, errors.Is(err, fs.ErrNotExist), "Close must remove the socket file")
|
||||
assert.Check(t, m.Open(ctx, path) != nil, "Open after Close must fail")
|
||||
|
||||
assert.NilError(t, hijacked.SetReadDeadline(time.Now().Add(10*time.Second)))
|
||||
_, err = br.ReadByte()
|
||||
assert.Check(t, is.ErrorIs(err, io.EOF), "Close must close hijacked connections")
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
// Package extensionapi provides the daemon API endpoint to extensions.
|
||||
//
|
||||
// The endpoint is a temporary shortcut until extensions can depend on specific
|
||||
// extension points for the daemon functionality they need.
|
||||
// Give new functionality for extensions its own extension point instead of
|
||||
// relying on this endpoint.
|
||||
package extensionapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/moby/extensions"
|
||||
daemonapiv0 "github.com/moby/moby/v2/extpoints/daemonapi/v0"
|
||||
)
|
||||
|
||||
// New declares a provider for the API endpoint returned by host.
|
||||
// host is called each time an extension resolves the endpoint.
|
||||
func New(host func(context.Context) (string, error)) extensions.Extension {
|
||||
return extensions.New(extensions.Declaration{
|
||||
ID: "org.mobyproject.daemon.api.v0",
|
||||
Providers: []extensions.Provider{
|
||||
daemonapiv0.Point.Provide(endpoint{host: host}),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
type endpoint struct {
|
||||
host func(context.Context) (string, error)
|
||||
}
|
||||
|
||||
func (e endpoint) Get(ctx context.Context, _ *daemonapiv0.GetRequest) (*daemonapiv0.GetResponse, error) {
|
||||
host, err := e.host(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &daemonapiv0.GetResponse{Host: host}, nil
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package daemonapiv0
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
|
||||
"github.com/moby/moby/client"
|
||||
)
|
||||
|
||||
// placeholderHost only selects the plain unix transport; the client dials the
|
||||
// endpoint the provider reports.
|
||||
// The client includes its host in connection errors, so the name points at
|
||||
// where the real endpoint comes from rather than looking like a real path.
|
||||
const placeholderHost = "unix:///resolved-by-daemonapi-v0.sock"
|
||||
|
||||
// NewClient returns an Engine API client for the endpoint api reports.
|
||||
// It is a temporary shortcut; prefer a specific extension point when one
|
||||
// covers what the extension needs.
|
||||
// Resolve api with Point.Single during initialization, after declaring
|
||||
// Point.Dependency().
|
||||
// The client asks api for the endpoint when it first connects, so the daemon
|
||||
// creates the socket only once an extension actually uses the API.
|
||||
// opts are applied after the endpoint, so they may override it.
|
||||
//
|
||||
// The daemon serves the API only after it finishes initializing, so API calls
|
||||
// made before then, including from Init, fail.
|
||||
// The caller owns the returned client and must close it.
|
||||
func NewClient(api DaemonAPI, opts ...client.Opt) (*client.Client, error) {
|
||||
d := &dialer{api: api, lock: make(chan struct{}, 1)}
|
||||
return client.New(append([]client.Opt{
|
||||
client.WithHost(placeholderHost),
|
||||
client.WithDialContext(d.dial),
|
||||
}, opts...)...)
|
||||
}
|
||||
|
||||
type dialer struct {
|
||||
api DaemonAPI
|
||||
|
||||
// lock is a one-slot semaphore rather than a mutex so that dials waiting
|
||||
// on a slow Get still honor their own context.
|
||||
lock chan struct{}
|
||||
path string
|
||||
}
|
||||
|
||||
func (d *dialer) dial(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
path, err := d.socketPath(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var nd net.Dialer
|
||||
return nd.DialContext(ctx, "unix", path)
|
||||
}
|
||||
|
||||
func (d *dialer) socketPath(ctx context.Context) (string, error) {
|
||||
select {
|
||||
case d.lock <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return "", ctx.Err()
|
||||
}
|
||||
defer func() { <-d.lock }()
|
||||
if d.path != "" {
|
||||
return d.path, nil
|
||||
}
|
||||
resp, err := d.api.Get(ctx, &GetRequest{})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("get daemon API endpoint: %w", err)
|
||||
}
|
||||
path, ok := strings.CutPrefix(resp.Host, "unix://")
|
||||
if !ok {
|
||||
return "", fmt.Errorf("unsupported daemon API endpoint %q", resp.Host)
|
||||
}
|
||||
d.path = path
|
||||
return path, nil
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package daemonapiv0_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"github.com/moby/moby/client"
|
||||
daemonapiv0 "github.com/moby/moby/v2/extpoints/daemonapi/v0"
|
||||
"gotest.tools/v3/assert"
|
||||
is "gotest.tools/v3/assert/cmp"
|
||||
)
|
||||
|
||||
type fakeAPI struct {
|
||||
host string
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (f *fakeAPI) Get(context.Context, *daemonapiv0.GetRequest) (*daemonapiv0.GetResponse, error) {
|
||||
f.calls.Add(1)
|
||||
return &daemonapiv0.GetResponse{Host: f.host}, nil
|
||||
}
|
||||
|
||||
func TestNewClientGetsEndpointOnFirstConnect(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// t.TempDir paths can exceed the unix socket path limit on macOS.
|
||||
dir, err := os.MkdirTemp("", "daemonapi")
|
||||
assert.NilError(t, err)
|
||||
t.Cleanup(func() { _ = os.RemoveAll(dir) })
|
||||
|
||||
sock := filepath.Join(dir, "api.sock")
|
||||
l, err := net.Listen("unix", sock)
|
||||
assert.NilError(t, err)
|
||||
srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
// Force a new connection per request so the second Ping dials again.
|
||||
w.Header().Set("Connection", "close")
|
||||
}))
|
||||
_ = srv.Listener.Close()
|
||||
srv.Listener = l
|
||||
srv.Start()
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
api := &fakeAPI{host: "unix://" + sock}
|
||||
c, err := daemonapiv0.NewClient(api)
|
||||
assert.NilError(t, err)
|
||||
t.Cleanup(func() { _ = c.Close() })
|
||||
assert.Check(t, is.Equal(api.calls.Load(), int32(0)))
|
||||
|
||||
for range 2 {
|
||||
_, err = c.Ping(t.Context(), client.PingOptions{})
|
||||
assert.NilError(t, err)
|
||||
}
|
||||
assert.Check(t, is.Equal(api.calls.Load(), int32(1)))
|
||||
}
|
||||
Generated
+21
@@ -0,0 +1,21 @@
|
||||
// Code generated by mobyextgen. DO NOT EDIT.
|
||||
|
||||
syntax = "proto3";
|
||||
|
||||
package org.mobyproject.extension.daemon.api.v0;
|
||||
|
||||
option go_package = "github.com/moby/moby/v2/extpoints/daemonapi/v0/protogen";
|
||||
|
||||
// DaemonAPI locates the Engine API endpoint the daemon serves for extensions.
|
||||
// Use [NewClient] instead of calling it directly.
|
||||
service DaemonAPI {
|
||||
// Get returns the endpoint.
|
||||
rpc Get(GetRequest) returns (GetResponse);
|
||||
}
|
||||
|
||||
message GetRequest {
|
||||
}
|
||||
|
||||
message GetResponse {
|
||||
string host = 1;
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
//go:generate go tool mobyextgen
|
||||
|
||||
// Package daemonapiv0 gives Moby extensions access to the daemon's Engine API
|
||||
// through a regular Moby API client.
|
||||
//
|
||||
// This is a temporary shortcut until Moby defines specific extension points
|
||||
// for the daemon functionality extensions need.
|
||||
// Once a specific point covers what an extension uses the API for, the
|
||||
// extension should depend on that point instead.
|
||||
package daemonapiv0
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/moby/extensions"
|
||||
)
|
||||
|
||||
// DaemonAPI locates the Engine API endpoint the daemon serves for extensions.
|
||||
// Use [NewClient] instead of calling it directly.
|
||||
type DaemonAPI interface {
|
||||
// Get returns the endpoint.
|
||||
Get(ctx context.Context, req *GetRequest) (*GetResponse, error)
|
||||
}
|
||||
|
||||
// Point is single-cardinality for the daemon API endpoint.
|
||||
// It is a temporary shortcut; prefer a specific point when one exists.
|
||||
// The daemon registers its builtin provider for extension dependencies,
|
||||
// including out-of-process extensions, but does not publish it on the public
|
||||
// daemon socket.
|
||||
//
|
||||
// The endpoint is a unix socket that only the daemon user can access; on
|
||||
// Windows, only Administrators and LocalSystem.
|
||||
// The daemon creates it on the first Get call, which a [NewClient] client
|
||||
// makes when it first connects.
|
||||
// Extensions run as the daemon user, so every extension can reach the socket
|
||||
// and gets unrestricted Engine API access, whether or not it declares a
|
||||
// dependency on this point.
|
||||
var Point = extensions.DefineSinglePoint[DaemonAPI]("org.mobyproject.extension.daemon.api.v0")
|
||||
|
||||
// GetRequest requests the daemon API endpoint.
|
||||
type GetRequest struct{}
|
||||
|
||||
// GetResponse contains the daemon API endpoint.
|
||||
type GetResponse struct {
|
||||
// Host is the endpoint in DOCKER_HOST form, such as unix:///path/api.sock.
|
||||
Host string `pb:"1"`
|
||||
}
|
||||
+165
@@ -0,0 +1,165 @@
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.36.12
|
||||
// protoc (unknown)
|
||||
// source: extpoints/daemonapi/v0/daemon_api.proto
|
||||
|
||||
package protogen
|
||||
|
||||
import (
|
||||
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
|
||||
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
|
||||
reflect "reflect"
|
||||
sync "sync"
|
||||
unsafe "unsafe"
|
||||
)
|
||||
|
||||
const (
|
||||
// Verify that this generated code is sufficiently up-to-date.
|
||||
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
|
||||
// Verify that runtime/protoimpl is sufficiently up-to-date.
|
||||
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
|
||||
)
|
||||
|
||||
type GetRequest struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *GetRequest) Reset() {
|
||||
*x = GetRequest{}
|
||||
mi := &file_extpoints_daemonapi_v0_daemon_api_proto_msgTypes[0]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *GetRequest) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*GetRequest) ProtoMessage() {}
|
||||
|
||||
func (x *GetRequest) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_extpoints_daemonapi_v0_daemon_api_proto_msgTypes[0]
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use GetRequest.ProtoReflect.Descriptor instead.
|
||||
func (*GetRequest) Descriptor() ([]byte, []int) {
|
||||
return file_extpoints_daemonapi_v0_daemon_api_proto_rawDescGZIP(), []int{0}
|
||||
}
|
||||
|
||||
type GetResponse struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *GetResponse) Reset() {
|
||||
*x = GetResponse{}
|
||||
mi := &file_extpoints_daemonapi_v0_daemon_api_proto_msgTypes[1]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *GetResponse) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*GetResponse) ProtoMessage() {}
|
||||
|
||||
func (x *GetResponse) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_extpoints_daemonapi_v0_daemon_api_proto_msgTypes[1]
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use GetResponse.ProtoReflect.Descriptor instead.
|
||||
func (*GetResponse) Descriptor() ([]byte, []int) {
|
||||
return file_extpoints_daemonapi_v0_daemon_api_proto_rawDescGZIP(), []int{1}
|
||||
}
|
||||
|
||||
func (x *GetResponse) GetHost() string {
|
||||
if x != nil {
|
||||
return x.Host
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
var File_extpoints_daemonapi_v0_daemon_api_proto protoreflect.FileDescriptor
|
||||
|
||||
const file_extpoints_daemonapi_v0_daemon_api_proto_rawDesc = "" +
|
||||
"\n" +
|
||||
"'extpoints/daemonapi/v0/daemon_api.proto\x12'org.mobyproject.extension.daemon.api.v0\"\f\n" +
|
||||
"\n" +
|
||||
"GetRequest\"!\n" +
|
||||
"\vGetResponse\x12\x12\n" +
|
||||
"\x04host\x18\x01 \x01(\tR\x04host2}\n" +
|
||||
"\tDaemonAPI\x12p\n" +
|
||||
"\x03Get\x123.org.mobyproject.extension.daemon.api.v0.GetRequest\x1a4.org.mobyproject.extension.daemon.api.v0.GetResponseB9Z7github.com/moby/moby/v2/extpoints/daemonapi/v0/protogenb\x06proto3"
|
||||
|
||||
var (
|
||||
file_extpoints_daemonapi_v0_daemon_api_proto_rawDescOnce sync.Once
|
||||
file_extpoints_daemonapi_v0_daemon_api_proto_rawDescData []byte
|
||||
)
|
||||
|
||||
func file_extpoints_daemonapi_v0_daemon_api_proto_rawDescGZIP() []byte {
|
||||
file_extpoints_daemonapi_v0_daemon_api_proto_rawDescOnce.Do(func() {
|
||||
file_extpoints_daemonapi_v0_daemon_api_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_extpoints_daemonapi_v0_daemon_api_proto_rawDesc), len(file_extpoints_daemonapi_v0_daemon_api_proto_rawDesc)))
|
||||
})
|
||||
return file_extpoints_daemonapi_v0_daemon_api_proto_rawDescData
|
||||
}
|
||||
|
||||
var file_extpoints_daemonapi_v0_daemon_api_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
|
||||
var file_extpoints_daemonapi_v0_daemon_api_proto_goTypes = []any{
|
||||
(*GetRequest)(nil), // 0: org.mobyproject.extension.daemon.api.v0.GetRequest
|
||||
(*GetResponse)(nil), // 1: org.mobyproject.extension.daemon.api.v0.GetResponse
|
||||
}
|
||||
var file_extpoints_daemonapi_v0_daemon_api_proto_depIdxs = []int32{
|
||||
0, // 0: org.mobyproject.extension.daemon.api.v0.DaemonAPI.Get:input_type -> org.mobyproject.extension.daemon.api.v0.GetRequest
|
||||
1, // 1: org.mobyproject.extension.daemon.api.v0.DaemonAPI.Get:output_type -> org.mobyproject.extension.daemon.api.v0.GetResponse
|
||||
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
|
||||
}
|
||||
|
||||
func init() { file_extpoints_daemonapi_v0_daemon_api_proto_init() }
|
||||
func file_extpoints_daemonapi_v0_daemon_api_proto_init() {
|
||||
if File_extpoints_daemonapi_v0_daemon_api_proto != nil {
|
||||
return
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
GoPackagePath: reflect.TypeFor[x]().PkgPath(),
|
||||
RawDescriptor: unsafe.Slice(unsafe.StringData(file_extpoints_daemonapi_v0_daemon_api_proto_rawDesc), len(file_extpoints_daemonapi_v0_daemon_api_proto_rawDesc)),
|
||||
NumEnums: 0,
|
||||
NumMessages: 2,
|
||||
NumExtensions: 0,
|
||||
NumServices: 1,
|
||||
},
|
||||
GoTypes: file_extpoints_daemonapi_v0_daemon_api_proto_goTypes,
|
||||
DependencyIndexes: file_extpoints_daemonapi_v0_daemon_api_proto_depIdxs,
|
||||
MessageInfos: file_extpoints_daemonapi_v0_daemon_api_proto_msgTypes,
|
||||
}.Build()
|
||||
File_extpoints_daemonapi_v0_daemon_api_proto = out.File
|
||||
file_extpoints_daemonapi_v0_daemon_api_proto_goTypes = nil
|
||||
file_extpoints_daemonapi_v0_daemon_api_proto_depIdxs = nil
|
||||
}
|
||||
+156
@@ -0,0 +1,156 @@
|
||||
// Code generated by mobyextgen. DO NOT EDIT.
|
||||
package protogen
|
||||
|
||||
import (
|
||||
context "context"
|
||||
extensions "github.com/moby/extensions"
|
||||
clientpoint "github.com/moby/extensions/clientpoint"
|
||||
serverpoint "github.com/moby/extensions/serverpoint"
|
||||
mobyextcontract "github.com/moby/moby/v2/extpoints/daemonapi/v0"
|
||||
grpc "google.golang.org/grpc"
|
||||
)
|
||||
|
||||
// serviceName is the point's fully-qualified gRPC service name.
|
||||
const serviceName = "org.mobyproject.extension.daemon.api.v0.DaemonAPI"
|
||||
|
||||
const (
|
||||
methodGet = "/" + serviceName + "/Get"
|
||||
)
|
||||
|
||||
// DaemonAPIServer is the server side of the point's gRPC service. It is the
|
||||
// proto-level shape of the point, not the point's Go interface: a contract
|
||||
// method returning a bare error returns an empty response message here.
|
||||
type DaemonAPIServer interface {
|
||||
// Get returns the endpoint.
|
||||
Get(context.Context, *GetRequest) (*GetResponse, error)
|
||||
}
|
||||
|
||||
// serviceDesc describes the point's gRPC service to a server. HandlerType is
|
||||
// what a registrar type-checks an implementation against, so registering the
|
||||
// wrong provider for this point is caught at registration.
|
||||
var serviceDesc = grpc.ServiceDesc{
|
||||
ServiceName: serviceName,
|
||||
HandlerType: (*DaemonAPIServer)(nil),
|
||||
Metadata: "extpoints/daemonapi/v0/daemon_api.proto",
|
||||
Methods: []grpc.MethodDesc{
|
||||
{MethodName: "Get", Handler: handleGet},
|
||||
},
|
||||
}
|
||||
|
||||
func handleGet(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) {
|
||||
in := new(GetRequest)
|
||||
if err := dec(in); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if interceptor == nil {
|
||||
return srv.(DaemonAPIServer).Get(ctx, in)
|
||||
}
|
||||
info := &grpc.UnaryServerInfo{Server: srv, FullMethod: methodGet}
|
||||
return interceptor(ctx, in, info, func(ctx context.Context, req any) (any, error) {
|
||||
return srv.(DaemonAPIServer).Get(ctx, req.(*GetRequest))
|
||||
})
|
||||
}
|
||||
|
||||
// DaemonAPIClient calls the point's gRPC service. It is exported so a client
|
||||
// outside the framework - one calling a service an extension publishes on the
|
||||
// API socket - can reach it with a plain gRPC client.
|
||||
type DaemonAPIClient interface {
|
||||
// Get returns the endpoint.
|
||||
Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (*GetResponse, error)
|
||||
}
|
||||
|
||||
// NewDaemonAPIClient returns a client for the point's gRPC service on cc.
|
||||
func NewDaemonAPIClient(cc grpc.ClientConnInterface) DaemonAPIClient { return &serviceClient{cc: cc} }
|
||||
|
||||
type serviceClient struct{ cc grpc.ClientConnInterface }
|
||||
|
||||
func (c *serviceClient) Get(ctx context.Context, in *GetRequest, opts ...grpc.CallOption) (*GetResponse, error) {
|
||||
out := new(GetResponse)
|
||||
if err := c.cc.Invoke(ctx, methodGet, in, out, append([]grpc.CallOption{grpc.StaticMethod()}, opts...)...); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ServerPoint serves the DaemonAPI point: it registers the point's gRPC service for
|
||||
// a provider with an SDK server. A binary passes it to (*sdk.Server).Register.
|
||||
var ServerPoint = serverpoint.Registration{
|
||||
Point: mobyextcontract.Point.ID(),
|
||||
Register: func(r grpc.ServiceRegistrar, impl any) {
|
||||
r.RegisterService(&serviceDesc, &grpcServer{impl: impl.(mobyextcontract.DaemonAPI)})
|
||||
},
|
||||
}
|
||||
|
||||
// ClientProvider builds a broker provider for the DaemonAPI point from an
|
||||
// out-of-process gRPC connection.
|
||||
func ClientProvider(conn grpc.ClientConnInterface) extensions.Provider {
|
||||
return mobyextcontract.Point.Provide(NewClient(conn))
|
||||
}
|
||||
|
||||
// ClientPoint registers ClientProvider for the DaemonAPI point with a host.
|
||||
// Single carries the contract's cardinality: the point admits one provider.
|
||||
var ClientPoint = clientpoint.Registration{Point: mobyextcontract.Point.ID(), Provider: ClientProvider, Single: true}
|
||||
|
||||
// NewClient returns a mobyextcontract.DaemonAPI that calls the DaemonAPI point over conn.
|
||||
func NewClient(conn grpc.ClientConnInterface) mobyextcontract.DaemonAPI {
|
||||
return &grpcClient{client: NewDaemonAPIClient(conn)}
|
||||
}
|
||||
|
||||
// grpcServer serves an implementation of the contract's Go interface.
|
||||
type grpcServer struct {
|
||||
impl mobyextcontract.DaemonAPI
|
||||
}
|
||||
|
||||
func (s *grpcServer) Get(ctx context.Context, req *GetRequest) (*GetResponse, error) {
|
||||
resp, err := s.impl.Get(ctx, getRequestFromProto(req))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return getResponseToProto(resp), nil
|
||||
}
|
||||
|
||||
type grpcClient struct {
|
||||
client DaemonAPIClient
|
||||
}
|
||||
|
||||
func (c *grpcClient) Get(ctx context.Context, req *mobyextcontract.GetRequest) (*mobyextcontract.GetResponse, error) {
|
||||
resp, err := c.client.Get(ctx, getRequestToProto(req))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return getResponseFromProto(resp), nil
|
||||
}
|
||||
|
||||
func getRequestToProto(in *mobyextcontract.GetRequest) *GetRequest {
|
||||
if in == nil {
|
||||
return nil
|
||||
}
|
||||
out := &GetRequest{}
|
||||
return out
|
||||
}
|
||||
|
||||
func getRequestFromProto(in *GetRequest) *mobyextcontract.GetRequest {
|
||||
if in == nil {
|
||||
return nil
|
||||
}
|
||||
out := &mobyextcontract.GetRequest{}
|
||||
return out
|
||||
}
|
||||
|
||||
func getResponseToProto(in *mobyextcontract.GetResponse) *GetResponse {
|
||||
if in == nil {
|
||||
return nil
|
||||
}
|
||||
out := &GetResponse{}
|
||||
out.Host = in.Host
|
||||
return out
|
||||
}
|
||||
|
||||
func getResponseFromProto(in *GetResponse) *mobyextcontract.GetResponse {
|
||||
if in == nil {
|
||||
return nil
|
||||
}
|
||||
out := &mobyextcontract.GetResponse{}
|
||||
out.Host = in.GetHost()
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package extension
|
||||
|
||||
import (
|
||||
"runtime"
|
||||
"testing"
|
||||
|
||||
extensionclient "github.com/moby/extensions/client"
|
||||
greeterv0 "github.com/moby/extensions/example/greeter/v0"
|
||||
greeterpb "github.com/moby/extensions/example/greeter/v0/protogen"
|
||||
"github.com/moby/moby/client"
|
||||
"github.com/moby/moby/v2/integration/extension/testdata/apiclient"
|
||||
"github.com/moby/moby/v2/internal/testutil"
|
||||
"github.com/moby/moby/v2/internal/testutil/daemon"
|
||||
"gotest.tools/v3/assert"
|
||||
"gotest.tools/v3/skip"
|
||||
)
|
||||
|
||||
func TestExtensionDaemonAPIClient(t *testing.T) {
|
||||
skip.If(t, testEnv.IsRemoteDaemon, "the extension binary must be on the daemon's host")
|
||||
|
||||
ctx := testutil.StartSpan(baseContext, t)
|
||||
|
||||
extDir := buildExtension(ctx, t, apiclient.ID, "./testdata/apiclient/cmd/apiclient")
|
||||
startArgs := []string{"--extension-dir", extDir}
|
||||
if testEnv.DaemonInfo.OSType == "linux" {
|
||||
startArgs = append(startArgs, "--iptables=false", "--ip6tables=false")
|
||||
}
|
||||
|
||||
d := daemon.New(t)
|
||||
d.Start(t, startArgs...)
|
||||
defer func() {
|
||||
if runtime.GOOS == "windows" {
|
||||
assert.NilError(t, d.Kill())
|
||||
return
|
||||
}
|
||||
d.Stop(t)
|
||||
}()
|
||||
|
||||
engine := d.NewClientT(t)
|
||||
info, err := engine.Info(ctx, client.InfoOptions{})
|
||||
assert.NilError(t, err)
|
||||
|
||||
extensions, err := extensionclient.New(engine, extensionclient.WithGRPCPoint(greeterpb.ClientPoint))
|
||||
assert.NilError(t, err)
|
||||
defer func() { assert.NilError(t, extensions.Close()) }()
|
||||
|
||||
greeter, err := extensionclient.Resolve(extensions, greeterv0.Point)
|
||||
assert.NilError(t, err)
|
||||
resp, err := greeter.Greet(ctx, &greeterv0.HelloRequest{})
|
||||
assert.NilError(t, err)
|
||||
assert.Equal(t, resp.Message, info.Info.ID)
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
// Package apiclient is the daemon API client integration-test fixture.
|
||||
package apiclient
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/moby/extensions"
|
||||
greeterv0 "github.com/moby/extensions/example/greeter/v0"
|
||||
greeterpb "github.com/moby/extensions/example/greeter/v0/protogen"
|
||||
servicev0 "github.com/moby/extensions/extpoints/service/v0"
|
||||
"github.com/moby/moby/client"
|
||||
daemonapiv0 "github.com/moby/moby/v2/extpoints/daemonapi/v0"
|
||||
servicegrpcv0 "github.com/moby/moby/v2/extpoints/servicegrpc/v0"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
// ID is the extension id and binary name.
|
||||
const ID = "org.mobyproject.example.apiclient.v1"
|
||||
|
||||
type extension struct {
|
||||
api *client.Client
|
||||
}
|
||||
|
||||
// Greet replies with the ID of the daemon reached through the API client.
|
||||
func (e *extension) Greet(ctx context.Context, _ *greeterv0.HelloRequest) (*greeterv0.HelloReply, error) {
|
||||
info, err := e.api.Info(ctx, client.InfoOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &greeterv0.HelloReply{Message: info.Info.ID}, nil
|
||||
}
|
||||
|
||||
func (e *extension) RegisterServices(r grpc.ServiceRegistrar) {
|
||||
greeterpb.ServerPoint.Register(r, e)
|
||||
}
|
||||
|
||||
func (e *extension) Declaration() extensions.Declaration {
|
||||
return extensions.Declaration{
|
||||
ID: ID,
|
||||
Providers: []extensions.Provider{
|
||||
servicegrpcv0.Point.Provide(e),
|
||||
servicev0.Offer(servicegrpcv0.Point),
|
||||
},
|
||||
Dependencies: []extensions.Dependency{
|
||||
daemonapiv0.Point.Dependency(),
|
||||
},
|
||||
Init: func(_ context.Context, _ extensions.Config, resolver extensions.Resolver) error {
|
||||
provider, err := daemonapiv0.Point.Single(resolver)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
api, err := daemonapiv0.NewClient(provider)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
e.api = api
|
||||
return nil
|
||||
},
|
||||
Shutdown: func(context.Context) error {
|
||||
return e.api.Close()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Extension publishes a greeter service backed by the daemon API.
|
||||
var Extension extensions.Extension = &extension{}
|
||||
@@ -0,0 +1,31 @@
|
||||
// Command apiclient wraps the daemon API client fixture for the integration test.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"github.com/moby/extensions/sdk"
|
||||
daemonapipb "github.com/moby/moby/v2/extpoints/daemonapi/v0/protogen"
|
||||
servicegrpcv0 "github.com/moby/moby/v2/extpoints/servicegrpc/v0"
|
||||
"github.com/moby/moby/v2/integration/extension/testdata/apiclient"
|
||||
)
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
srv := sdk.NewServer()
|
||||
if err := srv.Register(apiclient.Extension, servicegrpcv0.ServerPoint); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
srv.Depends(daemonapipb.ClientPoint)
|
||||
if err := srv.Listen(ctx); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user