Experimental BuildKit support

Makefile

@@ -31,6 +31,7 @@ DOCKER_ENVS := \
 	-e DOCKER_BUILD_ARGS \
 	-e DOCKER_BUILD_GOGC \
 	-e DOCKER_BUILD_PKGS \
+	-e DOCKER_BUILDKIT \
 	-e DOCKER_BASH_COMPLETION_PATH \
 	-e DOCKER_CLI_PATH \
 	-e DOCKER_DEBUG \

api/server/backend/build/backend.go

@@ -8,10 +8,12 @@ import (
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/backend"
 	"github.com/docker/docker/builder"
+	buildkit "github.com/docker/docker/builder/builder-next"
 	"github.com/docker/docker/builder/fscache"
 	"github.com/docker/docker/image"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 )
 
 // ImageComponent provides an interface for working with images
@@ -30,24 +32,39 @@ type Backend struct {
 	builder        Builder
 	fsCache        *fscache.FSCache
 	imageComponent ImageComponent
+	buildkit       *buildkit.Builder
 }
 
 // NewBackend creates a new build backend from components
-func NewBackend(components ImageComponent, builder Builder, fsCache *fscache.FSCache) (*Backend, error) {
-	return &Backend{imageComponent: components, builder: builder, fsCache: fsCache}, nil
+func NewBackend(components ImageComponent, builder Builder, fsCache *fscache.FSCache, buildkit *buildkit.Builder) (*Backend, error) {
+	return &Backend{imageComponent: components, builder: builder, fsCache: fsCache, buildkit: buildkit}, nil
 }
 
 // Build builds an image from a Source
 func (b *Backend) Build(ctx context.Context, config backend.BuildConfig) (string, error) {
 	options := config.Options
+	useBuildKit := options.Version == types.BuilderBuildKit
+
 	tagger, err := NewTagger(b.imageComponent, config.ProgressWriter.StdoutFormatter, options.Tags)
 	if err != nil {
 		return "", err
 	}
 
-	build, err := b.builder.Build(ctx, config)
-	if err != nil {
-		return "", err
+	var build *builder.Result
+	if useBuildKit {
+		build, err = b.buildkit.Build(ctx, config)
+		if err != nil {
+			return "", err
+		}
+	} else {
+		build, err = b.builder.Build(ctx, config)
+		if err != nil {
+			return "", err
+		}
+	}
+
+	if build == nil {
+		return "", nil
 	}
 
 	var imageID = build.ImageID
@@ -62,19 +79,48 @@ func (b *Backend) Build(ctx context.Context, config backend.BuildConfig) (string
 		}
 	}
 
-	stdout := config.ProgressWriter.StdoutFormatter
-	fmt.Fprintf(stdout, "Successfully built %s\n", stringid.TruncateID(imageID))
-	err = tagger.TagImages(image.ID(imageID))
+	if !useBuildKit {
+		stdout := config.ProgressWriter.StdoutFormatter
+		fmt.Fprintf(stdout, "Successfully built %s\n", stringid.TruncateID(imageID))
+		err = tagger.TagImages(image.ID(imageID))
+	}
 	return imageID, err
 }
 
 // PruneCache removes all cached build sources
 func (b *Backend) PruneCache(ctx context.Context) (*types.BuildCachePruneReport, error) {
-	size, err := b.fsCache.Prune(ctx)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to prune build cache")
+	eg, ctx := errgroup.WithContext(ctx)
+
+	var fsCacheSize uint64
+	eg.Go(func() error {
+		var err error
+		fsCacheSize, err = b.fsCache.Prune(ctx)
+		if err != nil {
+			return errors.Wrap(err, "failed to prune fscache")
+		}
+		return nil
+	})
+
+	var buildCacheSize int64
+	eg.Go(func() error {
+		var err error
+		buildCacheSize, err = b.buildkit.Prune(ctx)
+		if err != nil {
+			return errors.Wrap(err, "failed to prune build cache")
+		}
+		return nil
+	})
+
+	if err := eg.Wait(); err != nil {
+		return nil, err
 	}
-	return &types.BuildCachePruneReport{SpaceReclaimed: size}, nil
+
+	return &types.BuildCachePruneReport{SpaceReclaimed: fsCacheSize + uint64(buildCacheSize)}, nil
+}
+
+// Cancel cancels the build by ID
+func (b *Backend) Cancel(ctx context.Context, id string) error {
+	return b.buildkit.Cancel(ctx, id)
 }
 
 func squashBuild(build *builder.Result, imageComponent ImageComponent) (string, error) {

api/server/router/build/backend.go

@@ -15,6 +15,8 @@ type Backend interface {
 
 	// Prune build cache
 	PruneCache(context.Context) (*types.BuildCachePruneReport, error)
+
+	Cancel(context.Context, string) error
 }
 
 type experimentalProvider interface {

api/server/router/build/build.go

@@ -25,5 +25,6 @@ func (r *buildRouter) initRoutes() {
 	r.routes = []router.Route{
 		router.NewPostRoute("/build", r.postBuild, router.WithCancel),
 		router.NewPostRoute("/build/prune", r.postPrune, router.WithCancel),
+		router.NewPostRoute("/build/cancel", r.postCancel),
 	}
 }

api/server/router/build/build_routes.go

@@ -1,6 +1,7 @@
 package build // import "github.com/docker/docker/api/server/router/build"
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/base64"
@@ -145,10 +146,26 @@ func newImageBuildOptions(ctx context.Context, r *http.Request) (*types.ImageBui
 		options.CacheFrom = cacheFrom
 	}
 	options.SessionID = r.FormValue("session")
+	options.BuildID = r.FormValue("buildid")
+	builderVersion, err := parseVersion(r.FormValue("version"))
+	if err != nil {
+		return nil, err
+	}
+	options.Version = builderVersion
 
 	return options, nil
 }
 
+func parseVersion(s string) (types.BuilderVersion, error) {
+	if s == "" || s == string(types.BuilderV1) {
+		return types.BuilderV1, nil
+	}
+	if s == string(types.BuilderBuildKit) {
+		return types.BuilderBuildKit, nil
+	}
+	return "", errors.Errorf("invalid version %s", s)
+}
+
 func (br *buildRouter) postPrune(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
 	report, err := br.backend.PruneCache(ctx)
 	if err != nil {
@@ -157,6 +174,17 @@ func (br *buildRouter) postPrune(ctx context.Context, w http.ResponseWriter, r *
 	return httputils.WriteJSON(w, http.StatusOK, report)
 }
 
+func (br *buildRouter) postCancel(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
+	w.Header().Set("Content-Type", "application/json")
+
+	id := r.FormValue("id")
+	if id == "" {
+		return errors.Errorf("build ID not provided")
+	}
+
+	return br.backend.Cancel(ctx, id)
+}
+
 func (br *buildRouter) postBuild(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
 	var (
 		notVerboseBuffer = bytes.NewBuffer(nil)
@@ -165,18 +193,34 @@ func (br *buildRouter) postBuild(ctx context.Context, w http.ResponseWriter, r *
 
 	w.Header().Set("Content-Type", "application/json")
 
-	output := ioutils.NewWriteFlusher(w)
+	body := r.Body
+	var ww io.Writer = w
+	if body != nil {
+		// there is a possibility that output is written before request body
+		// has been fully read so we need to protect against it.
+		// this can be removed when
+		// https://github.com/golang/go/issues/15527
+		// https://github.com/golang/go/issues/22209
+		// has been fixed
+		body, ww = wrapOutputBufferedUntilRequestRead(body, ww)
+	}
+
+	output := ioutils.NewWriteFlusher(ww)
 	defer output.Close()
+
 	errf := func(err error) error {
+
 		if httputils.BoolValue(r, "q") && notVerboseBuffer.Len() > 0 {
 			output.Write(notVerboseBuffer.Bytes())
 		}
+
+		logrus.Debugf("isflushed %v", output.Flushed())
 		// Do not write the error in the http output if it's still empty.
 		// This prevents from writing a 200(OK) when there is an internal error.
 		if !output.Flushed() {
 			return err
 		}
-		_, err = w.Write(streamformatter.FormatError(err))
+		_, err = output.Write(streamformatter.FormatError(err))
 		if err != nil {
 			logrus.Warnf("could not write error response: %v", err)
 		}
@@ -205,10 +249,14 @@ func (br *buildRouter) postBuild(ctx context.Context, w http.ResponseWriter, r *
 		return progress.NewProgressReader(in, progressOutput, r.ContentLength, "Downloading context", buildOptions.RemoteContext)
 	}
 
+	if buildOptions.Version == types.BuilderBuildKit && !br.daemon.HasExperimental() {
+		return errdefs.InvalidParameter(errors.New("buildkit is only supported with experimental mode"))
+	}
+
 	wantAux := versions.GreaterThanOrEqualTo(version, "1.30")
 
 	imgID, err := br.backend.Build(ctx, backend.BuildConfig{
-		Source:         r.Body,
+		Source:         body,
 		Options:        buildOptions,
 		ProgressWriter: buildProgressWriter(out, wantAux, createProgressReader),
 	})
@@ -267,3 +315,102 @@ func buildProgressWriter(out io.Writer, wantAux bool, createProgressReader func(
 		ProgressReaderFunc: createProgressReader,
 	}
 }
+
+type flusher interface {
+	Flush()
+}
+
+func wrapOutputBufferedUntilRequestRead(rc io.ReadCloser, out io.Writer) (io.ReadCloser, io.Writer) {
+	var fl flusher = &ioutils.NopFlusher{}
+	if f, ok := out.(flusher); ok {
+		fl = f
+	}
+
+	w := &wcf{
+		buf:     bytes.NewBuffer(nil),
+		Writer:  out,
+		flusher: fl,
+	}
+	r := bufio.NewReader(rc)
+	_, err := r.Peek(1)
+	if err != nil {
+		return rc, out
+	}
+	rc = &rcNotifier{
+		Reader: r,
+		Closer: rc,
+		notify: w.notify,
+	}
+	return rc, w
+}
+
+type rcNotifier struct {
+	io.Reader
+	io.Closer
+	notify func()
+}
+
+func (r *rcNotifier) Read(b []byte) (int, error) {
+	n, err := r.Reader.Read(b)
+	if err != nil {
+		r.notify()
+	}
+	return n, err
+}
+
+func (r *rcNotifier) Close() error {
+	r.notify()
+	return r.Closer.Close()
+}
+
+type wcf struct {
+	io.Writer
+	flusher
+	mu      sync.Mutex
+	ready   bool
+	buf     *bytes.Buffer
+	flushed bool
+}
+
+func (w *wcf) Flush() {
+	w.mu.Lock()
+	w.flushed = true
+	if !w.ready {
+		w.mu.Unlock()
+		return
+	}
+	w.mu.Unlock()
+	w.flusher.Flush()
+}
+
+func (w *wcf) Flushed() bool {
+	w.mu.Lock()
+	b := w.flushed
+	w.mu.Unlock()
+	return b
+}
+
+func (w *wcf) Write(b []byte) (int, error) {
+	w.mu.Lock()
+	if !w.ready {
+		n, err := w.buf.Write(b)
+		w.mu.Unlock()
+		return n, err
+	}
+	w.mu.Unlock()
+	return w.Writer.Write(b)
+}
+
+func (w *wcf) notify() {
+	w.mu.Lock()
+	if !w.ready {
+		if w.buf.Len() > 0 {
+			io.Copy(w.Writer, w.buf)
+		}
+		if w.flushed {
+			w.flusher.Flush()
+		}
+		w.ready = true
+	}
+	w.mu.Unlock()
+}

api/server/router/system/system.go

@@ -2,6 +2,7 @@ package system // import "github.com/docker/docker/api/server/router/system"
 
 import (
 	"github.com/docker/docker/api/server/router"
+	buildkit "github.com/docker/docker/builder/builder-next"
 	"github.com/docker/docker/builder/fscache"
 )
 
@@ -11,15 +12,17 @@ type systemRouter struct {
 	backend Backend
 	cluster ClusterBackend
 	routes  []router.Route
-	builder *fscache.FSCache
+	fscache *fscache.FSCache // legacy
+	builder *buildkit.Builder
 }
 
 // NewRouter initializes a new system router
-func NewRouter(b Backend, c ClusterBackend, fscache *fscache.FSCache) router.Router {
+func NewRouter(b Backend, c ClusterBackend, fscache *fscache.FSCache, builder *buildkit.Builder) router.Router {
 	r := &systemRouter{
 		backend: b,
 		cluster: c,
-		builder: fscache,
+		fscache: fscache,
+		builder: builder,
 	}
 
 	r.routes = []router.Route{