Skip to content

feat: expectations #2840

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

Draft
wants to merge 6 commits into
base: next
Choose a base branch
from
Draft
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 @@ -6,14 +6,11 @@
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Locale;
import java.util.Objects;
import java.util.function.Predicate;
import java.util.regex.Pattern;

import io.fabric8.kubernetes.api.builder.Builder;
import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.Namespaced;
import io.fabric8.kubernetes.client.CustomResource;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.utils.Serialization;
Expand Down Expand Up @@ -73,36 +70,6 @@ public static String getNameFor(Class<? extends Reconciler> reconcilerClass) {
return getDefaultNameFor(reconcilerClass);
}

public static void checkIfCanAddOwnerReference(HasMetadata owner, HasMetadata resource) {
if (owner instanceof GenericKubernetesResource
|| resource instanceof GenericKubernetesResource) {
return;
}
if (owner instanceof Namespaced) {
if (!(resource instanceof Namespaced)) {
throw new OperatorException(
"Cannot add owner reference from a cluster scoped to a namespace scoped resource."
+ resourcesIdentifierDescription(owner, resource));
} else if (!Objects.equals(
owner.getMetadata().getNamespace(), resource.getMetadata().getNamespace())) {
throw new OperatorException(
"Cannot add owner reference between two resource in different namespaces."
+ resourcesIdentifierDescription(owner, resource));
}
}
}

private static String resourcesIdentifierDescription(HasMetadata owner, HasMetadata resource) {
return " Owner name: "
+ owner.getMetadata().getName()
+ " Kind: "
+ owner.getKind()
+ ", Resource name: "
+ resource.getMetadata().getName()
+ " Kind: "
+ resource.getKind();
}

