fixes for history API

api/services/control/control.proto

@@ -206,6 +206,7 @@ message BuildHistoryRecord {
 	bool pinned = 14;
 	int32 numCachedSteps = 15;
 	int32 numTotalSteps = 16;
+	int32 numCompletedSteps = 17;
 	// TODO: tags
 	// TODO: unclipped logs
 }

client/solve.go

@@ -49,6 +49,7 @@ type SolveOpt struct {
 	SessionPreInitialized bool             // TODO: refactor to better session syncing
 	Internal              bool
 	SourcePolicy          *spb.Policy
+	Ref                   string
 }
 
 type ExportEntry struct {
@@ -95,6 +96,9 @@ func (c *Client) solve(ctx context.Context, def *llb.Definition, runGateway runG
 	}
 
 	ref := identity.NewID()
+	if opt.Ref != "" {
+		ref = opt.Ref
+	}
 	eg, ctx := errgroup.WithContext(ctx)
 
 	statusContext, cancelStatus := context.WithCancel(context.Background())

cmd/buildctl/build.go

@@ -16,6 +16,7 @@ import (
 	"github.com/moby/buildkit/cmd/buildctl/build"
 	bccommon "github.com/moby/buildkit/cmd/buildctl/common"
 	gateway "github.com/moby/buildkit/frontend/gateway/client"
+	"github.com/moby/buildkit/identity"
 	"github.com/moby/buildkit/session"
 	"github.com/moby/buildkit/session/auth/authprovider"
 	"github.com/moby/buildkit/session/sshforward/sshprovider"
@@ -100,6 +101,10 @@ var buildCommand = cli.Command{
 			Name:  "source-policy-file",
 			Usage: "Read source policy file from a JSON file",
 		},
+		cli.StringFlag{
+			Name:  "ref-file",
+			Usage: "Write build ref to a file",
+		},
 	},
 }
 
