Respect annotation size limit for SSA last-applied.

To support CSA and SSA interoperability, SSA updates the CSA
last-applied annotation.

This change ensures we don't set a big last-applied annotation if the
value is over the annotation limits.

Also, make sure that it's possible to opt-out of this behavior by
setting the CSA annotation to "" the empty string.

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/lastappliedupdater.go

@@ -25,6 +25,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
+const totalAnnotationSizeLimitB int64 = 256 * (1 << 10) // 256 kB
+
 type lastAppliedUpdater struct {
 	fieldManager Manager
 }
@@ -78,8 +80,8 @@ func hasLastApplied(obj runtime.Object) bool {
 	if annotations == nil {
 		return false
 	}
-	_, ok := annotations[corev1.LastAppliedConfigAnnotation]
-	return ok
+	lastApplied, ok := annotations[corev1.LastAppliedConfigAnnotation]
+	return ok && len(lastApplied) > 0
 }
 
 func setLastApplied(obj runtime.Object, value string) error {
@@ -92,6 +94,9 @@ func setLastApplied(obj runtime.Object, value string) error {
 		annotations = map[string]string{}
 	}
 	annotations[corev1.LastAppliedConfigAnnotation] = value
+	if isAnnotationsValid(annotations) != nil {
+		delete(annotations, corev1.LastAppliedConfigAnnotation)
+	}
 	accessor.SetAnnotations(annotations)
 	return nil
 }
@@ -115,3 +120,14 @@ func buildLastApplied(obj runtime.Object) (string, error) {
 	}
 	return string(lastApplied), nil
 }
+
+func isAnnotationsValid(annotations map[string]string) error {
+	var totalSize int64
+	for k, v := range annotations {
+		totalSize += (int64)(len(k)) + (int64)(len(v))
+	}
+	if totalSize > (int64)(totalAnnotationSizeLimitB) {
+		return fmt.Errorf("annotations size %d is larger than limit %d", totalSize, totalAnnotationSizeLimitB)
+	}
+	return nil
+}

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager/lastappliedupdater_test.go

@@ -17,9 +17,13 @@ limitations under the License.
 package fieldmanager
 
 import (
+	"fmt"
 	"strings"
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/yaml"
@@ -89,3 +93,134 @@ spec:
 		t.Errorf("expected last applied annotation to be updated, but got: %q", lastApplied)
 	}
 }
+
+func TestLargeLastApplied(t *testing.T) {
+	tests := []struct {
+		name      string
+		oldObject *corev1.ConfigMap
+		newObject *corev1.ConfigMap
+	}{
+		{
+			name: "old object + new object last-applied annotation is too big",
+			oldObject: &corev1.ConfigMap{
+				TypeMeta: metav1.TypeMeta{
+					APIVersion: "v1",
+					Kind:       "ConfigMap",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "large-update-test-cm",
+					Namespace: "default",
+					Annotations: map[string]string{
+						corev1.LastAppliedConfigAnnotation: "nonempty",
+					},
+				},
+				Data: map[string]string{"k": "v"},
+			},
+			newObject: func() *corev1.ConfigMap {
+				cfg := &corev1.ConfigMap{
+					TypeMeta: metav1.TypeMeta{
+						APIVersion: "v1",
+						Kind:       "ConfigMap",
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "large-update-test-cm",
+						Namespace: "default",
+						Annotations: map[string]string{
+							corev1.LastAppliedConfigAnnotation: "nonempty",
+						},
+					},
+					Data: map[string]string{"k": "v"},
+				}
+				for i := 0; i < 9999; i++ {
+					unique := fmt.Sprintf("this-key-is-very-long-so-as-to-create-a-very-large-serialized-fieldset-%v", i)
+					cfg.Data[unique] = "A"
+				}
+				return cfg
+			}(),
+		},
+		{
+			name: "old object + new object annotations + new object last-applied annotation is too big",
+			oldObject: func() *corev1.ConfigMap {
+				cfg := &corev1.ConfigMap{
+					TypeMeta: metav1.TypeMeta{
+						APIVersion: "v1",
+						Kind:       "ConfigMap",
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "large-update-test-cm",
+						Namespace: "default",
+						Annotations: map[string]string{
+							corev1.LastAppliedConfigAnnotation: "nonempty",
+						},
+					},
+					Data: map[string]string{"k": "v"},
+				}
+				for i := 0; i < 2000; i++ {
+					unique := fmt.Sprintf("this-key-is-very-long-so-as-to-create-a-very-large-serialized-fieldset-%v", i)
+					cfg.Data[unique] = "A"
+				}
+				return cfg
+			}(),
+			newObject: func() *corev1.ConfigMap {
+				cfg := &corev1.ConfigMap{
+					TypeMeta: metav1.TypeMeta{
+						APIVersion: "v1",
+						Kind:       "ConfigMap",
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "large-update-test-cm",
+						Namespace: "default",
+						Annotations: map[string]string{
+							corev1.LastAppliedConfigAnnotation: "nonempty",
+						},
+					},
+					Data: map[string]string{"k": "v"},
+				}
+				for i := 0; i < 2000; i++ {
+					unique := fmt.Sprintf("this-key-is-very-long-so-as-to-create-a-very-large-serialized-fieldset-%v", i)
+					cfg.Data[unique] = "A"
+					cfg.ObjectMeta.Annotations[unique] = "A"
+				}
+				return cfg
+			}(),
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			f := NewTestFieldManager(schema.FromAPIVersionAndKind("v1", "ConfigMap"),
+				false,
+				func(m Manager) Manager {
+					return NewLastAppliedUpdater(m)
+				})
+
+			if err := f.Apply(test.oldObject, "kubectl", false); err != nil {
+				t.Errorf("Error applying object: %v", err)
+			}
+
+			lastApplied, err := getLastApplied(f.liveObj)
+			if err != nil {
+				t.Errorf("Failed to access last applied annotation: %v", err)
+			}
+			if len(lastApplied) == 0 || lastApplied == "nonempty" {
+				t.Errorf("Expected an updated last-applied annotation, but got: %q", lastApplied)
+			}
+
+			if err := f.Apply(test.newObject, "kubectl", false); err != nil {
+				t.Errorf("Error applying object: %v", err)
+			}
+
+			accessor := meta.NewAccessor()
+			annotations, err := accessor.Annotations(f.liveObj)
+			if err != nil {
+				t.Errorf("Failed to access annotations: %v", err)
+			}
+			if annotations == nil {
+				t.Errorf("No annotations on obj: %v", f.liveObj)
+			}
+			lastApplied, ok := annotations[corev1.LastAppliedConfigAnnotation]
+			if ok || len(lastApplied) > 0 {
+				t.Errorf("Expected no last applied annotation, but got last applied with length: %d", len(lastApplied))
+			}
+		})
+	}
+}