Merge pull request #374 from grokspawn/master

OLM-2726, OCPBUGS-643: downstreaming `opm serve` changes from operator-registry

Author: openshift-merge-robot

staging/operator-registry/alpha/action/generate_dockerfile.go

@@ -45,10 +45,11 @@ FROM {{.BaseImage}}
 
 # Configure the entrypoint and command
 ENTRYPOINT ["/bin/opm"]
-CMD ["serve", "/configs"]
+CMD ["serve", "/configs", "--cache-dir=/tmp/cache"]
 
-# Copy declarative config root into image at /configs
+# Copy declarative config root into image at /configs and pre-populate serve cache
 ADD {{.IndexDir}} /configs
+RUN ["/bin/opm", "serve", "/configs", "--cache-dir=/tmp/cache", "--cache-only"]
 
 # Set DC-specific label for the location of the DC root directory
 # in the image

staging/operator-registry/alpha/action/generate_dockerfile_test.go

@@ -50,10 +50,11 @@ FROM foo
 
 # Configure the entrypoint and command
 ENTRYPOINT ["/bin/opm"]
-CMD ["serve", "/configs"]
+CMD ["serve", "/configs", "--cache-dir=/tmp/cache"]
 
-# Copy declarative config root into image at /configs
+# Copy declarative config root into image at /configs and pre-populate serve cache
 ADD bar /configs
+RUN ["/bin/opm", "serve", "/configs", "--cache-dir=/tmp/cache", "--cache-only"]
 
 # Set DC-specific label for the location of the DC root directory
 # in the image
@@ -76,10 +77,11 @@ FROM foo
 
 # Configure the entrypoint and command
 ENTRYPOINT ["/bin/opm"]
-CMD ["serve", "/configs"]
+CMD ["serve", "/configs", "--cache-dir=/tmp/cache"]
 
-# Copy declarative config root into image at /configs
+# Copy declarative config root into image at /configs and pre-populate serve cache
 ADD bar /configs
+RUN ["/bin/opm", "serve", "/configs", "--cache-dir=/tmp/cache", "--cache-only"]
 
 # Set DC-specific label for the location of the DC root directory
 # in the image

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

