Merge pull request #533 from ijc/client-gateway

access gateway API from client

client/build.go

@@ -0,0 +1,95 @@
+package client
+
+import (
+ "context"
+
+ "github.com/moby/buildkit/client/buildid"
+ gateway "github.com/moby/buildkit/frontend/gateway/client"
+ "github.com/moby/buildkit/frontend/gateway/grpcclient"
+ gatewayapi "github.com/moby/buildkit/frontend/gateway/pb"
+ "github.com/moby/buildkit/session"
+ "github.com/moby/buildkit/util/apicaps"
+ "github.com/pkg/errors"
+ "google.golang.org/grpc"
+)
+
+func (c *Client) Build(ctx context.Context, opt SolveOpt, product string, buildFunc gateway.BuildFunc, statusChan chan *SolveStatus) (*SolveResponse, error) {
+ defer func() {
+ if statusChan != nil {
+ close(statusChan)
+ }
+ }()
+
+ if opt.Frontend != "" {
+ return nil, errors.New("invalid SolveOpt, Build interface cannot use Frontend")
+ }
+
+ if product == "" {
+ product = apicaps.ExportedProduct
+ }
+
+ feOpts := opt.FrontendAttrs
+ opt.FrontendAttrs = nil
+
+ workers, err := c.ListWorkers(ctx)
+ if err != nil {
+ return nil, errors.Wrap(err, "listing workers for Build")
+ }
+ var gworkers []gateway.WorkerInfo
+ for _, w := range workers {
+ gworkers = append(gworkers, gateway.WorkerInfo{
+ ID: w.ID,
+ Labels: w.Labels,
+ Platforms: w.Platforms,
+ })
+ }
+
+ cb := func(ref string, s *session.Session) error {
+ g, err := grpcclient.New(ctx, feOpts, s.ID(), product, c.gatewayClientForBuild(ref), gworkers)
+ if err != nil {
+ return err
+ }
+
+ if err := g.Run(ctx, buildFunc); err != nil {
+ return errors.Wrap(err, "failed to run Build function")
+ }
+ return nil
+ }
+
+ return c.solve(ctx, nil, cb, opt, statusChan)
+}
+
+func (c *Client) gatewayClientForBuild(buildid string) gatewayapi.LLBBridgeClient {
+ g := gatewayapi.NewLLBBridgeClient(c.conn)
+ return &gatewayClientForBuild{g, buildid}
+}
+
+type gatewayClientForBuild struct {
+ gateway gatewayapi.LLBBridgeClient
+ buildID string
+}
+
+func (g *gatewayClientForBuild) ResolveImageConfig(ctx context.Context, in *gatewayapi.ResolveImageConfigRequest, opts ...grpc.CallOption) (*gatewayapi.ResolveImageConfigResponse, error) {
+ ctx = buildid.AppendToOutgoingContext(ctx, g.buildID)
+ return g.gateway.ResolveImageConfig(ctx, in, opts...)
+}
+
+func (g *gatewayClientForBuild) Solve(ctx context.Context, in *gatewayapi.SolveRequest, opts ...grpc.CallOption) (*gatewayapi.SolveResponse, error) {
+ ctx = buildid.AppendToOutgoingContext(ctx, g.buildID)
+ return g.gateway.Solve(ctx, in, opts...)
+}
+
+func (g *gatewayClientForBuild) ReadFile(ctx context.Context, in *gatewayapi.ReadFileRequest, opts ...grpc.CallOption) (*gatewayapi.ReadFileResponse, error) {
+ ctx = buildid.AppendToOutgoingContext(ctx, g.buildID)
+ return g.gateway.ReadFile(ctx, in, opts...)
+}
+
+func (g *gatewayClientForBuild) Ping(ctx context.Context, in *gatewayapi.PingRequest, opts ...grpc.CallOption) (*gatewayapi.PongResponse, error) {
+ ctx = buildid.AppendToOutgoingContext(ctx, g.buildID)
+ return g.gateway.Ping(ctx, in, opts...)
+}
+
+func (g *gatewayClientForBuild) Return(ctx context.Context, in *gatewayapi.ReturnRequest, opts ...grpc.CallOption) (*gatewayapi.ReturnResponse, error) {
+ ctx = buildid.AppendToOutgoingContext(ctx, g.buildID)
+ return g.gateway.Return(ctx, in, opts...)
+}

