Merge branch 'main' into log-verbosity

Author: weinimo

config/manifests/bases/octavia-operator.clusterserviceversion.yaml

@@ -23,11 +23,40 @@ spec:
       displayName: Octavia API
       kind: OctaviaAPI
       name: octaviaapis.octavia.openstack.org
+      specDescriptors:
+      - description: TLS - Parameters related to the TLS
+        displayName: TLS
+        path: tls
+      - description: API tls type which encapsulates for API services
+        displayName: API
+        path: tls.api
+      - description: Ovn GenericService - holds the secret for the OvnDb client cert
+        displayName: Ovn
+        path: tls.ovn
       version: v1beta1
     - description: Octavia is the Schema for the octavia API
       displayName: Octavia
       kind: Octavia
       name: octavias.octavia.openstack.org
+      specDescriptors:
+      - description: TLS - Parameters related to the TLS
+        displayName: TLS
+        path: octaviaAPI.tls
+      - description: API tls type which encapsulates for API services
+        displayName: API
+        path: octaviaAPI.tls.api
+      - description: Ovn GenericService - holds the secret for the OvnDb client cert
+        displayName: Ovn
+        path: octaviaAPI.tls.ovn
+      - description: TLS - Parameters related to the TLS
+        displayName: TLS
+        path: octaviaHealthManager.tls
+      - description: TLS - Parameters related to the TLS
+        displayName: TLS
+        path: octaviaHousekeeping.tls
+      - description: TLS - Parameters related to the TLS
+        displayName: TLS
+        path: octaviaWorker.tls
       version: v1beta1
   description: Octavia Operator
   displayName: Octavia Operator

controllers/amphoracontroller_controller.go

@@ -19,7 +19,6 @@ package controllers
 import (
 	"context"
 	"fmt"
-	"sort"
 	"strings"
 	"time"
 
@@ -44,7 +43,6 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	k8s_errors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -578,46 +576,23 @@ func (r *OctaviaAmphoraControllerReconciler) generateServiceConfigMaps(
 			err.Error()))
 		return err
 	}
+
 	//
-	// TODO(beagles): Improve this with predictable IPs for the health managers because what is
-	// going to happen on start up is that health managers will restart each time a new one is deployed.
-	// The easiest strategy is to create a "hole" in the IP address range and control the
-	// allocation and configuration of an additional IP on each network attached interface. We will
-	// need a container in the Pod that has the ip command installed to do this however.
+	// Get the predicatable IPs from the HmConfigMap
 	//
-	healthManagerIPs, err := getPodIPs(
-		fmt.Sprintf("%s-%s", "octavia", octaviav1.HealthManager),
-		instance.Namespace,
-		r.Kclient,
-		&r.Log,
-	)
+	hmMap := &corev1.ConfigMap{}
+	err = helper.GetClient().Get(ctx, types.NamespacedName{Name: octavia.HmConfigMap, Namespace: instance.GetNamespace()}, hmMap)
 	if err != nil {
-		instance.Status.Conditions.Set(condition.FalseCondition(
-			condition.InputReadyCondition,
-			condition.ErrorReason,
-			condition.SeverityWarning,
-			condition.InputReadyErrorMessage,
-			err.Error()))
 		return err
 	}
