Add telemetry for data tiers (#63031)

This commit adds telemetry for our data tier formalization. This telemetry helps determine the
topology of the cluster with regard to the content, hot, warm, & cold tiers/roles.

An example of the telemetry looks like:

```
GET /_xpack/usage?human
{
  ...
  "data_tiers" : {
    "available" : true,
    "enabled" : true,
    "data_warm" : {
      ...
    },
    "data_cold" : {
      ...
    },
    "data_content" : {
      "node_count" : 1,
      "index_count" : 6,
      "total_shard_count" : 6,
      "primary_shard_count" : 6,
      "doc_count" : 71,
      "total_size" : "59.6kb",
      "total_size_bytes" : 61110,
      "primary_size" : "59.6kb",
      "primary_size_bytes" : 61110,
      "primary_shard_size_avg" : "9.9kb",
      "primary_shard_size_avg_bytes" : 10185,
      "primary_shard_size_median" : "8kb",
      "primary_shard_size_median_bytes" : 8254,
      "primary_shard_size_mad" : "7.2kb",
      "primary_shard_size_mad_bytes" : 7391
    },
    "data_hot" : {
       ...
    }
  }
}
```

The fields are as follows:

- node_count :: number of nodes with this tier/role
- index_count :: number of indices on this tier
- total_shard_count :: total number of shards for all nodes in this tier
- primary_shard_count :: number of primary shards for all nodes in this tier
- doc_count :: number of documents for all nodes in this tier
- total_size_bytes :: total number of bytes for all shards for all nodes in this tier
- primary_size_bytes :: number of bytes for all primary shards on all nodes in this tier
- primary_shard_size_avg_bytes :: average shard size for primary shard in this tier
- primary_shard_size_median_bytes :: median shard size for primary shard in this tier
- primary_shard_size_mad_bytes :: [median absolute deviation](https://en.wikipedia.org/wiki/Median_absolute_deviation) of shard size for primary shard in this tier

Relates to #60848

docs/reference/rest-api/info.asciidoc

@@ -146,6 +146,10 @@ Example response:
       "data_streams" : {
          "available" : true,
          "enabled" : true,
+      },
+      "data_tiers" : {
+         "available" : true,
+         "enabled" : true,
       }
    },
    "tagline" : "You know, for X"

docs/reference/rest-api/usage.asciidoc

@@ -285,6 +285,58 @@ GET /_xpack/usage
     "enabled" : true,
     "data_streams" : 0,
     "indices_count" : 0
+  },
+  "data_tiers" : {
+    "available" : true,
+    "enabled" : true,
+    "data_warm" : {
+      "node_count" : 0,
+      "index_count" : 0,
+      "total_shard_count" : 0,
+      "primary_shard_count" : 0,
+      "doc_count" : 0,
+      "total_size_bytes" : 0,
+      "primary_size_bytes" : 0,
+      "primary_shard_size_avg_bytes" : 0,
+      "primary_shard_size_median_bytes" : 0,
+      "primary_shard_size_mad_bytes" : 0
+    },
+    "data_cold" : {
+      "node_count" : 0,
+      "index_count" : 0,
+      "total_shard_count" : 0,
+      "primary_shard_count" : 0,
+      "doc_count" : 0,
+      "total_size_bytes" : 0,
+      "primary_size_bytes" : 0,
+      "primary_shard_size_avg_bytes" : 0,
+      "primary_shard_size_median_bytes" : 0,
+      "primary_shard_size_mad_bytes" : 0
+    },
+    "data_content" : {
+      "node_count" : 0,
+      "index_count" : 0,
+      "total_shard_count" : 0,
+      "primary_shard_count" : 0,
+      "doc_count" : 0,
+      "total_size_bytes" : 0,
+      "primary_size_bytes" : 0,
+      "primary_shard_size_avg_bytes" : 0,
+      "primary_shard_size_median_bytes" : 0,
+      "primary_shard_size_mad_bytes" : 0
+    },
+    "data_hot" : {
+      "node_count" : 0,
+      "index_count" : 0,
+      "total_shard_count" : 0,
+      "primary_shard_count" : 0,
+      "doc_count" : 0,
+      "total_size_bytes" : 0,
+      "primary_size_bytes" : 0,
+      "primary_shard_size_avg_bytes" : 0,
+      "primary_shard_size_median_bytes" : 0,
+      "primary_shard_size_mad_bytes" : 0
+    }
   }
 }
 ------------------------------------------------------------

x-pack/plugin/core/src/internalClusterTest/java/org/elasticsearch/xpack/cluster/routing/allocation/DataTierIT.java

@@ -16,13 +16,16 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.xpack.core.DataTier;
-import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
+import org.elasticsearch.xpack.core.DataTiersFeatureSetUsage;
+import org.elasticsearch.xpack.core.action.XPackUsageRequestBuilder;
+import org.elasticsearch.xpack.core.action.XPackUsageResponse;
 
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, numClientNodes = 0)
 public class DataTierIT extends ESIntegTestCase {
@@ -30,7 +33,7 @@ public class DataTierIT extends ESIntegTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return Collections.singleton(LocalStateCompositeXPackPlugin.class);
+        return Collections.singleton(DataTierTelemetryPlugin.class);
     }
 
     public void testDefaultIndexAllocateToContent() {
@@ -194,6 +197,61 @@ public class DataTierIT extends ESIntegTestCase {
         ensureYellow(index);
     }
 
+    public void testDataTierTelemetry() {
+        startContentOnlyNode();
+        startContentOnlyNode();
+        startHotOnlyNode();
+
+        client().admin().indices().prepareCreate(index)
+            .setSettings(Settings.builder()
+                .put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, "data_hot")
+                .put("index.number_of_shards", 2)
+                .put("index.number_of_replicas", 0))
+            .setWaitForActiveShards(0)
+            .get();
+
+        client().admin().indices().prepareCreate(index + "2")
+            .setSettings(Settings.builder()
+                .put("index.number_of_shards", 1)
+                .put("index.number_of_replicas", 1))
+            .setWaitForActiveShards(0)
+            .get();
+
+        ensureGreen();
+        client().prepareIndex(index).setSource("foo", "bar").get();
+        client().prepareIndex(index + "2").setSource("foo", "bar").get();
+        client().prepareIndex(index + "2").setSource("foo", "bar").get();
+        refresh(index, index + "2");
+
+        DataTiersFeatureSetUsage usage = getUsage();
+        // We can't guarantee that internal indices aren't created, so some of these are >= checks
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).nodeCount, equalTo(2));
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).indexCount, greaterThanOrEqualTo(1));
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).totalShardCount, greaterThanOrEqualTo(2));
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).primaryShardCount, greaterThanOrEqualTo(1));
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).docCount, greaterThanOrEqualTo(2L));
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).primaryByteCount, greaterThanOrEqualTo(1L));
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).primaryByteCountMedian, greaterThanOrEqualTo(1L));
+        assertThat(usage.getTierStats().get(DataTier.DATA_CONTENT).primaryShardBytesMAD, greaterThanOrEqualTo(0L));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).nodeCount, equalTo(1));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).indexCount, greaterThanOrEqualTo(1));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).totalShardCount, greaterThanOrEqualTo(2));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).primaryShardCount, greaterThanOrEqualTo(2));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).docCount, greaterThanOrEqualTo(1L));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).primaryByteCount, greaterThanOrEqualTo(1L));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).primaryByteCountMedian, greaterThanOrEqualTo(1L));
+        assertThat(usage.getTierStats().get(DataTier.DATA_HOT).primaryShardBytesMAD, greaterThanOrEqualTo(0L));
+    }
+
+    private DataTiersFeatureSetUsage getUsage() {
+        XPackUsageResponse usages = new XPackUsageRequestBuilder(client()).execute().actionGet();
+        return usages.getUsages().stream()
+            .filter(u -> u instanceof DataTiersFeatureSetUsage)
+            .findFirst()
+            .map(u -> (DataTiersFeatureSetUsage) u)
+            .orElseThrow();
+    }
+
     public void startDataNode() {
         Settings nodeSettings = Settings.builder()
             .putList("node.roles", Arrays.asList("master", "data", "ingest"))

x-pack/plugin/core/src/internalClusterTest/java/org/elasticsearch/xpack/cluster/routing/allocation/DataTierTelemetryPlugin.java

@@ -0,0 +1,79 @@
+/*
+ * 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.cluster.routing.allocation;
+
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.TransportAction;
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.LicenseService;
+import org.elasticsearch.protocol.xpack.XPackInfoRequest;
+import org.elasticsearch.protocol.xpack.XPackInfoResponse;
+import org.elasticsearch.protocol.xpack.XPackUsageRequest;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
+import org.elasticsearch.xpack.core.action.TransportXPackInfoAction;
+import org.elasticsearch.xpack.core.action.TransportXPackUsageAction;
+import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackUsageResponse;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * This plugin extends {@link LocalStateCompositeXPackPlugin} to only make the data tier telemetry
+ * available. This allows telemetry to be retrieved in integration tests where it would otherwise
+ * throw errors trying to retrieve all of the different telemetry types.
+ */
+public class DataTierTelemetryPlugin extends LocalStateCompositeXPackPlugin {
+
+    public static class DataTiersTransportXPackUsageAction extends TransportXPackUsageAction {
+        @Inject
+        public DataTiersTransportXPackUsageAction(ThreadPool threadPool, TransportService transportService,
+                                                  ClusterService clusterService, ActionFilters actionFilters,
+                                                  IndexNameExpressionResolver indexNameExpressionResolver, NodeClient client) {
+            super(threadPool, transportService, clusterService, actionFilters, indexNameExpressionResolver, client);
+        }
+        @Override
+        protected List<XPackUsageFeatureAction> usageActions() {
+            return Collections.singletonList(XPackUsageFeatureAction.DATA_TIERS);
+        }
+    }
+
+    public static class DataTiersTransportXPackInfoAction extends TransportXPackInfoAction {
+        @Inject
+        public DataTiersTransportXPackInfoAction(TransportService transportService, ActionFilters actionFilters,
+                                                 LicenseService licenseService, NodeClient client) {
+            super(transportService, actionFilters, licenseService, client);
+        }
+
+        @Override
+        protected List<XPackInfoFeatureAction> infoActions() {
+            return Collections.singletonList(XPackInfoFeatureAction.DATA_TIERS);
+        }
+    }
+
+    public DataTierTelemetryPlugin(final Settings settings, final Path configPath) {
+        super(settings, configPath);
+    }
+
+    @Override
+    protected Class<? extends TransportAction<XPackUsageRequest, XPackUsageResponse>> getUsageAction() {
+        return DataTiersTransportXPackUsageAction.class;
+    }
+
+    @Override
+    protected Class<? extends TransportAction<XPackInfoRequest, XPackInfoResponse>> getInfoAction() {
+        return DataTiersTransportXPackInfoAction.class;
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTier.java

@@ -16,6 +16,8 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.shard.IndexSettingProvider;
 import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
 
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.Set;
 
 /**
@@ -34,6 +36,8 @@ public class DataTier {
     public static final String DATA_WARM = "data_warm";
     public static final String DATA_COLD = "data_cold";
 
+    public static final Set<String> ALL_DATA_TIERS = new HashSet<>(Arrays.asList(DATA_CONTENT, DATA_HOT, DATA_WARM, DATA_COLD));
+
     /**
      * Returns true if the given tier name is a valid tier
      */

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTiersFeatureSetUsage.java

@@ -0,0 +1,195 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * {@link DataTiersFeatureSetUsage} represents the xpack usage for data tiers.
+ * This includes things like the number of nodes per tier, indices, shards, etc.
+ * See {@link TierSpecificStats} for the stats that are tracked on a per-tier
+ * basis.
+ */
+public class DataTiersFeatureSetUsage extends XPackFeatureSet.Usage {
+    private final Map<String, TierSpecificStats> tierStats;
+
+    public DataTiersFeatureSetUsage(StreamInput in) throws IOException {
+        super(in);
+        this.tierStats = in.readMap(StreamInput::readString, TierSpecificStats::new);
+    }
+
+    public DataTiersFeatureSetUsage(Map<String, TierSpecificStats> tierStats) {
+        super(XPackField.DATA_TIERS, true, true);
+        this.tierStats = tierStats;
+    }
+
+    @Override
+    public Version getMinimalSupportedVersion() {
+        return Version.V_7_10_0;
+    }
+
+    public Map<String, TierSpecificStats> getTierStats() {
+        return Collections.unmodifiableMap(tierStats);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeMap(tierStats, StreamOutput::writeString, (o, v) -> v.writeTo(o));
+    }
+
+    @Override
+    protected void innerXContent(XContentBuilder builder, Params params) throws IOException {
+        super.innerXContent(builder, params);
+        for (Map.Entry<String, TierSpecificStats> tierStats : tierStats.entrySet()) {
+            builder.field(tierStats.getKey(), tierStats.getValue());
+        }
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tierStats);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
+        if (getClass() != obj.getClass()) {
+            return false;
+        }
+        DataTiersFeatureSetUsage other = (DataTiersFeatureSetUsage) obj;
+        return Objects.equals(available, other.available) &&
+            Objects.equals(enabled, other.enabled) &&
+            Objects.equals(tierStats, other.tierStats);
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+
+    /**
+     * {@link TierSpecificStats} represents statistics about nodes in a single
+     * tier, for example, how many nodes there are, the index count, shard
+     * count, etc.
+     */
+    public static class TierSpecificStats implements Writeable, ToXContentObject {
+
+        public final int nodeCount;
+        public final int indexCount;
+        public final int totalShardCount;
+        public final int primaryShardCount;
+        public final long docCount;
+        public final long totalByteCount;
+        public final long primaryByteCount;
+        public final long primaryByteCountMedian;
+        public final long primaryShardBytesMAD;
+
+        public TierSpecificStats(StreamInput in) throws IOException {
+            this.nodeCount = in.readVInt();
+            this.indexCount = in.readVInt();
+            this.totalShardCount = in.readVInt();
+            this.primaryShardCount = in.readVInt();
+            this.docCount = in.readVLong();
+            this.totalByteCount = in.readVLong();
+            this.primaryByteCount = in.readVLong();
+            this.primaryByteCountMedian = in.readVLong();
+            this.primaryShardBytesMAD = in.readVLong();
+        }
+
+        public TierSpecificStats(int nodeCount, int indexCount, int totalShardCount, int primaryShardCount, long docCount,
+                                 long totalByteCount, long primaryByteCount, long primaryByteCountMedian, long primaryShardBytesMAD) {
+            this.nodeCount = nodeCount;
+            this.indexCount = indexCount;
+            this.totalShardCount = totalShardCount;
+            this.primaryShardCount = primaryShardCount;
+            this.docCount = docCount;
+            this.totalByteCount = totalByteCount;
+            this.primaryByteCount = primaryByteCount;
+            this.primaryByteCountMedian = primaryByteCountMedian;
+            this.primaryShardBytesMAD = primaryShardBytesMAD;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeVInt(this.nodeCount);
+            out.writeVInt(this.indexCount);
+            out.writeVInt(this.totalShardCount);
+            out.writeVInt(this.primaryShardCount);
+            out.writeVLong(this.docCount);
+            out.writeVLong(this.totalByteCount);
+            out.writeVLong(this.primaryByteCount);
+            out.writeVLong(this.primaryByteCountMedian);
+            out.writeVLong(this.primaryShardBytesMAD);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field("node_count", nodeCount);
+            builder.field("index_count", indexCount);
+            builder.field("total_shard_count", totalShardCount);
+            builder.field("primary_shard_count", primaryShardCount);
+            builder.field("doc_count", docCount);
+            builder.humanReadableField("total_size_bytes", "total_size", new ByteSizeValue(totalByteCount));
+            builder.humanReadableField("primary_size_bytes", "primary_size", new ByteSizeValue(primaryByteCount));
+            builder.humanReadableField("primary_shard_size_avg_bytes", "primary_shard_size_avg",
+                new ByteSizeValue(primaryShardCount == 0 ? 0 : (primaryByteCount / primaryShardCount)));
+            builder.humanReadableField("primary_shard_size_median_bytes", "primary_shard_size_median",
+                new ByteSizeValue(primaryByteCountMedian));
+            builder.humanReadableField("primary_shard_size_mad_bytes", "primary_shard_size_mad",
+                new ByteSizeValue(primaryShardBytesMAD));
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(this.nodeCount, this.indexCount, this.totalShardCount, this.primaryShardCount, this.totalByteCount,
+                this.primaryByteCount, this.docCount, this.primaryByteCountMedian, this.primaryShardBytesMAD);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null) {
+                return false;
+            }
+            if (getClass() != obj.getClass()) {
+                return false;
+            }
+            TierSpecificStats other = (TierSpecificStats) obj;
+            return nodeCount == other.nodeCount &&
+                indexCount == other.indexCount &&
+                totalShardCount == other.totalShardCount &&
+                primaryShardCount == other.primaryShardCount &&
+                docCount == other.docCount &&
+                totalByteCount == other.totalByteCount &&
+                primaryByteCount == other.primaryByteCount &&
+                primaryByteCountMedian == other.primaryByteCountMedian &&
+                primaryShardBytesMAD == other.primaryShardBytesMAD;
+        }
+
+        @Override
+        public String toString() {
+            return Strings.toString(this);
+        }
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTiersInfoTransportAction.java

@@ -0,0 +1,37 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackInfoFeatureTransportAction;
+
+public class DataTiersInfoTransportAction extends XPackInfoFeatureTransportAction {
+
+    @Inject
+    public DataTiersInfoTransportAction(TransportService transportService, ActionFilters actionFilters) {
+        super(XPackInfoFeatureAction.DATA_TIERS.name(), transportService, actionFilters);
+    }
+
+    @Override
+    public String name() {
+        return XPackField.DATA_TIERS;
+    }
+
+    @Override
+    public boolean available() {
+        return true;
+    }
+
+    @Override
+    public boolean enabled() {
+        return true;
+    }
+
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTiersUsageTransportAction.java

@@ -0,0 +1,144 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
+import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
+import org.elasticsearch.action.admin.indices.stats.CommonStatsFlags;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.cluster.routing.RoutingNode;
+import org.elasticsearch.cluster.routing.RoutingNodes;
+import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.ShardRoutingState;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.store.StoreStats;
+import org.elasticsearch.protocol.xpack.XPackUsageRequest;
+import org.elasticsearch.search.aggregations.metrics.TDigestState;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureResponse;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureTransportAction;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class DataTiersUsageTransportAction extends XPackUsageFeatureTransportAction {
+
+    private final Client client;
+
+    @Inject
+    public DataTiersUsageTransportAction(TransportService transportService, ClusterService clusterService,
+                                         ThreadPool threadPool, ActionFilters actionFilters,
+                                         IndexNameExpressionResolver indexNameExpressionResolver, Client client) {
+        super(XPackUsageFeatureAction.DATA_TIERS.name(), transportService, clusterService,
+            threadPool, actionFilters, indexNameExpressionResolver);
+        this.client = client;
+    }
+
+    @Override
+    protected void masterOperation(Task task, XPackUsageRequest request, ClusterState state,
+                                   ActionListener<XPackUsageFeatureResponse> listener) {
+        client.admin().cluster().prepareNodesStats()
+            .all()
+            .setIndices(CommonStatsFlags.ALL)
+            .execute(ActionListener.wrap(nodesStatsResponse -> {
+                final RoutingNodes routingNodes = state.getRoutingNodes();
+
+                // First separate the nodes into separate tiers, note that nodes *may* be duplicated
+                Map<String, List<NodeStats>> tierSpecificNodeStats = separateTiers(nodesStatsResponse);
+
+                // Generate tier specific stats for the nodes
+                Map<String, DataTiersFeatureSetUsage.TierSpecificStats> tierSpecificStats = tierSpecificNodeStats.entrySet()
+                    .stream().collect(Collectors.toMap(Map.Entry::getKey, ns -> calculateStats(ns.getValue(), routingNodes)));
+
+                listener.onResponse(new XPackUsageFeatureResponse(new DataTiersFeatureSetUsage(tierSpecificStats)));
+            }, listener::onFailure));
+    }
+
+    // Visible for testing
+    static Map<String, List<NodeStats>> separateTiers(NodesStatsResponse nodesStatsResponse) {
+        Map<String, List<NodeStats>> responses = new HashMap<>();
+        DataTier.ALL_DATA_TIERS.forEach(tier ->
+            responses.put(tier, nodesStatsResponse.getNodes().stream()
+                .filter(stats -> stats.getNode().getRoles().stream()
+                    .map(DiscoveryNodeRole::roleName)
+                    .anyMatch(rn -> rn.equals(tier)))
+                .collect(Collectors.toList())));
+        return responses;
+    }
+
+    private DataTiersFeatureSetUsage.TierSpecificStats calculateStats(List<NodeStats> nodesStats, RoutingNodes routingNodes) {
+        int nodeCount = 0;
+        int indexCount = 0;
+        int totalShardCount = 0;
+        long totalByteCount = 0;
+        long docCount = 0;
+        final AtomicInteger primaryShardCount = new AtomicInteger(0);
+        final AtomicLong primaryByteCount = new AtomicLong(0);
+        final TDigestState valueSketch = new TDigestState(1000);
+        for (NodeStats nodeStats : nodesStats) {
+            nodeCount++;
+            totalByteCount += nodeStats.getIndices().getStore().getSizeInBytes();
+            docCount += nodeStats.getIndices().getDocs().getCount();
+            String nodeId = nodeStats.getNode().getId();
+            final RoutingNode node = routingNodes.node(nodeId);
+            if (node != null) {
+                totalShardCount += node.shardsWithState(ShardRoutingState.STARTED).size();
+                Set<Index> indicesOnNode = node.shardsWithState(ShardRoutingState.STARTED).stream()
+                    .map(ShardRouting::index)
+                    .collect(Collectors.toSet());
+                indexCount += indicesOnNode.size();
+                indicesOnNode.forEach(index -> {
+                    nodeStats.getIndices().getShardStats(index).stream()
+                        .filter(shardStats -> shardStats.getPrimary().getStore() != null)
+                        .forEach(shardStats -> {
+                            StoreStats primaryStoreStats = shardStats.getPrimary().getStore();
+                            // If storeStats is null, it means this is not a replica
+                            primaryShardCount.incrementAndGet();
+                            long primarySize = primaryStoreStats.getSizeInBytes();
+                            primaryByteCount.addAndGet(primarySize);
+                            valueSketch.add(primarySize);
+                        });
+                });
+            }
+        }
+        long primaryShardSizeMedian = (long) valueSketch.quantile(0.5);
+        long primaryShardSizeMAD = computeMedianAbsoluteDeviation(valueSketch);
+        return new DataTiersFeatureSetUsage.TierSpecificStats(nodeCount, indexCount, totalShardCount, primaryShardCount.get(), docCount,
+            totalByteCount, primaryByteCount.get(), primaryShardSizeMedian, primaryShardSizeMAD);
+    }
+
+    // Visible for testing
+    static long computeMedianAbsoluteDeviation(TDigestState valuesSketch) {
+        if (valuesSketch.size() == 0) {
+            return 0;
+        } else {
+            final double approximateMedian = valuesSketch.quantile(0.5);
+            final TDigestState approximatedDeviationsSketch = new TDigestState(valuesSketch.compression());
+            valuesSketch.centroids().forEach(centroid -> {
+                final double deviation = Math.abs(approximateMedian - centroid.mean());
+                approximatedDeviationsSketch.add(deviation, centroid.count());
+            });
+
+            return (long) approximatedDeviationsSketch.quantile(0.5);
+        }
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java

@@ -507,7 +507,9 @@ public class XPackClientPlugin extends Plugin implements ActionPlugin, NetworkPl
             new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.SEARCHABLE_SNAPSHOTS,
                 SearchableSnapshotFeatureSetUsage::new),
             // Data Streams
-            new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.DATA_STREAMS, DataStreamFeatureSetUsage::new)
+            new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.DATA_STREAMS, DataStreamFeatureSetUsage::new),
+            // Data Tiers
+            new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.DATA_TIERS, DataTiersFeatureSetUsage::new)
         );
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackField.java

@@ -63,6 +63,8 @@ public final class XPackField {
     public static final String SEARCHABLE_SNAPSHOTS = "searchable_snapshots";
     /** Name constant for the data streams feature. */
     public static final String DATA_STREAMS = "data_streams";
+    /** Name constant for the data tiers feature. */
+    public static final String DATA_TIERS = "data_tiers";
 
     private XPackField() {}
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackPlugin.java

@@ -63,7 +63,9 @@ import org.elasticsearch.xpack.core.action.TransportReloadAnalyzersAction;
 import org.elasticsearch.xpack.core.action.TransportXPackInfoAction;
 import org.elasticsearch.xpack.core.action.TransportXPackUsageAction;
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
+import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
 import org.elasticsearch.xpack.core.action.XPackUsageAction;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
 import org.elasticsearch.xpack.core.action.XPackUsageResponse;
 import org.elasticsearch.xpack.core.async.DeleteAsyncResultAction;
 import org.elasticsearch.xpack.core.async.TransportDeleteAsyncResultAction;
@@ -281,6 +283,8 @@ public class XPackPlugin extends XPackClientPlugin implements ExtensiblePlugin,
         actions.add(new ActionHandler<>(DeleteAsyncResultAction.INSTANCE, TransportDeleteAsyncResultAction.class));
         actions.add(new ActionHandler<>(OpenPointInTimeAction.INSTANCE, TransportOpenPointInTimeAction.class));
         actions.add(new ActionHandler<>(ClosePointInTimeAction.INSTANCE, TransportClosePointInTimeAction.class));
+        actions.add(new ActionHandler<>(XPackInfoFeatureAction.DATA_TIERS, DataTiersInfoTransportAction.class));
+        actions.add(new ActionHandler<>(XPackUsageFeatureAction.DATA_TIERS, DataTiersUsageTransportAction.class));
         return actions;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/XPackInfoFeatureAction.java

@@ -45,13 +45,14 @@ public class XPackInfoFeatureAction extends ActionType<XPackInfoFeatureResponse>
     public static final XPackInfoFeatureAction ENRICH = new XPackInfoFeatureAction(XPackField.ENRICH);
     public static final XPackInfoFeatureAction SEARCHABLE_SNAPSHOTS = new XPackInfoFeatureAction(XPackField.SEARCHABLE_SNAPSHOTS);
     public static final XPackInfoFeatureAction DATA_STREAMS = new XPackInfoFeatureAction(XPackField.DATA_STREAMS);
+    public static final XPackInfoFeatureAction DATA_TIERS = new XPackInfoFeatureAction(XPackField.DATA_TIERS);
 
     public static final List<XPackInfoFeatureAction> ALL;
     static {
         final List<XPackInfoFeatureAction> actions = new ArrayList<>();
         actions.addAll(Arrays.asList(
             SECURITY, MONITORING, WATCHER, GRAPH, MACHINE_LEARNING, LOGSTASH, EQL, SQL, ROLLUP, INDEX_LIFECYCLE, SNAPSHOT_LIFECYCLE, CCR,
-            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, ENRICH, DATA_STREAMS, SEARCHABLE_SNAPSHOTS
+            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, ENRICH, DATA_STREAMS, SEARCHABLE_SNAPSHOTS, DATA_TIERS
         ));
         ALL = Collections.unmodifiableList(actions);
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/XPackUsageFeatureAction.java

@@ -45,13 +45,14 @@ public class XPackUsageFeatureAction extends ActionType<XPackUsageFeatureRespons
     public static final XPackUsageFeatureAction ENRICH = new XPackUsageFeatureAction(XPackField.ENRICH);
     public static final XPackUsageFeatureAction SEARCHABLE_SNAPSHOTS = new XPackUsageFeatureAction(XPackField.SEARCHABLE_SNAPSHOTS);
     public static final XPackUsageFeatureAction DATA_STREAMS = new XPackUsageFeatureAction(XPackField.DATA_STREAMS);
+    public static final XPackUsageFeatureAction DATA_TIERS = new XPackUsageFeatureAction(XPackField.DATA_TIERS);
 
     public static final List<XPackUsageFeatureAction> ALL;
     static {
         final List<XPackUsageFeatureAction> actions = new ArrayList<>();
         actions.addAll(Arrays.asList(
             SECURITY, MONITORING, WATCHER, GRAPH, MACHINE_LEARNING, LOGSTASH, EQL, SQL, ROLLUP, INDEX_LIFECYCLE, SNAPSHOT_LIFECYCLE, CCR,
-            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, DATA_STREAMS, SEARCHABLE_SNAPSHOTS
+            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, DATA_STREAMS, SEARCHABLE_SNAPSHOTS, DATA_TIERS
         ));
         ALL = Collections.unmodifiableList(actions);
     }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/DataTiersFeatureSetUsageTests.java

@@ -0,0 +1,48 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class DataTiersFeatureSetUsageTests extends AbstractWireSerializingTestCase<DataTiersFeatureSetUsage> {
+    @Override
+    protected Writeable.Reader<DataTiersFeatureSetUsage> instanceReader() {
+        return DataTiersFeatureSetUsage::new;
+    }
+
+    @Override
+    protected DataTiersFeatureSetUsage mutateInstance(DataTiersFeatureSetUsage instance) throws IOException {
+        return randomValueOtherThan(instance, DataTiersFeatureSetUsageTests::randomUsage);
+    }
+
+    @Override
+    protected DataTiersFeatureSetUsage createTestInstance() {
+        return randomUsage();
+    }
+
+    public static DataTiersFeatureSetUsage randomUsage() {
+        List<String> tiers = randomSubsetOf(DataTier.ALL_DATA_TIERS);
+        Map<String, DataTiersFeatureSetUsage.TierSpecificStats> stats = new HashMap<>();
+        tiers.forEach(tier ->
+            stats.put(tier, new DataTiersFeatureSetUsage.TierSpecificStats(randomIntBetween(1, 10),
+                randomIntBetween(5, 100),
+                randomIntBetween(0, 1000),
+                randomIntBetween(0, 1000),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong())));
+        return new DataTiersFeatureSetUsage(stats);
+    }
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/DataTiersUsageTransportActionTests.java

@@ -0,0 +1,67 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
+import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.search.aggregations.metrics.TDigestState;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class DataTiersUsageTransportActionTests extends ESTestCase {
+    public void testCalculateMAD() {
+        assertThat(DataTiersUsageTransportAction.computeMedianAbsoluteDeviation(new TDigestState(10)), equalTo(0L));
+
+        TDigestState sketch = new TDigestState(randomDoubleBetween(0, 1000, false));
+        sketch.add(1);
+        sketch.add(1);
+        sketch.add(2);
+        sketch.add(2);
+        sketch.add(4);
+        sketch.add(6);
+        sketch.add(9);
+        assertThat(DataTiersUsageTransportAction.computeMedianAbsoluteDeviation(sketch), equalTo(1L));
+    }
+
+    public void testSeparateTiers() {
+        NodeStats hotStats = fakeStats(DataTier.DATA_HOT_NODE_ROLE);
+        NodeStats coldStats = fakeStats(DataTier.DATA_COLD_NODE_ROLE);
+        NodeStats warmStats = fakeStats(DataTier.DATA_WARM_NODE_ROLE);
+        NodeStats warmStats2 = fakeStats(DataTier.DATA_WARM_NODE_ROLE);
+
+        NodesStatsResponse nodesStats = new NodesStatsResponse(new ClusterName("cluster"),
+            Arrays.asList(hotStats, coldStats, warmStats, warmStats2), Collections.emptyList());
+
+        Map<String, List<NodeStats>> tiers = DataTiersUsageTransportAction.separateTiers(nodesStats);
+        assertThat(tiers.keySet(), equalTo(DataTier.ALL_DATA_TIERS));
+        assertThat(tiers.get(DataTier.DATA_CONTENT), empty());
+        assertThat(tiers.get(DataTier.DATA_HOT), containsInAnyOrder(hotStats));
+        assertThat(tiers.get(DataTier.DATA_WARM), containsInAnyOrder(warmStats, warmStats2));
+        assertThat(tiers.get(DataTier.DATA_COLD), containsInAnyOrder(coldStats));
+    }
+
+    private static NodeStats fakeStats(DiscoveryNodeRole role) {
+        NodeStats stats = mock(NodeStats.class);
+        DiscoveryNode node = mock(DiscoveryNode.class);
+        when(node.getRoles()).thenReturn(Collections.singleton(role));
+        when(stats.getNode()).thenReturn(node);
+        return stats;
+    }
+}