Skip to content

Commit fbe0541

Browse files
smarterclaytonsmarterclayton
committed
Provide layers referenced by an image stream as layers subresource
Adds a new GET endpoint to an image stream as a subresource `layers` that returns an array of every layer referenced by the image stream and the tags and images included by the image. The subresource is fed by a store driven informer that caches and indexes only the layers. Clients get a 500 retry error if the cache has not initialized yet (the client will silently retry). Turns the registry access check for a given layer into an O(1) check instead of O(N) where N is the number of images in the image stream.
1 parent cbdaeb2 commit fbe0541

File tree

4 files changed

+293
-4
lines changed

4 files changed

+293
-4
lines changed

‎pkg/cmd/server/origin/legacy.go

+2
Original file line numberDiff line numberDiff line change
@@ -207,6 +207,8 @@ func LegacyStorage(storage map[schema.GroupVersion]map[string]rest.Storage) map[
207207

208208
case *imagestreametcd.REST:
209209
legacyStorage[resource] = &imagestreametcd.LegacyREST{REST: storage}
210+
case *imagestreametcd.LayersREST:
211+
delete(legacyStorage, resource)
210212

211213
case *routeetcd.REST:
212214
store := *storage.Store

‎pkg/image/apiserver/apiserver.go

+15-1
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@ type ExtraConfig struct {
5353
makeV1Storage sync.Once
5454
v1Storage map[string]rest.Storage
5555
v1StorageErr error
56+
startFns []func(<-chan struct{})
5657
}
5758

5859
type ImageAPIServerConfig struct {
@@ -107,6 +108,15 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
107108
return nil, err
108109
}
109110

111+
if err := s.GenericAPIServer.AddPostStartHook("image.openshift.io-apiserver-caches", func(context genericapiserver.PostStartHookContext) error {
112+
for _, fn := range c.ExtraConfig.startFns {
113+
go fn(context.StopCh)
114+
}
115+
return nil
116+
}); err != nil {
117+
return nil, err
118+
}
119+
110120
return s, nil
111121
}
112122

@@ -168,10 +178,13 @@ func (c *completedConfig) newV1RESTStorage() (map[string]rest.Storage, error) {
168178
whitelister = whitelist.WhitelistAllRegistries()
169179
}
170180

