Skip to content

Commit 95450aa

Browse files
csviricsviri
committed
merged master, google formatter
1 parent 6b9f071 commit 95450aa

File tree

13 files changed

+719
-666
lines changed

13 files changed

+719
-666
lines changed

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/Operator.java

Lines changed: 37 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,6 @@
1818
import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
1919
import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
2020
import io.javaoperatorsdk.operator.processing.retry.Retry;
21-
import org.slf4j.Logger;
22-
import org.slf4j.LoggerFactory;
23-
2421
import java.util.Arrays;
2522
import java.util.HashMap;
2623
import java.util.Map;
@@ -39,32 +36,45 @@ public Operator(KubernetesClient k8sClient) {
3936
this.k8sClient = k8sClient;
4037
}
4138

39+
public <R extends CustomResource> void registerControllerForAllNamespaces(
40+
ResourceController<R> controller) throws OperatorException {
41+
registerController(controller, true, null);
42+
}
4243

43-
public <R extends CustomResource> void registerControllerForAllNamespaces(ResourceController<R> controller) throws OperatorException {
44-
registerController(controller, true, null);
45-
}
46-
47-
public <R extends CustomResource> void registerController(ResourceController<R> controller, String... targetNamespaces) throws OperatorException {
48-
registerController(controller, false, null, targetNamespaces);
49-
}
44+
public <R extends CustomResource> void registerController(
45+
ResourceController<R> controller, String... targetNamespaces) throws OperatorException {
46+
registerController(controller, false, null, targetNamespaces);
47+
}
5048