public static String getNameFor(Reconciler reconciler) {
return getNameFor(reconciler.getClass());
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
package io.javaoperatorsdk.operator.api.config.informer;

public @interface Field {

String path();

String value();

boolean negated() default false;
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
package io.javaoperatorsdk.operator.api.config.informer;

import java.util.Arrays;
import java.util.List;

public class FieldSelector {
private final List<Field> fields;

public FieldSelector(List<Field> fields) {
this.fields = fields;
}

public FieldSelector(Field... fields) {
this.fields = Arrays.asList(fields);
}

public List<Field> getFields() {
return fields;
}

public record Field(String path, String value, boolean negated) {
public Field(String path, String value) {
this(path, value, false);
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
package io.javaoperatorsdk.operator.api.config.informer;

import java.util.ArrayList;
import java.util.List;

public class FieldSelectorBuilder {

private final List<FieldSelector.Field> fields = new ArrayList<>();

public FieldSelectorBuilder withField(String path, String value) {
fields.add(new FieldSelector.Field(path, value));
return this;
}

public FieldSelectorBuilder withoutField(String path, String value) {
fields.add(new FieldSelector.Field(path, value, true));
return this;
}

public FieldSelector build() {
return new FieldSelector(fields);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -113,4 +113,7 @@
* the informer cache.
*/
long informerListLimit() default NO_LONG_VALUE_SET;

/** Kubernetes field selector for additional resource filtering */
Field[] fieldSelector() default {};
}
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.javaoperatorsdk.operator.api.config.informer;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Set;
Expand Down Expand Up @@ -36,6 +37,7 @@ public class InformerConfiguration<R extends HasMetadata> {
private GenericFilter<? super R> genericFilter;
private ItemStore<R> itemStore;
private Long informerListLimit;
private FieldSelector fieldSelector;

protected InformerConfiguration(
Class<R> resourceClass,
Expand All @@ -48,7 +50,8 @@ protected InformerConfiguration(
OnDeleteFilter<? super R> onDeleteFilter,
GenericFilter<? super R> genericFilter,
ItemStore<R> itemStore,
Long informerListLimit) {
Long informerListLimit,
FieldSelector fieldSelector) {
this(resourceClass);
this.name = name;
this.namespaces = namespaces;
Expand All @@ -60,6 +63,7 @@ protected InformerConfiguration(
this.genericFilter = genericFilter;
this.itemStore = itemStore;
this.informerListLimit = informerListLimit;
this.fieldSelector = fieldSelector;
}

private InformerConfiguration(Class<R> resourceClass) {
Expand Down Expand Up @@ -93,7 +97,8 @@ public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
original.onDeleteFilter,
original.genericFilter,
original.itemStore,
original.informerListLimit)
original.informerListLimit,
original.fieldSelector)
.builder;
}

Expand Down Expand Up @@ -264,6 +269,10 @@ public Long getInformerListLimit() {
return informerListLimit;
}

public FieldSelector getFieldSelector() {
return fieldSelector;
}

@SuppressWarnings("UnusedReturnValue")
public class Builder {

Expand Down Expand Up @@ -329,6 +338,12 @@ public InformerConfiguration<R>.Builder initFromAnnotation(
final var informerListLimit =
informerListLimitValue == Constants.NO_LONG_VALUE_SET ? null : informerListLimitValue;
withInformerListLimit(informerListLimit);

withFieldSelector(
new FieldSelector(
Arrays.stream(informerConfig.fieldSelector())
.map(f -> new FieldSelector.Field(f.path(), f.value(), f.negated()))
.toList()));
}
return this;
}
Expand Down Expand Up @@ -424,5 +439,10 @@ public Builder withInformerListLimit(Long informerListLimit) {
InformerConfiguration.this.informerListLimit = informerListLimit;
return this;
}

public Builder withFieldSelector(FieldSelector fieldSelector) {
InformerConfiguration.this.fieldSelector = fieldSelector;
return this;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -265,6 +265,11 @@ public Builder<R> withInformerListLimit(Long informerListLimit) {
return this;
}

public Builder<R> withFieldSelector(FieldSelector fieldSelector) {
config.withFieldSelector(fieldSelector);
return this;
}

public void updateFrom(InformerConfiguration<R> informerConfig) {
if (informerConfig != null) {
final var informerConfigName = informerConfig.getName();
Expand All @@ -281,7 +286,8 @@ public void updateFrom(InformerConfiguration<R> informerConfig) {
.withOnUpdateFilter(informerConfig.getOnUpdateFilter())
.withOnDeleteFilter(informerConfig.getOnDeleteFilter())
.withGenericFilter(informerConfig.getGenericFilter())
.withInformerListLimit(informerConfig.getInformerListLimit());
.withInformerListLimit(informerConfig.getInformerListLimit())
.withFieldSelector(informerConfig.getFieldSelector());
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,17 @@
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.BiPredicate;

public abstract class BaseControl<T extends BaseControl<T>> {
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.expectation.Expectation;
import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationAdapter;
import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationContext;

public abstract class BaseControl<T extends BaseControl<T, P>, P extends HasMetadata> {

private Long scheduleDelay = null;
private Expectation<P> expectation;

public T rescheduleAfter(long delay) {
rescheduleAfter(Duration.ofMillis(delay));
Expand All @@ -25,4 +32,16 @@ public T rescheduleAfter(long delay, TimeUnit timeUnit) {
public Optional<Long> getScheduleDelay() {
return Optional.ofNullable(scheduleDelay);
}

public void expect(Expectation<P> expectation) {
this.expectation = expectation;
}

public void expect(BiPredicate<P, ExpectationContext<P>> expectation, Duration timeout) {
this.expectation = new ExpectationAdapter<>(expectation, timeout);
}

public Expectation<P> getExpectation() {
return expectation;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
package io.javaoperatorsdk.operator.api.reconciler;

import java.util.Optional;
import java.util.Set;
import java.util.stream.Stream;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;

public interface CacheAware<P extends HasMetadata> {
default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
return getSecondaryResource(expectedType, null);
}

<R> Set<R> getSecondaryResources(Class<R> expectedType);

default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
return getSecondaryResources(expectedType).stream();
}

<R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);

ControllerConfiguration<P> getControllerConfiguration();

/**
* Retrieves the primary resource.
*
* @return the primary resource associated with the current reconciliation
*/
P getPrimaryResource();

/**
* Retrieves the primary resource cache.
*
* @return the {@link IndexerResourceCache} associated with the associated {@link Reconciler} for
* this context
*/
@SuppressWarnings("unused")
IndexedResourceCache<P> getPrimaryCache();

EventSourceRetriever<P> eventSourceRetriever();
}
Original file line number Diff line number Diff line change
@@ -1,35 +1,18 @@
package io.javaoperatorsdk.operator.api.reconciler;

import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.stream.Stream;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;

public interface Context<P extends HasMetadata> {
public interface Context<P extends HasMetadata> extends CacheAware<P> {

Optional<RetryInfo> getRetryInfo();

default <R> Optional<R> getSecondaryResource(Class<R> expectedType) {
return getSecondaryResource(expectedType, null);
}

<R> Set<R> getSecondaryResources(Class<R> expectedType);

default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
return getSecondaryResources(expectedType).stream();
}

<R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName);

ControllerConfiguration<P> getControllerConfiguration();

/**
* Retrieve the {@link ManagedWorkflowAndDependentResourceContext} used to interact with {@link
* io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource}s and associated {@link
Expand All @@ -39,8 +22,6 @@ default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
*/
ManagedWorkflowAndDependentResourceContext managedWorkflowAndDependentResourceContext();

EventSourceRetriever<P> eventSourceRetriever();

KubernetesClient getClient();

/** ExecutorService initialized by framework for workflows. Used for workflow standalone mode. */
Expand Down Expand Up @@ -72,4 +53,6 @@ default <R> Stream<R> getSecondaryResourcesAsStream(Class<R> expectedType) {
* @return {@code true} is another reconciliation is already scheduled, {@code false} otherwise
*/
boolean isNextReconciliationImminent();

Optional<ExpectationResult<P>> expectationResult();
}
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedWorkflowAndDependentResourceContext;
import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
import io.javaoperatorsdk.operator.api.reconciler.expectation.ExpectationResult;
import io.javaoperatorsdk.operator.processing.Controller;
import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
import io.javaoperatorsdk.operator.processing.event.NoEventSourceForClassException;
Expand Down Expand Up @@ -119,6 +120,11 @@ public boolean isNextReconciliationImminent() {
.isNextReconciliationImminent(ResourceID.fromResource(primaryResource));
}

@Override
public Optional<ExpectationResult<P>> expectationResult() {
return Optional.empty();
}

public DefaultContext<P> setRetryInfo(RetryInfo retryInfo) {
this.retryInfo = retryInfo;
return this;
Expand Down
Loading