diff --git a/cmd/machine-config-controller/start.go b/cmd/machine-config-controller/start.go index cbc7308e5b..50343c58f8 100644 --- a/cmd/machine-config-controller/start.go +++ b/cmd/machine-config-controller/start.go @@ -92,28 +92,32 @@ func runStartCmd(_ *cobra.Command, _ []string) { return } - var inspectorFactory osimagestream.ImagesInspectorFactory - var inspectionCache *imageutils.FileInspectionCache - if startOpts.streamsCache != "" { - inspectionCache = imageutils.NewFileInspectionCache(path.Join(startOpts.streamsCache, "image-inspection.json"), 48*time.Hour) - } + syncer := imageutils.NewConfigMapCacheSyncer( + ctrlctx.KubeNamespacedInformerFactory.Core().V1().ConfigMaps(), + ctrlctx.ClientBuilder.KubeClientOrDie("inspection-cache-syncer"), + ctrlcommon.MCONamespace, + ctrlcommon.InspectionCacheConfigMapName, + osimagestream.NewCacheEntryFilter(), + osimagestream.NewImageStreamFileTransformer(), + ) + inspectionCache := imageutils.NewFileInspectionCache( + path.Join(startOpts.streamsCache, "image-inspection.json"), 48*time.Hour, syncer, + ) // OSImageStream must be the first controller to run: the blocking // EnsureOSImageStream call guarantees the CR exists before any other // controller starts, since render, node, and template depend on it // for OS image URLs. + var inspectorFactory osimagestream.ImagesInspectorFactory var cacheWarmer *pinnedimageset.CacheWarmer + var osImageStreamCtrl *osistreamctrl.Controller if osimagestream.IsFeatureEnabled(ctrlctx.FeatureGatesHandler) { - if inspectionCache != nil { - inspectorFactory = osimagestream.NewCachedImagesInspectorFactory( - &osimagestream.DefaultImagesInspectorFactory{}, - inspectionCache, - ) - } else { - inspectorFactory = &osimagestream.DefaultImagesInspectorFactory{} - } + inspectorFactory = osimagestream.NewCachedImagesInspectorFactory( + &osimagestream.DefaultImagesInspectorFactory{}, + inspectionCache, + ) - osImageStreamCtrl := osistreamctrl.New( + osImageStreamCtrl = osistreamctrl.New( ctrlctx.InformerFactory.Machineconfiguration().V1().OSImageStreams(), ctrlctx.InformerFactory.Machineconfiguration().V1().ControllerConfigs(), ctrlctx.ConfigInformerFactory.Config().V1().ClusterVersions(), @@ -128,41 +132,26 @@ func runStartCmd(_ *cobra.Command, _ []string) { ctrlctx.FeatureGatesHandler, inspectorFactory, ) + inspectionCache.RegisterEvicter(osImageStreamCtrl) - ctrlctx.InformerFactory.Start(ctx.Done()) - ctrlctx.ConfigInformerFactory.Start(ctx.Done()) - ctrlctx.OpenShiftConfigKubeNamespacedInformerFactory.Start(ctx.Done()) - ctrlctx.KubeNamespacedInformerFactory.Start(ctx.Done()) - ctrlctx.OperatorInformerFactory.Start(ctx.Done()) - - go osImageStreamCtrl.Run(ctx, 1) - - if err := osImageStreamCtrl.EnsureOSImageStream(ctx); err != nil { - klog.Fatalf("Failed to ensure OSImageStream: %v", err) - } - - if inspectionCache != nil { - inspectionCache.RegisterEvicter(osImageStreamCtrl) - - cacheWarmer = pinnedimageset.NewCacheWarmer( - ctrlctx.InformerFactory.Machineconfiguration().V1().PinnedImageSets().Lister(), - inspectorFactory, - ctrlcommon.NewSysContextFactory( - ctrlctx.InformerFactory.Machineconfiguration().V1().ControllerConfigs().Lister(), - ctrlctx.OpenShiftConfigKubeNamespacedInformerFactory.Core().V1().Secrets().Lister(), - ctrlctx.ConfigInformerFactory.Config().V1().Images().Lister(), - ctrlctx.OperatorInformerFactory.Operator().V1alpha1().ImageContentSourcePolicies().Lister(), - ctrlctx.ConfigInformerFactory.Config().V1().ImageDigestMirrorSets().Lister(), - ctrlctx.ConfigInformerFactory.Config().V1().ImageTagMirrorSets().Lister(), - ), - ) - inspectionCache.RegisterEvicter(cacheWarmer) - } + cacheWarmer = pinnedimageset.NewCacheWarmer( + ctrlctx.InformerFactory.Machineconfiguration().V1().PinnedImageSets().Lister(), + inspectorFactory, + ctrlcommon.NewSysContextFactory( + ctrlctx.InformerFactory.Machineconfiguration().V1().ControllerConfigs().Lister(), + ctrlctx.OpenShiftConfigKubeNamespacedInformerFactory.Core().V1().Secrets().Lister(), + ctrlctx.ConfigInformerFactory.Config().V1().Images().Lister(), + ctrlctx.OperatorInformerFactory.Operator().V1alpha1().ImageContentSourcePolicies().Lister(), + ctrlctx.ConfigInformerFactory.Config().V1().ImageDigestMirrorSets().Lister(), + ctrlctx.ConfigInformerFactory.Config().V1().ImageTagMirrorSets().Lister(), + ), + ) + inspectionCache.RegisterEvicter(cacheWarmer) } go ctrlcommon.StartMetricsListener(startOpts.promMetricsListenAddress, ctx.Done(), ctrlcommon.RegisterMCCMetrics, startOpts.tlsMinVersion, startOpts.tlsCipherSuites) - controllers := createControllers(ctrlctx, inspectionCache, inspectorFactory) + controllers := createControllers(ctrlctx, inspectionCache, inspectorFactory, cacheWarmer) draincontroller := drain.New( drain.DefaultConfig(), ctrlctx.KubeInformerFactory.Core().V1().Nodes(), @@ -189,15 +178,6 @@ func runStartCmd(_ *cobra.Command, _ []string) { klog.Fatalf("unable to start cert rotation controller: %v", err) } - pinnedImageSet := pinnedimageset.New( - ctrlctx.InformerFactory.Machineconfiguration().V1().PinnedImageSets(), - ctrlctx.InformerFactory.Machineconfiguration().V1().MachineConfigPools(), - ctrlctx.ClientBuilder.KubeClientOrDie("pinned-image-set-controller"), - ctrlctx.ClientBuilder.MachineConfigClientOrDie("pinned-image-set-controller"), - cacheWarmer, - ) - go pinnedImageSet.Run(ctx, 2) - // Start the shared factory informers that you need to use in your controller ctrlctx.InformerFactory.Start(ctrlctx.Stop) ctrlctx.KubeInformerFactory.Start(ctrlctx.Stop) @@ -211,6 +191,16 @@ func runStartCmd(_ *cobra.Command, _ []string) { close(ctrlctx.InformersStarted) + // Start the cache before any controller that consumes it has a chance to run. + inspectionCache.Start(ctx, 24*time.Hour, 10*time.Minute, 30*time.Second) + + if osImageStreamCtrl != nil { + go osImageStreamCtrl.Run(ctx, 1) + if err := osImageStreamCtrl.EnsureOSImageStream(ctx); err != nil { + klog.Fatalf("Failed to ensure OSImageStream: %v", err) + } + } + if ctrlcommon.IsBootImageControllerRequired(ctrlctx) { bootImageController := bootimagecontroller.New( ctrlctx.ClientBuilder.KubeClientOrDie("machine-set-boot-image-controller"), @@ -239,10 +229,6 @@ func runStartCmd(_ *cobra.Command, _ []string) { go draincontroller.Run(ctx, 5) go certrotationcontroller.Run(ctx, 1) - if inspectionCache != nil { - inspectionCache.StartEviction(ctx, 24*time.Hour, 10*time.Minute) - } - // wait here in this function until the context gets cancelled (which tells us when we are being shut down) <-ctx.Done() } @@ -266,7 +252,7 @@ func runStartCmd(_ *cobra.Command, _ []string) { panic("unreachable") } -func createControllers(ctx *ctrlcommon.ControllerContext, inspectionCache *imageutils.FileInspectionCache, inspectorFactory osimagestream.ImagesInspectorFactory) []ctrlcommon.Controller { +func createControllers(ctx *ctrlcommon.ControllerContext, inspectionCache *imageutils.FileInspectionCache, inspectorFactory osimagestream.ImagesInspectorFactory, pisCacheWarmer *pinnedimageset.CacheWarmer) []ctrlcommon.Controller { renderCtrl := render.New( ctx.InformerFactory.Machineconfiguration().V1().MachineConfigPools(), ctx.InformerFactory.Machineconfiguration().V1().MachineConfigs(), @@ -285,9 +271,7 @@ func createControllers(ctx *ctrlcommon.ControllerContext, inspectionCache *image ctx.FeatureGatesHandler, inspectorFactory, ) - if inspectionCache != nil { - inspectionCache.RegisterEvicter(renderCtrl) - } + inspectionCache.RegisterEvicter(renderCtrl) var controllers []ctrlcommon.Controller controllers = append(controllers, @@ -363,6 +347,13 @@ func createControllers(ctx *ctrlcommon.ControllerContext, inspectionCache *image ctx.ClientBuilder.KubeClientOrDie("internalreleaseimage-controller"), ctx.ClientBuilder.MachineConfigClientOrDie("internalreleaseimage-controller"), ), + pinnedimageset.New( + ctx.InformerFactory.Machineconfiguration().V1().PinnedImageSets(), + ctx.InformerFactory.Machineconfiguration().V1().MachineConfigPools(), + ctx.ClientBuilder.KubeClientOrDie("pinned-image-set-controller"), + ctx.ClientBuilder.MachineConfigClientOrDie("pinned-image-set-controller"), + pisCacheWarmer, + ), ) return controllers diff --git a/pkg/controller/common/constants.go b/pkg/controller/common/constants.go index b9357e861c..5565abed8d 100644 --- a/pkg/controller/common/constants.go +++ b/pkg/controller/common/constants.go @@ -198,4 +198,6 @@ const ( MachineConfigOperatorImagesConfigMapName string = "machine-config-operator-images" // The name of the machine-config-osimageurl ConfigMap. MachineConfigOSImageURLConfigMapName string = "machine-config-osimageurl" + // The name of the ConfigMap used to persist the image inspection cache across Pod restarts. + InspectionCacheConfigMapName string = "machine-config-image-inspection-cache" ) diff --git a/pkg/controller/pinnedimageset/cache_warmer_test.go b/pkg/controller/pinnedimageset/cache_warmer_test.go index f3641d3c9b..456d7864b5 100644 --- a/pkg/controller/pinnedimageset/cache_warmer_test.go +++ b/pkg/controller/pinnedimageset/cache_warmer_test.go @@ -71,7 +71,7 @@ func TestCacheWarmerWarmsOnPISChange(t *testing.T) { defer cancel() cache := imageutils.NewFileInspectionCache( - filepath.Join(t.TempDir(), "cache.json"), 48*time.Hour) + filepath.Join(t.TempDir(), "cache.json"), 48*time.Hour, nil) inspector := &fakeInspector{ inspectData: map[string]*types.ImageInspectInfo{ diff --git a/pkg/imageutils/cache_entry_transformer.go b/pkg/imageutils/cache_entry_transformer.go new file mode 100644 index 0000000000..974e9a7bfa --- /dev/null +++ b/pkg/imageutils/cache_entry_transformer.go @@ -0,0 +1,51 @@ +package imageutils + +// CacheEntryFilter decides whether a cache entry should be included in +// external persistence. Implementations must not mutate the input entry. +type CacheEntryFilter func(digest string, entry *InspectionCacheEntry) bool + +// CacheEntryTransformer returns a (possibly reduced) copy of an +// InspectionCacheEntry for external persistence. Implementations must not +// mutate the input entry. +type CacheEntryTransformer func(digest string, entry *InspectionCacheEntry) *InspectionCacheEntry + +// NewCacheFileTransformer returns a CacheEntryTransformer that applies a +// transformation function to a cached file matching the given path. Other +// files and labels are preserved. +func NewCacheFileTransformer(path string, transform func([]byte) ([]byte, error)) CacheEntryTransformer { + return func(_ string, entry *InspectionCacheEntry) *InspectionCacheEntry { + if entry.Files == nil { + return entry + } + _, ok := entry.Files[path] + if !ok { + return entry + } + + cp := entry.DeepCopy() + transformed, err := transform(cp.Files[path]) + if err != nil { + return entry + } + + cp.Files[path] = transformed + return cp + } +} + +// NewCacheEntryFilter returns a filter that accepts entries having at least +// one of the specified label keys. +func NewCacheEntryFilter(requiredLabelKeys ...string) CacheEntryFilter { + keys := make(map[string]struct{}, len(requiredLabelKeys)) + for _, k := range requiredLabelKeys { + keys[k] = struct{}{} + } + return func(_ string, entry *InspectionCacheEntry) bool { + for k := range entry.Labels { + if _, ok := keys[k]; ok { + return true + } + } + return false + } +} diff --git a/pkg/imageutils/cache_entry_transformer_test.go b/pkg/imageutils/cache_entry_transformer_test.go new file mode 100644 index 0000000000..f2c3846130 --- /dev/null +++ b/pkg/imageutils/cache_entry_transformer_test.go @@ -0,0 +1,40 @@ +package imageutils + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestEntryFilter_KeepsEntryWithMatchingLabel(t *testing.T) { + filter := NewCacheEntryFilter("io.openshift.os.streamclass", "io.openshift.release") + + entry := &InspectionCacheEntry{ + Labels: map[string]string{ + "io.openshift.os.streamclass": "rhel-9", + "vendor": "Red Hat", + }, + } + + assert.True(t, filter("sha256:aaa", entry)) +} + +func TestEntryFilter_ExcludesEntryWithoutMatchingLabels(t *testing.T) { + filter := NewCacheEntryFilter("io.openshift.os.streamclass", "io.openshift.release") + + entry := &InspectionCacheEntry{ + Labels: map[string]string{ + "vendor": "Red Hat", + "version": "9.4", + }, + } + + assert.False(t, filter("sha256:aaa", entry)) +} + +func TestEntryFilter_EmptyLabels(t *testing.T) { + filter := NewCacheEntryFilter("io.openshift.os.streamclass") + + assert.False(t, filter("sha256:aaa", &InspectionCacheEntry{Labels: nil})) + assert.False(t, filter("sha256:aaa", &InspectionCacheEntry{Labels: map[string]string{}})) +} diff --git a/pkg/imageutils/configmap_cache_syncer.go b/pkg/imageutils/configmap_cache_syncer.go new file mode 100644 index 0000000000..c216ca2447 --- /dev/null +++ b/pkg/imageutils/configmap_cache_syncer.go @@ -0,0 +1,197 @@ +package imageutils + +import ( + "context" + "encoding/json" + "fmt" + "time" + + annotations "github.com/openshift/api/annotations" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + coreinformersv1 "k8s.io/client-go/informers/core/v1" + clientset "k8s.io/client-go/kubernetes" + corev1listers "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/klog/v2" +) + +const ( + cacheConfigMapKey = "cache.json" + configMapMaxBytes = 1024 * 1024 // 1 MiB +) + +// ConfigMapCacheSyncer implements CacheSyncer by reading from a ConfigMap +// lister and writing via a kubeclient. An optional filter selects which entries +// to persist, and an optional transformer reduces their content before writing. +type ConfigMapCacheSyncer struct { + cmLister corev1listers.ConfigMapNamespaceLister + kubeclient clientset.Interface + namespace string + cmName string + filter CacheEntryFilter + transformer CacheEntryTransformer + hasSynced cache.InformerSynced + lastSaved string +} + +// NewConfigMapCacheSyncer creates a syncer that persists cache entries to the +// named ConfigMap. The filter, if non-nil, selects which entries to include. +// The transformer, if non-nil, is applied to each included entry before writing. +func NewConfigMapCacheSyncer( + cmInformer coreinformersv1.ConfigMapInformer, + kubeclient clientset.Interface, + namespace, cmName string, + filter CacheEntryFilter, + transformer CacheEntryTransformer, +) *ConfigMapCacheSyncer { + return &ConfigMapCacheSyncer{ + cmLister: cmInformer.Lister().ConfigMaps(namespace), + kubeclient: kubeclient, + namespace: namespace, + cmName: cmName, + filter: filter, + transformer: transformer, + hasSynced: cmInformer.Informer().HasSynced, + } +} + +// Start waits for the ConfigMap informer cache to sync, then launches +// the background sync loop. +func (s *ConfigMapCacheSyncer) Start(ctx context.Context, src SyncableCache, debounce time.Duration) { + if !cache.WaitForCacheSync(ctx.Done(), s.hasSynced) { + klog.Warning("ConfigMap informer cache sync timed out") + return + } + go s.syncLoop(ctx, src, debounce) +} + +func (s *ConfigMapCacheSyncer) syncLoop(ctx context.Context, src SyncableCache, debounce time.Duration) { + ch := src.SyncNotify() + for waitForNotify(ctx, ch) { + if !debounceDrain(ctx, ch, debounce) { + break + } + s.flush(ctx, src) + } + s.flush(context.Background(), src) +} + +func (s *ConfigMapCacheSyncer) flush(ctx context.Context, src SyncableCache) { + if err := s.save(ctx, src.Snapshot()); err != nil { + klog.Warningf("Failed to sync inspection cache to external store: %v", err) + } +} + +// waitForNotify blocks until a sync notification arrives or the context is +// cancelled. Returns true if a notification was received. +func waitForNotify(ctx context.Context, ch <-chan struct{}) bool { + select { + case <-ch: + return true + case <-ctx.Done(): + return false + } +} + +// debounceDrain drains further sync notifications until no new ones arrive +// for the given duration. Returns false if the context was cancelled. +func debounceDrain(ctx context.Context, ch <-chan struct{}, d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + for { + select { + case <-ch: + timer.Reset(d) + case <-timer.C: + return true + case <-ctx.Done(): + return false + } + } +} + +func (s *ConfigMapCacheSyncer) Load(_ context.Context) (map[string]*InspectionCacheEntry, error) { + cm, err := s.cmLister.Get(s.cmName) + if err != nil { + if apierrors.IsNotFound(err) { + return nil, nil + } + return nil, fmt.Errorf("getting inspection cache ConfigMap: %w", err) + } + + raw, ok := cm.Data[cacheConfigMapKey] + if !ok || raw == "" { + return nil, nil + } + + var file inspectionCacheFile + if err := json.Unmarshal([]byte(raw), &file); err != nil || file.Version != inspectionCacheVersion { + klog.Warningf("Ignoring inspection cache ConfigMap: invalid content") + return nil, nil + } + return file.Entries, nil +} + +func (s *ConfigMapCacheSyncer) save(ctx context.Context, entries map[string]*InspectionCacheEntry) error { + toSave := make(map[string]*InspectionCacheEntry, len(entries)) + for digest, entry := range entries { + if s.filter != nil && !s.filter(digest, entry) { + continue + } + if s.transformer != nil { + entry = s.transformer(digest, entry) + } + toSave[digest] = entry + } + + data, err := json.Marshal(&inspectionCacheFile{ + Version: inspectionCacheVersion, + Entries: toSave, + }) + if err != nil { + return fmt.Errorf("marshalling inspection cache for ConfigMap: %w", err) + } + + serialized := string(data) + if serialized == s.lastSaved { + return nil + } + + if len(data) > configMapMaxBytes { + klog.Warningf("Inspection cache too large for ConfigMap (%d bytes), skipping sync", len(data)) + return nil + } + + cmData := map[string]string{cacheConfigMapKey: string(data)} + + existing, err := s.kubeclient.CoreV1().ConfigMaps(s.namespace).Get(ctx, s.cmName, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: s.cmName, + Namespace: s.namespace, + Annotations: map[string]string{ + annotations.OpenShiftComponent: "Machine Config Operator", + }, + }, + Data: cmData, + } + if _, err := s.kubeclient.CoreV1().ConfigMaps(s.namespace).Create(ctx, cm, metav1.CreateOptions{}); err != nil { + return fmt.Errorf("creating inspection cache ConfigMap: %w", err) + } + s.lastSaved = serialized + return nil + } + if err != nil { + return fmt.Errorf("getting inspection cache ConfigMap: %w", err) + } + + existing.Data = cmData + if _, err := s.kubeclient.CoreV1().ConfigMaps(s.namespace).Update(ctx, existing, metav1.UpdateOptions{}); err != nil { + return fmt.Errorf("updating inspection cache ConfigMap: %w", err) + } + s.lastSaved = serialized + return nil +} diff --git a/pkg/imageutils/configmap_cache_syncer_test.go b/pkg/imageutils/configmap_cache_syncer_test.go new file mode 100644 index 0000000000..1ebef7a8cf --- /dev/null +++ b/pkg/imageutils/configmap_cache_syncer_test.go @@ -0,0 +1,156 @@ +package imageutils + +import ( + "context" + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes/fake" +) + +const testNamespace = "openshift-machine-config-operator" +const testCMName = "test-inspection-cache" + +func newFakeSyncer(t *testing.T, configMaps ...*corev1.ConfigMap) (*ConfigMapCacheSyncer, *fake.Clientset) { + t.Helper() + + fakeClient := fake.NewSimpleClientset() + for _, cm := range configMaps { + _, err := fakeClient.CoreV1().ConfigMaps(cm.Namespace).Create(context.Background(), cm, metav1.CreateOptions{}) + require.NoError(t, err) + } + + stopCh := make(chan struct{}) + t.Cleanup(func() { close(stopCh) }) + + factory := informers.NewSharedInformerFactory(fakeClient, 0) + syncer := NewConfigMapCacheSyncer(factory.Core().V1().ConfigMaps(), fakeClient, testNamespace, testCMName, nil, nil) + + factory.Start(stopCh) + factory.WaitForCacheSync(stopCh) + + return syncer, fakeClient +} + +func TestConfigMapCacheSyncer_LoadNotFound(t *testing.T) { + syncer, _ := newFakeSyncer(t) + entries, err := syncer.Load(context.Background()) + require.NoError(t, err) + assert.Nil(t, entries) +} + +func TestConfigMapCacheSyncer_LoadFromExisting(t *testing.T) { + cacheData := &inspectionCacheFile{ + Version: inspectionCacheVersion, + Entries: map[string]*InspectionCacheEntry{ + "sha256:aaa": {Labels: map[string]string{"k": "v"}}, + }, + } + raw, err := json.Marshal(cacheData) + require.NoError(t, err) + + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: testCMName, Namespace: testNamespace}, + Data: map[string]string{cacheConfigMapKey: string(raw)}, + } + + syncer, _ := newFakeSyncer(t, cm) + entries, err := syncer.Load(context.Background()) + require.NoError(t, err) + require.NotNil(t, entries) + assert.Equal(t, "v", entries["sha256:aaa"].Labels["k"]) +} + +func TestConfigMapCacheSyncer_SaveCreatesConfigMap(t *testing.T) { + syncer, client := newFakeSyncer(t) + + entries := map[string]*InspectionCacheEntry{ + "sha256:aaa": {Labels: map[string]string{"k": "v"}}, + } + + err := syncer.save(context.Background(), entries) + require.NoError(t, err) + + cm, err := client.CoreV1().ConfigMaps(testNamespace).Get(context.Background(), testCMName, metav1.GetOptions{}) + require.NoError(t, err) + assert.Contains(t, cm.Data[cacheConfigMapKey], "sha256:aaa") +} + +func TestConfigMapCacheSyncer_SaveUpdatesExisting(t *testing.T) { + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: testCMName, Namespace: testNamespace}, + Data: map[string]string{cacheConfigMapKey: `{"version":1,"entries":{}}`}, + } + + syncer, client := newFakeSyncer(t, cm) + + entries := map[string]*InspectionCacheEntry{ + "sha256:bbb": {Labels: map[string]string{"new": "entry"}}, + } + + err := syncer.save(context.Background(), entries) + require.NoError(t, err) + + updated, err := client.CoreV1().ConfigMaps(testNamespace).Get(context.Background(), testCMName, metav1.GetOptions{}) + require.NoError(t, err) + assert.Contains(t, updated.Data[cacheConfigMapKey], "sha256:bbb") +} + +func TestConfigMapCacheSyncer_SaveAppliesFilterAndTransformer(t *testing.T) { + fakeClient := fake.NewSimpleClientset() + + stopCh := make(chan struct{}) + t.Cleanup(func() { close(stopCh) }) + + factory := informers.NewSharedInformerFactory(fakeClient, 0) + filter := NewCacheEntryFilter("io.openshift.release") + syncer := NewConfigMapCacheSyncer(factory.Core().V1().ConfigMaps(), fakeClient, testNamespace, testCMName, filter, nil) + + factory.Start(stopCh) + factory.WaitForCacheSync(stopCh) + + entries := map[string]*InspectionCacheEntry{ + "sha256:release": {Labels: map[string]string{"io.openshift.release": "5.0.0"}}, + "sha256:random": {Labels: map[string]string{"vendor": "somebody"}}, + } + + err := syncer.save(context.Background(), entries) + require.NoError(t, err) + + cm, err := fakeClient.CoreV1().ConfigMaps(testNamespace).Get(context.Background(), testCMName, metav1.GetOptions{}) + require.NoError(t, err) + + assert.Contains(t, cm.Data[cacheConfigMapKey], "sha256:release") + assert.NotContains(t, cm.Data[cacheConfigMapKey], "sha256:random") +} + +func TestConfigMapCacheSyncer_SaveSkipsOversize(t *testing.T) { + syncer, client := newFakeSyncer(t) + + bigValue := make([]byte, configMapMaxBytes+1) + entries := map[string]*InspectionCacheEntry{ + "sha256:big": {Files: map[string][]byte{"/big": bigValue}}, + } + + err := syncer.save(context.Background(), entries) + require.NoError(t, err) + + _, err = client.CoreV1().ConfigMaps(testNamespace).Get(context.Background(), testCMName, metav1.GetOptions{}) + assert.Error(t, err, "ConfigMap should not be created when data exceeds limit") +} + +func TestConfigMapCacheSyncer_SaveSkipsDuplicate(t *testing.T) { + syncer, _ := newFakeSyncer(t) + + entries := map[string]*InspectionCacheEntry{ + "sha256:aaa": {Labels: map[string]string{"k": "v"}}, + } + + require.NoError(t, syncer.save(context.Background(), entries)) + require.NoError(t, syncer.save(context.Background(), entries)) +} diff --git a/pkg/imageutils/inspect_cache.go b/pkg/imageutils/inspect_cache.go index 0ec3eba425..5b0c27b352 100644 --- a/pkg/imageutils/inspect_cache.go +++ b/pkg/imageutils/inspect_cache.go @@ -24,7 +24,8 @@ type InspectionCacheEntry struct { CreatedAt time.Time `json:"createdAt,omitempty"` } -func (e *InspectionCacheEntry) deepCopy() *InspectionCacheEntry { +// DeepCopy returns a deep copy of the entry. +func (e *InspectionCacheEntry) DeepCopy() *InspectionCacheEntry { cp := &InspectionCacheEntry{ CreatedAt: e.CreatedAt, Labels: maps.Clone(e.Labels), @@ -46,6 +47,26 @@ type InspectionCache interface { Put(digest string, entry *InspectionCacheEntry) error } +// SyncableCache provides the read-side contract a CacheSyncer needs to +// observe changes and obtain snapshots for external persistence. +type SyncableCache interface { + // SyncNotify returns a channel that receives a value whenever the cache + // content changes and should be flushed. + SyncNotify() <-chan struct{} + // Snapshot returns a deep copy of all cache entries. + Snapshot() map[string]*InspectionCacheEntry +} + +// CacheSyncer provides external persistence for the inspection cache. +type CacheSyncer interface { + // Load retrieves persisted entries from the external store. + Load(ctx context.Context) (map[string]*InspectionCacheEntry, error) + // Start prepares the external store (e.g. waiting for informer sync), + // then launches a background goroutine that flushes cache snapshots + // whenever the cache signals a change. + Start(ctx context.Context, cache SyncableCache, debounce time.Duration) +} + // CacheEvicter determines which cached digests should be retained. // Retain receives all digests currently in the cache and returns the subset // that this evicter wants to keep. An entry is purged only if no registered @@ -61,22 +82,28 @@ type inspectionCacheFile struct { // FileInspectionCache is a file-backed InspectionCache. // It keeps an in-memory copy for fast reads and writes through to a JSON file on every Put. +// An optional CacheSyncer provides durable external persistence (e.g. a ConfigMap). type FileInspectionCache struct { - path string - mu sync.RWMutex - entries map[string]*InspectionCacheEntry - evicters []CacheEvicter - minAge time.Duration + path string + mu sync.RWMutex + entries map[string]*InspectionCacheEntry + evicters []CacheEvicter + minAge time.Duration + syncer CacheSyncer + syncNotify chan struct{} } // NewFileInspectionCache creates a cache backed by the given file path. // If the file exists it is loaded; a missing or corrupt file starts an empty cache. // minAge is the minimum time an entry must live before it can be evicted. -func NewFileInspectionCache(path string, minAge time.Duration) *FileInspectionCache { +// syncer, if non-nil, enables external persistence loaded on Start and flushed periodically. +func NewFileInspectionCache(path string, minAge time.Duration, syncer CacheSyncer) *FileInspectionCache { c := &FileInspectionCache{ - path: path, - entries: make(map[string]*InspectionCacheEntry), - minAge: minAge, + path: path, + entries: make(map[string]*InspectionCacheEntry), + minAge: minAge, + syncer: syncer, + syncNotify: make(chan struct{}, 1), } c.load() return c @@ -89,7 +116,7 @@ func (c *FileInspectionCache) Get(digest string) *InspectionCacheEntry { if e == nil { return nil } - return e.deepCopy() + return e.DeepCopy() } func (c *FileInspectionCache) Put(digest string, entry *InspectionCacheEntry) error { @@ -106,10 +133,11 @@ func (c *FileInspectionCache) Put(digest string, entry *InspectionCacheEntry) er existing.Files[k] = bytes.Clone(v) } } else { - cp := entry.deepCopy() + cp := entry.DeepCopy() cp.CreatedAt = time.Now() c.entries[digest] = cp } + c.notifySync() return c.saveLocked() } @@ -161,9 +189,65 @@ func (c *FileInspectionCache) RegisterEvicter(e CacheEvicter) { c.evicters = append(c.evicters, e) } -// StartEviction runs a background goroutine that periodically evicts entries -// older than minAge that no registered CacheEvicter wants to retain. -func (c *FileInspectionCache) StartEviction(ctx context.Context, interval, initialDelay time.Duration) { +// SyncNotify returns the channel that signals cache mutations. +func (c *FileInspectionCache) SyncNotify() <-chan struct{} { + return c.syncNotify +} + +// Snapshot returns a deep copy of all cache entries. +func (c *FileInspectionCache) Snapshot() map[string]*InspectionCacheEntry { + c.mu.RLock() + defer c.mu.RUnlock() + result := make(map[string]*InspectionCacheEntry, len(c.entries)) + for k, v := range c.entries { + result[k] = v.DeepCopy() + } + return result +} + +// Start loads persisted state from the syncer (if configured), then launches +// background goroutines for eviction and external sync. +func (c *FileInspectionCache) Start(ctx context.Context, evictionInterval, evictionDelay, syncInterval time.Duration) { + if c.syncer != nil { + c.syncer.Start(ctx, c, syncInterval) + c.loadFromSyncer(ctx) + } + c.startEviction(ctx, evictionInterval, evictionDelay) +} + +func (c *FileInspectionCache) notifySync() { + select { + case c.syncNotify <- struct{}{}: + default: + } +} + +func (c *FileInspectionCache) loadFromSyncer(ctx context.Context) { + entries, err := c.syncer.Load(ctx) + if err != nil { + klog.Warningf("Failed to load inspection cache from external store: %v", err) + return + } + if len(entries) == 0 { + return + } + + c.mu.Lock() + defer c.mu.Unlock() + for digest, entry := range entries { + existing, ok := c.entries[digest] + // Keep the local entry if it is the same age or newer. + if ok && !existing.CreatedAt.Before(entry.CreatedAt) { + continue + } + c.entries[digest] = entry.DeepCopy() + } + if err := c.saveLocked(); err != nil { + klog.Warningf("Failed to persist syncer entries to local cache file: %v", err) + } +} + +func (c *FileInspectionCache) startEviction(ctx context.Context, interval, initialDelay time.Duration) { go func() { select { case <-time.After(initialDelay): @@ -222,6 +306,7 @@ func (c *FileInspectionCache) evict() { } if evicted > 0 { + c.notifySync() klog.Infof("Inspection cache eviction: removed %d entries, %d retained", evicted, len(c.entries)) if err := c.saveLocked(); err != nil { klog.Warningf("Failed to persist inspection cache after eviction: %v", err) diff --git a/pkg/imageutils/inspect_cache_test.go b/pkg/imageutils/inspect_cache_test.go index 33443d430c..3b21039b2a 100644 --- a/pkg/imageutils/inspect_cache_test.go +++ b/pkg/imageutils/inspect_cache_test.go @@ -4,6 +4,7 @@ import ( "context" "os" "path/filepath" + "sync" "testing" "time" @@ -13,7 +14,7 @@ import ( func TestFileInspectionCache_PutAndGet(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") - cache := NewFileInspectionCache(path, 0) + cache := NewFileInspectionCache(path, 0, nil) entry := &InspectionCacheEntry{Labels: map[string]string{"k": "v"}} require.NoError(t, cache.Put("sha256:aaa", entry)) @@ -28,11 +29,11 @@ func TestFileInspectionCache_PutAndGet(t *testing.T) { func TestFileInspectionCache_PersistsAcrossInstances(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") - c1 := NewFileInspectionCache(path, 0) + c1 := NewFileInspectionCache(path, 0, nil) require.NoError(t, c1.Put("sha256:aaa", &InspectionCacheEntry{Labels: map[string]string{"a": "1"}})) require.NoError(t, c1.Put("sha256:bbb", &InspectionCacheEntry{Labels: map[string]string{"b": "2"}})) - c2 := NewFileInspectionCache(path, 0) + c2 := NewFileInspectionCache(path, 0, nil) got := c2.Get("sha256:aaa") require.NotNil(t, got) assert.Equal(t, "1", got.Labels["a"]) @@ -43,7 +44,7 @@ func TestFileInspectionCache_PersistsAcrossInstances(t *testing.T) { } func TestFileInspectionCache_MissingFile(t *testing.T) { - cache := NewFileInspectionCache(filepath.Join(t.TempDir(), "does-not-exist.json"), 0) + cache := NewFileInspectionCache(filepath.Join(t.TempDir(), "does-not-exist.json"), 0, nil) assert.Nil(t, cache.Get("sha256:any")) } @@ -51,7 +52,7 @@ func TestFileInspectionCache_CorruptFile(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") require.NoError(t, os.WriteFile(path, []byte("not json"), 0o644)) - cache := NewFileInspectionCache(path, 0) + cache := NewFileInspectionCache(path, 0, nil) assert.Nil(t, cache.Get("sha256:any")) require.NoError(t, cache.Put("sha256:aaa", &InspectionCacheEntry{Labels: map[string]string{"k": "v"}})) @@ -62,7 +63,7 @@ func TestFileInspectionCache_WrongVersion(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") require.NoError(t, os.WriteFile(path, []byte(`{"version":999,"entries":{"sha256:aaa":{"labels":{"k":"v"}}}}`), 0o644)) - cache := NewFileInspectionCache(path, 0) + cache := NewFileInspectionCache(path, 0, nil) assert.Nil(t, cache.Get("sha256:aaa")) } @@ -72,25 +73,25 @@ func (f retainFunc) Retain(digests []string) []string { return f(digests) } func TestFileInspectionCache_EvictNoEvicters(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") - cache := NewFileInspectionCache(path, 0) + cache := NewFileInspectionCache(path, 0, nil) require.NoError(t, cache.Put("sha256:aaa", &InspectionCacheEntry{Labels: map[string]string{"a": "1"}})) require.NoError(t, cache.Put("sha256:bbb", &InspectionCacheEntry{Labels: map[string]string{"b": "2"}})) ctx, cancel := context.WithCancel(context.Background()) defer cancel() - cache.StartEviction(ctx, 50*time.Millisecond, 0) + cache.Start(ctx, 50*time.Millisecond, 0, time.Hour) require.Eventually(t, func() bool { return cache.Get("sha256:aaa") == nil && cache.Get("sha256:bbb") == nil }, 5*time.Second, 50*time.Millisecond) - reloaded := NewFileInspectionCache(path, 0) + reloaded := NewFileInspectionCache(path, 0, nil) assert.Nil(t, reloaded.Get("sha256:aaa")) } func TestFileInspectionCache_EvictRetainsUnion(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") - cache := NewFileInspectionCache(path, 0) + cache := NewFileInspectionCache(path, 0, nil) require.NoError(t, cache.Put("sha256:aaa", &InspectionCacheEntry{Labels: map[string]string{"a": "1"}})) require.NoError(t, cache.Put("sha256:bbb", &InspectionCacheEntry{Labels: map[string]string{"b": "2"}})) require.NoError(t, cache.Put("sha256:ccc", &InspectionCacheEntry{Labels: map[string]string{"c": "3"}})) @@ -104,7 +105,7 @@ func TestFileInspectionCache_EvictRetainsUnion(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - cache.StartEviction(ctx, 50*time.Millisecond, 0) + cache.Start(ctx, 50*time.Millisecond, 0, time.Hour) require.Eventually(t, func() bool { return cache.Get("sha256:ccc") == nil @@ -113,7 +114,7 @@ func TestFileInspectionCache_EvictRetainsUnion(t *testing.T) { assert.NotNil(t, cache.Get("sha256:aaa")) assert.NotNil(t, cache.Get("sha256:bbb")) - reloaded := NewFileInspectionCache(path, 0) + reloaded := NewFileInspectionCache(path, 0, nil) assert.NotNil(t, reloaded.Get("sha256:aaa")) assert.NotNil(t, reloaded.Get("sha256:bbb")) assert.Nil(t, reloaded.Get("sha256:ccc")) @@ -121,12 +122,12 @@ func TestFileInspectionCache_EvictRetainsUnion(t *testing.T) { func TestFileInspectionCache_EvictRespectsMinAge(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") - cache := NewFileInspectionCache(path, 500*time.Millisecond) + cache := NewFileInspectionCache(path, 500*time.Millisecond, nil) require.NoError(t, cache.Put("sha256:aaa", &InspectionCacheEntry{Labels: map[string]string{"a": "1"}})) ctx, cancel := context.WithCancel(context.Background()) defer cancel() - cache.StartEviction(ctx, 50*time.Millisecond, 0) + cache.Start(ctx, 50*time.Millisecond, 0, time.Hour) require.Never(t, func() bool { return cache.Get("sha256:aaa") == nil @@ -139,7 +140,7 @@ func TestFileInspectionCache_EvictRespectsMinAge(t *testing.T) { func TestFileInspectionCache_PutMergesFiles(t *testing.T) { path := filepath.Join(t.TempDir(), "cache.json") - cache := NewFileInspectionCache(path, 0) + cache := NewFileInspectionCache(path, 0, nil) require.NoError(t, cache.Put("sha256:aaa", &InspectionCacheEntry{Labels: map[string]string{"k": "v"}})) require.NoError(t, cache.Put("sha256:aaa", &InspectionCacheEntry{Files: map[string][]byte{"/etc/config": []byte("data")}})) @@ -149,13 +150,167 @@ func TestFileInspectionCache_PutMergesFiles(t *testing.T) { assert.Equal(t, "v", got.Labels["k"]) assert.Equal(t, []byte("data"), got.Files["/etc/config"]) - reloaded := NewFileInspectionCache(path, 0) + reloaded := NewFileInspectionCache(path, 0, nil) got = reloaded.Get("sha256:aaa") require.NotNil(t, got) assert.Equal(t, "v", got.Labels["k"]) assert.Equal(t, []byte("data"), got.Files["/etc/config"]) } +type mockSyncer struct { + loadEntries map[string]*InspectionCacheEntry + loadErr error + + mu sync.Mutex + saved map[string]*InspectionCacheEntry + saveCount int +} + +func (m *mockSyncer) Load(_ context.Context) (map[string]*InspectionCacheEntry, error) { + return m.loadEntries, m.loadErr +} + +func (m *mockSyncer) record(snapshot map[string]*InspectionCacheEntry) { + m.mu.Lock() + defer m.mu.Unlock() + m.saved = snapshot + m.saveCount++ +} + +func (m *mockSyncer) state() (map[string]*InspectionCacheEntry, int) { + m.mu.Lock() + defer m.mu.Unlock() + return m.saved, m.saveCount +} + +func (m *mockSyncer) Start(ctx context.Context, src SyncableCache, debounce time.Duration) { + ch := src.SyncNotify() + go func() { + for waitForNotify(ctx, ch) { + if !debounceDrain(ctx, ch, debounce) { + break + } + m.record(src.Snapshot()) + } + m.record(src.Snapshot()) + }() +} + +// Verifies that entries from the syncer are loaded into memory on Start +// and persisted to the local cache file for subsequent instances. +func TestFileInspectionCache_StartLoadsFromSyncer(t *testing.T) { + syncer := &mockSyncer{ + loadEntries: map[string]*InspectionCacheEntry{ + "sha256:persisted": {Labels: map[string]string{"from": "configmap"}}, + }, + } + + path := filepath.Join(t.TempDir(), "cache.json") + cache := NewFileInspectionCache(path, 0, syncer) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cache.Start(ctx, time.Hour, time.Hour, time.Hour) + + got := cache.Get("sha256:persisted") + require.NotNil(t, got) + assert.Equal(t, "configmap", got.Labels["from"]) + + reloaded := NewFileInspectionCache(path, 0, nil) + got = reloaded.Get("sha256:persisted") + require.NotNil(t, got, "syncer entries must be persisted to the local cache file") + assert.Equal(t, "configmap", got.Labels["from"]) +} + +// Verifies that a Put triggers a sync flush to the external store. +func TestFileInspectionCache_StartSyncFlushesAfterPut(t *testing.T) { + syncer := &mockSyncer{} + + path := filepath.Join(t.TempDir(), "cache.json") + cache := NewFileInspectionCache(path, 0, syncer) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cache.Start(ctx, time.Hour, time.Hour, 50*time.Millisecond) + + require.NoError(t, cache.Put("sha256:new", &InspectionCacheEntry{Labels: map[string]string{"k": "v"}})) + + require.Eventually(t, func() bool { + _, count := syncer.state() + return count > 0 + }, 5*time.Second, 50*time.Millisecond) + + saved, _ := syncer.state() + assert.Contains(t, saved, "sha256:new") +} + +// Verifies that the syncer is not called when no mutations occur. +func TestFileInspectionCache_StartSyncNoFlushWithoutChanges(t *testing.T) { + syncer := &mockSyncer{} + + path := filepath.Join(t.TempDir(), "cache.json") + cache := NewFileInspectionCache(path, 0, syncer) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cache.Start(ctx, time.Hour, time.Hour, 50*time.Millisecond) + + require.Never(t, func() bool { + _, count := syncer.state() + return count > 0 + }, time.Second, 50*time.Millisecond) +} + +// Verifies that pending entries are flushed to the syncer on shutdown, +// even when the debounce interval has not elapsed. +func TestFileInspectionCache_StartSyncFlushOnShutdown(t *testing.T) { + syncer := &mockSyncer{} + + path := filepath.Join(t.TempDir(), "cache.json") + cache := NewFileInspectionCache(path, 0, syncer) + + ctx, cancel := context.WithCancel(context.Background()) + // Debounce set to time.Hour so the only trigger is the context cancellation. + cache.Start(ctx, time.Hour, time.Hour, time.Hour) + + require.NoError(t, cache.Put("sha256:flushed", &InspectionCacheEntry{Labels: map[string]string{"k": "v"}})) + cancel() + + require.Eventually(t, func() bool { + _, count := syncer.state() + return count > 0 + }, 5*time.Second, 50*time.Millisecond) + + saved, _ := syncer.state() + assert.Contains(t, saved, "sha256:flushed") +} + +// Verifies that eviction triggers a sync notification so the syncer +// flushes the updated state to external storage. +func TestFileInspectionCache_StartSyncEvictionNotifies(t *testing.T) { + syncer := &mockSyncer{} + + path := filepath.Join(t.TempDir(), "cache.json") + cache := NewFileInspectionCache(path, 0, syncer) + + require.NoError(t, cache.Put("sha256:evictme", &InspectionCacheEntry{Labels: map[string]string{"k": "v"}})) + + // Drain the notification from the Put above. + select { + case <-cache.syncNotify: + default: + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cache.Start(ctx, 50*time.Millisecond, 0, 50*time.Millisecond) + + require.Eventually(t, func() bool { + _, count := syncer.state() + return count > 0 + }, 5*time.Second, 50*time.Millisecond) +} + func TestDigestFromPullspec(t *testing.T) { tests := []struct { input string diff --git a/pkg/osimagestream/cache_entry_filter.go b/pkg/osimagestream/cache_entry_filter.go new file mode 100644 index 0000000000..2b7559984a --- /dev/null +++ b/pkg/osimagestream/cache_entry_filter.go @@ -0,0 +1,18 @@ +package osimagestream + +import ( + "github.com/openshift/machine-config-operator/pkg/imageutils" +) + +// NewCacheEntryFilter returns a filter that selects cache entries relevant +// for OS stream discovery: those carrying at least one OS, extensions, bootc, +// or release label. +func NewCacheEntryFilter() imageutils.CacheEntryFilter { + return imageutils.NewCacheEntryFilter( + coreOSLabelStreamClass, + coreOSLabelBootc, + coreOSLabelExtension, + releasePayloadLabel, + ) +} + diff --git a/pkg/osimagestream/cache_entry_filter_test.go b/pkg/osimagestream/cache_entry_filter_test.go new file mode 100644 index 0000000000..56b565dbfd --- /dev/null +++ b/pkg/osimagestream/cache_entry_filter_test.go @@ -0,0 +1,32 @@ +package osimagestream + +import ( + "testing" + + "github.com/openshift/machine-config-operator/pkg/imageutils" + "github.com/stretchr/testify/assert" +) + +func TestCacheEntryFilter(t *testing.T) { + filter := NewCacheEntryFilter() + + tests := []struct { + name string + labels map[string]string + want bool + }{ + {name: "accepts streamclass label", labels: map[string]string{"io.openshift.os.streamclass": "coreos"}, want: true}, + {name: "accepts bootc label", labels: map[string]string{"containers.bootc": "1"}, want: true}, + {name: "accepts extensions label", labels: map[string]string{"io.openshift.os.extensions": "true"}, want: true}, + {name: "accepts release label", labels: map[string]string{"io.openshift.release": "5.0.0"}, want: true}, + {name: "rejects unrelated labels", labels: map[string]string{"io.openshift.build.commit.id": "abc123"}, want: false}, + {name: "rejects empty labels", labels: map[string]string{}, want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + entry := &imageutils.InspectionCacheEntry{Labels: tt.labels} + assert.Equal(t, tt.want, filter("sha256:aaa", entry)) + }) + } +} diff --git a/pkg/osimagestream/imagestream.go b/pkg/osimagestream/imagestream.go new file mode 100644 index 0000000000..52bc0a50a3 --- /dev/null +++ b/pkg/osimagestream/imagestream.go @@ -0,0 +1,23 @@ +package osimagestream + +import "regexp" + +const ( + // osSourceAnnotation is the ImageStream tag annotation key that identifies + // tags built from the openshift/os repository. + osSourceAnnotation = "io.openshift.build.source-location" + + // osSourceRepo is the substring matched against the osSourceAnnotation value + // to identify OS-related ImageStream tags. + osSourceRepo = "github.com/openshift/os" + + // releaseImageStreamPath is the path inside a release payload image + // where the image-references ImageStream manifest is stored. + releaseImageStreamPath = "/release-manifests/image-references" +) + +var ( + // imageTagRegxpr matches ImageStream tag names that are considered OS or extensions images. + // Matches patterns like "rhel-coreos", "stream-coreos", "rhel-coreos-extensions", etc. + imageTagRegxpr = regexp.MustCompile(`^(rhel|stream)[a-zA-Z0-9.-]*-coreos[a-zA-Z0-9.-]*(-extensions[a-zA-Z0-9.-]*)?$`) +) diff --git a/pkg/osimagestream/imagestream_entry_transformer.go b/pkg/osimagestream/imagestream_entry_transformer.go new file mode 100644 index 0000000000..b307965e55 --- /dev/null +++ b/pkg/osimagestream/imagestream_entry_transformer.go @@ -0,0 +1,61 @@ +package osimagestream + +import ( + "bytes" + "fmt" + "strings" + + imagev1 "github.com/openshift/api/image/v1" + "github.com/openshift/client-go/image/clientset/versioned/scheme" + "github.com/openshift/machine-config-operator/pkg/imageutils" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/serializer/json" +) + +// NewImageStreamFileTransformer returns an EntryTransformer that strips +// unneeded tags from cached image-references files. Only tags that match +// the OS/extensions pattern or carry the OpenShift OS build annotation are +// kept, significantly reducing the manifest size. +func NewImageStreamFileTransformer() imageutils.CacheEntryTransformer { + return imageutils.NewCacheFileTransformer(releaseImageStreamPath, filterImageStreamTags) +} + +func filterImageStreamTags(data []byte) ([]byte, error) { + obj, err := runtime.Decode(scheme.Codecs.UniversalDecoder(imagev1.SchemeGroupVersion), data) + if err != nil { + return data, nil + } + + is, ok := obj.(*imagev1.ImageStream) + if !ok { + return data, nil + } + + filtered := make([]imagev1.TagReference, 0, len(is.Spec.Tags)) + for _, tag := range is.Spec.Tags { + if shouldKeepTag(tag) { + filtered = append(filtered, tag) + } + } + is.Spec.Tags = filtered + + serializer := json.NewSerializerWithOptions(json.DefaultMetaFactory, scheme.Scheme, scheme.Scheme, + json.SerializerOptions{Yaml: false, Pretty: false}) + var buf bytes.Buffer + if err := serializer.Encode(is, &buf); err != nil { + return nil, fmt.Errorf("encoding filtered ImageStream: %w", err) + } + return buf.Bytes(), nil +} + +func shouldKeepTag(tag imagev1.TagReference) bool { + if tag.From == nil || tag.From.Kind != "DockerImage" { + return false + } + if tag.Annotations != nil { + if source, ok := tag.Annotations[osSourceAnnotation]; ok && strings.Contains(source, osSourceRepo) { + return true + } + } + return imageTagRegxpr.MatchString(tag.Name) +} diff --git a/pkg/osimagestream/imagestream_entry_transformer_test.go b/pkg/osimagestream/imagestream_entry_transformer_test.go new file mode 100644 index 0000000000..451c7df4f5 --- /dev/null +++ b/pkg/osimagestream/imagestream_entry_transformer_test.go @@ -0,0 +1,150 @@ +package osimagestream + +import ( + "bytes" + "testing" + + imagev1 "github.com/openshift/api/image/v1" + "github.com/openshift/client-go/image/clientset/versioned/scheme" + "github.com/openshift/machine-config-operator/pkg/imageutils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime/serializer/json" +) + +func encodeImageStream(t *testing.T, is *imagev1.ImageStream) []byte { + t.Helper() + encoder := scheme.Codecs.EncoderForVersion( + json.NewSerializerWithOptions(json.DefaultMetaFactory, scheme.Scheme, scheme.Scheme, + json.SerializerOptions{Yaml: false}), + imagev1.SchemeGroupVersion, + ) + var buf bytes.Buffer + require.NoError(t, encoder.Encode(is, &buf)) + return buf.Bytes() +} + +func TestImageStreamFileTransformer_FiltersUnneededTags(t *testing.T) { + is := &imagev1.ImageStream{ + Spec: imagev1.ImageStreamSpec{ + Tags: []imagev1.TagReference{ + { + Name: "rhel-coreos", + From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/os@sha256:aaa"}, + }, + { + Name: "rhel-coreos-extensions", + From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/ext@sha256:bbb"}, + }, + { + Name: "machine-config-operator", + From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/mco@sha256:ccc"}, + }, + { + Name: "etcd", + From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/etcd@sha256:ddd"}, + }, + { + Name: "kube-apiserver", + From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/kas@sha256:eee"}, + }, + }, + }, + } + + entry := &imageutils.InspectionCacheEntry{ + Labels: map[string]string{"io.openshift.release": "5.0.0"}, + Files: map[string][]byte{releaseImageStreamPath: encodeImageStream(t, is)}, + } + + transformer := NewImageStreamFileTransformer() + result := transformer("sha256:release", entry) + require.NotNil(t, result) + + resultIS := decodeImageStream(t, result.Files[releaseImageStreamPath]) + tagNames := make([]string, 0, len(resultIS.Spec.Tags)) + for _, tag := range resultIS.Spec.Tags { + tagNames = append(tagNames, tag.Name) + } + + assert.Contains(t, tagNames, "rhel-coreos") + assert.Contains(t, tagNames, "rhel-coreos-extensions") + assert.NotContains(t, tagNames, "etcd") + assert.NotContains(t, tagNames, "kube-apiserver") + assert.NotContains(t, tagNames, "machine-config-operator") +} + +func TestImageStreamFileTransformer_AnnotationMatch(t *testing.T) { + is := &imagev1.ImageStream{ + Spec: imagev1.ImageStreamSpec{ + Tags: []imagev1.TagReference{ + { + Name: "custom-os", + From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/custom@sha256:aaa"}, + Annotations: map[string]string{"io.openshift.build.source-location": "https://github.com/openshift/os"}, + }, + { + Name: "unrelated", + From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/other@sha256:bbb"}, + }, + }, + }, + } + + entry := &imageutils.InspectionCacheEntry{ + Files: map[string][]byte{releaseImageStreamPath: encodeImageStream(t, is)}, + } + + transformer := NewImageStreamFileTransformer() + result := transformer("sha256:x", entry) + require.NotNil(t, result) + + resultIS := decodeImageStream(t, result.Files[releaseImageStreamPath]) + require.Len(t, resultIS.Spec.Tags, 1) + assert.Equal(t, "custom-os", resultIS.Spec.Tags[0].Name) +} + +func TestImageStreamFileTransformer_NonImageStreamFilePassesThrough(t *testing.T) { + entry := &imageutils.InspectionCacheEntry{ + Labels: map[string]string{"k": "v"}, + Files: map[string][]byte{"/other/path": []byte("untouched")}, + } + + transformer := NewImageStreamFileTransformer() + result := transformer("sha256:x", entry) + require.NotNil(t, result) + assert.Equal(t, []byte("untouched"), result.Files["/other/path"]) +} + +func TestImageStreamFileTransformer_DoesNotMutateInput(t *testing.T) { + is := &imagev1.ImageStream{ + Spec: imagev1.ImageStreamSpec{ + Tags: []imagev1.TagReference{ + {Name: "rhel-coreos", From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/os@sha256:aaa"}}, + {Name: "etcd", From: &corev1.ObjectReference{Kind: "DockerImage", Name: "quay.io/etcd@sha256:bbb"}}, + }, + }, + } + originalData := encodeImageStream(t, is) + + entry := &imageutils.InspectionCacheEntry{ + Files: map[string][]byte{releaseImageStreamPath: originalData}, + } + + transformer := NewImageStreamFileTransformer() + _ = transformer("sha256:x", entry) + + assert.Equal(t, originalData, entry.Files[releaseImageStreamPath]) +} + +func decodeImageStream(t *testing.T, data []byte) *imagev1.ImageStream { + t.Helper() + serializer := json.NewSerializerWithOptions(json.DefaultMetaFactory, scheme.Scheme, scheme.Scheme, + json.SerializerOptions{Yaml: false}) + obj, _, err := serializer.Decode(data, nil, nil) + require.NoError(t, err) + is, ok := obj.(*imagev1.ImageStream) + require.True(t, ok) + return is +} diff --git a/pkg/osimagestream/imagestream_provider.go b/pkg/osimagestream/imagestream_provider.go index 08b620899b..dd6e34f8e6 100644 --- a/pkg/osimagestream/imagestream_provider.go +++ b/pkg/osimagestream/imagestream_provider.go @@ -10,8 +10,6 @@ import ( "k8s.io/apimachinery/pkg/runtime" ) -const releaseImageStreamLocation = "/release-manifests/image-references" - // ImageStreamProvider provides access to an ImageStream resource. type ImageStreamProvider interface { ReadImageStream(ctx context.Context) (*imagev1.ImageStream, error) @@ -65,7 +63,7 @@ func (i *ImageStreamProviderNetwork) ReadImageStream(ctx context.Context) (*imag } func (i *ImageStreamProviderNetwork) fetchImageStream(ctx context.Context) (*imagev1.ImageStream, error) { - imageStreamBytes, err := i.imagesInspector.FetchImageFile(ctx, i.imageName, releaseImageStreamLocation) + imageStreamBytes, err := i.imagesInspector.FetchImageFile(ctx, i.imageName, releaseImageStreamPath) if err != nil { return nil, err } diff --git a/pkg/osimagestream/imagestream_source.go b/pkg/osimagestream/imagestream_source.go index f775722cba..cba0bb9bae 100644 --- a/pkg/osimagestream/imagestream_source.go +++ b/pkg/osimagestream/imagestream_source.go @@ -2,19 +2,12 @@ package osimagestream import ( "context" - "regexp" "strings" imagev1 "github.com/openshift/api/image/v1" mcfgv1 "github.com/openshift/api/machineconfiguration/v1" ) -var ( - // imageTagRegxpr matches ImageStream tag names that are considered OS or extensions images. - // Matches patterns like "rhel-coreos", "stream-coreos", "rhel-coreos-extensions", etc. - imageTagRegxpr = regexp.MustCompile(`^(rhel|stream)[a-zA-Z0-9.-]*-coreos[a-zA-Z0-9.-]*(-extensions[a-zA-Z0-9.-]*)?$`) -) - // ImageStreamStreamSource fetches OS image stream metadata from an OpenShift ImageStream resource. type ImageStreamStreamSource struct { discoverer *StreamDiscoverer @@ -46,7 +39,7 @@ func (r *ImageStreamStreamSource) filterImageTag(imageStream *imagev1.ImageStrea continue } if tag.Annotations != nil { - if source, ok := tag.Annotations["io.openshift.build.source-location"]; ok && strings.Contains(source, "github.com/openshift/os") { + if source, ok := tag.Annotations[osSourceAnnotation]; ok && strings.Contains(source, osSourceRepo) { imagesToParse = append(imagesToParse, tag.From.Name) continue } diff --git a/test/e2e-2of2/osimagestream_test.go b/test/e2e-2of2/osimagestream_test.go index b7d4665c2b..dbf361b264 100644 --- a/test/e2e-2of2/osimagestream_test.go +++ b/test/e2e-2of2/osimagestream_test.go @@ -117,7 +117,7 @@ func TestCachedInspectorFactory(t *testing.T) { require.NotEmpty(t, digest, "CVO release image must be digested") cachePath := filepath.Join(t.TempDir(), "test-cache.json") - cache := imageutils.NewFileInspectionCache(cachePath, 48*time.Hour) + cache := imageutils.NewFileInspectionCache(cachePath, 48*time.Hour, nil) factory := osimagestream.NewCachedImagesInspectorFactory( &osimagestream.DefaultImagesInspectorFactory{}, cache,