feature: support image pull progress timeout

Kubelet sends the PullImage request without timeout, because the image size
is unknown and timeout is hard to defined. The pulling request might run
into 0B/s speed, if containerd can't receive any packet in that connection.
For this case, the containerd should cancel the PullImage request.

Although containerd provides ingester manager to track the progress of pulling
request, for example `ctr image pull` shows the console progress bar, it needs
more CPU resources to open/read the ingested files to get status.

In order to support progress timeout feature with lower overhead, this
patch uses http.RoundTripper wrapper to track active progress. That
wrapper will increase active-request number and return the
countingReadCloser wrapper for http.Response.Body. Each bytes-read
can be count and the active-request number will be descreased when the
countingReadCloser wrapper has been closed. For the progress tracker,
it can check the active-request number and bytes-read at intervals. If
there is no any progress, the progress tracker should cancel the
request.

NOTE: For each blob data, the containerd will make sure that the content
writer is opened before sending http request to the registry. Therefore, the
progress reporter can rely on the active-request number.

fixed: containerd#4984

Signed-off-by: Wei Fu <[email protected]>

integration/build_local_containerd_helper_test.go

@@ -0,0 +1,209 @@
+/*
+   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 integration
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"github.com/containerd/containerd"
+	"github.com/containerd/containerd/content"
+	"github.com/containerd/containerd/leases"
+	"github.com/containerd/containerd/pkg/cri/constants"
+	"github.com/containerd/containerd/platforms"
+	"github.com/containerd/containerd/plugin"
+	"github.com/containerd/containerd/services"
+	ctrdsrv "github.com/containerd/containerd/services/server"
+	srvconfig "github.com/containerd/containerd/services/server/config"
+	"github.com/containerd/containerd/snapshots"
+
+	// NOTE: Importing containerd plugin(s) to build functionality in
+	// client side, which means there is no need to up server. It can
+	// prevent interference from testing with the same image.
+	containersapi "github.com/containerd/containerd/api/services/containers/v1"
+	diffapi "github.com/containerd/containerd/api/services/diff/v1"
+	imagesapi "github.com/containerd/containerd/api/services/images/v1"
+	introspectionapi "github.com/containerd/containerd/api/services/introspection/v1"
+	namespacesapi "github.com/containerd/containerd/api/services/namespaces/v1"
+	tasksapi "github.com/containerd/containerd/api/services/tasks/v1"
+	_ "github.com/containerd/containerd/diff/walking/plugin"
+	"github.com/containerd/containerd/events/exchange"
+	_ "github.com/containerd/containerd/events/plugin"
+	_ "github.com/containerd/containerd/gc/scheduler"
+	_ "github.com/containerd/containerd/leases/plugin"
+	_ "github.com/containerd/containerd/runtime/v2"
+	_ "github.com/containerd/containerd/runtime/v2/runc/options"
+	_ "github.com/containerd/containerd/services/containers"
+	_ "github.com/containerd/containerd/services/content"
+	_ "github.com/containerd/containerd/services/diff"
+	_ "github.com/containerd/containerd/services/events"
+	_ "github.com/containerd/containerd/services/images"
+	_ "github.com/containerd/containerd/services/introspection"
+	_ "github.com/containerd/containerd/services/leases"
+	_ "github.com/containerd/containerd/services/namespaces"
+	_ "github.com/containerd/containerd/services/snapshots"
+	_ "github.com/containerd/containerd/services/tasks"
+	_ "github.com/containerd/containerd/services/version"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	loadPluginOnce   sync.Once
+	loadedPlugins    []*plugin.Registration
+	loadedPluginsErr error
+)
+
+// buildLocalContainerdClient is to return containerd client with initialized
+// core plugins in local.
+func buildLocalContainerdClient(t *testing.T, tmpDir string) *containerd.Client {
+	ctx := context.Background()
+
+	// load plugins
+	loadPluginOnce.Do(func() {
+		loadedPlugins, loadedPluginsErr = ctrdsrv.LoadPlugins(ctx, &srvconfig.Config{})
+		assert.NoError(t, loadedPluginsErr)
+	})
+
+	// init plugins
+	var (
+		// TODO: Remove this in 2.0 and let event plugin crease it
+		events = exchange.NewExchange()
+
+		initialized = plugin.NewPluginSet()
+
+		// NOTE: plugin.Set doesn't provide the way to get all the same
+		// type plugins. lastInitContext is used to record the last
+		// initContext and work with getServicesOpts.
+		lastInitContext *plugin.InitContext
+
+		config = &srvconfig.Config{
+			Version: 2,
+			Root:    filepath.Join(tmpDir, "root"),
+			State:   filepath.Join(tmpDir, "state"),
+		}
+	)
+
+	for _, p := range loadedPlugins {
+		initContext := plugin.NewContext(
+			ctx,
+			p,
+			initialized,
+			config.Root,
+			config.State,
+		)
+		initContext.Events = events
+
+		// load the plugin specific configuration if it is provided
+		if p.Config != nil {
+			pc, err := config.Decode(p)
+			assert.NoError(t, err)
+
+			initContext.Config = pc
+		}
+
+		result := p.Init(initContext)
+		assert.NoError(t, initialized.Add(result))
+
+		_, err := result.Instance()
+		assert.NoError(t, err)
+
+		lastInitContext = initContext
+	}
+
+	servicesOpts, err := getServicesOpts(lastInitContext)
+	assert.NoError(t, err)
+
+	client, err := containerd.New(
+		"",
+		containerd.WithDefaultNamespace(constants.K8sContainerdNamespace),
+		containerd.WithDefaultPlatform(platforms.Default()),
+		containerd.WithServices(servicesOpts...),
+	)
+	assert.NoError(t, err)
+
+	return client
+}
+
+// getServicesOpts get service options from plugin context.
+//
+// TODO(fuweid): It is copied from pkg/cri/cri.go. Should we make it as helper?
+func getServicesOpts(ic *plugin.InitContext) ([]containerd.ServicesOpt, error) {
+	var opts []containerd.ServicesOpt
+	for t, fn := range map[plugin.Type]func(interface{}) containerd.ServicesOpt{
+		plugin.EventPlugin: func(i interface{}) containerd.ServicesOpt {
+			return containerd.WithEventService(i.(containerd.EventService))
+		},
+		plugin.LeasePlugin: func(i interface{}) containerd.ServicesOpt {
+			return containerd.WithLeasesService(i.(leases.Manager))
+		},
+	} {
+		i, err := ic.Get(t)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get %q plugin: %w", t, err)
+		}
+		opts = append(opts, fn(i))
+	}
+	plugins, err := ic.GetByType(plugin.ServicePlugin)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get service plugin: %w", err)
+	}
+
+	for s, fn := range map[string]func(interface{}) containerd.ServicesOpt{
+		services.ContentService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithContentStore(s.(content.Store))
+		},
+		services.ImagesService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithImageClient(s.(imagesapi.ImagesClient))
+		},
+		services.SnapshotsService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithSnapshotters(s.(map[string]snapshots.Snapshotter))
+		},
+		services.ContainersService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithContainerClient(s.(containersapi.ContainersClient))
+		},
+		services.TasksService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithTaskClient(s.(tasksapi.TasksClient))
+		},
+		services.DiffService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithDiffClient(s.(diffapi.DiffClient))
+		},
+		services.NamespacesService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithNamespaceClient(s.(namespacesapi.NamespacesClient))
+		},
+		services.IntrospectionService: func(s interface{}) containerd.ServicesOpt {
+			return containerd.WithIntrospectionClient(s.(introspectionapi.IntrospectionClient))
+		},
+	} {
+		p := plugins[s]
+		if p == nil {
+			return nil, fmt.Errorf("service %q not found", s)
+		}
+		i, err := p.Instance()
+		if err != nil {
+			return nil, fmt.Errorf("failed to get instance of service %q: %w", s, err)
+		}
+		if i == nil {
+			return nil, fmt.Errorf("instance of service %q not found", s)
+		}
+		opts = append(opts, fn(i))
+	}
+	return opts, nil
+}

integration/build_local_containerd_helper_test_linux.go

@@ -0,0 +1,23 @@
+/*
+   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 integration
+
+import (
+	// Register for linux platforms
+	_ "github.com/containerd/containerd/runtime/v1/linux"
+	_ "github.com/containerd/containerd/snapshots/overlay/plugin"
+)