fix(ROSAENG-64778): dvo and metrics clients leaking goroutines

pkg/dvo/builder.go

@@ -3,6 +3,7 @@ package dvo
 import (
 	"net/http"
 	"os"
+	"time"
 
 	"github.com/openshift/managed-upgrade-operator/pkg/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -39,6 +40,7 @@ func (dcb *dvoClientBuilder) New(c client.Client) (DvoClient, error) {
 	}
 
 	httpClient := http.Client{
+		Timeout:   30 * time.Second,
 		Transport: dvoTransport(),
 	}
 

pkg/dvo/client.go

@@ -11,7 +11,7 @@ import (
 )
 
 const (
-	// CLUSTERS_V1_PATH is a path to the OCM clusters service
+	// METRICS_API_PATH is the path to the DVO metrics endpoint
 	METRICS_API_PATH = "/metrics"
 )
 

pkg/metrics/metrics.go

@@ -155,17 +155,32 @@ func (mb *metricsBuilder) NewClient(c client.Client) (Metrics, error) {
 	return &Counter{
 		promTarget: promTarget,
 		promClient: http.Client{
+			Timeout: 30 * time.Second,
 			Transport: &prometheusRoundTripper{
 				token: *token,
 				tls:   tlsConfig,
+				transport: &http.Transport{
+					// Configure proxy using Go's standard environment variable handling
+					// Respects HTTP_PROXY, HTTPS_PROXY, and NO_PROXY environment variables
+					// See: https://pkg.go.dev/net/http#ProxyFromEnvironment
+					Proxy: http.ProxyFromEnvironment,
+					// Configure timeouts for reliable Prometheus communication
+					DialContext: (&net.Dialer{
+						Timeout:   30 * time.Second, // Maximum time to establish TCP connection
+						KeepAlive: 30 * time.Second, // TCP keep-alive probe interval
+					}).DialContext,
+					TLSHandshakeTimeout: 30 * time.Second, // Maximum time for TLS handshake (increased from 5s for proxy environments)
+					TLSClientConfig:     tlsConfig,
+				},
 			},
 		},
 	}, nil
 }
 
 type prometheusRoundTripper struct {
-	token string
-	tls   *tls.Config
+	token     string
+	tls       *tls.Config
+	transport *http.Transport
 }
 
 // MonitoringTLSConfig accepts a client.Client and returns a *tls.Config for monitoring services using the monitoring
