From e8382ece650c55d1fb1e006614f799bc5dc252b7 Mon Sep 17 00:00:00 2001 From: Tibor Vass Date: Tue, 2 Apr 2019 04:08:16 +0000 Subject: [PATCH 1/2] api: add undocumented /grpc endpoint to talk to GRPC services Signed-off-by: Tibor Vass --- api/server/router/grpc/backend.go | 8 +++++ api/server/router/grpc/grpc.go | 37 ++++++++++++++++++++ api/server/router/grpc/grpc_routes.go | 45 +++++++++++++++++++++++++ client/hijack.go | 11 ++++++ client/interface.go | 2 +- client/session.go | 18 ---------- integration/build/build_session_test.go | 5 ++- 7 files changed, 106 insertions(+), 20 deletions(-) create mode 100644 api/server/router/grpc/backend.go create mode 100644 api/server/router/grpc/grpc.go create mode 100644 api/server/router/grpc/grpc_routes.go delete mode 100644 client/session.go diff --git a/api/server/router/grpc/backend.go b/api/server/router/grpc/backend.go new file mode 100644 index 0000000000..52cc37cf2c --- /dev/null +++ b/api/server/router/grpc/backend.go @@ -0,0 +1,8 @@ +package grpc // import "github.com/docker/docker/api/server/router/grpc" + +import "google.golang.org/grpc" + +// Backend abstracts a registerable GRPC service. +type Backend interface { + RegisterGRPC(*grpc.Server) +} diff --git a/api/server/router/grpc/grpc.go b/api/server/router/grpc/grpc.go new file mode 100644 index 0000000000..496c19e28a --- /dev/null +++ b/api/server/router/grpc/grpc.go @@ -0,0 +1,37 @@ +package grpc // import "github.com/docker/docker/api/server/router/grpc" + +import ( + "github.com/docker/docker/api/server/router" + "golang.org/x/net/http2" + "google.golang.org/grpc" +) + +type grpcRouter struct { + routes []router.Route + grpcServer *grpc.Server + h2Server *http2.Server +} + +// NewRouter initializes a new grpc http router +func NewRouter(backends ...Backend) router.Router { + r := &grpcRouter{ + h2Server: &http2.Server{}, + grpcServer: grpc.NewServer(), + } + for _, b := range backends { + b.RegisterGRPC(r.grpcServer) + } + r.initRoutes() + return r +} + +// Routes returns the available routers to the session controller +func (r *grpcRouter) Routes() []router.Route { + return r.routes +} + +func (r *grpcRouter) initRoutes() { + r.routes = []router.Route{ + router.NewPostRoute("/grpc", r.serveGRPC), + } +} diff --git a/api/server/router/grpc/grpc_routes.go b/api/server/router/grpc/grpc_routes.go new file mode 100644 index 0000000000..1fe0da472f --- /dev/null +++ b/api/server/router/grpc/grpc_routes.go @@ -0,0 +1,45 @@ +package grpc // import "github.com/docker/docker/api/server/router/grpc" + +import ( + "context" + "net/http" + + "github.com/pkg/errors" + "golang.org/x/net/http2" +) + +func (gr *grpcRouter) serveGRPC(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error { + h, ok := w.(http.Hijacker) + if !ok { + return errors.New("handler does not support hijack") + } + proto := r.Header.Get("Upgrade") + if proto == "" { + return errors.New("no upgrade proto in request") + } + if proto != "h2c" { + return errors.Errorf("protocol %s not supported", proto) + } + + conn, _, err := h.Hijack() + if err != nil { + return err + } + resp := &http.Response{ + StatusCode: http.StatusSwitchingProtocols, + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{}, + } + resp.Header.Set("Connection", "Upgrade") + resp.Header.Set("Upgrade", proto) + + // set raw mode + conn.Write([]byte{}) + resp.Write(conn) + + // https://godoc.org/golang.org/x/net/http2#Server.ServeConn + // TODO: is it a problem that conn has already been written to? + gr.h2Server.ServeConn(conn, &http2.ServeConnOpts{Handler: gr.grpcServer}) + return nil +} diff --git a/client/hijack.go b/client/hijack.go index 0ac8248f2c..8609982739 100644 --- a/client/hijack.go +++ b/client/hijack.go @@ -38,6 +38,17 @@ func (cli *Client) postHijacked(ctx context.Context, path string, query url.Valu return types.HijackedResponse{Conn: conn, Reader: bufio.NewReader(conn)}, err } +// DialHijack returns a hijacked connection with negotiated protocol proto. +func (cli *Client) DialHijack(ctx context.Context, url, proto string, meta map[string][]string) (net.Conn, error) { + req, err := http.NewRequest("POST", url, nil) + if err != nil { + return nil, err + } + req = cli.addHeaders(req, meta) + + return cli.setupHijackConn(ctx, req, proto) +} + // fallbackDial is used when WithDialer() was not called. // See cli.Dialer(). func fallbackDial(proto, addr string, tlsConfig *tls.Config) (net.Conn, error) { diff --git a/client/interface.go b/client/interface.go index d190f8e58d..cde64be4b5 100644 --- a/client/interface.go +++ b/client/interface.go @@ -38,7 +38,7 @@ type CommonAPIClient interface { ServerVersion(ctx context.Context) (types.Version, error) NegotiateAPIVersion(ctx context.Context) NegotiateAPIVersionPing(types.Ping) - DialSession(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error) + DialHijack(ctx context.Context, url, proto string, meta map[string][]string) (net.Conn, error) Dialer() func(context.Context) (net.Conn, error) Close() error } diff --git a/client/session.go b/client/session.go deleted file mode 100644 index df199f3d03..0000000000 --- a/client/session.go +++ /dev/null @@ -1,18 +0,0 @@ -package client // import "github.com/docker/docker/client" - -import ( - "context" - "net" - "net/http" -) - -// DialSession returns a connection that can be used communication with daemon -func (cli *Client) DialSession(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error) { - req, err := http.NewRequest("POST", "/session", nil) - if err != nil { - return nil, err - } - req = cli.addHeaders(req, meta) - - return cli.setupHijackConn(ctx, req, proto) -} diff --git a/integration/build/build_session_test.go b/integration/build/build_session_test.go index 4f7ea2b0f6..db5124f4c7 100644 --- a/integration/build/build_session_test.go +++ b/integration/build/build_session_test.go @@ -3,6 +3,7 @@ package build import ( "context" "io/ioutil" + "net" "net/http" "strings" "testing" @@ -109,7 +110,9 @@ func testBuildWithSession(t *testing.T, client dclient.APIClient, daemonHost str g, ctx := errgroup.WithContext(ctx) g.Go(func() error { - return sess.Run(ctx, client.DialSession) + return sess.Run(ctx, func(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error) { + return client.DialHijack(ctx, "/session", "h2c", meta) + }) }) g.Go(func() error { From 05c5d20a2cd8b2073901828b83deaa493220bce6 Mon Sep 17 00:00:00 2001 From: Tibor Vass Date: Tue, 2 Apr 2019 04:08:51 +0000 Subject: [PATCH 2/2] grpc: register BuildKit controller to /grpc Signed-off-by: Tibor Vass --- api/server/backend/build/backend.go | 8 ++++++++ builder/builder-next/builder.go | 6 ++++++ cmd/dockerd/daemon.go | 11 +++++++++++ 3 files changed, 25 insertions(+) diff --git a/api/server/backend/build/backend.go b/api/server/backend/build/backend.go index 00fb5b8d2c..ff7a5a9b42 100644 --- a/api/server/backend/build/backend.go +++ b/api/server/backend/build/backend.go @@ -14,6 +14,7 @@ import ( "github.com/docker/docker/pkg/stringid" "github.com/pkg/errors" "golang.org/x/sync/errgroup" + "google.golang.org/grpc" ) // ImageComponent provides an interface for working with images @@ -40,6 +41,13 @@ func NewBackend(components ImageComponent, builder Builder, fsCache *fscache.FSC return &Backend{imageComponent: components, builder: builder, fsCache: fsCache, buildkit: buildkit}, nil } +// RegisterGRPC registers buildkit controller to the grpc server. +func (b *Backend) RegisterGRPC(s *grpc.Server) { + if b.buildkit != nil { + b.buildkit.RegisterGRPC(s) + } +} + // Build builds an image from a Source func (b *Backend) Build(ctx context.Context, config backend.BuildConfig) (string, error) { options := config.Options diff --git a/builder/builder-next/builder.go b/builder/builder-next/builder.go index 859a65f7f1..c6041d2d34 100644 --- a/builder/builder-next/builder.go +++ b/builder/builder-next/builder.go @@ -31,6 +31,7 @@ import ( "github.com/moby/buildkit/util/tracing" "github.com/pkg/errors" "golang.org/x/sync/errgroup" + "google.golang.org/grpc" grpcmetadata "google.golang.org/grpc/metadata" ) @@ -104,6 +105,11 @@ func New(opt Opt) (*Builder, error) { return b, nil } +// RegisterGRPC registers controller to the grpc server. +func (b *Builder) RegisterGRPC(s *grpc.Server) { + b.controller.Register(s) +} + // Cancel cancels a build using ID func (b *Builder) Cancel(ctx context.Context, id string) error { b.mu.Lock() diff --git a/cmd/dockerd/daemon.go b/cmd/dockerd/daemon.go index caef9d1c57..9a795046b6 100644 --- a/cmd/dockerd/daemon.go +++ b/cmd/dockerd/daemon.go @@ -21,6 +21,7 @@ import ( checkpointrouter "github.com/docker/docker/api/server/router/checkpoint" "github.com/docker/docker/api/server/router/container" distributionrouter "github.com/docker/docker/api/server/router/distribution" + grpcrouter "github.com/docker/docker/api/server/router/grpc" "github.com/docker/docker/api/server/router/image" "github.com/docker/docker/api/server/router/network" pluginrouter "github.com/docker/docker/api/server/router/plugin" @@ -481,6 +482,16 @@ func initRouter(opts routerOptions) { distributionrouter.NewRouter(opts.daemon.ImageService()), } + grpcBackends := []grpcrouter.Backend{} + for _, b := range []interface{}{opts.daemon, opts.buildBackend} { + if b, ok := b.(grpcrouter.Backend); ok { + grpcBackends = append(grpcBackends, b) + } + } + if len(grpcBackends) > 0 { + routers = append(routers, grpcrouter.NewRouter(grpcBackends...)) + } + if opts.daemon.NetworkControllerEnabled() { routers = append(routers, network.NewRouter(opts.daemon, opts.cluster)) }