log messages from allocation commands

core/src/main/java/org/elasticsearch/action/admin/cluster/reroute/TransportClusterRerouteAction.java

@@ -68,8 +68,18 @@ protected ClusterRerouteResponse newResponse() {
 
     @Override
     protected void masterOperation(final ClusterRerouteRequest request, final ClusterState state, final ActionListener<ClusterRerouteResponse> listener) {
+        ActionListener<ClusterRerouteResponse> logWrapper = ActionListener.wrap(
+            response -> {
+                if (request.dryRun() == false) {
+                    response.getExplanations().getYesDecisionMessages().forEach(logger::info);
+                }
+                listener.onResponse(response);
+            },
+            listener::onFailure
+        );
+
         clusterService.submitStateUpdateTask("cluster_reroute (api)", new ClusterRerouteResponseAckedClusterStateUpdateTask(logger,
-            allocationService, request, listener));
+            allocationService, request, logWrapper));
     }
 
     static class ClusterRerouteResponseAckedClusterStateUpdateTask extends AckedClusterStateUpdateTask<ClusterRerouteResponse> {

core/src/main/java/org/elasticsearch/cluster/routing/allocation/RoutingExplanations.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cluster.routing.allocation;
 
+import org.elasticsearch.cluster.routing.allocation.decider.Decision;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.ToXContent.Params;
@@ -28,6 +29,8 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
 
 /**
  * Class used to encapsulate a number of {@link RerouteExplanation}
@@ -49,6 +52,18 @@ public List<RerouteExplanation> explanations() {
         return this.explanations;
     }
 
+    /**
+     * Provides feedback from commands with a YES decision that should be displayed to the user after the command has been applied
+     */
+    public List<String> getYesDecisionMessages() {
+        return explanations().stream()
+            .filter(explanation -> explanation.decisions().type().equals(Decision.Type.YES))
+            .map(explanation -> explanation.command().getMessage())
+            .filter(Optional::isPresent)
+            .map(Optional::get)
+            .collect(Collectors.toList());
+    }
+
     /**
      * Read in a RoutingExplanations object
      */

core/src/main/java/org/elasticsearch/cluster/routing/allocation/command/AllocateEmptyPrimaryAllocationCommand.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.index.shard.ShardNotFoundException;
 
 import java.io.IOException;
+import java.util.Optional;
 
 /**
  * Allocates an unassigned empty primary shard to a specific node. Use with extreme care as this will result in data loss.
@@ -72,6 +73,11 @@ public String name() {
         return NAME;
     }
 
+    @Override
+    public Optional<String> getMessage() {
+        return Optional.of("Allocated an empty primary for [" + index + "][" + shardId + "] on node [" + node + "]");
+    }
+
     public static AllocateEmptyPrimaryAllocationCommand fromXContent(XContentParser parser) throws IOException {
         return new Builder().parse(parser).build();
     }
@@ -115,19 +121,22 @@ public RerouteExplanation execute(RoutingAllocation allocation, boolean explain)
         }
 
         if (shardRouting.recoverySource().getType() != RecoverySource.Type.EMPTY_STORE && acceptDataLoss == false) {
-            return explainOrThrowRejectedCommand(explain, allocation,
-                "allocating an empty primary for [" + index + "][" + shardId + "] can result in data loss. Please confirm by setting the accept_data_loss parameter to true");
+            String dataLossWarning = "allocating an empty primary for [" + index + "][" + shardId + "] can result in data loss. Please confirm " +
+                "by setting the accept_data_loss parameter to true";
+            return explainOrThrowRejectedCommand(explain, allocation, dataLossWarning);
         }
 
         UnassignedInfo unassignedInfoToUpdate = null;
         if (shardRouting.unassignedInfo().getReason() != UnassignedInfo.Reason.FORCED_EMPTY_PRIMARY) {
-            unassignedInfoToUpdate = new UnassignedInfo(UnassignedInfo.Reason.FORCED_EMPTY_PRIMARY,
-                "force empty allocation from previous reason " + shardRouting.unassignedInfo().getReason() + ", " + shardRouting.unassignedInfo().getMessage(),
+            String unassignedInfoMessage = "force empty allocation from previous reason " + shardRouting.unassignedInfo().getReason() +
+                ", " + shardRouting.unassignedInfo().getMessage();
+            unassignedInfoToUpdate = new UnassignedInfo(UnassignedInfo.Reason.FORCED_EMPTY_PRIMARY, unassignedInfoMessage,
                 shardRouting.unassignedInfo().getFailure(), 0, System.nanoTime(), System.currentTimeMillis(), false,
                 shardRouting.unassignedInfo().getLastAllocationStatus());
         }
 
-        initializeUnassignedShard(allocation, routingNodes, routingNode, shardRouting, unassignedInfoToUpdate, StoreRecoverySource.EMPTY_STORE_INSTANCE);
+        initializeUnassignedShard(allocation, routingNodes, routingNode, shardRouting, unassignedInfoToUpdate,
+            StoreRecoverySource.EMPTY_STORE_INSTANCE);
 
         return new RerouteExplanation(this, allocation.decision(Decision.YES, name() + " (allocation command)", "ignore deciders"));
     }

core/src/main/java/org/elasticsearch/cluster/routing/allocation/command/AllocateStalePrimaryAllocationCommand.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.index.shard.ShardNotFoundException;
 
 import java.io.IOException;
+import java.util.Optional;
 
 /**
  * Allocates an unassigned stale primary shard to a specific node. Use with extreme care as this will result in data loss.
@@ -70,6 +71,11 @@ public String name() {
         return NAME;
     }
 
+    @Override
+    public Optional<String> getMessage() {
+        return Optional.of("Allocated a stale primary for [" + index + "][" + shardId + "] on node [" + node + "]");
+    }
+
     public static AllocateStalePrimaryAllocationCommand fromXContent(XContentParser parser) throws IOException {
         return new Builder().parse(parser).build();
     }
@@ -113,8 +119,9 @@ public RerouteExplanation execute(RoutingAllocation allocation, boolean explain)
         }
 
         if (acceptDataLoss == false) {
-            return explainOrThrowRejectedCommand(explain, allocation,
-                "allocating an empty primary for [" + index + "][" + shardId + "] can result in data loss. Please confirm by setting the accept_data_loss parameter to true");
+            String dataLossWarning = "allocating an empty primary for [" + index + "][" + shardId + "] can result in data loss. Please " +
+                "confirm by setting the accept_data_loss parameter to true";
+            return explainOrThrowRejectedCommand(explain, allocation, dataLossWarning);
         }
 
         if (shardRouting.recoverySource().getType() != RecoverySource.Type.EXISTING_STORE) {

core/src/main/java/org/elasticsearch/cluster/routing/allocation/command/AllocationCommand.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.Optional;
 
 /**
  * A command to move shards in some way.
@@ -61,4 +62,12 @@ interface Parser<T extends AllocationCommand> {
     default String getWriteableName() {
         return name();
     }
+
+    /**
+     * Returns any feedback the command wants to provide for logging. This message should be appropriate to expose to the user after the
+     * command has been applied
+     */
+    default Optional<String> getMessage() {
+        return Optional.empty();
+    }
 }

