open-cluster-management/pkg/work/spoke/apply/server_side_apply.go

package apply

import (
	"context"
	"crypto/md5" //nolint:gosec
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"
	"time"

	jsonpatch "github.com/evanphx/json-patch"
	"github.com/itchyny/gojq"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/util/jsonpath"
	"k8s.io/klog/v2"

	workapiv1 "open-cluster-management.io/api/work/v1"
	"open-cluster-management.io/sdk-go/pkg/basecontroller/events"

	"open-cluster-management.io/ocm/pkg/work/helper"
)

type ServerSideApply struct {
	client dynamic.Interface
}

type ServerSideApplyConflictError struct {
	ssaErr error
}

func (e *ServerSideApplyConflictError) Error() string {
	return e.ssaErr.Error()
}

func NewServerSideApply(client dynamic.Interface) *ServerSideApply {
	return &ServerSideApply{client: client}
}

func (c *ServerSideApply) Apply(
	ctx context.Context,
	gvr schema.GroupVersionResource,
	requiredOriginal *unstructured.Unstructured,
	owner metav1.OwnerReference,
	applyOption *workapiv1.ManifestConfigOption,
	_ events.Recorder) (runtime.Object, error) {
	logger := klog.FromContext(ctx)
	// Currently, if the required object has zero creationTime in metadata, it will cause
	// kube-apiserver to increment generation even if nothing else changes. more details see:
	// https://github.com/kubernetes/kubernetes/issues/67610
	//
	// TODO Remove this after the above issue fixed in Kubernetes
	removeCreationTimeFromMetadata(requiredOriginal.Object, logger)

	force := false
	fieldManager := workapiv1.DefaultFieldManager
	var requiredHash string

	required := requiredOriginal.DeepCopy()

	if applyOption.UpdateStrategy.ServerSideApply != nil {
		force = applyOption.UpdateStrategy.ServerSideApply.Force
		if len(applyOption.UpdateStrategy.ServerSideApply.FieldManager) > 0 {
			fieldManager = applyOption.UpdateStrategy.ServerSideApply.FieldManager
		}

		ignoreFields := applyOption.UpdateStrategy.ServerSideApply.IgnoreFields
		if len(ignoreFields) > 0 {
			for _, field := range ignoreFields {
				// for IgnoreFieldsConditionOnSpokeChange, it will still be included when computing the hash. So when
				// hash dismatch, these fields will still the patched on the cluster.
				if field.Condition == workapiv1.IgnoreFieldsConditionOnSpokeChange {
					continue
				}

				// Process JSONPaths (existing behavior)
				for _, path := range field.JSONPaths {
					removeFieldByJSONPath(required.UnstructuredContent(), path, logger)
				}

				// Process JSONPointers (RFC 6901)
				for _, pointer := range field.JSONPointers {
					if err := removeFieldByJSONPointer(required, pointer, logger); err != nil {
						logger.Error(err, "failed to remove field by JSON Pointer", "pointer", pointer)
						return nil, NewIgnoreFieldError(err, "JSON Pointer error in '%s': %v", pointer, err)
					}
				}

				// Process JQPathExpressions
				for _, expr := range field.JQPathExpressions {
					if err := removeFieldByJQExpression(ctx, required, expr, logger); err != nil {
						logger.Error(err, "failed to remove field by JQ expression", "expression", expr)
						return nil, NewIgnoreFieldError(err, "JQ expression error in '%s': %v", expr, err)
					}
				}
			}
			requiredHash = hashOfResourceStruct(required)
			annotation := required.GetAnnotations()
			if annotation == nil {
				annotation = map[string]string{}
			}
			annotation[workapiv1.ManifestConfigSpecHashAnnotationKey] = requiredHash
			required.SetAnnotations(annotation)
			requiredOriginal.SetAnnotations(annotation)
		}
	}

	// only get existing resource and compare hash if the hash is computed.
	if len(requiredHash) > 0 {
		existing, err := c.client.Resource(gvr).Namespace(required.GetNamespace()).Get(
			ctx, required.GetName(), metav1.GetOptions{})
		switch {
		case apierrors.IsNotFound(err):
			// if object is not found, use requiredOriginal to apply so the ignore fields are kept when create
			required = requiredOriginal
		case err != nil:
			return nil, err
		case len(existing.GetAnnotations()) > 0:
			// skip the apply operation when the hash of the existing resource matches the required hash
			existingHash := existing.GetAnnotations()[workapiv1.ManifestConfigSpecHashAnnotationKey]
			if requiredHash == existingHash {
				// still needs to apply ownerref since it might be changed due to deleteoption update.
				err := helper.ApplyOwnerReferences(ctx, c.client, gvr, existing, owner)
				return existing, err
			}
		}
	}

	obj, err := c.client.
		Resource(gvr).
		Namespace(required.GetNamespace()).
		Apply(ctx, required.GetName(), required, metav1.ApplyOptions{FieldManager: fieldManager, Force: force})
	logger.Info("Server side applied",
		"gvr", gvr.String(), "resourceNamespace", required.GetNamespace(),
		"resourceName", required.GetName(), "fieldManager", fieldManager)

	if apierrors.IsConflict(err) {
		return obj, &ServerSideApplyConflictError{ssaErr: err}
	}

	if err == nil {
		err = helper.ApplyOwnerReferences(ctx, c.client, gvr, obj, owner)
	}

	return obj, err
}

