diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/indices/DataStream.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/indices/DataStream.java
index 76b5136bedaa6..1f81b233f0162 100644
--- a/client/rest-high-level/src/main/java/org/elasticsearch/client/indices/DataStream.java
+++ b/client/rest-high-level/src/main/java/org/elasticsearch/client/indices/DataStream.java
@@ -36,6 +36,7 @@ public final class DataStream {
     private final String timeStampField;
     private final List<String> indices;
     private final long generation;
+    private final boolean hidden;
     ClusterHealthStatus dataStreamStatus;
     @Nullable
     String indexTemplate;
@@ -45,7 +46,8 @@ public final class DataStream {
     private final Map<String, Object> metadata;
 
     public DataStream(String name, String timeStampField, List<String> indices, long generation, ClusterHealthStatus dataStreamStatus,
-                      @Nullable String indexTemplate, @Nullable String ilmPolicyName, @Nullable  Map<String, Object> metadata) {
+                      @Nullable String indexTemplate, @Nullable String ilmPolicyName, @Nullable  Map<String, Object> metadata,
+                      boolean hidden) {
         this.name = name;
         this.timeStampField = timeStampField;
         this.indices = indices;
@@ -54,6 +56,7 @@ public DataStream(String name, String timeStampField, List<String> indices, long
         this.indexTemplate = indexTemplate;
         this.ilmPolicyName = ilmPolicyName;
         this.metadata = metadata;
+        this.hidden = hidden;
     }
 
     public String getName() {
@@ -88,6 +91,10 @@ public Map<String, Object> getMetadata() {
         return metadata;
     }
 
+    public boolean isHidden() {
+        return hidden;
+    }
+
     public static final ParseField NAME_FIELD = new ParseField("name");
     public static final ParseField TIMESTAMP_FIELD_FIELD = new ParseField("timestamp_field");
     public static final ParseField INDICES_FIELD = new ParseField("indices");
@@ -96,6 +103,7 @@ public Map<String, Object> getMetadata() {
     public static final ParseField INDEX_TEMPLATE_FIELD = new ParseField("template");
     public static final ParseField ILM_POLICY_FIELD = new ParseField("ilm_policy");
     public static final ParseField METADATA_FIELD = new ParseField("_meta");
+    public static final ParseField HIDDEN_FIELD = new ParseField("hidden");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream",
@@ -110,7 +118,9 @@ public Map<String, Object> getMetadata() {
             String indexTemplate = (String) args[5];
             String ilmPolicy = (String) args[6];
             Map<String, Object> metadata = (Map<String, Object>) args[7];
-            return new DataStream(dataStreamName, timeStampField, indices, generation, status, indexTemplate, ilmPolicy, metadata);
+            Boolean hidden = (Boolean) args[8];
+            hidden = hidden != null && hidden;
+            return new DataStream(dataStreamName, timeStampField, indices, generation, status, indexTemplate, ilmPolicy, metadata, hidden);
         });
 
     static {
@@ -122,6 +132,7 @@ public Map<String, Object> getMetadata() {
         PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), INDEX_TEMPLATE_FIELD);
         PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), ILM_POLICY_FIELD);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), METADATA_FIELD);
+        PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), HIDDEN_FIELD);
     }
 
     public static DataStream fromXContent(XContentParser parser) throws IOException {
diff --git a/docs/reference/data-streams/change-mappings-and-settings.asciidoc b/docs/reference/data-streams/change-mappings-and-settings.asciidoc
index ddfada9a67281..9de3b893367a5 100644
--- a/docs/reference/data-streams/change-mappings-and-settings.asciidoc
+++ b/docs/reference/data-streams/change-mappings-and-settings.asciidoc
@@ -572,7 +572,8 @@ contains information about the stream's oldest backing index,
       ],
       "generation": 2,
       "status": "GREEN",
-      "template": "my-data-stream-template"
+      "template": "my-data-stream-template",
+      "hidden": false
     }
   ]
 }
diff --git a/docs/reference/indices/get-data-stream.asciidoc b/docs/reference/indices/get-data-stream.asciidoc
index 5f5e478cd6bb7..50768b9350ac1 100644
--- a/docs/reference/indices/get-data-stream.asciidoc
+++ b/docs/reference/indices/get-data-stream.asciidoc
@@ -209,7 +209,8 @@ The API returns the following response:
       "generation": 2,
       "status": "GREEN",
       "template": "my-index-template",
-      "ilm_policy": "my-lifecycle-policy"
+      "ilm_policy": "my-lifecycle-policy",
+      "hidden": false
     },
     {
       "name": "my-data-stream_two",
@@ -225,7 +226,8 @@ The API returns the following response:
       "generation": 1,
       "status": "YELLOW",
       "template": "my-index-template",
-      "ilm_policy": "my-lifecycle-policy"
+      "ilm_policy": "my-lifecycle-policy",
+      "hidden": false
     }
   ]
 }
diff --git a/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java b/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java
index e68c0f271e135..65fb2b75d693f 100644
--- a/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java
+++ b/server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java
@@ -225,6 +225,13 @@ public EnumSet<WildcardStates> getExpandWildcards() {
         return EnumSet.copyOf(expandWildcards);
     }
 
