Merge pull request #412 from deads2k/use-internal-lb

update CVO to inject internal loadbalancer for use by the CVO pod

install/0000_00_cluster-version-operator_03_deployment.yaml

@@ -45,9 +45,9 @@ spec:
             name: serving-cert
             readOnly: true
         env:
-          - name: KUBERNETES_SERVICE_PORT # allows CVO to communicate with apiserver directly on same host.
+          - name: KUBERNETES_SERVICE_PORT # allows CVO to communicate with apiserver directly on same host.  Is substituted with port from infrastructures.status.apiServerInternalURL if available.
             value: "6443"
-          - name: KUBERNETES_SERVICE_HOST # allows CVO to communicate with apiserver directly on same host.
+          - name: KUBERNETES_SERVICE_HOST # allows CVO to communicate with apiserver directly on same host.  Is substituted with hostname from infrastructures.status.apiServerInternalURL if available.
             value: "127.0.0.1"
           - name: NODE_NAME
             valueFrom:

lib/resourcebuilder/apps.go

@@ -3,6 +3,8 @@ package resourcebuilder
 import (
 	"context"
 	"fmt"
+	"net"
+	"net/url"
 	"strings"
 
 	configv1 "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1"
@@ -22,18 +24,20 @@ import (
 )
 
 type deploymentBuilder struct {
-	client      *appsclientv1.AppsV1Client
-	proxyGetter configv1.ProxiesGetter
-	raw         []byte
-	modifier    MetaV1ObjectModifierFunc
-	mode        Mode
+	client               *appsclientv1.AppsV1Client
+	proxyGetter          configv1.ProxiesGetter
+	infrastructureGetter configv1.InfrastructuresGetter
+	raw                  []byte
+	modifier             MetaV1ObjectModifierFunc
+	mode                 Mode
 }
 
 func newDeploymentBuilder(config *rest.Config, m lib.Manifest) Interface {
 	return &deploymentBuilder{
-		client:      appsclientv1.NewForConfigOrDie(withProtobuf(config)),
-		proxyGetter: configv1.NewForConfigOrDie(config),
-		raw:         m.Raw,
+		client:               appsclientv1.NewForConfigOrDie(withProtobuf(config)),
+		proxyGetter:          configv1.NewForConfigOrDie(config),
+		infrastructureGetter: configv1.NewForConfigOrDie(config),
+		raw:                  m.Raw,
 	}
 }
 
@@ -72,6 +76,43 @@ func (b *deploymentBuilder) Do(ctx context.Context) error {
 		}
 	}
 
