move replicas action functionality into AllocateAction (#32523)

Since replica counts and allocation rules are set separately, it is not always clear how many replicas are to be allocated in the allocate action. Moving the replicas action to occur at the same time as the allocate action, resolves this confusion that could end an undesired state. This means that the ReplicasAction is removed, and a new optional replicas parameter is added to AllocateAction.

Author: talevy

client/rest-high-level/src/test/java/org/elasticsearch/client/IndexLifecycleIT.java

@@ -68,7 +68,7 @@ public void testSetIndexLifecyclePolicy() throws Exception {
             "       \"cold\": {\n" +
             "         \"after\": \"2000s\",\n" +
             "         \"actions\": {\n" +
-            "          \"replicas\": {\n" +
+            "          \"allocate\": {\n" +
             "            \"number_of_replicas\": 0\n" +
             "          }\n" +
             "         }\n" +

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java

@@ -48,7 +48,6 @@
 import org.elasticsearch.xpack.core.indexlifecycle.LifecycleAction;
 import org.elasticsearch.xpack.core.indexlifecycle.LifecycleType;
 import org.elasticsearch.xpack.core.indexlifecycle.ReadOnlyAction;
-import org.elasticsearch.xpack.core.indexlifecycle.ReplicasAction;
 import org.elasticsearch.xpack.core.indexlifecycle.RolloverAction;
 import org.elasticsearch.xpack.core.indexlifecycle.ShrinkAction;
 import org.elasticsearch.xpack.core.indexlifecycle.TimeseriesLifecycleType;
@@ -405,7 +404,6 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, AllocateAction.NAME, AllocateAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, ForceMergeAction.NAME, ForceMergeAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, ReadOnlyAction.NAME, ReadOnlyAction::new),
-                new NamedWriteableRegistry.Entry(LifecycleAction.class, ReplicasAction.NAME, ReplicasAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, RolloverAction.NAME, RolloverAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, ShrinkAction.NAME, ShrinkAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, DeleteAction.NAME, DeleteAction::new)
@@ -449,7 +447,6 @@ public List<NamedXContentRegistry.Entry> getNamedXContent() {
                 new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(AllocateAction.NAME), AllocateAction::parse),
                 new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(ForceMergeAction.NAME), ForceMergeAction::parse),
                 new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(ReadOnlyAction.NAME), ReadOnlyAction::parse),
-                new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(ReplicasAction.NAME), ReplicasAction::parse),
                 new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(RolloverAction.NAME), RolloverAction::parse),
                 new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(ShrinkAction.NAME), ShrinkAction::parse),
                 new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(DeleteAction.NAME), DeleteAction::parse)

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/AllocateAction.java

@@ -27,20 +27,23 @@
 public class AllocateAction implements LifecycleAction {
 
     public static final String NAME = "allocate";
+    public static final ParseField NUMBER_OF_REPLICAS_FIELD = new ParseField("number_of_replicas");
     public static final ParseField INCLUDE_FIELD = new ParseField("include");
     public static final ParseField EXCLUDE_FIELD = new ParseField("exclude");
     public static final ParseField REQUIRE_FIELD = new ParseField("require");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<AllocateAction, Void> PARSER = new ConstructingObjectParser<>(NAME,
-            a -> new AllocateAction((Map<String, String>) a[0], (Map<String, String>) a[1], (Map<String, String>) a[2]));
+            a -> new AllocateAction((Integer) a[0], (Map<String, String>) a[1], (Map<String, String>) a[2], (Map<String, String>) a[3]));
 
     static {
+        PARSER.declareInt(ConstructingObjectParser.optionalConstructorArg(), NUMBER_OF_REPLICAS_FIELD);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.mapStrings(), INCLUDE_FIELD);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.mapStrings(), EXCLUDE_FIELD);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.mapStrings(), REQUIRE_FIELD);
     }
 
+    private final Integer numberOfReplicas;
     private final Map<String, String> include;
     private final Map<String, String> exclude;
     private final Map<String, String> require;
@@ -49,7 +52,7 @@ public static AllocateAction parse(XContentParser parser) {
         return PARSER.apply(parser, null);
     }
 
-    public AllocateAction(Map<String, String> include, Map<String, String> exclude, Map<String, String> require) {
+    public AllocateAction(Integer numberOfReplicas, Map<String, String> include, Map<String, String> exclude, Map<String, String> require) {
         if (include == null) {
             this.include = Collections.emptyMap();
         } else {
@@ -65,19 +68,27 @@ public AllocateAction(Map<String, String> include, Map<String, String> exclude,
         } else {
             this.require = require;
         }
-        if (this.include.isEmpty() && this.exclude.isEmpty() && this.require.isEmpty()) {
+        if (this.include.isEmpty() && this.exclude.isEmpty() && this.require.isEmpty() && numberOfReplicas == null) {
             throw new IllegalArgumentException(
                     "At least one of " + INCLUDE_FIELD.getPreferredName() + ", " + EXCLUDE_FIELD.getPreferredName() + " or "
                             + REQUIRE_FIELD.getPreferredName() + "must contain attributes for action " + NAME);
         }
+        if (numberOfReplicas != null && numberOfReplicas < 0) {
+            throw new IllegalArgumentException("[" + NUMBER_OF_REPLICAS_FIELD.getPreferredName() + "] must be >= 0");
+        }
+        this.numberOfReplicas = numberOfReplicas;
     }
 
     @SuppressWarnings("unchecked")
     public AllocateAction(StreamInput in) throws IOException {
-        this((Map<String, String>) in.readGenericValue(), (Map<String, String>) in.readGenericValue(),
+        this(in.readOptionalVInt(), (Map<String, String>) in.readGenericValue(), (Map<String, String>) in.readGenericValue(),
                 (Map<String, String>) in.readGenericValue());
     }
 
+    public Integer getNumberOfReplicas() {
+        return numberOfReplicas;
+    }
+
     public Map<String, String> getInclude() {
         return include;
     }
@@ -92,6 +103,7 @@ public Map<String, String> getRequire() {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        out.writeOptionalVInt(numberOfReplicas);
         out.writeGenericValue(include);
         out.writeGenericValue(exclude);
         out.writeGenericValue(require);
@@ -105,6 +117,9 @@ public String getWriteableName() {
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
+        if (numberOfReplicas != null) {
+            builder.field(NUMBER_OF_REPLICAS_FIELD.getPreferredName(), numberOfReplicas);
+        }
         builder.field(INCLUDE_FIELD.getPreferredName(), include);
         builder.field(EXCLUDE_FIELD.getPreferredName(), exclude);
         builder.field(REQUIRE_FIELD.getPreferredName(), require);
@@ -123,6 +138,9 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
         StepKey allocationRoutedKey = new StepKey(phase, NAME, AllocationRoutedStep.NAME);
 
         Settings.Builder newSettings = Settings.builder();
+        if (numberOfReplicas != null) {
+            newSettings.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, numberOfReplicas);
+        }
         include.forEach((key, value) -> newSettings.put(IndexMetaData.INDEX_ROUTING_INCLUDE_GROUP_SETTING.getKey() + key, value));
         exclude.forEach((key, value) -> newSettings.put(IndexMetaData.INDEX_ROUTING_EXCLUDE_GROUP_SETTING.getKey() + key, value));
         require.forEach((key, value) -> newSettings.put(IndexMetaData.INDEX_ROUTING_REQUIRE_GROUP_SETTING.getKey() + key, value));
@@ -140,7 +158,7 @@ public List<StepKey> toStepKeys(String phase) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(include, exclude, require);
+        return Objects.hash(numberOfReplicas, include, exclude, require);
     }
 
     @Override
@@ -152,7 +170,10 @@ public boolean equals(Object obj) {
             return false;
         }
         AllocateAction other = (AllocateAction) obj;
-        return Objects.equals(include, other.include) && Objects.equals(exclude, other.exclude) && Objects.equals(require, other.require);
+        return Objects.equals(numberOfReplicas, other.numberOfReplicas) &&
+            Objects.equals(include, other.include) &&
+            Objects.equals(exclude, other.exclude) &&
+            Objects.equals(require, other.require);
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/AllocationRoutedStep.java

@@ -54,15 +54,15 @@ public boolean getWaitOnAllShardCopies() {
 
     @Override
     public Result isConditionMet(Index index, ClusterState clusterState) {
-        if (ActiveShardCount.ALL.enoughShardsActive(clusterState, index.getName()) == false) {
-            logger.debug("[{}] lifecycle action for index [{}] cannot make progress because not all shards are active",
-                    getKey().getAction(), index.getName());
-            return new Result(false, new Info(-1, false));
-        }
         IndexMetaData idxMeta = clusterState.metaData().index(index);
         if (idxMeta == null) {
             throw new IndexNotFoundException("Index not found when executing " + getKey().getAction() + " lifecycle action.",
-                    index.getName());
+                index.getName());
+        }
+        if (ActiveShardCount.ALL.enoughShardsActive(clusterState, index.getName()) == false) {
+            logger.debug("[{}] lifecycle action for index [{}] cannot make progress because not all shards are active",
+                    getKey().getAction(), index.getName());
+            return new Result(false, new Info(idxMeta.getNumberOfReplicas(), -1, false));
         }
         // All the allocation attributes are already set so just need to check
         // if the allocation has happened
@@ -94,7 +94,7 @@ public Result isConditionMet(Index index, ClusterState clusterState) {
             logger.debug(
                     "[{}] lifecycle action for index [{}] waiting for [{}] shards " + "to be allocated to nodes matching the given filters",
                     getKey().getAction(), index, allocationPendingAllShards);
-            return new Result(false, new Info(allocationPendingAllShards, true));
+            return new Result(false, new Info(idxMeta.getNumberOfReplicas(), allocationPendingAllShards, true));
         } else {
             logger.debug("[{}] lifecycle action for index [{}] complete", getKey().getAction(), index);
             return new Result(true, null);
@@ -105,7 +105,7 @@ public Result isConditionMet(Index index, ClusterState clusterState) {
     public int hashCode() {
         return Objects.hash(super.hashCode(), waitOnAllShardCopies);
     }
-    
+
     @Override
     public boolean equals(Object obj) {
         if (obj == null) {
@@ -115,29 +115,33 @@ public boolean equals(Object obj) {
             return false;
         }
         AllocationRoutedStep other = (AllocationRoutedStep) obj;
-        return super.equals(obj) && 
+        return super.equals(obj) &&
                 Objects.equals(waitOnAllShardCopies, other.waitOnAllShardCopies);
     }
-    
+
     public static final class Info implements ToXContentObject {
 
+        private final long actualReplicas;
         private final long numberShardsLeftToAllocate;
         private final boolean allShardsActive;
         private final String message;
 
+        static final ParseField ACTUAL_REPLICAS = new ParseField("actual_replicas");
         static final ParseField SHARDS_TO_ALLOCATE = new ParseField("shards_left_to_allocate");
         static final ParseField ALL_SHARDS_ACTIVE = new ParseField("all_shards_active");
         static final ParseField MESSAGE = new ParseField("message");
         static final ConstructingObjectParser<Info, Void> PARSER = new ConstructingObjectParser<>("allocation_routed_step_info",
-                a -> new Info((long) a[0], (boolean) a[1]));
+                a -> new Info((long) a[0], (long) a[1], (boolean) a[2]));
         static {
+            PARSER.declareLong(ConstructingObjectParser.constructorArg(), ACTUAL_REPLICAS);
             PARSER.declareLong(ConstructingObjectParser.constructorArg(), SHARDS_TO_ALLOCATE);
             PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), ALL_SHARDS_ACTIVE);
             PARSER.declareString((i, s) -> {}, MESSAGE);
         }
 
-        public Info(long numberShardsLeftToMerge, boolean allShardsActive) {
-            this.numberShardsLeftToAllocate = numberShardsLeftToMerge;
+        public Info(long actualReplicas, long numberShardsLeftToAllocate, boolean allShardsActive) {
+            this.actualReplicas = actualReplicas;
+            this.numberShardsLeftToAllocate = numberShardsLeftToAllocate;
             this.allShardsActive = allShardsActive;
             if (allShardsActive == false) {
                 message = "Waiting for all shard copies to be active";
@@ -147,10 +151,14 @@ public Info(long numberShardsLeftToMerge, boolean allShardsActive) {
             }
         }
 
+        public long getActualReplicas() {
+            return actualReplicas;
+        }
+
         public long getNumberShardsLeftToAllocate() {
             return numberShardsLeftToAllocate;
         }
-        
+
         public boolean allShardsActive() {
             return allShardsActive;
         }
@@ -161,13 +169,14 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(MESSAGE.getPreferredName(), message);
             builder.field(SHARDS_TO_ALLOCATE.getPreferredName(), numberShardsLeftToAllocate);
             builder.field(ALL_SHARDS_ACTIVE.getPreferredName(), allShardsActive);
+            builder.field(ACTUAL_REPLICAS.getPreferredName(), actualReplicas);
             builder.endObject();
             return builder;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(numberShardsLeftToAllocate, allShardsActive);
+            return Objects.hash(actualReplicas, numberShardsLeftToAllocate, allShardsActive);
         }
 
         @Override
@@ -179,8 +188,9 @@ public boolean equals(Object obj) {
                 return false;
             }
             Info other = (Info) obj;
-            return Objects.equals(numberShardsLeftToAllocate, other.numberShardsLeftToAllocate) &&
-                    Objects.equals(allShardsActive, other.allShardsActive);
+            return Objects.equals(actualReplicas, other.actualReplicas) &&
+                Objects.equals(numberShardsLeftToAllocate, other.numberShardsLeftToAllocate) &&
+                Objects.equals(allShardsActive, other.allShardsActive);
         }
 
         @Override