<commit_before>/*
 * 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.marvel.agent.collector.shards;

import org.elasticsearch.Version;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.marvel.MarvelSettings;
import org.elasticsearch.marvel.MonitoredSystem;
import org.elasticsearch.marvel.agent.collector.AbstractCollectorTestCase;
import org.elasticsearch.marvel.agent.exporter.MonitoringDoc;
import org.elasticsearch.marvel.license.MarvelLicensee;

import java.util.Collection;

import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;

public class ShardsCollectorTests extends AbstractCollectorTestCase {

    public void testShardsCollectorNoIndices() throws Exception {
        Collection<MonitoringDoc> results = newShardsCollector().doCollect();
        assertThat(results, hasSize(0));
    }

    @Override
    protected Settings nodeSettings(int nodeOrdinal) {
        return Settings.builder()
                .put(super.nodeSettings(nodeOrdinal))
                .put(MarvelSettings.INDICES.getKey(), "test-shards*")
                .build();
    }

    public void testShardsCollectorOneIndex() throws Exception {
        int nbDocs = randomIntBetween(1, 20);
        for (int i = 0; i < nbDocs; i++) {
            client().prepareIndex("test-shards", "test").setSource("num", i).get();
        }

        waitForRelocation();
        securedEnsureGreen();
        securedRefresh();

        assertHitCount(client().prepareSearch().setSize(0).get(), nbDocs);

        Collection<MonitoringDoc> results = newShardsCollector().doCollect();
        assertThat(results, hasSize(getNumShards("test-shards").totalNumShards));

        final ClusterState clusterState = client().admin().cluster().prepareState().setMetaData(true).get().getState();

        int primaries = 0;
        int replicas = 0;

        for (MonitoringDoc monitoringDoc : results) {
            assertNotNull(monitoringDoc);
            assertThat(monitoringDoc, instanceOf(ShardMonitoringDoc.class));

            ShardMonitoringDoc shardMarvelDoc = (ShardMonitoringDoc) monitoringDoc;
            assertThat(shardMarvelDoc.getMonitoringId(), equalTo(MonitoredSystem.ES.getSystem()));
            assertThat(shardMarvelDoc.getMonitoringVersion(), equalTo(Version.CURRENT.toString()));
            assertThat(shardMarvelDoc.getClusterUUID(), equalTo(clusterState.metaData().clusterUUID()));
            assertThat(shardMarvelDoc.getTimestamp(), greaterThan(0L));
            assertThat(shardMarvelDoc.getSourceNode(), notNullValue());
            assertThat(shardMarvelDoc.getClusterStateUUID(), equalTo(clusterState.stateUUID()));

            ShardRouting shardRouting = shardMarvelDoc.getShardRouting();
            assertNotNull(shardRouting);
            assertThat(shardMarvelDoc.getShardRouting().assignedToNode(), is(true));

            if (shardRouting.primary()) {
                primaries++;
            } else {
                replicas++;
            }
        }

        int expectedPrimaries = getNumShards("test-shards").numPrimaries;
        int expectedReplicas = expectedPrimaries * getNumShards("test-shards").numReplicas;
        assertThat(primaries, equalTo(expectedPrimaries));
        assertThat(replicas, equalTo(expectedReplicas));
    }

    public void testShardsCollectorMultipleIndices() throws Exception {
        final String indexPrefix = "test-shards-";
        final int nbIndices = randomIntBetween(1, 3);
        final int[] nbDocsPerIndex = new int[nbIndices];

        for (int i = 0; i < nbIndices; i++) {
            String index = indexPrefix + String.valueOf(i);
            assertAcked(prepareCreate(index));

            nbDocsPerIndex[i] = randomIntBetween(1, 20);
            for (int j = 0; j < nbDocsPerIndex[i]; j++) {
                client().prepareIndex(index, "test").setSource("num", i).get();
            }
        }

        waitForRelocation();
        securedRefresh();

        int totalShards = 0;
        for (int i = 0; i < nbIndices; i++) {
            String index = indexPrefix + String.valueOf(i);

            assertHitCount(client().prepareSearch(index).setSize(0).get(), nbDocsPerIndex[i]);
            disableAllocation(index);
            totalShards += getNumShards(index).totalNumShards;
        }

        Collection<MonitoringDoc> results = newShardsCollector().doCollect();
        assertThat(results, hasSize(totalShards));

        final ClusterState clusterState = client().admin().cluster().prepareState().setMetaData(true).get().getState();

        for (MonitoringDoc doc : results) {
            assertNotNull(doc);
            assertThat(doc, instanceOf(ShardMonitoringDoc.class));

            ShardMonitoringDoc shardDoc = (ShardMonitoringDoc) doc;
            assertThat(shardDoc.getMonitoringId(), equalTo(MonitoredSystem.ES.getSystem()));
            assertThat(shardDoc.getMonitoringVersion(), equalTo(Version.CURRENT.toString()));
            assertThat(shardDoc.getClusterUUID(), equalTo(clusterState.metaData().clusterUUID()));
            assertThat(shardDoc.getTimestamp(), greaterThan(0L));
            assertThat(shardDoc.getSourceNode(), notNullValue());
            assertThat(shardDoc.getClusterStateUUID(), equalTo(clusterState.stateUUID()));

            ShardRouting shardRouting = shardDoc.getShardRouting();
            assertNotNull(shardRouting);
        }

        // Checks that a correct number of ShardMarvelDoc documents has been created for each index
        int[] shards = new int[nbIndices];
        for (MonitoringDoc monitoringDoc : results) {
            ShardRouting routing = ((ShardMonitoringDoc) monitoringDoc).getShardRouting();
            int index = Integer.parseInt(routing.getIndexName().substring(indexPrefix.length()));
            shards[index]++;
        }<commit_msg>Monitoring: Check for source_node only for assigned shard in test<commit_after>/*
 * 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.marvel.agent.collector.shards;

import org.elasticsearch.Version;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.marvel.MarvelSettings;
import org.elasticsearch.marvel.MonitoredSystem;
import org.elasticsearch.marvel.agent.collector.AbstractCollectorTestCase;
import org.elasticsearch.marvel.agent.exporter.MonitoringDoc;
import org.elasticsearch.marvel.license.MarvelLicensee;

import java.util.Collection;

import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;

public class ShardsCollectorTests extends AbstractCollectorTestCase {

    public void testShardsCollectorNoIndices() throws Exception {
        Collection<MonitoringDoc> results = newShardsCollector().doCollect();
        assertThat(results, hasSize(0));
    }

    @Override
    protected Settings nodeSettings(int nodeOrdinal) {
        return Settings.builder()
                .put(super.nodeSettings(nodeOrdinal))
                .put(MarvelSettings.INDICES.getKey(), "test-shards*")
                .build();
    }

    public void testShardsCollectorOneIndex() throws Exception {
        int nbDocs = randomIntBetween(1, 20);
        for (int i = 0; i < nbDocs; i++) {
            client().prepareIndex("test-shards", "test").setSource("num", i).get();
        }

        waitForRelocation();
        securedEnsureGreen();
        securedRefresh();

        assertHitCount(client().prepareSearch().setSize(0).get(), nbDocs);

        Collection<MonitoringDoc> results = newShardsCollector().doCollect();
        assertThat(results, hasSize(getNumShards("test-shards").totalNumShards));

        final ClusterState clusterState = client().admin().cluster().prepareState().setMetaData(true).get().getState();

        int primaries = 0;
        int replicas = 0;

        for (MonitoringDoc monitoringDoc : results) {
            assertNotNull(monitoringDoc);
            assertThat(monitoringDoc, instanceOf(ShardMonitoringDoc.class));

            ShardMonitoringDoc shardMarvelDoc = (ShardMonitoringDoc) monitoringDoc;
            assertThat(shardMarvelDoc.getMonitoringId(), equalTo(MonitoredSystem.ES.getSystem()));
            assertThat(shardMarvelDoc.getMonitoringVersion(), equalTo(Version.CURRENT.toString()));
            assertThat(shardMarvelDoc.getClusterUUID(), equalTo(clusterState.metaData().clusterUUID()));
            assertThat(shardMarvelDoc.getTimestamp(), greaterThan(0L));
            assertThat(shardMarvelDoc.getSourceNode(), notNullValue());
            assertThat(shardMarvelDoc.getClusterStateUUID(), equalTo(clusterState.stateUUID()));

            ShardRouting shardRouting = shardMarvelDoc.getShardRouting();
            assertNotNull(shardRouting);
            assertThat(shardMarvelDoc.getShardRouting().assignedToNode(), is(true));

            if (shardRouting.primary()) {
                primaries++;
            } else {
                replicas++;
            }
        }

        int expectedPrimaries = getNumShards("test-shards").numPrimaries;
        int expectedReplicas = expectedPrimaries * getNumShards("test-shards").numReplicas;
        assertThat(primaries, equalTo(expectedPrimaries));
        assertThat(replicas, equalTo(expectedReplicas));
    }

    public void testShardsCollectorMultipleIndices() throws Exception {
        final String indexPrefix = "test-shards-";
        final int nbIndices = randomIntBetween(1, 3);
        final int[] nbDocsPerIndex = new int[nbIndices];

        for (int i = 0; i < nbIndices; i++) {
            String index = indexPrefix + String.valueOf(i);
            assertAcked(prepareCreate(index));

            nbDocsPerIndex[i] = randomIntBetween(1, 20);
            for (int j = 0; j < nbDocsPerIndex[i]; j++) {
                client().prepareIndex(index, "test").setSource("num", i).get();
            }
        }

        waitForRelocation();
        securedRefresh();

        int totalShards = 0;
        for (int i = 0; i < nbIndices; i++) {
            String index = indexPrefix + String.valueOf(i);

            assertHitCount(client().prepareSearch(index).setSize(0).get(), nbDocsPerIndex[i]);
            disableAllocation(index);
            totalShards += getNumShards(index).totalNumShards;
        }

        Collection<MonitoringDoc> results = newShardsCollector().doCollect();
        assertThat(results, hasSize(totalShards));

        final ClusterState clusterState = client().admin().cluster().prepareState().setMetaData(true).get().getState();

        for (MonitoringDoc doc : results) {
            assertNotNull(doc);
            assertThat(doc, instanceOf(ShardMonitoringDoc.class));

            ShardMonitoringDoc shardDoc = (ShardMonitoringDoc) doc;
            assertThat(shardDoc.getMonitoringId(), equalTo(MonitoredSystem.ES.getSystem()));
            assertThat(shardDoc.getMonitoringVersion(), equalTo(Version.CURRENT.toString()));
            assertThat(shardDoc.getClusterUUID(), equalTo(clusterState.metaData().clusterUUID()));
            assertThat(shardDoc.getTimestamp(), greaterThan(0L));
            assertThat(shardDoc.getClusterStateUUID(), equalTo(clusterState.stateUUID()));

            ShardRouting shardRouting = shardDoc.getShardRouting();
            assertNotNull(shardRouting);

            if (shardRouting.assignedToNode()) {
                assertThat(shardDoc.getSourceNode(), notNullValue());
            } else {
                assertThat(shardDoc.getSourceNode(), nullValue());
            }
        }

        // Checks that a correct number of ShardMarvelDoc documents has been created for each index
        int[] shards = new int[nbIndices];
        for (MonitoringDoc monitoringDoc : results) {
            ShardRouting routing = ((ShardMonitoringDoc) monitoringDoc).getShardRouting();
            int index = Integer.parseInt(routing.getIndexName().substring(indexPrefix.length()));
            shards[index]++;
        }
{'fraction_non_alphanumeric': 0.11051414073576599, 'fraction_numerical': 0.0029848518767256176, 'mean_word_length': 3.3204384268214056, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3529588', 'n_tokens_mistral': 3640, 'n_tokens_neox': 3671, 'n_words': 663}