Skip to content

Commit 00d102d

Browse files
fuweidfuweid
committed
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: #4984 Signed-off-by: Wei Fu <[email protected]>
1 parent 320ef91 commit 00d102d

7 files changed

Lines changed: 910 additions & 6 deletions

File tree

‎integration/build_local_containerd_helper_test.go‎

Lines changed: 209 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
/*
2+
Copyright The containerd Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package integration
18+
19+
import (
20+
"context"
21+
"fmt"
22+
"path/filepath"
23+
"sync"
24+
"testing"
25+
26+
"github.com/containerd/containerd"
27+
"github.com/containerd/containerd/content"
28+
"github.com/containerd/containerd/leases"
29+
"github.com/containerd/containerd/pkg/cri/constants"
30+
"github.com/containerd/containerd/platforms"
31+
"github.com/containerd/containerd/plugin"
32+
"github.com/containerd/containerd/services"
33+
ctrdsrv "github.com/containerd/containerd/services/server"
34+
srvconfig "github.com/containerd/containerd/services/server/config"
35+
"github.com/containerd/containerd/snapshots"
36+
37+
// NOTE: Importing containerd plugin(s) to build functionality in
38+
// client side, which means there is no need to up server. It can
39+
// prevent interference from testing with the same image.
40+
containersapi "github.com/containerd/containerd/api/services/containers/v1"
41+
diffapi "github.com/containerd/containerd/api/services/diff/v1"
42+
imagesapi "github.com/containerd/containerd/api/services/images/v1"
43+
introspectionapi "github.com/containerd/containerd/api/services/introspection/v1"
44+
namespacesapi "github.com/containerd/containerd/api/services/namespaces/v1"
45+
tasksapi "github.com/containerd/containerd/api/services/tasks/v1"
46+
_ "github.com/containerd/containerd/diff/walking/plugin"
47+
"github.com/containerd/containerd/events/exchange"
48+
_ "github.com/containerd/containerd/events/plugin"
49+
_ "github.com/containerd/containerd/gc/scheduler"
50+
_ "github.com/containerd/containerd/leases/plugin"
51+
_ "github.com/containerd/containerd/runtime/v2"
52+
_ "github.com/containerd/containerd/runtime/v2/runc/options"
53+
_ "github.com/containerd/containerd/services/containers"
54+
_ "github.com/containerd/containerd/services/content"
55+
_ "github.com/containerd/containerd/services/diff"
56+
_ "github.com/containerd/containerd/services/events"
57+
_ "github.com/containerd/containerd/services/images"
58+
_ "github.com/containerd/containerd/services/introspection"
59+
_ "github.com/containerd/containerd/services/leases"
60+
_ "github.com/containerd/containerd/services/namespaces"
61+
_ "github.com/containerd/containerd/services/snapshots"
62+
_ "github.com/containerd/containerd/services/tasks"
63+
_ "github.com/containerd/containerd/services/version"
64+
65+
"github.com/stretchr/testify/assert"
66+
)
67+
68+
var (
69+
loadPluginOnce sync.Once
70+
loadedPlugins []*plugin.Registration
71+
loadedPluginsErr error
72+
)
73+
74+
// buildLocalContainerdClient is to return containerd client with initialized
75+
// core plugins in local.
76+
func buildLocalContainerdClient(t *testing.T, tmpDir string) *containerd.Client {
77+
ctx := context.Background()
78+
79+
// load plugins
80+
loadPluginOnce.Do(func() {
81+
loadedPlugins, loadedPluginsErr = ctrdsrv.LoadPlugins(ctx, &srvconfig.Config{})
82+
assert.NoError(t, loadedPluginsErr)
83+
})
84+
85+
// init plugins
86+
var (
87+
// TODO: Remove this in 2.0 and let event plugin crease it
88+
events = exchange.NewExchange()
89+
90+
initialized = plugin.NewPluginSet()
91+
92+
// NOTE: plugin.Set doesn't provide the way to get all the same
93+
// type plugins. lastInitContext is used to record the last
94+
// initContext and work with getServicesOpts.
95+
lastInitContext *plugin.InitContext
96+
97+
config = &srvconfig.Config{
98+
Version: 2,
99+
Root: filepath.Join(tmpDir, "root"),
100+
State: filepath.Join(tmpDir, "state"),
101+
}
102+
)
103+
104+
for _, p := range loadedPlugins {
105+
initContext := plugin.NewContext(
106+
ctx,
107+
p,
108+
initialized,
109+
config.Root,
110+
config.State,
111+
)
112+
initContext.Events = events
113+
114+
// load the plugin specific configuration if it is provided
115+
if p.Config != nil {
116+
pc, err := config.Decode(p)
117+
assert.NoError(t, err)
118+
119+
initContext.Config = pc
120+
}
121+
122+
result := p.Init(initContext)
123+
assert.NoError(t, initialized.Add(result))
124+
125+
_, err := result.Instance()
126+
assert.NoError(t, err)
127+
128+
lastInitContext = initContext
129+
}
130+
131+
servicesOpts, err := getServicesOpts(lastInitContext)
132+
assert.NoError(t, err)
133+
134+
client, err := containerd.New(
135+
"",
136+
containerd.WithDefaultNamespace(constants.K8sContainerdNamespace),
137+
containerd.WithDefaultPlatform(platforms.Default()),
138+
containerd.WithServices(servicesOpts...),
139+
)
140+
assert.NoError(t, err)
141+
142+
return client
143+
}
144+
145+
// getServicesOpts get service options from plugin context.
146+
//
147+
// TODO(fuweid): It is copied from pkg/cri/cri.go. Should we make it as helper?
148+
func getServicesOpts(ic *plugin.InitContext) ([]containerd.ServicesOpt, error) {
149+
var opts []containerd.ServicesOpt
150+
for t, fn := range map[plugin.Type]func(interface{}) containerd.ServicesOpt{
151+
plugin.EventPlugin: func(i interface{}) containerd.ServicesOpt {
152+
return containerd.WithEventService(i.(containerd.EventService))
153+
},
154+
plugin.LeasePlugin: func(i interface{}) containerd.ServicesOpt {
155+
return containerd.WithLeasesService(i.(leases.Manager))
156+
},
157+
} {
158+
i, err := ic.Get(t)
159+
if err != nil {
160+
return nil, fmt.Errorf("failed to get %q plugin: %w", t, err)
161+
}
162+
opts = append(opts, fn(i))
163+
}
164+
plugins, err := ic.GetByType(plugin.ServicePlugin)
165+
if err != nil {
166+
return nil, fmt.Errorf("failed to get service plugin: %w", err)
167+
}
168+
169+
for s, fn := range map[string]func(interface{}) containerd.ServicesOpt{
170+
services.ContentService: func(s interface{}) containerd.ServicesOpt {
171+
return containerd.WithContentStore(s.(content.Store))
172+
},
173+
services.ImagesService: func(s interface{}) containerd.ServicesOpt {
174+
return containerd.WithImageClient(s.(imagesapi.ImagesClient))
175+
},
176+
services.SnapshotsService: func(s interface{}) containerd.ServicesOpt {
177+
return containerd.WithSnapshotters(s.(map[string]snapshots.Snapshotter))
178+
},
179+
services.ContainersService: func(s interface{}) containerd.ServicesOpt {
180+
return containerd.WithContainerClient(s.(containersapi.ContainersClient))
181+
},
182+
services.TasksService: func(s interface{}) containerd.ServicesOpt {
183+
return containerd.WithTaskClient(s.(tasksapi.TasksClient))
184+
},
185+
services.DiffService: func(s interface{}) containerd.ServicesOpt {
186+
return containerd.WithDiffClient(s.(diffapi.DiffClient))
187+
},
188+
services.NamespacesService: func(s interface{}) containerd.ServicesOpt {
189+
return containerd.WithNamespaceClient(s.(namespacesapi.NamespacesClient))
190+
},
191+
services.IntrospectionService: func(s interface{}) containerd.ServicesOpt {
192+
return containerd.WithIntrospectionClient(s.(introspectionapi.IntrospectionClient))
193+
},
194+
} {
195+
p := plugins[s]
196+
if p == nil {
197+
return nil, fmt.Errorf("service %q not found", s)
198+
}
199+
i, err := p.Instance()
200+
if err != nil {
201+
return nil, fmt.Errorf("failed to get instance of service %q: %w", s, err)
202+
}
203+
if i == nil {
204+
return nil, fmt.Errorf("instance of service %q not found", s)
205+
}
206+
opts = append(opts, fn(i))
207+
}
208+
return opts, nil
209+
}

‎integration/build_local_containerd_helper_test_linux.go‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
/*
2+
Copyright The containerd Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package integration
18+
19+
import (
20+
// Register for linux platforms
21+
_ "github.com/containerd/containerd/runtime/v1/linux"
22+
_ "github.com/containerd/containerd/snapshots/overlay/plugin"
23+
)

0 commit comments

Comments
 (0)