const (
	// DefaultJQExecutionTimeout is the default timeout for jq expression execution
	DefaultJQExecutionTimeout = 1 * time.Second
)

// IgnoreFieldError represents an error that occurred during ignoreFields processing
// This error type is used to distinguish ignoreFields errors from other apply errors
// so they can be reported with the specific AppliedManifestSSAIgnoreFieldError reason
type IgnoreFieldError struct {
	message string
	cause   error
}

func (e *IgnoreFieldError) Error() string {
	return e.message
}

func (e *IgnoreFieldError) Unwrap() error {
	return e.cause
}

// NewIgnoreFieldError creates a new IgnoreFieldError wrapping the given cause
func NewIgnoreFieldError(cause error, format string, args ...interface{}) error {
	return &IgnoreFieldError{
		message: fmt.Sprintf(format, args...),
		cause:   cause,
	}
}

// removeFieldByJSONPath remove the field from object by json path. The json path should not point to a
// list, since removing list from the object and apply would bring unexpected behavior.
func removeFieldByJSONPath(obj interface{}, path string, logger klog.Logger) {
	listKeys := strings.Split(path, ".")
	if len(listKeys) == 0 {
		return
	}
	lastKey := listKeys[len(listKeys)-1]
	pathWithoutLastKey := strings.TrimSuffix(path, "."+lastKey)
	finder := jsonpath.New("ignoreFields").AllowMissingKeys(true)
	if err := finder.Parse(fmt.Sprintf("{%s}", pathWithoutLastKey)); err != nil {
		logger.Error(err, "parse jsonpath", "path", pathWithoutLastKey)
	}
	results, err := finder.FindResults(obj)
	if err != nil {
		logger.Error(err, "find jsonpath", "path", pathWithoutLastKey)
	}
	for _, result := range results {
		for _, r := range result {
			mapResult, ok := r.Interface().(map[string]interface{})
			if !ok {
				continue
			}
			delete(mapResult, lastKey)
		}
	}
}

