Changing wait_for_snapshot to check start time rather than finish time

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

@@ -6,6 +6,8 @@
  */
 package org.elasticsearch.xpack.core.ilm;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.xcontent.ToXContentObject;
@@ -25,6 +27,7 @@
 public class WaitForSnapshotStep extends ClusterStateWaitStep {
 
     static final String NAME = "wait-for-snapshot";
+    private static final Logger logger = LogManager.getLogger(WaitForSnapshotStep.class);
 
     private static final String MESSAGE_FIELD = "message";
     private static final String POLICY_NOT_EXECUTED_MESSAGE = "waiting for policy '%s' to be executed since %s";
@@ -57,10 +60,30 @@ public Result isConditionMet(Index index, ClusterState clusterState) {
             throw error(POLICY_NOT_FOUND_MESSAGE, policy);
         }
         SnapshotLifecyclePolicyMetadata snapPolicyMeta = snapMeta.getSnapshotConfigurations().get(policy);
-        if (snapPolicyMeta.getLastSuccess() == null || snapPolicyMeta.getLastSuccess().getTimestamp() < phaseTime) {
+        if (snapPolicyMeta.getLastSuccess() == null || snapPolicyMeta.getLastSuccess().getSnapshotStartTimestamp() == null ||
+            snapPolicyMeta.getLastSuccess().getSnapshotStartTimestamp() < phaseTime) {
+            if (snapPolicyMeta.getLastSuccess() == null) {
+                logger.debug("skipping ILM policy execution because there is no last snapshot success, phase time: {}", phaseTime);
+            } else if (snapPolicyMeta.getLastSuccess().getSnapshotStartTimestamp() == null) {
+                /*
+                 * This is because we are running in mixed cluster mode, and the snapshot was taken on an older master, which then went
+                 * down before this check could happen. We'll wait until a snapshot is taken on this newer master before passing this check.
+                 */
+                logger.debug("skipping ILM policy execution because no last snapshot start date, phase time: {}", phaseTime);
+            }
+            else {
+                logger.debug("skipping ILM policy execution because snapshot start time {} is before phase time {}, snapshot timestamp " +
+                        "is {}",
+                    snapPolicyMeta.getLastSuccess().getSnapshotStartTimestamp(),
+                    phaseTime,
+                    snapPolicyMeta.getLastSuccess().getSnapshotFinishTimestamp());
+            }
             return new Result(false, notExecutedMessage(phaseTime));
         }
-
+        logger.debug("executing policy because snapshot start time {} is after phase time {}, snapshot timestamp is {}",
+            snapPolicyMeta.getLastSuccess().getSnapshotStartTimestamp(),
+            phaseTime,
+            snapPolicyMeta.getLastSuccess().getSnapshotFinishTimestamp());
         return new Result(true, null);
     }
 

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

@@ -7,16 +7,18 @@
 
 package org.elasticsearch.xpack.core.slm;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.AbstractDiffable;
 import org.elasticsearch.cluster.Diffable;
-import org.elasticsearch.common.xcontent.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ParseField;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.core.Nullable;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -29,19 +31,22 @@ public class SnapshotInvocationRecord extends AbstractDiffable<SnapshotInvocatio
     implements Writeable, ToXContentObject, Diffable<SnapshotInvocationRecord> {
 
     static final ParseField SNAPSHOT_NAME = new ParseField("snapshot_name");
+    static final ParseField START_TIMESTAMP = new ParseField("start_time");
     static final ParseField TIMESTAMP = new ParseField("time");
     static final ParseField DETAILS = new ParseField("details");
 
     private String snapshotName;
-    private long timestamp;
+    private Long snapshotStartTimestamp;
+    private long snapshotFinishTimestamp;
     private String details;
 
     public static final ConstructingObjectParser<SnapshotInvocationRecord, String> PARSER =
         new ConstructingObjectParser<>("snapshot_policy_invocation_record", true,
-            a -> new SnapshotInvocationRecord((String) a[0], (long) a[1], (String) a[2]));
+            a -> new SnapshotInvocationRecord((String) a[0], (Long) a[1], (long) a[2], (String) a[3]));
 
     static {
         PARSER.declareString(ConstructingObjectParser.constructorArg(), SNAPSHOT_NAME);
+        PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), START_TIMESTAMP);
         PARSER.declareLong(ConstructingObjectParser.constructorArg(), TIMESTAMP);
         PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), DETAILS);
     }
@@ -50,24 +55,35 @@ public static SnapshotInvocationRecord parse(XContentParser parser, String name)
         return PARSER.apply(parser, name);
     }
 
