Skip to content

feat: add a DependentResource implementation that's also an EventSource #1093

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 15 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,12 @@
public interface EventSourceInitializer<P extends HasMetadata> {

/**
* Prepares a list of {@link EventSource} implementations to be registered by the SDK.
* Prepares a map of {@link EventSource} implementations keyed by the name with which they need to
* be registered by the SDK.
*
* @param context a {@link EventSourceContext} providing access to information useful to event
* sources
* @return list of event sources to register
* @return a map of event sources to register
*/
Map<String, EventSource> prepareEventSources(EventSourceContext<P> context);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

import io.fabric8.kubernetes.api.model.HasMetadata;

@SuppressWarnings("rawtypes")
public class UpdateControl<T extends HasMetadata> extends BaseControl<UpdateControl<T>> {

private final T resource;
Expand Down Expand Up @@ -32,8 +31,7 @@ public static <T extends HasMetadata> UpdateControl<T> updateResource(T customRe
return new UpdateControl<>(customResource, false, true);
}

public static <T extends HasMetadata> UpdateControl<T> updateStatus(
T customResource) {
public static <T extends HasMetadata> UpdateControl<T> updateStatus(T customResource) {
return new UpdateControl<>(customResource, true, false);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,4 @@ public interface EventSourceProvider<P extends HasMetadata> {
* @return the initiated event source.
*/
EventSource initEventSource(EventSourceContext<P> context);

EventSource getEventSource();
}
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,6 @@ public UpdateControl<P> execute() throws Exception {
.collect(Collectors.joining("\n")),
exceptions);
}

return reconciler.reconcile(resource, context);
}
});
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,7 @@

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter;
import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationEventFilter;
import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
import io.javaoperatorsdk.operator.api.reconciler.dependent.*;
import io.javaoperatorsdk.operator.processing.event.ResourceID;

public abstract class AbstractDependentResource<R, P extends HasMetadata>
Expand Down Expand Up @@ -68,89 +63,20 @@ public ReconcileResult<R> reconcile(P primary, Context<P> context) {

protected R handleCreate(R desired, P primary, Context<P> context) {
ResourceID resourceID = ResourceID.fromResource(primary);
R created = null;
try {
prepareEventFiltering(desired, resourceID);
created = creator.create(desired, primary, context);
cacheAfterCreate(resourceID, created);
return created;
} catch (RuntimeException e) {
cleanupAfterEventFiltering(desired, resourceID, created);
throw e;
}
}

private void cleanupAfterEventFiltering(R desired, ResourceID resourceID, R created) {
if (isFilteringEventSource()) {
eventSourceAsRecentOperationEventFilter()
.cleanupOnCreateOrUpdateEventFiltering(resourceID, created);
}
}

private void cacheAfterCreate(ResourceID resourceID, R created) {
if (isRecentOperationCacheFiller()) {
eventSourceAsRecentOperationCacheFiller().handleRecentResourceCreate(resourceID, created);
}
R created = creator.create(desired, primary, context);
cacheAfterCreate(resourceID, created);
return created;
}

private void cacheAfterUpdate(R actual, ResourceID resourceID, R updated) {
if (isRecentOperationCacheFiller()) {
eventSourceAsRecentOperationCacheFiller().handleRecentResourceUpdate(resourceID, updated,
actual);
}
}
protected abstract void cacheAfterCreate(ResourceID resourceID, R created);

private void prepareEventFiltering(R desired, ResourceID resourceID) {
if (isFilteringEventSource()) {
eventSourceAsRecentOperationEventFilter().prepareForCreateOrUpdateEventFiltering(resourceID,
desired);
}
}
protected abstract void cacheAfterUpdate(R actual, ResourceID resourceID, R updated);

protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
ResourceID resourceID = ResourceID.fromResource(primary);
R updated = null;
try {
prepareEventFiltering(desired, resourceID);
updated = updater.update(actual, desired, primary, context);
cacheAfterUpdate(actual, resourceID, updated);
return updated;
} catch (RuntimeException e) {
cleanupAfterEventFiltering(desired, resourceID, updated);
throw e;
}
}

