[v1.5.10] s4: Fix 478

Author: bfbachmann

cluster/addons.go

@@ -560,7 +560,7 @@ func (c *Cluster) StoreAddonConfigMap(ctx context.Context, addonYaml string, add
 	select {
 	case <-timeout:
 		return updated, nil
-	case <-time.After(time.Second * UpdateStateTimeout):
+	case <-time.After(UpdateStateTimeout):
 		return updated, fmt.Errorf("[addons] Timeout waiting for kubernetes to be ready")
 	}
 }

cluster/cluster.go

@@ -86,8 +86,9 @@ const (
 	AuthnWebhookProvider   = "webhook"
 	StateConfigMapName     = "cluster-state"
 	FullStateConfigMapName = "full-cluster-state"
-	UpdateStateTimeout     = 30
-	GetStateTimeout        = 30
+	FullStateSecretName    = "full-cluster-state"
+	UpdateStateTimeout     = time.Second * 30
+	GetStateTimeout        = time.Second * 30
 	RewriteWorkers         = 5
 	SyncWorkers            = 10
 	NoneAuthorizationMode  = "none"

cluster/state.go

@@ -3,6 +3,7 @@ package cluster
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -11,8 +12,6 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/client-go/transport"
-
 	"github.com/rancher/rke/hosts"
 	"github.com/rancher/rke/k8s"
 	"github.com/rancher/rke/log"
@@ -22,13 +21,22 @@ import (
 	"github.com/sirupsen/logrus"
 	"gopkg.in/yaml.v2"
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/transport"
 )
 
 const (
 	stateFileExt = ".rkestate"
 	certDirExt   = "_certs"
 )
 
+var (
+	ErrFullStateIsNil = errors.New("fullState argument cannot be nil")
+)
+
 type FullState struct {
 	DesiredState State `json:"desiredState,omitempty"`
 	CurrentState State `json:"currentState,omitempty"`
@@ -79,45 +87,152 @@ func (c *Cluster) GetStateFileFromConfigMap(ctx context.Context) (string, error)
 		}
 		return stateFile, nil
 	}
-	return "", fmt.Errorf("Unable to get ConfigMap with cluster state from any Control Plane host")
+	return "", fmt.Errorf("[state] Unable to get ConfigMap with cluster state from any Control Plane host")
 }
 
