feat: access controller event source

caffeine-bounded-cache-support/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>java-operator-sdk</artifactId>
     <groupId>io.javaoperatorsdk</groupId>
-    <version>4.4.3-SNAPSHOT</version>
+    <version>4.5.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

micrometer-support/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>java-operator-sdk</artifactId>
     <groupId>io.javaoperatorsdk</groupId>
-    <version>4.4.3-SNAPSHOT</version>
+    <version>4.5.0-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

operator-framework-bom/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>operator-framework-bom</artifactId>
-    <version>4.4.3-SNAPSHOT</version>
+    <version>4.5.0-SNAPSHOT</version>
     <name>Operator SDK - Bill of Materials</name>
     <packaging>pom</packaging>
     <description>Java SDK for implementing Kubernetes operators</description>

operator-framework-core/pom.xml

@@ -6,7 +6,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>4.4.3-SNAPSHOT</version>
+    <version>4.5.0-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/LeaderElectionManager.java

@@ -66,19 +66,28 @@ private void init(LeaderElectionConfiguration config) {
                 config.getLeaseDuration(),
                 config.getRenewDeadline(),
                 config.getRetryPeriod(),
-                leaderCallbacks(),
+                leaderCallbacks(config),
                 true,
                 config.getLeaseName()))
         .build();
   }
 
 
 