@SuppressWarnings("unchecked")
private RecentOperationEventFilter<R> eventSourceAsRecentOperationEventFilter() {
return (RecentOperationEventFilter<R>) ((EventSourceProvider<P>) this).getEventSource();
}

@SuppressWarnings("unchecked")
private RecentOperationCacheFiller<R> eventSourceAsRecentOperationCacheFiller() {
return (RecentOperationCacheFiller<R>) ((EventSourceProvider<P>) this).getEventSource();
}

@SuppressWarnings("unchecked")
// this cannot be done in constructor since event source might be initialized later
protected boolean isFilteringEventSource() {
if (this instanceof EventSourceProvider) {
final var eventSource = ((EventSourceProvider<P>) this).getEventSource();
return eventSource instanceof RecentOperationEventFilter;
} else {
return false;
}
}

@SuppressWarnings("unchecked")
// this cannot be done in constructor since event source might be initialized later
protected boolean isRecentOperationCacheFiller() {
if (this instanceof EventSourceProvider) {
final var eventSource = ((EventSourceProvider<P>) this).getEventSource();
return eventSource instanceof RecentOperationCacheFiller;
} else {
return false;
}
R updated = updater.update(actual, desired, primary, context);
cacheAfterUpdate(actual, resourceID, updated);
return updated;
}

