Merge pull request #43534 from thaJeztah/bump_containerd_1.6.3

vendor: github.com/containerd/containerd v1.6.3

vendor.mod

@@ -18,7 +18,7 @@ require (
 	github.com/bsphere/le_go v0.0.0-20170215134836-7a984a84b549
 	github.com/cloudflare/cfssl v0.0.0-20180323000720-5d63dbd981b5
 	github.com/containerd/cgroups v1.0.3
-	github.com/containerd/containerd v1.6.2
+	github.com/containerd/containerd v1.6.3
 	github.com/containerd/continuity v0.2.2
 	github.com/containerd/fifo v1.0.0
 	github.com/containerd/typeurl v1.0.2

vendor.sum

@@ -233,8 +233,8 @@ github.com/containerd/containerd v1.5.1/go.mod h1:0DOxVqwDy2iZvrZp2JUx/E+hS0UNTV
 github.com/containerd/containerd v1.5.7/go.mod h1:gyvv6+ugqY25TiXxcZC3L5yOeYgEw0QMhscqVp1AR9c=
 github.com/containerd/containerd v1.5.8/go.mod h1:YdFSv5bTFLpG2HIYmfqDpSYYTDX+mc5qtSuYx1YUb/s=
 github.com/containerd/containerd v1.6.1/go.mod h1:1nJz5xCZPusx6jJU8Frfct988y0NpumIq9ODB0kLtoE=
-github.com/containerd/containerd v1.6.2 h1:pcaPUGbYW8kBw6OgIZwIVIeEhdWVrBzsoCfVJ5BjrLU=
-github.com/containerd/containerd v1.6.2/go.mod h1:sidY30/InSE1j2vdD1ihtKoJz+lWdaXMdiAeIupaf+s=
+github.com/containerd/containerd v1.6.3 h1:JfgUEIAH07xDWk6kqz0P3ArZt+KJ9YeihSC9uyFtSKg=
+github.com/containerd/containerd v1.6.3/go.mod h1:gCVGrYRYFm2E8GmuUIbj/NGD7DLZQLzSJQazjVKDOig=
 github.com/containerd/continuity v0.0.0-20190426062206-aaeac12a7ffc/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=
 github.com/containerd/continuity v0.0.0-20190815185530-f2a389ac0a02/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=
 github.com/containerd/continuity v0.0.0-20191127005431-f65d91d395eb/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=

vendor/github.com/containerd/containerd/Vagrantfile

@@ -15,9 +15,10 @@
 #   See the License for the specific language governing permissions and
 #   limitations under the License.
 
-# Vagrantfile for cgroup2 and SELinux
+# Vagrantfile for Fedora and EL
 Vagrant.configure("2") do |config|
-  config.vm.box = "fedora/35-cloud-base"
+  config.vm.box = ENV["BOX"] || "fedora/35-cloud-base"
+  config.vm.box_version = ENV["BOX_VERSION"]
   memory = 4096
   cpus = 2
   config.vm.provider :virtualbox do |v|
@@ -71,22 +72,31 @@ Vagrant.configure("2") do |config|
     SHELL
   end
 
+  # EL does not have /usr/local/{bin,sbin} in the PATH by default
+  config.vm.provision "setup-etc-environment", type: "shell", run: "once" do |sh|
+    sh.upload_path = "/tmp/vagrant-setup-etc-environment"
+    sh.inline = <<~SHELL
+        #!/usr/bin/env bash
+        set -eux -o pipefail
+        cat >> /etc/environment <<EOF
+PATH=/usr/local/go/bin:/usr/local/bin:/usr/local/sbin:$PATH
+EOF
+        source /etc/environment
+        SHELL
+  end
+
   # To re-run this provisioner, installing a different version of go:
   #   GO_VERSION="1.14.6" vagrant up --provision-with=install-golang
   #
   config.vm.provision "install-golang", type: "shell", run: "once" do |sh|
     sh.upload_path = "/tmp/vagrant-install-golang"
     sh.env = {
-        'GO_VERSION': ENV['GO_VERSION'] || "1.17.8",
+        'GO_VERSION': ENV['GO_VERSION'] || "1.17.9",
     }
     sh.inline = <<~SHELL
         #!/usr/bin/env bash
         set -eux -o pipefail
         curl -fsSL "https://dl.google.com/go/go${GO_VERSION}.linux-amd64.tar.gz" | tar Cxz /usr/local
-        cat >> /etc/environment <<EOF
-PATH=/usr/local/go/bin:$PATH
-EOF
-        source /etc/environment
         cat >> /etc/profile.d/sh.local <<EOF
 GOPATH=\\$HOME/go
 PATH=\\$GOPATH/bin:\\$PATH
@@ -218,6 +228,7 @@ EOF
         set -eux -o pipefail
         rm -rf /var/lib/containerd-test /run/containerd-test
         cd ${GOPATH}/src/github.com/containerd/containerd
+        go test -v -count=1 -race ./metrics/cgroups
         make integration EXTRA_TESTFLAGS="-timeout 15m -no-criu -test.v" TEST_RUNTIME=io.containerd.runc.v2 RUNC_FLAVOR=$RUNC_FLAVOR
     SHELL
   end

vendor/github.com/containerd/containerd/image.go

@@ -28,6 +28,7 @@ import (
 	"github.com/containerd/containerd/diff"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/pkg/kmutex"
 	"github.com/containerd/containerd/platforms"
 	"github.com/containerd/containerd/rootfs"
 	"github.com/containerd/containerd/snapshots"
@@ -287,6 +288,10 @@ type UnpackConfig struct {
 	// CheckPlatformSupported is whether to validate that a snapshotter
 	// supports an image's platform before unpacking
 	CheckPlatformSupported bool
+	// DuplicationSuppressor is used to make sure that there is only one
+	// in-flight fetch request or unpack handler for a given descriptor's
+	// digest or chain ID.
+	DuplicationSuppressor kmutex.KeyedLocker
 }
 
 // UnpackOpt provides configuration for unpack
@@ -300,6 +305,14 @@ func WithSnapshotterPlatformCheck() UnpackOpt {
 	}
 }
 
+// WithUnpackDuplicationSuppressor sets `DuplicationSuppressor` on the UnpackConfig.
+func WithUnpackDuplicationSuppressor(suppressor kmutex.KeyedLocker) UnpackOpt {
+	return func(ctx context.Context, uc *UnpackConfig) error {
+		uc.DuplicationSuppressor = suppressor
+		return nil
+	}
+}
+
 func (i *image) Unpack(ctx context.Context, snapshotterName string, opts ...UnpackOpt) error {
 	ctx, done, err := i.client.WithLease(ctx)
 	if err != nil {

vendor/github.com/containerd/containerd/pkg/kmutex/kmutex.go

@@ -0,0 +1,105 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+// Package kmutex provides synchronization primitives to lock/unlock resource by unique key.
+package kmutex
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"golang.org/x/sync/semaphore"
+)
+
+// KeyedLocker is the interface for acquiring locks based on string.
+type KeyedLocker interface {
+	Lock(ctx context.Context, key string) error
+	Unlock(key string)
+}
+
+func New() KeyedLocker {
+	return newKeyMutex()
+}
+
+func newKeyMutex() *keyMutex {
+	return &keyMutex{
+		locks: make(map[string]*klock),
+	}
+}
+
+type keyMutex struct {
+	mu sync.Mutex
+
+	locks map[string]*klock
+}
+
+type klock struct {
+	*semaphore.Weighted
+	ref int
+}
+
+func (km *keyMutex) Lock(ctx context.Context, key string) error {
+	km.mu.Lock()
+
+	l, ok := km.locks[key]
+	if !ok {
+		km.locks[key] = &klock{
+			Weighted: semaphore.NewWeighted(1),
+		}
+		l = km.locks[key]
+	}
+	l.ref++
+	km.mu.Unlock()
+
+	if err := l.Acquire(ctx, 1); err != nil {
+		km.mu.Lock()
+		defer km.mu.Unlock()
+
+		l.ref--
+
+		if l.ref < 0 {
+			panic(fmt.Errorf("kmutex: release of unlocked key %v", key))
+		}
+
+		if l.ref == 0 {
+			delete(km.locks, key)
+		}
+		return err
+	}
+	return nil
+}
+
+func (km *keyMutex) Unlock(key string) {
+	km.mu.Lock()
+	defer km.mu.Unlock()
+
+	l, ok := km.locks[key]
+	if !ok {
+		panic(fmt.Errorf("kmutex: unlock of unlocked key %v", key))
+	}
+	l.Release(1)
+
+	l.ref--
+
+	if l.ref < 0 {
+		panic(fmt.Errorf("kmutex: released of unlocked key %v", key))
+	}
+
+	if l.ref == 0 {
+		delete(km.locks, key)
+	}
+}

vendor/github.com/containerd/containerd/pkg/kmutex/noop.go

@@ -0,0 +1,33 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package kmutex
+
+import "context"
+
+func NewNoop() KeyedLocker {
+	return &noopMutex{}
+}
+
+type noopMutex struct {
+}
+
+func (*noopMutex) Lock(_ context.Context, _ string) error {
+	return nil
+}
+
+func (*noopMutex) Unlock(_ string) {
+}

vendor/github.com/containerd/containerd/unpacker.go

@@ -32,6 +32,7 @@ import (
 	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/mount"
+	"github.com/containerd/containerd/pkg/kmutex"
 	"github.com/containerd/containerd/platforms"
 	"github.com/containerd/containerd/snapshots"
 	"github.com/opencontainers/go-digest"
@@ -59,7 +60,9 @@ func (c *Client) newUnpacker(ctx context.Context, rCtx *RemoteContext) (*unpacke
 	if err != nil {
 		return nil, err
 	}
-	var config UnpackConfig
+	var config = UnpackConfig{
+		DuplicationSuppressor: kmutex.NewNoop(),
+	}
 	for _, o := range rCtx.UnpackOpts {
 		if err := o(ctx, &config); err != nil {
 			return nil, err
@@ -127,15 +130,20 @@ func (u *unpacker) unpack(
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-EachLayer:
-	for i, desc := range layers {
+	doUnpackFn := func(i int, desc ocispec.Descriptor) error {
 		parent := identity.ChainID(chain)
 		chain = append(chain, diffIDs[i])
-
 		chainID := identity.ChainID(chain).String()
+
+		unlock, err := u.lockSnChainID(ctx, chainID)
+		if err != nil {
+			return err
+		}
+		defer unlock()
+
 		if _, err := sn.Stat(ctx, chainID); err == nil {
 			// no need to handle
-			continue
+			return nil
 		} else if !errdefs.IsNotFound(err) {
 			return fmt.Errorf("failed to stat snapshot %s: %w", chainID, err)
 		}
@@ -167,7 +175,7 @@ EachLayer:
 						log.G(ctx).WithField("key", key).WithField("chainid", chainID).Debug("extraction snapshot already exists, chain id not found")
 					} else {
 						// no need to handle, snapshot now found with chain id
-						continue EachLayer
+						return nil
 					}
 				} else {
 					return fmt.Errorf("failed to prepare extraction snapshot %q: %w", key, err)
@@ -227,7 +235,7 @@ EachLayer:
 		if err = sn.Commit(ctx, chainID, key, opts...); err != nil {
 			abort()
 			if errdefs.IsAlreadyExists(err) {
-				continue
+				return nil
 			}
 			return fmt.Errorf("failed to commit snapshot %s: %w", key, err)
 		}
@@ -243,7 +251,13 @@ EachLayer:
 		if _, err := cs.Update(ctx, cinfo, "labels.containerd.io/uncompressed"); err != nil {
 			return err
 		}
+		return nil
+	}
 
+	for i, desc := range layers {
+		if err := doUnpackFn(i, desc); err != nil {
+			return err
+		}
 	}
 
 	chainID := identity.ChainID(chain).String()
@@ -271,17 +285,22 @@ func (u *unpacker) fetch(ctx context.Context, h images.Handler, layers []ocispec
 		desc := desc
 		i := i
 
-		if u.limiter != nil {
-			if err := u.limiter.Acquire(ctx, 1); err != nil {
-				return err
-			}
+		if err := u.acquire(ctx); err != nil {
+			return err
 		}
 
 		eg.Go(func() error {
-			_, err := h.Handle(ctx2, desc)
-			if u.limiter != nil {
-				u.limiter.Release(1)
+			unlock, err := u.lockBlobDescriptor(ctx2, desc)
+			if err != nil {
+				u.release()
+				return err
 			}
+
+			_, err = h.Handle(ctx2, desc)
+
+			unlock()
+			u.release()
+
 			if err != nil && !errors.Is(err, images.ErrSkipDesc) {
 				return err
 			}
@@ -306,7 +325,13 @@ func (u *unpacker) handlerWrapper(
 			layers = map[digest.Digest][]ocispec.Descriptor{}
 		)
 		return images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
+			unlock, err := u.lockBlobDescriptor(ctx, desc)
+			if err != nil {
+				return nil, err
+			}
+
 			children, err := f.Handle(ctx, desc)
+			unlock()
 			if err != nil {
 				return children, err
 			}
@@ -349,6 +374,50 @@ func (u *unpacker) handlerWrapper(
 	}, eg
 }
 
+func (u *unpacker) acquire(ctx context.Context) error {
+	if u.limiter == nil {
+		return nil
+	}
+	return u.limiter.Acquire(ctx, 1)
+}
+
+func (u *unpacker) release() {
+	if u.limiter == nil {
+		return
+	}
+	u.limiter.Release(1)
+}
+
+func (u *unpacker) lockSnChainID(ctx context.Context, chainID string) (func(), error) {
+	key := u.makeChainIDKeyWithSnapshotter(chainID)
+
+	if err := u.config.DuplicationSuppressor.Lock(ctx, key); err != nil {
+		return nil, err
+	}
+	return func() {
+		u.config.DuplicationSuppressor.Unlock(key)
+	}, nil
+}
+
+func (u *unpacker) lockBlobDescriptor(ctx context.Context, desc ocispec.Descriptor) (func(), error) {
+	key := u.makeBlobDescriptorKey(desc)
+
+	if err := u.config.DuplicationSuppressor.Lock(ctx, key); err != nil {
+		return nil, err
+	}
+	return func() {
+		u.config.DuplicationSuppressor.Unlock(key)
+	}, nil
+}
+
+func (u *unpacker) makeChainIDKeyWithSnapshotter(chainID string) string {
+	return fmt.Sprintf("sn://%s/%v", u.snapshotter, chainID)
+}
+
+func (u *unpacker) makeBlobDescriptorKey(desc ocispec.Descriptor) string {
+	return fmt.Sprintf("blob://%v", desc.Digest)
+}
+
 func uniquePart() string {
 	t := time.Now()
 	var b [3]byte

vendor/github.com/containerd/containerd/version/version.go

@@ -23,7 +23,7 @@ var (
 	Package = "github.com/containerd/containerd"
 
 	// Version holds the complete version number. Filled in at linking time.
-	Version = "1.6.2+unknown"
+	Version = "1.6.3+unknown"
 
 	// Revision is filled with the VCS (e.g. git) revision being used to build
 	// the program at linking time.

vendor/modules.txt

@@ -151,7 +151,7 @@ github.com/containerd/cgroups/v2/stats
 # github.com/containerd/console v1.0.3
 ## explicit; go 1.13
 github.com/containerd/console
-# github.com/containerd/containerd v1.6.2
+# github.com/containerd/containerd v1.6.3
 ## explicit; go 1.17
 github.com/containerd/containerd
 github.com/containerd/containerd/api/events
@@ -200,6 +200,7 @@ github.com/containerd/containerd/oci
 github.com/containerd/containerd/pkg/apparmor
 github.com/containerd/containerd/pkg/cap
 github.com/containerd/containerd/pkg/dialer
+github.com/containerd/containerd/pkg/kmutex
 github.com/containerd/containerd/pkg/seccomp
 github.com/containerd/containerd/pkg/userns
 github.com/containerd/containerd/platforms