Merge pull request #249 from jim-minter/monitorfixes

Monitor fixes

pkg/monitor/cache.go

@@ -56,7 +56,7 @@ func (mon *monitor) fixDocs() {
 // iff it is in a bucket owned by us.  Caller must hold mon.mu.Lock.
 func (mon *monitor) fixDoc(doc *api.OpenShiftClusterDocument) {
 	v := mon.docs[doc.ID]
-	_, ours := mon.buckets[mon.docs[doc.ID].doc.Bucket]
+	_, ours := mon.buckets[v.doc.Bucket]
 
 	if !ours && v.stop != nil {
 		close(v.stop)

pkg/monitor/cluster/alerts.go

@@ -16,31 +16,35 @@ import (
 	"github.com/Azure/ARO-RP/pkg/util/portforward"
 )
 
-const (
-	// alertNamespace is the namespace where the alert manager pod is living
-	alertNamespace string = "openshift-monitoring"
-	// alertPod is the pod to query
-	alertPod string = "alertmanager-main-0"
-	// alertServiceEndpoint is the service name to query
-	alertServiceEndpoint string = "http://alertmanager-main.openshift-monitoring.svc:9093/api/v2/alerts"
-)
-
 func (mon *Monitor) emitPrometheusAlerts(ctx context.Context) error {
-	hc := &http.Client{
-		Transport: &http.Transport{
-			DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
-				_, port, err := net.SplitHostPort(address)
-				if err != nil {
-					return nil, err
-				}
+	var resp *http.Response
+	var err error
 
-				// TODO: try other pods if -0 isn't available?
-				return portforward.DialContext(ctx, mon.env, mon.oc, alertNamespace, alertPod, port)
+	for i := 0; i < 3; i++ {
+		hc := &http.Client{
+			Transport: &http.Transport{
+				DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
+					_, port, err := net.SplitHostPort(address)
+					if err != nil {
+						return nil, err
+					}
+
+					return portforward.DialContext(ctx, mon.env, mon.oc, "openshift-monitoring", fmt.Sprintf("alertmanager-main-%d", i), port)
+				},
 			},
-		},
-	}
+		}
 
-	resp, err := hc.Get(alertServiceEndpoint)
+		var req *http.Request
+		req, err = http.NewRequestWithContext(ctx, http.MethodGet, "http://alertmanager-main.openshift-monitoring.svc:9093/api/v2/alerts", nil)
+		if err != nil {
+			return err
+		}
+
+		resp, err = hc.Do(req)
+		if err == nil {
+			break
+		}
+	}
 	if err != nil {
 		return err
 	}

pkg/monitor/worker.go

@@ -122,6 +122,7 @@ func (mon *monitor) worker(stop <-chan struct{}, delay time.Duration, id string)
 	t := time.NewTicker(time.Minute)
 	defer t.Stop()
 
+out:
 	for {
 		mon.mu.RLock()
 		v := mon.docs[id]
@@ -142,7 +143,7 @@ func (mon *monitor) worker(stop <-chan struct{}, delay time.Duration, id string)
 		select {
 		case <-t.C:
 		case <-stop:
-			break
+			break out
 		}
 	}
 

pkg/util/portforward/portforward.go

@@ -78,11 +78,19 @@ func DialContext(ctx context.Context, env env.Interface, oc *api.OpenShiftCluste
 		return nil, err
 	}
 
-	if resp.StatusCode != http.StatusSwitchingProtocols ||
-		resp.Header.Get(httpstream.HeaderConnection) != httpstream.HeaderUpgrade ||
-		resp.Header.Get(httpstream.HeaderUpgrade) != spdy.HeaderSpdy31 {
+	if resp.StatusCode != http.StatusSwitchingProtocols {
 		tlsConn.Close()
-		return nil, fmt.Errorf("unexpected http response")
+		return nil, fmt.Errorf("unexpected http status code %d", resp.StatusCode)
+	}
+
+	if resp.Header.Get(httpstream.HeaderConnection) != httpstream.HeaderUpgrade {
+		tlsConn.Close()
+		return nil, fmt.Errorf("unexpected http header %s: %s", httpstream.HeaderConnection, resp.Header.Get(httpstream.HeaderConnection))
+	}
+
+	if resp.Header.Get(httpstream.HeaderUpgrade) != spdy.HeaderSpdy31 {
+		tlsConn.Close()
+		return nil, fmt.Errorf("unexpected http header %s: %s", httpstream.HeaderUpgrade, resp.Header.Get(httpstream.HeaderUpgrade))
 	}
 
 	// 5. Negotiate SPDY