opm serve: use pre-existing cache, if set and up-to-date (openshift#1005)

* opm serve: use pre-existing cache, if set and up-to-date

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

* refactor to leave NewQuerier function untouched

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

Signed-off-by: Joe Lanford <[email protected]>
Upstream-repository: operator-registry
Upstream-commit: 494b68e62a814a891821aeb2bd28f33abc1624ff

Author: joelanford

pkg/manifests/csv.yaml

@@ -5,7 +5,7 @@ metadata:
   name: packageserver
   namespace: openshift-operator-lifecycle-manager
   labels:
-    olm.version: 0.0.0-f8dd7c4e6b8762e41dd59f9e727c52acd9cffe59
+    olm.version: 0.0.0-7829ec69880557119c98ba10ad52cb9264e4f278
     olm.clusteroperator.name: operator-lifecycle-manager-packageserver
   annotations:
     include.release.openshift.io/self-managed-high-availability: "true"
@@ -159,7 +159,7 @@ spec:
                                 - packageserver
                         topologyKey: "kubernetes.io/hostname"
   maturity: alpha
-  version: 0.0.0-f8dd7c4e6b8762e41dd59f9e727c52acd9cffe59
+  version: 0.0.0-7829ec69880557119c98ba10ad52cb9264e4f278
   apiservicedefinitions:
     owned:
       - group: packages.operators.coreos.com

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

@@ -5,7 +5,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"net"
 	"net/http"
 	endpoint "net/http/pprof"
@@ -20,18 +19,17 @@ 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
-	cacheEnforceIntegrity bool
+	configDir string
+	cacheDir  string
+	cacheOnly bool
 
 	port           string
 	terminationLog string
@@ -61,19 +59,15 @@ startup. Changes made to the declarative config after the this command starts
 will not be reflected in the served content.
 `,
 		Args: cobra.ExactArgs(1),
-		PreRun: func(_ *cobra.Command, args []string) {
+		PreRunE: func(_ *cobra.Command, args []string) error {
 			s.configDir = args[0]
 			if s.debug {
 				logger.SetLevel(logrus.DebugLevel)
 			}
+			return nil
 		},
-		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)
-			}
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			return s.run(cmd.Context())
 		},
 	}
 
@@ -83,15 +77,12 @@ 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
 }
 
 func (s *serve) run(ctx context.Context) error {
 	p := newProfilerInterface(s.pprofAddr, s.logger)
-	if err := p.startEndpoint(); err != nil {
-		return fmt.Errorf("could not start pprof endpoint: %v", err)
-	}
+	p.startEndpoint()
 	if err := p.startCpuProfileCache(); err != nil {
 		return fmt.Errorf("could not start CPU profile: %v", err)
 	}
@@ -109,45 +100,18 @@ func (s *serve) run(ctx context.Context) error {
 
 	s.logger = s.logger.WithFields(logrus.Fields{"configs": s.configDir, "port": s.port})
 
-	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)
+	store, err := registry.NewQuerierFromFS(os.DirFS(s.configDir), s.cacheDir)
+	defer store.Close()
 	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
 	}
 
 	lis, err := net.Listen("tcp", ":"+s.port)
 	if err != nil {
-		return fmt.Errorf("failed to listen: %s", err)
+		s.logger.Fatalf("failed to listen: %s", err)
 	}
 
 	grpcServer := grpc.NewServer()
@@ -161,9 +125,7 @@ func (s *serve) run(ctx context.Context) error {
 		return grpcServer.Serve(lis)
 	}, func() {
 		grpcServer.GracefulStop()
-		if err := p.stopEndpoint(ctx); err != nil {
-			s.logger.Warnf("error shutting down pprof server: %v", err)
-		}
+		p.stopEndpoint(p.logger.Context)
 	})
 
 }
@@ -181,8 +143,7 @@ type profilerInterface struct {
 	cacheReady bool
 	cacheLock  sync.RWMutex
 
-	logger   *logrus.Entry
-	closeErr chan error
+	logger *logrus.Entry
 }
 
 func newProfilerInterface(a string, log *logrus.Entry) *profilerInterface {
@@ -197,10 +158,10 @@ func (p *profilerInterface) isEnabled() bool {
 	return p.addr != ""
 }
 
-func (p *profilerInterface) startEndpoint() error {
+func (p *profilerInterface) startEndpoint() {
 	// short-circuit if not enabled
 	if !p.isEnabled() {
-		return nil
+		return
 	}
 
 	mux := http.NewServeMux()
@@ -216,22 +177,14 @@ func (p *profilerInterface) startEndpoint() error {
 		Handler: mux,
 	}
 
-	lis, err := net.Listen("tcp", p.addr)
-	if err != nil {
-		return err
-	}
-
-	p.closeErr = make(chan error)
+	// goroutine exits with main
 	go func() {
-		p.closeErr <- func() error {
-			p.logger.Info("starting pprof endpoint")
-			if err := p.server.Serve(lis); err != nil && !errors.Is(err, http.ErrServerClosed) {
-				return err
-			}
-			return nil
-		}()
+
+		p.logger.Info("starting pprof endpoint")
+		if err := p.server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			p.logger.Fatal(err)
+		}
 	}()
-	return nil
 }
 
 func (p *profilerInterface) startCpuProfileCache() error {
@@ -265,14 +218,10 @@ func (p *profilerInterface) httpHandler(w http.ResponseWriter, r *http.Request)
 	w.Write(p.cache.Bytes())
 }
 
-func (p *profilerInterface) stopEndpoint(ctx context.Context) error {
-	if !p.isEnabled() {
-		return nil
-	}
+func (p *profilerInterface) stopEndpoint(ctx context.Context) {
 	if err := p.server.Shutdown(ctx); err != nil {
-		return err
+		p.logger.Fatal(err)
 	}
-	return <-p.closeErr
 }
 
 func (p *profilerInterface) isCacheReady() bool {