-    public SnapshotInvocationRecord(String snapshotName, long timestamp, String details) {
+    public SnapshotInvocationRecord(String snapshotName, Long snapshotStartTimestamp, long snapshotFinishTimestamp, String details) {
         this.snapshotName = Objects.requireNonNull(snapshotName, "snapshot name must be provided");
-        this.timestamp = timestamp;
+        this.snapshotStartTimestamp = snapshotStartTimestamp;
+        this.snapshotFinishTimestamp = snapshotFinishTimestamp;
         this.details = details;
     }
 
     public SnapshotInvocationRecord(StreamInput in) throws IOException {
         this.snapshotName = in.readString();
-        this.timestamp = in.readVLong();
+        if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+            this.snapshotStartTimestamp = in.readOptionalVLong();
+        } else {
+            this.snapshotStartTimestamp = null;
+        }
+        this.snapshotFinishTimestamp = in.readVLong();
         this.details = in.readOptionalString();
     }
 
     public String getSnapshotName() {
         return snapshotName;
     }
 
-    public long getTimestamp() {
-        return timestamp;
+    @Nullable
+    public Long getSnapshotStartTimestamp() {
+        return snapshotStartTimestamp;
+    }
+
+    public long getSnapshotFinishTimestamp() {
+        return snapshotFinishTimestamp;
     }
 
     public String getDetails() {
@@ -77,7 +93,10 @@ public String getDetails() {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(snapshotName);
-        out.writeVLong(timestamp);
+        if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+            out.writeOptionalVLong(snapshotStartTimestamp);
+        }
+        out.writeVLong(snapshotFinishTimestamp);
         out.writeOptionalString(details);
     }
 
@@ -86,7 +105,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.startObject();
         {
             builder.field(SNAPSHOT_NAME.getPreferredName(), snapshotName);
-            builder.timeField(TIMESTAMP.getPreferredName(), "time_string", timestamp);
+            if (snapshotStartTimestamp != null) {
+                builder.timeField(START_TIMESTAMP.getPreferredName(), "start_time_string", snapshotStartTimestamp);
+            }
+            builder.timeField(TIMESTAMP.getPreferredName(), "time_string", snapshotFinishTimestamp);
             if (Objects.nonNull(details)) {
                 builder.field(DETAILS.getPreferredName(), details);
             }
@@ -100,13 +122,14 @@ public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         SnapshotInvocationRecord that = (SnapshotInvocationRecord) o;
-        return getTimestamp() == that.getTimestamp() &&
+        return getSnapshotFinishTimestamp() == that.getSnapshotFinishTimestamp() &&
+            Objects.equals(getSnapshotStartTimestamp(), that.getSnapshotStartTimestamp()) &&
             Objects.equals(getSnapshotName(), that.getSnapshotName()) &&
             Objects.equals(getDetails(), that.getDetails());
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(getSnapshotName(), getTimestamp(), getDetails());
+        return Objects.hash(getSnapshotName(), getSnapshotStartTimestamp(), getSnapshotFinishTimestamp(), getDetails());
     }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/WaitForSnapshotStepTests.java

@@ -95,13 +95,30 @@ public void testSlmPolicyNotExecuted() throws IOException {
     }
 
     public void testSlmPolicyExecutedBeforeStep() throws IOException {
+        // The snapshot was started and finished before the phase time, so we do not expect the step to finish:
+        assertSlmPolicyExecuted(false, false);
+    }
+
+    public void testSlmPolicyExecutedAfterStep() throws IOException {
+        // The snapshot was started and finished after the phase time, so we do expect the step to finish:
+        assertSlmPolicyExecuted(true, true);
+    }
+
+    public void testSlmPolicyNotExecutedWhenStartIsBeforePhaseTime() throws IOException {
+        // The snapshot was started before the phase time and finished after, so we do expect the step to finish:
+        assertSlmPolicyExecuted(false, true);
+    }
+
+    private void assertSlmPolicyExecuted(boolean startTimeAfterPhaseTime, boolean finishTimeAfterPhaseTime) throws IOException {
         long phaseTime = randomLong();
 
         WaitForSnapshotStep instance = createRandomInstance();
         SnapshotLifecyclePolicyMetadata slmPolicy = SnapshotLifecyclePolicyMetadata.builder()
             .setModifiedDate(randomLong())
             .setPolicy(new SnapshotLifecyclePolicy("", "", "", "", null, null))
-            .setLastSuccess(new SnapshotInvocationRecord("", phaseTime - 10, ""))
+            .setLastSuccess(new SnapshotInvocationRecord("",
+                phaseTime + (startTimeAfterPhaseTime ? 10 : -100),
+                phaseTime + (finishTimeAfterPhaseTime ? 100 : -10), ""))
             .build();
         SnapshotLifecycleMetadata smlMetadata = new SnapshotLifecycleMetadata(Map.of(instance.getPolicy(), slmPolicy),
             OperationMode.RUNNING, null);
@@ -115,18 +132,25 @@ public void testSlmPolicyExecutedBeforeStep() throws IOException {
         Metadata.Builder meta = Metadata.builder().indices(indices.build()).putCustom(SnapshotLifecycleMetadata.TYPE, smlMetadata);
         ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).metadata(meta).build();
         ClusterStateWaitStep.Result result = instance.isConditionMet(indexMetadata.getIndex(), clusterState);
-        assertFalse(result.isComplete());
-        assertTrue(getMessage(result).contains("to be executed"));
+        if (startTimeAfterPhaseTime) {
+            assertTrue(result.isComplete());
+            assertNull(result.getInfomationContext());
+        } else {
+            assertFalse(result.isComplete());
+            assertTrue(getMessage(result).contains("to be executed"));
+        }
     }
 
-    public void testSlmPolicyExecutedAfterStep() throws IOException {
+    public void testNullStartTime() throws IOException {
         long phaseTime = randomLong();
 
         WaitForSnapshotStep instance = createRandomInstance();
         SnapshotLifecyclePolicyMetadata slmPolicy = SnapshotLifecyclePolicyMetadata.builder()
             .setModifiedDate(randomLong())
             .setPolicy(new SnapshotLifecyclePolicy("", "", "", "", null, null))
-            .setLastSuccess(new SnapshotInvocationRecord("", phaseTime + 10, ""))
+            .setLastSuccess(new SnapshotInvocationRecord("",
+                null,
+                phaseTime + 100, ""))
             .build();
         SnapshotLifecycleMetadata smlMetadata = new SnapshotLifecycleMetadata(Map.of(instance.getPolicy(), slmPolicy),
             OperationMode.RUNNING, null);
@@ -140,8 +164,8 @@ public void testSlmPolicyExecutedAfterStep() throws IOException {
         Metadata.Builder meta = Metadata.builder().indices(indices.build()).putCustom(SnapshotLifecycleMetadata.TYPE, smlMetadata);
         ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).metadata(meta).build();
         ClusterStateWaitStep.Result result = instance.isConditionMet(indexMetadata.getIndex(), clusterState);
-        assertTrue(result.isComplete());
-        assertNull(result.getInfomationContext());
+        assertFalse(result.isComplete());
+        assertTrue(getMessage(result).contains("to be executed"));
     }
 
     private String getMessage(ClusterStateWaitStep.Result result) throws IOException {

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/slm/SnapshotInvocationRecordTests.java

@@ -37,15 +37,17 @@ protected SnapshotInvocationRecord mutateInstance(SnapshotInvocationRecord insta
             case 0:
                 return new SnapshotInvocationRecord(
                     randomValueOtherThan(instance.getSnapshotName(), () -> randomAlphaOfLengthBetween(2,10)),
-                    instance.getTimestamp(),
+                    instance.getSnapshotFinishTimestamp() - 100,
+                    instance.getSnapshotFinishTimestamp(),
                     instance.getDetails());
             case 1:
+                long timestamp = randomValueOtherThan(instance.getSnapshotFinishTimestamp(), ESTestCase::randomNonNegativeLong);
                 return new SnapshotInvocationRecord(instance.getSnapshotName(),
-                    randomValueOtherThan(instance.getTimestamp(), ESTestCase::randomNonNegativeLong),
+                    timestamp - 100, timestamp,
                     instance.getDetails());
             case 2:
                 return new SnapshotInvocationRecord(instance.getSnapshotName(),
-                    instance.getTimestamp(),
+                    instance.getSnapshotFinishTimestamp() - 100, instance.getSnapshotFinishTimestamp(),
                     randomValueOtherThan(instance.getDetails(), () -> randomAlphaOfLengthBetween(2,10)));
             default:
                 throw new AssertionError("failure, got illegal switch case");
@@ -55,8 +57,18 @@ protected SnapshotInvocationRecord mutateInstance(SnapshotInvocationRecord insta
     public static SnapshotInvocationRecord randomSnapshotInvocationRecord() {
         return new SnapshotInvocationRecord(
             randomAlphaOfLengthBetween(5,10),
+            randomNonNegativeNullableLong(),
             randomNonNegativeLong(),
             randomBoolean() ? null : randomAlphaOfLengthBetween(5, 10));
     }
 
+    private static Long randomNonNegativeNullableLong() {
+        long value = randomLong();
+        if (value < 0) {
+            return null;
+        } else {
+            return value;
+        }
+    }
+
 }