Improve error message for non append-only writes that target data stream (#60887)

Backport of #60874 to 7.9 branch.

Closes #60581

Author: martijnvg

server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

@@ -89,6 +89,7 @@
 import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyMap;
+import static org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.EXCLUDED_DATA_STREAMS_KEY;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
 
@@ -577,7 +578,7 @@ private boolean addFailureIfIndexIsUnavailable(DocWriteRequest<?> request, int i
             if (concreteIndex == null) {
                 try {
                     concreteIndex = concreteIndices.resolveIfAbsent(request);
-                } catch (IndexClosedException | IndexNotFoundException ex) {
+                } catch (IndexClosedException | IndexNotFoundException | IllegalArgumentException ex) {
                     addFailure(request, idx, ex);
                     return true;
                 }
@@ -623,8 +624,16 @@ Index resolveIfAbsent(DocWriteRequest<?> request) {
             Index concreteIndex = indices.get(request.index());
             if (concreteIndex == null) {
                 boolean includeDataStreams = request.opType() == DocWriteRequest.OpType.CREATE;
-                concreteIndex = indexNameExpressionResolver.concreteWriteIndex(state, request.indicesOptions(), request.indices()[0],
-                    false, includeDataStreams);
+                try {
+                    concreteIndex = indexNameExpressionResolver.concreteWriteIndex(state, request.indicesOptions(),
+                        request.indices()[0], false, includeDataStreams);
+                } catch (IndexNotFoundException e) {
+                    if (includeDataStreams == false && e.getMetadataKeys().contains(EXCLUDED_DATA_STREAMS_KEY)) {
+                        throw new IllegalArgumentException("only write ops with an op_type of create are allowed in data streams");
+                    } else {
+                        throw e;
+                    }
+                }
                 indices.put(request.index(), concreteIndex);
             }
             return concreteIndex;

server/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -50,6 +51,8 @@
 
 import java.io.IOException;
 
+import static org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.EXCLUDED_DATA_STREAMS_KEY;
+
 public abstract class TransportInstanceSingleOperationAction<
             Request extends InstanceShardOperationRequest<Request>,
             Response extends ActionResponse
@@ -143,7 +146,15 @@ protected void doStart(ClusterState clusterState) {
                         throw blockException;
                     }
                 }
-                request.concreteIndex(indexNameExpressionResolver.concreteWriteIndex(clusterState, request).getName());
+                try {
+                    request.concreteIndex(indexNameExpressionResolver.concreteWriteIndex(clusterState, request).getName());
+                } catch (IndexNotFoundException e) {
+                    if (request.includeDataStreams() == false && e.getMetadataKeys().contains(EXCLUDED_DATA_STREAMS_KEY)) {
+                        throw new IllegalArgumentException("only write ops with an op_type of create are allowed in data streams");
+                    } else {
+                        throw e;
+                    }
+                }
                 resolveRequest(clusterState, request);
                 blockException = checkRequestBlock(clusterState, request);
                 if (blockException != null) {

server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java

@@ -60,6 +60,8 @@
 
 public class IndexNameExpressionResolver {
 
+    public static final String EXCLUDED_DATA_STREAMS_KEY = "es.excluded_ds";
+
     private final DateMathExpressionResolver dateMathExpressionResolver = new DateMathExpressionResolver();
     private final WildcardExpressionResolver wildcardExpressionResolver = new WildcardExpressionResolver();
     private final List<ExpressionResolver> expressionResolvers =
@@ -208,6 +210,7 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
             }
         }
 
+        boolean excludedDataStreams = false;
         final Set<Index> concreteIndices = new HashSet<>(expressions.size());
         for (String expression : expressions) {
             IndexAbstraction indexAbstraction = metadata.getIndicesLookup().get(expression);
@@ -232,6 +235,7 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
                 }
             } else if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM &&
                         context.includeDataStreams() == false) {
+                excludedDataStreams = true;
                 continue;
             }
 
@@ -273,6 +277,10 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
         if (options.allowNoIndices() == false && concreteIndices.isEmpty()) {
             IndexNotFoundException infe = new IndexNotFoundException((String)null);
             infe.setResources("index_expression", indexExpressions);
+            if (excludedDataStreams) {
+                // Allows callers to handle IndexNotFoundException differently based on whether data streams were excluded.
+                infe.addMetadata(EXCLUDED_DATA_STREAMS_KEY, "true");
+            }
             throw infe;
         }
         return concreteIndices.toArray(new Index[concreteIndices.size()]);

x-pack/plugin/data-streams/qa/rest/src/test/resources/rest-api-spec/test/data-streams/20_unsupported_apis.yml

@@ -196,3 +196,50 @@
       indices.delete_data_stream:
         name: simple-data-stream1
   - is_true: acknowledged
+
+---
+"Non append-only writes into a data stream":
+  - skip:
+      version: " - 7.8.99"
+      reason: "data streams only supported in 7.9+"
+      features: allowed_warnings
+
+  - do:
+      allowed_warnings:
+        - "index template [my-template1] has index patterns [logs-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template1
+        body:
+          index_patterns: [logs-*]
+          data_stream: {}
+
+  - do:
+      catch: bad_request
+      index:
+        index:  logs-foobar
+        id: "1"
+        body:
+          '@timestamp': '2020-12-12'
+
+  - do:
+      bulk:
+        body:
+          - index:
+              _index: logs-foobar
+              _id: "1"
+          - '@timestamp': '2020-12-12'
+          - create:
+              _index: logs-foobar
+              _id: "1"
+          - '@timestamp': '2020-12-12'
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: illegal_argument_exception }
+  - match: { items.0.index.error.reason: "only write ops with an op_type of create are allowed in data streams" }
+  - match: { items.1.create.result: created }
+  - match: { items.1.create._index: .ds-logs-foobar-000001 }
+
+  - do:
+      indices.delete_data_stream:
+        name: logs-foobar
+  - is_true: acknowledged