// removeFieldByJSONPointer removes the field from object using RFC 6901 JSON Pointer.
func removeFieldByJSONPointer(obj *unstructured.Unstructured, pointer string, logger klog.Logger) error {
	// Marshal the object to JSON
	objJSON, err := obj.MarshalJSON()
	if err != nil {
		return fmt.Errorf("failed to marshal object: %w", err)
	}

	// Create a JSON Patch to remove the field
	patchData, err := json.Marshal([]map[string]string{{"op": "remove", "path": pointer}})
	if err != nil {
		return fmt.Errorf("failed to create patch: %w", err)
	}

	patch, err := jsonpatch.DecodePatch(patchData)
	if err != nil {
		return fmt.Errorf("failed to decode patch: %w", err)
	}

	// Apply the patch
	patchedJSON, err := patch.Apply(objJSON)
	if err != nil {
		// Silently ignore if path doesn't exist (using proper error checking)
		if errors.Is(err, jsonpatch.ErrMissing) {
			logger.V(4).Info("JSON Pointer path not found, skipping", "pointer", pointer)
			return nil
		}
		return fmt.Errorf("failed to apply JSON Pointer patch: %w", err)
	}

	// Unmarshal back to the object
	if err := obj.UnmarshalJSON(patchedJSON); err != nil {
		return fmt.Errorf("failed to unmarshal patched object: %w", err)
	}

	return nil
}

// removeFieldByJQExpression removes fields from object using jq path expression.
func removeFieldByJQExpression(ctx context.Context, obj *unstructured.Unstructured, expression string, logger klog.Logger) error {
	// Parse the jq expression wrapped in del()
	jqDeletionQuery, err := gojq.Parse(fmt.Sprintf("del(%s)", expression))
	if err != nil {
		return fmt.Errorf("failed to parse jq expression: %w", err)
	}

	// Compile the query
	jqDeletionCode, err := gojq.Compile(jqDeletionQuery)
	if err != nil {
		return fmt.Errorf("failed to compile jq expression: %w", err)
	}

	// Get object as map
	dataJSON := obj.UnstructuredContent()

	// Execute with timeout derived from parent context
	// This will cancel when EITHER parent context is cancelled OR timeout expires
	timeoutCtx, cancel := context.WithTimeout(ctx, DefaultJQExecutionTimeout)
	defer cancel()

	iter := jqDeletionCode.RunWithContext(timeoutCtx, dataJSON)
	first, ok := iter.Next()
	if !ok {
		return fmt.Errorf("jq expression did not return any data")
	}

	// Check for errors
	if errVal, ok := first.(error); ok {
		if errors.Is(errVal, context.DeadlineExceeded) {
			return fmt.Errorf("jq expression execution timed out after %v", DefaultJQExecutionTimeout)
		}
		if errors.Is(errVal, context.Canceled) {
			return fmt.Errorf("jq expression execution cancelled: %w", errVal)
		}
		return fmt.Errorf("jq expression returned error: %w", errVal)
	}

	// Check for multiple results
	_, ok = iter.Next()
	if ok {
		return fmt.Errorf("jq expression returned multiple objects")
	}

	// Set the modified content back
	if resultMap, ok := first.(map[string]interface{}); ok {
		obj.SetUnstructuredContent(resultMap)
	} else {
		return fmt.Errorf("jq expression result is not a valid object")
	}

	return nil
}

// detect changes in a resource by caching a hash of the string representation of the resource
// note: some changes in a resource e.g. nil vs empty, will not be detected this way
func hashOfResourceStruct(o interface{}) string {
	oString := fmt.Sprintf("%v", o)
	h := md5.New() //nolint:gosec
	if _, err := io.WriteString(h, oString); err != nil {
		return ""
	}
	rval := fmt.Sprintf("%x", h.Sum(nil))
	return rval
}

func removeCreationTimeFromMetadata(obj map[string]interface{}, logger klog.Logger) {
	if metadata, found := obj["metadata"]; found {
		if metaObj, ok := metadata.(map[string]interface{}); ok {
			creationTimestamp, ok := metaObj["creationTimestamp"]
			if ok && creationTimestamp == nil {
				unstructured.RemoveNestedField(metaObj, "creationTimestamp")
			}
		}
	}

	for _, v := range obj {
		switch val := v.(type) {
		case map[string]interface{}:
			removeCreationTimeFromMetadata(val, logger)
		case []interface{}:
			for _, item := range val {
				if itemObj, ok := item.(map[string]interface{}); ok {
					removeCreationTimeFromMetadata(itemObj, logger)
				}
			}
		}
	}
}