client/build_test.go

@@ -0,0 +1,178 @@
+package client
+
+import (
+ "context"
+ "io/ioutil"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/moby/buildkit/client/llb"
+ "github.com/moby/buildkit/frontend/gateway/client"
+ gatewayapi "github.com/moby/buildkit/frontend/gateway/pb"
+ "github.com/moby/buildkit/identity"
+ "github.com/moby/buildkit/util/testutil/integration"
+ "github.com/pkg/errors"
+ "github.com/stretchr/testify/require"
+)
+
+func TestClientGatewayIntegration(t *testing.T) {
+ integration.Run(t, []integration.Test{
+ testClientGatewaySolve,
+ testClientGatewayFailedSolve,
+ testClientGatewayEmptySolve,
+ testNoBuildID,
+ testUnknownBuildID,
+ })
+}
+
+func testClientGatewaySolve(t *testing.T, sb integration.Sandbox) {
+ t.Parallel()
+ requiresLinux(t)
+
+ ctx := context.TODO()
+
+ c, err := New(ctx, sb.Address())
+ require.NoError(t, err)
+ defer c.Close()
+
+ product := "buildkit_test"
+ optKey := "test-string"
+
+ b := func(ctx context.Context, c client.Client) (*client.Result, error) {
+ if c.BuildOpts().Product != product {
+ return nil, errors.Errorf("expected product %q, got %q", product, c.BuildOpts().Product)
+ }
+ opts := c.BuildOpts().Opts
+ testStr, ok := opts[optKey]
+ if !ok {
+ return nil, errors.Errorf(`build option %q missing`, optKey)
+ }
+
+ run := llb.Image("busybox:latest").Run(
+ llb.ReadonlyRootFS(),
+ llb.Args([]string{"/bin/sh", "-ec", `echo -n '` + testStr + `' > /out/foo`}),
+ )
+ st := run.AddMount("/out", llb.Scratch())
+
+ def, err := st.Marshal()
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to marshal state")
+ }
+
+ r, err := c.Solve(ctx, client.SolveRequest{
+ Definition: def.ToPB(),
+ })
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to solve")
+ }
+
+ read, err := r.Ref.ReadFile(ctx, client.ReadRequest{
+ Filename: "/foo",
+ })
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to read result")
+ }
+ if testStr != string(read) {
+ return nil, errors.Errorf("read back %q, expected %q", string(read), testStr)
+ }
+ return r, nil
+ }
+
+ tmpdir, err := ioutil.TempDir("", "buildkit")
+ require.NoError(t, err)
+ defer os.RemoveAll(tmpdir)
+
+ testStr := "This is a test"
+
+ _, err = c.Build(ctx, SolveOpt{
+ Exporter: ExporterLocal,
+ ExporterOutputDir: tmpdir,
+ FrontendAttrs: map[string]string{
+ optKey: testStr,
+ },
+ }, product, b, nil)
+ require.NoError(t, err)
+
+ read, err := ioutil.ReadFile(filepath.Join(tmpdir, "foo"))
+ require.NoError(t, err)
+ require.Equal(t, testStr, string(read))
+
+ checkAllReleasable(t, c, sb, true)
+}
+
+func testClientGatewayFailedSolve(t *testing.T, sb integration.Sandbox) {
+ t.Parallel()
+ requiresLinux(t)
+
+ ctx := context.TODO()
+
+ c, err := New(ctx, sb.Address())
+ require.NoError(t, err)
+ defer c.Close()
+
+ b := func(ctx context.Context, c client.Client) (*client.Result, error) {
+ return nil, errors.New("expected to fail")
+ }
+
+ _, err = c.Build(ctx, SolveOpt{}, "", b, nil)
+ require.Error(t, err)
+ require.Contains(t, err.Error(), "expected to fail")
+}
+
+func testClientGatewayEmptySolve(t *testing.T, sb integration.Sandbox) {
+ t.Parallel()
+ requiresLinux(t)
+
+ ctx := context.TODO()
+
+ c, err := New(ctx, sb.Address())
+ require.NoError(t, err)
+ defer c.Close()
+
+ b := func(ctx context.Context, c client.Client) (*client.Result, error) {
+ r, err := c.Solve(ctx, client.SolveRequest{})
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to solve")
+ }
+ if r.Ref != nil || r.Refs != nil || r.Metadata != nil {
+ return nil, errors.Errorf("got unexpected non-empty result %+v", r)
+ }
+ return r, nil
+ }
+
+ _, err = c.Build(ctx, SolveOpt{}, "", b, nil)
+ require.NoError(t, err)
+}
+
+func testNoBuildID(t *testing.T, sb integration.Sandbox) {
+ t.Parallel()
+ requiresLinux(t)
+
+ ctx := context.TODO()
+
+ c, err := New(ctx, sb.Address())
+ require.NoError(t, err)
+ defer c.Close()
+
+ g := gatewayapi.NewLLBBridgeClient(c.conn)
+ _, err = g.Ping(ctx, &gatewayapi.PingRequest{})
+ require.Error(t, err)
+ require.Contains(t, err.Error(), "no buildid found in context")
+}
+
+func testUnknownBuildID(t *testing.T, sb integration.Sandbox) {
+ t.Parallel()
+ requiresLinux(t)
+
+ ctx := context.TODO()
+
+ c, err := New(ctx, sb.Address())
+ require.NoError(t, err)
+ defer c.Close()
+
+ g := c.gatewayClientForBuild(t.Name() + identity.NewID())
+ _, err = g.Ping(ctx, &gatewayapi.PingRequest{})
+ require.Error(t, err)
+ require.Contains(t, err.Error(), "no such job")
+}