51-
@SuppressWarnings("rawtypes")
52-
private <R extends CustomResource> void registerController(ResourceController<R> controller,
53-
boolean watchAllNamespaces, Retry retry, String... targetNamespaces) throws OperatorException {
54-
Class<R> resClass = getCustomResourceClass(controller);
55-
CustomResourceDefinitionContext crd = getCustomResourceDefinitionForController(controller);
56-
KubernetesDeserializer.registerCustomKind(crd.getVersion(), crd.getKind(), resClass);
57-
String finalizer = ControllerUtils.getFinalizer(controller);
58-
MixedOperation client = k8sClient.customResources(crd, resClass, CustomResourceList.class, ControllerUtils.getCustomResourceDoneableClass(controller));
59-
EventDispatcher eventDispatcher = new EventDispatcher(controller,
60-
finalizer, new EventDispatcher.CustomResourceFacade(client));
61-
62-
63-
CustomResourceCache customResourceCache = new CustomResourceCache();
64-
DefaultEventHandler defaultEventHandler = new DefaultEventHandler(customResourceCache, eventDispatcher, controller.getClass().getName(), retry);
65-
DefaultEventSourceManager eventSourceManager = new DefaultEventSourceManager(defaultEventHandler, retry != null);
66-
defaultEventHandler.setDefaultEventSourceManager(eventSourceManager);
67-
eventDispatcher.setEventSourceManager(eventSourceManager);
49+
@SuppressWarnings("rawtypes")
50+
private <R extends CustomResource> void registerController(
51+
ResourceController<R> controller,
52+
boolean watchAllNamespaces,
53+
Retry retry,
54+
String... targetNamespaces)
55+
throws OperatorException {
56+
Class<R> resClass = getCustomResourceClass(controller);
57+
CustomResourceDefinitionContext crd = getCustomResourceDefinitionForController(controller);
58+
KubernetesDeserializer.registerCustomKind(crd.getVersion(), crd.getKind(), resClass);
59+
String finalizer = ControllerUtils.getFinalizer(controller);
60+
MixedOperation client =
61+
k8sClient.customResources(
62+
crd,
63+
resClass,
64+
CustomResourceList.class,
65+
ControllerUtils.getCustomResourceDoneableClass(controller));
66+
EventDispatcher eventDispatcher =
67+
new EventDispatcher(
68+
controller, finalizer, new EventDispatcher.CustomResourceFacade(client));
69+
70+
CustomResourceCache customResourceCache = new CustomResourceCache();
71+
DefaultEventHandler defaultEventHandler =
72+
new DefaultEventHandler(
73+
customResourceCache, eventDispatcher, controller.getClass().getName(), retry);
74+
DefaultEventSourceManager eventSourceManager =
75+
new DefaultEventSourceManager(defaultEventHandler, retry != null);
76+
defaultEventHandler.setDefaultEventSourceManager(eventSourceManager);
77+
eventDispatcher.setEventSourceManager(eventSourceManager);
6878

6979
customResourceClients.put(resClass, (CustomResourceOperationsImpl) client);
7080

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java

Lines changed: 103 additions & 94 deletions
Original file line numberDiff line numberDiff line change
@@ -8,25 +8,18 @@
88
import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
99
import io.javaoperatorsdk.operator.processing.event.Event;
1010
import io.javaoperatorsdk.operator.processing.event.EventHandler;
11-
import java.util.HashSet;
12-
import java.util.Optional;
13-
import java.util.Set;
1411
import io.javaoperatorsdk.operator.processing.retry.Retry;
1512
import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
16-
import org.slf4j.Logger;
17-
import org.slf4j.LoggerFactory;
18-
1913
import java.util.*;
14+
import java.util.HashSet;
15+
import java.util.Optional;
16+
import java.util.Set;
2017
import java.util.concurrent.ScheduledThreadPoolExecutor;
2118
import java.util.concurrent.ThreadFactory;
2219
import java.util.concurrent.locks.ReentrantLock;
2320
import org.slf4j.Logger;
2421
import org.slf4j.LoggerFactory;
2522

26-
import static io.javaoperatorsdk.operator.EventListUtils.containsCustomResourceDeletedEvent;
27-
import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
28-
import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
29-
3023
/**
3124
* Event handler that makes sure that events are processed in a "single threaded" way per resource
3225
* UID, while buffering events which are received during an execution.
@@ -35,26 +28,32 @@ public class DefaultEventHandler implements EventHandler {
3528

3629
private static final Logger log = LoggerFactory.getLogger(DefaultEventHandler.class);
3730

38-
private final CustomResourceCache customResourceCache;
39-
private final EventBuffer eventBuffer;
40-
private final Set<String> underProcessing = new HashSet<>();
41-
private final ScheduledThreadPoolExecutor executor;
42-
private final EventDispatcher eventDispatcher;
43-
private final Retry retry;
44-
private final Map<String, RetryExecution> retryState = new HashMap<>();
45-
private DefaultEventSourceManager defaultEventSourceManager;
31+
private final CustomResourceCache customResourceCache;
32+
private final EventBuffer eventBuffer;
33+
private final Set<String> underProcessing = new HashSet<>();
34+
private final ScheduledThreadPoolExecutor executor;
35+
private final EventDispatcher eventDispatcher;
36+
private final Retry retry;
37+
private final Map<String, RetryExecution> retryState = new HashMap<>();
38+
private DefaultEventSourceManager defaultEventSourceManager;
4639

4740
private final ReentrantLock lock = new ReentrantLock();
4841

49-
public DefaultEventHandler(CustomResourceCache customResourceCache, EventDispatcher eventDispatcher, String relatedControllerName,
50-
Retry retry) {
51-
this.customResourceCache = customResourceCache;
52-
this.eventDispatcher = eventDispatcher;
53-
this.retry = retry;
54-
eventBuffer = new EventBuffer();
55-
executor = new ScheduledThreadPoolExecutor(5, new ThreadFactory() {
56-
@Override
57-
public Thread newThread(Runnable runnable) {
42+
public DefaultEventHandler(
43+
CustomResourceCache customResourceCache,
44+
EventDispatcher eventDispatcher,
45+
String relatedControllerName,
46+
Retry retry) {
47+
this.customResourceCache = customResourceCache;
48+
this.eventDispatcher = eventDispatcher;
49+
this.retry = retry;
50+
eventBuffer = new EventBuffer();
51+
executor =
52+
new ScheduledThreadPoolExecutor(
53+
5,
54+
new ThreadFactory() {
55+
@Override
56+
public Thread newThread(Runnable runnable) {
5857
return new Thread(runnable, "EventHandler-" + relatedControllerName);
5958
}
6059
});
@@ -101,80 +100,90 @@ private void executeBufferedEvents(String customResourceUid) {
101100
}
102101
}
103102

104-
void eventProcessingFinished(ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
105-
try {
106-
lock.lock();
107-
log.debug("Event processing finished. Scope: {}", executionScope);
108-
unsetUnderExecution(executionScope.getCustomResourceUid());
109-
110-
if (retry != null && postExecutionControl.exceptionDuringExecution()) {
111-
handleRetryOnException(executionScope, postExecutionControl);
112-
} else if (retry != null) {
113-
handleSuccessfulExecutionRegardingRetry(executionScope);
114-
}
115-
116-
if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
117-
cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
118-
} else {
119-
cacheUpdatedResourceIfChanged(executionScope, postExecutionControl);
120-
executeBufferedEvents(executionScope.getCustomResourceUid());
121-
}
122-
} finally {
123-
lock.unlock();
124-
}
103+
void eventProcessingFinished(
104+
ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
105+
try {
106+
lock.lock();
107+
log.debug("Event processing finished. Scope: {}", executionScope);
108+
unsetUnderExecution(executionScope.getCustomResourceUid());
109+
110+
if (retry != null && postExecutionControl.exceptionDuringExecution()) {
111+
handleRetryOnException(executionScope, postExecutionControl);
112+
} else if (retry != null) {
113+
handleSuccessfulExecutionRegardingRetry(executionScope);
114+
}
115+
116+
if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
117+
cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
118+
} else {
119+
cacheUpdatedResourceIfChanged(executionScope, postExecutionControl);
120+
executeBufferedEvents(executionScope.getCustomResourceUid());
121+
}
122+
} finally {
123+
lock.unlock();
125124
}
125+
}
126126

127-
/**
128-
* Regarding the events there are 2 approaches we can take. Either retry always when there are new events (received meanwhile retry
129-
* is in place or already in buffer) instantly or always wait according to the retry timing if there was an exception.
130-
*/
131-
private void handleRetryOnException(ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
132-
RetryExecution execution = getOrInitRetryExecution(executionScope);
133-
boolean newEventsExists = eventBuffer.newEventsExists(executionScope.getCustomResourceUid());
134-
eventBuffer.putBackEvents(executionScope.getCustomResourceUid(), executionScope.getEvents());
135-
136-
Optional<Long> nextDelay = execution.nextDelay();
137-
if (newEventsExists) {
138-
executeBufferedEvents(executionScope.getCustomResourceUid());
139-
return;
140-
}
141-
nextDelay.ifPresent(delay ->
142-
defaultEventSourceManager.getRetryTimerEventSource()
143-
.scheduleOnce(executionScope.getCustomResource(), delay));
127+
/**
128+
* Regarding the events there are 2 approaches we can take. Either retry always when there are new
129+
* events (received meanwhile retry is in place or already in buffer) instantly or always wait
130+
* according to the retry timing if there was an exception.
131+
*/
132+
private void handleRetryOnException(
133+
ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
134+
RetryExecution execution = getOrInitRetryExecution(executionScope);
135+
boolean newEventsExists = eventBuffer.newEventsExists(executionScope.getCustomResourceUid());
136+
eventBuffer.putBackEvents(executionScope.getCustomResourceUid(), executionScope.getEvents());
137+
138+
Optional<Long> nextDelay = execution.nextDelay();
139+
if (newEventsExists) {
140+
executeBufferedEvents(executionScope.getCustomResourceUid());
141+
return;
144142
}
143+
nextDelay.ifPresent(
144+
delay ->
145+
defaultEventSourceManager
146+
.getRetryTimerEventSource()
147+
.scheduleOnce(executionScope.getCustomResource(), delay));
148+
}
145149

146-
private void handleSuccessfulExecutionRegardingRetry(ExecutionScope executionScope) {
147-
retryState.remove(executionScope.getCustomResourceUid());
148-
defaultEventSourceManager.getRetryTimerEventSource().cancelOnceSchedule(executionScope.getCustomResourceUid());
149-
}
150+
private void handleSuccessfulExecutionRegardingRetry(ExecutionScope executionScope) {
151+
retryState.remove(executionScope.getCustomResourceUid());
152+
defaultEventSourceManager
153+
.getRetryTimerEventSource()
154+
.cancelOnceSchedule(executionScope.getCustomResourceUid());
155+
}
150156

151-
private RetryExecution getOrInitRetryExecution(ExecutionScope executionScope) {
152-
RetryExecution retryExecution = retryState.get(executionScope.getCustomResourceUid());
153-
if (retryExecution == null) {
154-
retryExecution = retry.initExecution();
155-
retryState.put(executionScope.getCustomResourceUid(), retryExecution);
156-
}
157-
return retryExecution;
157+
private RetryExecution getOrInitRetryExecution(ExecutionScope executionScope) {
158+
RetryExecution retryExecution = retryState.get(executionScope.getCustomResourceUid());
159+
if (retryExecution == null) {
160+
retryExecution = retry.initExecution();
161+
retryState.put(executionScope.getCustomResourceUid(), retryExecution);
158162
}
163+
return retryExecution;
164+
}
159165

160-
/**
161-
* Here we try to cache the latest resource after an update. The goal is to solve a concurrency issue we've seen:
162-
* If an execution is finished, where we updated a custom resource, but there are other events already buffered for next
163-
* execution, we might not get the newest custom resource from CustomResource event source in time. Thus we execute
164-
* the next batch of events but with a non up to date CR. Here we cache the latest CustomResource from the update
165-
* execution so we make sure its already used in the up-coming execution.
166-
* <p>
167-
* Note that this is an improvement, not a bug fix. This situation can happen naturally, we just make the execution more
168-
* efficient, and avoid questions about conflicts.
169-
* <p>
170-
* Note that without the conditional locking in the cache, there is a very minor chance that we would override an
171-
* additional change coming from a different client.
172-
*/
173-
private void cacheUpdatedResourceIfChanged(ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
174-
if (postExecutionControl.customResourceUpdatedDuringExecution()) {
175-
CustomResource originalCustomResource = executionScope.getCustomResource();
176-
CustomResource customResourceAfterExecution = postExecutionControl.getUpdatedCustomResource().get();
177-
String originalResourceVersion = getVersion(originalCustomResource);
166+
/**
167+
* Here we try to cache the latest resource after an update. The goal is to solve a concurrency
168+
* issue we've seen: If an execution is finished, where we updated a custom resource, but there
169+
* are other events already buffered for next execution, we might not get the newest custom
170+
* resource from CustomResource event source in time. Thus we execute the next batch of events but
171+
* with a non up to date CR. Here we cache the latest CustomResource from the update execution so
172+
* we make sure its already used in the up-coming execution.
173+
*
174+
* <p>Note that this is an improvement, not a bug fix. This situation can happen naturally, we
175+
* just make the execution more efficient, and avoid questions about conflicts.
176+
*
177+
* <p>Note that without the conditional locking in the cache, there is a very minor chance that we
178+
* would override an additional change coming from a different client.
179+
*/
180+
private void cacheUpdatedResourceIfChanged(
181+
ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
182+
if (postExecutionControl.customResourceUpdatedDuringExecution()) {
183+
CustomResource originalCustomResource = executionScope.getCustomResource();
184+
CustomResource customResourceAfterExecution =
185+
postExecutionControl.getUpdatedCustomResource().get();
186+
String originalResourceVersion = getVersion(originalCustomResource);
178187

179188
log.debug(
180189
"Trying to update resource cache from update response for resource uid: {} new version: {} old version: {}",

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java

Lines changed: 22 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,38 +1,36 @@
11
package io.javaoperatorsdk.operator.processing;
22

33
import io.javaoperatorsdk.operator.processing.event.Event;
4-
import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEvent;
5-
64
import java.util.*;
75

86
class EventBuffer {
97

10-
private final Map<String, List<Event>> events = new HashMap<>();
8+
private final Map<String, List<Event>> events = new HashMap<>();
119

12-
public void addEvent(Event event) {
13-
String uid = event.getRelatedCustomResourceUid();
14-
List<Event> crEvents = events.computeIfAbsent(uid, (id) -> new ArrayList<>(1));
15-
crEvents.add(event);
16-
}
10+
public void addEvent(Event event) {
11+
String uid = event.getRelatedCustomResourceUid();
12+
List<Event> crEvents = events.computeIfAbsent(uid, (id) -> new ArrayList<>(1));
13+
crEvents.add(event);
14+
}
1715

18-
public boolean newEventsExists(String resourceId) {
19-
return !events.get(resourceId).isEmpty();
20-
}
16+
public boolean newEventsExists(String resourceId) {
17+
return !events.get(resourceId).isEmpty();
18+
}
2119

22-
public void putBackEvents(String resourceUid, List<Event> oldEvents) {
23-
events.get(resourceUid).addAll(0, oldEvents);
24-
}
20+
public void putBackEvents(String resourceUid, List<Event> oldEvents) {
21+
events.get(resourceUid).addAll(0, oldEvents);
22+
}
2523

26-
public boolean containsEvents(String customResourceId) {
27-
return events.get(customResourceId) != null;
28-
}
24+
public boolean containsEvents(String customResourceId) {
25+
return events.get(customResourceId) != null;
26+
}
2927

30-
public List<Event> getAndRemoveEventsForExecution(String resourceUid) {
31-
List<Event> crEvents = events.remove(resourceUid);
32-
return crEvents == null ? Collections.emptyList() : crEvents;
33-
}
28+
public List<Event> getAndRemoveEventsForExecution(String resourceUid) {
29+
List<Event> crEvents = events.remove(resourceUid);
30+
return crEvents == null ? Collections.emptyList() : crEvents;
31+
}
3432

35-
public void cleanup(String resourceUid) {
36-
events.remove(resourceUid);
37-
}
33+
public void cleanup(String resourceUid) {
34+
events.remove(resourceUid);
35+
}
3836
}

0 commit comments

Comments
 (0)