Skip to content

Effort to fix testDataStreamLifecycleDownsampleRollingRestart #123769 #125478

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Mar 28, 2025
Merged

Effort to fix testDataStreamLifecycleDownsampleRollingRestart #123769 #125478

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
a969ed0
Use a cluster state listener to detect the downsampling status to avo…
gmarouli Mar 24, 2025
0caf4dd
Reduce the amount of docs we are indexing to avoid flakiness
gmarouli Mar 24, 2025
38c1882
Bring the timeouts closer to the original
gmarouli Mar 24, 2025
4377b72
Increase the timeout when waiting for the downsampling to start
gmarouli Mar 24, 2025
091c227
Overload `safeGet` and `safeAwait` to also accept timeouts
gmarouli Mar 26, 2025
403b310
Merge branch 'main' into test-fix-123769
gmarouli Mar 26, 2025
e6504e2
Merge with main
gmarouli Mar 28, 2025
77ed7ff
Make the status check stricter
gmarouli Mar 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions muted-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,9 +254,6 @@ tests:
- class: org.elasticsearch.test.rest.yaml.RcsCcsCommonYamlTestSuiteIT
method: test {p0=search.vectors/41_knn_search_bbq_hnsw/Test knn search}
issue: https://github.com/elastic/elasticsearch/issues/123727
- class: org.elasticsearch.xpack.downsample.DataStreamLifecycleDownsampleDisruptionIT
method: testDataStreamLifecycleDownsampleRollingRestart
issue: https://github.com/elastic/elasticsearch/issues/123769
- class: org.elasticsearch.xpack.searchablesnapshots.FrozenSearchableSnapshotsIntegTests
method: testCreateAndRestorePartialSearchableSnapshot
issue: https://github.com/elastic/elasticsearch/issues/123773
Expand Down
87 changes: 52 additions & 35 deletions ...st/java/org/elasticsearch/xpack/downsample/DataStreamLifecycleDownsampleDisruptionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,37 +11,36 @@
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.rollover.RolloverAction;
import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
import org.elasticsearch.action.downsample.DownsampleConfig;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.TestPlainActionFuture;
import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.datastreams.DataStreamsPlugin;
import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.test.ClusterServiceUtils;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.InternalTestCluster;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;

import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_DOWNSAMPLE_STATUS;
import static org.elasticsearch.xpack.downsample.DataStreamLifecycleDriver.getBackingIndices;
import static org.elasticsearch.xpack.downsample.DataStreamLifecycleDriver.putTSDBIndexTemplate;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;

