|
|
@@ -11,6 +11,7 @@ package org.elasticsearch.cluster.service;
|
|
|
import org.apache.logging.log4j.Level;
|
|
|
import org.apache.logging.log4j.LogManager;
|
|
|
import org.apache.logging.log4j.Logger;
|
|
|
+import org.apache.logging.log4j.core.LogEvent;
|
|
|
import org.apache.lucene.util.SetOnce;
|
|
|
import org.elasticsearch.ElasticsearchException;
|
|
|
import org.elasticsearch.Version;
|
|
|
@@ -1093,11 +1094,11 @@ public class MasterServiceTests extends ESTestCase {
|
|
|
final AtomicReference<ClusterState> clusterStateRef = new AtomicReference<>(initialClusterState);
|
|
|
masterService.setClusterStatePublisher((clusterStatePublicationEvent, publishListener, ackListener) -> {
|
|
|
ClusterServiceUtils.setAllElapsedMillis(clusterStatePublicationEvent);
|
|
|
- if (clusterStatePublicationEvent.getSummary().contains("test5")) {
|
|
|
+ if (clusterStatePublicationEvent.getSummary().toString().contains("test5")) {
|
|
|
relativeTimeInMillis += MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY).millis()
|
|
|
+ randomLongBetween(1, 1000000);
|
|
|
}
|
|
|
- if (clusterStatePublicationEvent.getSummary().contains("test6")) {
|
|
|
+ if (clusterStatePublicationEvent.getSummary().toString().contains("test6")) {
|
|
|
relativeTimeInMillis += MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING.get(Settings.EMPTY).millis()
|
|
|
+ randomLongBetween(1, 1000000);
|
|
|
throw new ElasticsearchException("simulated error during slow publication which should trigger logging");
|
|
|
@@ -1663,6 +1664,146 @@ public class MasterServiceTests extends ESTestCase {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ @TestLogging(
|
|
|
+ value = "org.elasticsearch.cluster.service:DEBUG",
|
|
|
+ reason = "to ensure that we log the right batch description, which only happens at DEBUG level"
|
|
|
+ )
|
|
|
+ public void testBatchedUpdateSummaryLogging() throws Exception {
|
|
|
+ MockLogAppender mockAppender = new MockLogAppender();
|
|
|
+ mockAppender.start();
|
|
|
+
|
|
|
+ Logger masterServiceLogger = LogManager.getLogger(MasterService.class);
|
|
|
+ Loggers.addAppender(masterServiceLogger, mockAppender);
|
|
|
+ try (MasterService masterService = createMasterService(true)) {
|
|
|
+
|
|
|
+ final var barrier = new CyclicBarrier(2);
|
|
|
+ final var blockingTask = new ClusterStateUpdateTask() {
|
|
|
+ @Override
|
|
|
+ public ClusterState execute(ClusterState currentState) throws Exception {
|
|
|
+ barrier.await(10, TimeUnit.SECONDS);
|
|
|
+ barrier.await(10, TimeUnit.SECONDS);
|
|
|
+ return currentState;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void onFailure(Exception e) {
|
|
|
+ fail();
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ class Task implements ClusterStateTaskListener {
|
|
|
+ private final String description;
|
|
|
+
|
|
|
+ Task(String description) {
|
|
|
+ this.description = description;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void onFailure(Exception e) {
|
|
|
+ fail();
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public String toString() {
|
|
|
+ return description;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ class Executor implements ClusterStateTaskExecutor<Task> {
|
|
|
+
|
|
|
+ final Semaphore semaphore = new Semaphore(0);
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public ClusterState execute(ClusterState currentState, List<TaskContext<Task>> taskContexts) {
|
|
|
+ for (final var taskContext : taskContexts) {
|
|
|
+ taskContext.success(ActionListener.wrap(semaphore::release));
|
|
|
+ }
|
|
|
+ return currentState;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ masterService.submitUnbatchedStateUpdateTask("block", blockingTask);
|
|
|
+ barrier.await(10, TimeUnit.SECONDS);
|
|
|
+
|
|
|
+ final var smallBatchExecutor = new Executor();
|
|
|
+ for (int source = 0; source < 2; source++) {
|
|
|
+ for (int task = 0; task < 2; task++) {
|
|
|
+ masterService.submitStateUpdateTask(
|
|
|
+ "source-" + source,
|
|
|
+ new Task("task-" + task),
|
|
|
+ ClusterStateTaskConfig.build(Priority.NORMAL),
|
|
|
+ smallBatchExecutor
|
|
|
+ );
|
|
|
+ }
|
|
|
+ mockAppender.addExpectation(
|
|
|
+ new MockLogAppender.SeenEventExpectation(
|
|
|
+ "mention of tasks source-" + source,
|
|
|
+ MasterService.class.getCanonicalName(),
|
|
|
+ Level.DEBUG,
|
|
|
+ "executing cluster state update for [*source-" + source + "[task-0, task-1]*"
|
|
|
+ )
|
|
|
+ );
|
|
|
+ }
|
|
|
+
|
|
|
+ final var manySourceExecutor = new Executor();
|
|
|
+ for (int source = 0; source < 1024; source++) {
|
|
|
+ for (int task = 0; task < 2; task++) {
|
|
|
+ masterService.submitStateUpdateTask(
|
|
|
+ "source-" + source,
|
|
|
+ new Task("task-" + task),
|
|
|
+ ClusterStateTaskConfig.build(Priority.NORMAL),
|
|
|
+ manySourceExecutor
|
|
|
+ );
|
|
|
+ }
|
|
|
+ }
|
|
|
+ mockAppender.addExpectation(
|
|
|
+ new MockLogAppender.SeenEventExpectation(
|
|
|
+ "truncated description of batch with many sources",
|
|
|
+ MasterService.class.getCanonicalName(),
|
|
|
+ Level.DEBUG,
|
|
|
+ "executing cluster state update for [* ... (1024 in total, *) (2048 tasks in total)]"
|
|
|
+ ) {
|
|
|
+ @Override
|
|
|
+ public boolean innerMatch(LogEvent event) {
|
|
|
+ return event.getMessage().getFormattedMessage().length() < BatchSummary.MAX_TASK_DESCRIPTION_CHARS + 200;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ );
|
|
|
+
|
|
|
+ final var manyTasksPerSourceExecutor = new Executor();
|
|
|
+ for (int task = 0; task < 2048; task++) {
|
|
|
+ masterService.submitStateUpdateTask(
|
|
|
+ "unique-source",
|
|
|
+ new Task("task-" + task),
|
|
|
+ ClusterStateTaskConfig.build(Priority.NORMAL),
|
|
|
+ manyTasksPerSourceExecutor
|
|
|
+ );
|
|
|
+ }
|
|
|
+ mockAppender.addExpectation(
|
|
|
+ new MockLogAppender.SeenEventExpectation(
|
|
|
+ "truncated description of batch with many tasks from a single source",
|
|
|
+ MasterService.class.getCanonicalName(),
|
|
|
+ Level.DEBUG,
|
|
|
+ "executing cluster state update for [unique-source[task-0, task-1, task-2, task-3, task-4, * ... (2048 in total, *)]]"
|
|
|
+ ) {
|
|
|
+ @Override
|
|
|
+ public boolean innerMatch(LogEvent event) {
|
|
|
+ return event.getMessage().getFormattedMessage().length() < 1500;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ );
|
|
|
+
|
|
|
+ barrier.await(10, TimeUnit.SECONDS);
|
|
|
+ assertTrue(smallBatchExecutor.semaphore.tryAcquire(4, 10, TimeUnit.SECONDS));
|
|
|
+ assertTrue(manySourceExecutor.semaphore.tryAcquire(2048, 10, TimeUnit.SECONDS));
|
|
|
+ assertTrue(manyTasksPerSourceExecutor.semaphore.tryAcquire(2048, 10, TimeUnit.SECONDS));
|
|
|
+ mockAppender.assertAllExpectationsMatched();
|
|
|
+ } finally {
|
|
|
+ Loggers.removeAppender(masterServiceLogger, mockAppender);
|
|
|
+ mockAppender.stop();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* Returns the cluster state that the master service uses (and that is provided by the discovery layer)
|
|
|
*/
|
David Turner