Merge pull request #540 from stevekuznetsov/skuznets/cherry-pick-memory

OCPBUGS-17791: opm: always serve pprof endpoints, improve server allocations (#1129)

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

@@ -36,8 +36,9 @@ type serve struct {
 	port           string
 	terminationLog string
 
-	debug     bool
-	pprofAddr string
+	debug           bool
+	pprofAddr       string
+	captureProfiles bool
 
 	logger *logrus.Entry
 }
@@ -80,7 +81,8 @@ will not be reflected in the served content.
 	cmd.Flags().BoolVar(&s.debug, "debug", false, "enable debug logging")
 	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.pprofAddr, "pprof-addr", "localhost:6060", "address of startup profiling endpoint (addr:port format)")
+	cmd.Flags().BoolVar(&s.captureProfiles, "pprof-capture-profiles", false, "capture pprof CPU profiles")
 	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), ")
@@ -92,8 +94,10 @@ func (s *serve) run(ctx context.Context) error {
 	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)
+	if s.captureProfiles {
+		if err := p.startCpuProfileCache(); err != nil {
+			return fmt.Errorf("could not start CPU profile: %v", err)
+		}
 	}
 
 	// Immediately set up termination log

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

@@ -159,13 +159,16 @@ func (q *JSON) existingDigest() (string, error) {
 }
 
 func (q *JSON) computeDigest(fbcFsys fs.FS) (string, error) {
+	// We are not sensitive to the size of this buffer, we just need it to be shared.
+	// For simplicity, do the same as io.Copy() would.
+	buf := make([]byte, 32*1024)
 	computedHasher := fnv.New64a()
-	if err := fsToTar(computedHasher, fbcFsys); err != nil {
+	if err := fsToTar(computedHasher, fbcFsys, buf); err != nil {
 		return "", err
 	}
 
 	if cacheFS, err := fs.Sub(os.DirFS(q.baseDir), jsonDir); err == nil {
-		if err := fsToTar(computedHasher, cacheFS); err != nil && !errors.Is(err, os.ErrNotExist) {
+		if err := fsToTar(computedHasher, cacheFS, buf); err != nil && !errors.Is(err, os.ErrNotExist) {
 			return "", fmt.Errorf("compute hash: %v", err)
 		}
 	}

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

@@ -13,7 +13,12 @@ import (
 // This function unsets user and group information in the tar archive so that readers
 // of archives produced by this function do not need to account for differences in
 // permissions between source and destination filesystems.
-func fsToTar(w io.Writer, fsys fs.FS) error {
+func fsToTar(w io.Writer, fsys fs.FS, buf []byte) error {
+	if buf == nil || len(buf) == 0 {
+		// We are not sensitive to the size of this buffer, we just need it to be shared.
+		// For simplicity, do the same as io.Copy() would.
+		buf = make([]byte, 32*1024)
+	}
 	tw := tar.NewWriter(w)
 	if err := fs.WalkDir(fsys, ".", func(path string, d fs.DirEntry, err error) error {
 		if err != nil {
@@ -52,7 +57,7 @@ func fsToTar(w io.Writer, fsys fs.FS) error {
 			return fmt.Errorf("open file %q: %v", path, err)
 		}
 		defer f.Close()
-		if _, err := io.Copy(tw, f); err != nil {
+		if _, err := io.CopyBuffer(tw, f, buf); err != nil {
 			return fmt.Errorf("write tar data for %q: %v", path, err)
 		}
 		return nil

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

@@ -51,7 +51,7 @@ func Test_fsToTar(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			w := bytes.Buffer{}
-			err := fsToTar(&w, tc.fsys())
+			err := fsToTar(&w, tc.fsys(), nil)
 			tc.expect(t, w.Bytes(), err)
 		})
 	}