181+
imageLayerIndex := imagestreametcd.NewImageLayerIndex(imageStorage)
182+
c.ExtraConfig.startFns = append(c.ExtraConfig.startFns, imageLayerIndex.Run)
183+
171184
imageRegistry := image.NewRegistry(imageStorage)
172185
imageSignatureStorage := imagesignature.NewREST(imageClient.Image())
173186
imageStreamSecretsStorage := imagesecret.NewREST(coreClient)
174-
imageStreamStorage, imageStreamStatusStorage, internalImageStreamStorage, err := imagestreametcd.NewREST(c.GenericConfig.RESTOptionsGetter, c.ExtraConfig.RegistryHostnameRetriever, authorizationClient.SubjectAccessReviews(), c.ExtraConfig.LimitVerifier, whitelister)
187+
imageStreamStorage, imageStreamLayersStorage, imageStreamStatusStorage, internalImageStreamStorage, err := imagestreametcd.NewREST(c.GenericConfig.RESTOptionsGetter, c.ExtraConfig.RegistryHostnameRetriever, authorizationClient.SubjectAccessReviews(), c.ExtraConfig.LimitVerifier, whitelister, imageLayerIndex)
175188
if err != nil {
176189
return nil, fmt.Errorf("error building REST storage: %v", err)
177190
}
@@ -206,6 +219,7 @@ func (c *completedConfig) newV1RESTStorage() (map[string]rest.Storage, error) {
206219
v1Storage["imagesignatures"] = imageSignatureStorage
207220
v1Storage["imageStreams/secrets"] = imageStreamSecretsStorage
208221
v1Storage["imageStreams"] = imageStreamStorage
222+
v1Storage["imageStreams/layers"] = imageStreamLayersStorage
209223
v1Storage["imageStreams/status"] = imageStreamStatusStorage
210224
v1Storage["imageStreamImports"] = imageStreamImportStorage
211225
v1Storage["imageStreamImages"] = imageStreamImageStorage

‎pkg/image/registry/imagestream/etcd/etcd.go

+117-3
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package etcd
22

33
import (
4+
"k8s.io/apimachinery/pkg/api/errors"
45
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
56
"k8s.io/apimachinery/pkg/runtime"
67
apirequest "k8s.io/apiserver/pkg/endpoints/request"
@@ -46,7 +47,8 @@ func NewREST(
4647
subjectAccessReviewRegistry authorizationclient.SubjectAccessReviewInterface,
4748
limitVerifier imageadmission.LimitVerifier,
4849
registryWhitelister whitelist.RegistryWhitelister,
49-
) (*REST, *StatusREST, *InternalREST, error) {
50+
imageLayerIndex ImageLayerIndex,
51+
) (*REST, *LayersREST, *StatusREST, *InternalREST, error) {
5052
store := registry.Store{
5153
NewFunc: func() runtime.Object { return &imageapi.ImageStream{} },
5254
NewListFunc: func() runtime.Object { return &imageapi.ImageStreamList{} },
@@ -71,9 +73,11 @@ func NewREST(
7173
AttrFunc: storage.AttrFunc(storage.DefaultNamespaceScopedAttr).WithFieldMutation(imageapi.ImageStreamSelector),
7274
}
7375
if err := store.CompleteWithOptions(options); err != nil {
74-
return nil, nil, nil, err
76+
return nil, nil, nil, nil, err
7577
}
7678

79+
layersREST := &LayersREST{index: imageLayerIndex, store: &store}
80+
7781
statusStrategy := imagestream.NewStatusStrategy(strategy)
7882
statusStore := store
7983
statusStore.Decorator = nil
@@ -88,7 +92,7 @@ func NewREST(
8892
internalStore.UpdateStrategy = internalStrategy
8993

9094
internalREST := &InternalREST{store: &internalStore}
91-
return rest, statusREST, internalREST, nil
95+
return rest, layersREST, statusREST, internalREST, nil
9296
}
9397

9498
// StatusREST implements the REST endpoint for changing the status of an image stream.
@@ -138,6 +142,116 @@ func (r *InternalREST) Update(ctx apirequest.Context, name string, objInfo rest.
138142
return r.store.Update(ctx, name, objInfo, createValidation, updateValidation)
139143
}
140144

145+
// LayersREST implements the REST endpoint for changing both the spec and status of an image stream.
146+
type LayersREST struct {
147+
store *registry.Store
148+
index ImageLayerIndex
149+
}
150+
151+
var _ rest.Getter = &LayersREST{}
152+
153+
func (r *LayersREST) New() runtime.Object {
154+
return &imageapi.ImageStreamLayers{}
155+
}
156+
157+
// Get returns the layers for an image stream.
158+
func (r *LayersREST) Get(ctx apirequest.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
159+
if !r.index.HasSynced() {
160+
return nil, errors.NewServerTimeout(r.store.DefaultQualifiedResource, "get", 2)
161+
}
162+
obj, err := r.store.Get(ctx, name, options)
163+
if err != nil {
164+
return nil, err
165+
}
166+
is := obj.(*imageapi.ImageStream)
167+
isl := &imageapi.ImageStreamLayers{
168+
ObjectMeta: is.ObjectMeta,
169+
}
170+
171+
existing, existingBlobs := make(map[string]int), make(map[string]int)
172+
existingImage, existingImageBlobs := make(map[string][]int), make(map[string][]int)
173+
for tag, status := range is.Status.Tags {
174+
var seen map[string]struct{}
175+
for _, item := range status.Items {
176+
if len(item.Image) == 0 {
177+
continue
178+
}
179+
if _, ok := seen[item.Image]; ok {
180+
continue
181+
}
182+
if len(status.Items) > 1 {
183+
if seen == nil {
184+
seen = make(map[string]struct{})
185+
}
186+
seen[item.Image] = struct{}{}
187+
}
188+
if indices, ok := existingImage[item.Image]; ok {
189+
for _, index := range indices {
190+
ref := &isl.Layers[index]
191+
ref.Tags = append(ref.Tags, tag)
192+
}
193+
if indices, ok := existingImageBlobs[item.Image]; ok {
194+
for _, index := range indices {
195+
ref := &isl.Blobs[index]
196+
ref.Tags = append(ref.Tags, tag)
197+
}
198+
}
199+
continue
200+
}
201+
202+
obj, _, _ := r.index.GetIndexer().GetByKey(item.Image)
203+
entry, ok := obj.(*ImageLayers)
204+
if !ok {
205+
continue
206+
}
207+
indices := make([]int, 0, len(entry.Layers))
208+
for _, layer := range entry.Layers {
209+
if index, ok := existing[layer.Name]; ok {
210+
ref := &isl.Layers[index]
211+
ref.Tags = append(ref.Tags, tag)
212+
ref.ImageIDs = append(ref.ImageIDs, item.Image)
213+
indices = append(indices, index)
214+
continue
215+
}
216+
index := len(isl.Layers)
217+
existing[layer.Name] = index
218+
indices = append(indices, index)
219+
isl.Layers = append(isl.Layers, imageapi.ImageLayerReference{
220+
Name: layer.Name,
221+
LayerSize: layer.LayerSize,
222+
MediaType: layer.MediaType,
223+
Tags: []string{tag},
224+
ImageIDs: []string{item.Image},
225+
})
226+
}
227+
existingImage[item.Image] = indices
228+
229+
blobIndices := make([]int, 0, 1)
230+
if layer := entry.Manifest; layer != nil {
231+
if index, ok := existingBlobs[layer.Name]; ok {
232+
ref := &isl.Blobs[index]
233+
ref.Tags = append(ref.Tags, tag)
234+
ref.ImageIDs = append(ref.ImageIDs, item.Image)
235+
blobIndices = append(blobIndices, index)
236+
continue
237+
}
238+
index := len(isl.Blobs)
239+
existingBlobs[layer.Name] = index
240+
blobIndices = append(blobIndices, index)
241+
isl.Blobs = append(isl.Blobs, imageapi.ImageLayerReference{
242+
Name: layer.Name,
243+
LayerSize: layer.LayerSize,
244+
MediaType: layer.MediaType,
245+
Tags: []string{tag},
246+
ImageIDs: []string{item.Image},
247+
})
248+
}
249+
existingImageBlobs[item.Image] = blobIndices
250+
}
251+
}
252+
return isl, nil
253+
}
254+
141255
// LegacyREST allows us to wrap and alter some behavior
142256
type LegacyREST struct {
143257
*REST

‎pkg/image/registry/imagestream/etcd/image.go

+159
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
package etcd
2+
3+
import (
4+
"fmt"
5+
6+
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
7+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
8+
"k8s.io/apimachinery/pkg/runtime"
9+
"k8s.io/apimachinery/pkg/runtime/schema"
10+
"k8s.io/apimachinery/pkg/types"
11+
"k8s.io/apimachinery/pkg/watch"
12+
apirequest "k8s.io/apiserver/pkg/endpoints/request"
13+
"k8s.io/apiserver/pkg/registry/rest"
14+
"k8s.io/client-go/tools/cache"
15+
16+
imageapi "github.com/openshift/origin/pkg/image/apis/image"
17+
)
18+
19+
type ImageLayerIndex interface {
20+
cache.SharedIndexInformer
21+
}
22+
23+
type ImageStore interface {
24+
rest.Watcher
25+
rest.Lister
26+
}
27+
28+
func NewImageLayerIndex(store ImageStore) ImageLayerIndex {
29+
ctx := apirequest.NewContext()
30+
informer := cache.NewSharedIndexInformer(&cache.ListWatch{
31+
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
32+
obj, err := store.List(ctx, &metainternalversion.ListOptions{
33+
ResourceVersion: options.ResourceVersion,
34+
})
35+
if err != nil {
36+
return nil, err
37+
}
38+
list, ok := obj.(*imageapi.ImageList)
39+
if !ok {
40+
return nil, fmt.Errorf("unexpected store type %T for layer index", obj)
41+
}
42+
out := &metainternalversion.List{
43+
Items: make([]runtime.Object, len(list.Items)),
44+
}
45+
for i, image := range list.Items {
46+
out.Items[i] = &ImageLayers{
47+
Name: image.Name,
48+
Layers: image.DockerImageLayers,
49+
Manifest: manifestFromImage(&image),
50+
}
51+
}
52+
return out, nil
53+
},
54+
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
55+
w, err := store.Watch(ctx, &metainternalversion.ListOptions{
56+
ResourceVersion: options.ResourceVersion,
57+
})
58+
if err != nil {
59+
return nil, err
60+
}
61+
return watch.Filter(w, func(in watch.Event) (out watch.Event, keep bool) {
62+
if in.Object == nil {
63+
return in, true
64+
}
65+
image, ok := in.Object.(*imageapi.Image)
66+
if !ok {
67+
return in, true
68+
}
69+
in.Object = &ImageLayers{
70+
Name: image.Name,
71+
Layers: image.DockerImageLayers,
72+
Manifest: manifestFromImage(image),
73+
}
74+
return in, true
75+
}), nil
76+
},
77+
}, &ImageLayers{}, 0, cache.Indexers{
78+
"layers": func(obj interface{}) ([]string, error) {
79+
entry, ok := obj.(*ImageLayers)
80+
if !ok {
81+
return nil, fmt.Errorf("unexpected cache object %T", obj)
82+
}
83+
keys := make([]string, 0, len(entry.Layers))
84+
for _, layer := range entry.Layers {
85+
keys = append(keys, layer.Name)
86+
}
87+
return keys, nil
88+
},
89+
})
90+
return informer
91+
}
92+
93+
// manifestFromImage attempts to find a manifest blob description from
94+
// an image. Images older than schema2 in Docker do not have a manifest blob.
95+
func manifestFromImage(image *imageapi.Image) *imageapi.ImageLayer {
96+
if image.DockerImageManifestMediaType != "application/vnd.docker.distribution.manifest.v2+json" {
97+
return nil
98+
}
99+
return &imageapi.ImageLayer{
100+
Name: image.DockerImageMetadata.ID,
101+
MediaType: image.DockerImageManifestMediaType,
102+
}
103+
}
104+
105+
type ImageLayers struct {
106+
Name string
107+
Manifest *imageapi.ImageLayer
108+
Layers []imageapi.ImageLayer
109+
}
110+
111+
var _ metav1.Object = &ImageLayers{}
112+
113+
func (l *ImageLayers) GetObjectKind() schema.ObjectKind { return &metav1.TypeMeta{} }
114+
func (l *ImageLayers) DeepCopyObject() runtime.Object {
115+
var layers []imageapi.ImageLayer
116+
if l.Layers != nil {
117+
layers = make([]imageapi.ImageLayer, len(l.Layers))
118+
copy(layers, l.Layers)
119+
}
120+
return &ImageLayers{
121+
Name: l.Name,
122+
Layers: layers,
123+
}
124+
}
125+
126+
// TODO: fix client-go/cache to not need this
127+
128+
func (l *ImageLayers) GetNamespace() string { return "" }
129+
func (l *ImageLayers) SetNamespace(namespace string) {}
130+
func (l *ImageLayers) GetName() string { return l.Name }
131+
func (l *ImageLayers) SetName(name string) {}
132+
func (l *ImageLayers) GetGenerateName() string { return "" }
133+
func (l *ImageLayers) SetGenerateName(name string) {}
134+
func (l *ImageLayers) GetUID() types.UID { return "" }
135+
func (l *ImageLayers) SetUID(uid types.UID) {}
136+
func (l *ImageLayers) GetResourceVersion() string { return "" }
137+
func (l *ImageLayers) SetResourceVersion(version string) {}
138+
func (l *ImageLayers) GetGeneration() int64 { return 0 }
139+
func (l *ImageLayers) SetGeneration(generation int64) {}
140+
func (l *ImageLayers) GetSelfLink() string { return "" }
141+
func (l *ImageLayers) SetSelfLink(selfLink string) {}
142+
func (l *ImageLayers) GetCreationTimestamp() metav1.Time { return metav1.Time{} }
143+
func (l *ImageLayers) SetCreationTimestamp(timestamp metav1.Time) {}
144+
func (l *ImageLayers) GetDeletionTimestamp() *metav1.Time { return nil }
145+
func (l *ImageLayers) SetDeletionTimestamp(timestamp *metav1.Time) {}
146+
func (l *ImageLayers) GetDeletionGracePeriodSeconds() *int64 { return nil }
147+
func (l *ImageLayers) SetDeletionGracePeriodSeconds(*int64) {}
148+
func (l *ImageLayers) GetLabels() map[string]string { return nil }
149+
func (l *ImageLayers) SetLabels(labels map[string]string) {}
150+
func (l *ImageLayers) GetAnnotations() map[string]string { return nil }
151+
func (l *ImageLayers) SetAnnotations(annotations map[string]string) {}
152+
func (l *ImageLayers) GetInitializers() *metav1.Initializers { return nil }
153+
func (l *ImageLayers) SetInitializers(initializers *metav1.Initializers) {}
154+
func (l *ImageLayers) GetFinalizers() []string { return nil }
155+
func (l *ImageLayers) SetFinalizers(finalizers []string) {}
156+
func (l *ImageLayers) GetOwnerReferences() []metav1.OwnerReference { return nil }
157+
func (l *ImageLayers) SetOwnerReferences([]metav1.OwnerReference) {}
158+
func (l *ImageLayers) GetClusterName() string { return "" }
159+
func (l *ImageLayers) SetClusterName(clusterName string) {}

0 commit comments

Comments
 (0)