-func SaveFullStateToKubernetes(ctx context.Context, kubeCluster *Cluster, fullState *FullState) error {
-	k8sClient, err := k8s.NewClient(kubeCluster.LocalKubeConfigPath, kubeCluster.K8sWrapTransport)
-	if err != nil {
-		return fmt.Errorf("Failed to create Kubernetes Client: %v", err)
-	}
+// SaveFullStateToK8s saves the full cluster state to a k8s secret. If any errors that occur on attempts to update
+// the secret will be retired up until some limit.
+func SaveFullStateToK8s(ctx context.Context, k8sClient kubernetes.Interface, fullState *FullState) error {
 	log.Infof(ctx, "[state] Saving full cluster state to Kubernetes")
-	stateFile, err := json.Marshal(*fullState)
+
+	if fullState == nil {
+		return ErrFullStateIsNil
+	}
+
+	secrets := k8sClient.CoreV1().Secrets(metav1.NamespaceSystem)
+	configMaps := k8sClient.CoreV1().ConfigMaps(metav1.NamespaceSystem)
+	stateBytes, err := json.Marshal(fullState)
 	if err != nil {
-		return err
+		return fmt.Errorf("[state] error marshalling full state to JSON: %w", err)
 	}
-	timeout := make(chan bool, 1)
-	go func() {
-		for {
-			_, err := k8s.UpdateConfigMap(k8sClient, stateFile, FullStateConfigMapName)
+
+	// Back off for 1s between attempts.
+	backoff := wait.Backoff{
+		Duration: time.Second,
+		Steps:    int(UpdateStateTimeout.Seconds()),
+	}
+
+	// Try to create or update the secret and delete the old configmap in k8s, if it still exists.
+	saveState := func(ctx context.Context) (bool, error) {
+		// Check if the secret already exists.
+		existingSecret, err := secrets.Get(ctx, FullStateSecretName, metav1.GetOptions{})
+		if err == nil {
+			// The secret already exists, update it.
+			existingSecretCopy := existingSecret.DeepCopy()
+			existingSecretCopy.Data[FullStateSecretName] = stateBytes
+			if _, err := secrets.Update(ctx, existingSecretCopy, metav1.UpdateOptions{}); err != nil {
+				return false, fmt.Errorf("[state] error updating secret: %w", err)
+			}
+		} else if apierrors.IsNotFound(err) {
+			// The secret does not exist, create it.
+			_, err := secrets.Create(ctx, &v1.Secret{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      FullStateSecretName,
+					Namespace: metav1.NamespaceSystem,
+				},
+				Data: map[string][]byte{
+					FullStateSecretName: stateBytes,
+				},
+			}, metav1.CreateOptions{})
 			if err != nil {
-				time.Sleep(time.Second * 5)
-				continue
+				return false, fmt.Errorf("[state] error creating secret: %w", err)
 			}
-			log.Infof(ctx, "[state] Successfully Saved full cluster state to Kubernetes ConfigMap: %s", FullStateConfigMapName)
-			timeout <- true
-			break
+		} else {
+			return false, fmt.Errorf("[state] error getting secret: %w", err)
 		}
-	}()
-	select {
-	case <-timeout:
-		return nil
-	case <-time.After(time.Second * UpdateStateTimeout):
-		return fmt.Errorf("[state] Timeout waiting for kubernetes to be ready")
+
+		// Delete the old configmap.
+		err = configMaps.Delete(ctx, FullStateConfigMapName, metav1.DeleteOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return false, fmt.Errorf("[state] error deleting configmap: %w", err)
+		}
+
+		return true, nil
 	}
+
+	// Retry until success or backoff.Steps has been reached ctx is cancelled.
+	if err = wait.ExponentialBackoffWithContext(ctx, backoff, saveState); err != nil {
+		return fmt.Errorf("[state] error updating secret: %w", err)
+	}
+
+	return nil
+}
+
+// GetFullStateFromK8s fetches the full cluster state from the k8s cluster.
+// In earlier versions of RKE, the full cluster state was stored in a configmap, but it has since been moved
+// to a secret. This function tries fetching it from the secret first and will fall back on the configmap if the secret
+// doesn't exist.
+func GetFullStateFromK8s(ctx context.Context, k8sClient kubernetes.Interface) (*FullState, error) {
+	// Back off for 1s between attempts.
+	backoff := wait.Backoff{
+		Duration: time.Second,
+		Steps:    int(GetStateTimeout.Seconds()),
+	}
+
+	// Try to fetch secret or configmap in k8s.
+	var fullState FullState
+	getState := func(ctx context.Context) (bool, error) {
+		fullStateBytes, err := getFullStateBytesFromSecret(ctx, k8sClient, FullStateSecretName)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				logrus.Debug("full-state secret not found, falling back to configmap")
+
+				fullStateBytes, err = getFullStateBytesFromConfigMap(ctx, k8sClient, FullStateConfigMapName)
+				if err != nil {
+					return false, fmt.Errorf("[state] error getting full state from configmap: %w", err)
+				}
+			} else {
+				return false, fmt.Errorf("[state] error getting full state from secret: %w", err)
+			}
+		}
+
+		if err := json.Unmarshal(fullStateBytes, &fullState); err != nil {
+			return false, fmt.Errorf("[state] error unmarshalling full state from JSON: %w", err)
+		}
+
+		return true, nil
+	}
+
+	// Retry until success or backoff.Steps has been reached or ctx is cancelled.
+	err := wait.ExponentialBackoffWithContext(ctx, backoff, getState)
+	return &fullState, err
+}
+
+// getFullStateBytesFromConfigMap fetches the full state from the configmap with the given name in the kube-system namespace.
+func getFullStateBytesFromConfigMap(ctx context.Context, k8sClient kubernetes.Interface, name string) ([]byte, error) {
+	confMap, err := k8sClient.CoreV1().ConfigMaps(metav1.NamespaceSystem).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("[state] error getting configmap %s: %w", name, err)
+	}
+
+	data, ok := confMap.Data[name]
+	if !ok {
+		return nil, fmt.Errorf("[state] expected configmap %s to have field %s, but none was found", name, name)
+	}
+
+	return []byte(data), nil
+}
+
+// getFullStateBytesFromSecret fetches the full state from the secret with the given name in the kube-system namespace.
+func getFullStateBytesFromSecret(ctx context.Context, k8sClient kubernetes.Interface, name string) ([]byte, error) {
+	secret, err := k8sClient.CoreV1().Secrets(metav1.NamespaceSystem).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("[state] error getting secret %s: %w", name, err)
+	}
+
+	data, ok := secret.Data[name]
+	if !ok {
+		return nil, fmt.Errorf("[state] expected secret %s to have field %s, but none was found", name, name)
+	}
+
+	return data, nil
 }
 
 func GetStateFromKubernetes(ctx context.Context, kubeCluster *Cluster) (*Cluster, error) {
 	log.Infof(ctx, "[state] Fetching cluster state from Kubernetes")
 	k8sClient, err := k8s.NewClient(kubeCluster.LocalKubeConfigPath, kubeCluster.K8sWrapTransport)
 	if err != nil {
-		return nil, fmt.Errorf("Failed to create Kubernetes Client: %v", err)
+		return nil, fmt.Errorf("[state] Failed to create Kubernetes Client: %v", err)
 	}
 	var cfgMap *v1.ConfigMap
 	var currentCluster Cluster
@@ -139,26 +254,26 @@ func GetStateFromKubernetes(ctx context.Context, kubeCluster *Cluster) (*Cluster
 		clusterData := cfgMap.Data[StateConfigMapName]
 		err := yaml.Unmarshal([]byte(clusterData), &currentCluster)
 		if err != nil {
-			return nil, fmt.Errorf("Failed to unmarshal cluster data")
+			return nil, fmt.Errorf("[state] Failed to unmarshal cluster data")
 		}
 		return &currentCluster, nil
-	case <-time.After(time.Second * GetStateTimeout):
+	case <-time.After(GetStateTimeout):
 		log.Infof(ctx, "Timed out waiting for kubernetes cluster to get state")
-		return nil, fmt.Errorf("Timeout waiting for kubernetes cluster to get state")
+		return nil, fmt.Errorf("[state] Timeout waiting for kubernetes cluster to get state")
 	}
 }
 
 func GetK8sVersion(localConfigPath string, k8sWrapTransport transport.WrapperFunc) (string, error) {
 	logrus.Debugf("[version] Using %s to connect to Kubernetes cluster..", localConfigPath)
 	k8sClient, err := k8s.NewClient(localConfigPath, k8sWrapTransport)
 	if err != nil {
-		return "", fmt.Errorf("Failed to create Kubernetes Client: %v", err)
+		return "", fmt.Errorf("[state] Failed to create Kubernetes Client: %v", err)
 	}
 	discoveryClient := k8sClient.DiscoveryClient
 	logrus.Debugf("[version] Getting Kubernetes server version..")
 	serverVersion, err := discoveryClient.ServerVersion()
 	if err != nil {
-		return "", fmt.Errorf("Failed to get Kubernetes server version: %v", err)
+		return "", fmt.Errorf("[state] Failed to get Kubernetes server version: %v", err)
 	}
 	return fmt.Sprintf("%#v", *serverVersion), nil
 }
