diff --git a/docs/changelog/124352.yaml b/docs/changelog/124352.yaml new file mode 100644 index 0000000000000..35657a32b5dcd --- /dev/null +++ b/docs/changelog/124352.yaml @@ -0,0 +1,6 @@ +pr: 124352 +summary: Add `force_merge_max_num_segments` option to downsample api and downsample + ilm action +area: Downsampling +type: enhancement +issues: [] diff --git a/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java b/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java index a39f117a10a52..dbfddf1e5a9c6 100644 --- a/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java +++ b/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java @@ -1038,7 +1038,7 @@ public void testLifecycleAppliedToFailureStore() throws Exception { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(10), - new DownsampleConfig(new DateHistogramInterval("10m")) + new DownsampleConfig(new DateHistogramInterval("10m"), null) ) ) ) diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/MetadataIndexTemplateServiceTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/MetadataIndexTemplateServiceTests.java index b5f9fbbabfeef..219c6b7acbcd2 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/MetadataIndexTemplateServiceTests.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/MetadataIndexTemplateServiceTests.java @@ -257,7 +257,7 @@ private static DataStreamLifecycle.Downsampling randomNonEmptyDownsampling() { List rounds = new ArrayList<>(); var previous = new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(randomIntBetween(1, 365)), - new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h")) + new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h"), null) ); rounds.add(previous); for (int i = 0; i < count; i++) { @@ -271,7 +271,8 @@ private static DataStreamLifecycle.Downsampling randomNonEmptyDownsampling() { private static DataStreamLifecycle.Downsampling.Round nextRound(DataStreamLifecycle.Downsampling.Round previous) { var after = TimeValue.timeValueDays(previous.after().days() + randomIntBetween(1, 10)); var fixedInterval = new DownsampleConfig( - new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms") + new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms"), + null ); return new DataStreamLifecycle.Downsampling.Round(after, fixedInterval); } diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleFixtures.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleFixtures.java index 0ab105a467ab3..5de628386ff93 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleFixtures.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleFixtures.java @@ -154,7 +154,7 @@ private static DataStreamLifecycle.Downsampling randomDownsampling() { List rounds = new ArrayList<>(); var previous = new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(randomIntBetween(1, 365)), - new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h")) + new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h"), null) ); rounds.add(previous); for (int i = 0; i < count; i++) { @@ -170,7 +170,8 @@ private static DataStreamLifecycle.Downsampling randomDownsampling() { private static DataStreamLifecycle.Downsampling.Round nextRound(DataStreamLifecycle.Downsampling.Round previous) { var after = TimeValue.timeValueDays(previous.after().days() + randomIntBetween(1, 10)); var fixedInterval = new DownsampleConfig( - new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms") + new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms"), + null ); return new DataStreamLifecycle.Downsampling.Round(after, fixedInterval); } diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java index 8173e8e93887d..369d92f0b6e94 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java @@ -1227,7 +1227,7 @@ public void testDownsampling() throws Exception { DataStreamLifecycle.newBuilder() .downsampling( new Downsampling( - List.of(new Round(TimeValue.timeValueMillis(0), new DownsampleConfig(new DateHistogramInterval("5m")))) + List.of(new Round(TimeValue.timeValueMillis(0), new DownsampleConfig(new DateHistogramInterval("5m"), null))) ) ) .dataRetention(TimeValue.MAX_VALUE) @@ -1365,7 +1365,7 @@ public void testDownsamplingWhenTargetIndexNameClashYieldsException() throws Exc DataStreamLifecycle.newBuilder() .downsampling( new Downsampling( - List.of(new Round(TimeValue.timeValueMillis(0), new DownsampleConfig(new DateHistogramInterval("5m")))) + List.of(new Round(TimeValue.timeValueMillis(0), new DownsampleConfig(new DateHistogramInterval("5m"), null))) ) ) .dataRetention(TimeValue.MAX_VALUE) @@ -1616,7 +1616,7 @@ private ClusterState downsampleSetup(String dataStreamName, IndexMetadata.Downsa DataStreamLifecycle.newBuilder() .downsampling( new Downsampling( - List.of(new Round(TimeValue.timeValueMillis(0), new DownsampleConfig(new DateHistogramInterval("5m")))) + List.of(new Round(TimeValue.timeValueMillis(0), new DownsampleConfig(new DateHistogramInterval("5m"), null))) ) ) .dataRetention(TimeValue.timeValueMillis(1)) diff --git a/server/src/main/java/org/elasticsearch/TransportVersions.java b/server/src/main/java/org/elasticsearch/TransportVersions.java index 36ec1979ccb28..2bddc630de9d4 100644 --- a/server/src/main/java/org/elasticsearch/TransportVersions.java +++ b/server/src/main/java/org/elasticsearch/TransportVersions.java @@ -184,6 +184,7 @@ static TransportVersion def(int id) { public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM = def(9_023_0_00); public static final TransportVersion MAX_OPERATION_SIZE_REJECTIONS_ADDED = def(9_024_0_00); public static final TransportVersion RETRY_ILM_ASYNC_ACTION_REQUIRE_ERROR = def(9_025_0_00); + public static final TransportVersion DOWNSAMPLE_FORCE_MERGE_MAX_NUM_SEGMENTS_PARAMETER = def(9_026_0_00); /* * STOP! READ THIS FIRST! No, really, diff --git a/server/src/main/java/org/elasticsearch/action/downsample/DownsampleConfig.java b/server/src/main/java/org/elasticsearch/action/downsample/DownsampleConfig.java index 43d722cdcae3b..db651ac5e9422 100644 --- a/server/src/main/java/org/elasticsearch/action/downsample/DownsampleConfig.java +++ b/server/src/main/java/org/elasticsearch/action/downsample/DownsampleConfig.java @@ -9,6 +9,7 @@ package org.elasticsearch.action.downsample; +import org.elasticsearch.TransportVersions; import org.elasticsearch.cluster.metadata.IndexMetadata; import org.elasticsearch.common.Rounding; import org.elasticsearch.common.Strings; @@ -31,6 +32,7 @@ import java.util.Objects; import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg; +import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg; /** * This class holds the configuration details of a DownsampleAction that downsamples time series @@ -56,11 +58,14 @@ public class DownsampleConfig implements NamedWriteable, ToXContentObject { private static final String NAME = "downsample/action/config"; public static final String FIXED_INTERVAL = "fixed_interval"; + public static final String FORCE_MERGE_MAX_NUM_SEGMENTS = "force_merge_max_num_segments"; public static final String TIME_ZONE = "time_zone"; public static final String DEFAULT_TIMEZONE = ZoneId.of("UTC").getId(); + private static final int DEFAULT_MAX_NUM_SEGMENTS = 1; private static final String timestampField = DataStreamTimestampFieldMapper.DEFAULT_PATH; private final DateHistogramInterval fixedInterval; + private final int forceMergeMaxNumSegments; private final String timeZone = DEFAULT_TIMEZONE; private final String intervalType = FIXED_INTERVAL; @@ -68,8 +73,9 @@ public class DownsampleConfig implements NamedWriteable, ToXContentObject { static { PARSER = new ConstructingObjectParser<>(NAME, a -> { DateHistogramInterval fixedInterval = (DateHistogramInterval) a[0]; + Integer forceMergeMaxNumSegments = (Integer) a[1]; if (fixedInterval != null) { - return new DownsampleConfig(fixedInterval); + return new DownsampleConfig(fixedInterval, forceMergeMaxNumSegments); } else { throw new IllegalArgumentException("Parameter [" + FIXED_INTERVAL + "] is required."); } @@ -81,13 +87,19 @@ public class DownsampleConfig implements NamedWriteable, ToXContentObject { new ParseField(FIXED_INTERVAL), ObjectParser.ValueType.STRING ); + PARSER.declareField( + optionalConstructorArg(), + p -> p.intValue(), + new ParseField(FORCE_MERGE_MAX_NUM_SEGMENTS), + ObjectParser.ValueType.INT + ); } /** * Create a new {@link DownsampleConfig} using the given configuration parameters. * @param fixedInterval the fixed interval to use for computing the date histogram for the rolled up documents (required). */ - public DownsampleConfig(final DateHistogramInterval fixedInterval) { + public DownsampleConfig(final DateHistogramInterval fixedInterval, Integer forceMergeMaxNumSegments) { if (fixedInterval == null) { throw new IllegalArgumentException("Parameter [" + FIXED_INTERVAL + "] is required."); } @@ -95,10 +107,20 @@ public DownsampleConfig(final DateHistogramInterval fixedInterval) { // validate interval createRounding(this.fixedInterval.toString(), this.timeZone); + + if (forceMergeMaxNumSegments == null) { + forceMergeMaxNumSegments = 1; + } + this.forceMergeMaxNumSegments = forceMergeMaxNumSegments; } public DownsampleConfig(final StreamInput in) throws IOException { fixedInterval = new DateHistogramInterval(in); + if (in.getTransportVersion().onOrAfter(TransportVersions.DOWNSAMPLE_FORCE_MERGE_MAX_NUM_SEGMENTS_PARAMETER)) { + forceMergeMaxNumSegments = in.readInt(); + } else { + forceMergeMaxNumSegments = DEFAULT_MAX_NUM_SEGMENTS; + } } /** @@ -135,6 +157,9 @@ public static void validateSourceAndTargetIntervals(DownsampleConfig source, Dow @Override public void writeTo(final StreamOutput out) throws IOException { fixedInterval.writeTo(out); + if (out.getTransportVersion().onOrAfter(TransportVersions.DOWNSAMPLE_FORCE_MERGE_MAX_NUM_SEGMENTS_PARAMETER)) { + out.writeInt(forceMergeMaxNumSegments); + } } /** @@ -180,6 +205,10 @@ public Rounding.Prepared createRounding() { return createRounding(fixedInterval.toString(), timeZone); } + public int getForceMergeMaxNumSegments() { + return forceMergeMaxNumSegments; + } + @Override public String getWriteableName() { return NAME; @@ -195,7 +224,11 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa } public XContentBuilder toXContentFragment(final XContentBuilder builder) throws IOException { - return builder.field(FIXED_INTERVAL, fixedInterval.toString()); + builder.field(FIXED_INTERVAL, fixedInterval.toString()); + if (forceMergeMaxNumSegments != DEFAULT_MAX_NUM_SEGMENTS) { + builder.field(FORCE_MERGE_MAX_NUM_SEGMENTS, forceMergeMaxNumSegments); + } + return builder; } public static DownsampleConfig fromXContent(final XContentParser parser) throws IOException { @@ -212,13 +245,14 @@ public boolean equals(final Object other) { } final DownsampleConfig that = (DownsampleConfig) other; return Objects.equals(fixedInterval, that.fixedInterval) + && Objects.equals(forceMergeMaxNumSegments, that.forceMergeMaxNumSegments) && Objects.equals(intervalType, that.intervalType) && ZoneId.of(timeZone, ZoneId.SHORT_IDS).getRules().equals(ZoneId.of(that.timeZone, ZoneId.SHORT_IDS).getRules()); } @Override public int hashCode() { - return Objects.hash(fixedInterval, intervalType, ZoneId.of(timeZone)); + return Objects.hash(fixedInterval, forceMergeMaxNumSegments, intervalType, ZoneId.of(timeZone)); } @Override diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamLifecycle.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamLifecycle.java index fbee25d0f5453..5f4473af6e636 100644 --- a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamLifecycle.java +++ b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamLifecycle.java @@ -488,7 +488,7 @@ public record Round(TimeValue after, DownsampleConfig config) implements Writeab private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( "downsampling_round", false, - (args, unused) -> new Round((TimeValue) args[0], new DownsampleConfig((DateHistogramInterval) args[1])) + (args, unused) -> new Round((TimeValue) args[0], new DownsampleConfig((DateHistogramInterval) args[1], null)) ); static { diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamLifecycleTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamLifecycleTests.java index 7cf3768329ddf..82a53c1a2586f 100644 --- a/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamLifecycleTests.java +++ b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamLifecycleTests.java @@ -182,11 +182,11 @@ public void testInvalidDownsamplingConfiguration() { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(10), - new DownsampleConfig(new DateHistogramInterval("2h")) + new DownsampleConfig(new DateHistogramInterval("2h"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(3), - new DownsampleConfig(new DateHistogramInterval("2h")) + new DownsampleConfig(new DateHistogramInterval("2h"), null) ) ) ) @@ -203,11 +203,11 @@ public void testInvalidDownsamplingConfiguration() { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(10), - new DownsampleConfig(new DateHistogramInterval("2h")) + new DownsampleConfig(new DateHistogramInterval("2h"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(30), - new DownsampleConfig(new DateHistogramInterval("2h")) + new DownsampleConfig(new DateHistogramInterval("2h"), null) ) ) ) @@ -221,11 +221,11 @@ public void testInvalidDownsamplingConfiguration() { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(10), - new DownsampleConfig(new DateHistogramInterval("2h")) + new DownsampleConfig(new DateHistogramInterval("2h"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(30), - new DownsampleConfig(new DateHistogramInterval("3h")) + new DownsampleConfig(new DateHistogramInterval("3h"), null) ) ) ) @@ -248,7 +248,7 @@ public void testInvalidDownsamplingConfiguration() { .map( i -> new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(i), - new DownsampleConfig(new DateHistogramInterval(i + "h")) + new DownsampleConfig(new DateHistogramInterval(i + "h"), randomBoolean() ? null : randomIntBetween(-1, 128)) ) ) .toList() @@ -264,7 +264,7 @@ public void testInvalidDownsamplingConfiguration() { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(10), - new DownsampleConfig(new DateHistogramInterval("2m")) + new DownsampleConfig(new DateHistogramInterval("2m"), null) ) ) ) @@ -420,9 +420,10 @@ static DataStreamLifecycle.Downsampling randomDownsampling() { default -> { var count = randomIntBetween(0, 9); List rounds = new ArrayList<>(); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); var previous = new DataStreamLifecycle.Downsampling.Round( randomTimeValue(1, 365, TimeUnit.DAYS), - new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h")) + new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h"), forceMergeMaxNumSegments) ); rounds.add(previous); for (int i = 0; i < count; i++) { @@ -437,8 +438,10 @@ static DataStreamLifecycle.Downsampling randomDownsampling() { private static DataStreamLifecycle.Downsampling.Round nextRound(DataStreamLifecycle.Downsampling.Round previous) { var after = TimeValue.timeValueDays(previous.after().days() + randomIntBetween(1, 10)); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); var fixedInterval = new DownsampleConfig( - new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms") + new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms"), + forceMergeMaxNumSegments ); return new DataStreamLifecycle.Downsampling.Round(after, fixedInterval); } diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java index f7a2530075713..46eada2a5e068 100644 --- a/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java +++ b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java @@ -1647,15 +1647,15 @@ public void testGetDownsampleRounds() { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(2000), - new DownsampleConfig(new DateHistogramInterval("10m")) + new DownsampleConfig(new DateHistogramInterval("10m"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(3200), - new DownsampleConfig(new DateHistogramInterval("100m")) + new DownsampleConfig(new DateHistogramInterval("100m"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(3500), - new DownsampleConfig(new DateHistogramInterval("1000m")) + new DownsampleConfig(new DateHistogramInterval("1000m"), null) ) ) ) @@ -1708,15 +1708,15 @@ public void testGetDownsampleRounds() { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(2000), - new DownsampleConfig(new DateHistogramInterval("10m")) + new DownsampleConfig(new DateHistogramInterval("10m"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(3200), - new DownsampleConfig(new DateHistogramInterval("100m")) + new DownsampleConfig(new DateHistogramInterval("100m"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(3500), - new DownsampleConfig(new DateHistogramInterval("1000m")) + new DownsampleConfig(new DateHistogramInterval("1000m"), null) ) ) ) diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateServiceTests.java index 1c60a9ff7cea3..737f2301a2913 100644 --- a/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateServiceTests.java +++ b/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateServiceTests.java @@ -1501,7 +1501,7 @@ public void testResolveLifecycle() throws Exception { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueDays(30), - new DownsampleConfig(new DateHistogramInterval("3h")) + new DownsampleConfig(new DateHistogramInterval("3h"), null) ) ) ) diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java index d1e142a20cd5c..52ff3bedf02f9 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java @@ -52,10 +52,11 @@ public class DownsampleAction implements LifecycleAction { public static final TimeValue DEFAULT_WAIT_TIMEOUT = new TimeValue(1, TimeUnit.DAYS); private static final ParseField FIXED_INTERVAL_FIELD = new ParseField(DownsampleConfig.FIXED_INTERVAL); private static final ParseField WAIT_TIMEOUT_FIELD = new ParseField("wait_timeout"); + private static final ParseField FORCE_MERGE_MAX_NUM_SEGMENTS_FIELD = new ParseField("force_merge_max_num_segments"); private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( NAME, - a -> new DownsampleAction((DateHistogramInterval) a[0], (TimeValue) a[1]) + a -> new DownsampleAction((DateHistogramInterval) a[0], (TimeValue) a[1], (Integer) a[2]) ); static { @@ -71,21 +72,24 @@ public class DownsampleAction implements LifecycleAction { WAIT_TIMEOUT_FIELD, ObjectParser.ValueType.STRING ); + PARSER.declareField(optionalConstructorArg(), p -> p.intValue(), FORCE_MERGE_MAX_NUM_SEGMENTS_FIELD, ObjectParser.ValueType.INT); } private final DateHistogramInterval fixedInterval; private final TimeValue waitTimeout; + private final Integer forceMergeMaxNumSegments; public static DownsampleAction parse(XContentParser parser) { return PARSER.apply(parser, null); } - public DownsampleAction(final DateHistogramInterval fixedInterval, final TimeValue waitTimeout) { + public DownsampleAction(final DateHistogramInterval fixedInterval, final TimeValue waitTimeout, Integer forceMergeMaxNumSegments) { if (fixedInterval == null) { throw new IllegalArgumentException("Parameter [" + FIXED_INTERVAL_FIELD.getPreferredName() + "] is required."); } this.fixedInterval = fixedInterval; this.waitTimeout = waitTimeout == null ? DEFAULT_WAIT_TIMEOUT : waitTimeout; + this.forceMergeMaxNumSegments = forceMergeMaxNumSegments; } public DownsampleAction(StreamInput in) throws IOException { @@ -93,7 +97,10 @@ public DownsampleAction(StreamInput in) throws IOException { new DateHistogramInterval(in), in.getTransportVersion().onOrAfter(TransportVersions.V_8_10_X) ? TimeValue.parseTimeValue(in.readString(), WAIT_TIMEOUT_FIELD.getPreferredName()) - : DEFAULT_WAIT_TIMEOUT + : DEFAULT_WAIT_TIMEOUT, + in.getTransportVersion().onOrAfter(TransportVersions.DOWNSAMPLE_FORCE_MERGE_MAX_NUM_SEGMENTS_PARAMETER) + ? in.readOptionalInt() + : null ); } @@ -105,6 +112,9 @@ public void writeTo(StreamOutput out) throws IOException { } else { out.writeString(DEFAULT_WAIT_TIMEOUT.getStringRep()); } + if (out.getTransportVersion().onOrAfter(TransportVersions.DOWNSAMPLE_FORCE_MERGE_MAX_NUM_SEGMENTS_PARAMETER)) { + out.writeOptionalInt(forceMergeMaxNumSegments); + } } @Override @@ -112,6 +122,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws builder.startObject(); builder.field(FIXED_INTERVAL_FIELD.getPreferredName(), fixedInterval.toString()); builder.field(WAIT_TIMEOUT_FIELD.getPreferredName(), waitTimeout.getStringRep()); + if (forceMergeMaxNumSegments != null) { + builder.field(FORCE_MERGE_MAX_NUM_SEGMENTS_FIELD.getPreferredName(), forceMergeMaxNumSegments); + } builder.endObject(); return builder; } @@ -129,6 +142,10 @@ public TimeValue waitTimeout() { return waitTimeout; } + public Integer getForceMergeMaxNumSegments() { + return forceMergeMaxNumSegments; + } + @Override public boolean isSafeAction() { return false; @@ -220,7 +237,14 @@ public List toSteps(Client client, String phase, StepKey nextStepKey) { ); // Here is where the actual downsample action takes place - DownsampleStep downsampleStep = new DownsampleStep(downsampleKey, waitForDownsampleIndexKey, client, fixedInterval, waitTimeout); + DownsampleStep downsampleStep = new DownsampleStep( + downsampleKey, + waitForDownsampleIndexKey, + client, + fixedInterval, + waitTimeout, + forceMergeMaxNumSegments + ); // Wait until the downsampled index is recovered. We again wait until the configured threshold is breached and // if the downsampled index has not successfully recovered until then, we rewind to the "cleanup-downsample-index" @@ -305,12 +329,13 @@ public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) return false; DownsampleAction that = (DownsampleAction) o; - return Objects.equals(this.fixedInterval, that.fixedInterval); + return Objects.equals(this.fixedInterval, that.fixedInterval) + && Objects.equals(this.forceMergeMaxNumSegments, that.forceMergeMaxNumSegments); } @Override public int hashCode() { - return Objects.hash(fixedInterval); + return Objects.hash(fixedInterval, forceMergeMaxNumSegments); } @Override diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleStep.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleStep.java index 2056c41c0152f..0a99cab33890a 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleStep.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleStep.java @@ -33,17 +33,20 @@ public class DownsampleStep extends AsyncActionStep { private final DateHistogramInterval fixedInterval; private final TimeValue waitTimeout; + private final Integer forceMergeMaxNumSegments; public DownsampleStep( final StepKey key, final StepKey nextStepKey, final Client client, final DateHistogramInterval fixedInterval, - final TimeValue waitTimeout + final TimeValue waitTimeout, + final Integer forceMergeMaxNumSegments ) { super(key, nextStepKey, client); this.fixedInterval = fixedInterval; this.waitTimeout = waitTimeout; + this.forceMergeMaxNumSegments = forceMergeMaxNumSegments; } @Override @@ -89,7 +92,7 @@ public void performAction( } void performDownsampleIndex(String indexName, String downsampleIndexName, ActionListener listener) { - DownsampleConfig config = new DownsampleConfig(fixedInterval); + DownsampleConfig config = new DownsampleConfig(fixedInterval, forceMergeMaxNumSegments); DownsampleAction.Request request = new DownsampleAction.Request( TimeValue.MAX_VALUE, indexName, @@ -109,9 +112,13 @@ public TimeValue getWaitTimeout() { return waitTimeout; } + public Integer getForceMergeMaxNumSegments() { + return forceMergeMaxNumSegments; + } + @Override public int hashCode() { - return Objects.hash(super.hashCode(), fixedInterval, waitTimeout); + return Objects.hash(super.hashCode(), fixedInterval, waitTimeout, forceMergeMaxNumSegments); } @Override @@ -126,7 +133,10 @@ public boolean equals(Object obj) { return false; } DownsampleStep other = (DownsampleStep) obj; - return super.equals(obj) && Objects.equals(fixedInterval, other.fixedInterval) && Objects.equals(waitTimeout, other.waitTimeout); + return super.equals(obj) + && Objects.equals(fixedInterval, other.fixedInterval) + && Objects.equals(waitTimeout, other.waitTimeout) + && Objects.equals(forceMergeMaxNumSegments, other.forceMergeMaxNumSegments); } } diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/downsample/DownsampleActionConfigTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/downsample/DownsampleActionConfigTests.java index a32b43214e34f..6bbc4c668b3a4 100644 --- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/downsample/DownsampleActionConfigTests.java +++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/downsample/DownsampleActionConfigTests.java @@ -8,7 +8,6 @@ import org.elasticsearch.action.downsample.DownsampleConfig; import org.elasticsearch.common.io.stream.Writeable; -import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval; import org.elasticsearch.test.AbstractXContentSerializingTestCase; import org.elasticsearch.xcontent.XContentParser; import org.elasticsearch.xpack.core.rollup.ConfigTestHelpers; @@ -30,7 +29,8 @@ protected DownsampleConfig mutateInstance(DownsampleConfig instance) { } public static DownsampleConfig randomConfig() { - return new DownsampleConfig(ConfigTestHelpers.randomInterval()); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + return new DownsampleConfig(ConfigTestHelpers.randomInterval(), forceMergeMaxNumSegments); } @Override @@ -44,12 +44,13 @@ protected DownsampleConfig doParseInstance(final XContentParser parser) throws I } public void testEmptyFixedInterval() { - Exception e = expectThrows(IllegalArgumentException.class, () -> new DownsampleConfig((DateHistogramInterval) null)); + Exception e = expectThrows(IllegalArgumentException.class, () -> new DownsampleConfig(null, null)); assertThat(e.getMessage(), equalTo("Parameter [fixed_interval] is required.")); } public void testEmptyTimezone() { - DownsampleConfig config = new DownsampleConfig(ConfigTestHelpers.randomInterval()); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + DownsampleConfig config = new DownsampleConfig(ConfigTestHelpers.randomInterval(), forceMergeMaxNumSegments); assertEquals("UTC", config.getTimeZone()); } } diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleActionTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleActionTests.java index 7cb93803de4ee..76dde4cc3e812 100644 --- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleActionTests.java +++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleActionTests.java @@ -36,7 +36,8 @@ public class DownsampleActionTests extends AbstractActionTestCase throw new AssertionError("Illegal randomisation branch"); } - return new DownsampleStep(key, nextKey, instance.getClient(), fixedInterval, timeout); + return new DownsampleStep(key, nextKey, instance.getClient(), fixedInterval, timeout, null); } @Override @@ -81,7 +82,8 @@ public DownsampleStep copyInstance(DownsampleStep instance) { instance.getNextStepKey(), instance.getClient(), instance.getFixedInterval(), - instance.getWaitTimeout() + instance.getWaitTimeout(), + instance.getForceMergeMaxNumSegments() ); } @@ -256,7 +258,15 @@ public void testNextStepKey() { StepKey nextKey = randomStepKey(); DateHistogramInterval fixedInterval = ConfigTestHelpers.randomInterval(); TimeValue timeout = DownsampleAction.DEFAULT_WAIT_TIMEOUT; - DownsampleStep completeStep = new DownsampleStep(randomStepKey(), nextKey, client, fixedInterval, timeout) { + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + DownsampleStep completeStep = new DownsampleStep( + randomStepKey(), + nextKey, + client, + fixedInterval, + timeout, + forceMergeMaxNumSegments + ) { void performDownsampleIndex(String indexName, String downsampleIndexName, ActionListener listener) { listener.onResponse(null); } @@ -271,7 +281,15 @@ void performDownsampleIndex(String indexName, String downsampleIndexName, Action StepKey nextKey = randomStepKey(); DateHistogramInterval fixedInterval = ConfigTestHelpers.randomInterval(); TimeValue timeout = DownsampleAction.DEFAULT_WAIT_TIMEOUT; - DownsampleStep doubleInvocationStep = new DownsampleStep(randomStepKey(), nextKey, client, fixedInterval, timeout) { + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + DownsampleStep doubleInvocationStep = new DownsampleStep( + randomStepKey(), + nextKey, + client, + fixedInterval, + timeout, + forceMergeMaxNumSegments + ) { void performDownsampleIndex(String indexName, String downsampleIndexName, ActionListener listener) { listener.onFailure( new IllegalStateException( diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleTypeTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleTypeTests.java index 57bddad8479bf..7358b4c45d64f 100644 --- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleTypeTests.java +++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleTypeTests.java @@ -76,7 +76,7 @@ public class TimeseriesLifecycleTypeTests extends ESTestCase { // same phase private static final MigrateAction TEST_MIGRATE_ACTION = MigrateAction.DISABLED; public static final TimeValue TIMEOUT = new TimeValue(1, TimeUnit.MINUTES); - private static final DownsampleAction TEST_DOWNSAMPLE_ACTION = new DownsampleAction(DateHistogramInterval.DAY, TIMEOUT); + private static final DownsampleAction TEST_DOWNSAMPLE_ACTION = new DownsampleAction(DateHistogramInterval.DAY, TIMEOUT, null); public void testValidatePhases() { boolean invalid = randomBoolean(); @@ -360,12 +360,12 @@ public void testValidateDownsamplingAction() { Phase warmPhase = new Phase( "warm", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT, null)) ); Phase coldPhase = new Phase( "cold", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT, null)) ); IllegalArgumentException e = expectThrows( @@ -382,12 +382,12 @@ public void testValidateDownsamplingAction() { Phase warmPhase = new Phase( "warm", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT, null)) ); Phase coldPhase = new Phase( "cold", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.minutes(30), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.minutes(30), TIMEOUT, null)) ); IllegalArgumentException e = expectThrows( @@ -404,12 +404,12 @@ public void testValidateDownsamplingAction() { Phase warmPhase = new Phase( "warm", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(1), TIMEOUT, null)) ); Phase coldPhase = new Phase( "cold", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.minutes(130), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.minutes(130), TIMEOUT, null)) ); IllegalArgumentException e = expectThrows( @@ -430,18 +430,18 @@ public void testValidateDownsamplingAction() { RolloverAction.NAME, TEST_ROLLOVER_ACTION, DownsampleAction.NAME, - new DownsampleAction(DateHistogramInterval.minutes(10), TIMEOUT) + new DownsampleAction(DateHistogramInterval.minutes(10), TIMEOUT, null) ) ); Phase warmPhase = new Phase( "warm", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.minutes(30), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.minutes(30), TIMEOUT, null)) ); Phase coldPhase = new Phase( "cold", TimeValue.ZERO, - Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(2), TIMEOUT)) + Map.of(DownsampleAction.NAME, new DownsampleAction(DateHistogramInterval.hours(2), TIMEOUT, null)) ); // This is a valid interval combination diff --git a/x-pack/plugin/downsample/src/internalClusterTest/java/org/elasticsearch/xpack/downsample/ILMDownsampleDisruptionIT.java b/x-pack/plugin/downsample/src/internalClusterTest/java/org/elasticsearch/xpack/downsample/ILMDownsampleDisruptionIT.java index b94ed992fe852..4134f21ab2e25 100644 --- a/x-pack/plugin/downsample/src/internalClusterTest/java/org/elasticsearch/xpack/downsample/ILMDownsampleDisruptionIT.java +++ b/x-pack/plugin/downsample/src/internalClusterTest/java/org/elasticsearch/xpack/downsample/ILMDownsampleDisruptionIT.java @@ -130,7 +130,7 @@ public void setup(final String sourceIndex, int numOfShards, int numOfReplicas, new Phase( "warm", TimeValue.ZERO, - Map.of("downsample", new org.elasticsearch.xpack.core.ilm.DownsampleAction(DateHistogramInterval.HOUR, null)) + Map.of("downsample", new org.elasticsearch.xpack.core.ilm.DownsampleAction(DateHistogramInterval.HOUR, null, null)) ) ); LifecyclePolicy policy = new LifecyclePolicy(POLICY_NAME, phases); @@ -148,7 +148,8 @@ public void testILMDownsampleRollingRestart() throws Exception { final String sourceIndex = randomAlphaOfLength(10).toLowerCase(Locale.ROOT); long startTime = LocalDateTime.parse("1993-09-09T18:00:00").atZone(ZoneId.of("UTC")).toInstant().toEpochMilli(); setup(sourceIndex, 1, 0, startTime); - final DownsampleConfig config = new DownsampleConfig(randomInterval()); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + final DownsampleConfig config = new DownsampleConfig(randomInterval(), forceMergeMaxNumSegments); final SourceSupplier sourceSupplier = () -> { final String ts = randomDateForInterval(config.getInterval(), startTime); double counterValue = DATE_FORMATTER.parseMillis(ts); diff --git a/x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/TransportDownsampleAction.java b/x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/TransportDownsampleAction.java index 8ea64ab38a7a5..8b10e04b95080 100644 --- a/x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/TransportDownsampleAction.java +++ b/x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/TransportDownsampleAction.java @@ -267,7 +267,7 @@ protected void masterOperation( final TaskId parentTask = new TaskId(clusterService.localNode().getId(), task.getId()); // Short circuit if target index has been downsampled: final String downsampleIndexName = request.getTargetIndex(); - if (canShortCircuit(downsampleIndexName, parentTask, request.getWaitTimeout(), startTime, projectMetadata, listener)) { + if (canShortCircuit(downsampleIndexName, parentTask, request, startTime, projectMetadata, listener)) { logger.info("Skipping downsampling, because a previous execution already completed downsampling"); return; } @@ -390,7 +390,7 @@ protected void masterOperation( if (canShortCircuit( request.getTargetIndex(), parentTask, - request.getWaitTimeout(), + request, startTime, clusterService.state().metadata().getProject(projectMetadata.id()), listener @@ -426,7 +426,7 @@ protected void masterOperation( private boolean canShortCircuit( String targetIndexName, TaskId parentTask, - TimeValue waitTimeout, + DownsampleAction.Request request, long startTime, ProjectMetadata projectMetadata, ActionListener listener @@ -460,7 +460,7 @@ private boolean canShortCircuit( listener, parentTask, targetIndexMetadata.getIndex().getName(), - waitTimeout, + request, startTime ) ); @@ -619,14 +619,7 @@ private void updateTargetIndexSettingStep( .indices() .updateSettings( updateSettingsReq, - new UpdateDownsampleIndexSettingsActionListener( - projectId, - listener, - parentTask, - downsampleIndexName, - request.getWaitTimeout(), - startTime - ) + new UpdateDownsampleIndexSettingsActionListener(projectId, listener, parentTask, downsampleIndexName, request, startTime) ); } @@ -819,7 +812,10 @@ private static void validateDownsamplingInterval(MapperService mapperService, Do String interval = meta.get(config.getIntervalType()); if (interval != null) { try { - DownsampleConfig sourceConfig = new DownsampleConfig(new DateHistogramInterval(interval)); + DownsampleConfig sourceConfig = new DownsampleConfig( + new DateHistogramInterval(interval), + config.getForceMergeMaxNumSegments() + ); DownsampleConfig.validateSourceAndTargetIntervals(sourceConfig, config); } catch (IllegalArgumentException exception) { e.addValidationError("Source index is a downsampled index. " + exception.getMessage()); @@ -1004,7 +1000,7 @@ class UpdateDownsampleIndexSettingsActionListener implements ActionListener listener; final TaskId parentTask; final String downsampleIndexName; - final TimeValue timeout; + final DownsampleAction.Request request; final long startTime; UpdateDownsampleIndexSettingsActionListener( @@ -1012,14 +1008,14 @@ class UpdateDownsampleIndexSettingsActionListener implements ActionListener listener, final TaskId parentTask, final String downsampleIndexName, - final TimeValue timeout, + final DownsampleAction.Request request, final long startTime ) { this.projectId = projectId; this.listener = listener; this.parentTask = parentTask; this.downsampleIndexName = downsampleIndexName; - this.timeout = timeout; + this.request = request; this.startTime = startTime; } @@ -1031,7 +1027,7 @@ public void onResponse(final AcknowledgedResponse response) { .indices() .refresh( request, - new RefreshDownsampleIndexActionListener(projectId, listener, parentTask, downsampleIndexName, timeout, startTime) + new RefreshDownsampleIndexActionListener(projectId, listener, parentTask, downsampleIndexName, this.request, startTime) ); } @@ -1052,7 +1048,7 @@ class RefreshDownsampleIndexActionListener implements ActionListener actionListener; private final TaskId parentTask; private final String downsampleIndexName; - private final TimeValue timeout; + private final DownsampleAction.Request request; private final long startTime; RefreshDownsampleIndexActionListener( @@ -1060,14 +1056,14 @@ class RefreshDownsampleIndexActionListener implements ActionListener actionListener, TaskId parentTask, final String downsampleIndexName, - final TimeValue timeout, + final DownsampleAction.Request request, final long startTime ) { this.projectId = projectId; this.actionListener = actionListener; this.parentTask = parentTask; this.downsampleIndexName = downsampleIndexName; - this.timeout = timeout; + this.request = request; this.startTime = startTime; } @@ -1080,7 +1076,7 @@ public void onResponse(final BroadcastResponse response) { taskQueue.submitTask( "update-downsample-metadata [" + downsampleIndexName + "]", new DownsampleClusterStateUpdateTask( - new ForceMergeActionListener(parentTask, downsampleIndexName, startTime, actionListener) + new ForceMergeActionListener(parentTask, request, downsampleIndexName, startTime, actionListener) ) { @Override @@ -1102,7 +1098,7 @@ public ClusterState execute(ClusterState currentState) { return ClusterState.builder(currentState).putProjectMetadata(projectBuilder).build(); } }, - timeout + request.getWaitTimeout() ); } @@ -1121,16 +1117,19 @@ class ForceMergeActionListener implements ActionListener { final ActionListener actionListener; private final TaskId parentTask; + private final DownsampleAction.Request request; private final String downsampleIndexName; private final long startTime; ForceMergeActionListener( final TaskId parentTask, + DownsampleAction.Request request, final String downsampleIndexName, final long startTime, final ActionListener onFailure ) { this.parentTask = parentTask; + this.request = request; this.downsampleIndexName = downsampleIndexName; this.startTime = startTime; this.actionListener = onFailure; @@ -1139,7 +1138,12 @@ class ForceMergeActionListener implements ActionListener { @Override public void onResponse(final AcknowledgedResponse response) { ForceMergeRequest request = new ForceMergeRequest(downsampleIndexName); - request.maxNumSegments(1); + // flush defaults to true, this is just to declare intent. The translog should be purged and a lucene commit should happen. + request.flush(true); + int maxNumSegments = this.request.getDownsampleConfig().getForceMergeMaxNumSegments(); + if (maxNumSegments != -1) { + request.maxNumSegments(maxNumSegments); + } request.setParentTask(parentTask); client.admin().indices().forceMerge(request, ActionListener.wrap(mergeIndexResp -> { actionListener.onResponse(AcknowledgedResponse.TRUE); diff --git a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java index f0b55e8398b05..f6fbf785702cb 100644 --- a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java +++ b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java @@ -270,7 +270,8 @@ public void setup() throws IOException { } public void testDownsampleIndex() throws Exception { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + DownsampleConfig config = new DownsampleConfig(randomInterval(), forceMergeMaxNumSegments); SourceSupplier sourceSupplier = () -> { String ts = randomDateForInterval(config.getInterval()); double labelDoubleValue = DATE_FORMATTER.parseMillis(ts); @@ -329,7 +330,8 @@ public void testDownsampleIndex() throws Exception { } public void testDownsampleIndexWithFlattenedAndMultiFieldDimensions() throws Exception { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + DownsampleConfig config = new DownsampleConfig(randomInterval(), forceMergeMaxNumSegments); SourceSupplier sourceSupplier = () -> { String ts = randomDateForInterval(config.getInterval()); double labelDoubleValue = DATE_FORMATTER.parseMillis(ts); @@ -366,7 +368,8 @@ public void testDownsampleIndexWithFlattenedAndMultiFieldDimensions() throws Exc public void testDownsampleOfDownsample() throws Exception { int intervalMinutes = randomIntBetween(10, 120); - DownsampleConfig config = new DownsampleConfig(DateHistogramInterval.minutes(intervalMinutes)); + Integer forceMergeMaxNumSegments = randomBoolean() ? null : randomIntBetween(-1, 128); + DownsampleConfig config = new DownsampleConfig(DateHistogramInterval.minutes(intervalMinutes), forceMergeMaxNumSegments); SourceSupplier sourceSupplier = () -> { String ts = randomDateForInterval(config.getInterval()); double labelDoubleValue = DATE_FORMATTER.parseMillis(ts); @@ -402,7 +405,7 @@ public void testDownsampleOfDownsample() throws Exception { // Downsample the downsample index. The downsampling interval is a multiple of the previous downsampling interval. String downsampleIndex2 = downsampleIndex + "-2"; - DownsampleConfig config2 = new DownsampleConfig(DateHistogramInterval.minutes(intervalMinutes * randomIntBetween(2, 50))); + DownsampleConfig config2 = new DownsampleConfig(DateHistogramInterval.minutes(intervalMinutes * randomIntBetween(2, 50)), null); downsample(downsampleIndex, downsampleIndex2, config2); assertDownsampleIndex(downsampleIndex, downsampleIndex2, config2); } @@ -436,7 +439,7 @@ public void testCopyIndexSettings() throws IOException { var updateSettingsReq = new UpdateSettingsRequest(settings, sourceIndex); assertAcked(indicesAdmin().updateSettings(updateSettingsReq).actionGet()); - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> { String ts = randomDateForInterval(config.getInterval()); return XContentFactory.jsonBuilder() @@ -464,7 +467,7 @@ public void testCopyIndexSettings() throws IOException { } public void testNullSourceIndexName() { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); ActionRequestValidationException exception = expectThrows( ActionRequestValidationException.class, () -> downsample(null, downsampleIndex, config) @@ -473,7 +476,7 @@ public void testNullSourceIndexName() { } public void testNullDownsampleIndexName() { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); ActionRequestValidationException exception = expectThrows( ActionRequestValidationException.class, () -> downsample(sourceIndex, null, config) @@ -490,7 +493,7 @@ public void testNullDownsampleConfig() { } public void testDownsampleSparseMetrics() throws Exception { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> { XContentBuilder builder = XContentFactory.jsonBuilder() .startObject() @@ -511,7 +514,7 @@ public void testDownsampleSparseMetrics() throws Exception { } public void testCannotDownsampleToExistingIndex() throws Exception { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); prepareSourceIndex(sourceIndex, true); // Create an empty index with the same name as the downsample index @@ -524,7 +527,7 @@ public void testCannotDownsampleToExistingIndex() throws Exception { } public void testDownsampleEmptyIndex() throws Exception { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); // Source index has been created in the setup() method prepareSourceIndex(sourceIndex, true); downsample(sourceIndex, downsampleIndex, config); @@ -551,21 +554,21 @@ public void testDownsampleIndexWithNoMetrics() throws Exception { ) .get(); - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); prepareSourceIndex(sourceIndex, true); downsample(sourceIndex, downsampleIndex, config); assertDownsampleIndex(sourceIndex, downsampleIndex, config); } public void testCannotDownsampleWriteableIndex() { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); // Source index has been created in the setup() method and is empty and still writable Exception exception = expectThrows(ElasticsearchException.class, () -> downsample(sourceIndex, downsampleIndex, config)); assertThat(exception.getMessage(), containsString("Downsample requires setting [index.blocks.write = true] for index")); } public void testCannotDownsampleMissingIndex() { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); IndexNotFoundException exception = expectThrows( IndexNotFoundException.class, () -> downsample("missing-index", downsampleIndex, config) @@ -575,7 +578,7 @@ public void testCannotDownsampleMissingIndex() { } public void testCannotDownsampleWhileOtherDownsampleInProgress() throws Exception { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder() .startObject() .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval())) @@ -630,7 +633,7 @@ public void onFailure(Exception e) { } public void testDownsampleDatastream() throws Exception { - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); String dataStreamName = createDataStream(); final Instant now = Instant.now(); @@ -666,7 +669,7 @@ public void testDownsampleDatastream() throws Exception { public void testCancelDownsampleIndexer() throws IOException { // create downsample config and index documents into source index - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder() .startObject() .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval())) @@ -718,7 +721,7 @@ public void testCancelDownsampleIndexer() throws IOException { public void testDownsampleBulkFailed() throws IOException { // create downsample config and index documents into source index - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder() .startObject() .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval())) @@ -786,7 +789,7 @@ public void testDownsampleBulkFailed() throws IOException { public void testTooManyBytesInFlight() throws IOException { // create downsample config and index documents into source index - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder() .startObject() .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval())) @@ -840,7 +843,7 @@ public void testTooManyBytesInFlight() throws IOException { public void testDownsampleStats() throws Exception { final PersistentTasksService persistentTasksService = mock(PersistentTasksService.class); - final DownsampleConfig config = new DownsampleConfig(randomInterval()); + final DownsampleConfig config = new DownsampleConfig(randomInterval(), null); final SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder() .startObject() .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval())) @@ -897,7 +900,7 @@ public void testDownsampleStats() throws Exception { public void testResumeDownsample() throws IOException { // create downsample config and index documents into source index - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder() .startObject() .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval())) @@ -973,7 +976,7 @@ public void testResumeDownsample() throws IOException { public void testResumeDownsamplePartial() throws IOException { // create downsample config and index documents into source index - DownsampleConfig config = new DownsampleConfig(randomInterval()); + DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> XContentFactory.jsonBuilder() .startObject() .field(FIELD_TIMESTAMP, randomDateForInterval(config.getInterval())) @@ -1620,7 +1623,7 @@ private String createDataStream() throws Exception { } public void testConcurrentDownsample() throws Exception { - final DownsampleConfig config = new DownsampleConfig(randomInterval()); + final DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> { String ts = randomDateForInterval(config.getInterval()); double labelDoubleValue = DATE_FORMATTER.parseMillis(ts); @@ -1699,7 +1702,7 @@ public void testConcurrentDownsample() throws Exception { } public void testDuplicateDownsampleRequest() throws Exception { - final DownsampleConfig config = new DownsampleConfig(randomInterval()); + final DownsampleConfig config = new DownsampleConfig(randomInterval(), null); SourceSupplier sourceSupplier = () -> { String ts = randomDateForInterval(config.getInterval()); double labelDoubleValue = DATE_FORMATTER.parseMillis(ts); diff --git a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleDataStreamTests.java b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleDataStreamTests.java index 96b3fc0311835..d41fcdedcd147 100644 --- a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleDataStreamTests.java +++ b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleDataStreamTests.java @@ -102,7 +102,7 @@ public void testDataStreamDownsample() throws ExecutionException, InterruptedExc rolloverResponse.getOldIndex(), downsampleTargetIndex, TIMEOUT, - new DownsampleConfig(DateHistogramInterval.HOUR) + new DownsampleConfig(DateHistogramInterval.HOUR, null) ); final AcknowledgedResponse downsampleResponse = indicesAdmin().execute(DownsampleAction.INSTANCE, downsampleRequest).actionGet(); diff --git a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardPersistentTaskExecutorTests.java b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardPersistentTaskExecutorTests.java index c132912da133a..ab214ed4ffdef 100644 --- a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardPersistentTaskExecutorTests.java +++ b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardPersistentTaskExecutorTests.java @@ -78,7 +78,7 @@ public void testGetAssignment() { .build(); var params = new DownsampleShardTaskParams( - new DownsampleConfig(new DateHistogramInterval("1h")), + new DownsampleConfig(new DateHistogramInterval("1h"), null), shardId.getIndexName(), 1, 1, @@ -110,7 +110,7 @@ public void testGetAssignmentMissingIndex() { var missingShardId = new ShardId(new Index("another_index", "uid"), 0); var params = new DownsampleShardTaskParams( - new DownsampleConfig(new DateHistogramInterval("1h")), + new DownsampleConfig(new DateHistogramInterval("1h"), null), missingShardId.getIndexName(), 1, 1, diff --git a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardTaskParamsTests.java b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardTaskParamsTests.java index 7d79ae720271e..b445967a2aa2c 100644 --- a/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardTaskParamsTests.java +++ b/x-pack/plugin/downsample/src/test/java/org/elasticsearch/xpack/downsample/DownsampleShardTaskParamsTests.java @@ -29,7 +29,7 @@ protected DownsampleShardTaskParams createTestInstance() { long endTime = startTime + randomLongBetween(1000, 10_000); String[] dimensions = randomBoolean() ? generateRandomStringArray(5, 5, false, true) : new String[] {}; return new DownsampleShardTaskParams( - new DownsampleConfig(randomFrom(DateHistogramInterval.HOUR, DateHistogramInterval.DAY)), + new DownsampleConfig(randomFrom(DateHistogramInterval.HOUR, DateHistogramInterval.DAY), null), randomAlphaOfLength(5), startTime, endTime, @@ -44,7 +44,7 @@ protected DownsampleShardTaskParams createTestInstance() { protected DownsampleShardTaskParams mutateInstance(DownsampleShardTaskParams in) throws IOException { return switch (between(0, 7)) { case 0 -> new DownsampleShardTaskParams( - new DownsampleConfig(randomFrom(DateHistogramInterval.WEEK, DateHistogramInterval.MONTH)), + new DownsampleConfig(randomFrom(DateHistogramInterval.WEEK, DateHistogramInterval.MONTH), null), in.downsampleIndex(), in.indexStartTimeMillis(), in.indexEndTimeMillis(), diff --git a/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ilm/actions/DownsampleActionIT.java b/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ilm/actions/DownsampleActionIT.java index d4ecff4238591..70246e5529ac1 100644 --- a/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ilm/actions/DownsampleActionIT.java +++ b/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ilm/actions/DownsampleActionIT.java @@ -196,7 +196,12 @@ public void testRollupIndex() throws Exception { // Create the ILM policy String phaseName = randomFrom("warm", "cold"); DateHistogramInterval fixedInterval = ConfigTestHelpers.randomInterval(); - createNewSingletonPolicy(client(), policy, phaseName, new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT)); + createNewSingletonPolicy( + client(), + policy, + phaseName, + new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT, null) + ); // Create a time series index managed by the policy createIndex(index, alias, policy, true); @@ -230,7 +235,7 @@ public void testRollupIndexInTheHotPhaseWithoutRollover() { client(), policy, "hot", - new DownsampleAction(ConfigTestHelpers.randomInterval(), DownsampleAction.DEFAULT_WAIT_TIMEOUT) + new DownsampleAction(ConfigTestHelpers.randomInterval(), DownsampleAction.DEFAULT_WAIT_TIMEOUT, null) ) ); assertTrue( @@ -248,7 +253,7 @@ public void testRollupIndexInTheHotPhaseAfterRollover() throws Exception { RolloverAction.NAME, new RolloverAction(null, null, null, 1L, null, null, null, null, null, null), DownsampleAction.NAME, - new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT) + new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT, null) ); Map phases = Map.of("hot", new Phase("hot", TimeValue.ZERO, hotActions)); LifecyclePolicy lifecyclePolicy = new LifecyclePolicy(policy, phases); @@ -307,7 +312,12 @@ public void testRollupIndexInTheHotPhaseAfterRollover() throws Exception { public void testTsdbDataStreams() throws Exception { // Create the ILM policy DateHistogramInterval fixedInterval = ConfigTestHelpers.randomInterval(); - createNewSingletonPolicy(client(), policy, "warm", new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT)); + createNewSingletonPolicy( + client(), + policy, + "warm", + new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT, null) + ); // Create a template Request createIndexTemplateRequest = new Request("POST", "/_index_template/" + dataStream); @@ -355,7 +365,12 @@ public void testTsdbDataStreams() throws Exception { public void testILMWaitsForTimeSeriesEndTimeToLapse() throws Exception { // Create the ILM policy DateHistogramInterval fixedInterval = ConfigTestHelpers.randomInterval(); - createNewSingletonPolicy(client(), policy, "warm", new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT)); + createNewSingletonPolicy( + client(), + policy, + "warm", + new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT, null) + ); // Create a template Request createIndexTemplateRequest = new Request("POST", "/_index_template/" + dataStream); @@ -398,7 +413,12 @@ public void testRollupNonTSIndex() throws Exception { // Create the ILM policy String phaseName = randomFrom("warm", "cold"); DateHistogramInterval fixedInterval = ConfigTestHelpers.randomInterval(); - createNewSingletonPolicy(client(), policy, phaseName, new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT)); + createNewSingletonPolicy( + client(), + policy, + phaseName, + new DownsampleAction(fixedInterval, DownsampleAction.DEFAULT_WAIT_TIMEOUT, null) + ); // Create a non TSDB managed index createIndex(index, alias, policy, false); diff --git a/x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/integration/DataStreamLifecycleDownsamplingSecurityIT.java b/x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/integration/DataStreamLifecycleDownsamplingSecurityIT.java index 458dee693c80a..9c63043c19e6d 100644 --- a/x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/integration/DataStreamLifecycleDownsamplingSecurityIT.java +++ b/x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/integration/DataStreamLifecycleDownsamplingSecurityIT.java @@ -125,11 +125,11 @@ public void testDownsamplingAuthorized() throws Exception { List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(0), - new DownsampleConfig(new DateHistogramInterval("5m")) + new DownsampleConfig(new DateHistogramInterval("5m"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueSeconds(10), - new DownsampleConfig(new DateHistogramInterval("10m")) + new DownsampleConfig(new DateHistogramInterval("10m"), null) ) ) ) @@ -417,11 +417,11 @@ public static class SystemDataStreamWithDownsamplingConfigurationPlugin extends List.of( new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueMillis(0), - new DownsampleConfig(new DateHistogramInterval("5m")) + new DownsampleConfig(new DateHistogramInterval("5m"), null) ), new DataStreamLifecycle.Downsampling.Round( TimeValue.timeValueSeconds(10), - new DownsampleConfig(new DateHistogramInterval("10m")) + new DownsampleConfig(new DateHistogramInterval("10m"), null) ) ) ) diff --git a/x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/AbstractSqlQueryRequest.java b/x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/AbstractSqlQueryRequest.java index eed36f417d7fb..85bc335a07418 100644 --- a/x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/AbstractSqlQueryRequest.java +++ b/x-pack/plugin/sql/sql-action/src/main/java/org/elasticsearch/xpack/sql/action/AbstractSqlQueryRequest.java @@ -6,7 +6,6 @@ */ package org.elasticsearch.xpack.sql.action; -import org.elasticsearch.TransportVersions; import org.elasticsearch.action.ActionRequestValidationException; import org.elasticsearch.action.CompositeIndicesRequest; import org.elasticsearch.common.Strings;