x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java

@@ -230,36 +230,45 @@ public void testOtherWriteOps() throws Exception {
             );
             BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
             assertThat(bulkResponse.getItems(), arrayWithSize(1));
-            assertThat(bulkResponse.getItems()[0].getFailure().getMessage(), containsString("no such index [null]"));
+            assertThat(
+                bulkResponse.getItems()[0].getFailure().getMessage(),
+                containsString("only write ops with an op_type of create are allowed in data streams")
+            );
         }
         {
             BulkRequest bulkRequest = new BulkRequest().add(new DeleteRequest(dataStreamName, "_id"));
             BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
             assertThat(bulkResponse.getItems(), arrayWithSize(1));
-            assertThat(bulkResponse.getItems()[0].getFailure().getMessage(), containsString("no such index [null]"));
+            assertThat(
+                bulkResponse.getItems()[0].getFailure().getMessage(),
+                containsString("only write ops with an op_type of create are allowed in data streams")
+            );
         }
         {
             BulkRequest bulkRequest = new BulkRequest().add(
                 new UpdateRequest(dataStreamName, "_id").doc("{\"@timestamp1\": \"2020-12-12\"}", XContentType.JSON)
             );
             BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
             assertThat(bulkResponse.getItems(), arrayWithSize(1));
-            assertThat(bulkResponse.getItems()[0].getFailure().getMessage(), containsString("no such index [null]"));
+            assertThat(
+                bulkResponse.getItems()[0].getFailure().getMessage(),
+                containsString("only write ops with an op_type of create are allowed in data streams")
+            );
         }
         {
             IndexRequest indexRequest = new IndexRequest(dataStreamName).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON);
-            Exception e = expectThrows(IndexNotFoundException.class, () -> client().index(indexRequest).actionGet());
-            assertThat(e.getMessage(), equalTo("no such index [null]"));
+            Exception e = expectThrows(IllegalArgumentException.class, () -> client().index(indexRequest).actionGet());
+            assertThat(e.getMessage(), equalTo("only write ops with an op_type of create are allowed in data streams"));
         }
         {
             UpdateRequest updateRequest = new UpdateRequest(dataStreamName, "_id").doc("{}", XContentType.JSON);
-            Exception e = expectThrows(IndexNotFoundException.class, () -> client().update(updateRequest).actionGet());
-            assertThat(e.getMessage(), equalTo("no such index [null]"));
+            Exception e = expectThrows(IllegalArgumentException.class, () -> client().update(updateRequest).actionGet());
+            assertThat(e.getMessage(), equalTo("only write ops with an op_type of create are allowed in data streams"));
         }
         {
             DeleteRequest deleteRequest = new DeleteRequest(dataStreamName, "_id");
-            Exception e = expectThrows(IndexNotFoundException.class, () -> client().delete(deleteRequest).actionGet());
-            assertThat(e.getMessage(), equalTo("no such index [null]"));
+            Exception e = expectThrows(IllegalArgumentException.class, () -> client().delete(deleteRequest).actionGet());
+            assertThat(e.getMessage(), equalTo("only write ops with an op_type of create are allowed in data streams"));
         }
         {
             IndexRequest indexRequest = new IndexRequest(dataStreamName).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON)