feat: shard selector support (#3429)

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

docs/content/en/docs/documentation/operations/configuration.md

@@ -324,6 +324,7 @@ All controller-level keys are prefixed with `josdk.controller.<controller-name>.
 | `josdk.controller.<name>.finalizer` | `String` | Finalizer string added to managed resources |
 | `josdk.controller.<name>.generation-aware` | `Boolean` | Skip reconciliation when the resource generation has not changed |
 | `josdk.controller.<name>.label-selector` | `String` | Label selector to filter watched resources |
+| `josdk.controller.<name>.shard-selector` | `String` | Shard selector to filter watched resources for sharding across operator replicas |
 | `josdk.controller.<name>.max-reconciliation-interval` | `Duration` | Maximum interval between reconciliations even without events |
 | `josdk.controller.<name>.field-manager` | `String` | Field manager name used for SSA operations |
 | `josdk.controller.<name>.trigger-reconciler-on-all-events` | `Boolean` | Trigger reconciliation on every event, not only meaningful changes |
@@ -333,6 +334,7 @@ All controller-level keys are prefixed with `josdk.controller.<controller-name>.
 | Key | Type | Description |
 |---|---|---|
 | `josdk.controller.<name>.informer.label-selector` | `String` | Label selector for the primary resource informer (alias for `label-selector`) |
+| `josdk.controller.<name>.informer.shard-selector` | `String` | Shard selector for the primary resource informer (alias for `shard-selector`) |
 | `josdk.controller.<name>.informer.list-limit` | `Long` | Page size for paginated informer list requests; omit for no pagination |
 
 #### Retry

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

@@ -136,6 +136,11 @@ public ControllerConfigurationOverrider<R> withLabelSelector(String labelSelecto
     return this;
   }
 
+  public ControllerConfigurationOverrider<R> withShardSelector(String shardSelector) {
+    config.withShardSelector(shardSelector);
+    return this;
+  }
+
   public ControllerConfigurationOverrider<R> withReconciliationMaxInterval(
       Duration reconciliationMaxInterval) {
     this.reconciliationMaxInterval = reconciliationMaxInterval;

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

@@ -71,6 +71,18 @@
    */
   String labelSelector() default NO_VALUE_SET;
 
+  /**
+   * Optional shard selector used to restrict the set of resources the associated informer will act
+   * upon to a single shard, typically when the same workload is split across several operator
+   * instances. Just like {@link #labelSelector()} it is expressed as a label selector and can be
+   * made of multiple comma separated requirements that act as a logical AND operator. When both a
+   * label selector and a shard selector are set, the resulting informer only watches resources
+   * matching both (the two selectors are combined with a logical AND).
+   *
+   * @return the shard selector
+   */
+  String shardSelector() default NO_VALUE_SET;
+
   /**
    * Optional {@link OnAddFilter} to filter add events sent to the associated informer
    *

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

@@ -47,6 +47,7 @@ public class InformerConfiguration<R extends HasMetadata> {
   private Set<String> namespaces;
   private Boolean followControllerNamespaceChanges;
   private String labelSelector;
+  private String shardSelector;
   private OnAddFilter<? super R> onAddFilter;
   private OnUpdateFilter<? super R> onUpdateFilter;
   private OnDeleteFilter<? super R> onDeleteFilter;
@@ -62,6 +63,7 @@ protected InformerConfiguration(
       Set<String> namespaces,
       boolean followControllerNamespaceChanges,
       String labelSelector,
+      String shardSelector,
       OnAddFilter<? super R> onAddFilter,
       OnUpdateFilter<? super R> onUpdateFilter,
       OnDeleteFilter<? super R> onDeleteFilter,
@@ -77,6 +79,7 @@ protected InformerConfiguration(
     this.namespaces = namespaces;
     this.followControllerNamespaceChanges = followControllerNamespaceChanges;
     this.labelSelector = labelSelector;
+    this.shardSelector = shardSelector;
     this.onAddFilter = onAddFilter;
     this.onUpdateFilter = onUpdateFilter;
     this.onDeleteFilter = onDeleteFilter;
@@ -113,6 +116,7 @@ public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
             original.namespaces,
             original.followControllerNamespaceChanges,
             original.labelSelector,
+            original.shardSelector,
             original.onAddFilter,
             original.onUpdateFilter,
             original.onDeleteFilter,
@@ -125,11 +129,6 @@ public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
         .builder;
   }
 
-  public static String ensureValidLabelSelector(String labelSelector) {
-    // might want to implement validation here?
-    return labelSelector;
-  }
-
   public static boolean allNamespacesWatched(Set<String> namespaces) {
     failIfNotValid(namespaces);
     return DEFAULT_NAMESPACES_SET.equals(namespaces);
@@ -251,6 +250,20 @@ public String getLabelSelector() {
     return labelSelector;
   }
 
+  /**
+   * Retrieves the shard selector that is used, in addition to the {@link #getLabelSelector() label
+   * selector}, to restrict which resources are actually watched by the associated informer.
+   * Typically used to assign a subset (shard) of the resources to a given operator instance. It is
+   * expressed using the same syntax as a label selector. See the official documentation on the <a
+   * href="https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/">topic</a> for
+   * more details on syntax.
+   *
+   * @return the shard selector filtering watched resources
+   */
+  public String getShardSelector() {
+    return shardSelector;
+  }
+
   public OnAddFilter<? super R> getOnAddFilter() {
     return onAddFilter;
   }
@@ -353,6 +366,11 @@ public InformerConfiguration<R>.Builder initFromAnnotation(
         var labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
         withLabelSelector(labelSelector);
 
+        final var shardFromAnnotation = informerConfig.shardSelector();
+        var shardSelector =
+            Constants.NO_VALUE_SET.equals(shardFromAnnotation) ? null : shardFromAnnotation;
+        withShardSelector(shardSelector);
+
         withOnAddFilter(
             Utils.instantiate(informerConfig.onAddFilter(), OnAddFilter.class, context));
 
@@ -442,7 +460,12 @@ public Builder withFollowControllerNamespacesChanges(boolean followChanges) {
     }
 
     public Builder withLabelSelector(String labelSelector) {
-      InformerConfiguration.this.labelSelector = ensureValidLabelSelector(labelSelector);
+      InformerConfiguration.this.labelSelector = labelSelector;
+      return this;
+    }
+
+    public Builder withShardSelector(String shardSelector) {
+      InformerConfiguration.this.shardSelector = shardSelector;
       return this;
     }
 

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

@@ -251,6 +251,11 @@ public Builder<R> withLabelSelector(String labelSelector) {
       return this;
     }
 
+    public Builder<R> withShardSelector(String shardSelector) {
+      config.withShardSelector(shardSelector);
+      return this;
+    }
+
     public Builder<R> withOnAddFilter(OnAddFilter<? super R> onAddFilter) {
       config.withOnAddFilter(onAddFilter);
       return this;
@@ -308,6 +313,7 @@ public void updateFrom(InformerConfiguration<R> informerConfig) {
             .withFollowControllerNamespacesChanges(
                 informerConfig.getFollowControllerNamespaceChanges())
             .withLabelSelector(informerConfig.getLabelSelector())
+            .withShardSelector(informerConfig.getShardSelector())
             .withItemStore(informerConfig.getItemStore())
             .withOnAddFilter(informerConfig.getOnAddFilter())
             .withOnUpdateFilter(informerConfig.getOnUpdateFilter())

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

@@ -136,13 +136,18 @@ public void changeNamespaces(Set<String> namespaces) {
   private InformerWrapper<R> createEventSourceForNamespace(String namespace) {
     final InformerWrapper<R> source;
     final var labelSelector = configuration.getInformerConfig().getLabelSelector();
+    final var shardSelector = configuration.getInformerConfig().getShardSelector();
     if (namespace.equals(WATCH_ALL_NAMESPACES)) {
-      final var filteredBySelectorClient = client.inAnyNamespace().withLabelSelector(labelSelector);
+      final var filteredBySelectorClient =
+          client.inAnyNamespace().withLabelSelector(labelSelector).withShardSelector(shardSelector);
       source = createEventSource(filteredBySelectorClient, eventHandler, WATCH_ALL_NAMESPACES);
     } else {
       source =
           createEventSource(
-              client.inNamespace(namespace).withLabelSelector(labelSelector),
+              client
+                  .inNamespace(namespace)
+                  .withLabelSelector(labelSelector)
+                  .withShardSelector(shardSelector),
               eventHandler,
               namespace);
     }
@@ -275,12 +280,14 @@ public List<R> byIndex(String indexName, String indexKey) {
   @Override
   public String toString() {
     final var informerConfig = configuration.getInformerConfig();
-    final var selector = informerConfig.getLabelSelector();
+    final var labelSelector = informerConfig.getLabelSelector();
+    final var shardSelector = informerConfig.getShardSelector();
     return "InformerManager ["
         + ReconcilerUtilsInternal.getResourceTypeNameWithVersion(configuration.getResourceClass())
         + "] watching: "
         + informerConfig.getEffectiveNamespaces(controllerConfiguration)
-        + (selector != null ? " selector: " + selector : "");
+        + (labelSelector != null ? " label selector: " + labelSelector : "")
+        + (shardSelector != null ? " shard selector: " + shardSelector : "");
   }
 
   public Map<String, InformerHealthIndicator> informerHealthIndicators() {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java

@@ -84,6 +84,7 @@ public static <T extends HasMetadata> KubernetesClient client(
     when(nonNamespaceOperation.withLabelSelector(nullable(String.class))).thenReturn(filterable);
     when(resources.inAnyNamespace()).thenReturn(inAnyNamespace);
     when(inAnyNamespace.withLabelSelector(nullable(String.class))).thenReturn(filterable);
+    when(filterable.withShardSelector(nullable(String.class))).thenReturn(filterable);
     SharedIndexInformer<T> informer = mock(SharedIndexInformer.class);
     CompletableFuture<Void> informerStartRes = new CompletableFuture<>();
     informerStartRes.complete(null);

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverriderTest.java

@@ -181,6 +181,20 @@ void itemStorePreserved() {
     assertNotNull(configuration.getInformerConfig().getItemStore());
   }
 
+  @Test
+  void shardSelectorShouldBePropagated() {
+    var configuration = createConfiguration(new WatchCurrentReconciler());
+    assertNull(configuration.getInformerConfig().getShardSelector());
+
+    final var shardSelector = "shard=1";
+    configuration =
+        ControllerConfigurationOverrider.override(configuration)
+            .withShardSelector(shardSelector)
+            .build();
+
+    assertEquals(shardSelector, configuration.getInformerConfig().getShardSelector());
+  }
+
   @Test
   void configuredDependentShouldNotChangeOnParentOverrideEvenWhenInitialConfigIsSame() {
     var configuration = createConfiguration(new OverriddenNSOnDepReconciler());

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java

@@ -73,6 +73,19 @@ void nullLabelSelectorByDefault() {
     assertNull(informerConfig.getLabelSelector());
   }
 
+  @Test
+  void nullShardSelectorByDefault() {
+    final var informerConfig = InformerConfiguration.builder(ConfigMap.class).build();
+    assertNull(informerConfig.getShardSelector());
+  }
+
+  @Test
+  void shardSelectorIsSetOnBuilder() {
+    final var informerConfig =
+        InformerConfiguration.builder(ConfigMap.class).withShardSelector("shard=1").build();
+    assertEquals("shard=1", informerConfig.getShardSelector());
+  }
+
   @Test
   void shouldWatchAllNamespacesByDefaultForControllers() {
     final var informerConfig = InformerConfiguration.builder(ConfigMap.class).buildForController();

operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java

@@ -143,6 +143,8 @@ public static ConfigLoader getDefault() {
               ControllerConfigurationOverrider::withGenerationAware),
           new ConfigBinding<>(
               "label-selector", String.class, ControllerConfigurationOverrider::withLabelSelector),
+          new ConfigBinding<>(
+              "shard-selector", String.class, ControllerConfigurationOverrider::withShardSelector),
           new ConfigBinding<>(
               "max-reconciliation-interval",
               Duration.class,
@@ -157,6 +159,10 @@ public static ConfigLoader getDefault() {
               "informer.label-selector",
               String.class,
               ControllerConfigurationOverrider::withLabelSelector),
+          new ConfigBinding<>(
+              "informer.shard-selector",
+              String.class,
+              ControllerConfigurationOverrider::withShardSelector),
           new ConfigBinding<>(
               "informer.list-limit",
               Long.class,