introduce pprof server to manager

Author: zqzten

pkg/config/v1alpha1/types.go

@@ -68,6 +68,10 @@ type ControllerManagerConfigurationSpec struct {
 // +optional
 Health ControllerHealth `json:"health,omitempty"`
 
+// Pprof contains the controller pprof configuration
+// +optional
+Pprof ControllerPprof `json:"pprof,omitempty"`
+
 // Webhook contains the controllers webhook configuration
 // +optional
 Webhook ControllerWebhook `json:"webhook,omitempty"`
@@ -121,6 +125,15 @@ type ControllerHealth struct {
 LivenessEndpointName string `json:"livenessEndpointName,omitempty"`
 }
 
+// ControllerPprof defines the pprof configs.
+type ControllerPprof struct {
+// BindAddress is the TCP address that the controller should bind to
+// for serving pprof
+// It can be set to "" or "0" to disable the pprof serving.
+// +optional
+BindAddress string `json:"bindAddress,omitempty"`
+}
+
 // ControllerWebhook defines the webhook server for the controller.
 type ControllerWebhook struct {
 // Port is the port that the webhook server serves at.

pkg/config/v1alpha1/zz_generated.deepcopy.go

@@ -22,6 +22,7 @@ import (
 "fmt"
 "net"
 "net/http"
+"net/http/pprof"
 "sync"
 "sync/atomic"
 "time"
@@ -106,6 +107,9 @@ type controllerManager struct {
 // Healthz probe handler
 healthzHandler *healthz.Handler
 
+// pprofListener is used to serve pprof
+pprofListener net.Listener
+
 // controllerOptions are the global controller options.
 controllerOptions v1alpha1.ControllerConfigurationSpec
 
@@ -355,6 +359,19 @@ func (cm *controllerManager) serveHealthProbes() {
 go cm.httpServe("health probe", cm.logger, server, cm.healthProbeListener)
 }
 
+func (cm *controllerManager) servePprof() {
+mux := http.NewServeMux()
+server := httpserver.New(mux)
+
+mux.HandleFunc("/debug/pprof/", pprof.Index)
+mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
+mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
+
+go cm.httpServe("pprof", cm.logger, server, cm.pprofListener)
+}
+
 func (cm *controllerManager) httpServe(kind string, log logr.Logger, server *http.Server, ln net.Listener) {
 log = log.WithValues("kind", kind, "addr", ln.Addr())
 
@@ -450,6 +467,11 @@ func (cm *controllerManager) Start(ctx context.Context) (err error) {
 cm.serveHealthProbes()
 }
 
+// Serve pprof.
+if cm.pprofListener != nil {
+cm.servePprof()
+}
+
 // First start any webhook servers, which includes conversion, validation, and defaulting
 // webhooks that are registered.
 //

pkg/manager/internal.go

@@ -227,6 +227,10 @@ type Options struct {
 // Liveness probe endpoint name, defaults to "healthz"
 LivenessEndpointName string
 
+// PprofBindAddress is the TCP address that the controller should bind to
+// for serving pprof
+PprofBindAddress string
+
 // Port is the port that the webhook server serves at.
 // It is used to set webhook.Server.Port if WebhookServer is not set.
 Port int
@@ -299,6 +303,7 @@ type Options struct {
 newResourceLock func(config *rest.Config, recorderProvider recorder.Provider, options leaderelection.Options) (resourcelock.Interface, error)
 newMetricsListener func(addr string) (net.Listener, error)
 newHealthProbeListener func(addr string) (net.Listener, error)
+newPprofListener func(addr string) (net.Listener, error)
 }
 
 // BaseContextFunc is a function used to provide a base Context to Runnables
@@ -403,6 +408,13 @@ func New(config *rest.Config, options Options) (Manager, error) {
 return nil, err
 }
 
+// Create pprof listener. This will throw an error if the bind
+// address is invalid or already in use.
+pprofListener, err := options.newPprofListener(options.PprofBindAddress)
+if err != nil {
+return nil, err
+}
+
 errChan := make(chan error)
 runnables := newRunnables(options.BaseContext, errChan)
 
@@ -428,6 +440,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 healthProbeListener: healthProbeListener,
 readinessEndpointName: options.ReadinessEndpointName,
 livenessEndpointName: options.LivenessEndpointName,
+pprofListener: pprofListener,
 gracefulShutdownTimeout: *options.GracefulShutdownTimeout,
 internalProceduresStop: make(chan struct{}),
 leaderElectionStopped: make(chan struct{}),
@@ -477,6 +490,10 @@ func (o Options) AndFrom(loader config.ControllerManagerConfiguration) (Options,
 o.LivenessEndpointName = newObj.Health.LivenessEndpointName
 }
 
+if o.PprofBindAddress == "" && newObj.Pprof.BindAddress != "" {
+o.PprofBindAddress = newObj.Pprof.BindAddress
+}
+
 if o.Port == 0 && newObj.Webhook.Port != nil {
 o.Port = *newObj.Webhook.Port
 }
@@ -561,6 +578,19 @@ func defaultHealthProbeListener(addr string) (net.Listener, error) {
 return ln, nil
 }
 
+// defaultPprofListener creates the default pprof listener bound to the given address.
+func defaultPprofListener(addr string) (net.Listener, error) {
+if addr == "" || addr == "0" {
+return nil, nil
+}
+
+ln, err := net.Listen("tcp", addr)
+if err != nil {
+return nil, fmt.Errorf("error listening on %s: %w", addr, err)
+}
+return ln, nil
+}
+
 // defaultBaseContext is used as the BaseContext value in Options if one
 // has not already been set.
 func defaultBaseContext() context.Context {
@@ -621,6 +651,10 @@ func setOptionsDefaults(options Options) Options {
 options.newHealthProbeListener = defaultHealthProbeListener
 }
 
+if options.newPprofListener == nil {
+options.newPprofListener = defaultPprofListener
+}
+
 if options.GracefulShutdownTimeout == nil {
 gracefulShutdownTimeout := defaultGracefulShutdownPeriod
 options.GracefulShutdownTimeout = &gracefulShutdownTimeout

pkg/manager/manager.go

@@ -147,6 +147,9 @@ var _ = Describe("manger.Manager", func() {
 ReadinessEndpointName: "/readyz",
 LivenessEndpointName: "/livez",
 },
+Pprof: v1alpha1.ControllerPprof{
+BindAddress: ":6080",
+},
 Webhook: v1alpha1.ControllerWebhook{
 Port: &port,
 Host: "localhost",
@@ -171,6 +174,7 @@ var _ = Describe("manger.Manager", func() {
 Expect(m.HealthProbeBindAddress).To(Equal("6060"))
 Expect(m.ReadinessEndpointName).To(Equal("/readyz"))
 Expect(m.LivenessEndpointName).To(Equal("/livez"))
+Expect(m.PprofBindAddress).To(Equal(":6080"))
 Expect(m.Port).To(Equal(port))
 Expect(m.Host).To(Equal("localhost"))
 Expect(m.CertDir).To(Equal("/certs"))
@@ -203,6 +207,9 @@ var _ = Describe("manger.Manager", func() {
 ReadinessEndpointName: "/readyz",
 LivenessEndpointName: "/livez",
 },
+Pprof: v1alpha1.ControllerPprof{
+BindAddress: ":6080",
+},
 Webhook: v1alpha1.ControllerWebhook{
 Port: &port,
 Host: "localhost",
@@ -225,6 +232,7 @@ var _ = Describe("manger.Manager", func() {
 HealthProbeBindAddress: "5000",
 ReadinessEndpointName: "/readiness",
 LivenessEndpointName: "/liveness",
+PprofBindAddress: ":6000",
 Port: 8080,
 Host: "example.com",
 CertDir: "/pki",
@@ -244,6 +252,7 @@ var _ = Describe("manger.Manager", func() {
 Expect(m.HealthProbeBindAddress).To(Equal("5000"))
 Expect(m.ReadinessEndpointName).To(Equal("/readiness"))
 Expect(m.LivenessEndpointName).To(Equal("/liveness"))
+Expect(m.PprofBindAddress).To(Equal(":6000"))
 Expect(m.Port).To(Equal(8080))
 Expect(m.Host).To(Equal("example.com"))
 Expect(m.CertDir).To(Equal("/pki"))
@@ -282,6 +291,7 @@ var _ = Describe("manger.Manager", func() {
 LeaderElectionID: "test-leader-election-id-2",
 HealthProbeBindAddress: "0",
 MetricsBindAddress: "0",
+PprofBindAddress: "0",
 })
 Expect(err).To(BeNil())
 
@@ -327,6 +337,7 @@ var _ = Describe("manger.Manager", func() {
 LeaderElectionID: "test-leader-election-id-3",
 HealthProbeBindAddress: "0",
 MetricsBindAddress: "0",
+PprofBindAddress: "0",
 })
 Expect(err).To(BeNil())
 
@@ -361,6 +372,7 @@ var _ = Describe("manger.Manager", func() {
 },
 HealthProbeBindAddress: "0",
 MetricsBindAddress: "0",
+PprofBindAddress: "0",
 })
 Expect(err).ToNot(HaveOccurred())
 Expect(m1).ToNot(BeNil())
@@ -381,6 +393,7 @@ var _ = Describe("manger.Manager", func() {
 },
 HealthProbeBindAddress: "0",
 MetricsBindAddress: "0",
+PprofBindAddress: "0",
 })
 Expect(err).ToNot(HaveOccurred())
 Expect(m2).ToNot(BeNil())
@@ -1395,6 +1408,94 @@ var _ = Describe("manger.Manager", func() {
 })
 })
 
+Context("should start serving pprof", func() {
+var listener net.Listener
+var opts Options
+
+BeforeEach(func() {
+listener = nil
+opts = Options{
+newPprofListener: func(addr string) (net.Listener, error) {
+var err error
+listener, err = defaultPprofListener(addr)
+return listener, err
+},
+}
+})
+
+AfterEach(func() {
+if listener != nil {
+listener.Close()
+}
+})
+
+It("should stop serving pprof when stop is called", func() {
+opts.PprofBindAddress = ":0"
+m, err := New(cfg, opts)
+Expect(err).NotTo(HaveOccurred())
+
+ctx, cancel := context.WithCancel(context.Background())
+go func() {
+defer GinkgoRecover()
+Expect(m.Start(ctx)).NotTo(HaveOccurred())
+}()
+<-m.Elected()
+
+// Check the pprof started
+endpoint := fmt.Sprintf("http://%s", listener.Addr().String())
+_, err = http.Get(endpoint)
+Expect(err).NotTo(HaveOccurred())
+
+// Shutdown the server
+cancel()
+
+// Expect the pprof server to shutdown
+Eventually(func() error {
+_, err = http.Get(endpoint)
+return err
+}).ShouldNot(Succeed())
+})
+
+It("should serve pprof endpoints", func() {
+opts.PprofBindAddress = ":0"
+m, err := New(cfg, opts)
+Expect(err).NotTo(HaveOccurred())
+
+ctx, cancel := context.WithCancel(context.Background())
+defer cancel()
+go func() {
+defer GinkgoRecover()
+Expect(m.Start(ctx)).NotTo(HaveOccurred())
+}()
+<-m.Elected()
+
+pprofIndexEndpoint := fmt.Sprintf("http://%s/debug/pprof/", listener.Addr().String())
+resp, err := http.Get(pprofIndexEndpoint)
+Expect(err).NotTo(HaveOccurred())
+Expect(resp.StatusCode).To(Equal(200))
+
+pprofCmdlineEndpoint := fmt.Sprintf("http://%s/debug/pprof/cmdline", listener.Addr().String())
+resp, err = http.Get(pprofCmdlineEndpoint)
+Expect(err).NotTo(HaveOccurred())
+Expect(resp.StatusCode).To(Equal(200))
+
+pprofProfileEndpoint := fmt.Sprintf("http://%s/debug/pprof/profile", listener.Addr().String())
+resp, err = http.Get(pprofProfileEndpoint)
+Expect(err).NotTo(HaveOccurred())
+Expect(resp.StatusCode).To(Equal(200))
+
+pprofSymbolEndpoint := fmt.Sprintf("http://%s/debug/pprof/symbol", listener.Addr().String())
+resp, err = http.Get(pprofSymbolEndpoint)
+Expect(err).NotTo(HaveOccurred())
+Expect(resp.StatusCode).To(Equal(200))
+
+pprofTraceEndpoint := fmt.Sprintf("http://%s/debug/pprof/trace", listener.Addr().String())
+resp, err = http.Get(pprofTraceEndpoint)
+Expect(err).NotTo(HaveOccurred())
+Expect(resp.StatusCode).To(Equal(200))
+})
+})
+
 Describe("Add", func() {
 It("should immediately start the Component if the Manager has already Started another Component",
 func() {