@@ -115,7 +115,7 @@ func serveFunc(cmd *cobra.Command, _ []string) error {
 
 	lis, err := net.Listen("tcp", ":"+port)
 	if err != nil {
-		logger.Fatalf("failed to listen: %s", err)
+		return fmt.Errorf("failed to listen: %s", err)
 	}
 
 	timeout, err := cmd.Flags().GetString("timeout-seconds")

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

@@ -6,19 +6,17 @@ import (
 	"errors"
 	"fmt"
 	"net"
-	"os"
-	"sync"
-
 	"net/http"
 	endpoint "net/http/pprof"
+	"os"
 	"runtime/pprof"
+	"sync"
 
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
 
-	"github.com/operator-framework/operator-registry/alpha/declcfg"
 	"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/lib/dns"
@@ -30,6 +28,8 @@ import (
 
 type serve struct {
 	configDir string
+	cacheDir  string
+	cacheOnly bool
 
 	port           string
 	terminationLog string
@@ -75,12 +75,16 @@ 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().StringVarP(&s.port, "port", "p", "50051", "port number to serve on")
 	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")
 	return cmd
 }
 
 func (s *serve) run(ctx context.Context) error {
 	p := newProfilerInterface(s.pprofAddr, s.logger)
-	p.startEndpoint()
+	if err := p.startEndpoint(); err != nil {
+		return fmt.Errorf("could not start pprof endpoint: %v", err)
+	}
 	if err := p.startCpuProfileCache(); err != nil {
 		return fmt.Errorf("could not start CPU profile: %v", err)
 	}
@@ -98,24 +102,18 @@ func (s *serve) run(ctx context.Context) error {
 
 	s.logger = s.logger.WithFields(logrus.Fields{"configs": s.configDir, "port": s.port})
 
-	cfg, err := declcfg.LoadFS(os.DirFS(s.configDir))
-	if err != nil {
-		return fmt.Errorf("load declarative config directory: %v", err)
-	}
-
-	m, err := declcfg.ConvertToModel(*cfg)
-	if err != nil {
-		return fmt.Errorf("could not build index model from declarative config: %v", err)
-	}
-	store, err := registry.NewQuerier(m)
+	store, err := registry.NewQuerierFromFS(os.DirFS(s.configDir), s.cacheDir)
 	defer store.Close()
 	if err != nil {
 		return err
 	}
+	if s.cacheOnly {
+		return nil
+	}
 
 	lis, err := net.Listen("tcp", ":"+s.port)
 	if err != nil {
-		s.logger.Fatalf("failed to listen: %s", err)
+		return fmt.Errorf("failed to listen: %s", err)
 	}
 
 	grpcServer := grpc.NewServer()
@@ -129,7 +127,9 @@ func (s *serve) run(ctx context.Context) error {
 		return grpcServer.Serve(lis)
 	}, func() {
 		grpcServer.GracefulStop()
-		p.stopEndpoint(p.logger.Context)
+		if err := p.stopEndpoint(ctx); err != nil {
+			s.logger.Warnf("error shutting down pprof server: %v", err)
+		}
 	})
 
 }
@@ -147,7 +147,8 @@ type profilerInterface struct {
 	cacheReady bool
 	cacheLock  sync.RWMutex
 
-	logger *logrus.Entry
+	logger   *logrus.Entry
+	closeErr chan error
 }
 
 func newProfilerInterface(a string, log *logrus.Entry) *profilerInterface {
@@ -162,10 +163,10 @@ func (p *profilerInterface) isEnabled() bool {
 	return p.addr != ""
 }
 
-func (p *profilerInterface) startEndpoint() {
+func (p *profilerInterface) startEndpoint() error {
 	// short-circuit if not enabled
 	if !p.isEnabled() {
-		return
+		return nil
 	}
 
 	mux := http.NewServeMux()
@@ -181,14 +182,22 @@ func (p *profilerInterface) startEndpoint() {
 		Handler: mux,
 	}
 
-	// goroutine exits with main
-	go func() {
+	lis, err := net.Listen("tcp", p.addr)
+	if err != nil {
+		return err
+	}
 
-		p.logger.Info("starting pprof endpoint")
-		if err := p.server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			p.logger.Fatal(err)
-		}
+	p.closeErr = make(chan error)
+	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
+		}()
 	}()
+	return nil
 }
 
 func (p *profilerInterface) startCpuProfileCache() error {
@@ -222,10 +231,14 @@ func (p *profilerInterface) httpHandler(w http.ResponseWriter, r *http.Request)
 	w.Write(p.cache.Bytes())
 }
 
-func (p *profilerInterface) stopEndpoint(ctx context.Context) {
+func (p *profilerInterface) stopEndpoint(ctx context.Context) error {
+	if !p.isEnabled() {
+		return nil
+	}
 	if err := p.server.Shutdown(ctx); err != nil {
-		p.logger.Fatal(err)
+		return err
 	}
+	return <-p.closeErr
 }
 
 func (p *profilerInterface) isCacheReady() bool {

staging/operator-registry/go.mod

@@ -35,6 +35,7 @@ require (
 	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3
 	golang.org/x/net v0.0.0-20220407224826-aac1ed45d8e3
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
+	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad
 	google.golang.org/grpc v1.45.0
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v0.0.0-20200709232328-d8193ee9cc3e
 	google.golang.org/protobuf v1.28.0
@@ -148,7 +149,6 @@ require (
 	go.opentelemetry.io/proto/otlp v0.7.0 // indirect
 	golang.org/x/crypto v0.0.0-20220408190544-5352b0902921 // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
-	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect