Add analytics plugin usage stats to _xpack/usage

docs/reference/rest-api/usage.asciidoc

@@ -16,7 +16,7 @@ Provides usage information about the installed {xpack} features.
 === {api-description-title}
 
 This API provides information about which features are currently enabled and
-available under the current license and some usage statistics. 
+available under the current license and some usage statistics.
 
 [discrete]
 [[usage-api-query-parms]]
@@ -263,7 +263,14 @@ GET /_xpack/usage
   },
   "analytics" : {
     "available" : true,
-    "enabled" : true
+    "enabled" : true,
+    "stats": {
+      "boxplot_usage" : 0,
+      "top_metrics_usage" : 0,
+      "cumulative_cardinality_usage" : 0,
+      "t_test_usage" : 0,
+      "string_stats_usage" : 0
+    }
   }
 }
 ------------------------------------------------------------

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/AnalyticsPlugin.java

@@ -75,7 +75,7 @@ public List<PipelineAggregationSpec> getPipelineAggregations() {
                 CumulativeCardinalityPipelineAggregationBuilder.NAME,
                 CumulativeCardinalityPipelineAggregationBuilder::new,
                 CumulativeCardinalityPipelineAggregator::new,
-                usage.track(AnalyticsUsage.Item.CUMULATIVE_CARDINALITY,
+                usage.track(AnalyticsStatsAction.Item.CUMULATIVE_CARDINALITY,
                         checkLicense(CumulativeCardinalityPipelineAggregationBuilder.PARSER)))
         );
     }
@@ -86,24 +86,24 @@ public List<AggregationSpec> getAggregations() {
             new AggregationSpec(
                 StringStatsAggregationBuilder.NAME,
                 StringStatsAggregationBuilder::new,
-                usage.track(AnalyticsUsage.Item.STRING_STATS, checkLicense(StringStatsAggregationBuilder.PARSER)))
+                usage.track(AnalyticsStatsAction.Item.STRING_STATS, checkLicense(StringStatsAggregationBuilder.PARSER)))
                 .addResultReader(InternalStringStats::new)
                 .setAggregatorRegistrar(StringStatsAggregationBuilder::registerAggregators),
             new AggregationSpec(
                 BoxplotAggregationBuilder.NAME,
                 BoxplotAggregationBuilder::new,
-                usage.track(AnalyticsUsage.Item.BOXPLOT, checkLicense(BoxplotAggregationBuilder.PARSER)))
+                usage.track(AnalyticsStatsAction.Item.BOXPLOT, checkLicense(BoxplotAggregationBuilder.PARSER)))
                 .addResultReader(InternalBoxplot::new)
                 .setAggregatorRegistrar(BoxplotAggregationBuilder::registerAggregators),
             new AggregationSpec(
                 TopMetricsAggregationBuilder.NAME,
                 TopMetricsAggregationBuilder::new,
-                usage.track(AnalyticsUsage.Item.TOP_METRICS, checkLicense(TopMetricsAggregationBuilder.PARSER)))
+                usage.track(AnalyticsStatsAction.Item.TOP_METRICS, checkLicense(TopMetricsAggregationBuilder.PARSER)))
                 .addResultReader(InternalTopMetrics::new),
             new AggregationSpec(
                 TTestAggregationBuilder.NAME,
                 TTestAggregationBuilder::new,
-                usage.track(AnalyticsUsage.Item.T_TEST, checkLicense(TTestAggregationBuilder.PARSER)))
+                usage.track(AnalyticsStatsAction.Item.T_TEST, checkLicense(TTestAggregationBuilder.PARSER)))
                 .addResultReader(InternalTTest::new)
         );
     }
@@ -137,7 +137,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
             ResourceWatcherService resourceWatcherService, ScriptService scriptService, NamedXContentRegistry xContentRegistry,
             Environment environment, NodeEnvironment nodeEnvironment, NamedWriteableRegistry namedWriteableRegistry,
             IndexNameExpressionResolver indexNameExpressionResolver) {
-        return singletonList(new AnalyticsUsage());
+        return singletonList(usage);
     }
 
     @Override

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/AnalyticsUsage.java

@@ -8,54 +8,32 @@
 
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.xcontent.ContextParser;
+import org.elasticsearch.xpack.core.analytics.EnumCounters;
 import org.elasticsearch.xpack.core.analytics.action.AnalyticsStatsAction;
 
-import java.util.EnumMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
 /**
  * Tracks usage of the Analytics aggregations.
  */
 public class AnalyticsUsage {
-    /**
-     * Items to track.
-     */
-    public enum Item {
-        BOXPLOT,
-        CUMULATIVE_CARDINALITY,
-        STRING_STATS,
-        TOP_METRICS,
-        T_TEST;
-    }
 
-    private final Map<Item, AtomicLong> trackers = new EnumMap<>(Item.class);
+    private final EnumCounters<AnalyticsStatsAction.Item> counters = new EnumCounters<>(AnalyticsStatsAction.Item.class);
 
     public AnalyticsUsage() {
-        for (Item item: Item.values()) {
-            trackers.put(item, new AtomicLong(0));
-        }
     }
 
     /**
      * Track successful parsing.
      */
-    public <C, T> ContextParser<C, T> track(Item item, ContextParser<C, T> realParser) {
-        AtomicLong usage = trackers.get(item);
+    public <C, T> ContextParser<C, T> track(AnalyticsStatsAction.Item item, ContextParser<C, T> realParser) {
         return (parser, context) -> {
             T value = realParser.parse(parser, context);
             // Intentionally doesn't count unless the parser returns cleanly.
-            usage.incrementAndGet();
+            counters.inc(item);
             return value;
         };
     }
 
     public AnalyticsStatsAction.NodeResponse stats(DiscoveryNode node) {
-        return new AnalyticsStatsAction.NodeResponse(node,
-                trackers.get(Item.BOXPLOT).get(),
-                trackers.get(Item.CUMULATIVE_CARDINALITY).get(),
-                trackers.get(Item.STRING_STATS).get(),
-                trackers.get(Item.TOP_METRICS).get(),
-                trackers.get(Item.T_TEST).get());
+        return new AnalyticsStatsAction.NodeResponse(node, counters);
     }
 }

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/action/AnalyticsUsageTransportAction.java

@@ -7,6 +7,7 @@
 
 import org.elasticsearch.action.ActionListener;
 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.service.ClusterService;
@@ -20,26 +21,49 @@
 import org.elasticsearch.xpack.core.action.XPackUsageFeatureResponse;
 import org.elasticsearch.xpack.core.action.XPackUsageFeatureTransportAction;
 import org.elasticsearch.xpack.core.analytics.AnalyticsFeatureSetUsage;
+import org.elasticsearch.xpack.core.analytics.EnumCounters;
+import org.elasticsearch.xpack.core.analytics.action.AnalyticsStatsAction;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
 
 public class AnalyticsUsageTransportAction extends XPackUsageFeatureTransportAction {
     private final XPackLicenseState licenseState;
+    private final Client client;
 
     @Inject
     public AnalyticsUsageTransportAction(TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
                                          ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                                         XPackLicenseState licenseState) {
+                                         XPackLicenseState licenseState, Client client) {
         super(XPackUsageFeatureAction.ANALYTICS.name(), transportService, clusterService,
             threadPool, actionFilters, indexNameExpressionResolver);
         this.licenseState = licenseState;
+        this.client = client;
     }
 
     @Override
     protected void masterOperation(Task task, XPackUsageRequest request, ClusterState state,
                                    ActionListener<XPackUsageFeatureResponse> listener) {
         boolean available = licenseState.isDataScienceAllowed();
+        if (available) {
+            AnalyticsStatsAction.Request statsRequest = new AnalyticsStatsAction.Request();
+            statsRequest.setParentTask(clusterService.localNode().getId(), task.getId());
+            client.execute(AnalyticsStatsAction.INSTANCE, statsRequest, ActionListener.wrap(r ->
+                    listener.onResponse(new XPackUsageFeatureResponse(usageFeatureResponse(true, true, r))),
+                listener::onFailure));
+        } else {
+            AnalyticsFeatureSetUsage usage = new AnalyticsFeatureSetUsage(false, true, Collections.emptyMap());
+            listener.onResponse(new XPackUsageFeatureResponse(usage));
+        }
+    }
 
-        AnalyticsFeatureSetUsage usage =
-            new AnalyticsFeatureSetUsage(available, true);
-        listener.onResponse(new XPackUsageFeatureResponse(usage));
+    static AnalyticsFeatureSetUsage usageFeatureResponse(boolean available, boolean enabled, AnalyticsStatsAction.Response r) {
+        List<EnumCounters<AnalyticsStatsAction.Item>> countersPerNode = r.getNodes()
+            .stream()
+            .map(AnalyticsStatsAction.NodeResponse::getStats)
+            .collect(Collectors.toList());
+        EnumCounters<AnalyticsStatsAction.Item> mergedCounters = EnumCounters.merge(AnalyticsStatsAction.Item.class, countersPerNode);
+        return new AnalyticsFeatureSetUsage(available, enabled, mergedCounters.toMap());
     }
 }

x-pack/plugin/analytics/src/test/java/org/elasticsearch/xpack/analytics/action/AnalyticsInfoTransportActionTests.java

@@ -5,28 +5,54 @@
  */
 package org.elasticsearch.xpack.analytics.action;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.XPackFeatureSet;
 import org.elasticsearch.xpack.core.action.XPackUsageFeatureResponse;
 import org.elasticsearch.xpack.core.analytics.AnalyticsFeatureSetUsage;
+import org.elasticsearch.xpack.core.analytics.action.AnalyticsStatsAction;
 import org.junit.Before;
+import org.mockito.stubbing.Answer;
+
+import java.util.Collections;
 
 import static org.hamcrest.core.Is.is;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
 public class AnalyticsInfoTransportActionTests extends ESTestCase {
 
     private XPackLicenseState licenseState;
+    private Task task;
+    private ClusterService clusterService;
+    private ClusterName clusterName;
 
     @Before
     public void init() {
         licenseState = mock(XPackLicenseState.class);
+        task = mock(Task.class);
+        when(task.getId()).thenReturn(randomLong());
+        clusterService = mock(ClusterService.class);
+        DiscoveryNode discoveryNode = mock(DiscoveryNode.class);
+        when(discoveryNode.getId()).thenReturn(randomAlphaOfLength(10));
+        when(clusterService.localNode()).thenReturn(discoveryNode);
+        clusterName = mock(ClusterName.class);
     }
 
     public void testAvailable() throws Exception {
@@ -35,37 +61,59 @@ public void testAvailable() throws Exception {
         boolean available = randomBoolean();
         when(licenseState.isDataScienceAllowed()).thenReturn(available);
         assertThat(featureSet.available(), is(available));
-
-        AnalyticsUsageTransportAction usageAction = new AnalyticsUsageTransportAction(mock(TransportService.class), null, null,
-            mock(ActionFilters.class), null, licenseState);
+        Client client = mockClient();
+        AnalyticsUsageTransportAction usageAction = new AnalyticsUsageTransportAction(mock(TransportService.class), clusterService, null,
+            mock(ActionFilters.class), null, licenseState, client);
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(null, null, null, future);
+        usageAction.masterOperation(task, null, null, future);
         XPackFeatureSet.Usage usage = future.get().getUsage();
         assertThat(usage.available(), is(available));
 
         BytesStreamOutput out = new BytesStreamOutput();
         usage.writeTo(out);
         XPackFeatureSet.Usage serializedUsage = new AnalyticsFeatureSetUsage(out.bytes().streamInput());
         assertThat(serializedUsage.available(), is(available));
+        if (available) {
+            verify(client, times(1)).execute(any(), any(), any());
+        }
+        verifyNoMoreInteractions(client);
     }
 
     public void testEnabled() throws Exception {
         AnalyticsInfoTransportAction featureSet = new AnalyticsInfoTransportAction(
             mock(TransportService.class), mock(ActionFilters.class), licenseState);
         assertThat(featureSet.enabled(), is(true));
         assertTrue(featureSet.enabled());
-
+        boolean available = randomBoolean();
+        when(licenseState.isDataScienceAllowed()).thenReturn(available);
+        Client client = mockClient();
         AnalyticsUsageTransportAction usageAction = new AnalyticsUsageTransportAction(mock(TransportService.class),
-            null, null, mock(ActionFilters.class), null, licenseState);
+            clusterService, null, mock(ActionFilters.class), null, licenseState, client);
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(null, null, null, future);
+        usageAction.masterOperation(task, null, null, future);
         XPackFeatureSet.Usage usage = future.get().getUsage();
         assertTrue(usage.enabled());
 
         BytesStreamOutput out = new BytesStreamOutput();
         usage.writeTo(out);
         XPackFeatureSet.Usage serializedUsage = new AnalyticsFeatureSetUsage(out.bytes().streamInput());
         assertTrue(serializedUsage.enabled());
+        if (available) {
+            verify(client, times(1)).execute(any(), any(), any());
+        }
+        verifyNoMoreInteractions(client);
+    }
+
+    private Client mockClient() {
+        Client client = mock(Client.class);
+        doAnswer((Answer<Void>) invocation -> {
+            @SuppressWarnings("unchecked")
+            ActionListener<AnalyticsStatsAction.Response> listener =
+                (ActionListener<AnalyticsStatsAction.Response>) invocation.getArguments()[2];
+            listener.onResponse(new AnalyticsStatsAction.Response(clusterName, Collections.emptyList(), Collections.emptyList()));
+            return null;
+        }).when(client).execute(eq(AnalyticsStatsAction.INSTANCE), any(), any());
+        return client;
     }
 
 }

x-pack/plugin/analytics/src/test/java/org/elasticsearch/xpack/analytics/action/AnalyticsStatsActionNodeResponseTests.java

@@ -0,0 +1,49 @@
+/*
+ * 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.analytics.action;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.xpack.core.analytics.EnumCounters;
+import org.elasticsearch.xpack.core.analytics.action.AnalyticsStatsAction;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class AnalyticsStatsActionNodeResponseTests extends AbstractWireSerializingTestCase<AnalyticsStatsAction.NodeResponse> {
+
+    @Override
+    protected Writeable.Reader<AnalyticsStatsAction.NodeResponse> instanceReader() {
+        return AnalyticsStatsAction.NodeResponse::new;
+    }
+
+    @Override
+    protected AnalyticsStatsAction.NodeResponse createTestInstance() {
+        String nodeName = randomAlphaOfLength(10);
+        DiscoveryNode node = new DiscoveryNode(nodeName, buildNewFakeTransportAddress(), Version.CURRENT);
+        EnumCounters<AnalyticsStatsAction.Item> counters = new EnumCounters<>(AnalyticsStatsAction.Item.class);
+        for (AnalyticsStatsAction.Item item : AnalyticsStatsAction.Item.values()) {
+            if (randomBoolean()) {
+                counters.inc(item, randomLongBetween(0, 1000));
+            }
+        }
+        return new AnalyticsStatsAction.NodeResponse(node, counters);
+    }
+
+    public void testItemEnum() {
+        int i = 0;
+        // We rely on the ordinals for serialization, so they shouldn't change between version
+        assertThat(AnalyticsStatsAction.Item.BOXPLOT.ordinal(), equalTo(i++));
+        assertThat(AnalyticsStatsAction.Item.CUMULATIVE_CARDINALITY.ordinal(), equalTo(i++));
+        assertThat(AnalyticsStatsAction.Item.STRING_STATS.ordinal(), equalTo(i++));
+        assertThat(AnalyticsStatsAction.Item.TOP_METRICS.ordinal(), equalTo(i++));
+        assertThat(AnalyticsStatsAction.Item.T_TEST.ordinal(), equalTo(i++));
+        // Please add tests for newly added items here
+        assertThat(AnalyticsStatsAction.Item.values().length, equalTo(i));
+    }
+}