protected R desired(P primary, Context<P> context) {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
package io.javaoperatorsdk.operator.processing.dependent;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.OperatorException;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationEventFilter;
import io.javaoperatorsdk.operator.processing.event.EventHandler;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.javaoperatorsdk.operator.processing.event.source.EventSource;

public abstract class AbstractEventSourceHolderDependentResource<R, P extends HasMetadata, T extends EventSource>
extends AbstractDependentResource<R, P> implements EventSource, EventSourceProvider<P> {
private T eventSource;
private boolean isFilteringEventSource;
private boolean isCacheFillerEventSource;

@Override
public void start() throws OperatorException {
eventSource.start();
}

@Override
public void stop() throws OperatorException {
eventSource.stop();
}

public EventSource initEventSource(EventSourceContext<P> context) {
// some sub-classes (e.g. KubernetesDependentResource) can have their event source created
// before this method is called in the managed case, so only create the event source if it
// hasn't already been set
if (eventSource == null) {
eventSource = createEventSource(context);
}

// but we still need to record which interfaces the event source implements even if it has
// already been set before this method is called
isFilteringEventSource = eventSource instanceof RecentOperationEventFilter;
isCacheFillerEventSource = eventSource instanceof RecentOperationCacheFiller;
return this;
}

protected abstract T createEventSource(EventSourceContext<P> context);

protected void setEventSource(T eventSource) {
this.eventSource = eventSource;
}

@Override
public void setEventHandler(EventHandler handler) {
eventSource.setEventHandler(handler);
}

protected T eventSource() {
return eventSource;
}

protected R handleCreate(R desired, P primary, Context<P> context) {
ResourceID resourceID = ResourceID.fromResource(primary);
R created = null;
try {
prepareEventFiltering(desired, resourceID);
created = super.handleCreate(desired, primary, context);
return created;
} catch (RuntimeException e) {
cleanupAfterEventFiltering(desired, resourceID, created);
throw e;
}
}

protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
ResourceID resourceID = ResourceID.fromResource(primary);
R updated = null;
try {
prepareEventFiltering(desired, resourceID);
updated = super.handleUpdate(actual, desired, primary, context);
return updated;
} catch (RuntimeException e) {
cleanupAfterEventFiltering(desired, resourceID, updated);
throw e;
}
}


protected void cacheAfterCreate(ResourceID resourceID, R created) {
if (isCacheFillerEventSource) {
recentOperationCacheFiller().handleRecentResourceCreate(resourceID, created);
}
}

protected void cacheAfterUpdate(R actual, ResourceID resourceID, R updated) {
if (isCacheFillerEventSource) {
recentOperationCacheFiller().handleRecentResourceUpdate(resourceID, updated, actual);
}
}

private void prepareEventFiltering(R desired, ResourceID resourceID) {
if (isFilteringEventSource) {
recentOperationEventFilter().prepareForCreateOrUpdateEventFiltering(resourceID, desired);
}
}

private void cleanupAfterEventFiltering(R desired, ResourceID resourceID, R created) {
if (isFilteringEventSource) {
recentOperationEventFilter().cleanupOnCreateOrUpdateEventFiltering(resourceID, created);
}
}

@SuppressWarnings("unchecked")
private RecentOperationEventFilter<R> recentOperationEventFilter() {
return (RecentOperationEventFilter<R>) eventSource;
}

@SuppressWarnings("unchecked")
private RecentOperationCacheFiller<R> recentOperationCacheFiller() {
return (RecentOperationCacheFiller<R>) eventSource;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -3,29 +3,20 @@
import java.util.Optional;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceProvider;
import io.javaoperatorsdk.operator.processing.dependent.AbstractDependentResource;
import io.javaoperatorsdk.operator.processing.dependent.AbstractEventSourceHolderDependentResource;
import io.javaoperatorsdk.operator.processing.event.ExternalResourceCachingEventSource;
import io.javaoperatorsdk.operator.processing.event.source.EventSource;

public abstract class AbstractCachingDependentResource<R, P extends HasMetadata>
extends AbstractDependentResource<R, P>
implements EventSourceProvider<P> {

protected ExternalResourceCachingEventSource<R, P> eventSource;
extends
AbstractEventSourceHolderDependentResource<R, P, ExternalResourceCachingEventSource<R, P>> {
private final Class<R> resourceType;

protected AbstractCachingDependentResource(Class<R> resourceType) {
this.resourceType = resourceType;
}

public Optional<R> fetchResource(P primaryResource) {
return eventSource.getAssociated(primaryResource);
}

@Override
public EventSource getEventSource() {
return eventSource;
return eventSource().getAssociated(primaryResource);
}

@Override
Expand All @@ -35,6 +26,6 @@ public Class<R> resourceType() {

@Override
public Optional<R> getResource(P primaryResource) {
return eventSource.getAssociated(primaryResource);
return eventSource().getAssociated(primaryResource);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -59,17 +59,13 @@ public final void delete(P primary, Context<P> context) {
protected abstract void deleteResource(P primary, Context<P> context);

@Override
protected R handleCreate(R desired, P primary, Context<P> context) {
var res = this.creator.create(desired, primary, context);
cache.put(ResourceID.fromResource(primary), res);
return res;
protected void cacheAfterCreate(ResourceID resourceID, R created) {
cache.put(resourceID, created);
}

@Override
protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
var res = updater.update(actual, desired, primary, context);
cache.put(ResourceID.fromResource(primary), res);
return res;
protected void cacheAfterUpdate(R actual, ResourceID resourceID, R updated) {
cache.put(resourceID, updated);
}

public Matcher.Result<R> match(R actualResource, P primary, Context<P> context) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.processing.event.source.EventSource;
import io.javaoperatorsdk.operator.processing.event.ExternalResourceCachingEventSource;
import io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource;

public abstract class PerResourcePollingDependentResource<R, P extends HasMetadata>
Expand All @@ -17,9 +17,9 @@ public PerResourcePollingDependentResource(Class<R> resourceType, long pollingPe
}

@Override
public EventSource initEventSource(EventSourceContext<P> context) {
eventSource = new PerResourcePollingEventSource<>(this, context.getPrimaryCache(),
protected ExternalResourceCachingEventSource<R, P> createEventSource(
EventSourceContext<P> context) {
return new PerResourcePollingEventSource<>(this, context.getPrimaryCache(),
getPollingPeriod(), resourceType());
return eventSource;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.processing.event.ExternalResourceCachingEventSource;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.javaoperatorsdk.operator.processing.event.source.EventSource;
import io.javaoperatorsdk.operator.processing.event.source.polling.PollingEventSource;

public abstract class PollingDependentResource<R, P extends HasMetadata>
Expand All @@ -21,8 +21,8 @@ public PollingDependentResource(Class<R> resourceType, long pollingPeriod) {
}

@Override
public EventSource initEventSource(EventSourceContext<P> context) {
eventSource = new PollingEventSource<>(this, getPollingPeriod(), resourceType());
return eventSource;
protected ExternalResourceCachingEventSource<R, P> createEventSource(
EventSourceContext<P> context) {
return new PollingEventSource<>(this, getPollingPeriod(), resourceType());
}
}
Loading