@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, numClientNodes = 4)
public class DataStreamLifecycleDownsampleDisruptionIT extends ESIntegTestCase {
private static final Logger logger = LogManager.getLogger(DataStreamLifecycleDownsampleDisruptionIT.class);
public static final int DOC_COUNT = 50_000;
public static final int DOC_COUNT = 25_000;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you know why we need so many docs in the first place? Is the reason purely that we want the downsample operation to take some time so we have a chance to disrupt the cluster during the downsampling? If so, I feel like a more targeted approach would be better. For instance, we could delay some actions by intercepting them - that's a fairly common practice in internal cluster tests. It's going to require some more complexity, but I think it'll have a higher value as we'll be more in control of when the disruption happens. What do you think?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure about that. Reducing the amount of indexed documents very much and accessing internal state to determine when to introduce the disruption can also reduce the value of the test because it becomes more staged.

On the other hand, I do not know what is the lowest count that makes sense in this test and if there is an internal task or something that we could leverage to better position the rolling restart. That is why this PR is on the conservative side. But I can follow up on it and see where it goes.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mentioned that you ran the test locally to check which doc counts still caused the disruption to happen. Do we have an idea of whether the disruption even happens in CI - with the 50k docs and the 25k docs? Because the test also handles the situation where the downsampling already completed before we start the disruption, I feel we have no proof that the test actually tests what it's supposed to do.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Kind of, I sampled it but this does not guarantee that it will always be like that. I checked if the status after the disruption was started at least once.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Did you check on CI as well? If it runs - based on timing - on our laptops in a certain way, it doesn't prove it runs on CI in the same way.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No I haven't checked it because I do not know how, as far as I know we do not have test logs from successful builds. So unless I make it fail I do not know how to get that data. How would you test it?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The only way I can think of is to change the test to change the first ensureDownsamplingStatus to only expect the started status. That will cause the test to fail if the downsampling already completed (which makes the test worthless).

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done, we can monitor if it starts failing now with this assertion and evaluate how to change the test. I will keep an eye on it.


@Override
protected Collection<Class<? extends Plugin>> nodePlugins() {
Expand All @@ -55,7 +54,6 @@ protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
return settings.build();
}

@TestLogging(value = "org.elasticsearch.datastreams.lifecycle:TRACE", reason = "debugging")
public void testDataStreamLifecycleDownsampleRollingRestart() throws Exception {
final InternalTestCluster cluster = internalCluster();
cluster.startMasterOnlyNodes(1);
Expand Down Expand Up @@ -88,38 +86,57 @@ public void testDataStreamLifecycleDownsampleRollingRestart() throws Exception {
// testing so DSL doesn't have to wait for the end_time to lapse)
putTSDBIndexTemplate(client(), dataStreamName, null, null, lifecycle);
client().execute(RolloverAction.INSTANCE, new RolloverRequest(dataStreamName, null)).actionGet();
String sourceIndex = getBackingIndices(client(), dataStreamName).get(0);
final String targetIndex = "downsample-5m-" + sourceIndex;

// DSL runs every second and it has to tail forcemerge the index (2 seconds) and mark it as read-only (2s) before it starts
// downsampling. This sleep here tries to get as close as possible to having disruption during the downsample execution.
long sleepTime = randomLongBetween(3000, 4500);
logger.info("-> giving data stream lifecycle [{}] millis to make some progress before starting the disruption", sleepTime);
Thread.sleep(sleepTime);
List<String> backingIndices = getBackingIndices(client(), dataStreamName);
// first generation index
String sourceIndex = backingIndices.get(0);
/**
* DLM runs every second and it has to tail forcemerge the index (2 seconds) and mark it as read-only (2s) before it starts
* downsampling. We try to detect if the downsampling has started by checking the downsample status in the target index.
*/
logger.info("-> Waiting for the data stream lifecycle to start the downsampling operation before starting the disruption.");
ensureDownsamplingStatus(
targetIndex,
Set.of(IndexMetadata.DownsampleTaskStatus.STARTED, IndexMetadata.DownsampleTaskStatus.SUCCESS),
TimeValue.timeValueMillis(4500)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd personally be inclined to up this to TimeValue.timeValueSeconds(10) to allow for even more leniency. In happy flows, that doesn't have a (negative) impact as you've already optimized to use a cluster state listener instead of an exponential backoff. If there is an actual bug, waiting a few seconds won't have a negative impact either. If there's just a timing issue (i.e. slow CI server or w/e), waiting a few seconds more can have a positive impact. What do you think?

);

logger.info("-> Starting the disruption.");
internalCluster().rollingRestart(new InternalTestCluster.RestartCallback() {
});

// if the source index has already been downsampled and moved into the data stream just use its name directly
final String targetIndex = sourceIndex.startsWith("downsample-5m-") ? sourceIndex : "downsample-5m-" + sourceIndex;
assertBusy(() -> {
try {
GetSettingsResponse getSettingsResponse = cluster.client()
.admin()
.indices()
.getSettings(
new GetSettingsRequest(TEST_REQUEST_TIMEOUT).indices(targetIndex).indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN)
)
.actionGet();
Settings indexSettings = getSettingsResponse.getIndexToSettings().get(targetIndex);
assertThat(indexSettings, is(notNullValue()));
assertThat(IndexMetadata.INDEX_DOWNSAMPLE_STATUS.get(indexSettings), is(IndexMetadata.DownsampleTaskStatus.SUCCESS));
assertEquals("5m", IndexMetadata.INDEX_DOWNSAMPLE_INTERVAL.get(indexSettings));
} catch (Exception e) {
throw new AssertionError(e);
}
}, 120, TimeUnit.SECONDS);
ensureDownsamplingStatus(targetIndex, Set.of(IndexMetadata.DownsampleTaskStatus.SUCCESS), TimeValue.timeValueSeconds(120));
ensureGreen(targetIndex);
logger.info("-> Relocation has finished");
}

private void ensureDownsamplingStatus(
String downsampledIndex,
Set<IndexMetadata.DownsampleTaskStatus> expectedStatuses,
TimeValue timeout
) {
final var clusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
final IndexMetadata.DownsampleTaskStatus[] downsamplingStatus = new IndexMetadata.DownsampleTaskStatus[1];
final var listener = ClusterServiceUtils.addTemporaryStateListener(clusterService, clusterState -> {
final var indexMetadata = clusterState.metadata().getProject().index(downsampledIndex);
if (indexMetadata == null) {
return false;
}
downsamplingStatus[0] = INDEX_DOWNSAMPLE_STATUS.get(indexMetadata.getSettings());
return expectedStatuses.contains(downsamplingStatus[0]);
});
try {
final var future = new TestPlainActionFuture<Void>();
listener.addListener(future);
future.get(timeout.getMillis(), TimeUnit.MILLISECONDS);
logger.info("-> Downsampling status for index [{}] is [{}]", downsampledIndex, downsamplingStatus[0]);
} catch (Exception e) {
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
throw new AssertionError(
"Error while waiting for " + expectedStatuses + " but found '" + downsamplingStatus[0] + "'. " + e.getMessage(),
e
);
}
}
}