fix(thin): reconcile post-ready ThinRuntime fuse template updates

ThinEngine.SyncRuntime was an empty stub, while pkg/ddc/base/syncs.go calls it on every
reconciliation. The fuse values were therefore only ever rendered once, during setup, so editing a
ready ThinRuntime's spec.fuse never reached the rendered values ConfigMap or the fuse DaemonSet and
operators had to patch the generated DaemonSet by hand.

Implement it following the JuiceFS shape, treating the helm values ConfigMap as the last synced
state: re-render the desired value from the ThinRuntime and its ThinRuntimeProfile, diff it against
that last synced state, push the differences into the fuse DaemonSet, and only then commit the
advanced value back to the ConfigMap, so an interrupted sync is retried instead of forgotten.

Covered fields are resources, image, imageTag, imagePullPolicy, envs (which is also how fuse options
reach the pod), lifecycle, pod labels and annotations, volumes and volumeMounts. nodeSelector is
left alone because transformFuse injects the fuse scheduling label CSI relies on, and configValue is
already reconciled by updateFuseConfigOnChange.

The fuse DaemonSet uses the OnDelete update strategy, so the template is updated without restarting
running fuse pods. The strategy is verified before every sync, and the change is surfaced as a
FuseTemplateUpdated event that spells out the rollout semantics.

parseFuseOptions now sorts the rendered mount options. Its map iteration order was previously
unobservable, but SyncRuntime would otherwise see a different mount options env variable on every
reconciliation and keep updating the DaemonSet forever.

Also add utils.TransformInternalResourcesToCoreV1Resources, the missing inverse of
TransformCoreV1ResourcesToInternalResources, needed to compare a value that round tripped through
the values ConfigMap against the live DaemonSet.

Fixes #6150

Signed-off-by: cheyang <cheyang.cy@alibaba-inc.com>

Author: cheyang

pkg/common/constants.go