core/src/test/java/org/elasticsearch/cluster/allocation/ClusterRerouteIT.java

@@ -19,20 +19,24 @@
 
 package org.elasticsearch.cluster.allocation;
 
+import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.IOUtils;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
+import org.elasticsearch.action.admin.cluster.reroute.TransportClusterRerouteAction;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardRoutingState;
 import org.elasticsearch.cluster.routing.allocation.RerouteExplanation;
 import org.elasticsearch.cluster.routing.allocation.RoutingExplanations;
 import org.elasticsearch.cluster.routing.allocation.command.AllocateEmptyPrimaryAllocationCommand;
+import org.elasticsearch.cluster.routing.allocation.command.AllocationCommand;
 import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
 import org.elasticsearch.cluster.routing.allocation.decider.Decision;
 import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
@@ -50,6 +54,7 @@
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.test.MockLogAppender;
 
 import java.nio.file.Path;
 import java.util.Arrays;
@@ -63,6 +68,7 @@
 import static org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertBlocked;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 
@@ -304,6 +310,84 @@ public void testRerouteExplain() {
         assertThat(explanation.decisions().type(), equalTo(Decision.Type.YES));
     }
 
+    public void testMessageLogging() throws Exception{
+        final Settings settings = Settings.builder()
+            .put(EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING.getKey(), Allocation.NONE.name())
+            .put(EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING.getKey(), EnableAllocationDecider.Rebalance.NONE.name())
+            .build();
+
+        final String nodeName1 = internalCluster().startNode(settings);
+        assertThat(cluster().size(), equalTo(1));
+        ClusterHealthResponse healthResponse = client().admin().cluster().prepareHealth().setWaitForNodes("1")
+            .execute().actionGet();
+        assertThat(healthResponse.isTimedOut(), equalTo(false));
+
+        final String nodeName2 = internalCluster().startNode(settings);
+        assertThat(cluster().size(), equalTo(2));
+        healthResponse = client().admin().cluster().prepareHealth().setWaitForNodes("2").execute().actionGet();
+        assertThat(healthResponse.isTimedOut(), equalTo(false));
+
+        final String indexName = "test_index";
+        client().admin().indices().prepareCreate(indexName).setWaitForActiveShards(ActiveShardCount.NONE)
+            .setSettings(Settings.builder()
+            .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 2)
+            .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 1))
+        .execute().actionGet();
+
+        Logger actionLogger = Loggers.getLogger(TransportClusterRerouteAction.class);
+
+        MockLogAppender dryRunMockLog = new MockLogAppender();
+        dryRunMockLog.start();
+        dryRunMockLog.addExpectation(
+            new MockLogAppender.UnseenEventExpectation("no completed message logged on dry run",
+                TransportClusterRerouteAction.class.getName(), Level.INFO, "Allocated an empty primary*")
+        );
+        Loggers.addAppender(actionLogger, dryRunMockLog);
+
+        AllocationCommand dryRunAllocation = new AllocateEmptyPrimaryAllocationCommand(indexName, 0, nodeName1, true);
+        ClusterRerouteResponse dryRunResponse = client().admin().cluster().prepareReroute()
+            .setExplain(randomBoolean())
+            .setDryRun(true)
+            .add(dryRunAllocation)
+            .execute().actionGet();
+
+        // during a dry run, messages exist but are not logged or exposed
+        assertThat(dryRunResponse.getExplanations().getYesDecisionMessages(), hasSize(1));
+        assertThat(dryRunResponse.getExplanations().getYesDecisionMessages().get(0), containsString("Allocated an empty primary"));
+
+        dryRunMockLog.assertAllExpectationsMatched();
+        dryRunMockLog.stop();
+        Loggers.removeAppender(actionLogger, dryRunMockLog);
+
+        MockLogAppender allocateMockLog = new MockLogAppender();
+        allocateMockLog.start();
+        allocateMockLog.addExpectation(
+            new MockLogAppender.SeenEventExpectation("message for first allocate empty primary",
+                TransportClusterRerouteAction.class.getName(), Level.INFO, "Allocated an empty primary*" + nodeName1 + "*")
+        );
+        allocateMockLog.addExpectation(
+            new MockLogAppender.UnseenEventExpectation("no message for second allocate empty primary",
+                TransportClusterRerouteAction.class.getName(), Level.INFO, "Allocated an empty primary*" + nodeName2 + "*")
+        );
+        Loggers.addAppender(actionLogger, allocateMockLog);
+
+        AllocationCommand yesDecisionAllocation = new AllocateEmptyPrimaryAllocationCommand(indexName, 0, nodeName1, true);
+        AllocationCommand noDecisionAllocation = new AllocateEmptyPrimaryAllocationCommand("noexist", 1, nodeName2, true);
+        ClusterRerouteResponse response = client().admin().cluster().prepareReroute()
+            .setExplain(true) // so we get a NO decision back rather than an exception
+            .add(yesDecisionAllocation)
+            .add(noDecisionAllocation)
+            .execute().actionGet();
+
+        assertThat(response.getExplanations().getYesDecisionMessages(), hasSize(1));
+        assertThat(response.getExplanations().getYesDecisionMessages().get(0), containsString("Allocated an empty primary"));
+        assertThat(response.getExplanations().getYesDecisionMessages().get(0), containsString(nodeName1));
+
+        allocateMockLog.assertAllExpectationsMatched();
+        allocateMockLog.stop();
+        Loggers.removeAppender(actionLogger, allocateMockLog);
+    }
+
     public void testClusterRerouteWithBlocks() throws Exception {
         List<String> nodesIds = internalCluster().startNodes(2);