+	// if we detect the CVO deployment we need to replace the KUBERNETES_SERVICE_HOST env var with the internal load
+	// balancer to be resilient to kube-apiserver rollouts that cause the localhost server to become non-responsive for
+	// multiple minutes.
+	if deployment.Namespace == "openshift-cluster-version" && deployment.Name == "cluster-version-operator" {
+		infrastructureConfig, err := b.infrastructureGetter.Infrastructures().Get("cluster", metav1.GetOptions{})
+		// not found just means that we don't have infrastructure configuration yet, so we should tolerate not found and avoid substitution
+		if err != nil && !errors.IsNotFound(err) {
+			return err
+		}
+		if !errors.IsNotFound(err) {
+			lbURL, err := url.Parse(infrastructureConfig.Status.APIServerInternalURL)
+			if err != nil {
+				return err
+			}
+			// if we have any error and have empty strings, substitution below will do nothing and leave the manifest specified value
+			// errors can happen when the port is not specified, in which case we have a host and we write that into the env vars
+			lbHost, lbPort, err := net.SplitHostPort(lbURL.Host)
+			if err != nil {
+				if strings.Contains(err.Error(), "missing port in address") {
+					lbHost = lbURL.Host
+					lbPort = ""
+				} else {
+					return err
+				}
+			}
+			err = updatePodSpecWithInternalLoadBalancerKubeService(
+				&deployment.Spec.Template.Spec,
+				[]string{"cluster-version-operator"},
+				lbHost,
+				lbPort,
+			)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
 	_, updated, err := resourceapply.ApplyDeployment(b.client, deployment)
 	if err != nil {
 		return err

lib/resourcebuilder/podspec.go

@@ -45,3 +45,78 @@ func updatePodSpecWithProxy(podSpec *corev1.PodSpec, containerNames []string, ht
 	return nil
 
 }
+
+// updatePodSpecWithInternalLoadBalancerKubeService mutates the input podspec by setting the KUBERNETES_SERVICE_HOST to the internal
+// loadbalancer endpoint and the KUBERNETES_SERVICE_PORT to the specified port
+func updatePodSpecWithInternalLoadBalancerKubeService(podSpec *corev1.PodSpec, containerNames []string, internalLoadBalancerHost, internalLoadBalancerPort string) error {
+	hasInternalLoadBalancer := len(internalLoadBalancerHost) > 0
+	if !hasInternalLoadBalancer {
+		return nil
+	}
+
+	for _, containerName := range containerNames {
+		found := false
+		for i := range podSpec.Containers {
+			if podSpec.Containers[i].Name != containerName {
+				continue
+			}
+			found = true
+
+			podSpec.Containers[i].Env = setKubeServiceValue(podSpec.Containers[i].Env, internalLoadBalancerHost, internalLoadBalancerPort)
+		}
+		for i := range podSpec.InitContainers {
+			if podSpec.InitContainers[i].Name != containerName {
+				continue
+			}
+			found = true
+
+			podSpec.InitContainers[i].Env = setKubeServiceValue(podSpec.Containers[i].Env, internalLoadBalancerHost, internalLoadBalancerPort)
+		}
+
+		if !found {
+			return fmt.Errorf("requested injection for non-existent container: %q", containerName)
+		}
+	}
+
+	return nil
+}
+
+// setKubeServiceValue replaces values if they are present and adds them if they are not
+func setKubeServiceValue(in []corev1.EnvVar, internalLoadBalancerHost, internalLoadBalancerPort string) []corev1.EnvVar {
+	ret := []corev1.EnvVar{}
+
+	portVal := "443"
+	if len(internalLoadBalancerPort) != 0 {
+		portVal = internalLoadBalancerPort
+	}
+
+	foundPort := false
+	foundHost := false
+	for j := range in {
+		ret = append(ret, *in[j].DeepCopy())
+		if ret[j].Name == "KUBERNETES_SERVICE_PORT" {
+			foundPort = true
+			ret[j].Value = portVal
+		}
+		if ret[j].Name == "KUBERNETES_SERVICE_HOST" {
+			foundHost = true
+			ret[j].Value = internalLoadBalancerHost
+		}
+	}
+
+	if !foundPort {
+		ret = append(ret, corev1.EnvVar{
+			Name:  "KUBERNETES_SERVICE_PORT",
+			Value: portVal,
+		})
+	}
+
+	if !foundHost {
+		ret = append(ret, corev1.EnvVar{
+			Name:  "KUBERNETES_SERVICE_HOST",
+			Value: internalLoadBalancerHost,
+		})
+	}
+
+	return ret
+}

lib/resourcebuilder/podspec_test.go

@@ -110,3 +110,159 @@ func TestUpdatePodSpecWithProxy(t *testing.T) {
 		})
 	}
 }
