github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-03 17:30:00 +00:00
Code Issues Projects Releases Wiki Activity

storage/util: introduce AnnotateInitialEventsEndBookmark function

Browse Source 
// AnnotateInitialEventsEndBookmark adds a special annotation to the given object
// which indicates that the initial events have been sent.
//
// Note that this function assumes that the obj's annotation
// field is a reference type (i.e. a map).
This commit is contained in:
Lukasz Szaszkiewicz 2023-08-24 12:19:10 +02:00
parent 02e51b27a9
commit 47d9a47a08
2 changed files with 20 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
staging/src/k8s.io/apiserver/pkg/storage/cacher/cache_watcher.go
View File

@ -22,7 +22,6 @@ import (
"sync" "sync"
"time" "time"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/runtime/schema"
utilruntime "k8s.io/apimachinery/pkg/util/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@ -365,17 +364,10 @@ func (c *cacheWatcher) convertToWatchEvent(event *watchCacheEvent) *watch.Event
if event.Type == watch.Bookmark { if event.Type == watch.Bookmark {
e := &watch.Event{Type: watch.Bookmark, Object: event.Object.DeepCopyObject()} e := &watch.Event{Type: watch.Bookmark, Object: event.Object.DeepCopyObject()}
if !c.wasBookmarkAfterRvSent() { if !c.wasBookmarkAfterRvSent() {
objMeta, err := meta.Accessor(e.Object) if err := storage.AnnotateInitialEventsEndBookmark(e.Object); err != nil {
if err != nil {
utilruntime.HandleError(fmt.Errorf("error while accessing object's metadata gr: %v, identifier: %v, obj: %#v, err: %v", c.groupResource, c.identifier, e.Object, err)) utilruntime.HandleError(fmt.Errorf("error while accessing object's metadata gr: %v, identifier: %v, obj: %#v, err: %v", c.groupResource, c.identifier, e.Object, err))
return nil return nil
} }
objAnnotations := objMeta.GetAnnotations()
if objAnnotations == nil {
objAnnotations = map[string]string{}
}
objAnnotations["k8s.io/initial-events-end"] = "true"
objMeta.SetAnnotations(objAnnotations)
} }
return e return e
} }

19
staging/src/k8s.io/apiserver/pkg/storage/util.go
View File

@ -122,3 +122,22 @@ func GetCurrentResourceVersionFromStorage(ctx context.Context, storage Interface
} }
return uint64(currentResourceVersion), nil return uint64(currentResourceVersion), nil
} }
// AnnotateInitialEventsEndBookmark adds a special annotation to the given object
// which indicates that the initial events have been sent.
//
// Note that this function assumes that the obj's annotation
// field is a reference type (i.e. a map).
func AnnotateInitialEventsEndBookmark(obj runtime.Object) error {
objMeta, err := meta.Accessor(obj)
if err != nil {
return err
}
objAnnotations := objMeta.GetAnnotations()
if objAnnotations == nil {
objAnnotations = map[string]string{}
}
objAnnotations["k8s.io/initial-events-end"] = "true"
objMeta.SetAnnotations(objAnnotations)
return nil
}