client/buildid/metadata.go

@@ -0,0 +1,29 @@
+package buildid
+
+import (
+ "context"
+
+ "google.golang.org/grpc/metadata"
+)
+
+var metadataKey = "buildkit-controlapi-buildid"
+
+func AppendToOutgoingContext(ctx context.Context, id string) context.Context {
+ if id != "" {
+ return metadata.AppendToOutgoingContext(ctx, metadataKey, id)
+ }
+ return ctx
+}
+
+func FromIncomingContext(ctx context.Context) string {
+ md, ok := metadata.FromIncomingContext(ctx)
+ if !ok {
+ return ""
+ }
+
+ if ids := md.Get(metadataKey); len(ids) == 1 {
+ return ids[0]
+ }
+
+ return ""
+}

client/solve.go

@@ -54,6 +54,16 @@ func (c *Client) Solve(ctx context.Context, def *llb.Definition, opt SolveOpt, s
  return nil, errors.Errorf("invalid definition for frontend %s", opt.Frontend)
  }
 
+ return c.solve(ctx, def, nil, opt, statusChan)
+}
+
+type runGatewayCB func(ref string, s *session.Session) error
+
+func (c *Client) solve(ctx context.Context, def *llb.Definition, runGateway runGatewayCB, opt SolveOpt, statusChan chan *SolveStatus) (*SolveResponse, error) {
+ if def != nil && runGateway != nil {
+ return nil, errors.New("invalid with def and cb")
+ }
+
  syncedDirs, err := prepareSyncedDirs(def, opt.LocalDirs)
  if err != nil {
  return nil, err
@@ -112,8 +122,12 @@ func (c *Client) Solve(ctx context.Context, def *llb.Definition, opt SolveOpt, s
  return s.Run(statusContext, grpchijack.Dialer(c.controlClient()))
  })
 
+ solveCtx, cancelSolve := context.WithCancel(ctx)
  var res *SolveResponse
  eg.Go(func() error {
+ ctx := solveCtx
+ defer cancelSolve()
+
  defer func() { // make sure the Status ends cleanly on build errors
  go func() {
  <-time.After(3 * time.Second)
@@ -150,6 +164,28 @@ func (c *Client) Solve(ctx context.Context, def *llb.Definition, opt SolveOpt, s
  return nil
  })
 
+ if runGateway != nil {
+ eg.Go(func() error {
+ err := runGateway(ref, s)
+ if err == nil {
+ return nil
+ }
+
+ // If the callback failed then the main
+ // `Solve` (called above) should error as
+ // well. However as a fallback we wait up to
+ // 5s for that to happen before failing this
+ // goroutine.
+ select {
+ case <-solveCtx.Done():
+ case <-time.After(5 * time.Second):
+ cancelSolve()
+ }
+
+ return err
+ })
+ }
+
  eg.Go(func() error {
  stream, err := c.controlClient().Status(statusContext, &controlapi.StatusRequest{
  Ref: ref,