Merge pull request #482 from grokspawn/cherry-pick-opm-caching

OCPBUGS-12263: cherry-pick pull request refactor FBC caching (#1051) f…

staging/operator-registry/cmd/opm/serve/serve.go

@@ -3,6 +3,7 @@ package serve
 import (
 	"context"
 	"fmt"
+	"io"
 	"net"
 	"os"
 
@@ -13,17 +14,18 @@ import (
 
 	"github.com/operator-framework/operator-registry/pkg/api"
 	health "github.com/operator-framework/operator-registry/pkg/api/grpc_health_v1"
+	"github.com/operator-framework/operator-registry/pkg/cache"
 	"github.com/operator-framework/operator-registry/pkg/lib/dns"
 	"github.com/operator-framework/operator-registry/pkg/lib/graceful"
 	"github.com/operator-framework/operator-registry/pkg/lib/log"
-	"github.com/operator-framework/operator-registry/pkg/registry"
 	"github.com/operator-framework/operator-registry/pkg/server"
 )
 
 type serve struct {
-	configDir string
-	cacheDir  string
-	cacheOnly bool
+	configDir             string
+	cacheDir              string
+	cacheOnly             bool
+	cacheEnforceIntegrity bool
 
 	port           string
 	terminationLog string
@@ -47,15 +49,19 @@ startup. Changes made to the declarative config after the this command starts
 will not be reflected in the served content.
 `,
 		Args: cobra.ExactArgs(1),
-		PreRunE: func(_ *cobra.Command, args []string) error {
+		PreRun: func(_ *cobra.Command, args []string) {
 			s.configDir = args[0]
 			if s.debug {
 				logger.SetLevel(logrus.DebugLevel)
 			}
-			return nil
 		},
-		RunE: func(cmd *cobra.Command, _ []string) error {
-			return s.run(cmd.Context())
+		Run: func(cmd *cobra.Command, _ []string) {
+			if !cmd.Flags().Changed("cache-enforce-integrity") {
+				s.cacheEnforceIntegrity = s.cacheDir != "" && !s.cacheOnly
+			}
+			if err := s.run(cmd.Context()); err != nil {
+				logger.Fatal(err)
+			}
 		},
 	}
 
@@ -64,6 +70,7 @@ will not be reflected in the served content.
 	cmd.Flags().StringVarP(&s.terminationLog, "termination-log", "t", "/dev/termination-log", "path to a container termination log file")
 	cmd.Flags().StringVar(&s.cacheDir, "cache-dir", "", "if set, sync and persist server cache directory")
 	cmd.Flags().BoolVar(&s.cacheOnly, "cache-only", false, "sync the serve cache and exit without serving")
+	cmd.Flags().BoolVar(&s.cacheEnforceIntegrity, "cache-enforce-integrity", false, "exit with error if cache is not present or has been invalidated. (default: true when --cache-dir is set and --cache-only is false, false otherwise), ")
 	return cmd
 }
 
@@ -81,11 +88,38 @@ func (s *serve) run(ctx context.Context) error {
 
 	s.logger = s.logger.WithFields(logrus.Fields{"configs": s.configDir, "port": s.port})
 
-	store, err := registry.NewQuerierFromFS(os.DirFS(s.configDir), s.cacheDir)
-	defer store.Close()
+	if s.cacheDir == "" && s.cacheEnforceIntegrity {
+		return fmt.Errorf("--cache-dir must be specified with --cache-enforce-integrity")
+	}
+
+	if s.cacheDir == "" {
+		s.cacheDir, err = os.MkdirTemp("", "opm-serve-cache-")
+		if err != nil {
+			return err
+		}
+		defer os.RemoveAll(s.cacheDir)
+	}
+
+	store, err := cache.New(s.cacheDir)
 	if err != nil {
 		return err
 	}
+	if storeCloser, ok := store.(io.Closer); ok {
+		defer storeCloser.Close()
+	}
+	if s.cacheEnforceIntegrity {
+		if err := store.CheckIntegrity(os.DirFS(s.configDir)); err != nil {
+			return err
+		}
+		if err := store.Load(); err != nil {
+			return err
+		}
+	} else {
+		if err := cache.LoadOrRebuild(store, os.DirFS(s.configDir)); err != nil {
+			return err
+		}
+	}
+
 	if s.cacheOnly {
 		return nil
 	}

staging/operator-registry/pkg/cache/cache.go

@@ -0,0 +1,104 @@
+package cache
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	"github.com/operator-framework/operator-registry/pkg/api"
+	"github.com/operator-framework/operator-registry/pkg/registry"
+)
+
+type Cache interface {
+	registry.GRPCQuery
+
+	CheckIntegrity(fbc fs.FS) error
+	Build(fbc fs.FS) error
+	Load() error
+}
+
+func LoadOrRebuild(c Cache, fbc fs.FS) error {
+	if err := c.CheckIntegrity(fbc); err != nil {
+		if err := c.Build(fbc); err != nil {
+			return err
+		}
+	}
+	return c.Load()
+}
+
+// New creates a new Cache. It chooses a cache implementation based
+// on the files it finds in the cache directory, with a preference for the
+// latest iteration of the cache implementation. It returns an error if
+// cacheDir exists and contains unexpected files.
+func New(cacheDir string) (Cache, error) {
+	entries, err := os.ReadDir(cacheDir)
+	if err != nil && !errors.Is(err, os.ErrNotExist) {
+		return nil, fmt.Errorf("detect cache format: read cache directory: %v", err)
+	}
+	jsonCache := sets.NewString(jsonDir, jsonDigestFile)
+
+	found := sets.NewString()
+	for _, e := range entries {
+		found.Insert(e.Name())
+	}
+
+	// Preference (and currently only supported) is the JSON-based cache implementation.
+	if found.IsSuperset(jsonCache) || len(entries) == 0 {
+		return NewJSON(cacheDir), nil
+	}
+
+	// Anything else is unexpected.
+	return nil, fmt.Errorf("cache directory has unexpected contents")
+}
+
+func ensureEmptyDir(dir string, mode os.FileMode) error {
+	if err := os.MkdirAll(dir, mode); err != nil {
+		return err
+	}
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		return err
+	}
+	for _, entry := range entries {
+		if err := os.RemoveAll(filepath.Join(dir, entry.Name())); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func doesBundleProvide(ctx context.Context, c Cache, pkgName, chName, bundleName, group, version, kind string) (bool, error) {
+	apiBundle, err := c.GetBundle(ctx, pkgName, chName, bundleName)
+	if err != nil {
+		return false, fmt.Errorf("get bundle %q: %v", bundleName, err)
+	}
+	for _, gvk := range apiBundle.ProvidedApis {
+		if gvk.Group == group && gvk.Version == version && gvk.Kind == kind {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+type sliceBundleSender []*api.Bundle
+
+func (s *sliceBundleSender) Send(b *api.Bundle) error {
+	*s = append(*s, b)
+	return nil
+}
+
+func listBundles(ctx context.Context, c Cache) ([]*api.Bundle, error) {
+	var bundleSender sliceBundleSender
+
+	err := c.SendBundles(ctx, &bundleSender)
+	if err != nil {
+		return nil, err
+	}
+
+	return bundleSender, nil
+}