+    /**
+     * @return a copy of the {@link Option}s that these indices options will use
+     */
+    public EnumSet<Option> getOptions() {
+        return EnumSet.copyOf(options);
+    }
+
     public void writeIndicesOptions(StreamOutput out) throws IOException {
         out.writeEnumSet(options);
         out.writeEnumSet(expandWildcards);
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplate.java b/server/src/main/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplate.java
index 9b0d7449c49c3..e88796afc9d7b 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplate.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplate.java
@@ -29,7 +29,6 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;
-import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
@@ -279,20 +278,33 @@ public String toString() {
 
     public static class DataStreamTemplate implements Writeable, ToXContentObject {
 
-        private static final ObjectParser<DataStreamTemplate, Void> PARSER = new ObjectParser<>(
+        private static final ParseField HIDDEN = new ParseField("hidden");
+
+        public static final ConstructingObjectParser<DataStreamTemplate, Void> PARSER = new ConstructingObjectParser<>(
             "data_stream_template",
-            DataStreamTemplate::new
-        );
+            false,
+            a -> new DataStreamTemplate(a[0] != null && (boolean) a[0]));
+
+        static {
+            PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), HIDDEN);
+        }
+
+        private final boolean hidden;
 
         public DataStreamTemplate() {
+            this(false);
         }
 
-        public String getTimestampField() {
-            return FIXED_TIMESTAMP_FIELD;
+        public DataStreamTemplate(boolean hidden) {
+            this.hidden = hidden;
         }
 
-        DataStreamTemplate(StreamInput in) {
-            this();
+        DataStreamTemplate(StreamInput in) throws IOException {
+            hidden = in.getVersion().onOrAfter(DataStream.HIDDEN_VERSION) && in.readBoolean();
+        }
+
+        public String getTimestampField() {
+            return FIXED_TIMESTAMP_FIELD;
         }
 
         /**
@@ -303,13 +315,21 @@ public Map<String, Object> getDataStreamMappingSnippet() {
             return Map.of(MapperService.SINGLE_MAPPING_NAME, Map.of("_data_stream_timestamp", Map.of("enabled", true)));
         }
 
+        public boolean isHidden() {
+            return hidden;
+        }
+
         @Override
         public void writeTo(StreamOutput out) throws IOException {
+            if (out.getVersion().onOrAfter(DataStream.HIDDEN_VERSION)) {
+                out.writeBoolean(hidden);
+            }
         }
 
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
+            builder.field("hidden", hidden);
             builder.endObject();
             return builder;
         }
@@ -317,12 +337,14 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
-            return o != null && getClass() == o.getClass();
+            if (o == null || getClass() != o.getClass()) return false;
+            DataStreamTemplate that = (DataStreamTemplate) o;
+            return hidden == that.hidden;
         }
 
         @Override
         public int hashCode() {
-            return DataStreamTemplate.class.hashCode();
+            return Objects.hash(hidden);
         }
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
index 8437679a2453a..0faaa6f4d9283 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java
@@ -43,19 +43,27 @@
 public final class DataStream extends AbstractDiffable<DataStream> implements ToXContentObject {
 
     public static final String BACKING_INDEX_PREFIX = ".ds-";
+    public static final Version HIDDEN_VERSION = Version.V_8_0_0;
 
     private final String name;
     private final TimestampField timeStampField;
     private final List<Index> indices;
     private final long generation;
     private final Map<String, Object> metadata;
+    private final boolean hidden;
 
     public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation, Map<String, Object> metadata) {
+        this(name, timeStampField, indices, generation, metadata, false);
+    }
+
+    public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation, Map<String, Object> metadata,
+                      boolean hidden) {
         this.name = name;
         this.timeStampField = timeStampField;
         this.indices = Collections.unmodifiableList(indices);
         this.generation = generation;
         this.metadata = metadata;
+        this.hidden = hidden;
         assert indices.size() > 0;
     }
 
@@ -88,6 +96,10 @@ public Map<String, Object> getMetadata() {
         return metadata;
     }
 
+    public boolean isHidden() {
+        return hidden;
+    }
+
     /**
      * Performs a rollover on a {@code DataStream} instance and returns a new instance containing
      * the updated list of backing indices and incremented generation.
@@ -100,7 +112,7 @@ public DataStream rollover(Index newWriteIndex) {
         assert newWriteIndex.getName().equals(getDefaultBackingIndexName(name, generation + 1));
         List<Index> backingIndices = new ArrayList<>(indices);
         backingIndices.add(newWriteIndex);
-        return new DataStream(name, timeStampField, backingIndices, generation + 1, metadata);
+        return new DataStream(name, timeStampField, backingIndices, generation + 1, metadata, hidden);
     }
 
     /**
@@ -114,7 +126,7 @@ public DataStream removeBackingIndex(Index index) {
         List<Index> backingIndices = new ArrayList<>(indices);
         backingIndices.remove(index);
         assert backingIndices.size() == indices.size() - 1;
-        return new DataStream(name, timeStampField, backingIndices, generation, metadata);
+        return new DataStream(name, timeStampField, backingIndices, generation, metadata, hidden);
     }
 
     /**
@@ -139,7 +151,7 @@ public DataStream replaceBackingIndex(Index existingBackingIndex, Index newBacki
                 "it is the write index", existingBackingIndex.getName(), name));
         }
         backingIndices.set(backingIndexPosition, newBackingIndex);
-        return new DataStream(name, timeStampField, backingIndices, generation, metadata);
+        return new DataStream(name, timeStampField, backingIndices, generation, metadata, hidden);
     }
 
     /**
@@ -156,7 +168,8 @@ public static String getDefaultBackingIndexName(String dataStreamName, long gene
 
     public DataStream(StreamInput in) throws IOException {
         this(in.readString(), new TimestampField(in), in.readList(Index::new), in.readVLong(),
-            in.getVersion().onOrAfter(Version.V_7_11_0) ? in.readMap(): null);
+            in.getVersion().onOrAfter(Version.V_7_11_0) ? in.readMap(): null,
+            in.getVersion().onOrAfter(HIDDEN_VERSION) && in.readBoolean());
     }
 
     public static Diff<DataStream> readDiffFrom(StreamInput in) throws IOException {
@@ -172,6 +185,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_7_11_0)) {
             out.writeMap(metadata);
         }
+        if (out.getVersion().onOrAfter(HIDDEN_VERSION)) {
+            out.writeBoolean(hidden);
+        }
     }
 
     public static final ParseField NAME_FIELD = new ParseField("name");
@@ -179,11 +195,12 @@ public void writeTo(StreamOutput out) throws IOException {
     public static final ParseField INDICES_FIELD = new ParseField("indices");
     public static final ParseField GENERATION_FIELD = new ParseField("generation");
     public static final ParseField METADATA_FIELD = new ParseField("_meta");
+    public static final ParseField HIDDEN_FIELD = new ParseField("hidden");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream",
         args -> new DataStream((String) args[0], (TimestampField) args[1], (List<Index>) args[2], (Long) args[3],
-            (Map<String, Object>) args[4]));
+            (Map<String, Object>) args[4], args[5] != null && (boolean) args[5]));
 
     static {
         PARSER.declareString(ConstructingObjectParser.constructorArg(), NAME_FIELD);
@@ -191,6 +208,7 @@ public void writeTo(StreamOutput out) throws IOException {
         PARSER.declareObjectArray(ConstructingObjectParser.constructorArg(), (p, c) -> Index.fromXContent(p), INDICES_FIELD);
         PARSER.declareLong(ConstructingObjectParser.constructorArg(), GENERATION_FIELD);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), METADATA_FIELD);
+        PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), HIDDEN_FIELD);
     }
 
     public static DataStream fromXContent(XContentParser parser) throws IOException {
@@ -207,6 +225,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (metadata != null) {
             builder.field(METADATA_FIELD.getPreferredName(), metadata);
         }
+        builder.field(HIDDEN_FIELD.getPreferredName(), hidden);
         builder.endObject();
         return builder;
     }
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstraction.java b/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstraction.java
index af4066fb9acde..4b32c5ff429b7 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstraction.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstraction.java
@@ -358,7 +358,7 @@ public DataStream getParentDataStream() {
 
         @Override
         public boolean isHidden() {
-            return false;
+            return dataStream.isHidden();
         }
 
         @Override
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstractionResolver.java b/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstractionResolver.java
index da1917f783ab4..9638d3f7a5e88 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstractionResolver.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/IndexAbstractionResolver.java
@@ -131,23 +131,18 @@ public static boolean isIndexVisible(String expression, String index, IndicesOpt
             throw new IllegalStateException("could not resolve index abstraction [" + index + "]");
         }
         final boolean isHidden = indexAbstraction.isHidden();
+        boolean isVisible = isHidden == false || indicesOptions.expandWildcardsHidden() || isVisibleDueToImplicitHidden(expression, index);
         if (indexAbstraction.getType() == IndexAbstraction.Type.ALIAS) {
             //it's an alias, ignore expandWildcardsOpen and expandWildcardsClosed.
             //complicated to support those options with aliases pointing to multiple indices...
-            if (indicesOptions.ignoreAliases()) {
-                return false;
-            } else if (isHidden == false || indicesOptions.expandWildcardsHidden() || isVisibleDueToImplicitHidden(expression, index)) {
-                return true;
-            } else {
-                return false;
-            }
+            return isVisible && indicesOptions.ignoreAliases() == false;
         }
         if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM) {
-            return includeDataStreams;
+            return isVisible && includeDataStreams;
         }
         assert indexAbstraction.getIndices().size() == 1 : "concrete index must point to a single index";
         IndexMetadata indexMetadata = indexAbstraction.getIndices().get(0);
-        if (isHidden && indicesOptions.expandWildcardsHidden() == false && isVisibleDueToImplicitHidden(expression, index) == false) {
+        if (isVisible == false) {
             return false;
         }
 
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java
index 41e1192d78847..9fba4fbfe35e5 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java
@@ -159,8 +159,9 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
         if (dataStreamName.toLowerCase(Locale.ROOT).equals(dataStreamName) == false) {
             throw new IllegalArgumentException("data_stream [" + dataStreamName + "] must be lowercase");
         }
-        if (dataStreamName.startsWith(".")) {
-            throw new IllegalArgumentException("data_stream [" + dataStreamName + "] must not start with '.'");
+        if (dataStreamName.startsWith(DataStream.BACKING_INDEX_PREFIX)) {
+            throw new IllegalArgumentException("data_stream [" + dataStreamName + "] must not start with '"
+                + DataStream.BACKING_INDEX_PREFIX + "'");
         }
 
         ComposableIndexTemplate template = lookupTemplateForDataStream(dataStreamName, currentState.metadata());
@@ -189,8 +190,9 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
         DataStream.TimestampField timestampField = new DataStream.TimestampField(fieldName);
         List<Index> dsBackingIndices = backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList());
         dsBackingIndices.add(writeIndex.getIndex());
+        boolean hidden = template.getDataStreamTemplate().isHidden();
         DataStream newDataStream = new DataStream(dataStreamName, timestampField, dsBackingIndices, 1L,
-                                                  template.metadata() != null ? Map.copyOf(template.metadata()) : null);
+                                                  template.metadata() != null ? Map.copyOf(template.metadata()) : null, hidden);
         Metadata.Builder builder = Metadata.builder(currentState.metadata()).put(newDataStream);
         logger.info("adding data stream [{}] with write index [{}] and backing indices [{}]", dataStreamName,
             writeIndex.getIndex().getName(),
diff --git a/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java b/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java
index 9dc8c630fcecd..efa43cf4d4bc3 100644
--- a/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java
+++ b/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java
@@ -604,7 +604,7 @@ static DataStream updateDataStream(DataStream dataStream, Metadata.Builder metad
             .map(i -> metadata.get(renameIndex(i.getName(), request, true)).getIndex())
             .collect(Collectors.toList());
         return new DataStream(dataStreamName, dataStream.getTimeStampField(), updatedIndices, dataStream.getGeneration(),
-            dataStream.getMetadata());
+            dataStream.getMetadata(), dataStream.isHidden());
     }
 
     public static RestoreInProgress updateRestoreStateWithDeletedIndices(RestoreInProgress oldRestore, Set<Index> deletedIndices) {
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java
index d592d861ae9b7..da6df7a532f77 100644
--- a/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java
@@ -2091,6 +2091,31 @@ public void testDataStreamsWithRegularIndexAndAlias() {
         assertThat(result[2].getName(), equalTo("logs-foobarbaz-0"));
     }
 
+    public void testHiddenDataStreams() {
+        final String dataStream1 = "logs-foobar";
+        IndexMetadata index1 = createBackingIndex(dataStream1, 1).build();
+        IndexMetadata index2 = createBackingIndex(dataStream1, 2).build();
+        IndexMetadata justAnIndex = IndexMetadata.builder("logs-foobarbaz-0")
+            .settings(ESTestCase.settings(Version.CURRENT))
+            .numberOfShards(1)
+            .numberOfReplicas(1)
+            .build();
+
+        ClusterState state = ClusterState.builder(new ClusterName("_name"))
+            .metadata(Metadata.builder()
+                .put(index1, false)
+                .put(index2, false)
+                .put(justAnIndex, false)
+                .put(new DataStream(dataStream1, createTimestampField("@timestamp"),
+                    List.of(index1.getIndex(), index2.getIndex()), 2, Collections.emptyMap(), true))).build();
+
+        Index[] result = indexNameExpressionResolver.concreteIndices(state, IndicesOptions.strictExpandHidden(), true, "logs-*");
+        assertThat(result, arrayContainingInAnyOrder(index1.getIndex(), index2.getIndex(), justAnIndex.getIndex() ));
+
+        result = indexNameExpressionResolver.concreteIndices(state, IndicesOptions.strictExpandOpen(), true, "logs-*");
+        assertThat(result, arrayContaining(justAnIndex.getIndex()));
+    }
+
     public void testDataStreamsNames() {
         final String dataStream1 = "logs-foobar";
         final String dataStream2 = "other-foobar";
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamServiceTests.java
index 4c7159166cb8b..b8abd6182d5cf 100644
--- a/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamServiceTests.java
+++ b/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamServiceTests.java
@@ -102,13 +102,13 @@ public void testCreateDataStreamWithUppercaseCharacters() throws Exception {
 
     public void testCreateDataStreamStartingWithPeriod() throws Exception {
         final MetadataCreateIndexService metadataCreateIndexService = getMetadataCreateIndexService();
-        final String dataStreamName = ".may_not_start_with_period";
+        final String dataStreamName = ".ds-may_not_start_with_ds";
         ClusterState cs = ClusterState.builder(new ClusterName("_name")).build();
         CreateDataStreamClusterStateUpdateRequest req =
             new CreateDataStreamClusterStateUpdateRequest(dataStreamName, TimeValue.ZERO, TimeValue.ZERO);
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
             () -> MetadataCreateDataStreamService.createDataStream(metadataCreateIndexService, cs, req));
-        assertThat(e.getMessage(), containsString("data_stream [" + dataStreamName + "] must not start with '.'"));
+        assertThat(e.getMessage(), containsString("data_stream [" + dataStreamName + "] must not start with '.ds-'"));
     }
 
     public void testCreateDataStreamNoTemplate() throws Exception {
diff --git a/test/framework/src/main/java/org/elasticsearch/cluster/DataStreamTestHelper.java b/test/framework/src/main/java/org/elasticsearch/cluster/DataStreamTestHelper.java
index 8f7871e736c1e..8ba80f0466984 100644
--- a/test/framework/src/main/java/org/elasticsearch/cluster/DataStreamTestHelper.java
+++ b/test/framework/src/main/java/org/elasticsearch/cluster/DataStreamTestHelper.java
@@ -109,7 +109,7 @@ public static DataStream randomInstance() {
         if (randomBoolean()) {
             metadata = Map.of("key", "value");
         }
-        return new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation, metadata);
+        return new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation, metadata, randomBoolean());
     }
 
     /**
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/DeleteDataStreamAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/DeleteDataStreamAction.java
index 47f64a4b9e1fa..23cc8e2f2559f 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/DeleteDataStreamAction.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/DeleteDataStreamAction.java
@@ -12,6 +12,7 @@
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.MasterNodeRequest;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.regex.Regex;
@@ -43,6 +44,7 @@ public static class Request extends MasterNodeRequest<Request> implements Indice
         // empty response can be returned in case wildcards were used or
         // 404 status code returned in case no wildcard were used.
         private final boolean wildcardExpressionsOriginallySpecified;
+        private IndicesOptions indicesOptions = IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
 
         public Request(String[] names) {
             this.names = Objects.requireNonNull(names);
@@ -66,6 +68,9 @@ public Request(StreamInput in) throws IOException {
             super(in);
             this.names = in.readStringArray();
             this.wildcardExpressionsOriginallySpecified = in.getVersion().onOrAfter(Version.V_7_10_0) && in.readBoolean();
+            if (in.getVersion().onOrAfter(DataStream.HIDDEN_VERSION)) {
+                this.indicesOptions = IndicesOptions.readIndicesOptions(in);
+            }
         }
 
         @Override
@@ -75,6 +80,9 @@ public void writeTo(StreamOutput out) throws IOException {
             if (out.getVersion().onOrAfter(Version.V_7_10_0)) {
                 out.writeBoolean(wildcardExpressionsOriginallySpecified);
             }
+            if (out.getVersion().onOrAfter(DataStream.HIDDEN_VERSION)) {
+                indicesOptions.writeIndicesOptions(out);
+            }
         }
 
         @Override
@@ -83,12 +91,12 @@ public boolean equals(Object o) {
             if (o == null || getClass() != o.getClass()) return false;
             Request request = (Request) o;
             return wildcardExpressionsOriginallySpecified == request.wildcardExpressionsOriginallySpecified &&
-                Arrays.equals(names, request.names);
+                Arrays.equals(names, request.names) && indicesOptions.equals(request.indicesOptions);
         }
 
         @Override
         public int hashCode() {
-            int result = Objects.hash(wildcardExpressionsOriginallySpecified);
+            int result = Objects.hash(wildcardExpressionsOriginallySpecified, indicesOptions);
             result = 31 * result + Arrays.hashCode(names);
             return result;
         }
@@ -100,9 +108,12 @@ public String[] indices() {
 
         @Override
         public IndicesOptions indicesOptions() {
-            // this doesn't really matter since data stream name resolution isn't affected by IndicesOptions and
-            // a data stream's backing indices are retrieved from its metadata
-            return IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
+            return indicesOptions;
+        }
+
+        public IndicesRequest indicesOptions(IndicesOptions indicesOptions) {
+            this.indicesOptions = indicesOptions;
+            return this;
         }
 
         @Override
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/GetDataStreamAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/GetDataStreamAction.java
index 8f0a12b805217..1ee0f6edac3b7 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/GetDataStreamAction.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/GetDataStreamAction.java
@@ -38,6 +38,7 @@ private GetDataStreamAction() {
     public static class Request extends MasterNodeReadRequest<Request> implements IndicesRequest.Replaceable {
 
         private String[] names;
+        private IndicesOptions indicesOptions = IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
 
         public Request(String[] names) {
             this.names = names;
@@ -55,12 +56,18 @@ public ActionRequestValidationException validate() {
         public Request(StreamInput in) throws IOException {
             super(in);
             this.names = in.readOptionalStringArray();
+            if (in.getVersion().onOrAfter(DataStream.HIDDEN_VERSION)) {
+                this.indicesOptions = IndicesOptions.readIndicesOptions(in);
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeOptionalStringArray(names);
+            if (out.getVersion().onOrAfter(DataStream.HIDDEN_VERSION)) {
+                indicesOptions.writeIndicesOptions(out);
+            }
         }
 
         @Override
@@ -68,12 +75,15 @@ public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Request request = (Request) o;
-            return Arrays.equals(names, request.names);
+            return Arrays.equals(names, request.names) &&
+                indicesOptions.equals(request.indicesOptions);
         }
 
         @Override
         public int hashCode() {
-            return Arrays.hashCode(names);
+            int result = Objects.hash(indicesOptions);
+            result = 31 * result + Arrays.hashCode(names);
+            return result;
         }
 
         @Override
@@ -83,9 +93,12 @@ public String[] indices() {
 
         @Override
         public IndicesOptions indicesOptions() {
-            // this doesn't really matter since data stream name resolution isn't affected by IndicesOptions and
-            // a data stream's backing indices are retrieved from its metadata
-            return IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
+            return indicesOptions;
+        }
+
+        public Request indicesOptions(IndicesOptions indicesOptions){
+            this.indicesOptions = indicesOptions;
+            return this;
         }
 
         @Override
@@ -108,6 +121,7 @@ public static class DataStreamInfo extends AbstractDiffable<DataStreamInfo> impl
             public static final ParseField STATUS_FIELD = new ParseField("status");
             public static final ParseField INDEX_TEMPLATE_FIELD = new ParseField("template");
             public static final ParseField ILM_POLICY_FIELD = new ParseField("ilm_policy");
+            public static final ParseField HIDDEN_FIELD = new ParseField("hidden");
 
             DataStream dataStream;
             ClusterHealthStatus dataStreamStatus;
@@ -169,6 +183,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 if (ilmPolicyName != null) {
                     builder.field(ILM_POLICY_FIELD.getPreferredName(), ilmPolicyName);
                 }
+                builder.field(HIDDEN_FIELD.getPreferredName(), dataStream.isHidden());
                 builder.endObject();
                 return builder;
             }
diff --git a/x-pack/plugin/data-streams/qa/rest/build.gradle b/x-pack/plugin/data-streams/qa/rest/build.gradle
index 1b65032dd021d..9ddc0858b59e3 100644
--- a/x-pack/plugin/data-streams/qa/rest/build.gradle
+++ b/x-pack/plugin/data-streams/qa/rest/build.gradle
@@ -1,4 +1,5 @@
 apply plugin: 'elasticsearch.yaml-rest-test'
+apply plugin: 'elasticsearch.java-rest-test'
 
 restResources {
   restApi {
diff --git a/x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamUpgradeRestIT.java b/x-pack/plugin/data-streams/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamUpgradeRestIT.java
similarity index 100%
rename from x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamUpgradeRestIT.java
rename to x-pack/plugin/data-streams/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamUpgradeRestIT.java
diff --git a/x-pack/plugin/data-streams/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsRestIT.java b/x-pack/plugin/data-streams/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsRestIT.java
new file mode 100644
index 0000000000000..5bfdd29edef4d
--- /dev/null
+++ b/x-pack/plugin/data-streams/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsRestIT.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.datastreams;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.common.xcontent.support.XContentMapValues;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.junit.After;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+
+public class DataStreamsRestIT extends ESRestTestCase {
+
+    @After
+    public void cleanUp() throws IOException {
+        adminClient().performRequest(new Request("DELETE", "_data_stream/*?expand_wildcards=hidden"));
+    }
+
+    public void testHiddenDataStream() throws IOException {
+        // Create a template
+        Request putComposableIndexTemplateRequest = new Request("POST", "/_index_template/hidden");
+        putComposableIndexTemplateRequest.setJsonEntity("{\"index_patterns\": [\"hidden\"], \"data_stream\": {\"hidden\": true}}");
+        assertOK(client().performRequest(putComposableIndexTemplateRequest));
+
+        Request createDocRequest = new Request("POST", "/hidden/_doc?refresh=true");
+        createDocRequest.setJsonEntity("{ \"@timestamp\": \"2020-10-22\", \"a\": 1 }");
+
+        assertOK(client().performRequest(createDocRequest));
+
+        Request getDataStreamsRequest = new Request("GET", "/_data_stream?expand_wildcards=hidden");
+        Response response = client().performRequest(getDataStreamsRequest);
+        Map<String, Object> dataStreams = entityAsMap(response);
+        assertEquals(Collections.singletonList("hidden"), XContentMapValues.extractValue("data_streams.name", dataStreams));
+        assertEquals(Collections.singletonList("hidden"), XContentMapValues.extractValue("data_streams.template", dataStreams));
+        assertEquals(Collections.singletonList(1), XContentMapValues.extractValue("data_streams.generation", dataStreams));
+        assertEquals(Collections.singletonList(true), XContentMapValues.extractValue("data_streams.hidden", dataStreams));
+
+        Request searchRequest = new Request("GET", "/hidd*/_search");
+        response = client().performRequest(searchRequest);
+        Map<String, Object> results = entityAsMap(response);
+        assertEquals(0, XContentMapValues.extractValue("hits.total.value", results));
+
+        searchRequest = new Request("GET", "/hidd*/_search?expand_wildcards=open,hidden");
+        response = client().performRequest(searchRequest);
+        results = entityAsMap(response);
+        assertEquals(1, XContentMapValues.extractValue("hits.total.value", results));
+    }
+
+    public void testHiddenDataStreamImplicitHiddenSearch() throws IOException {
+        // Create a template
+        Request putComposableIndexTemplateRequest = new Request("POST", "/_index_template/hidden");
+        putComposableIndexTemplateRequest.setJsonEntity("{\"index_patterns\": [\".hidden\"], \"data_stream\": {\"hidden\": true}}");
+        assertOK(client().performRequest(putComposableIndexTemplateRequest));
+
+        Request createDocRequest = new Request("POST", "/.hidden/_doc?refresh=true");
+        createDocRequest.setJsonEntity("{ \"@timestamp\": \"2020-10-22\", \"a\": 1 }");
+
+        assertOK(client().performRequest(createDocRequest));
+
+        Request getDataStreamsRequest = new Request("GET", "/_data_stream?expand_wildcards=hidden");
+        Response response = client().performRequest(getDataStreamsRequest);
+        Map<String, Object> dataStreams = entityAsMap(response);
+        assertEquals(Collections.singletonList(".hidden"), XContentMapValues.extractValue("data_streams.name", dataStreams));
+        assertEquals(Collections.singletonList("hidden"), XContentMapValues.extractValue("data_streams.template", dataStreams));
+        assertEquals(Collections.singletonList(1), XContentMapValues.extractValue("data_streams.generation", dataStreams));
+        assertEquals(Collections.singletonList(true), XContentMapValues.extractValue("data_streams.hidden", dataStreams));
+
+        Request searchRequest = new Request("GET", "/.hidd*/_search");
+        response = client().performRequest(searchRequest);
+        Map<String, Object> results = entityAsMap(response);
+        assertEquals(1, XContentMapValues.extractValue("hits.total.value", results));
+    }
+}
diff --git a/x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsRestIT.java b/x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsClientYamlTestSuiteIT.java
similarity index 79%
rename from x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsRestIT.java
rename to x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsClientYamlTestSuiteIT.java
index 1d40326391ace..ddca84d5be7cd 100644
--- a/x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsRestIT.java
+++ b/x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/DataStreamsClientYamlTestSuiteIT.java
@@ -10,9 +10,9 @@
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
 
-public class DataStreamsRestIT extends ESClientYamlSuiteTestCase {
+public class DataStreamsClientYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
 
-    public DataStreamsRestIT(final ClientYamlTestCandidate testCandidate) {
+    public DataStreamsClientYamlTestSuiteIT(final ClientYamlTestCandidate testCandidate) {
         super(testCandidate);
     }
 
diff --git a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/DataStreamsActionUtil.java b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/DataStreamsActionUtil.java
new file mode 100644
index 0000000000000..f7e12bfd07925
--- /dev/null
+++ b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/DataStreamsActionUtil.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.datastreams.action;
+
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+
+import java.util.EnumSet;
+import java.util.List;
+
+public class DataStreamsActionUtil {
+
+    /**
+     * Gets data streams names, expanding wildcards using {@link IndicesOptions} provided.
+     * For data streams we only care for {@link IndicesOptions.WildcardStates}.HIDDEN state (we can't have closed or open data streams),
+     * but we have to have either OPEN or CLOSE to have any names returned from {@link IndexNameExpressionResolver}. So here we always
+     * add OPEN to make sure that happens.
+     */
+    public static List<String> getDataStreamNames(
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        ClusterState currentState,
+        String[] names,
+        IndicesOptions indicesOptions
+    ) {
+        EnumSet<IndicesOptions.WildcardStates> expandWildcards = indicesOptions.getExpandWildcards();
+        expandWildcards.add(IndicesOptions.WildcardStates.OPEN);
+        indicesOptions = new IndicesOptions(indicesOptions.getOptions(), expandWildcards);
+        return indexNameExpressionResolver.dataStreamNames(currentState, indicesOptions, names);
+    }
+}
diff --git a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/DeleteDataStreamTransportAction.java b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/DeleteDataStreamTransportAction.java
index f7be4e2a7d7f6..0a321fbc3255a 100644
--- a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/DeleteDataStreamTransportAction.java
+++ b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/DeleteDataStreamTransportAction.java
@@ -35,8 +35,11 @@
 
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
+import static org.elasticsearch.xpack.datastreams.action.DataStreamsActionUtil.getDataStreamNames;
+
 public class DeleteDataStreamTransportAction extends AcknowledgedTransportMasterNodeAction<DeleteDataStreamAction.Request> {
 
     private static final Logger LOGGER = LogManager.getLogger(DeleteDataStreamTransportAction.class);
@@ -100,9 +103,8 @@ static ClusterState removeDataStream(
         ClusterState currentState,
         DeleteDataStreamAction.Request request
     ) {
-        Set<String> dataStreams = new HashSet<>(
-            indexNameExpressionResolver.dataStreamNames(currentState, request.indicesOptions(), request.getNames())
-        );
+        List<String> names = getDataStreamNames(indexNameExpressionResolver, currentState, request.getNames(), request.indicesOptions());
+        Set<String> dataStreams = new HashSet<>(names);
         Set<String> snapshottingDataStreams = SnapshotsService.snapshottingDataStreams(currentState, dataStreams);
 
         if (dataStreams.isEmpty()) {
diff --git a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/GetDataStreamsTransportAction.java b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/GetDataStreamsTransportAction.java
index cb14e5610552d..f60efa582ff78 100644
--- a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/GetDataStreamsTransportAction.java
+++ b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/action/GetDataStreamsTransportAction.java
@@ -98,7 +98,7 @@ static List<DataStream> getDataStreams(
         IndexNameExpressionResolver iner,
         GetDataStreamAction.Request request
     ) {
-        List<String> results = iner.dataStreamNames(clusterState, request.indicesOptions(), request.getNames());
+        List<String> results = DataStreamsActionUtil.getDataStreamNames(iner, clusterState, request.getNames(), request.indicesOptions());
         Map<String, DataStream> dataStreams = clusterState.metadata().dataStreams();
 
         return results.stream().map(dataStreams::get).sorted(Comparator.comparing(DataStream::getName)).collect(Collectors.toList());
diff --git a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestDeleteDataStreamAction.java b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestDeleteDataStreamAction.java
index 1a142453c45c5..4ceef6c36b619 100644
--- a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestDeleteDataStreamAction.java
+++ b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestDeleteDataStreamAction.java
@@ -5,14 +5,14 @@
  */
 package org.elasticsearch.xpack.datastreams.rest;
 
-import org.elasticsearch.xpack.core.action.DeleteDataStreamAction;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xpack.core.action.DeleteDataStreamAction;
 
-import java.io.IOException;
 import java.util.List;
 
 public class RestDeleteDataStreamAction extends BaseRestHandler {
@@ -28,10 +28,11 @@ public List<Route> routes() {
     }
 
     @Override
-    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
         DeleteDataStreamAction.Request deleteDataStreamRequest = new DeleteDataStreamAction.Request(
             Strings.splitStringByCommaToArray(request.param("name"))
         );
+        deleteDataStreamRequest.indicesOptions(IndicesOptions.fromRequest(request, deleteDataStreamRequest.indicesOptions()));
         return channel -> client.execute(DeleteDataStreamAction.INSTANCE, deleteDataStreamRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestGetDataStreamsAction.java b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestGetDataStreamsAction.java
index 5efdc63bd5f18..0d80a5e6fdd56 100644
--- a/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestGetDataStreamsAction.java
+++ b/x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/rest/RestGetDataStreamsAction.java
@@ -5,14 +5,14 @@
  */
 package org.elasticsearch.xpack.datastreams.rest;
 
-import org.elasticsearch.xpack.core.action.GetDataStreamAction;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xpack.core.action.GetDataStreamAction;
 
-import java.io.IOException;
 import java.util.List;
 
 public class RestGetDataStreamsAction extends BaseRestHandler {
@@ -28,10 +28,11 @@ public List<Route> routes() {
     }
 
     @Override
-    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
         GetDataStreamAction.Request getDataStreamsRequest = new GetDataStreamAction.Request(
             Strings.splitStringByCommaToArray(request.param("name"))
         );
+        getDataStreamsRequest.indicesOptions(IndicesOptions.fromRequest(request, getDataStreamsRequest.indicesOptions()));
         return channel -> client.execute(GetDataStreamAction.INSTANCE, getDataStreamsRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.delete_data_stream.json b/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.delete_data_stream.json
index c7b562f370969..c9e7f995c3a0a 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.delete_data_stream.json
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.delete_data_stream.json
@@ -21,6 +21,19 @@
         }
       ]
     },
-    "params":{}
+    "params":{
+      "expand_wildcards":{
+        "type":"enum",
+        "options":[
+          "open",
+          "closed",
+          "hidden",
+          "none",
+          "all"
+        ],
+        "default":"open",
+        "description":"Whether wildcard expressions should get expanded to open or closed indices (default: open)"
+      }
+    }
   }
 }
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.get_data_stream.json b/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.get_data_stream.json
index b75ae14a48b60..ab0d4de8bb779 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.get_data_stream.json
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/api/indices.get_data_stream.json
@@ -28,6 +28,18 @@
       ]
     },
     "params":{
+      "expand_wildcards":{
+        "type":"enum",
+        "options":[
+          "open",
+          "closed",
+          "hidden",
+          "none",
+          "all"
+        ],
+        "default":"open",
+        "description":"Whether wildcard expressions should get expanded to open or closed indices (default: open)"
+      }
     }
   }
 }
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_basic.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_basic.yml
index 235553581f864..96bcfd9555341 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_basic.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_basic.yml
@@ -27,6 +27,16 @@ setup:
                   type: date_nanos
           data_stream: {}
 
+  - do:
+      allowed_warnings:
+        - "index template [my-template3] has index patterns [.hidden-data-stream,hidden-data-stream] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template3] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template3
+        body:
+          index_patterns: [.hidden-data-stream,hidden-data-stream]
+          data_stream:
+            hidden: true
+
 ---
 "Create data stream":
   - skip:
@@ -53,12 +63,14 @@ setup:
   - match: { data_streams.0.indices.0.index_name: '.ds-simple-data-stream1-000001' }
   - match: { data_streams.0.status: 'GREEN' }
   - match: { data_streams.0.template: 'my-template1' }
+  - match: { data_streams.0.hidden: false }
   - match: { data_streams.1.name: simple-data-stream2 }
   - match: { data_streams.1.timestamp_field.name: '@timestamp' }
   - match: { data_streams.0.generation: 1 }
   - length: { data_streams.1.indices: 1 }
   - match: { data_streams.1.indices.0.index_name: '.ds-simple-data-stream2-000001' }
   - match: { data_streams.1.template: 'my-template2' }
+  - match: { data_streams.0.hidden: false }
 
   - do:
       indices.get_mapping:
@@ -101,6 +113,89 @@ setup:
         name: simple-data-stream2
   - is_true: acknowledged
 
+---
+"Create hidden data stream":
+  - skip:
+      version: " - 7.99.99"
+      reason: "re-enable in 7.11 when backported"
+  - do:
+      indices.create_data_stream:
+        name: hidden-data-stream
+  - is_true: acknowledged
+
+  - do:
+      indices.create_data_stream:
+        name: .hidden-data-stream
+  - is_true: acknowledged
+
+  - do:
+      indices.get_data_stream:
+        name: "*"
+  - length: { data_streams: 0 }
+
+  - do:
+      indices.get_data_stream:
+        name: "*"
+        expand_wildcards: hidden
+  - length: { data_streams: 2 }
+  - match: { data_streams.0.name: .hidden-data-stream }
+  - match: { data_streams.0.hidden: true }
+  - match: { data_streams.1.name: hidden-data-stream }
+  - match: { data_streams.1.hidden: true }
+
+  - do:
+      index:
+        index: hidden-data-stream
+        body:
+          '@timestamp': '2020-12-12'
+          foo: bar
+
+  - do:
+      indices.refresh:
+        index: hidden-data-stream
+
+  - do:
+      search:
+        index: hidden-data-*
+        body: { query: { match_all: {} } }
+  - length:   { hits.hits: 0  }
+
+  - do:
+      search:
+        index: hidden-data-*
+        expand_wildcards: open,hidden
+        body: { query: { match_all: { } } }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._source.foo: 'bar' }
+
+  - do:
+      index:
+        index: .hidden-data-stream
+        body:
+          '@timestamp': '2020-12-12'
+          foo: bar
+
+  - do:
+      indices.refresh:
+        index: .hidden-data-stream
+
+  - do:
+      search:
+        index: .hidden-data-*
+        body: { query: { match_all: { } } }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._source.foo: 'bar' }
+
+  - do:
+      indices.delete_data_stream:
+        name: hidden-data-stream
+  - is_true: acknowledged
+
+  - do:
+      indices.delete_data_stream:
+        name: .hidden-data-stream
+  - is_true: acknowledged
+
 ---
 "Create data stream with invalid name":
   - skip: