Skip to content

Commit d0d67fd

Browse files
shawkinsshawkins
committed
remove the use of the previous annotation via locking
Signed-off-by: Steve Hawkins <shawkins@redhat.com>
1 parent c310e6b commit d0d67fd

File tree

12 files changed

+199
-208
lines changed

12 files changed

+199
-208
lines changed

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java‎

Lines changed: 3 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,8 @@
4646
import io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflowFactory;
4747
import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource;
4848

49+
import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSIONS;
50+
4951
/** An interface from which to retrieve configuration information. */
5052
public interface ConfigurationService {
5153

@@ -447,19 +449,6 @@ default Set<Class<? extends HasMetadata>> defaultNonSSAResource() {
447449
return defaultNonSSAResources();
448450
}
449451

450-
/**
451-
* If a javaoperatorsdk.io/previous annotation should be used so that the operator sdk can detect
452-
* events from its own updates of dependent resources and then filter them.
453-
*
454-
* <p>Disable this if you want to react to your own dependent resource updates
455-
*
456-
* @return if special annotation should be used for dependent resource to filter events
457-
* @since 4.5.0
458-
*/
459-
default boolean previousAnnotationForDependentResourcesEventFiltering() {
460-
return true;
461-
}
462-
463452
/**
464453
* For dependent resources, the framework can add an annotation to filter out events resulting
465454
* directly from the framework's operation. There are, however, some resources that do not follow
@@ -500,7 +489,7 @@ default Set<Class<? extends HasMetadata>> withPreviousAnnotationForDependentReso
500489
* @since 4.5.0
501490
*/
502491
default boolean parseResourceVersionsForEventFilteringAndCaching() {
503-
return true;
492+
return DEFAULT_COMPARABLE_RESOURCE_VERSIONS;
504493
}
505494

506495
/**

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -331,13 +331,6 @@ public Set<Class<? extends HasMetadata>> defaultNonSSAResources() {
331331
defaultNonSSAResource, ConfigurationService::defaultNonSSAResources);
332332
}
333333

334-
@Override
335-
public boolean previousAnnotationForDependentResourcesEventFiltering() {
336-
return overriddenValueOrDefault(
337-
previousAnnotationForDependentResources,
338-
ConfigurationService::previousAnnotationForDependentResourcesEventFiltering);
339-
}
340-
341334
@Override
342335
public boolean parseResourceVersionsForEventFilteringAndCaching() {
343336
return overriddenValueOrDefault(

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter;
3434
import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers;
3535

36+
import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSIONS;
3637
import static io.javaoperatorsdk.operator.api.reconciler.Constants.SAME_AS_CONTROLLER_NAMESPACES_SET;
3738
import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_ALL_NAMESPACE_SET;
3839
import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_CURRENT_NAMESPACE_SET;
@@ -140,7 +141,7 @@ public Optional<KubernetesClient> getKubernetesClient() {
140141
}
141142

142143
@Override
143-
public boolean parseResourceVersionsForEventFilteringAndCaching() {
144+
public boolean comparableResourceVersions() {
144145
return this.comparableResourceVersions;
145146
}
146147
}
@@ -156,7 +157,7 @@ class Builder<R extends HasMetadata> {
156157
private PrimaryToSecondaryMapper<?> primaryToSecondaryMapper;
157158
private SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper;
158159
private KubernetesClient kubernetesClient;
159-
private boolean comparableResourceVersions = true;
160+
private boolean comparableResourceVersions = DEFAULT_COMPARABLE_RESOURCE_VERSIONS;
160161

161162
private Builder(Class<R> resourceClass, Class<? extends HasMetadata> primaryResourceClass) {
162163
this(resourceClass, primaryResourceClass, null);
@@ -294,8 +295,8 @@ public Builder<R> withFieldSelector(FieldSelector fieldSelector) {
294295
return this;
295296
}
296297

297-
public Builder<R> parseResourceVersionsForEventFilteringAndCaching(boolean parse) {
298-
this.comparableResourceVersions = parse;
298+
public Builder<R> withComparableResourceVersions(boolean comparableResourceVersions) {
299+
this.comparableResourceVersions = comparableResourceVersions;
299300
return this;
300301
}
301302

@@ -343,5 +344,5 @@ public InformerEventSourceConfiguration<R> build() {
343344
}
344345
}
345346

346-
boolean parseResourceVersionsForEventFilteringAndCaching();
347+
boolean comparableResourceVersions();
347348
}

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Constants.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ public final class Constants {
4141
public static final String RESOURCE_GVK_KEY = "josdk.resource.gvk";
4242
public static final String CONTROLLER_NAME = "controller.name";
4343
public static final boolean DEFAULT_FOLLOW_CONTROLLER_NAMESPACE_CHANGES = true;
44+
public static final boolean DEFAULT_COMPARABLE_RESOURCE_VERSIONS = true;
4445

4546
private Constants() {}
4647
}

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java‎

Lines changed: 21 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,6 @@ public abstract class KubernetesDependentResource<R extends HasMetadata, P exten
5555
private final boolean garbageCollected = this instanceof GarbageCollected;
5656
private KubernetesDependentResourceConfig<R> kubernetesDependentResourceConfig;
5757
private volatile Boolean useSSA;
58-
private volatile Boolean usePreviousAnnotationForEventFiltering;
5958

6059
public KubernetesDependentResource() {}
6160

@@ -72,6 +71,27 @@ public void configureWith(KubernetesDependentResourceConfig<R> config) {
7271
this.kubernetesDependentResourceConfig = config;
7372
}
7473

74+
@Override
75+
protected R handleCreate(R desired, P primary, Context<P> context) {
76+
return eventSource()
77+
.orElseThrow()
78+
.updateAndCacheResource(
79+
desired,
80+
context,
81+
toCreate -> KubernetesDependentResource.super.handleCreate(toCreate, primary, context));
82+
}
83+
84+
@Override
85+
protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
86+
return eventSource()
87+
.orElseThrow()
88+
.updateAndCacheResource(
89+
desired,
90+
context,
91+
toUpdate ->
92+
KubernetesDependentResource.super.handleUpdate(actual, toUpdate, primary, context));
93+
}
94+
7595
@SuppressWarnings("unused")
7696
public R create(R desired, P primary, Context<P> context) {
7797
if (useSSA(context)) {
@@ -158,14 +178,6 @@ protected void addMetadata(
158178
} else {
159179
annotations.remove(InformerEventSource.PREVIOUS_ANNOTATION_KEY);
160180
}
161-
} else if (usePreviousAnnotation(context)) { // set a new one
162-
eventSource()
163-
.orElseThrow()
164-
.addPreviousAnnotation(
165-
Optional.ofNullable(actualResource)
166-
.map(r -> r.getMetadata().getResourceVersion())
167-
.orElse(null),
168-
target);
169181
}
170182
addReferenceHandlingMetadata(target, primary);
171183
}
@@ -181,22 +193,6 @@ protected boolean useSSA(Context<P> context) {
181193
return useSSA;
182194
}
183195

184-
private boolean usePreviousAnnotation(Context<P> context) {
185-
if (usePreviousAnnotationForEventFiltering == null) {
186-
usePreviousAnnotationForEventFiltering =
187-
context
188-
.getControllerConfiguration()
189-
.getConfigurationService()
190-
.previousAnnotationForDependentResourcesEventFiltering()
191-
&& !context
192-
.getControllerConfiguration()
193-
.getConfigurationService()
194-
.withPreviousAnnotationForDependentResourcesBlocklist()
195-
.contains(this.resourceType());
196-
}
197-
return usePreviousAnnotationForEventFiltering;
198-
}
199-
200196
@Override
201197
protected void handleDelete(P primary, R secondary, Context<P> context) {
202198
if (secondary != null) {

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java‎

Lines changed: 37 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
import java.util.Optional;
1919
import java.util.Set;
20-
import java.util.UUID;
20+
import java.util.function.UnaryOperator;
2121
import java.util.stream.Collectors;
2222

2323
import org.slf4j.Logger;
@@ -35,6 +35,8 @@
3535
import io.javaoperatorsdk.operator.processing.event.ResourceID;
3636
import io.javaoperatorsdk.operator.processing.event.source.PrimaryToSecondaryMapper;
3737

38+
import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSIONS;
39+
3840
/**
3941
* Wraps informer(s) so they are connected to the eventing system of the framework. Note that since
4042
* this is built on top of Fabric8 client Informers, it also supports caching resources using
@@ -78,18 +80,17 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
7880
// we need direct control for the indexer to propagate the just update resource also to the index
7981
private final PrimaryToSecondaryIndex<R> primaryToSecondaryIndex;
8082
private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
81-
private final String id = UUID.randomUUID().toString();
8283

8384
public InformerEventSource(
8485
InformerEventSourceConfiguration<R> configuration, EventSourceContext<P> context) {
8586
this(
8687
configuration,
8788
configuration.getKubernetesClient().orElse(context.getClient()),
88-
configuration.parseResourceVersionsForEventFilteringAndCaching());
89+
configuration.comparableResourceVersions());
8990
}
9091

9192
InformerEventSource(InformerEventSourceConfiguration<R> configuration, KubernetesClient client) {
92-
this(configuration, client, true);
93+
this(configuration, client, DEFAULT_COMPARABLE_RESOURCE_VERSIONS);
9394
}
9495

9596
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -122,6 +123,22 @@ private InformerEventSource(
122123
genericFilter = informerConfig.getGenericFilter();
123124
}
124125

126+
public R updateAndCacheResource(
127+
R resourceToUpdate, Context<?> context, UnaryOperator<R> updateMethod) {
128+
ResourceID id = ResourceID.fromResource(resourceToUpdate);
129+
if (log.isDebugEnabled()) {
130+
log.debug("Update and cache: {}", id);
131+
}
132+
try {
133+
temporaryResourceCache.startModifying(id);
134+
var updated = updateMethod.apply(resourceToUpdate);
135+
handleRecentResourceUpdate(id, updated, resourceToUpdate);
136+
return updated;
137+
} finally {
138+
temporaryResourceCache.doneModifying(id);
139+
}
140+
}
141+
125142
@Override
126143
public void onAdd(R newResource) {
127144
if (log.isDebugEnabled()) {
@@ -131,9 +148,7 @@ public void onAdd(R newResource) {
131148
resourceType().getSimpleName(),
132149
newResource.getMetadata().getResourceVersion());
133150
}
134-
primaryToSecondaryIndex.onAddOrUpdate(newResource);
135-
onAddOrUpdate(
136-
Operation.ADD, newResource, null, () -> InformerEventSource.super.onAdd(newResource));
151+
onAddOrUpdate(Operation.ADD, newResource, null);
137152
}
138153

139154
@Override
@@ -146,16 +161,11 @@ public void onUpdate(R oldObject, R newObject) {
146161
newObject.getMetadata().getResourceVersion(),
147162
oldObject.getMetadata().getResourceVersion());
148163
}
149-
primaryToSecondaryIndex.onAddOrUpdate(newObject);
150-
onAddOrUpdate(
151-
Operation.UPDATE,
152-
newObject,
153-
oldObject,
154-
() -> InformerEventSource.super.onUpdate(oldObject, newObject));
164+
onAddOrUpdate(Operation.UPDATE, newObject, oldObject);
155165
}
156166

157167
@Override
158-
public void onDelete(R resource, boolean b) {
168+
public synchronized void onDelete(R resource, boolean b) {
159169
if (log.isDebugEnabled()) {
160170
log.debug(
161171
"On delete event received for resource id: {} type: {}",
@@ -177,55 +187,28 @@ public synchronized void start() {
177187
manager().list().forEach(primaryToSecondaryIndex::onAddOrUpdate);
178188
}
179189

180-
private synchronized void onAddOrUpdate(
181-
Operation operation, R newObject, R oldObject, Runnable superOnOp) {
190+
private synchronized void onAddOrUpdate(Operation operation, R newObject, R oldObject) {
191+
primaryToSecondaryIndex.onAddOrUpdate(newObject);
182192
var resourceID = ResourceID.fromResource(newObject);
183193

184-
if (canSkipEvent(newObject, oldObject, resourceID)) {
194+
if (temporaryResourceCache.onAddOrUpdateEvent(newObject)) {
185195
log.debug(
186196
"Skipping event propagation for {}, since was a result of a reconcile action. Resource"
187197
+ " ID: {}",
188198
operation,
189199
ResourceID.fromResource(newObject));
190-
superOnOp.run();
200+
} else if (eventAcceptedByFilter(operation, newObject, oldObject)) {
201+
log.debug(
202+
"Propagating event for {}, resource with same version not result of a reconciliation."
203+
+ " Resource ID: {}",
204+
operation,
205+
resourceID);
206+
propagateEvent(newObject);
191207
} else {
192-
superOnOp.run();
193-
if (eventAcceptedByFilter(operation, newObject, oldObject)) {
194-
log.debug(
195-
"Propagating event for {}, resource with same version not result of a reconciliation."
196-
+ " Resource ID: {}",
197-
operation,
198-
resourceID);
199-
propagateEvent(newObject);
200-
} else {
201-
log.debug("Event filtered out for operation: {}, resourceID: {}", operation, resourceID);
202-
}
208+
log.debug("Event filtered out for operation: {}, resourceID: {}", operation, resourceID);
203209
}
204210
}
205211

206-
private boolean canSkipEvent(R newObject, R oldObject, ResourceID resourceID) {
207-
return temporaryResourceCache.canSkipEvent(resourceID, newObject)
208-
|| isEventKnownFromAnnotation(newObject, oldObject);
209-
}
210-
211-
private boolean isEventKnownFromAnnotation(R newObject, R oldObject) {
212-
String previous = newObject.getMetadata().getAnnotations().get(PREVIOUS_ANNOTATION_KEY);
213-
boolean known = false;
214-
if (previous != null) {
215-
String[] parts = previous.split(",");
216-
if (id.equals(parts[0])) {
217-
if (oldObject == null && parts.length == 1) {
218-
known = true;
219-
} else if (oldObject != null
220-
&& parts.length == 2
221-
&& oldObject.getMetadata().getResourceVersion().equals(parts[1])) {
222-
known = true;
223-
}
224-
}
225-
}
226-
return known;
227-
}
228-
229212
private void propagateEvent(R object) {
230213
var primaryResourceIdSet =
231214
configuration().getSecondaryToPrimaryMapper().toPrimaryResourceIDs(object);
@@ -273,13 +256,13 @@ public Set<R> getSecondaryResources(P primary) {
273256
}
274257

275258
@Override
276-
public synchronized void handleRecentResourceUpdate(
259+
public void handleRecentResourceUpdate(
277260
ResourceID resourceID, R resource, R previousVersionOfResource) {
278261
handleRecentCreateOrUpdate(Operation.UPDATE, resource, previousVersionOfResource);
279262
}
280263

281264
@Override
282-
public synchronized void handleRecentResourceCreate(ResourceID resourceID, R resource) {
265+
public void handleRecentResourceCreate(ResourceID resourceID, R resource) {
283266
handleRecentCreateOrUpdate(Operation.ADD, resource, null);
284267
}
285268

@@ -313,22 +296,6 @@ private boolean acceptedByDeleteFilters(R resource, boolean b) {
313296
&& (genericFilter == null || genericFilter.accept(resource));
314297
}
315298

316-
/**
317-
* Add an annotation to the resource so that the subsequent will be omitted
318-
*
319-
* @param resourceVersion null if there is no prior version
320-
* @param target mutable resource that will be returned
321-
*/
322-
public R addPreviousAnnotation(String resourceVersion, R target) {
323-
target
324-
.getMetadata()
325-
.getAnnotations()
326-
.put(
327-
PREVIOUS_ANNOTATION_KEY,
328-
id + Optional.ofNullable(resourceVersion).map(rv -> "," + rv).orElse(""));
329-
return target;
330-
}
331-
332299
private enum Operation {
333300
ADD,
334301
UPDATE

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -221,11 +221,6 @@ public Optional<R> get(ResourceID resourceID) {
221221
: r);
222222
}
223223

224-
public Optional<String> getLastSyncResourceVersion(Optional<String> namespace) {
225-
return getSource(namespace.orElse(WATCH_ALL_NAMESPACES))
226-
.map(source -> source.getLastSyncResourceVersion());
227-
}
228-
229224
@Override
230225
public Stream<ResourceID> keys() {
231226
return sources.values().stream().flatMap(Cache::keys);

0 commit comments

Comments
 (0)