@@ -174,11 +289,11 @@ func RebuildState(ctx context.Context, kubeCluster *Cluster, oldState *FullState
 	if flags.CustomCerts {
 		certBundle, err := pki.ReadCertsAndKeysFromDir(flags.CertificateDir)
 		if err != nil {
-			return nil, fmt.Errorf("Failed to read certificates from dir [%s]: %v", flags.CertificateDir, err)
+			return nil, fmt.Errorf("[state] Failed to read certificates from dir [%s]: %v", flags.CertificateDir, err)
 		}
 		// make sure all custom certs are included
 		if err := pki.ValidateBundleContent(rkeConfig, certBundle, flags.ClusterFilePath, flags.ConfigDir); err != nil {
-			return nil, fmt.Errorf("Failed to validates certificates from dir [%s]: %v", flags.CertificateDir, err)
+			return nil, fmt.Errorf("[state] Failed to validates certificates from dir [%s]: %v", flags.CertificateDir, err)
 		}
 		newState.DesiredState.CertificatesBundle = certBundle
 		newState.CurrentState = oldState.CurrentState
@@ -207,11 +322,11 @@ func RebuildState(ctx context.Context, kubeCluster *Cluster, oldState *FullState
 func (s *FullState) WriteStateFile(ctx context.Context, statePath string) error {
 	stateFile, err := json.MarshalIndent(s, "", "  ")
 	if err != nil {
-		return fmt.Errorf("Failed to Marshal state object: %v", err)
+		return fmt.Errorf("[state] Failed to Marshal state object: %v", err)
 	}
 	logrus.Tracef("Writing state file: %s", stateFile)
 	if err := os.WriteFile(statePath, stateFile, 0600); err != nil {
-		return fmt.Errorf("Failed to write state file: %v", err)
+		return fmt.Errorf("[state] Failed to write state file: %v", err)
 	}
 	log.Infof(ctx, "Successfully Deployed state file at [%s]", statePath)
 	return nil
@@ -264,19 +379,19 @@ func ReadStateFile(ctx context.Context, statePath string) (*FullState, error) {
 	rkeFullState := &FullState{}
 	fp, err := filepath.Abs(statePath)
 	if err != nil {
-		return rkeFullState, fmt.Errorf("failed to lookup current directory name: %v", err)
+		return rkeFullState, fmt.Errorf("[state] failed to lookup current directory name: %v", err)
 	}
 	file, err := os.Open(fp)
 	if err != nil {
-		return rkeFullState, fmt.Errorf("Can not find RKE state file: %v", err)
+		return rkeFullState, fmt.Errorf("[state] Can not find RKE state file: %v", err)
 	}
 	defer file.Close()
 	buf, err := io.ReadAll(file)
 	if err != nil {
-		return rkeFullState, fmt.Errorf("failed to read state file: %v", err)
+		return rkeFullState, fmt.Errorf("[state] failed to read state file: %v", err)
 	}
 	if err := json.Unmarshal(buf, rkeFullState); err != nil {
-		return rkeFullState, fmt.Errorf("failed to unmarshal the state file: %v", err)
+		return rkeFullState, fmt.Errorf("[state] failed to unmarshal the state file: %v", err)
 	}
 	rkeFullState.DesiredState.CertificatesBundle = pki.TransformPEMToObject(rkeFullState.DesiredState.CertificatesBundle)
 	rkeFullState.CurrentState.CertificatesBundle = pki.TransformPEMToObject(rkeFullState.CurrentState.CertificatesBundle)
@@ -322,7 +437,7 @@ func buildFreshState(ctx context.Context, kubeCluster *Cluster, newState *FullSt
 	// Get the certificate Bundle
 	certBundle, err := pki.GenerateRKECerts(ctx, *rkeConfig, "", "")
 	if err != nil {
-		return fmt.Errorf("Failed to generate certificate bundle: %v", err)
+		return fmt.Errorf("[state] Failed to generate certificate bundle: %v", err)
 	}
 	newState.DesiredState.CertificatesBundle = certBundle
 	if isEncryptionEnabled(rkeConfig) {