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

* pkg/cache: use a shared buffer to limit allocations

Previously, new buffers were allocated on each file we read in, which
was unnecessary and wasteful.

Signed-off-by: Steve Kuznetsov <[email protected]>

* cmd/opm: serve pprof endpoints by default

There is no substantial runtime cost to serving pprof endpoints, and
when things hit the fan and we need to investigate performance in situ,
there is no time to restart pods and change flags. Capturing profiles
remains opt-in, since those are costly.

Signed-off-by: Steve Kuznetsov <[email protected]>

---------

Signed-off-by: Steve Kuznetsov <[email protected]>

Upstream-repository: operator-registry

Upstream-commit: 68e13df96590977370ffcd1a8e9ff76e0f2a03f2
Signed-off-by: Steve Kuznetsov <[email protected]>

pkg/manifests/csv.yaml

@@ -5,7 +5,7 @@ metadata:
  name: packageserver
  namespace: openshift-operator-lifecycle-manager
  labels:
- olm.version: 0.0.0-9f342c1be2d693ab5335ed27d10411cca471c937
+ olm.version: 0.19.0
  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-9f342c1be2d693ab5335ed27d10411cca471c937
+ version: 0.19.0
  apiservicedefinitions:
  owned:
  - group: packages.operators.coreos.com

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

@@ -165,13 +165,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)
  })
  }