+
+func TestUpdatePodSpecWithInternalLoadBalancerKubeService(t *testing.T) {
+	tests := []struct {
+		name string
+
+		input          *corev1.PodSpec
+		containerNames []string
+		lbHost, lbPort string
+
+		expectedErr string
+		expected    *corev1.PodSpec
+	}{
+		{
+			name: "no lbhost val",
+			input: &corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name: "foo",
+					},
+				},
+			},
+			expected: &corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name: "foo",
+					},
+				},
+			},
+		},
+		{
+			name:           "host and port, add to container and mutate",
+			containerNames: []string{"foo", "init-foo"},
+			lbHost:         "lbhost-val",
+			lbPort:         "lbport-val",
+			input: &corev1.PodSpec{
+				InitContainers: []corev1.Container{
+					{
+						Name: "init-foo",
+						Env: []corev1.EnvVar{
+							{Name: "KUBERNETES_SERVICE_PORT", Value: "oldport-val"},
+							{Name: "KUBERNETES_SERVICE_HOST", Value: "oldhost-val"},
+						},
+					},
+					{
+						Name: "init-bar",
+					},
+				},
+				Containers: []corev1.Container{
+					{
+						Name: "foo",
+					},
+					{
+						Name: "bar",
+					},
+				},
+			},
+			expected: &corev1.PodSpec{
+				InitContainers: []corev1.Container{
+					{
+						Name: "init-foo",
+						Env: []corev1.EnvVar{
+							{Name: "KUBERNETES_SERVICE_PORT", Value: "lbport-val"},
+							{Name: "KUBERNETES_SERVICE_HOST", Value: "lbhost-val"},
+						},
+					},
+					{
+						Name: "init-bar",
+					},
+				},
+				Containers: []corev1.Container{
+					{
+						Name: "foo",
+						Env: []corev1.EnvVar{
+							{Name: "KUBERNETES_SERVICE_PORT", Value: "lbport-val"},
+							{Name: "KUBERNETES_SERVICE_HOST", Value: "lbhost-val"},
+						},
+					},
+					{
+						Name: "bar",
+					},
+				},
+			},
+		},
+		{
+			name:           "lbhost only, add to container and mutate",
+			containerNames: []string{"foo", "init-foo"},
+			lbHost:         "lbhost-val",
+			input: &corev1.PodSpec{
+				InitContainers: []corev1.Container{
+					{
+						Name: "init-foo",
+						Env: []corev1.EnvVar{
+							{Name: "KUBERNETES_SERVICE_PORT", Value: "oldport-val"},
+							{Name: "KUBERNETES_SERVICE_HOST", Value: "oldhost-val"},
+						}},
+					{
+						Name: "init-bar",
+					},
+				},
+				Containers: []corev1.Container{
+					{
+						Name: "foo",
+					},
+					{
+						Name: "bar",
+					},
+				},
+			},
+			expected: &corev1.PodSpec{
+				InitContainers: []corev1.Container{
+					{
+						Name: "init-foo",
+						Env: []corev1.EnvVar{
+							{Name: "KUBERNETES_SERVICE_PORT", Value: "443"},
+							{Name: "KUBERNETES_SERVICE_HOST", Value: "lbhost-val"},
+						},
+					},
+					{
+						Name: "init-bar",
+					},
+				},
+				Containers: []corev1.Container{
+					{
+						Name: "foo",
+						Env: []corev1.EnvVar{
+							{Name: "KUBERNETES_SERVICE_PORT", Value: "443"},
+							{Name: "KUBERNETES_SERVICE_HOST", Value: "lbhost-val"},
+						},
+					},
+					{
+						Name: "bar",
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := updatePodSpecWithInternalLoadBalancerKubeService(test.input, test.containerNames, test.lbHost, test.lbPort)
+			switch {
+			case err == nil && len(test.expectedErr) == 0:
+			case err != nil && len(test.expectedErr) == 0:
+				t.Fatal(err)
+			case err == nil && len(test.expectedErr) != 0:
+				t.Fatal(err)
+			case err != nil && len(test.expectedErr) != 0 && err.Error() != test.expectedErr:
+				t.Fatal(err)
+			}
+
+			if !reflect.DeepEqual(test.input, test.expected) {
+				t.Error(diff.ObjectDiff(test.input, test.expected))
+			}
+		})
+	}
+}