nydusify: support --push-chunk-size option

Reference: containerd/containerd#9405

Will replace containerd dep to upstream version if the PR can be merged.

Signed-off-by: Yan Song <[email protected]>

contrib/nydusify/cmd/nydusify.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/containerd/containerd/reference/docker"
 	"github.com/docker/distribution/reference"
+	"github.com/dustin/go-humanize"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
@@ -1046,6 +1047,12 @@ func main() {
 					Usage: "Copy images for specific platforms, for example: 'linux/amd64,linux/arm64'",
 				},
 
+				&cli.StringFlag{
+					Name:  "push-chunk-size",
+					Value: "0MB",
+					Usage: "Chunk size for pushing a blob layer in chunked",
+				},
+
 				&cli.StringFlag{
 					Name:    "work-dir",
 					Value:   "./tmp",
@@ -1067,6 +1074,14 @@ func main() {
 					return err
 				}
 
+				pushChunkSize, err := humanize.ParseBytes(c.String("push-chunk-size"))
+				if err != nil {
+					return errors.Wrap(err, "invalid --push-chunk-size option")
+				}
+				if pushChunkSize > 0 {
+					logrus.Infof("will copy layer with chunk size %s", c.String("push-chunk-size"))
+				}
+
 				opt := copier.Opt{
 					WorkDir:        c.String("work-dir"),
 					NydusImagePath: c.String("nydus-image"),
@@ -1081,6 +1096,8 @@ func main() {
 
 					AllPlatforms: c.Bool("all-platforms"),
 					Platforms:    c.String("platform"),
+
+					PushChunkSize: int64(pushChunkSize),
 				}
 
 				return copier.Copy(context.Background(), opt)

contrib/nydusify/go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 // indirect
 	github.com/AdamKorcz/go-118-fuzz-build v0.0.0-20230306123547-8075edf89bb0 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
-	github.com/Microsoft/hcsshim v0.11.0 // indirect
+	github.com/Microsoft/hcsshim v0.11.4 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.10 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.24 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.30 // indirect
@@ -118,3 +118,5 @@ require (
 	gopkg.in/square/go-jose.v2 v2.5.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+replace github.com/containerd/containerd => github.com/nydusaccelerator/containerd v0.0.0-20231121100328-6c4d1f35ac28

contrib/nydusify/go.sum

@@ -6,8 +6,8 @@ github.com/AdamKorcz/go-118-fuzz-build v0.0.0-20230306123547-8075edf89bb0/go.mod
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow=
 github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM=
-github.com/Microsoft/hcsshim v0.11.0 h1:7EFNIY4igHEXUdj1zXgAyU3fLc7QfOKHbkldRVTBdiM=
-github.com/Microsoft/hcsshim v0.11.0/go.mod h1:OEthFdQv/AD2RAdzR6Mm1N1KPCztGKDurW1Z8b8VGMM=
+github.com/Microsoft/hcsshim v0.11.4 h1:68vKo2VN8DE9AdN4tnkWnmdhqdbpUFM8OF3Airm7fz8=
+github.com/Microsoft/hcsshim v0.11.4/go.mod h1:smjE4dvqPX9Zldna+t5FG3rnoHhaB7QYxPRqGcpAD9w=
 github.com/aliyun/aliyun-oss-go-sdk v2.2.6+incompatible h1:KXeJoM1wo9I/6xPTyt6qCxoSZnmASiAjlrr0dyTUKt8=
 github.com/aliyun/aliyun-oss-go-sdk v2.2.6+incompatible/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8=
 github.com/aws/aws-sdk-go-v2 v1.17.6 h1:Y773UK7OBqhzi5VDXMi1zVGsoj+CVHs2eaC2bDsLwi0=

contrib/nydusify/pkg/converter/converter.go

@@ -77,7 +77,7 @@ func Convert(ctx context.Context, opt Opt) error {
 	if err != nil {
 		return errors.Wrap(err, "create temp directory")
 	}
-	pvd, err := provider.New(tmpDir, hosts(opt), opt.CacheMaxRecords, opt.CacheVersion, platformMC)
+	pvd, err := provider.New(tmpDir, hosts(opt), opt.CacheMaxRecords, opt.CacheVersion, platformMC, 0)
 	if err != nil {
 		return err
 	}

contrib/nydusify/pkg/converter/provider/provider.go

@@ -6,15 +6,20 @@ package provider
 
 import (
 	"context"
+	"crypto/tls"
+	"net"
+	"net/http"
 	"os"
 	"path/filepath"
 	"sync"
+	"time"
 
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/content"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/platforms"
 	"github.com/containerd/containerd/remotes"
+	"github.com/containerd/containerd/remotes/docker"
 	"github.com/goharbor/acceleration-service/pkg/cache"
 	accelcontent "github.com/goharbor/acceleration-service/pkg/content"
 	"github.com/goharbor/acceleration-service/pkg/remote"
@@ -32,9 +37,10 @@ type Provider struct {
 	platformMC   platforms.MatchComparer
 	cacheSize    int
 	cacheVersion string
+	chunkSize    int64
 }
 
-func New(root string, hosts remote.HostFunc, cacheSize uint, cacheVersion string, platformMC platforms.MatchComparer) (*Provider, error) {
+func New(root string, hosts remote.HostFunc, cacheSize uint, cacheVersion string, platformMC platforms.MatchComparer, chunkSize int64) (*Provider, error) {
 	contentDir := filepath.Join(root, "content")
 	if err := os.MkdirAll(contentDir, 0755); err != nil {
 		return nil, err
@@ -51,9 +57,52 @@ func New(root string, hosts remote.HostFunc, cacheSize uint, cacheVersion string
 		cacheSize:    int(cacheSize),
 		platformMC:   platformMC,
 		cacheVersion: cacheVersion,
+		chunkSize:    chunkSize,
 	}, nil
 }
 
+func newDefaultClient(skipTLSVerify bool) *http.Client {
+	return &http.Client{
+		Transport: &http.Transport{
+			Proxy: http.ProxyFromEnvironment,
+			DialContext: (&net.Dialer{
+				Timeout:   30 * time.Second,
+				KeepAlive: 30 * time.Second,
+				DualStack: true,
+			}).DialContext,
+			MaxIdleConns:          10,
+			IdleConnTimeout:       30 * time.Second,
+			TLSHandshakeTimeout:   10 * time.Second,
+			ExpectContinueTimeout: 5 * time.Second,
+			DisableKeepAlives:     true,
+			TLSNextProto:          make(map[string]func(authority string, c *tls.Conn) http.RoundTripper),
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: skipTLSVerify,
+			},
+		},
+	}
+}
+
+func newResolver(insecure, plainHTTP bool, credFunc remote.CredentialFunc, chunkSize int64) remotes.Resolver {
+	registryHosts := docker.ConfigureDefaultRegistries(
+		docker.WithAuthorizer(
+			docker.NewDockerAuthorizer(
+				docker.WithAuthClient(newDefaultClient(insecure)),
+				docker.WithAuthCreds(credFunc),
+			),
+		),
+		docker.WithClient(newDefaultClient(insecure)),
+		docker.WithPlainHTTP(func(host string) (bool, error) {
+			return plainHTTP, nil
+		}),
+		docker.WithChunkSize(chunkSize),
+	)
+
+	return docker.NewResolver(docker.ResolverOptions{
+		Hosts: registryHosts,
+	})
+}
+
 func (pvd *Provider) UsePlainHTTP() {
 	pvd.usePlainHTTP = true
 }
@@ -63,7 +112,7 @@ func (pvd *Provider) Resolver(ref string) (remotes.Resolver, error) {
 	if err != nil {
 		return nil, err
 	}
-	return remote.NewResolver(insecure, pvd.usePlainHTTP, credFunc), nil
+	return newResolver(insecure, pvd.usePlainHTTP, credFunc, pvd.chunkSize), nil
 }
 
 func (pvd *Provider) Pull(ctx context.Context, ref string) error {

contrib/nydusify/pkg/copier/copier.go

@@ -57,6 +57,8 @@ type Opt struct {
 
 	AllPlatforms bool
 	Platforms    string
+
+	PushChunkSize int64
 }
 
 type output struct {
@@ -277,7 +279,7 @@ func Copy(ctx context.Context, opt Opt) error {
 	if err != nil {
 		return errors.Wrap(err, "create temp directory")
 	}
-	pvd, err := provider.New(tmpDir, hosts(opt), 200, "v1", platformMC)
+	pvd, err := provider.New(tmpDir, hosts(opt), 200, "v1", platformMC, opt.PushChunkSize)
 	if err != nil {
 		return err
 	}

smoke/tests/image_test.go

@@ -127,7 +127,7 @@ func (i *ImageTestSuite) TestConvertAndCopyImage(t *testing.T, ctx tool.Context,
 	// Copy image
 	targetCopied := fmt.Sprintf("%s_copied", target)
 	copyCmd := fmt.Sprintf(
-		"%s %s copy --source %s --target %s --nydus-image %s --work-dir %s",
+		"%s %s copy --source %s --target %s --nydus-image %s --work-dir %s --chunk-size 1MB",
 		ctx.Binary.Nydusify, logLevel, target, targetCopied, ctx.Binary.Builder, ctx.Env.WorkDir,
 	)
 	tool.RunWithoutOutput(t, copyCmd)