refactor FBC caching (#1051)

This commit improves maintainability and extensibility of the FBC
caching used to serve the GRPC API.

It:
- introduces a new cache package and interface, which defines
  primitives for checking cache integrity, building the cache, and
  loading the cache.
- moves the existing registry.Querier to an implementation of the cache
  interface, called cache.JSON

This refactor also resolves two outstanding issues:
1. The current code contains a bug that causes panics when certain
   failures occur loading the cache.
2. The current code is hardcoded to always rebuild the cache if it is
   unreadable or its digest doesn't contain the expected value.

This commit:
1. Refactors and removes the cases the led to the panic situation
2. Adds a new flag, --cache-enforce-integrity, that causes `opm serve`
   to exit with a failure if the cache is unreadable or its digest
   doesn't contain the expected value. This is useful in production
   contexts when one always expects the cache to be pre-populated
   correctly.

Signed-off-by: Joe Lanford <[email protected]>

Signed-off-by: Joe Lanford <[email protected]>

Author: joelanford

cmd/opm/serve/serve.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	endpoint "net/http/pprof"
@@ -19,17 +20,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
@@ -59,15 +61,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)
+			}
 		},
 	}
 
@@ -77,6 +83,7 @@ will not be reflected in the served content.
 	cmd.Flags().StringVar(&s.pprofAddr, "pprof-addr", "", "address of startup profiling endpoint (addr:port format)")
 	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
 }
 
@@ -102,11 +109,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
 	}

go.mod

@@ -34,7 +34,7 @@ require (
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4
 	golang.org/x/net v0.0.0-20220722155237-a158d28d115b
 	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
-	golang.org/x/sys v0.0.0-20220907062415-87db552b00fd
+	golang.org/x/sys v0.0.0-20221010170243-090e33056c14
 	google.golang.org/grpc v1.47.0
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v0.0.0-20200709232328-d8193ee9cc3e
 	google.golang.org/protobuf v1.28.0
@@ -94,6 +94,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/snappy v0.0.3 // indirect
 	github.com/google/cel-go v0.12.4 // indirect
 	github.com/google/gnostic v0.5.7-v3refs // indirect
 	github.com/google/gofuzz v1.1.0 // indirect
@@ -108,7 +109,7 @@ require (
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect
-	github.com/klauspost/compress v1.11.13 // indirect
+	github.com/klauspost/compress v1.12.3 // indirect
 	github.com/mailru/easyjson v0.7.6 // indirect
 	github.com/mattn/go-isatty v0.0.12 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect

go.sum

@@ -338,6 +338,8 @@ github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu
 github.com/golang/snappy v0.0.0-20170215233205-553a64147049/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.3 h1:fHPg5GQYlCeLIPB9BZqMVR5nR9A+IM5zcgeTdjMYmLA=
+github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/cel-go v0.12.4 h1:YINKfuHZ8n72tPOqSPZBwGiDpew2CJS48mdM5W8LZQU=
@@ -468,8 +470,8 @@ github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvW
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.11.13 h1:eSvu8Tmq6j2psUJqJrLcWH6K3w5Dwc+qipbaA6eVEN4=
-github.com/klauspost/compress v1.11.13/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
+github.com/klauspost/compress v1.12.3 h1:G5AfA94pHPysR56qqrkO2pxEexdDzrpFJ6yt/VqWxVU=
+github.com/klauspost/compress v1.12.3/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -942,8 +944,8 @@ golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220907062415-87db552b00fd h1:AZeIEzg+8RCELJYq8w+ODLVxFgLMMigSwO/ffKPEd9U=
-golang.org/x/sys v0.0.0-20220907062415-87db552b00fd/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20221010170243-090e33056c14 h1:k5II8e6QD8mITdi+okbbmR/cIyEbeXLBhy5Ha4nevyc=
+golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 h1:JGgROgKl9N8DuW20oFS5gxc+lE67/N3FcwmBPMe7ArY=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=

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
+}