Repository: Azure/ARO-RP (go)

Issue:
Monitoring degraded
Looks like prometheus might be degraded in production:
```
github.com/Azure/ARO-RP/pkg/monitor/cluster.(*Monitor).emitPrometheusAlerts-fm: Get "http://alertmanager-main.openshift-monitoring.svc:9093/api/v2/alerts": unexpected http status code 200 
```

Pull request description:
4.5+ prometheus monitoring fix
Fixes #1178 

Adds basic e2e coverage

Patch:
diff --git a/pkg/api/validate/openshiftcluster_validatedynamic.go b/pkg/api/validate/openshiftcluster_validatedynamic.go
index ebd2d19c3..bbd26506f 100644
--- a/pkg/api/validate/openshiftcluster_validatedynamic.go
+++ b/pkg/api/validate/openshiftcluster_validatedynamic.go
@@ -322,10 +322,6 @@ func (dv *openShiftClusterDynamicValidator) validateSubnet(ctx context.Context,
 func (dv *openShiftClusterDynamicValidator) validateVnet(ctx context.Context, vnet *mgmtnetwork.VirtualNetwork) error {
 	dv.log.Print("validateVnet")
 
-	if !strings.EqualFold(*vnet.Location, dv.oc.Location) {
-		return api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodeInvalidLinkedVNet, "", "The vnet location '%s' must match the cluster location '%s'.", *vnet.Location, dv.oc.Location)
-	}
-
 	master, err := dv.validateSubnet(ctx, vnet, "properties.masterProfile.subnetId", dv.oc.Properties.MasterProfile.SubnetID)
 	if err != nil {
 		return err
diff --git a/pkg/monitor/cluster/cluster.go b/pkg/monitor/cluster/cluster.go
index 792499c72..379655894 100644
--- a/pkg/monitor/cluster/cluster.go
+++ b/pkg/monitor/cluster/cluster.go
@@ -17,27 +17,26 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 
 	"github.com/Azure/ARO-RP/pkg/api"
 	"github.com/Azure/ARO-RP/pkg/metrics"
 	aroclient "github.com/Azure/ARO-RP/pkg/operator/clientset/versioned/typed/aro.openshift.io/v1alpha1"
-	"github.com/Azure/ARO-RP/pkg/proxy"
-	"github.com/Azure/ARO-RP/pkg/util/restconfig"
 )
 
 type Monitor struct {
-	dialer    proxy.Dialer
 	log       *logrus.Entry
 	hourlyRun bool
 
 	oc   *api.OpenShiftCluster
 	dims map[string]string
 
-	cli       kubernetes.Interface
-	configcli configclient.Interface
-	mcocli    mcoclient.Interface
-	m         metrics.Interface
-	arocli    aroclient.AroV1alpha1Interface
+	restconfig *rest.Config
+	cli        kubernetes.Interface
+	configcli  configclient.Interface
+	mcocli     mcoclient.Interface
+	m          metrics.Interface
+	arocli     aroclient.AroV1alpha1Interface
 
 	// access below only via the helper functions in cache.go
 	cache struct {
@@ -48,7 +47,7 @@ type Monitor struct {
 	}
 }
 
-func NewMonitor(ctx context.Context, dialer proxy.Dialer, log *logrus.Entry, oc *api.OpenShiftCluster, m metrics.Interface, hourlyRun bool) (*Monitor, error) {
+func NewMonitor(ctx context.Context, log *logrus.Entry, restConfig *rest.Config, oc *api.OpenShiftCluster, m metrics.Interface, hourlyRun bool) (*Monitor, error) {
 	r, err := azure.ParseResourceID(oc.ID)
 	if err != nil {
 		return nil, err
@@ -61,11 +60,6 @@ func NewMonitor(ctx context.Context, dialer proxy.Dialer, log *logrus.Entry, oc
 		"resourceName":   r.ResourceName,
 	}
 
-	restConfig, err := restconfig.RestConfig(dialer, oc)
-	if err != nil {
-		return nil, err
-	}
-
 	cli, err := kubernetes.NewForConfig(restConfig)
 	if err != nil {
 		return nil, err
@@ -87,28 +81,29 @@ func NewMonitor(ctx context.Context, dialer proxy.Dialer, log *logrus.Entry, oc
 	}
 
 	return &Monitor{
-		dialer:    dialer,
 		log:       log,
 		hourlyRun: hourlyRun,
 
 		oc:   oc,
 		dims: dims,
 
-		cli:       cli,
-		configcli: configcli,
-		mcocli:    mcocli,
-		arocli:    arocli,
-		m:         m,
+		restconfig: restConfig,
+		cli:        cli,
+		configcli:  configcli,
+		mcocli:     mcocli,
+		arocli:     arocli,
+		m:          m,
 	}, nil
 }
 
 // Monitor checks the API server health of a cluster
-func (mon *Monitor) Monitor(ctx context.Context) {
+func (mon *Monitor) Monitor(ctx context.Context) (errs []error) {
 	mon.log.Debug("monitoring")
 
 	// If API is not returning 200, don't need to run the next checks
 	statusCode, err := mon.emitAPIServerHealthzCode(ctx)
 	if err != nil {
+		errs = append(errs, err)
 		mon.log.Printf("%s: %s", runtime.FuncForPC(reflect.ValueOf(mon.emitAPIServerHealthzCode).Pointer()).Name(), err)
 		mon.emitGauge("monitor.clustererrors", 1, map[string]string{"monitor": runtime.FuncForPC(reflect.ValueOf(mon.emitAPIServerHealthzCode).Pointer()).Name()})
 	}
@@ -135,11 +130,14 @@ func (mon *Monitor) Monitor(ctx context.Context) {
 	} {
 		err = f(ctx)
 		if err != nil {
+			errs = append(errs, err)
 			mon.log.Printf("%s: %s", runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name(), err)
 			mon.emitGauge("monitor.clustererrors", 1, map[string]string{"monitor": runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name()})
 			// keep going
 		}
 	}
+
+	return
 }
 
 func (mon *Monitor) emitFloat(m string, value float64, dims map[string]string) {
diff --git a/pkg/monitor/cluster/prometheusalerts.go b/pkg/monitor/cluster/prometheusalerts.go
index e35563219..448e32883 100644
--- a/pkg/monitor/cluster/prometheusalerts.go
+++ b/pkg/monitor/cluster/prometheusalerts.go
@@ -30,7 +30,7 @@ func (mon *Monitor) emitPrometheusAlerts(ctx context.Context) error {
 						return nil, err
 					}
 
-					return portforward.DialContext(ctx, mon.log, mon.dialer, mon.oc, "openshift-monitoring", fmt.Sprintf("alertmanager-main-%d", i), port)
+					return portforward.DialContext(ctx, mon.log, mon.restconfig, "openshift-monitoring", fmt.Sprintf("alertmanager-main-%d", i), port)
 				},
 				// HACK: without this, keepalive connections don't get closed,
 				// resulting in excessive open TCP connections, lots of
diff --git a/pkg/monitor/worker.go b/pkg/monitor/worker.go
index eab46c3c2..a7d4042e8 100644
--- a/pkg/monitor/worker.go
+++ b/pkg/monitor/worker.go
@@ -15,6 +15,7 @@ import (
 	"github.com/Azure/ARO-RP/pkg/monitor/cluster"
 	utillog "github.com/Azure/ARO-RP/pkg/util/log"
 	"github.com/Azure/ARO-RP/pkg/util/recover"
+	"github.com/Azure/ARO-RP/pkg/util/restconfig"
 )
 
 // listBuckets reads our bucket allocation from the master
@@ -191,7 +192,13 @@ func (mon *monitor) workOne(ctx context.Context, log *logrus.Entry, doc *api.Ope
 	ctx, cancel := context.WithTimeout(ctx, 50*time.Second)
 	defer cancel()
 
-	c, err := cluster.NewMonitor(ctx, mon.dialer, log, doc.OpenShiftCluster, mon.clusterm, hourlyRun)
+	restConfig, err := restconfig.RestConfig(mon.dialer, doc.OpenShiftCluster)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+
+	c, err := cluster.NewMonitor(ctx, log, restConfig, doc.OpenShiftCluster, mon.clusterm, hourlyRun)
 	if err != nil {
 		log.Error(err)
 		return
diff --git a/pkg/util/portforward/dialer.go b/pkg/util/portforward/dialer.go
index 302f13a53..23589991f 100644
--- a/pkg/util/portforward/dialer.go
+++ b/pkg/util/portforward/dialer.go
@@ -8,26 +8,20 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"net"
 	"net/http"
 	"net/url"
+	"time"
 
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
 	"k8s.io/client-go/rest"
-
-	"github.com/Azure/ARO-RP/pkg/api"
-	"github.com/Azure/ARO-RP/pkg/proxy"
-	"github.com/Azure/ARO-RP/pkg/util/restconfig"
+	"k8s.io/client-go/tools/portforward"
 )
 
 // dialSpdy connects to the specified path on the API server of oc and
 // negotiates SPDY
-func dialSpdy(ctx context.Context, dialer proxy.Dialer, oc *api.OpenShiftCluster, path string) (httpstream.Connection, error) {
-	restconfig, err := restconfig.RestConfig(dialer, oc)
-	if err != nil {
-		return nil, err
-	}
-
+func dialSpdy(ctx context.Context, restconfig *rest.Config, path string) (httpstream.Connection, error) {
 	// 1. Connect to the API server via private endpoint (and via proxy in
 	//    development mode)
 	clusterURL, err := url.Parse(restconfig.Host)
@@ -35,7 +29,15 @@ func dialSpdy(ctx context.Context, dialer proxy.Dialer, oc *api.OpenShiftCluster
 		return nil, err
 	}
 
-	rawConn, err := dialer.DialContext(ctx, "tcp", oc.Properties.NetworkProfile.PrivateEndpointIP+":"+clusterURL.Port())
+	dialContext := restconfig.Dial
+	if dialContext == nil {
+		dialContext = (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext
+	}
+
+	rawConn, err := dialContext(ctx, "tcp", clusterURL.Host)
 	if err != nil {
 		return nil, err
 	}
@@ -46,6 +48,9 @@ func dialSpdy(ctx context.Context, dialer proxy.Dialer, oc *api.OpenShiftCluster
 		rawConn.Close()
 		return nil, err
 	}
+	if tlsConfig == nil {
+		tlsConfig = &tls.Config{}
+	}
 	tlsConfig.ServerName = clusterURL.Hostname()
 
 	tlsConn := tls.Client(rawConn, tlsConfig)
@@ -62,7 +67,11 @@ func dialSpdy(ctx context.Context, dialer proxy.Dialer, oc *api.OpenShiftCluster
 		return nil, err
 	}
 	req.Header.Add(httpstream.HeaderConnection, httpstream.HeaderUpgrade)
+	req.Header.Add(httpstream.HeaderProtocolVersion, portforward.PortForwardProtocolV1Name)
 	req.Header.Add(httpstream.HeaderUpgrade, spdy.HeaderSpdy31)
+	if restconfig.BearerToken != "" {
+		req.Header.Add("Authorization", "Bearer "+restconfig.BearerToken)
+	}
 
 	err = req.Write(tlsConn)
 	if err != nil {
diff --git a/pkg/util/portforward/exec.go b/pkg/util/portforward/exec.go
deleted file mode 100644
index 647be9276..000000000
--- a/pkg/util/portforward/exec.go
+++ /dev/null
@@ -1,54 +0,0 @@
-package portforward
-
-// Copyright (c) Microsoft Corporation.
-// Licensed under the Apache License 2.0.
-
-import (
-	"context"
-	"io"
-	"net/http"
-	"net/url"
-
-	"github.com/sirupsen/logrus"
-	v1 "k8s.io/api/core/v1"
-
-	"github.com/Azure/ARO-RP/pkg/api"
-	"github.com/Azure/ARO-RP/pkg/proxy"
-)
-
-// ExecStdout executes a command in the given namespace/pod/container and
-// streams its stdout.
-func ExecStdout(ctx context.Context, log *logrus.Entry, dialer proxy.Dialer, oc *api.OpenShiftCluster, namespace, pod, container string, command []string) (io.ReadCloser, error) {
-	v := url.Values{
-		"container": []string{container},
-		"command":   command,
-		"stdout":    []string{"true"},
-	}
-
-	spdyConn, err := dialSpdy(ctx, dialer, oc, "/api/v1/namespaces/"+namespace+"/pods/"+pod+"/exec?"+v.Encode())
-	if err != nil {
-		return nil, err
-	}
-
-	// Connect the error stream, r/o
-	errorStream, err := spdyConn.CreateStream(http.Header{
-		v1.StreamType: []string{v1.StreamTypeError},
-	})
-	if err != nil {
-		spdyConn.Close()
-		return nil, err
-	}
-	errorStream.Close() // this actually means CloseWrite()
-
-	// Connect the stdout stream, r/o
-	stdoutStream, err := spdyConn.CreateStream(http.Header{
-		v1.StreamType: []string{v1.StreamTypeStdout},
-	})
-	if err != nil {
-		spdyConn.Close()
-		return nil, err
-	}
-	stdoutStream.Close() // this actually means CloseWrite()
-
-	return newStreamConn(log, spdyConn, stdoutStream, errorStream), nil
-}
diff --git a/pkg/util/portforward/portforward.go b/pkg/util/portforward/portforward.go
index 4cdb5a33a..553560b67 100644
--- a/pkg/util/portforward/portforward.go
+++ b/pkg/util/portforward/portforward.go
@@ -10,14 +10,12 @@ import (
 
 	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
-
-	"github.com/Azure/ARO-RP/pkg/api"
-	"github.com/Azure/ARO-RP/pkg/proxy"
+	"k8s.io/client-go/rest"
 )
 
 // DialContext returns a connection to the specified cluster/namespace/pod/port.
-func DialContext(ctx context.Context, log *logrus.Entry, dialer proxy.Dialer, oc *api.OpenShiftCluster, namespace, pod, port string) (net.Conn, error) {
-	spdyConn, err := dialSpdy(ctx, dialer, oc, "/api/v1/namespaces/"+namespace+"/pods/"+pod+"/portforward")
+func DialContext(ctx context.Context, log *logrus.Entry, restconfig *rest.Config, namespace, pod, port string) (net.Conn, error) {
+	spdyConn, err := dialSpdy(ctx, restconfig, "/api/v1/namespaces/"+namespace+"/pods/"+pod+"/portforward")
 	if err != nil {
 		return nil, err
 	}
diff --git a/vendor/k8s.io/client-go/tools/portforward/doc.go b/vendor/k8s.io/client-go/tools/portforward/doc.go
new file mode 100644
index 000000000..2f5340634
--- /dev/null
+++ b/vendor/k8s.io/client-go/tools/portforward/doc.go
@@ -0,0 +1,19 @@
+/*
+Copyright 2015 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package portforward adds support for SSH-like port forwarding from the client's
+// local host to remote containers.
+package portforward // import "k8s.io/client-go/tools/portforward"
diff --git a/vendor/k8s.io/client-go/tools/portforward/portforward.go b/vendor/k8s.io/client-go/tools/portforward/portforward.go
new file mode 100644
index 000000000..ffc0bcac7
--- /dev/null
+++ b/vendor/k8s.io/client-go/tools/portforward/portforward.go
@@ -0,0 +1,429 @@
+/*
+Copyright 2015 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package portforward
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/apimachinery/pkg/util/runtime"
+)
+
+// PortForwardProtocolV1Name is the subprotocol used for port forwarding.
+// TODO move to API machinery and re-unify with kubelet/server/portfoward
+const PortForwardProtocolV1Name = "portforward.k8s.io"
+
+// PortForwarder knows how to listen for local connections and forward them to
+// a remote pod via an upgraded HTTP request.
+type PortForwarder struct {
+	addresses []listenAddress
+	ports     []ForwardedPort
+	stopChan  <-chan struct{}
+
+	dialer        httpstream.Dialer
+	streamConn    httpstream.Connection
+	listeners     []io.Closer
+	Ready         chan struct{}
+	requestIDLock sync.Mutex
+	requestID     int
+	out           io.Writer
+	errOut        io.Writer
+}
+
+// ForwardedPort contains a Local:Remote port pairing.
+type ForwardedPort struct {
+	Local  uint16
+	Remote uint16
+}
+
+/*
+	valid port specifications:
+
+	5000
+	- forwards from localhost:5000 to pod:5000
+
+	8888:5000
+	- forwards from localhost:8888 to pod:5000
+
+	0:5000
+	:5000
+	- selects a random available local port,
+	  forwards from localhost:<random port> to pod:5000
+*/
+func parsePorts(ports []string) ([]ForwardedPort, error) {
+	var forwards []ForwardedPort
+	for _, portString := range ports {
+		parts := strings.Split(portString, ":")
+		var localString, remoteString string
+		if len(parts) == 1 {
+			localString = parts[0]
+			remoteString = parts[0]
+		} else if len(parts) == 2 {
+			localString = parts[0]
+			if localString == "" {
+				// support :5000
+				localString = "0"
+			}
+			remoteString = parts[1]
+		} else {
+			return nil, fmt.Errorf("invalid port format '%s'", portString)
+		}
+
+		localPort, err := strconv.ParseUint(localString, 10, 16)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing local port '%s': %s", localString, err)
+		}
+
+		remotePort, err := strconv.ParseUint(remoteString, 10, 16)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing remote port '%s': %s", remoteString, err)
+		}
+		if remotePort == 0 {
+			return nil, fmt.Errorf("remote port must be > 0")
+		}
+
+		forwards = append(forwards, ForwardedPort{uint16(localPort), uint16(remotePort)})
+	}
+
+	return forwards, nil
+}
+
+type listenAddress struct {
+	address     string
+	protocol    string
+	failureMode string
+}
+
+func parseAddresses(addressesToParse []string) ([]listenAddress, error) {
+	var addresses []listenAddress
+	parsed := make(map[string]listenAddress)
+	for _, address := range addressesToParse {
+		if address == "localhost" {
+			if _, exists := parsed["127.0.0.1"]; !exists {
+				ip := listenAddress{address: "127.0.0.1", protocol: "tcp4", failureMode: "all"}
+				parsed[ip.address] = ip
+			}
+			if _, exists := parsed["::1"]; !exists {
+				ip := listenAddress{address: "::1", protocol: "tcp6", failureMode: "all"}
+				parsed[ip.address] = ip
+			}
+		} else if net.ParseIP(address).To4() != nil {
+			parsed[address] = listenAddress{address: address, protocol: "tcp4", failureMode: "any"}
+		} else if net.ParseIP(address) != nil {
+			parsed[address] = listenAddress{address: address, protocol: "tcp6", failureMode: "any"}
+		} else {
+			return nil, fmt.Errorf("%s is not a valid IP", address)
+		}
+	}
+	addresses = make([]listenAddress, len(parsed))
+	id := 0
+	for _, v := range parsed {
+		addresses[id] = v
+		id++
+	}
+	// Sort addresses before returning to get a stable order
+	sort.Slice(addresses, func(i, j int) bool { return addresses[i].address < addresses[j].address })
+
+	return addresses, nil
+}
+
+// New creates a new PortForwarder with localhost listen addresses.
+func New(dialer httpstream.Dialer, ports []string, stopChan <-chan struct{}, readyChan chan struct{}, out, errOut io.Writer) (*PortForwarder, error) {
+	return NewOnAddresses(dialer, []string{"localhost"}, ports, stopChan, readyChan, out, errOut)
+}
+
+// NewOnAddresses creates a new PortForwarder with custom listen addresses.
+func NewOnAddresses(dialer httpstream.Dialer, addresses []string, ports []string, stopChan <-chan struct{}, readyChan chan struct{}, out, errOut io.Writer) (*PortForwarder, error) {
+	if len(addresses) == 0 {
+		return nil, errors.New("you must specify at least 1 address")
+	}
+	parsedAddresses, err := parseAddresses(addresses)
+	if err != nil {
+		return nil, err
+	}
+	if len(ports) == 0 {
+		return nil, errors.New("you must specify at least 1 port")
+	}
+	parsedPorts, err := parsePorts(ports)
+	if err != nil {
+		return nil, err
+	}
+	return &PortForwarder{
+		dialer:    dialer,
+		addresses: parsedAddresses,
+		ports:     parsedPorts,
+		stopChan:  stopChan,
+		Ready:     readyChan,
+		out:       out,
+		errOut:    errOut,
+	}, nil
+}
+
+// ForwardPorts formats and executes a port forwarding request. The connection will remain
+// open until stopChan is closed.
+func (pf *PortForwarder) ForwardPorts() error {
+	defer pf.Close()
+
+	var err error
+	pf.streamConn, _, err = pf.dialer.Dial(PortForwardProtocolV1Name)
+	if err != nil {
+		return fmt.Errorf("error upgrading connection: %s", err)
+	}
+	defer pf.streamConn.Close()
+
+	return pf.forward()
+}
+
+// forward dials the remote host specific in req, upgrades the request, starts
+// listeners for each port specified in ports, and forwards local connections
+// to the remote host via streams.
+func (pf *PortForwarder) forward() error {
+	var err error
+
+	listenSuccess := false
+	for i := range pf.ports {
+		port := &pf.ports[i]
+		err = pf.listenOnPort(port)
+		switch {
+		case err == nil:
+			listenSuccess = true
+		default:
+			if pf.errOut != nil {
+				fmt.Fprintf(pf.errOut, "Unable to listen on port %d: %v\n", port.Local, err)
+			}
+		}
+	}
+
+	if !listenSuccess {
+		return fmt.Errorf("unable to listen on any of the requested ports: %v", pf.ports)
+	}
+
+	if pf.Ready != nil {
+		close(pf.Ready)
+	}
+
+	// wait for interrupt or conn closure
+	select {
+	case <-pf.stopChan:
+	case <-pf.streamConn.CloseChan():
+		runtime.HandleError(errors.New("lost connection to pod"))
+	}
+
+	return nil
+}
+
+// listenOnPort delegates listener creation and waits for connections on requested bind addresses.
+// An error is raised based on address groups (default and localhost) and their failure modes
+func (pf *PortForwarder) listenOnPort(port *ForwardedPort) error {
+	var errors []error
+	failCounters := make(map[string]int, 2)
+	successCounters := make(map[string]int, 2)
+	for _, addr := range pf.addresses {
+		err := pf.listenOnPortAndAddress(port, addr.protocol, addr.address)
+		if err != nil {
+			errors = append(errors, err)
+			failCounters[addr.failureMode]++
+		} else {
+			successCounters[addr.failureMode]++
+		}
+	}
+	if successCounters["all"] == 0 && failCounters["all"] > 0 {
+		return fmt.Errorf("%s: %v", "Listeners failed to create with the following errors", errors)
+	}
+	if failCounters["any"] > 0 {
+		return fmt.Errorf("%s: %v", "Listeners failed to create with the following errors", errors)
+	}
+	return nil
+}
+
+// listenOnPortAndAddress delegates listener creation and waits for new connections
+// in the background f
+func (pf *PortForwarder) listenOnPortAndAddress(port *ForwardedPort, protocol string, address string) error {
+	listener, err := pf.getListener(protocol, address, port)
+	if err != nil {
+		return err
+	}
+	pf.listeners = append(pf.listeners, listener)
+	go pf.waitForConnection(listener, *port)
+	return nil
+}
+
+// getListener creates a listener on the interface targeted by the given hostname on the given port with
+// the given protocol. protocol is in net.Listen style which basically admits values like tcp, tcp4, tcp6
+func (pf *PortForwarder) getListener(protocol string, hostname string, port *ForwardedPort) (net.Listener, error) {
+	listener, err := net.Listen(protocol, net.JoinHostPort(hostname, strconv.Itoa(int(port.Local))))
+	if err != nil {
+		return nil, fmt.Errorf("unable to create listener: Error %s", err)
+	}
+	listenerAddress := listener.Addr().String()
+	host, localPort, _ := net.SplitHostPort(listenerAddress)
+	localPortUInt, err := strconv.ParseUint(localPort, 10, 16)
+
+	if err != nil {
+		fmt.Fprintf(pf.out, "Failed to forward from %s:%d -> %d\n", hostname, localPortUInt, port.Remote)
+		return nil, fmt.Errorf("error parsing local port: %s from %s (%s)", err, listenerAddress, host)
+	}
+	port.Local = uint16(localPortUInt)
+	if pf.out != nil {
+		fmt.Fprintf(pf.out, "Forwarding from %s -> %d\n", net.JoinHostPort(hostname, strconv.Itoa(int(localPortUInt))), port.Remote)
+	}
+
+	return listener, nil
+}
+
+// waitForConnection waits for new connections to listener and handles them in
+// the background.
+func (pf *PortForwarder) waitForConnection(listener net.Listener, port ForwardedPort) {
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			// TODO consider using something like https://github.com/hydrogen18/stoppableListener?
+			if !strings.Contains(strings.ToLower(err.Error()), "use of closed network connection") {
+				runtime.HandleError(fmt.Errorf("error accepting connection on port %d: %v", port.Local, err))
+			}
+			return
+		}
+		go pf.handleConnection(conn, port)
+	}
+}
+
+func (pf *PortForwarder) nextRequestID() int {
+	pf.requestIDLock.Lock()
+	defer pf.requestIDLock.Unlock()
+	id := pf.requestID
+	pf.requestID++
+	return id
+}
+
+// handleConnection copies data between the local connection and the stream to
+// the remote server.
+func (pf *PortForwarder) handleConnection(conn net.Conn, port ForwardedPort) {
+	defer conn.Close()
+
+	if pf.out != nil {
+		fmt.Fprintf(pf.out, "Handling connection for %d\n", port.Local)
+	}
+
+	requestID := pf.nextRequestID()
+
+	// create error stream
+	headers := http.Header{}
+	headers.Set(v1.StreamType, v1.StreamTypeError)
+	headers.Set(v1.PortHeader, fmt.Sprintf("%d", port.Remote))
+	headers.Set(v1.PortForwardRequestIDHeader, strconv.Itoa(requestID))
+	errorStream, err := pf.streamConn.CreateStream(headers)
+	if err != nil {
+		runtime.HandleError(fmt.Errorf("error creating error stream for port %d -> %d: %v", port.Local, port.Remote, err))
+		return
+	}
+	// we're not writing to this stream
+	errorStream.Close()
+
+	errorChan := make(chan error)
+	go func() {
+		message, err := ioutil.ReadAll(errorStream)
+		switch {
+		case err != nil:
+			errorChan <- fmt.Errorf("error reading from error stream for port %d -> %d: %v", port.Local, port.Remote, err)
+		case len(message) > 0:
+			errorChan <- fmt.Errorf("an error occurred forwarding %d -> %d: %v", port.Local, port.Remote, string(message))
+		}
+		close(errorChan)
+	}()
+
+	// create data stream
+	headers.Set(v1.StreamType, v1.StreamTypeData)
+	dataStream, err := pf.streamConn.CreateStream(headers)
+	if err != nil {
+		runtime.HandleError(fmt.Errorf("error creating forwarding stream for port %d -> %d: %v", port.Local, port.Remote, err))
+		return
+	}
+
+	localError := make(chan struct{})
+	remoteDone := make(chan struct{})
+
+	go func() {
+		// Copy from the remote side to the local port.
+		if _, err := io.Copy(conn, dataStream); err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			runtime.HandleError(fmt.Errorf("error copying from remote stream to local connection: %v", err))
+		}
+
+		// inform the select below that the remote copy is done
+		close(remoteDone)
+	}()
+
+	go func() {
+		// inform server we're not sending any more data after copy unblocks
+		defer dataStream.Close()
+
+		// Copy from the local port to the remote side.
+		if _, err := io.Copy(dataStream, conn); err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			runtime.HandleError(fmt.Errorf("error copying from local connection to remote stream: %v", err))
+			// break out of the select below without waiting for the other copy to finish
+			close(localError)
+		}
+	}()
+
+	// wait for either a local->remote error or for copying from remote->local to finish
+	select {
+	case <-remoteDone:
+	case <-localError:
+	}
+
+	// always expect something on errorChan (it may be nil)
+	err = <-errorChan
+	if err != nil {
+		runtime.HandleError(err)
+	}
+}
+
+// Close stops all listeners of PortForwarder.
+func (pf *PortForwarder) Close() {
+	// stop all listeners
+	for _, l := range pf.listeners {
+		if err := l.Close(); err != nil {
+			runtime.HandleError(fmt.Errorf("error closing listener: %v", err))
+		}
+	}
+}
+
+// GetPorts will return the ports that were forwarded; this can be used to
+// retrieve the locally-bound port in cases where the input was port 0. This
+// function will signal an error if the Ready channel is nil or if the
+// listeners are not ready yet; this function will succeed after the Ready
+// channel has been closed.
+func (pf *PortForwarder) GetPorts() ([]ForwardedPort, error) {
+	if pf.Ready == nil {
+		return nil, fmt.Errorf("no Ready channel provided")
+	}
+	select {
+	case <-pf.Ready:
+		return pf.ports, nil
+	default:
+		return nil, fmt.Errorf("listeners not ready")
+	}
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 3eb31cede..ceff1573c 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1196,6 +1196,7 @@ k8s.io/client-go/tools/leaderelection
 k8s.io/client-go/tools/leaderelection/resourcelock
 k8s.io/client-go/tools/metrics
 k8s.io/client-go/tools/pager
+k8s.io/client-go/tools/portforward
 k8s.io/client-go/tools/record
 k8s.io/client-go/tools/record/util
 k8s.io/client-go/tools/reference


Test patch:
diff --git a/pkg/api/validate/openshiftcluster_validatedynamic_test.go b/pkg/api/validate/openshiftcluster_validatedynamic_test.go
index ed68f952e..98d2449d3 100644
--- a/pkg/api/validate/openshiftcluster_validatedynamic_test.go
+++ b/pkg/api/validate/openshiftcluster_validatedynamic_test.go
@@ -191,13 +191,6 @@ func TestValidateVnet(t *testing.T) {
 				oc.Properties.ProvisioningState = api.ProvisioningStateCreating
 			},
 		},
-		{
-			name: "missing subnet (master)",
-			modifyVnet: func(vnet *mgmtnetwork.VirtualNetwork) {
-				vnet.Location = to.StringPtr("westeurope")
-			},
-			wantErr: "400: InvalidLinkedVNet: : The vnet location 'westeurope' must match the cluster location 'eastus'.",
-		},
 		{
 			name: "missing subnet (master)",
 			modifyVnet: func(vnet *mgmtnetwork.VirtualNetwork) {
@@ -350,7 +343,6 @@ func TestValidateVnet(t *testing.T) {
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			oc := &api.OpenShiftCluster{
-				Location: "eastus",
 				Properties: api.OpenShiftClusterProperties{
 					ClusterProfile: api.ClusterProfile{
 						ResourceGroupID: resourceGroupID,
@@ -371,8 +363,7 @@ func TestValidateVnet(t *testing.T) {
 			}
 
 			vnet := &mgmtnetwork.VirtualNetwork{
-				Location: to.StringPtr("eastus"),
-				ID:       &vnetID,
+				ID: &vnetID,
 				VirtualNetworkPropertiesFormat: &mgmtnetwork.VirtualNetworkPropertiesFormat{
 					DhcpOptions: &mgmtnetwork.DhcpOptions{
 						DNSServers: &[]string{},
diff --git a/test/e2e/monitor.go b/test/e2e/monitor.go
new file mode 100644
index 000000000..8630f4c71
--- /dev/null
+++ b/test/e2e/monitor.go
@@ -0,0 +1,29 @@
+package e2e
+
+// Copyright (c) Microsoft Corporation.
+// Licensed under the Apache License 2.0.
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/Azure/ARO-RP/pkg/api"
+	"github.com/Azure/ARO-RP/pkg/metrics/noop"
+	"github.com/Azure/ARO-RP/pkg/monitor/cluster"
+)
+
+var _ = Describe("Monitor", func() {
+	Specify("a monitor run should not return any errors", func() {
+		ctx := context.Background()
+
+		mon, err := cluster.NewMonitor(ctx, log, clients.RestConfig, &api.OpenShiftCluster{
+			ID: resourceIDFromEnv(),
+		}, &noop.Noop{}, true)
+		Expect(err).NotTo(HaveOccurred())
+
+		errs := mon.Monitor(ctx)
+		Expect(errs).To(HaveLen(0))
+	})
+})
diff --git a/test/e2e/setup.go b/test/e2e/setup.go
index da1e9b6b0..173424379 100644
--- a/test/e2e/setup.go
+++ b/test/e2e/setup.go
@@ -17,6 +17,7 @@ import (
 	machineapiclient "github.com/openshift/machine-api-operator/pkg/generated/clientset/versioned"
 	"github.com/sirupsen/logrus"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/client-go/tools/clientcmd/api/latest"
@@ -41,6 +42,7 @@ type clientSet struct {
 	ActivityLogs      insights.ActivityLogsClient
 	VirtualNetworks   network.VirtualNetworksClient
 
+	RestConfig  *rest.Config
 	Kubernetes  kubernetes.Interface
 	MachineAPI  machineapiclient.Interface
 	AROClusters aroclient.AroV1alpha1Interface
@@ -119,6 +121,7 @@ func newClientSet(ctx context.Context) (*clientSet, error) {
 		ActivityLogs:      insights.NewActivityLogsClient(im.SubscriptionID(), authorizer),
 		VirtualNetworks:   network.NewVirtualNetworksClient(im.SubscriptionID(), authorizer),
 
+		RestConfig:  restconfig,
 		Kubernetes:  cli,
 		MachineAPI:  machineapicli,
 		AROClusters: arocli,