pkg/cvo/metrics: Graceful server shutdown

pkg/cvo/availableupdates.go

@@ -2,7 +2,6 @@ package cvo
 
 import (
 	"crypto/tls"
-	"crypto/x509"
 	"fmt"
 	"net/url"
 	"runtime"
@@ -11,7 +10,6 @@ import (
 	"github.com/blang/semver"
 	"github.com/google/uuid"
 	"k8s.io/apimachinery/pkg/api/equality"
-	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/klog"
 
@@ -224,54 +222,3 @@ func calculateAvailableUpdatesStatus(clusterID string, proxyURL *url.URL, tlsCon
 		LastTransitionTime: metav1.Now(),
 	}
 }
-
-// getHTTPSProxyURL returns a url.URL object for the configured
-// https proxy only. It can be nil if does not exist or there is an error.
-func (optr *Operator) getHTTPSProxyURL() (*url.URL, string, error) {
-	proxy, err := optr.proxyLister.Get("cluster")
-
-	if errors.IsNotFound(err) {
-		return nil, "", nil
-	}
-	if err != nil {
-		return nil, "", err
-	}
-
-	if &proxy.Spec != nil {
-		if proxy.Spec.HTTPSProxy != "" {
-			proxyURL, err := url.Parse(proxy.Spec.HTTPSProxy)
-			if err != nil {
-				return nil, "", err
-			}
-			return proxyURL, proxy.Spec.TrustedCA.Name, nil
-		}
-	}
-	return nil, "", nil
-}
-
-func (optr *Operator) getTLSConfig(cmNameRef string) (*tls.Config, error) {
-	cm, err := optr.cmConfigLister.Get(cmNameRef)
-
-	if err != nil {
-		return nil, err
-	}
-
-	certPool, _ := x509.SystemCertPool()
-	if certPool == nil {
-		certPool = x509.NewCertPool()
-	}
-
-	if cm.Data["ca-bundle.crt"] != "" {
-		if ok := certPool.AppendCertsFromPEM([]byte(cm.Data["ca-bundle.crt"])); !ok {
-			return nil, fmt.Errorf("unable to add ca-bundle.crt certificates")
-		}
-	} else {
-		return nil, nil
-	}
-
-	config := &tls.Config{
-		RootCAs: certPool,
-	}
-
-	return config, nil
-}

pkg/cvo/cvo.go

@@ -169,7 +169,6 @@ func New(
 	proxyInformer configinformersv1.ProxyInformer,
 	client clientset.Interface,
 	kubeClient kubernetes.Interface,
-	enableMetrics bool,
 	exclude string,
 ) *Operator {
 	eventBroadcaster := record.NewBroadcaster()
@@ -214,11 +213,6 @@ func New(
 	// make sure this is initialized after all the listers are initialized
 	optr.upgradeableChecks = optr.defaultUpgradeableChecks()
 
-	if enableMetrics {
-		if err := optr.registerMetrics(coInformer.Informer()); err != nil {
-			panic(err)
-		}
-	}
 	return optr
 }
 

pkg/cvo/egress.go

@@ -0,0 +1,61 @@
+package cvo
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"net/url"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+)
+
+// getHTTPSProxyURL returns a url.URL object for the configured
+// https proxy only. It can be nil if does not exist or there is an error.
+func (optr *Operator) getHTTPSProxyURL() (*url.URL, string, error) {
+	proxy, err := optr.proxyLister.Get("cluster")
+
+	if errors.IsNotFound(err) {
+		return nil, "", nil
+	}
+	if err != nil {
+		return nil, "", err
+	}
+
+	if &proxy.Spec != nil {
+		if proxy.Spec.HTTPSProxy != "" {
+			proxyURL, err := url.Parse(proxy.Spec.HTTPSProxy)
+			if err != nil {
+				return nil, "", err
+			}
+			return proxyURL, proxy.Spec.TrustedCA.Name, nil
+		}
+	}
+	return nil, "", nil
+}
+
+func (optr *Operator) getTLSConfig(cmNameRef string) (*tls.Config, error) {
+	cm, err := optr.cmConfigLister.Get(cmNameRef)
+
+	if err != nil {
+		return nil, err
+	}
+
+	certPool, _ := x509.SystemCertPool()
+	if certPool == nil {
+		certPool = x509.NewCertPool()
+	}
+
+	if cm.Data["ca-bundle.crt"] != "" {
+		if ok := certPool.AppendCertsFromPEM([]byte(cm.Data["ca-bundle.crt"])); !ok {
+			return nil, fmt.Errorf("unable to add ca-bundle.crt certificates")
+		}
+	} else {
+		return nil, nil
+	}
+
+	config := &tls.Config{
+		RootCAs: certPool,
+	}
+
+	return config, nil
+}

pkg/cvo/metrics.go

@@ -1,21 +1,30 @@
 package cvo
 
 import (
+	"context"
+	"crypto/tls"
+	"net"
+	"net/http"
 	"time"
 
+	"github.com/cockroachdb/cmux"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog"
 
 	configv1 "github.com/openshift/api/config/v1"
 	"github.com/openshift/cluster-version-operator/lib/resourcemerge"
 	"github.com/openshift/cluster-version-operator/pkg/internal"
 )
 
-func (optr *Operator) registerMetrics(coInformer cache.SharedInformer) error {
+// RegisterMetrics initializes metrics and registers them with the
+// Prometheus implementation.
+func (optr *Operator) RegisterMetrics(coInformer cache.SharedInformer) error {
 	m := newOperatorMetrics(optr)
 	coInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		UpdateFunc: m.clusterOperatorChanged,
@@ -86,6 +95,88 @@ version for 'cluster', or empty for 'initial'.
 	}
 }
 
+// RunMetrics launches an server bound to listenAddress serving
+// Prometheus metrics at /metrics over HTTP, and, if tlsConfig is
+// non-nil, also over HTTPS.  Continues serving until runContext.Done()
+// and then attempts a clean shutdown limited by shutdownContext.Done().
+// Assumes runContext.Done() occurs before or simultaneously with
+// shutdownContext.Done().
+func RunMetrics(runContext context.Context, shutdownContext context.Context, listenAddress string, tlsConfig *tls.Config) error {
+	handler := http.NewServeMux()
+	handler.Handle("/metrics", promhttp.Handler())
+	server := &http.Server{
+		Handler: handler,
+	}
+
+	tcpListener, err := net.Listen("tcp", listenAddress)
+	if err != nil {
+		return err
+	}
+
+	// if a TLS connection was requested, set up a connection mux that will send TLS requests to
+	// the TLS server but send HTTP requests to the HTTP server. Preserves the ability for legacy
+	// HTTP, needed during upgrade, while still allowing TLS certs and end to end metrics protection.
+	mux := cmux.New(tcpListener)
+
+	errorChannel := make(chan error, 1)
+	errorChannelCount := 1
+
+	go func() {
+		// match HTTP first
+		httpListener := mux.Match(cmux.HTTP1())
+		klog.Infof("Metrics port listening for HTTP on %v", listenAddress)
+		errorChannel <- server.Serve(httpListener)
+	}()
+
+	if tlsConfig != nil {
+		errorChannelCount++
+		go func() {
+			tlsListener := tls.NewListener(mux.Match(cmux.Any()), tlsConfig)
+			klog.Infof("Metrics port listening for HTTPS on %v", listenAddress)
+			errorChannel <- server.Serve(tlsListener)
+		}()
+	}
+
+	errorChannelCount++
+	go func() {
+		errorChannel <- mux.Serve()
+	}()
+
+	shutdown := false
+	var loopError error
+	for errorChannelCount > 0 {
+		if shutdown {
+			err := <-errorChannel
+			errorChannelCount--
+			if err != nil && err != http.ErrServerClosed && err != cmux.ErrListenerClosed {
+				if loopError == nil {
+					loopError = err
+				} else if err != nil { // log the error we are discarding
+					klog.Errorf("Failed to gracefully shut down metrics server: %s", err)
+				}
+			}
+		} else {
+			select {
+			case <-runContext.Done(): // clean shutdown
+			case err := <-errorChannel: // crashed before a shutdown was requested
+				errorChannelCount--
+				if err != nil && err != http.ErrServerClosed && err != cmux.ErrListenerClosed {
+					loopError = err
+				}
+			}
+			shutdown = true
+			shutdownError := server.Shutdown(shutdownContext)
+			if loopError == nil {
+				loopError = shutdownError
+			} else if shutdownError != nil { // log the error we are discarding
+				klog.Errorf("Failed to gracefully shut down metrics server: %s", shutdownError)
+			}
+		}
+	}
+
+	return loopError
+}
+
 type conditionKey struct {
 	Name string
 	Type string