-
-	// TODO(beagles): come up with a way to preallocate or ensure
-	// a stable list of IPs.
-
-	if instance.Spec.Role == octaviav1.HealthManager {
-		// TODO(gthiemonge) This is fine to leave this list empty in the HM when
-		// we use redis, because the HM doesn't create any LBs, but if we drop
-		// redis, failovers will be triggered in the HM
-		templateParameters["ControllerIPList"] = ""
-	} else if len(healthManagerIPs) == 0 {
-		return fmt.Errorf("Health manager ports are not ready yet")
-	} else {
-		withPorts := make([]string, len(healthManagerIPs))
-		for idx, val := range healthManagerIPs {
-			withPorts[idx] = fmt.Sprintf("%s:5555", val)
+	var ipAddresses []string
+	for key, val := range hmMap.Data {
+		if strings.HasPrefix(key, "hm_") {
+			ipAddresses = append(ipAddresses, fmt.Sprintf("%s:5555", val))
 		}
-		templateParameters["ControllerIPList"] = strings.Join(withPorts, ",")
 	}
+	ipAddressString := strings.Join(ipAddresses, ",")
+	templateParameters["ControllerIPList"] = ipAddressString
 
 	spec := instance.Spec
 	templateParameters["ServiceUser"] = spec.ServiceUser
@@ -727,46 +702,6 @@ func (r *OctaviaAmphoraControllerReconciler) SetupWithManager(mgr ctrl.Manager)
 		Complete(r)
 }
 
-func listHealthManagerPods(name string, ns string, client kubernetes.Interface, log *logr.Logger) (*corev1.PodList, error) {
-	listOptions := metav1.ListOptions{
-		LabelSelector: fmt.Sprintf("%s=%s", common.AppSelector, name),
-		FieldSelector: "status.phase==Running",
-	}
-	log.V(1).Info(fmt.Sprintf("Listing pods using label selector %s and field selector %s", listOptions.LabelSelector, listOptions.FieldSelector))
-	pods, err := client.CoreV1().Pods(ns).List(context.Background(), listOptions)
-	if err != nil {
-		return nil, err
-	}
-	return pods, nil
-}
-
-func getPodIPs(name string, ns string, client kubernetes.Interface, log *logr.Logger) ([]string, error) {
-	//
-	// Get the IPs for the network attachments for these PODs.
-	//
-	var result []string
-	pods, err := listHealthManagerPods(name, ns, client, log)
-	if err != nil {
-		return nil, err
-	}
-	for _, pod := range pods.Items {
-		annotations := pod.GetAnnotations()
-		networkStatusList, err := nad.GetNetworkStatusFromAnnotation(annotations)
-		if err != nil {
-			log.Error(err, fmt.Sprintf("Unable to get network annotations from %s", annotations))
-			return nil, err
-		}
-		for _, networkStatus := range networkStatusList {
-			netAttachName := fmt.Sprintf("%s/%s", ns, octavia.LbNetworkAttachmentName)
-			if networkStatus.Name == netAttachName {
-				result = append(result, networkStatus.IPs[0])
-			}
-		}
-	}
-	sort.Strings(result)
-	return result, nil
-}
-
 func (r *OctaviaAmphoraControllerReconciler) findObjectsForSrc(ctx context.Context, src client.Object) []reconcile.Request {
 	requests := []reconcile.Request{}
 

controllers/octavia_controller.go

@@ -43,6 +43,7 @@ import (
 	octaviav1 "github.com/openstack-k8s-operators/octavia-operator/api/v1beta1"
 	"github.com/openstack-k8s-operators/octavia-operator/pkg/octavia"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -186,7 +187,6 @@ func (r *OctaviaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (re
 		condition.UnknownCondition(octaviav1.OctaviaAPIReadyCondition, condition.InitReason, octaviav1.OctaviaAPIReadyInitMessage),
 		condition.UnknownCondition(condition.NetworkAttachmentsReadyCondition, condition.InitReason, condition.NetworkAttachmentsReadyInitMessage),
 		condition.UnknownCondition(condition.ExposeServiceReadyCondition, condition.InitReason, condition.ExposeServiceReadyInitMessage),
-		condition.UnknownCondition(condition.DeploymentReadyCondition, condition.InitReason, condition.DeploymentReadyInitMessage),
 		condition.UnknownCondition(octaviav1.OctaviaAmphoraCertsReadyCondition, condition.InitReason, octaviav1.OctaviaAmphoraCertsReadyInitMessage),
 		condition.UnknownCondition(octaviav1.OctaviaQuotasReadyCondition, condition.InitReason, octaviav1.OctaviaQuotasReadyInitMessage),
 		condition.UnknownCondition(octaviav1.OctaviaAmphoraSSHReadyCondition, condition.InitReason, octaviav1.OctaviaAmphoraSSHReadyInitMessage),
@@ -504,9 +504,12 @@ func (r *OctaviaReconciler) reconcileNormal(ctx context.Context, instance *octav
 			condition.InputReadyWaitingMessage))
 		return ctrl.Result{RequeueAfter: time.Duration(10) * time.Second}, nil
 	}
+	instance.Status.Conditions.MarkTrue(
+		condition.RabbitMqTransportURLReadyCondition,
+		condition.RabbitMqTransportURLReadyMessage)
 	instance.Status.Conditions.MarkTrue(condition.InputReadyCondition, condition.InputReadyMessage)
 
-	err = octavia.EnsureAmphoraCerts(ctx, instance, helper, &Log)
+	err = octavia.EnsureAmphoraCerts(ctx, instance, helper)
 	if err != nil {
 		instance.Status.Conditions.Set(condition.FalseCondition(
 			octaviav1.OctaviaAmphoraCertsReadyCondition,
@@ -673,6 +676,109 @@ func (r *OctaviaReconciler) reconcileNormal(ctx context.Context, instance *octav
 		return ctrl.Result{}, err
 	}
 
+	nodeConfigMap := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      octavia.HmConfigMap,
+			Namespace: instance.GetNamespace(),
+			Labels:    labels.GetLabels(instance, labels.GetGroupLabel(instance.ObjectMeta.Name), map[string]string{}),
+		},
+		Data: make(map[string]string),
+	}
+
+	// Look for existing config map and if exists, read existing data and match
+	// against nodes.
+	foundMap := &corev1.ConfigMap{}
+	err = helper.GetClient().Get(ctx, types.NamespacedName{Name: octavia.HmConfigMap, Namespace: instance.GetNamespace()},
+		foundMap)
+	if err != nil {
+		if k8s_errors.IsNotFound(err) {
+			Log.Info(fmt.Sprintf("Port map %s doesn't exist, creating.", octavia.HmConfigMap))
+		} else {
+			return ctrl.Result{}, err
+		}
+	} else {
+		Log.Info("Retrieved existing map, updating..")
+		nodeConfigMap.Data = foundMap.Data
+	}
+
+	//
+	// Predictable IPs.
+	//
+	// NOTE(beagles): refactoring this might be nice. This could also  be
+	// optimized but the data sets are small (nodes an IP ranges are less than
+	// 100) so optimization might be a waste.
+	//
+	predictableIPParams, err := octavia.GetPredictableIPAM(networkParameters)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+	// Get a list of the nodes in the cluster
+
+	// TODO(beagles):
+	// * confirm whether or not this lists only the nodes we want (i.e. ones
+	// that will host the daemonset)
+	// * do we want to provide a mechanism to temporarily disabling this list
+	// for maintenance windows where nodes might be "coming and going"
+
+	nodes, _ := helper.GetKClient().CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+	updatedMap := make(map[string]string)
+	allocatedIPs := make(map[string]bool)
+	var predictableIPsRequired []string
+
+	// First scan existing allocations so we can keep existing allocations.
+	// Keeping track of what's required and what already exists. If a node is
+	// removed from the cluster, it's IPs will not be added to the allocated
+	// list and are effectively recycled.
+	for _, node := range nodes.Items {
+		Log.Info(fmt.Sprintf("cluster node name %s", node.Name))
+		portName := fmt.Sprintf("hm_%s", node.Name)
+		if ipValue, ok := nodeConfigMap.Data[portName]; ok {
+			updatedMap[portName] = ipValue
+			allocatedIPs[ipValue] = true
+			Log.Info(fmt.Sprintf("%s has IP mapping %s: %s", node.Name, portName, ipValue))
+		} else {
+			predictableIPsRequired = append(predictableIPsRequired, portName)
+		}
+		portName = fmt.Sprintf("rsyslog_%s", node.Name)
+		if ipValue, ok := nodeConfigMap.Data[portName]; ok {
+			updatedMap[portName] = ipValue
+			allocatedIPs[ipValue] = true
+			Log.Info(fmt.Sprintf("%s has IP mapping %s: %s", node.Name, portName, ipValue))
+		} else {
+			predictableIPsRequired = append(predictableIPsRequired, portName)
+		}
+	}
+	// Get new IPs using the range from predictableIPParmas minus the
+	// allocatedIPs captured above.
+	Log.Info(fmt.Sprintf("Allocating %d predictable IPs", len(predictableIPsRequired)))
+	for _, portName := range predictableIPsRequired {
+		hmPort, err := octavia.GetNextIP(predictableIPParams, allocatedIPs)
+		if err != nil {
+			// An error here is really unexpected- it means either we have
+			// messed up the allocatedIPs list or the range we are assuming is
+			// too small for the number of health managers and rsyslog
+			// containers.
+			return ctrl.Result{}, err
+		}
+		updatedMap[portName] = hmPort
+	}
+
+	mapLabels := labels.GetLabels(instance, labels.GetGroupLabel(instance.ObjectMeta.Name), map[string]string{})
+	_, err = controllerutil.CreateOrPatch(ctx, helper.GetClient(), nodeConfigMap, func() error {
+		nodeConfigMap.Labels = util.MergeStringMaps(nodeConfigMap.Labels, mapLabels)
+		nodeConfigMap.Data = updatedMap
+		err := controllerutil.SetControllerReference(instance, nodeConfigMap, helper.GetScheme())
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+
+	if err != nil {
+		Log.Info("Unable to create config map for health manager ports...")
+		return ctrl.Result{}, err
+	}
+
 	octaviaHealthManager, op, err := r.amphoraControllerDaemonSetCreateOrUpdate(instance, networkInfo,
 		ampImageOwnerID, instance.Spec.OctaviaHealthManager, octaviav1.HealthManager)
 	if err != nil {
@@ -823,6 +929,8 @@ func (r *OctaviaReconciler) reconcileNormal(ctx context.Context, instance *octav
 		octaviav1.OctaviaAmphoraImagesReadyCondition,
 		octaviav1.OctaviaAmphoraImagesReadyCompleteMessage)
 
+	instance.Status.Conditions.MarkTrue(condition.ExposeServiceReadyCondition, condition.ExposeServiceReadyMessage)
+
 	// create Deployment - end
 
 	// We reached the end of the Reconcile, update the Ready condition based on
@@ -965,6 +1073,7 @@ func (r *OctaviaReconciler) reconcileAmphoraImages(
 ) (ctrl.Result, error) {
 	Log := r.GetLogger(ctx)
 
+	var ctrlResult ctrl.Result
 	if instance.Spec.AmphoraImageContainerImage == "" {
 		if instance.Status.Hash[octaviav1.ImageUploadHash] != "" {
 			Log.Info("Reseting image upload hash")
@@ -986,36 +1095,6 @@ func (r *OctaviaReconciler) reconcileAmphoraImages(
 		common.AppSelector: octavia.ServiceName + "-image",
 	}
 
-	Log.Info("Initializing amphora image upload deployment")
-	depl := deployment.NewDeployment(
-		octavia.ImageUploadDeployment(instance, serviceLabels),
-		time.Duration(5)*time.Second,
-	)
-	ctrlResult, err := depl.CreateOrPatch(ctx, helper)
-	if err != nil {
-		instance.Status.Conditions.Set(condition.FalseCondition(
-			condition.DeploymentReadyCondition,
-			condition.ErrorReason,
-			condition.SeverityWarning,
-			condition.DeploymentReadyErrorMessage,
-			err.Error()))
-		return ctrlResult, err
-	} else if (ctrlResult != ctrl.Result{}) {
-		instance.Status.Conditions.Set(condition.FalseCondition(
-			condition.DeploymentReadyCondition,
-			condition.RequestedReason,
-			condition.SeverityInfo,
-			condition.DeploymentReadyRunningMessage))
-		return ctrlResult, nil
-	}
-
-	readyCount := depl.GetDeployment().Status.ReadyReplicas
-	if readyCount == 0 {
-		// Not ready, wait for the next loop
-		Log.Info("Image Upload Pod not ready")
-		return ctrl.Result{Requeue: true, RequeueAfter: 1 * time.Second}, nil
-	}
-
 	exportLabels := util.MergeStringMaps(
 		serviceLabels,
 		map[string]string{
@@ -1076,6 +1155,35 @@ func (r *OctaviaReconciler) reconcileAmphoraImages(
 			condition.ExposeServiceReadyRunningMessage))
 		return ctrlResult, nil
 	}
+	Log.Info("Initializing amphora image upload deployment")
+	depl := deployment.NewDeployment(
+		octavia.ImageUploadDeployment(instance, serviceLabels),
+		time.Duration(5)*time.Second,
+	)
+	ctrlResult, err = depl.CreateOrPatch(ctx, helper)
+	if err != nil {
+		instance.Status.Conditions.Set(condition.FalseCondition(
+			octaviav1.OctaviaAmphoraImagesReadyCondition,
+			condition.ErrorReason,
+			condition.SeverityWarning,
+			octaviav1.OctaviaAmphoraImagesReadyErrorMessage,
+			err.Error()))
+		return ctrlResult, err
+	} else if (ctrlResult != ctrl.Result{}) {
+		instance.Status.Conditions.Set(condition.FalseCondition(
+			octaviav1.OctaviaAmphoraImagesReadyCondition,
+			condition.ErrorReason,
+			condition.SeverityWarning,
+			octaviav1.OctaviaAmphoraImagesReadyErrorMessage,
+			err.Error()))
+		return ctrlResult, nil
+	}
+	readyCount := depl.GetDeployment().Status.ReadyReplicas
+	if readyCount == 0 {
+		// Not ready, wait for the next loop
+		Log.Info("Image Upload Pod not ready")
+		return ctrl.Result{Requeue: true, RequeueAfter: 1 * time.Second}, nil
+	}
 	endpoint, err := svc.GetAPIEndpoint(nil, nil, "")
 	if err != nil {
 		return ctrl.Result{}, err