11package store
22
33import (
4+ "context"
45 "fmt"
56 "strconv"
67 "time"
@@ -27,7 +28,7 @@ type baseSourceStore struct {
2728 receivedWorks workqueue.TypedRateLimitingInterface [* workv1.ManifestWork ]
2829}
2930
30- func (bs * baseSourceStore ) HandleReceivedResource (action types.ResourceAction , work * workv1.ManifestWork ) error {
31+ func (bs * baseSourceStore ) HandleReceivedResource (_ context. Context , action types.ResourceAction , work * workv1.ManifestWork ) error {
3132 switch action {
3233 case types .StatusModified :
3334 bs .receivedWorks .Add (work )
@@ -50,26 +51,26 @@ func newWorkProcessor(works workqueue.TypedRateLimitingInterface[*workv1.Manifes
5051 }
5152}
5253
53- func (b * workProcessor ) run (stopCh <- chan struct {} ) {
54+ func (b * workProcessor ) run (ctx context. Context ) {
5455 defer b .works .ShutDown ()
5556
5657 // start a goroutine to handle the works from the queue
5758 // the .Until will re-kick the runWorker one second after the runWorker completes
58- go wait .Until ( b .runWorker , time .Second , stopCh )
59+ go wait .UntilWithContext ( ctx , b .runWorker , time .Second )
5960
6061 // wait until we're told to stop
61- <- stopCh
62+ <- ctx . Done ()
6263}
6364
64- func (b * workProcessor ) runWorker () {
65+ func (b * workProcessor ) runWorker (ctx context. Context ) {
6566 // hot loop until we're told to stop. processNextEvent will automatically wait until there's work available, so
6667 // we don't worry about secondary waits
67- for b .processNextWork () {
68+ for b .processNextWork (ctx ) {
6869 }
6970}
7071
7172// processNextWork deals with one key off the queue.
72- func (b * workProcessor ) processNextWork () bool {
73+ func (b * workProcessor ) processNextWork (ctx context. Context ) bool {
7374 // pull the next event item from queue.
7475 // events queue blocks until it can return an item to be processed
7576 key , quit := b .works .Get ()
@@ -79,7 +80,7 @@ func (b *workProcessor) processNextWork() bool {
7980 }
8081 defer b .works .Done (key )
8182
82- if err := b .handleWork (key ); err != nil {
83+ if err := b .handleWork (ctx , key ); err != nil {
8384 // we failed to handle the work, we should requeue the item to work on later
8485 // this method will add a backoff to avoid hotlooping on particular items
8586 b .works .AddRateLimited (key )
@@ -91,8 +92,9 @@ func (b *workProcessor) processNextWork() bool {
9192 return true
9293}
9394
94- func (b * workProcessor ) handleWork (work * workv1.ManifestWork ) error {
95- lastWork := b .getWork (work .UID )
95+ func (b * workProcessor ) handleWork (ctx context.Context , work * workv1.ManifestWork ) error {
96+ logger := klog .FromContext (ctx ).WithValues ("manifestWorkNamespace" , work .Namespace , "manifestWorkName" , work .Name )
97+ lastWork := b .getWork (ctx , work .UID )
9698 if lastWork == nil {
9799 // the work is not found from the local cache and it has been deleted by the agent,
98100 // ignore this work.
@@ -116,20 +118,20 @@ func (b *workProcessor) handleWork(work *workv1.ManifestWork) error {
116118
117119 lastResourceVersion , err := strconv .Atoi (lastWork .ResourceVersion )
118120 if err != nil {
119- klog . Errorf ( "invalid resource version for work %s/%s, %v" , lastWork . Namespace , lastWork . Name , err )
121+ logger . Error ( err , "invalid resource version for work" )
120122 return nil
121123 }
122124
123125 resourceVersion , err := strconv .Atoi (work .ResourceVersion )
124126 if err != nil {
125- klog . Errorf ( "invalid resource version for work %s/%s, %v" , lastWork . Namespace , lastWork . Name , err )
127+ logger . Error ( err , "invalid resource version for work" )
126128 return nil
127129 }
128130
129131 // the current work's version is maintained on source and the agent's work is newer than source, ignore
130132 if lastResourceVersion != 0 && resourceVersion > lastResourceVersion {
131- klog . Warningf ("the work %s/%s resource version %d is great than its generation %d , ignore" ,
132- lastWork . Namespace , lastWork . Name , resourceVersion , lastResourceVersion )
133+ logger . Info ("the work resource version is great than its generation, ignore" ,
134+ "agentResourceVersion" , resourceVersion , "sourceResourceVersion" , lastResourceVersion )
133135 return nil
134136 }
135137
@@ -140,13 +142,13 @@ func (b *workProcessor) handleWork(work *workv1.ManifestWork) error {
140142 sequenceID := work .Annotations [common .CloudEventsSequenceIDAnnotationKey ]
141143 greater , err := utils .CompareSnowflakeSequenceIDs (lastSequenceID , sequenceID )
142144 if err != nil {
143- klog . Errorf ( "invalid sequenceID for work %s/%s, %v" , lastWork . Namespace , lastWork . Name , err )
145+ logger . Error ( err , "invalid sequenceID for work" )
144146 return nil
145147 }
146148
147149 if ! greater {
148- klog . Warningf ("the work %s/%s current sequenceID %s is less than its last %s , ignore" ,
149- lastWork . Namespace , lastWork . Name , sequenceID , lastSequenceID )
150+ logger . Info ("the work current sequenceID is less than its last, ignore" ,
151+ "currentSequenceID" , sequenceID , "lastSequenceID" , lastSequenceID )
150152 return nil
151153 }
152154
@@ -163,10 +165,11 @@ func (b *workProcessor) handleWork(work *workv1.ManifestWork) error {
163165 return b .store .Update (updatedWork )
164166}
165167
166- func (b * workProcessor ) getWork (uid kubetypes.UID ) * workv1.ManifestWork {
168+ func (b * workProcessor ) getWork (ctx context.Context , uid kubetypes.UID ) * workv1.ManifestWork {
169+ logger := klog .FromContext (ctx )
167170 works , err := b .store .ListAll ()
168171 if err != nil {
169- klog . Errorf ( "failed to lists works, %v" , err )
172+ logger . Error ( err , "failed to lists works" )
170173 return nil
171174 }
172175
0 commit comments