@@ -40,6 +40,8 @@ const (
 
 	FuseUmountDuplicate = "UnmountDuplicateMountpoint"
 
+	FuseTemplateUpdated = "FuseTemplateUpdated"
+
 	RuntimeDeprecated = "RuntimeDeprecated"
 
 	RuntimeWithSecretNotSupported = "RuntimeWithSecretNotSupported"

pkg/ddc/thin/sync_runtime.go

@@ -16,8 +16,311 @@
 
 package thin
 
-import cruntime "github.com/fluid-cloudnative/fluid/pkg/runtime"
+import (
+	"context"
+	"reflect"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/retry"
+
+	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
+	"github.com/fluid-cloudnative/fluid/pkg/common"
+	cruntime "github.com/fluid-cloudnative/fluid/pkg/runtime"
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
+	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
+	runtimeOpts "github.com/fluid-cloudnative/fluid/pkg/utils/runtimes/options"
+)
+
+// fuseContainerName is the name of the fuse container rendered by the thin chart,
+// see charts/thin/templates/fuse/daemonset.yaml.
+const fuseContainerName = "thin-fuse"
+
+// SyncRuntime reconciles the fuse template fields of a ThinRuntime that has already been set up.
+// Without it, editing a ready ThinRuntime's spec.fuse never reaches the rendered values or the fuse
+// DaemonSet, because they are only ever generated once during setup.
+//
+// The helm values ConfigMap is treated as the last synced state: the desired state is re-rendered
+// from the ThinRuntime and its ThinRuntimeProfile, diffed against that last synced state, pushed to
+// the DaemonSet, and only then committed back to the ConfigMap. Doing it in that order means an
+// interrupted sync is retried on the next reconciliation instead of being silently forgotten.
+//
+// The fuse DaemonSet uses the OnDelete update strategy, so updating its template does not restart
+// running fuse pods. They pick up the new template whenever they are deleted, which is why the
+// change is also surfaced as an event.
+func (t *ThinEngine) SyncRuntime(ctx cruntime.ReconcileRequestContext) (changed bool, err error) {
+	if runtimeOpts.ShouldSkipSyncingRuntime() {
+		t.Log.V(1).Info("Skipping runtime sync due to CONTROLLER_SKIP_SYNCING_RUNTIME being enabled")
+		return
+	}
+
+	runtime, err := t.getRuntime()
+	if err != nil {
+		return
+	}
+
+	// The engine is cached across reconciliations, so t.runtime and t.runtimeProfile may be stale.
+	profile, err := utils.GetThinRuntimeProfile(t.Client, runtime.Spec.ThinRuntimeProfileName)
+	if err != nil {
+		if apierrs.IsNotFound(err) {
+			t.Log.Info("ThinRuntimeProfile not found, skip syncing the runtime spec",
+				"profile", runtime.Spec.ThinRuntimeProfileName)
+			return false, nil
+		}
+		return false, err
+	}
+
+	latestValue, err := t.transform(runtime, profile)
+	if err != nil {
+		return
+	}
+
+	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+		valueToSync, innerErr := t.GetValueFromConfigmap()
+		if innerErr != nil {
+			return innerErr
+		}
+		if valueToSync == nil {
+			// The user opted out of the values ConfigMap, so there is no last synced state to diff
+			// against. Degrade to not syncing rather than failing the whole reconciliation.
+			t.Log.Info("Helm value configmap not found, skip syncing the runtime spec",
+				"configmap", t.getHelmValuesConfigMapName())
+			return nil
+		}
+
+		fuseChanged, innerErr := t.syncFuseSpec(valueToSync, latestValue)
+		if innerErr != nil {
+			return innerErr
+		}
+
+		changed = fuseChanged
+		if !changed {
+			return nil
+		}
+
+		t.Log.Info("Committing the changed value to the helm value configmap", "name", t.name, "namespace", t.namespace)
+		if innerErr = t.SaveValueToConfigmap(valueToSync); innerErr != nil {
+			t.Log.Error(innerErr, "failed to save the changed value to the helm value configmap")
+			return innerErr
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		t.Log.Error(err, "Failed to sync the runtime spec")
+		return false, err
+	}
+
+	if changed {
+		t.recordFuseTemplateUpdated(runtime)
+	}
 
-func (t ThinEngine) SyncRuntime(ctx cruntime.ReconcileRequestContext) (changed bool, err error) {
 	return
 }
+
+func (t *ThinEngine) recordFuseTemplateUpdated(runtime *datav1alpha1.ThinRuntime) {
+	if t.Recorder == nil {
+		return
+	}
+	t.Recorder.Eventf(runtime, corev1.EventTypeNormal, common.FuseTemplateUpdated,
+		"Updated the template of fuse daemonset %s. Its update strategy is OnDelete, so running fuse pods keep the previous template until they are deleted.",
+		t.getFuseName())
+}
+
+// syncFuseSpec pushes the differences between the last synced value and the latest value into the
+// fuse DaemonSet. oldValue is advanced in place for every field that was actually pushed, so that
+// the caller can commit it as the new last synced state.
+func (t *ThinEngine) syncFuseSpec(oldValue, latestValue *ThinValue) (changed bool, err error) {
+	t.Log.V(1).Info("entering syncFuseSpec")
+	defer func() {
+		t.Log.V(1).Info("exiting syncFuseSpec")
+	}()
+
+	fuses, err := kubeclient.GetDaemonset(t.Client, t.getFuseName(), t.namespace)
+	if err != nil {
+		return false, err
+	}
+
+	if fuses.Spec.UpdateStrategy.Type != appsv1.OnDeleteDaemonSetStrategyType {
+		// Updating the template of a RollingUpdate daemonset would restart the running fuse pods and
+		// break the applications mounting them, so switch the strategy first and let the resulting
+		// update event trigger a new reconciliation.
+		t.Log.V(1).Info("Fuse daemonset's update strategy is not safe to sync fuse spec",
+			"updateStrategy", fuses.Spec.UpdateStrategy.Type)
+		if err = kubeclient.UpdateDaemonSetUpdateStrategy(t.Client, fuses.Name, fuses.Namespace,
+			appsv1.DaemonSetUpdateStrategy{Type: appsv1.OnDeleteDaemonSetStrategyType}); err != nil {
+			return false, err
+		}
+		t.Log.Info("syncFuseSpec: successfully updated fuse daemonset's update strategy to OnDelete",
+			"fuse ds", types.NamespacedName{Namespace: fuses.Namespace, Name: fuses.Name})
+		return false, nil
+	}
+
+	fusesToUpdate := fuses.DeepCopy()
+	changed, err = t.checkAndSetFuseChanges(oldValue, latestValue, fusesToUpdate)
+	if err != nil {
+		return false, err
+	}
+	if !changed {
+		t.Log.V(1).Info("syncFuseSpec: no differences detected about fuse")
+		return false, nil
+	}
+
+	if reflect.DeepEqual(fuses, fusesToUpdate) {
+		t.Log.V(1).Info("syncFuseSpec: no differences detected about fuse after equality check")
+		return false, nil
+	}
+
+	t.Log.Info("syncFuseSpec: some fields are changed in fuse, try to update the fuse daemonset",
+		"fuse ds", types.NamespacedName{Namespace: fusesToUpdate.Namespace, Name: fusesToUpdate.Name})
+	if err = t.Client.Update(context.TODO(), fusesToUpdate); err != nil {
+		t.Log.Error(err, "syncFuseSpec: failed to update the fuse daemonset spec",
+			"fuse ds", types.NamespacedName{Namespace: fusesToUpdate.Namespace, Name: fusesToUpdate.Name})
+		return false, err
+	}
+
+	return true, nil
+}
+
+// checkAndSetFuseChanges applies the supported fuse template changes onto fusesToUpdate.
+//
+// Fields the thin chart renders verbatim (resources, image, imagePullPolicy) are compared against
+// the live daemonset, so that manual drift is corrected as well. Fields the chart merges with
+// entries of its own (envs, volumes, volumeMounts, labels, annotations) are compared against the
+// last synced value instead, and only the value-derived entries are replaced, so that the chart's
+// own entries survive.
+//
+// Some fuse fields are deliberately left out:
+//   - nodeSelector, because transformFuse injects the fuse scheduling label that CSI relies on to
+//     place fuse pods, so changing it after creation breaks mounting.
+//   - hostNetwork, hostPID, targetPath, ports, command, args and the probes, whose post-ready change
+//     semantics need their own discussion.
+//   - configValue, which updateFuseConfigOnChange already reconciles.
+func (t *ThinEngine) checkAndSetFuseChanges(oldValue, latestValue *ThinValue, fusesToUpdate *appsv1.DaemonSet) (changed bool, err error) {
+	// volumes
+	if !isSliceEqual(oldValue.Fuse.Volumes, latestValue.Fuse.Volumes) {
+		t.Log.Info("syncFuseSpec: volumes changed", "old", oldValue.Fuse.Volumes, "new", latestValue.Fuse.Volumes)
+		fusesToUpdate.Spec.Template.Spec.Volumes = append(
+			utils.GetVolumesDifference(fusesToUpdate.Spec.Template.Spec.Volumes, oldValue.Fuse.Volumes),
+			latestValue.Fuse.Volumes...)
+		oldValue.Fuse.Volumes = latestValue.Fuse.Volumes
+		changed = true
+	}
+
+	// labels
+	if !isMapEqual(oldValue.Fuse.Labels, latestValue.Fuse.Labels) {
+		t.Log.Info("syncFuseSpec: labels changed", "old", oldValue.Fuse.Labels, "new", latestValue.Fuse.Labels)
+		fusesToUpdate.Spec.Template.Labels = utils.UnionMapsWithOverride(
+			utils.GetMapsDifference(fusesToUpdate.Spec.Template.Labels, oldValue.Fuse.Labels),
+			latestValue.Fuse.Labels)
+		oldValue.Fuse.Labels = latestValue.Fuse.Labels
+		changed = true
+	}
+
+	// annotations
+	if !isMapEqual(oldValue.Fuse.Annotations, latestValue.Fuse.Annotations) {
+		t.Log.Info("syncFuseSpec: annotations changed", "old", oldValue.Fuse.Annotations, "new", latestValue.Fuse.Annotations)
+		fusesToUpdate.Spec.Template.Annotations = utils.UnionMapsWithOverride(
+			utils.GetMapsDifference(fusesToUpdate.Spec.Template.Annotations, oldValue.Fuse.Annotations),
+			latestValue.Fuse.Annotations)
+		oldValue.Fuse.Annotations = latestValue.Fuse.Annotations
+		changed = true
+	}
+
+	containerIdx := utils.GetContainerIndex(fusesToUpdate.Spec.Template.Spec.Containers, fuseContainerName)
+	if containerIdx < 0 {
+		t.Log.Info("syncFuseSpec: fuse container not found in the fuse daemonset, skip syncing the container spec",
+			"container", fuseContainerName)
+		return changed, nil
+	}
+	container := &fusesToUpdate.Spec.Template.Spec.Containers[containerIdx]
+
+	// resources
+	latestResources, err := utils.TransformInternalResourcesToCoreV1Resources(latestValue.Fuse.Resources)
+	if err != nil {
+		return false, err
+	}
+	if !utils.ResourceRequirementsEqual(container.Resources, latestResources) {
+		t.Log.Info("syncFuseSpec: resources changed", "old", container.Resources, "new", latestResources)
+		container.Resources = latestResources
+		oldValue.Fuse.Resources = latestValue.Fuse.Resources
+		changed = true
+	}
+
+	// image
+	if latestImage := composeImage(latestValue.Fuse.Image, latestValue.Fuse.ImageTag); container.Image != latestImage {
+		t.Log.Info("syncFuseSpec: image changed", "old", container.Image, "new", latestImage)
+		container.Image = latestImage
+		oldValue.Fuse.Image = latestValue.Fuse.Image
+		oldValue.Fuse.ImageTag = latestValue.Fuse.ImageTag
+		changed = true
+	}
+
+	// imagePullPolicy
+	// An empty value means the chart falls back to the Kubernetes default, so leave the daemonset
+	// alone instead of clearing a policy that is already in effect.
+	if latestPullPolicy := corev1.PullPolicy(latestValue.Fuse.ImagePullPolicy); latestPullPolicy != "" &&
+		container.ImagePullPolicy != latestPullPolicy {
+		t.Log.Info("syncFuseSpec: image pull policy changed", "old", container.ImagePullPolicy, "new", latestPullPolicy)
+		container.ImagePullPolicy = latestPullPolicy
+		oldValue.Fuse.ImagePullPolicy = latestValue.Fuse.ImagePullPolicy
+		changed = true
+	}
+
+	// envs
+	if !isSliceEqual(oldValue.Fuse.Envs, latestValue.Fuse.Envs) {
+		t.Log.Info("syncFuseSpec: env variables changed", "old", oldValue.Fuse.Envs, "new", latestValue.Fuse.Envs)
+		container.Env = append(
+			utils.GetEnvsDifference(container.Env, oldValue.Fuse.Envs),
+			latestValue.Fuse.Envs...)
+		oldValue.Fuse.Envs = latestValue.Fuse.Envs
+		changed = true
+	}
+
+	// volumeMounts
+	if !isSliceEqual(oldValue.Fuse.VolumeMounts, latestValue.Fuse.VolumeMounts) {
+		t.Log.Info("syncFuseSpec: volume mounts changed", "old", oldValue.Fuse.VolumeMounts, "new", latestValue.Fuse.VolumeMounts)
+		container.VolumeMounts = append(
+			utils.GetVolumeMountsDifference(container.VolumeMounts, oldValue.Fuse.VolumeMounts),
+			latestValue.Fuse.VolumeMounts...)
+		oldValue.Fuse.VolumeMounts = latestValue.Fuse.VolumeMounts
+		changed = true
+	}
+
+	// lifecycle
+	if !reflect.DeepEqual(oldValue.Fuse.Lifecycle, latestValue.Fuse.Lifecycle) {
+		t.Log.Info("syncFuseSpec: lifecycle changed", "old", oldValue.Fuse.Lifecycle, "new", latestValue.Fuse.Lifecycle)
+		container.Lifecycle = latestValue.Fuse.Lifecycle
+		oldValue.Fuse.Lifecycle = latestValue.Fuse.Lifecycle
+		changed = true
+	}
+
+	return changed, nil
+}
+
+func composeImage(image, imageTag string) string {
+	if len(imageTag) == 0 {
+		return image
+	}
+	return image + ":" + imageTag
+}
+
+// isSliceEqual treats a nil slice and an empty slice as equal, because a value that round trips
+// through the values ConfigMap loses that distinction.
+func isSliceEqual[T any](a, b []T) bool {
+	if len(a) == 0 && len(b) == 0 {
+		return true
+	}
+	return reflect.DeepEqual(a, b)
+}
+
+// isMapEqual treats a nil map and an empty map as equal, for the same reason as isSliceEqual.
+func isMapEqual(a, b map[string]string) bool {
+	if len(a) == 0 && len(b) == 0 {
+		return true
+	}
+	return reflect.DeepEqual(a, b)
+}