/*
 * 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.monitoring.resolver.node;

import org.apache.lucene.util.Constants;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import org.elasticsearch.test.ESIntegTestCase.Scope;
import org.elasticsearch.xpack.monitoring.MonitoringSettings;
import org.elasticsearch.xpack.monitoring.collector.node.NodeStatsCollector;
import org.elasticsearch.xpack.monitoring.exporter.local.LocalExporter;
import org.elasticsearch.xpack.monitoring.test.MonitoringIntegTestCase;
import org.junit.After;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.hamcrest.Matchers.greaterThan;

// numClientNodes is set to 0 because Client nodes don't have Filesystem stats
@ClusterScope(scope = Scope.TEST, numClientNodes = 0, transportClientRatio = 0.0)
public class NodeStatsTests extends MonitoringIntegTestCase {

    @Override
    protected Settings nodeSettings(int nodeOrdinal) {
        return Settings.builder()
                .put(super.nodeSettings(nodeOrdinal))
                .put(MonitoringSettings.INTERVAL.getKey(), "-1")
                .put(MonitoringSettings.COLLECTORS.getKey(), NodeStatsCollector.NAME)
                .put("xpack.monitoring.exporters.default_local.type", LocalExporter.TYPE)
                .build();
    }

    @After
    public void cleanup() throws Exception {
        updateMonitoringInterval(-1, TimeUnit.SECONDS);
        wipeMonitoringIndices();
    }

    public void testNodeStats() throws Exception {
        logger.debug("--> creating some indices for future node stats");
        final int numDocs = between(50, 150);
        for (int i = 0; i < numDocs; i++) {
            client().prepareIndex("test", "foo").setSource("value", randomInt()).get();
        }

        flush();
        refresh();

        updateMonitoringInterval(3L, TimeUnit.SECONDS);
        waitForMonitoringIndices();

        awaitMonitoringDocsCount(greaterThan(0L), NodeStatsResolver.TYPE);

        logger.debug("--> searching for monitoring documents of type [{}]", NodeStatsResolver.TYPE);
        SearchResponse response = client().prepareSearch().setTypes(NodeStatsResolver.TYPE).get();
        assertThat(response.getHits().getTotalHits(), greaterThan(0L));

        logger.debug("--> checking that every document contains the expected fields");

        for (SearchHit searchHit : response.getHits().getHits()) {
            Map<String, Object> fields = searchHit.sourceAsMap();

            for (String filter : nodeStatsFilters(watcherEnabled)) {
                if (Constants.WINDOWS) {
                    // load average is unavailable on Windows
                    if ("node_stats.os.cpu.load_average.1m".equals(filter)) {
                        continue;
                    }
                }
                if (filter.startsWith("node_stats.fs")) {
                    // NORELEASE what is going on here?
                    continue;
                }

                assertContains(filter, fields);
            }
        }

        logger.debug("--> node stats successfully collected");
    }

    /**
     * Optionally exclude {@link NodeStatsResolver#FILTERS} that require Watcher to be enabled.
     *
     * @param includeWatcher {@code true} to keep watcher filters.
     * @return Never {@code null} or empty.
     * @see #watcherEnabled
     */
    private static Set<String> nodeStatsFilters(boolean includeWatcher) {
        if (includeWatcher) {
            return NodeStatsResolver.FILTERS;
        }

        return NodeStatsResolver.FILTERS.stream().filter(s -> s.contains("watcher") == false).collect(Collectors.toSet());
    }

    @Override
    protected boolean enableWatcher() {
        // currently this is the only Monitoring test that expects Watcher to be enabled.
        // Once this becomes the default, then this should be removed.
        return randomBoolean();
    }
}

@@ -1936,81 +1936,8 @@
 n %7B%0A
-        logger.debug(%22--%3E creating some indices for future node stats%22);%0A
     
@@ -2331,109 +2331,8 @@
 );%0A%0A
-        logger.debug(%22--%3E searching for monitoring documents of type %5B%7B%7D%5D%22, NodeStatsResolver.TYPE);%0A
     
@@ -2503,96 +2503,8 @@
 );%0A%0A
-        logger.debug(%22--%3E checking that every document contains the expected fields%22);%0A%0A
     
@@ -2828,16 +2828,34 @@
      if 
+(filter.startsWith
 (%22node_s
@@ -2882,26 +2882,9 @@
 rage
-.1m%22.equals(filter
+%22
 )) %7B
@@ -2978,105 +2978,170 @@
     
-if (filter.
+// fs 
 sta
-r
 ts
-With(%22node_stats.fs%22)) %7B%0A                    // NORELEASE what is going on here?
+ and cgroup stats are not reported on every node%0A                if (filter.startsWith(%22node_stats.fs%22) %7C%7C filter.startsWith(%22node_stats.os.cgroup%22)) %7B
 %0A   
@@ -3261,72 +3261,8 @@
    %7D
-%0A%0A        logger.debug(%22--%3E node stats successfully collected%22);
 %0A   

Fix test that fails with Cgroups missing on some machines