Skip to content

Commit 99d578f

Browse files
enarhaenarha
authored
Use server-side apply for managing annotations (#1411)
This is part of broader effort to switch all Tekton components to using Server-Side Apply patch to improve performance (less retries when merge patches are combined with resourceVersion) and reliability (when merge patches are used without resourceVersion). The underlying problem is that in the case of PipelineRun and TaskRun objects, there are multiple controllers operating on those objects and that leads to issues such as one controller overriding the changes (annotations, labels) of other controllers.
1 parent 3c34df5 commit 99d578f

7 files changed

Lines changed: 463 additions & 28 deletions

File tree

‎pkg/chains/annotations.go‎

Lines changed: 41 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import (
1717
"context"
1818
"fmt"
1919
"strconv"
20+
"strings"
2021

2122
"github.com/pkg/errors"
2223
"github.com/tektoncd/chains/pkg/chains/objects"
@@ -26,6 +27,8 @@ import (
2627
)
2728

2829
const (
30+
// ChainsAnnotationPrefix is the prefix for all Chains annotations
31+
ChainsAnnotationPrefix = "chains.tekton.dev/"
2932
// ChainsAnnotation is the standard annotation to indicate a TR has been signed.
3033
ChainsAnnotation = "chains.tekton.dev/signed"
3134
RetryAnnotation = "chains.tekton.dev/retries"
@@ -63,6 +66,10 @@ func reconciledFromAnnotations(annotations map[string]string) bool {
6366
// MarkSigned marks a Tekton object as signed.
6467
func MarkSigned(ctx context.Context, obj objects.TektonObject, ps versioned.Interface, annotations map[string]string) error {
6568
if _, ok := obj.GetAnnotations()[ChainsAnnotation]; ok {
69+
// Object is already signed, but we may still need to apply additional annotations
70+
if len(annotations) > 0 {
71+
return AddAnnotation(ctx, obj, ps, ChainsAnnotation, "true", annotations)
72+
}
6673
return nil
6774
}
6875
return AddAnnotation(ctx, obj, ps, ChainsAnnotation, "true", annotations)
@@ -97,18 +104,47 @@ func AddRetry(ctx context.Context, obj objects.TektonObject, ps versioned.Interf
97104
}
98105

99106
func AddAnnotation(ctx context.Context, obj objects.TektonObject, ps versioned.Interface, key, value string, annotations map[string]string) error {
100-
// Use patch instead of update to help prevent race conditions.
101-
if annotations == nil {
102-
annotations = map[string]string{}
107+
// Get current annotations from API server to ensure we have the latest state
108+
currentAnnotations, err := obj.GetLatestAnnotations(ctx, ps)
109+
if err != nil {
110+
return err
111+
}
112+
113+
// Start with existing chains annotations, ignore annotations from other controllers,
114+
// so we do not take ownership of them.
115+
mergedAnnotations := make(map[string]string)
116+
for k, v := range currentAnnotations {
117+
if strings.HasPrefix(k, ChainsAnnotationPrefix) {
118+
mergedAnnotations[k] = v
119+
}
120+
}
121+
122+
// Add the new chains annotations, they all must be chains annotations
123+
for k, v := range annotations {
124+
if !strings.HasPrefix(k, ChainsAnnotationPrefix) {
125+
return fmt.Errorf("invalid annotation key %q: all annotations must have prefix %q", k, ChainsAnnotationPrefix)
126+
}
127+
mergedAnnotations[k] = v
103128
}
104-
annotations[key] = value
105-
patchBytes, err := patch.GetAnnotationsPatch(annotations)
129+
130+
// Add the specific key-value pair, again it must be chains annotation
131+
if !strings.HasPrefix(key, ChainsAnnotationPrefix) {
132+
return fmt.Errorf("invalid annotation key %q: all annotations must have prefix %q", key, ChainsAnnotationPrefix)
133+
}
134+
mergedAnnotations[key] = value
135+
136+
patchBytes, err := patch.GetAnnotationsPatch(mergedAnnotations, obj)
106137
if err != nil {
107138
return err
108139
}
109140
err = obj.Patch(ctx, ps, patchBytes)
110141
if err != nil {
111142
return err
112143
}
144+
145+
// Note: Ideally here we'll update the in-memory object to keep it consistent through
146+
// the reconciliation loop. It hasn't been done to preserve the existing controller behavior
147+
// and maintain compatibility with existing tests. This could be revisited in the future.
148+
113149
return nil
114150
}

‎pkg/chains/annotations_test.go‎

Lines changed: 158 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -183,7 +183,7 @@ func TestMarkSigned(t *testing.T) {
183183

184184
// Now mark it as signed.
185185
extra := map[string]string{
186-
"foo": "bar",
186+
"chains.tekton.dev/extra": "bar",
187187
}
188188

189189
if err := MarkSigned(ctx, tt.object, c, extra); err != nil {
@@ -198,7 +198,7 @@ func TestMarkSigned(t *testing.T) {
198198
if _, ok := signed.GetAnnotations()[ChainsAnnotation]; !ok {
199199
t.Error("Object not signed.")
200200
}
201-
if signed.GetAnnotations()["foo"] != "bar" {
201+
if signed.GetAnnotations()["chains.tekton.dev/extra"] != "bar" {
202202
t.Error("Extra annotations not applied")
203203
}
204204
})
@@ -372,3 +372,159 @@ func TestAddRetry(t *testing.T) {
372372
})
373373
}
374374
}
375+
376+
// TestAddAnnotationValidation tests the new validation logic for annotations
377+
func TestAddAnnotationValidation(t *testing.T) {
378+
tests := []struct {
379+
name string
380+
key string
381+
value string
382+
annotations map[string]string
383+
wantErr bool
384+
wantErrContains string
385+
}{
386+
{
387+
name: "valid chains annotation key",
388+
key: "chains.tekton.dev/test",
389+
value: "value",
390+
wantErr: false,
391+
},
392+
{
393+
name: "invalid annotation key without prefix",
394+
key: "invalid-key",
395+
value: "value",
396+
wantErr: true,
397+
wantErrContains: "invalid annotation key",
398+
},
399+
{
400+
name: "invalid annotation in map without prefix",
401+
key: "chains.tekton.dev/test",
402+
value: "value",
403+
annotations: map[string]string{"invalid": "value"},
404+
wantErr: true,
405+
wantErrContains: "invalid annotation key",
406+
},
407+
{
408+
name: "valid annotations with prefix",
409+
key: "chains.tekton.dev/test",
410+
value: "value",
411+
annotations: map[string]string{
412+
"chains.tekton.dev/extra1": "value1",
413+
"chains.tekton.dev/extra2": "value2",
414+
},
415+
wantErr: false,
416+
},
417+
{
418+
name: "mixed valid and invalid annotations",
419+
key: "chains.tekton.dev/test",
420+
value: "value",
421+
annotations: map[string]string{
422+
"chains.tekton.dev/extra1": "value1",
423+
"invalid": "value2",
424+
},
425+
wantErr: true,
426+
wantErrContains: "invalid annotation key",
427+
},
428+
}
429+
430+
for _, tt := range tests {
431+
t.Run(tt.name, func(t *testing.T) {
432+
ctx, _ := rtesting.SetupFakeContext(t)
433+
c := fakepipelineclient.Get(ctx)
434+
435+
obj := objects.NewTaskRunObjectV1(&v1.TaskRun{
436+
ObjectMeta: metav1.ObjectMeta{
437+
Name: "test-taskrun",
438+
Namespace: "default",
439+
},
440+
})
441+
442+
tekton.CreateObject(t, ctx, c, obj)
443+
444+
err := AddAnnotation(ctx, obj, c, tt.key, tt.value, tt.annotations)
445+
446+
if tt.wantErr {
447+
if err == nil {
448+
t.Errorf("AddAnnotation() expected error but got none")
449+
} else if tt.wantErrContains != "" && !contains(err.Error(), tt.wantErrContains) {
450+
t.Errorf("AddAnnotation() error = %v, wantErrContains %v", err, tt.wantErrContains)
451+
}
452+
} else {
453+
if err != nil {
454+
t.Errorf("AddAnnotation() unexpected error = %v", err)
455+
}
456+
}
457+
})
458+
}
459+
}
460+
461+
// TestAnnotationPreservation tests that only Chains annotations are preserved
462+
func TestAnnotationPreservation(t *testing.T) {
463+
ctx, _ := rtesting.SetupFakeContext(t)
464+
c := fakepipelineclient.Get(ctx)
465+
466+
// Create object with mixed annotations
467+
obj := objects.NewTaskRunObjectV1(&v1.TaskRun{
468+
ObjectMeta: metav1.ObjectMeta{
469+
Name: "test-taskrun",
470+
Namespace: "default",
471+
Annotations: map[string]string{
472+
"chains.tekton.dev/existing": "keep-me",
473+
"tekton.dev/other": "ignore-me",
474+
"kubernetes.io/annotation": "ignore-me",
475+
"chains.tekton.dev/another": "keep-me-too",
476+
},
477+
},
478+
})
479+
480+
tekton.CreateObject(t, ctx, c, obj)
481+
482+
// Add a new annotation
483+
err := AddAnnotation(ctx, obj, c, "chains.tekton.dev/new", "new-value", nil)
484+
if err != nil {
485+
t.Fatalf("AddAnnotation() error = %v", err)
486+
}
487+
488+
// Verify the result
489+
updated, err := tekton.GetObject(t, ctx, c, obj)
490+
if err != nil {
491+
t.Fatalf("Get() error = %v", err)
492+
}
493+
494+
annotations := updated.GetAnnotations()
495+
496+
// Should have all chains annotations
497+
expectedChains := map[string]string{
498+
"chains.tekton.dev/existing": "keep-me",
499+
"chains.tekton.dev/another": "keep-me-too",
500+
"chains.tekton.dev/new": "new-value",
501+
}
502+
503+
for k, v := range expectedChains {
504+
if got := annotations[k]; got != v {
505+
t.Errorf("Expected annotation %s=%s, got %s", k, v, got)
506+
}
507+
}
508+
509+
// Should still have non-chains annotations (they weren't removed, just not included in patch)
510+
expectedNonChains := map[string]string{
511+
"tekton.dev/other": "ignore-me",
512+
"kubernetes.io/annotation": "ignore-me",
513+
}
514+
515+
for k, v := range expectedNonChains {
516+
if got := annotations[k]; got != v {
517+
t.Errorf("Expected non-chains annotation %s=%s to be preserved, got %s", k, v, got)
518+
}
519+
}
520+
}
521+
522+
// Helper function to check if string contains substring
523+
func contains(s, substr string) bool {
524+
for i := 0; i <= len(s)-len(substr); i++ {
525+
if s[i:i+len(substr)] == substr {
526+
return true
527+
}
528+
}
529+
return false
530+
}

‎pkg/chains/objects/objects.go‎

Lines changed: 33 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,15 +22,24 @@ import (
2222

2323
v1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1"
2424
"github.com/tektoncd/pipeline/pkg/client/clientset/versioned"
25+
apierrors "k8s.io/apimachinery/pkg/api/errors"
2526
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
2627
"k8s.io/apimachinery/pkg/runtime"
2728
"k8s.io/apimachinery/pkg/types"
2829
"knative.dev/pkg/apis"
30+
"knative.dev/pkg/logging"
31+
"knative.dev/pkg/ptr"
2932
)
3033

3134
// Label added to TaskRuns identifying the associated pipeline Task
3235
const PipelineTaskLabel = "tekton.dev/pipelineTask"
3336

37+
// patchOptions contains the default patch options
38+
var patchOptions = metav1.PatchOptions{
39+
FieldManager: "tekton-chains-controller",
40+
Force: ptr.Bool(false),
41+
}
42+
3443
// Object is used as a base object of all Kubernetes objects
3544
// ref: https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.9.4/pkg/client#Object
3645
type Object interface {
@@ -121,8 +130,19 @@ func (tro *TaskRunObjectV1) GetObject() interface{} {
121130

122131
// Patch the original TaskRun object
123132
func (tro *TaskRunObjectV1) Patch(ctx context.Context, clientSet versioned.Interface, patchBytes []byte) error {
133+
logger := logging.FromContext(ctx)
124134
_, err := clientSet.TektonV1().TaskRuns(tro.Namespace).Patch(
125-
ctx, tro.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
135+
ctx, tro.Name, types.ApplyPatchType, patchBytes, patchOptions)
136+
if apierrors.IsConflict(err) {
137+
// Since we only update the list of annotations we manage, there shouldn't be any conflicts unless
138+
// another controller/client is updating our annotations. We log the issue and force patch.
139+
logger.Warnf("failed to patch object %s/%s due to Server-Side Apply patch conflict, using force patch.", tro.Namespace, tro.Name)
140+
// use a copy to avoid changing the global var
141+
patchOptionsForce := patchOptions
142+
patchOptionsForce.Force = ptr.Bool(true)
143+
_, err = clientSet.TektonV1().TaskRuns(tro.Namespace).Patch(
144+
ctx, tro.Name, types.ApplyPatchType, patchBytes, patchOptionsForce)
145+
}
126146
return err
127147
}
128148

@@ -260,8 +280,19 @@ func (pro *PipelineRunObjectV1) GetObject() interface{} {
260280

261281
// Patch the original PipelineRun object
262282
func (pro *PipelineRunObjectV1) Patch(ctx context.Context, clientSet versioned.Interface, patchBytes []byte) error {
283+
logger := logging.FromContext(ctx)
263284
_, err := clientSet.TektonV1().PipelineRuns(pro.Namespace).Patch(
264-
ctx, pro.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
285+
ctx, pro.Name, types.ApplyPatchType, patchBytes, patchOptions)
286+
if apierrors.IsConflict(err) {
287+
// Since we only update the list of annotations we manage, there shouldn't be any conflicts unless
288+
// another controller/client is updating our annotations. We log the issue and force patch.
289+
logger.Warnf("failed to patch object %s/%s due to Server-Side Apply patch conflict, using force patch.", pro.Namespace, pro.Name)
290+
// use a copy to avoid changing the global var
291+
patchOptionsForce := patchOptions
292+
patchOptionsForce.Force = ptr.Bool(true)
293+
_, err = clientSet.TektonV1().PipelineRuns(pro.Namespace).Patch(
294+
ctx, pro.Name, types.ApplyPatchType, patchBytes, patchOptionsForce)
295+
}
265296
return err
266297
}
267298

‎pkg/chains/storage/tekton/tekton.go‎

Lines changed: 30 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import (
1717
"context"
1818
"encoding/base64"
1919
"fmt"
20+
"strings"
2021

2122
intoto "github.com/in-toto/attestation/go/v1"
2223
"github.com/tektoncd/chains/pkg/chains/objects"
@@ -30,6 +31,8 @@ import (
3031
)
3132

3233
const (
34+
// ChainsAnnotationPrefix is the prefix for all Chains annotations
35+
ChainsAnnotationPrefix = "chains.tekton.dev/"
3336
StorageBackendTekton = "tekton"
3437
PayloadAnnotationFormat = "chains.tekton.dev/payload-%s"
3538
SignatureAnnotationFormat = "chains.tekton.dev/signature-%s"
@@ -168,13 +171,28 @@ func (s *Storer) Store(ctx context.Context, req *api.StoreRequest[objects.Tekton
168171
if key == "" {
169172
key = string(obj.GetUID())
170173
}
171-
patchBytes, err := patch.GetAnnotationsPatch(map[string]string{
172-
// Base64 encode both the signature and the payload
173-
fmt.Sprintf(PayloadAnnotationFormat, key): base64.StdEncoding.EncodeToString(req.Bundle.Content),
174-
fmt.Sprintf(SignatureAnnotationFormat, key): base64.StdEncoding.EncodeToString(req.Bundle.Signature),
175-
fmt.Sprintf(CertAnnotationsFormat, key): base64.StdEncoding.EncodeToString(req.Bundle.Cert),
176-
fmt.Sprintf(ChainAnnotationFormat, key): base64.StdEncoding.EncodeToString(req.Bundle.Chain),
177-
})
174+
175+
// Get current annotations from API server to ensure we have the latest state
176+
currentAnnotations, err := obj.GetLatestAnnotations(ctx, s.client)
177+
if err != nil {
178+
return nil, err
179+
}
180+
181+
// Merge existing annotations with new Chains annotations
182+
mergedAnnotations := make(map[string]string)
183+
for k, v := range currentAnnotations {
184+
if strings.HasPrefix(k, ChainsAnnotationPrefix) {
185+
mergedAnnotations[k] = v
186+
}
187+
}
188+
189+
// Add Chains-specific annotations
190+
mergedAnnotations[fmt.Sprintf(PayloadAnnotationFormat, key)] = base64.StdEncoding.EncodeToString(req.Bundle.Content)
191+
mergedAnnotations[fmt.Sprintf(SignatureAnnotationFormat, key)] = base64.StdEncoding.EncodeToString(req.Bundle.Signature)
192+
mergedAnnotations[fmt.Sprintf(CertAnnotationsFormat, key)] = base64.StdEncoding.EncodeToString(req.Bundle.Cert)
193+
mergedAnnotations[fmt.Sprintf(ChainAnnotationFormat, key)] = base64.StdEncoding.EncodeToString(req.Bundle.Chain)
194+
195+
patchBytes, err := patch.GetAnnotationsPatch(mergedAnnotations, obj)
178196
if err != nil {
179197
return nil, err
180198
}
@@ -183,5 +201,10 @@ func (s *Storer) Store(ctx context.Context, req *api.StoreRequest[objects.Tekton
183201
if patchErr != nil {
184202
return nil, patchErr
185203
}
204+
205+
// Note: Ideally here we'll update the in-memory object to keep it consistent through
206+
// the reconciliation loop. It hasn't been done to preserve the existing controller behavior
207+
// and maintain compatibility with existing tests. This could be revisited in the future.
208+
186209
return &api.StoreResponse{}, nil
187210
}

0 commit comments

Comments
 (0)