|
8 | 8 | */ |
9 | 9 | package org.elasticsearch.datastreams; |
10 | 10 |
|
| 11 | +import org.apache.logging.log4j.LogManager; |
| 12 | +import org.apache.logging.log4j.Logger; |
| 13 | +import org.apache.logging.log4j.core.LogEvent; |
| 14 | +import org.apache.logging.log4j.core.appender.AbstractAppender; |
| 15 | +import org.apache.logging.log4j.core.filter.RegexFilter; |
| 16 | +import org.apache.logging.log4j.message.Message; |
11 | 17 | import org.elasticsearch.cluster.ClusterState; |
12 | 18 | import org.elasticsearch.cluster.metadata.DataStream; |
13 | 19 | import org.elasticsearch.cluster.metadata.DataStreamTestHelper; |
14 | 20 | import org.elasticsearch.cluster.metadata.ProjectId; |
15 | 21 | import org.elasticsearch.cluster.metadata.ProjectMetadata; |
16 | 22 | import org.elasticsearch.cluster.service.ClusterService; |
| 23 | +import org.elasticsearch.common.logging.Loggers; |
17 | 24 | import org.elasticsearch.common.settings.ClusterSettings; |
18 | 25 | import org.elasticsearch.common.settings.Settings; |
19 | 26 | import org.elasticsearch.core.TimeValue; |
|
23 | 30 | import org.elasticsearch.threadpool.TestThreadPool; |
24 | 31 | import org.elasticsearch.threadpool.ThreadPool; |
25 | 32 | import org.junit.After; |
| 33 | +import org.junit.AfterClass; |
26 | 34 | import org.junit.Before; |
| 35 | +import org.junit.BeforeClass; |
27 | 36 |
|
28 | 37 | import java.time.Duration; |
29 | 38 | import java.time.Instant; |
30 | 39 | import java.time.temporal.ChronoUnit; |
31 | 40 | import java.time.temporal.TemporalAmount; |
| 41 | +import java.util.ArrayList; |
| 42 | +import java.util.HashMap; |
32 | 43 | import java.util.List; |
| 44 | +import java.util.Map; |
33 | 45 | import java.util.Set; |
34 | 46 |
|
| 47 | +import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName; |
| 48 | +import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.createIndexMetadata; |
35 | 49 | import static org.hamcrest.Matchers.containsString; |
36 | 50 | import static org.hamcrest.Matchers.equalTo; |
37 | 51 | import static org.hamcrest.Matchers.not; |
|
42 | 56 |
|
43 | 57 | public class UpdateTimeSeriesRangeServiceTests extends ESTestCase { |
44 | 58 |
|
| 59 | + static MockAppender appender; |
| 60 | + static Logger testLogger1 = LogManager.getLogger(UpdateTimeSeriesRangeService.class); |
| 61 | + |
| 62 | + @BeforeClass |
| 63 | + public static void classInit() throws IllegalAccessException { |
| 64 | + appender = new MockAppender("mock_appender"); |
| 65 | + appender.start(); |
| 66 | + Loggers.addAppender(testLogger1, appender); |
| 67 | + } |
| 68 | + |
| 69 | + @AfterClass |
| 70 | + public static void classCleanup() { |
| 71 | + Loggers.removeAppender(testLogger1, appender); |
| 72 | + appender.stop(); |
| 73 | + } |
| 74 | + |
45 | 75 | private ThreadPool threadPool; |
46 | 76 | private UpdateTimeSeriesRangeService instance; |
47 | 77 |
|
@@ -199,6 +229,70 @@ public void testUpdateTimeSeriesTemporalRangeMultipleDataStream() { |
199 | 229 | assertThat(getEndTime(project, dataStreamName3, 0), equalTo(start)); |
200 | 230 | } |
201 | 231 |
|
| 232 | + public void testUpdateTimeSeriesTemporalOneBadDataStream() { |
| 233 | + String dataStreamName1 = "logs-app1"; |
| 234 | + String dataStreamName2 = "logs-app2-broken"; |
| 235 | + String dataStreamName3 = "logs-app3"; |
| 236 | + Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS); |
| 237 | + |
| 238 | + Instant start = now.minus(90, ChronoUnit.MINUTES); |
| 239 | + Instant end = start.plus(30, ChronoUnit.MINUTES); |
| 240 | + final var projectId = randomProjectIdOrDefault(); |
| 241 | + ProjectMetadata.Builder mbBuilder = ProjectMetadata.builder(projectId); |
| 242 | + for (String dataStreamName : List.of(dataStreamName1, dataStreamName2, dataStreamName3)) { |
| 243 | + DataStreamTestHelper.getClusterStateWithDataStream(mbBuilder, dataStreamName, List.of(new Tuple<>(start, end))); |
| 244 | + } |
| 245 | + |
| 246 | + Settings settings = Settings.builder().put("index.mode", "logsdb").build(); |
| 247 | + var im = createIndexMetadata(getDefaultBackingIndexName(dataStreamName2, 2, start.toEpochMilli()), true, settings, 0); |
| 248 | + mbBuilder.put(im, true); |
| 249 | + var ds2 = mbBuilder.dataStreamMetadata().dataStreams().get(dataStreamName2); |
| 250 | + var ds2Indices = new ArrayList<>(ds2.getIndices()); |
| 251 | + ds2Indices.add(im.getIndex()); |
| 252 | + var copy = new HashMap<>(mbBuilder.dataStreamMetadata().dataStreams()); |
| 253 | + copy.put( |
| 254 | + dataStreamName2, |
| 255 | + new DataStream( |
| 256 | + ds2.getName(), |
| 257 | + ds2Indices, |
| 258 | + 2, |
| 259 | + ds2.getMetadata(), |
| 260 | + ds2.isHidden(), |
| 261 | + ds2.isReplicated(), |
| 262 | + ds2.isSystem(), |
| 263 | + ds2.isAllowCustomRouting(), |
| 264 | + ds2.getIndexMode(), |
| 265 | + ds2.getDataLifecycle(), |
| 266 | + ds2.getDataStreamOptions(), |
| 267 | + ds2.getFailureIndices(), |
| 268 | + ds2.rolloverOnWrite(), |
| 269 | + ds2.getAutoShardingEvent() |
| 270 | + ) |
| 271 | + ); |
| 272 | + mbBuilder.dataStreams(copy, Map.of()); |
| 273 | + |
| 274 | + now = now.minus(45, ChronoUnit.MINUTES); |
| 275 | + ClusterState before = ClusterState.builder(ClusterState.EMPTY_STATE).putProjectMetadata(mbBuilder).build(); |
| 276 | + ClusterState result = instance.updateTimeSeriesTemporalRange(before, now); |
| 277 | + assertThat(result, not(sameInstance(before))); |
| 278 | + final var project = result.getMetadata().getProject(projectId); |
| 279 | + final var expectedEndTime = now.plus(35, ChronoUnit.MINUTES).truncatedTo(ChronoUnit.SECONDS); |
| 280 | + assertThat(getEndTime(project, dataStreamName1, 0), equalTo(expectedEndTime)); |
| 281 | + assertThat(getEndTime(project, dataStreamName2, 0), equalTo(end)); // failed to update end_time, because broken data stream |
| 282 | + assertThat(getEndTime(project, dataStreamName3, 0), equalTo(expectedEndTime)); |
| 283 | + |
| 284 | + String message = appender.getLastEventAndReset().getMessage().getFormattedMessage(); |
| 285 | + assertThat( |
| 286 | + message, |
| 287 | + equalTo( |
| 288 | + "unable to update [index.time_series.end_time] for data stream [logs-app2-broken] and " |
| 289 | + + "backing index [" |
| 290 | + + im.getIndex().getName() |
| 291 | + + "]" |
| 292 | + ) |
| 293 | + ); |
| 294 | + } |
| 295 | + |
202 | 296 | public void testUpdateTimeSeriesTemporalRange_multipleProjects() { |
203 | 297 | String dataStreamName = "logs-app1"; |
204 | 298 | Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS); |
@@ -253,4 +347,27 @@ static Instant getStartTime(ProjectMetadata project, String dataStreamName, int |
253 | 347 | return IndexSettings.TIME_SERIES_START_TIME.get(indexSettings); |
254 | 348 | } |
255 | 349 |
|
| 350 | + static class MockAppender extends AbstractAppender { |
| 351 | + public LogEvent lastEvent; |
| 352 | + |
| 353 | + MockAppender(final String name) throws IllegalAccessException { |
| 354 | + super(name, RegexFilter.createFilter(".*(\n.*)*", new String[0], false, null, null), null, false); |
| 355 | + } |
| 356 | + |
| 357 | + @Override |
| 358 | + public void append(LogEvent event) { |
| 359 | + lastEvent = event.toImmutable(); |
| 360 | + } |
| 361 | + |
| 362 | + Message lastMessage() { |
| 363 | + return lastEvent.getMessage(); |
| 364 | + } |
| 365 | + |
| 366 | + public LogEvent getLastEventAndReset() { |
| 367 | + LogEvent toReturn = lastEvent; |
| 368 | + lastEvent = null; |
| 369 | + return toReturn; |
| 370 | + } |
| 371 | + } |
| 372 | + |
256 | 373 | } |
0 commit comments