@@ -201,21 +216,7 @@ func MonitoringTLSConfig(c client.Client) (*tls.Config, error) {
 
 func (prt *prometheusRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
 	req.Header.Add("Authorization", "Bearer "+prt.token)
-	transport := http.Transport{
-		// Configure proxy using Go's standard environment variable handling
-		// Respects HTTP_PROXY, HTTPS_PROXY, and NO_PROXY environment variables
-		// See: https://pkg.go.dev/net/http#ProxyFromEnvironment
-		Proxy: http.ProxyFromEnvironment,
-
-		// Configure timeouts for reliable Prometheus communication
-		DialContext: (&net.Dialer{
-			Timeout:   30 * time.Second, // Maximum time to establish TCP connection
-			KeepAlive: 30 * time.Second, // TCP keep-alive probe interval
-		}).DialContext,
-		TLSHandshakeTimeout: 30 * time.Second, // Maximum time for TLS handshake (increased from 5s for proxy environments)
-		TLSClientConfig:     prt.tls,
-	}
-	return transport.RoundTrip(req)
+	return prt.transport.RoundTrip(req)
 }
 
 type Counter struct {

pkg/metrics/metrics_suite_test.go

@@ -0,0 +1,13 @@
+package metrics
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestMetrics(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Metrics Suite")
+}

pkg/metrics/metrics_test.go

@@ -0,0 +1,146 @@
+package metrics
+
+import (
+	"encoding/pem"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"runtime"
+	"strconv"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	gomock "go.uber.org/mock/gomock"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/openshift/managed-upgrade-operator/util/mocks"
+)
+
+func newMockMetricsClient(server *httptest.Server) *Counter {
+	mockCtrl := gomock.NewController(GinkgoT())
+	mockClient := mocks.NewMockClient(mockCtrl)
+
+	u, _ := url.Parse(server.URL)
+	port, _ := strconv.Atoi(u.Port())
+	svcPort := int32(port) //nolint:gosec
+
+	caCertPEM := pem.EncodeToMemory(&pem.Block{
+		Type:  "CERTIFICATE",
+		Bytes: server.Certificate().Raw,
+	})
+
+	mockClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ interface{}, _ interface{}, obj interface{}, _ ...interface{}) error {
+			switch o := obj.(type) {
+			case *corev1.Service:
+				*o = corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      promApp,
+						Namespace: MonitoringNS,
+					},
+					Spec: corev1.ServiceSpec{
+						Ports: []corev1.ServicePort{
+							{Name: "web", Port: svcPort},
+						},
+					},
+				}
+			case *corev1.ConfigMap:
+				*o = corev1.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      MonitoringCAConfigMapName,
+						Namespace: MonitoringNS,
+					},
+					Data: map[string]string{
+						MonitoringConfigField: string(caCertPEM),
+					},
+				}
+			}
+			return nil
+		},
+	).AnyTimes()
+
+	mockClient.EXPECT().List(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ interface{}, obj interface{}, _ ...interface{}) error {
+			if sl, ok := obj.(*corev1.SecretList); ok {
+				sl.Items = []corev1.Secret{
+					{
+						ObjectMeta: metav1.ObjectMeta{Name: "prometheus-k8s-token-test"},
+						Data:       map[string][]byte{corev1.ServiceAccountTokenKey: []byte("test-token")},
+					},
+				}
+			}
+			return nil
+		},
+	).AnyTimes()
+
+	mc, err := NewBuilder().NewClient(mockClient)
+	Expect(err).NotTo(HaveOccurred())
+
+	counter, ok := mc.(*Counter)
+	Expect(ok).To(BeTrue())
+	counter.promTarget = u.Host
+
+	return counter
+}
+
+var _ = Describe("Counter", func() {
+	var (
+		server *httptest.Server
+	)
+
+	AfterEach(func() {
+		if server != nil {
+			server.Close()
+		}
+	})
+
+	It("does not leak goroutines from Query", func() {
+		server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"status":"success","data":{"result":[]}}`))
+		}))
+		counter := newMockMetricsClient(server)
+
+		runtime.GC()
+		before := runtime.NumGoroutine()
+
+		for i := 0; i < 100; i++ {
+			_, err := counter.Query("up")
+			Expect(err).NotTo(HaveOccurred())
+		}
+
+		runtime.GC()
+		after := runtime.NumGoroutine()
+		leaked := after - before
+		Expect(leaked).To(BeNumerically("<=", 10),
+			"goroutine leak: %d before, %d after (%d leaked)", before, after, leaked)
+	})
+
+	It("returns parsed results from Query", func() {
+		server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			Expect(r.URL.Path).To(Equal("/api/v1/query"))
+			Expect(r.URL.Query().Get("query")).To(Equal("up"))
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"status":"success","data":{"result":[{"metric":{"alertname":"TestAlert"},"value":[1,"1"]}]}}`))
+		}))
+		counter := newMockMetricsClient(server)
+
+		resp, err := counter.Query("up")
+		Expect(err).NotTo(HaveOccurred())
+		Expect(resp.Data.Result).To(HaveLen(1))
+		Expect(resp.Data.Result[0].Metric["alertname"]).To(Equal("TestAlert"))
+	})
+
+	It("sets the Authorization header on Query", func() {
+		server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			Expect(r.Header.Get("Authorization")).To(Equal("Bearer test-token"))
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"status":"success","data":{"result":[]}}`))
+		}))
+		counter := newMockMetricsClient(server)
+
+		_, err := counter.Query("up")
+		Expect(err).NotTo(HaveOccurred())
+	})
+})

pkg/upgraders/healthcheck_pdb.go

@@ -54,7 +54,7 @@ func checkPodDisruptionBudgets(c client.Client, logger logr.Logger) ([]PDBDetail
 	}
 
 	for _, pdb := range pdbList.Items {
-		if !strings.HasPrefix(pdb.Namespace, "openshift-*") || checkNamespaceExistsInArray(namespaceException, pdb.Namespace) {
+		if !strings.HasPrefix(pdb.Namespace, "openshift-") || checkNamespaceExistsInArray(namespaceException, pdb.Namespace) {
 			pdbDetail := PDBDetails{}
 			pdbDetail.Name = pdb.Name
 			pdbDetail.Namespace = pdb.Namespace