@@ -209,6 +214,8 @@ func buildAction(clicontext *cli.Context) error {
 
 	eg, ctx := errgroup.WithContext(bccommon.CommandContext(clicontext))
 
+	ref := identity.NewID()
+
 	solveOpt := client.SolveOpt{
 		Exports: exports,
 		// LocalDirs is set later
@@ -220,6 +227,7 @@ func buildAction(clicontext *cli.Context) error {
 		Session:             attachable,
 		AllowedEntitlements: allowed,
 		SourcePolicy:        srcPol,
+		Ref:                 ref,
 	}
 
 	solveOpt.FrontendAttrs, err = build.ParseOpt(clicontext.StringSlice("opt"))
@@ -255,6 +263,13 @@ func buildAction(clicontext *cli.Context) error {
 		}
 	}
 
+	refFile := clicontext.String("ref-file")
+	if refFile != "" {
+		defer func() {
+			continuity.AtomicWriteFile(refFile, []byte(ref), 0666)
+		}()
+	}
+
 	// not using shared context to not disrupt display but let is finish reporting errors
 	pw, err := progresswriter.NewPrinter(context.TODO(), os.Stderr, clicontext.String("progress"))
 	if err != nil {

control/control.go

@@ -44,6 +44,7 @@ import (
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 )
 
@@ -248,6 +249,9 @@ func (c *Controller) Export(ctx context.Context, req *tracev1.ExportTraceService
 }
 
 func (c *Controller) ListenBuildHistory(req *controlapi.BuildHistoryRequest, srv controlapi.Control_ListenBuildHistoryServer) error {
+	if err := sendTimestampHeader(srv); err != nil {
+		return err
+	}
 	return c.history.Listen(srv.Context(), req, func(h *controlapi.BuildHistoryEvent) error {
 		if err := srv.Send(h); err != nil {
 			return err
@@ -426,6 +430,8 @@ func (c *Controller) Solve(ctx context.Context, req *controlapi.SolveRequest) (*
 	}, llbsolver.ExporterRequest{
 		Exporter:       expi,
 		CacheExporters: cacheExporters,
+		Type:           req.Exporter,
+		Attrs:          req.ExporterAttrs,
 	}, req.Entitlements, procs, req.Internal, req.SourcePolicy)
 	if err != nil {
 		return nil, err
@@ -436,6 +442,9 @@ func (c *Controller) Solve(ctx context.Context, req *controlapi.SolveRequest) (*
 }
 
 func (c *Controller) Status(req *controlapi.StatusRequest, stream controlapi.Control_StatusServer) error {
+	if err := sendTimestampHeader(stream); err != nil {
+		return err
+	}
 	ch := make(chan *client.SolveStatus, 8)
 
 	eg, ctx := errgroup.WithContext(stream.Context())
@@ -645,3 +654,9 @@ func (cs *roContentStore) Update(ctx context.Context, info content.Info, fieldpa
 func (cs *roContentStore) Abort(ctx context.Context, ref string) error {
 	return errors.Errorf("read-only content store")
 }
+
+const timestampKey = "buildkit-current-timestamp"
+
+func sendTimestampHeader(srv grpc.ServerStream) error {
+	return srv.SendHeader(metadata.Pairs(timestampKey, time.Now().Format(time.RFC3339Nano)))
+}

solver/llbsolver/history.go

@@ -45,9 +45,10 @@ type HistoryQueue struct {
 }
 
 type StatusImportResult struct {
-	Descriptor     ocispecs.Descriptor
-	NumCachedSteps int
-	NumTotalSteps  int
+	Descriptor        ocispecs.Descriptor
+	NumCachedSteps    int
+	NumCompletedSteps int
+	NumTotalSteps     int
 }
 
 func NewHistoryQueue(opt HistoryQueueOpt) *HistoryQueue {
@@ -461,16 +462,23 @@ func (h *HistoryQueue) ImportStatus(ctx context.Context, ch chan *client.SolveSt
 		}
 	}()
 
-	vtxMap := make(map[digest.Digest]bool)
+	type vtxInfo struct {
+		cached    bool
+		completed bool
+	}
+	vtxMap := make(map[digest.Digest]*vtxInfo)
 
 	buf := make([]byte, 32*1024)
 	for st := range ch {
 		for _, vtx := range st.Vertexes {
 			if _, ok := vtxMap[vtx.Digest]; !ok {
-				vtxMap[vtx.Digest] = false
+				vtxMap[vtx.Digest] = &vtxInfo{}
 			}
 			if vtx.Cached {
-				vtxMap[vtx.Digest] = true
+				vtxMap[vtx.Digest].cached = true
+			}
+			if vtx.Completed != nil {
+				vtxMap[vtx.Digest].completed = true
 			}
 		}
 
@@ -502,16 +510,21 @@ func (h *HistoryQueue) ImportStatus(ctx context.Context, ch chan *client.SolveSt
 	}
 
 	numCached := 0
-	for _, cached := range vtxMap {
-		if cached {
+	numCompleted := 0
+	for _, info := range vtxMap {
+		if info.cached {
 			numCached++
 		}
+		if info.completed {
+			numCompleted++
+		}
 	}
 
 	return &StatusImportResult{
-		Descriptor:     *desc,
-		NumCachedSteps: numCached,
-		NumTotalSteps:  len(vtxMap),
+		Descriptor:        *desc,
+		NumCachedSteps:    numCached,
+		NumCompletedSteps: numCompleted,
+		NumTotalSteps:     len(vtxMap),
 	}, release, nil
 }
 

solver/llbsolver/solver.go

@@ -51,6 +51,8 @@ const (
 )
 
 type ExporterRequest struct {
+	Type           string
+	Attrs          map[string]string
 	Exporter       exporter.ExporterInstance
 	CacheExporters []RemoteCacheExporter
 }
@@ -139,7 +141,7 @@ func (s *Solver) Bridge(b solver.Builder) frontend.FrontendLLBBridge {
 	return s.bridge(b)
 }
 
-func (s *Solver) recordBuildHistory(ctx context.Context, id string, req frontend.SolveRequest, j *solver.Job) (func(*Result, exporter.DescriptorReference, error) error, error) {
+func (s *Solver) recordBuildHistory(ctx context.Context, id string, req frontend.SolveRequest, exp ExporterRequest, j *solver.Job) (func(*Result, exporter.DescriptorReference, error) error, error) {
 	var stopTrace func() []tracetest.SpanStub
 
 	if s := trace.SpanFromContext(ctx); s.SpanContext().IsValid() {
@@ -157,6 +159,14 @@ func (s *Solver) recordBuildHistory(ctx context.Context, id string, req frontend
 		FrontendAttrs: req.FrontendOpt,
 		CreatedAt:     &st,
 	}
+
+	if exp.Type != "" {
+		rec.Exporters = []*controlapi.Exporter{{
+			Type:  exp.Type,
+			Attrs: exp.Attrs,
+		}}
+	}
+
 	if err := s.history.Update(ctx, &controlapi.BuildHistoryEvent{
 		Type:   controlapi.BuildHistoryEventType_STARTED,
 		Record: rec,
@@ -177,6 +187,9 @@ func (s *Solver) recordBuildHistory(ctx context.Context, id string, req frontend
 			}
 		}
 
+		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+		defer cancel()
+
 		var mu sync.Mutex
 		ch := make(chan *client.SolveStatus)
 		eg, ctx2 := errgroup.WithContext(ctx)
@@ -282,6 +295,7 @@ func (s *Solver) recordBuildHistory(ctx context.Context, id string, req frontend
 				MediaType: st.Descriptor.MediaType,
 			}
 			rec.NumCachedSteps = int32(st.NumCachedSteps)
+			rec.NumCompletedSteps = int32(st.NumCompletedSteps)
 			rec.NumTotalSteps = int32(st.NumTotalSteps)
 			mu.Unlock()
 			return nil
@@ -334,10 +348,6 @@ func (s *Solver) recordBuildHistory(ctx context.Context, id string, req frontend
 			}
 		}
 
-		if err != nil {
-			return err
-		}
-
 		if stopTrace == nil {
 			logrus.Warn("no trace recorder found, skipping")
 			return err
@@ -414,7 +424,7 @@ func (s *Solver) Solve(ctx context.Context, id string, sessionID string, req fro
 	if internal {
 		defer j.CloseProgress()
 	} else {
-		rec, err1 := s.recordBuildHistory(ctx, id, req, j)
+		rec, err1 := s.recordBuildHistory(ctx, id, req, exp, j)
 		if err != nil {
 			defer j.CloseProgress()
 			return nil, err1