Add http transport for cincinnati to enable proxy

jcpowermac · jcpowermac · commit ea5e3bc86ea5 · 2019-07-16T08:29:01.000-04:00
- Modified Cincinnati client to accept a url.URL
so that a https proxy can be set
- Created new function getHTTPSProxyURL() that
retrieves the proxy config, creates the url.URL and returns the
url ptr.
- Modified tests to include proxy lister
- Added ProxyLister to Operator struct, modified New()
- Modified start.go for ProxyLister operator struct change
diff --git a/pkg/cincinnati/cincinnati.go b/pkg/cincinnati/cincinnati.go
@@ -20,12 +20,13 @@ const (
 // Client is a Cincinnati client which can be used to fetch update graphs from
 // an upstream Cincinnati stack.
 type Client struct {
-	id uuid.UUID
+	id       uuid.UUID
+	proxyURL *url.URL
 }
 
 // NewClient creates a new Cincinnati client with the given client identifier.
-func NewClient(id uuid.UUID) Client {
-	return Client{id: id}
+func NewClient(id uuid.UUID, proxyURL *url.URL) Client {
+	return Client{id: id, proxyURL: proxyURL}
 }
 
 // Update is a single node from the update graph.
@@ -39,6 +40,7 @@ type Update node
 // the current version and their payloads indicate from where the actual update
 // image can be downloaded.
 func (c Client) GetUpdates(upstream string, channel string, version semver.Version) ([]Update, error) {
+	transport := http.Transport{}
 	// Prepare parametrized cincinnati query.
 	cincinnatiURL, err := url.Parse(upstream)
 	if err != nil {
@@ -57,7 +59,11 @@ func (c Client) GetUpdates(upstream string, channel string, version semver.Versi
 	}
 	req.Header.Add("Accept", GraphMediaType)
 
-	client := http.Client{}
+	if c.proxyURL != nil {
+		transport.Proxy = http.ProxyURL(c.proxyURL)
+	}
+
+	client := http.Client{Transport: &transport}
 	resp, err := client.Do(req)
 	if err != nil {
 		return nil, err
diff --git a/pkg/cincinnati/cincinnati_test.go b/pkg/cincinnati/cincinnati_test.go
@@ -10,8 +10,8 @@ import (
 	"testing"
 
 	"github.com/blang/semver"
-	_ "k8s.io/klog" // integration tests set glog flags.
 	"github.com/google/uuid"
+	_ "k8s.io/klog" // integration tests set glog flags.
 )
 
 func TestGetUpdates(t *testing.T) {
@@ -120,8 +120,9 @@ func TestGetUpdates(t *testing.T) {
 
 			ts := httptest.NewServer(http.HandlerFunc(handler))
 			defer ts.Close()
+			var proxyURL *url.URL
 
-			c := NewClient(clientID)
+			c := NewClient(clientID, proxyURL)
 
 			updates, err := c.GetUpdates(ts.URL, channelName, semver.MustParse(test.version))
 			if test.err == "" {
diff --git a/pkg/cvo/availableupdates.go b/pkg/cvo/availableupdates.go
@@ -4,9 +4,12 @@ import (
 	"fmt"
 	"time"
 
+	"net/url"
+
 	"github.com/blang/semver"
-	"k8s.io/klog"
 	"github.com/google/uuid"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/klog"
 
 	"k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -36,7 +39,12 @@ func (optr *Operator) syncAvailableUpdates(config *configv1.ClusterVersion) erro
 		return nil
 	}
 
-	updates, condition := calculateAvailableUpdatesStatus(string(config.Spec.ClusterID), upstream, channel, optr.releaseVersion)
+	proxyURL, err := optr.getHTTPSProxyURL()
+	if err != nil {
+		return err
+	}
+
+	updates, condition := calculateAvailableUpdatesStatus(string(config.Spec.ClusterID), proxyURL, upstream, channel, optr.releaseVersion)
 
 	if usedDefaultUpstream {
 		upstream = ""
@@ -103,7 +111,7 @@ func (optr *Operator) getAvailableUpdates() *availableUpdates {
 	return optr.availableUpdates
 }
 
-func calculateAvailableUpdatesStatus(clusterID, upstream, channel, version string) ([]configv1.Update, configv1.ClusterOperatorStatusCondition) {
+func calculateAvailableUpdatesStatus(clusterID string, proxyURL *url.URL, upstream, channel, version string) ([]configv1.Update, configv1.ClusterOperatorStatusCondition) {
 	if len(upstream) == 0 {
 		return nil, configv1.ClusterOperatorStatusCondition{
 			Type: configv1.RetrievedUpdates, Status: configv1.ConditionFalse, Reason: "NoUpstream",
@@ -134,7 +142,7 @@ func calculateAvailableUpdatesStatus(clusterID, upstream, channel, version strin
 		}
 	}
 
-	updates, err := checkForUpdate(clusterID, upstream, channel, currentVersion)
+	updates, err := checkForUpdate(clusterID, proxyURL, upstream, channel, currentVersion)
 	if err != nil {
 		klog.V(2).Infof("Upstream server %s could not return available updates: %v", upstream, err)
 		return nil, configv1.ClusterOperatorStatusCondition{
@@ -159,13 +167,38 @@ func calculateAvailableUpdatesStatus(clusterID, upstream, channel, version strin
 	}
 }
 
-func checkForUpdate(clusterID, upstream, channel string, currentVersion semver.Version) ([]cincinnati.Update, error) {
+func checkForUpdate(clusterID string, proxyURL *url.URL, upstream, channel string, currentVersion semver.Version) ([]cincinnati.Update, error) {
 	uuid, err := uuid.Parse(string(clusterID))
 	if err != nil {
 		return nil, err
 	}
+
 	if len(upstream) == 0 {
 		return nil, fmt.Errorf("no upstream URL set for cluster version")
 	}
-	return cincinnati.NewClient(uuid).GetUpdates(upstream, channel, currentVersion)
+	return cincinnati.NewClient(uuid, proxyURL).GetUpdates(upstream, channel, currentVersion)
+}
+
+// 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, 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, nil
+		}
+	}
+	return nil, nil
 }
diff --git a/pkg/cvo/cvo.go b/pkg/cvo/cvo.go
@@ -10,8 +10,8 @@ import (
 	"github.com/openshift/cluster-version-operator/pkg/verify"
 
 	"github.com/blang/semver"
-	"k8s.io/klog"
 	"github.com/google/uuid"
+	"k8s.io/klog"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -101,6 +101,7 @@ type Operator struct {
 
 	cvLister    configlistersv1.ClusterVersionLister
 	coLister    configlistersv1.ClusterOperatorLister
+	proxyLister configlistersv1.ProxyLister
 	cacheSynced []cache.InformerSynced
 
 	// queue tracks applying updates to a cluster.
@@ -135,6 +136,7 @@ func New(
 	minimumInterval time.Duration,
 	cvInformer configinformersv1.ClusterVersionInformer,
 	coInformer configinformersv1.ClusterOperatorInformer,
+	proxyInformer configinformersv1.ProxyInformer,
 	client clientset.Interface,
 	kubeClient kubernetes.Interface,
 	enableMetrics bool,
@@ -162,6 +164,9 @@ func New(
 		availableUpdatesQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "availableupdates"),
 	}
 
+	optr.proxyLister = proxyInformer.Lister()
+	proxyInformer.Informer().AddEventHandler(optr.eventHandler())
+
 	cvInformer.Informer().AddEventHandler(optr.eventHandler())
 
 	optr.coLister = coInformer.Lister()
diff --git a/pkg/cvo/cvo_test.go b/pkg/cvo/cvo_test.go
@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/davecgh/go-spew/spew"
-	"k8s.io/klog"
 	"github.com/google/uuid"
 	apiextv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	apiextclientv1 "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1beta1"
@@ -27,6 +26,7 @@ import (
 	"k8s.io/client-go/rest"
 	ktesting "k8s.io/client-go/testing"
 	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog"
 
 	configv1 "github.com/openshift/api/config/v1"
 	clientset "github.com/openshift/client-go/config/clientset/versioned"
@@ -42,15 +42,35 @@ var (
 	defaultCompletionTime = metav1.Time{Time: time.Unix(2, 0)}
 )
 
+type clientProxyLister struct {
+	client clientset.Interface
+}
+
+func (c *clientProxyLister) Get(name string) (*configv1.Proxy, error) {
+	return c.client.ConfigV1().Proxies().Get(name, metav1.GetOptions{})
+}
+
+func (c *clientProxyLister) List(selector labels.Selector) (ret []*configv1.Proxy, err error) {
+	list, err := c.client.ConfigV1().Proxies().List(metav1.ListOptions{LabelSelector: selector.String()})
+	if err != nil {
+		return nil, err
+	}
+	var items []*configv1.Proxy
+	for i := range list.Items {
+		items = append(items, &list.Items[i])
+	}
+	return items, nil
+}
+
 type clientCVLister struct {
 	client clientset.Interface
 }
 
 func (c *clientCVLister) Get(name string) (*configv1.ClusterVersion, error) {
-	return c.client.Config().ClusterVersions().Get(name, metav1.GetOptions{})
+	return c.client.ConfigV1().ClusterVersions().Get(name, metav1.GetOptions{})
 }
 func (c *clientCVLister) List(selector labels.Selector) (ret []*configv1.ClusterVersion, err error) {
-	list, err := c.client.Config().ClusterVersions().List(metav1.ListOptions{LabelSelector: selector.String()})
+	list, err := c.client.ConfigV1().ClusterVersions().List(metav1.ListOptions{LabelSelector: selector.String()})
 	if err != nil {
 		return nil, err
 	}
@@ -61,15 +81,32 @@ func (c *clientCVLister) List(selector labels.Selector) (ret []*configv1.Cluster
 	return items, nil
 }
 
+type proxyLister struct {
+	Err   error
+	Items []*configv1.Proxy
+}
+
+func (r *proxyLister) List(selector labels.Selector) (ret []*configv1.Proxy, err error) {
+	return r.Items, r.Err
+}
+func (r *proxyLister) Get(name string) (*configv1.Proxy, error) {
+	for _, s := range r.Items {
+		if s.Name == name {
+			return s, nil
+		}
+	}
+	return nil, errors.NewNotFound(schema.GroupResource{}, name)
+}
+
 type clientCOLister struct {
 	client clientset.Interface
 }
 
 func (c *clientCOLister) Get(name string) (*configv1.ClusterOperator, error) {
-	return c.client.Config().ClusterOperators().Get(name, metav1.GetOptions{})
+	return c.client.ConfigV1().ClusterOperators().Get(name, metav1.GetOptions{})
 }
 func (c *clientCOLister) List(selector labels.Selector) (ret []*configv1.ClusterOperator, err error) {
-	list, err := c.client.Config().ClusterOperators().List(metav1.ListOptions{LabelSelector: selector.String()})
+	list, err := c.client.ConfigV1().ClusterOperators().List(metav1.ListOptions{LabelSelector: selector.String()})
 	if err != nil {
 		return nil, err
 	}
@@ -1959,6 +1996,7 @@ func TestOperator_sync(t *testing.T) {
 			if tt.init != nil {
 				tt.init(optr)
 			}
+			optr.proxyLister = &clientProxyLister{client: optr.client}
 			optr.cvLister = &clientCVLister{client: optr.client}
 			optr.coLister = &clientCOLister{client: optr.client}
 			if optr.configSync == nil {
@@ -2332,6 +2370,7 @@ func TestOperator_availableUpdatesSync(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			optr := tt.optr
 			optr.queue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+			optr.proxyLister = &clientProxyLister{client: optr.client}
 			optr.coLister = &clientCOLister{client: optr.client}
 			optr.cvLister = &clientCVLister{client: optr.client}
 
diff --git a/pkg/start/start.go b/pkg/start/start.go
@@ -13,9 +13,9 @@ import (
 	"syscall"
 	"time"
 
-	"k8s.io/klog"
 	"github.com/google/uuid"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"k8s.io/klog"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -325,6 +325,7 @@ func (o *Options) NewControllerContext(cb *ClientBuilder) *Context {
 			resyncPeriod(o.ResyncInterval)(),
 			cvInformer.Config().V1().ClusterVersions(),
 			sharedInformers.Config().V1().ClusterOperators(),
+			sharedInformers.Config().V1().Proxies(),
 			cb.ClientOrDie(o.Namespace),
 			cb.KubeClientOrDie(o.Namespace, useProtobuf),
 			o.EnableMetrics,
