2
2
3
3
4
4
import com .github .containersolutions .operator .processing .retry .Retry ;
5
- import com .google .common .util .concurrent .ThreadFactoryBuilder ;
6
5
import io .fabric8 .kubernetes .client .CustomResource ;
7
6
import io .fabric8 .kubernetes .client .KubernetesClientException ;
8
7
import io .fabric8 .kubernetes .client .Watcher ;
11
10
12
11
import java .util .Optional ;
13
12
import java .util .concurrent .ScheduledThreadPoolExecutor ;
14
- import java .util .concurrent .ThreadFactory ;
15
13
import java .util .concurrent .TimeUnit ;
16
14
import java .util .concurrent .locks .ReentrantLock ;
17
15
@@ -48,11 +46,7 @@ public class EventScheduler implements Watcher<CustomResource> {
48
46
public EventScheduler (EventDispatcher eventDispatcher , Retry retry ) {
49
47
this .eventDispatcher = eventDispatcher ;
50
48
this .retry = retry ;
51
- ThreadFactory threadFactory = new ThreadFactoryBuilder ()
52
- .setNameFormat ("event-consumer-%d" )
53
- .setDaemon (false )
54
- .build ();
55
- executor = new ScheduledThreadPoolExecutor (1 , threadFactory );
49
+ executor = new ScheduledThreadPoolExecutor (1 );
56
50
executor .setRemoveOnCancelPolicy (true );
57
51
}
58
52
@@ -61,14 +55,13 @@ public void eventReceived(Watcher.Action action, CustomResource resource) {
61
55
log .debug ("Event received for action: {}, {}: {}" , action .toString ().toLowerCase (), resource .getClass ().getSimpleName (),
62
56
resource .getMetadata ().getName ());
63
57
CustomResourceEvent event = new CustomResourceEvent (action , resource , retry );
64
- scheduleEvent (event );
58
+ scheduleEventFromApi (event );
65
59
}
66
60
67
- void scheduleEvent (CustomResourceEvent event ) {
68
- log .trace ("Current queue size {}" , executor .getQueue ().size ());
69
- log .debug ("Scheduling event: {}" , event );
61
+ void scheduleEventFromApi (CustomResourceEvent event ) {
70
62
try {
71
63
lock .lock ();
64
+ log .debug ("Scheduling event from Api: {}" , event );
72
65
if (event .getResource ().getMetadata ().getDeletionTimestamp () != null && event .getAction () == Action .DELETED ) {
73
66
// Note that we always use finalizers, we want to process delete event just in corner case,
74
67
// when we are not able to add finalizer (lets say because of optimistic locking error, and the resource was deleted instantly).
@@ -89,30 +82,39 @@ void scheduleEvent(CustomResourceEvent event) {
89
82
eventStore .addOrReplaceEventAsNotScheduled (event );
90
83
return ;
91
84
}
85
+ scheduleEventForExecution (event );
86
+ log .trace ("Scheduling event from API finished: {}" , event );
87
+ } finally {
88
+ lock .unlock ();
89
+ }
90
+ }
92
91
92
+ private void scheduleEventForExecution (CustomResourceEvent event ) {
93
+ try {
94
+ lock .lock ();
95
+ log .trace ("Current queue size {}" , executor .getQueue ().size ());
96
+ log .debug ("Scheduling event for execution: {}" , event );
93
97
Optional <Long > nextBackOff = event .nextBackOff ();
94
98
if (!nextBackOff .isPresent ()) {
95
99
log .warn ("Event max retry limit reached. Will be discarded. {}" , event );
96
100
return ;
97
101
}
98
- log .debug ("Creating scheduled task for event: {}" , event );
99
102
eventStore .addEventUnderProcessing (event );
100
103
executor .schedule (new EventConsumer (event , eventDispatcher , this ),
101
104
nextBackOff .get (), TimeUnit .MILLISECONDS );
105
+ log .trace ("Scheduled task for event: {}" , event );
102
106
} finally {
103
- log .debug ("Scheduling event finished: {}" , event );
104
107
lock .unlock ();
105
108
}
106
109
}
107
110
108
111
void eventProcessingFinishedSuccessfully (CustomResourceEvent event ) {
109
112
try {
110
113
lock .lock ();
111
- log .debug ("Event processing successful for event: {}" , event );
112
114
eventStore .removeEventUnderProcessing (event .resourceUid ());
113
115
if (eventStore .containsNotScheduledEvent (event .resourceUid ())) {
114
- log .debug ("Scheduling recent event for processing processing : {}" , event );
115
- scheduleEvent ( eventStore . removeEventNotScheduled ( event .resourceUid () ));
116
+ log .debug ("Scheduling recent event for processing: {}" , event );
117
+ scheduleNotYetScheduledEventForExecution ( event .resourceUid ());
116
118
}
117
119
} finally {
118
120
lock .unlock ();
@@ -124,19 +126,22 @@ void eventProcessingFailed(CustomResourceEvent event) {
124
126
lock .lock ();
125
127
eventStore .removeEventUnderProcessing (event .resourceUid ());
126
128
if (eventStore .containsNotScheduledEvent (event .resourceUid ())) {
127
- CustomResourceEvent notScheduledEvent = eventStore .removeEventNotScheduled (event .resourceUid ());
128
- log .debug ("Event processing failed. Scheduling the most recent event. Failed event: {}," +
129
- " Most recent event: {}" , event , notScheduledEvent );
130
- scheduleEvent (notScheduledEvent );
129
+ log .debug ("Event processing failed. Scheduling the most recent event. Failed event: {}" , event );
130
+ scheduleNotYetScheduledEventForExecution (event .resourceUid ());
131
131
} else {
132
132
log .debug ("Event processing failed. Attempting to re-schedule the event: {}" , event );
133
- scheduleEvent (event );
133
+ scheduleEventForExecution (event );
134
134
}
135
135
} finally {
136
136
lock .unlock ();
137
137
}
138
138
}
139
139
140
+ private void scheduleNotYetScheduledEventForExecution (String uuid ) {
141
+ CustomResourceEvent notScheduledEvent = eventStore .removeEventNotScheduled (uuid );
142
+ scheduleEventForExecution (notScheduledEvent );
143
+ }
144
+
140
145
@ Override
141
146
public void onClose (KubernetesClientException e ) {
142
147
log .error ("Error: " , e );
0 commit comments