Add pprof debug endpoint (#1064)

Allow users to opt-in to exposing pprof debugging endpoints on the operator's metrics port.
Users may add the `ENABLE_DEBUG_PPROF` env variable to the operator Pod to enable the pprof endpoints.
This is an opt-in pattern, as memory/CPU profiling can have a non-insignificant impact on runtime perf,
and could be a denial of service vector. These endpoints should not be enabled in production.

With this feature enabled, users can access the pprof output at `<OPERATOR_URL>:<METRICS_PORT>/debug/pprof`. For example, after
port-forwarding to the operator Pod:
```
go tool pprof -inuse_space -web http://localhost:9782/debug/pprof/heap
```

Author: coro

Dockerfile

@@ -13,6 +13,7 @@ COPY main.go main.go
 COPY api/ api/
 COPY controllers/ controllers/
 COPY internal/ internal/
+COPY pkg/ pkg/
 
 # Build
 RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 GO111MODULE=on go build -a -tags timetzdata -o manager main.go

Makefile

@@ -25,7 +25,7 @@ kubebuilder-assets: $(KUBEBUILDER_ASSETS)
 
 .PHONY: unit-tests
 unit-tests: install-tools $(KUBEBUILDER_ASSETS) generate fmt vet manifests ## Run unit tests
-	ginkgo -r --randomize-all api/ internal/
+	ginkgo -r --randomize-all api/ internal/ pkg/
 
 .PHONY: integration-tests
 integration-tests: install-tools $(KUBEBUILDER_ASSETS) generate fmt vet manifests ## Run integration tests

main.go

@@ -11,6 +11,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"github.com/rabbitmq/cluster-operator/pkg/profiling"
 	"os"
 	"strconv"
 	"time"
@@ -113,6 +114,17 @@ func main() {
 		os.Exit(1)
 	}
 
+	if enableDebugPprof, ok := os.LookupEnv("ENABLE_DEBUG_PPROF"); ok {
+		pprofEnabled, err := strconv.ParseBool(enableDebugPprof)
+		if err == nil && pprofEnabled {
+			mgr, err = profiling.AddDebugPprofEndpoints(mgr)
+			if err != nil {
+				log.Error(err, "unable to add debug endpoints to manager")
+				os.Exit(1)
+			}
+		}
+	}
+
 	clusterConfig := config.GetConfigOrDie()
 
 	err = (&controllers.RabbitmqClusterReconciler{

pkg/profiling/pprof.go

@@ -0,0 +1,30 @@
+package profiling
+
+import (
+	"net/http"
+	"net/http/pprof"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+func AddDebugPprofEndpoints(mgr ctrl.Manager) (ctrl.Manager, error) {
+	pprofEndpoints := map[string]http.HandlerFunc{
+		"/debug/pprof":              http.HandlerFunc(pprof.Index),
+		"/debug/pprof/allocs":       http.HandlerFunc(pprof.Index),
+		"/debug/pprof/block":        http.HandlerFunc(pprof.Index),
+		"/debug/pprof/cmdline":      http.HandlerFunc(pprof.Cmdline),
+		"/debug/pprof/goroutine":    http.HandlerFunc(pprof.Index),
+		"/debug/pprof/heap":         http.HandlerFunc(pprof.Index),
+		"/debug/pprof/mutex":        http.HandlerFunc(pprof.Index),
+		"/debug/pprof/profile":      http.HandlerFunc(pprof.Profile),
+		"/debug/pprof/symbol":       http.HandlerFunc(pprof.Symbol),
+		"/debug/pprof/threadcreate": http.HandlerFunc(pprof.Index),
+		"/debug/pprof/trace":        http.HandlerFunc(pprof.Trace),
+	}
+	for path, handler := range pprofEndpoints {
+		err := mgr.AddMetricsExtraHandler(path, handler)
+		if err != nil {
+			return mgr, err
+		}
+	}
+	return mgr, nil
+}

pkg/profiling/pprof_test.go

@@ -0,0 +1,53 @@
+package profiling_test
+
+import (
+	"context"
+	"fmt"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/rabbitmq/cluster-operator/pkg/profiling"
+	"io"
+	"net/http"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+var _ = Describe("Pprof", func() {
+
+	var (
+		opts            ctrl.Options
+		mgr             ctrl.Manager
+		err             error
+		metricsEndpoint string
+	)
+
+	BeforeEach(func() {
+		metricsEndpoint, err = getFreePort()
+		opts = ctrl.Options{
+			MetricsBindAddress: metricsEndpoint,
+		}
+		mgr, err = ctrl.NewManager(cfg, opts)
+		Expect(err).NotTo(HaveOccurred())
+		mgr, err = profiling.AddDebugPprofEndpoints(mgr)
+		Expect(err).NotTo(HaveOccurred())
+
+	})
+
+	It("should serve extra endpoints", func() {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		go func() {
+			defer GinkgoRecover()
+			Expect(mgr.Start(ctx)).NotTo(HaveOccurred())
+		}()
+		<-mgr.Elected()
+		endpoint := fmt.Sprintf("http://%s/debug/pprof", metricsEndpoint)
+		resp, err := http.Get(endpoint)
+		Expect(err).NotTo(HaveOccurred())
+		defer resp.Body.Close()
+		Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+		body, err := io.ReadAll(resp.Body)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(string(body)).NotTo(BeEmpty())
+	})
+})

pkg/profiling/profiling_suite_test.go

@@ -0,0 +1,52 @@
+package profiling_test
+
+import (
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"net"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var testenv *envtest.Environment
+var cfg *rest.Config
+var clientset *kubernetes.Clientset
+
+func TestProfiling(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Profiling Suite")
+}
+
+var _ = BeforeSuite(func() {
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
+
+	testenv = &envtest.Environment{}
+
+	var err error
+	cfg, err = testenv.Start()
+	Expect(err).NotTo(HaveOccurred())
+
+})
+
+var _ = AfterSuite(func() {
+	Expect(testenv.Stop()).To(Succeed())
+})
+
+func getFreePort() (string, error) {
+	addr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:0")
+	if err != nil {
+		return "", err
+	}
+
+	l, err := net.ListenTCP("tcp", addr)
+	if err != nil {
+		return "", err
+	}
+	defer l.Close()
+	return l.Addr().String(), nil
+}