Start http server for serving metrics

Author: JoelSpeed

pkg/manager/internal.go

@@ -17,11 +17,14 @@ limitations under the License.
 package manager
 
 import (
+	"context"
 	"fmt"
 	"net"
+	"net/http"
 	"sync"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
@@ -30,6 +33,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
@@ -161,6 +165,32 @@ func (cm *controllerManager) GetRESTMapper() meta.RESTMapper {
 	return cm.mapper
 }
 
+func (cm *controllerManager) serveMetrics(stop <-chan struct{}) {
+	handler := promhttp.HandlerFor(metrics.Registry, promhttp.HandlerOpts{
+		ErrorHandling: promhttp.HTTPErrorOnError,
+	})
+	// TODO(JoelSpeed): Use existing Kubernetes machinery for serving metrics
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", handler)
+	server := http.Server{
+		Handler: mux,
+	}
+	// Run the server
+	go func() {
+		if err := server.Serve(cm.metricsListener); err != nil {
+			cm.errChan <- err
+		}
+	}()
+
+	// Shutdown the server when stop is closed
+	select {
+	case <-stop:
+		if err := server.Shutdown(context.Background()); err != nil {
+			cm.errChan <- err
+		}
+	}
+}
+
 func (cm *controllerManager) Start(stop <-chan struct{}) error {
 	if cm.resourceLock != nil {
 		err := cm.startLeaderElection(stop)
@@ -197,6 +227,11 @@ func (cm *controllerManager) start(stop <-chan struct{}) {
 		}
 	}()
 
+	// Start the metrics server
+	if cm.metricsListener != nil {
+		go cm.serveMetrics(stop)
+	}
+
 	// Wait for the caches to sync.
 	// TODO(community): Check the return value and write a test
 	cm.cache.WaitForCacheSync(stop)

pkg/manager/manager_test.go

@@ -18,11 +18,14 @@ package manager
 
 import (
 	"fmt"
+	"io/ioutil"
 	"net"
+	"net/http"
 
 	"github.com/go-logr/logr"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/prometheus/client_golang/prometheus"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
@@ -265,6 +268,132 @@ var _ = Describe("manger.Manager", func() {
 				newResourceLock:         fakeleaderelection.NewResourceLock,
 			})
 		})
+
+		Context("should start serving metrics", func() {
+			var listener net.Listener
+			var opts Options
+
+			BeforeEach(func() {
+				listener = nil
+				opts = Options{
+					newMetricsListener: func(addr string) (net.Listener, error) {
+						var err error
+						listener, err = metrics.NewListener(addr)
+						return listener, err
+					},
+				}
+			})
+
+			AfterEach(func() {
+				if listener != nil {
+					listener.Close()
+				}
+			})
+
+			It("should stop serving metrics when stop is called", func(done Done) {
+				opts.MetricsBindAddress = ":0"
+				m, err := New(cfg, opts)
+				Expect(err).NotTo(HaveOccurred())
+
+				s := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					Expect(m.Start(s)).NotTo(HaveOccurred())
+					close(done)
+				}()
+
+				// Check the metrics started
+				endpoint := fmt.Sprintf("http://%s", listener.Addr().String())
+				_, err = http.Get(endpoint)
+				Expect(err).NotTo(HaveOccurred())
+
+				// Shutdown the server
+				close(s)
+
+				// Expect the metrics server to shutdown
+				Eventually(func() error {
+					_, err = http.Get(endpoint)
+					return err
+				}).ShouldNot(Succeed())
+			})
+
+			It("should serve metrics endpoint", func(done Done) {
+				opts.MetricsBindAddress = ":0"
+				m, err := New(cfg, opts)
+				Expect(err).NotTo(HaveOccurred())
+
+				s := make(chan struct{})
+				defer close(s)
+				go func() {
+					defer GinkgoRecover()
+					Expect(m.Start(s)).NotTo(HaveOccurred())
+					close(done)
+				}()
+
+				metricsEndpoint := fmt.Sprintf("http://%s/metrics", listener.Addr().String())
+				resp, err := http.Get(metricsEndpoint)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(resp.StatusCode).To(Equal(200))
+			})
+
+			It("should not serve anything other than metrics endpoint", func(done Done) {
+				opts.MetricsBindAddress = ":0"
+				m, err := New(cfg, opts)
+				Expect(err).NotTo(HaveOccurred())
+
+				s := make(chan struct{})
+				defer close(s)
+				go func() {
+					defer GinkgoRecover()
+					Expect(m.Start(s)).NotTo(HaveOccurred())
+					close(done)
+				}()
+
+				endpoint := fmt.Sprintf("http://%s/should-not-exist", listener.Addr().String())
+				resp, err := http.Get(endpoint)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(resp.StatusCode).To(Equal(404))
+			})
+
+			It("should serve metrics in its registry", func(done Done) {
+				one := prometheus.NewCounter(prometheus.CounterOpts{
+					Name: "test_one",
+					Help: "test metric for testing",
+				})
+				one.Set(1)
+				err := metrics.Registry.Register(one)
+				Expect(err).NotTo(HaveOccurred())
+
+				opts.MetricsBindAddress = ":0"
+				m, err := New(cfg, opts)
+				Expect(err).NotTo(HaveOccurred())
+
+				s := make(chan struct{})
+				defer close(s)
+				go func() {
+					defer GinkgoRecover()
+					Expect(m.Start(s)).NotTo(HaveOccurred())
+					close(done)
+				}()
+
+				metricsEndpoint := fmt.Sprintf("http://%s/metrics", listener.Addr().String())
+				resp, err := http.Get(metricsEndpoint)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(resp.StatusCode).To(Equal(200))
+
+				data, err := ioutil.ReadAll(resp.Body)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(string(data)).To(ContainSubstring("%s\n%s\n%s\n",
+					`# HELP test_one test metric for testing`,
+					`# TYPE test_one counter`,
+					`test_one 1`,
+				))
+
+				// Unregister will return false if the metric was never registered
+				ok := metrics.Registry.Unregister(one)
+				Expect(ok).To(BeTrue())
+			})
+		})
 	})
 
 	Describe("Add", func() {