Merge pull request containerd#9630 from AkihiroSuda/transfer-desc

transfer: add OCI descriptors to Progress structure

Author: fuweid

api/next.pb.txt

@@ -7231,6 +7231,7 @@ file {
 file {
   name: "github.com/containerd/containerd/api/types/transfer/progress.proto"
   package: "containerd.types.transfer"
+  dependency: "github.com/containerd/containerd/api/types/descriptor.proto"
   message_type {
     name: "Progress"
     field {
@@ -7268,6 +7269,14 @@ file {
       type: TYPE_INT64
       json_name: "total"
     }
+    field {
+      name: "desc"
+      number: 6
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".containerd.types.Descriptor"
+      json_name: "desc"
+    }
   }
   options {
     go_package: "github.com/containerd/containerd/v2/api/types/transfer"

api/types/transfer/progress.pb.go

@@ -18,6 +18,8 @@ syntax = "proto3";
 
 package containerd.types.transfer;
 
+import "github.com/containerd/containerd/api/types/descriptor.proto";
+
 option go_package = "github.com/containerd/containerd/v2/api/types/transfer";
 
 message Progress {
@@ -26,4 +28,5 @@ message Progress {
 	repeated string parents = 3;
 	int64 progress = 4;
 	int64 total = 5;
+	containerd.types.Descriptor desc = 6;
 }

api/types/transfer/progress.proto

@@ -206,6 +206,18 @@ type progressNode struct {
 	root     bool
 }
 
+func (n *progressNode) mainDesc() *ocispec.Descriptor {
+	if n.Desc != nil {
+		return n.Desc
+	}
+	for _, c := range n.children {
+		if desc := c.mainDesc(); desc != nil {
+			return desc
+		}
+	}
+	return nil
+}
+
 // ProgressHandler continuously updates the output with job progress
 // by checking status in the content store.
 func ProgressHandler(ctx context.Context, out io.Writer) (transfer.ProgressFunc, func()) {
@@ -329,6 +341,11 @@ func ProgressHandler(ctx context.Context, out io.Writer) (transfer.ProgressFunc,
 
 func DisplayHierarchy(w io.Writer, status string, roots []*progressNode, start time.Time) {
 	total := displayNode(w, "", roots)
+	for _, r := range roots {
+		if desc := r.mainDesc(); desc != nil {
+			fmt.Fprintf(w, "%s %s\n", desc.MediaType, desc.Digest)
+		}
+	}
 	// Print the Status line
 	fmt.Fprintf(w, "%s\telapsed: %-4.1fs\ttotal: %7.6v\t(%v)\t\n",
 		status,

cmd/ctr/commands/images/pull.go

@@ -128,6 +128,7 @@ func (ts *localTransferService) importStream(ctx context.Context, i transfer.Ima
 	}
 
 	for _, desc := range descriptors {
+		desc := desc
 		imgs, err := is.Store(ctx, desc, ts.images)
 		if err != nil {
 			if errdefs.IsNotFound(err) {
@@ -142,6 +143,7 @@ func (ts *localTransferService) importStream(ctx context.Context, i transfer.Ima
 				tops.Progress(transfer.Progress{
 					Event: "saved",
 					Name:  img.Name,
+					Desc:  &desc,
 				})
 			}
 		}

pkg/transfer/local/import.go

@@ -106,6 +106,7 @@ func (j *ProgressTracker) HandleProgress(ctx context.Context, pf transfer.Progre
 							Parents:  job.parents,
 							Progress: status.Offset,
 							Total:    status.Total,
+							Desc:     &job.desc,
 						})
 						job.progress = status.Offset
 						job.state = jobInProgress
@@ -122,6 +123,7 @@ func (j *ProgressTracker) HandleProgress(ctx context.Context, pf transfer.Progre
 							Parents:  job.parents,
 							Progress: job.desc.Size,
 							Total:    job.desc.Size,
+							Desc:     &job.desc,
 						})
 
 					}
@@ -165,6 +167,7 @@ func (j *ProgressTracker) HandleProgress(ctx context.Context, pf transfer.Progre
 					//Digest:   desc.Digest.String(),
 					Progress: 0,
 					Total:    update.desc.Size,
+					Desc:     &job.desc,
 				})
 			}
 			if update.exists {
@@ -173,6 +176,7 @@ func (j *ProgressTracker) HandleProgress(ctx context.Context, pf transfer.Progre
 					Name:     remotes.MakeRefKey(ctx, update.desc),
 					Progress: update.desc.Size,
 					Total:    update.desc.Size,
+					Desc:     &job.desc,
 				})
 				job.state = jobComplete
 				job.progress = job.desc.Size

pkg/transfer/local/progress.go

@@ -63,6 +63,7 @@ func (ts *localTransferService) push(ctx context.Context, ig transfer.ImageGette
 			Event: "pushing content",
 			Name:  img.Name,
 			//Digest: img.Target.Digest.String(),
+			Desc: &img.Target,
 		})
 	}
 
@@ -109,9 +110,11 @@ func (ts *localTransferService) push(ctx context.Context, ig transfer.ImageGette
 			Event: "pushed content",
 			Name:  img.Name,
 			//Digest: img.Target.Digest.String(),
+			Desc: &img.Target,
 		})
 		tops.Progress(transfer.Progress{
 			Event: fmt.Sprintf("Completed push to %s", p),
+			Desc:  &img.Target,
 		})
 	}
 

pkg/transfer/local/push.go

@@ -25,11 +25,13 @@ import (
 
 	transferapi "github.com/containerd/containerd/v2/api/services/transfer/v1"
 	transfertypes "github.com/containerd/containerd/v2/api/types/transfer"
+	"github.com/containerd/containerd/v2/oci"
 	"github.com/containerd/containerd/v2/pkg/streaming"
 	"github.com/containerd/containerd/v2/pkg/transfer"
 	tstreaming "github.com/containerd/containerd/v2/pkg/transfer/streaming"
 	"github.com/containerd/log"
 	"github.com/containerd/typeurl/v2"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 )
 
 type proxyTransferrer struct {
@@ -74,12 +76,18 @@ func (p *proxyTransferrer) Transfer(ctx context.Context, src interface{}, dst in
 				}
 				switch v := i.(type) {
 				case *transfertypes.Progress:
+					var descp *ocispec.Descriptor
+					if v.Desc != nil {
+						desc := oci.DescriptorFromProto(v.Desc)
+						descp = &desc
+					}
 					o.Progress(transfer.Progress{
 						Event:    v.Event,
 						Name:     v.Name,
 						Parents:  v.Parents,
 						Progress: v.Progress,
 						Total:    v.Total,
+						Desc:     descp,
 					})
 				default:
 					log.G(ctx).Warnf("unhandled progress object %T: %v", i, a.GetTypeUrl())

pkg/transfer/proxy/transfer.go

@@ -132,5 +132,5 @@ type Progress struct {
 	Parents  []string
 	Progress int64
 	Total    int64
-	// Descriptor?
+	Desc     *ocispec.Descriptor // since containerd v2.0
 }

pkg/transfer/transfer.go

@@ -20,8 +20,10 @@ import (
 	"context"
 
 	transferapi "github.com/containerd/containerd/v2/api/services/transfer/v1"
+	"github.com/containerd/containerd/v2/api/types"
 	transferTypes "github.com/containerd/containerd/v2/api/types/transfer"
 	"github.com/containerd/containerd/v2/errdefs"
+	"github.com/containerd/containerd/v2/oci"
 	"github.com/containerd/containerd/v2/pkg/streaming"
 	"github.com/containerd/containerd/v2/pkg/transfer"
 	tplugins "github.com/containerd/containerd/v2/pkg/transfer/plugins"
@@ -92,12 +94,17 @@ func (s *service) Transfer(ctx context.Context, req *transferapi.TransferRequest
 			defer stream.Close()
 
 			pf := func(p transfer.Progress) {
+				var descp *types.Descriptor
+				if p.Desc != nil {
+					descp = oci.DescriptorToProto(*p.Desc)
+				}
 				progress, err := typeurl.MarshalAny(&transferTypes.Progress{
 					Event:    p.Event,
 					Name:     p.Name,
 					Parents:  p.Parents,
 					Progress: p.Progress,
 					Total:    p.Total,
+					Desc:     descp,
 				})
 				if err != nil {
 					log.G(ctx).WithError(err).Warnf("event could not be marshaled: %v/%v", p.Event, p.Name)