-  private LeaderCallbacks leaderCallbacks() {
+  private LeaderCallbacks leaderCallbacks(LeaderElectionConfiguration config) {
     return new LeaderCallbacks(
-        this::startLeading,
-        this::stopLeading,
-        leader -> log.info("New leader with identity: {}", leader));
+        () -> {
+          config.getLeaderCallbacks().ifPresent(LeaderCallbacks::onStartLeading);
+          LeaderElectionManager.this.startLeading();
+        },
+        () -> {
+          config.getLeaderCallbacks().ifPresent(LeaderCallbacks::onStopLeading);
+          LeaderElectionManager.this.stopLeading();
+        },
+        leader -> {
+          config.getLeaderCallbacks().ifPresent(cb -> cb.onNewLeader(leader));
+          log.info("New leader with identity: {}", leader);
+        });
   }
 
   private void startLeading() {

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

@@ -3,6 +3,8 @@
 import java.time.Duration;
 import java.util.Optional;
 
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
+
 public class LeaderElectionConfiguration {
 
   public static final Duration LEASE_DURATION_DEFAULT_VALUE = Duration.ofSeconds(15);
@@ -17,13 +19,15 @@ public class LeaderElectionConfiguration {
   private final Duration renewDeadline;
   private final Duration retryPeriod;
 
+  private final LeaderCallbacks leaderCallbacks;
+
   public LeaderElectionConfiguration(String leaseName, String leaseNamespace, String identity) {
     this(
         leaseName,
         leaseNamespace,
         LEASE_DURATION_DEFAULT_VALUE,
         RENEW_DEADLINE_DEFAULT_VALUE,
-        RETRY_PERIOD_DEFAULT_VALUE, identity);
+        RETRY_PERIOD_DEFAULT_VALUE, identity, null);
   }
 
   public LeaderElectionConfiguration(String leaseName, String leaseNamespace) {
@@ -32,7 +36,7 @@ public LeaderElectionConfiguration(String leaseName, String leaseNamespace) {
         leaseNamespace,
         LEASE_DURATION_DEFAULT_VALUE,
         RENEW_DEADLINE_DEFAULT_VALUE,
-        RETRY_PERIOD_DEFAULT_VALUE, null);
+        RETRY_PERIOD_DEFAULT_VALUE, null, null);
   }
 
   public LeaderElectionConfiguration(String leaseName) {
@@ -41,7 +45,7 @@ public LeaderElectionConfiguration(String leaseName) {
         null,
         LEASE_DURATION_DEFAULT_VALUE,
         RENEW_DEADLINE_DEFAULT_VALUE,
-        RETRY_PERIOD_DEFAULT_VALUE, null);
+        RETRY_PERIOD_DEFAULT_VALUE, null, null);
   }
 
   public LeaderElectionConfiguration(
@@ -50,7 +54,7 @@ public LeaderElectionConfiguration(
       Duration leaseDuration,
       Duration renewDeadline,
       Duration retryPeriod) {
-    this(leaseName, leaseNamespace, leaseDuration, renewDeadline, retryPeriod, null);
+    this(leaseName, leaseNamespace, leaseDuration, renewDeadline, retryPeriod, null, null);
   }
 
   public LeaderElectionConfiguration(
@@ -59,13 +63,15 @@ public LeaderElectionConfiguration(
       Duration leaseDuration,
       Duration renewDeadline,
       Duration retryPeriod,
-      String identity) {
+      String identity,
+      LeaderCallbacks leaderCallbacks) {
     this.leaseName = leaseName;
     this.leaseNamespace = leaseNamespace;
     this.leaseDuration = leaseDuration;
     this.renewDeadline = renewDeadline;
     this.retryPeriod = retryPeriod;
     this.identity = identity;
+    this.leaderCallbacks = leaderCallbacks;
   }
 
   public Optional<String> getLeaseNamespace() {
@@ -91,4 +97,8 @@ public Duration getRetryPeriod() {
   public Optional<String> getIdentity() {
     return Optional.ofNullable(identity);
   }
+
+  public Optional<LeaderCallbacks> getLeaderCallbacks() {
+    return Optional.ofNullable(leaderCallbacks);
+  }
 }

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

@@ -0,0 +1,61 @@
+package io.javaoperatorsdk.operator.api.config;
+
+import java.time.Duration;
+
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
+
+import static io.javaoperatorsdk.operator.api.config.LeaderElectionConfiguration.*;
+
+public final class LeaderElectionConfigurationBuilder {
+
+  private String leaseName;
+  private String leaseNamespace;
+  private String identity;
+  private Duration leaseDuration = LEASE_DURATION_DEFAULT_VALUE;
+  private Duration renewDeadline = RENEW_DEADLINE_DEFAULT_VALUE;
+  private Duration retryPeriod = RETRY_PERIOD_DEFAULT_VALUE;
+  private LeaderCallbacks leaderCallbacks;
+
+  private LeaderElectionConfigurationBuilder(String leaseName) {
+    this.leaseName = leaseName;
+  }
+
+  public static LeaderElectionConfigurationBuilder aLeaderElectionConfiguration(String leaseName) {
+    return new LeaderElectionConfigurationBuilder(leaseName);
+  }
+
+  public LeaderElectionConfigurationBuilder withLeaseNamespace(String leaseNamespace) {
+    this.leaseNamespace = leaseNamespace;
+    return this;
+  }
+
+  public LeaderElectionConfigurationBuilder withIdentity(String identity) {
+    this.identity = identity;
+    return this;
+  }
+
+  public LeaderElectionConfigurationBuilder withLeaseDuration(Duration leaseDuration) {
+    this.leaseDuration = leaseDuration;
+    return this;
+  }
+
+  public LeaderElectionConfigurationBuilder withRenewDeadline(Duration renewDeadline) {
+    this.renewDeadline = renewDeadline;
+    return this;
+  }
+
+  public LeaderElectionConfigurationBuilder withRetryPeriod(Duration retryPeriod) {
+    this.retryPeriod = retryPeriod;
+    return this;
+  }
+
+  public LeaderElectionConfigurationBuilder withLeaderCallbacks(LeaderCallbacks leaderCallbacks) {
+    this.leaderCallbacks = leaderCallbacks;
+    return this;
+  }
+
+  public LeaderElectionConfiguration build() {
+    return new LeaderElectionConfiguration(leaseName, leaseNamespace, leaseDuration, renewDeadline,
+        retryPeriod, identity, leaderCallbacks);
+  }
+}

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

@@ -9,7 +9,7 @@ public abstract class BaseControl<T extends BaseControl<T>> {
   private Long scheduleDelay = null;
 
   public T rescheduleAfter(long delay) {
-    this.scheduleDelay = delay;
+    rescheduleAfter(Duration.ofMillis(delay));
     return (T) this;
   }
 

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

@@ -1,16 +1,17 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.time.Duration;
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
-public class ErrorStatusUpdateControl<P extends HasMetadata> {
+public class ErrorStatusUpdateControl<P extends HasMetadata>
+    extends BaseControl<ErrorStatusUpdateControl<P>> {
 
   private final P resource;
   private final boolean patch;
   private boolean noRetry = false;
 
-
   public static <T extends HasMetadata> ErrorStatusUpdateControl<T> patchStatus(T resource) {
     return new ErrorStatusUpdateControl<>(resource, true);
   }
@@ -49,4 +50,16 @@ public boolean isNoRetry() {
   public boolean isPatch() {
     return patch;
   }
+
+  /**
+   * If re-scheduled using this method, it is not considered as retry, it effectively cancels retry.
+   *
+   * @param delay for next execution
+   * @return ErrorStatusUpdateControl
+   */
+  @Override
+  public ErrorStatusUpdateControl<P> rescheduleAfter(Duration delay) {
+    withNoRetry();
+    return super.rescheduleAfter(delay);
+  }
 }

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

@@ -0,0 +1,50 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+import java.util.function.Function;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+
+/**
+ * Uses a custom index of {@link InformerEventSource} to access the target resource. The index needs
+ * to be explicitly created when the event source is defined. This approach improves the performance
+ * to access the resource.
+ */
+public class IndexDiscriminator<R extends HasMetadata, P extends HasMetadata>
+    implements ResourceDiscriminator<R, P> {
+
+  private final String indexName;
+  private final String eventSourceName;
+  private final Function<P, String> keyMapper;
+
+  public IndexDiscriminator(String indexName, Function<P, String> keyMapper) {
+    this(indexName, null, keyMapper);
+  }
+
+  public IndexDiscriminator(String indexName, String eventSourceName,
+      Function<P, String> keyMapper) {
+    this.indexName = indexName;
+    this.eventSourceName = eventSourceName;
+    this.keyMapper = keyMapper;
+  }
+
+  @Override
+  public Optional<R> distinguish(Class<R> resource,
+      P primary,
+      Context<P> context) {
+
+    InformerEventSource<R, P> eventSource =
+        (InformerEventSource<R, P>) context
+            .eventSourceRetriever()
+            .getResourceEventSourceFor(resource, eventSourceName);
+    var resources = eventSource.byIndex(indexName, keyMapper.apply(primary));
+    if (resources.isEmpty()) {
+      return Optional.empty();
+    } else if (resources.size() > 1) {
+      throw new IllegalStateException("More than one resource found");
+    } else {
+      return Optional.of(resources.get(0));
+    }
+  }
+}

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

@@ -5,21 +5,41 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.Cache;
 
 public class ResourceIDMatcherDiscriminator<R extends HasMetadata, P extends HasMetadata>
     implements ResourceDiscriminator<R, P> {
 
+
+  private final String eventSourceName;
   private final Function<P, ResourceID> mapper;
 
   public ResourceIDMatcherDiscriminator(Function<P, ResourceID> mapper) {
+    this(null, mapper);
+  }
+
+  public ResourceIDMatcherDiscriminator(String eventSourceName, Function<P, ResourceID> mapper) {
+    this.eventSourceName = eventSourceName;
     this.mapper = mapper;
   }
 
+  @SuppressWarnings("unchecked")
   @Override
   public Optional<R> distinguish(Class<R> resource, P primary, Context<P> context) {
     var resourceID = mapper.apply(primary);
-    return context.getSecondaryResourcesAsStream(resource)
-        .filter(resourceID::isSameResource)
-        .findFirst();
+    if (eventSourceName != null) {
+      return ((Cache<R>) context.eventSourceRetriever().getResourceEventSourceFor(resource,
+          eventSourceName))
+          .get(resourceID);
+    } else {
+      var eventSources = context.eventSourceRetriever().getResourceEventSourcesFor(resource);
+      if (eventSources.size() == 1) {
+        return ((Cache<R>) eventSources.get(0)).get(resourceID);
+      } else {
+        return context.getSecondaryResourcesAsStream(resource)
+            .filter(resourceID::isSameResource)
+            .findFirst();
+      }
+    }
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java

@@ -44,7 +44,7 @@ public interface DependentResource<R, P extends HasMetadata> {
    * @param eventSourceContext context of event source initialization
    * @return an optional event source
    */
-  default Optional<ResourceEventSource<R, P>> eventSource(
+  default Optional<? extends ResourceEventSource<R, P>> eventSource(
       EventSourceContext<P> eventSourceContext) {
     return Optional.empty();
   }