/*
 * 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.shield.audit.index;

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import org.elasticsearch.test.ESIntegTestCase.Scope;
import org.elasticsearch.test.InternalTestCluster;
import org.elasticsearch.test.ShieldIntegTestCase;
import org.elasticsearch.test.ShieldSettingsSource;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.test.InternalTestCluster.clusterName;
import static org.hamcrest.Matchers.is;

/**
 * This test checks to ensure that the IndexAuditTrail starts properly when indexing to a remote cluster
 */
@ClusterScope(scope = Scope.TEST)
public class RemoteIndexAuditTrailStartingTests extends ShieldIntegTestCase {

    public static final String SECOND_CLUSTER_NODE_PREFIX = "remote_" + SUITE_CLUSTER_NODE_PREFIX;

    private InternalTestCluster remoteCluster;

    private final boolean useSSL = randomBoolean();
    private final boolean localAudit = randomBoolean();
    private final String outputs = randomFrom("index", "logfile", "index,logfile");

    @Override
    public boolean sslTransportEnabled() {
        return useSSL;
    }

    @Override
    public Settings nodeSettings(int nodeOrdinal) {
        return Settings.builder()
                .put(super.nodeSettings(nodeOrdinal))
                .put("shield.audit.enabled", localAudit)
                .put("shield.audit.outputs", outputs)
                .build();
    }

    @Override
    protected Set<String> excludeTemplates() {
        return Collections.singleton(IndexAuditTrail.INDEX_TEMPLATE_NAME);
    }

    @Override
    public void beforeIndexDeletion() {
        // For this test, this is a NO-OP because the index audit trail will continue to capture events and index after
        // the tests have completed. The default implementation of this method expects that nothing is performing operations
        // after the test has completed
    }

    @Before
    public void startRemoteCluster() throws IOException {
        final List<String> addresses = new ArrayList<>();
        // get addresses for current cluster
        NodesInfoResponse response = client().admin().cluster().prepareNodesInfo().execute().actionGet();
        final String clusterName = response.getClusterNameAsString();
        for (NodeInfo nodeInfo : response.getNodes()) {
            InetSocketTransportAddress address = (InetSocketTransportAddress) nodeInfo.getTransport().address().publishAddress();
            addresses.add(address.address().getHostString() + ":" + address.address().getPort());
        }

        // create another cluster
        String cluster2Name = clusterName(Scope.SUITE.name(), randomLong());

        // Setup a second test cluster with randomization for number of nodes, shield enabled, and SSL
        final int numNodes = randomIntBetween(2, 3);
        ShieldSettingsSource cluster2SettingsSource = new ShieldSettingsSource(numNodes, useSSL, systemKey(), createTempDir(), Scope.SUITE) {
            @Override
            public Settings nodeSettings(int nodeOrdinal) {
                Settings.Builder builder = Settings.builder()
                        .put(super.nodeSettings(nodeOrdinal))
                        .put("shield.audit.enabled", true)
                        .put("shield.audit.outputs", randomFrom("index", "index,logfile"))
                        .putArray("shield.audit.index.client.hosts", addresses.toArray(new String[addresses.size()]))
                        .put("shield.audit.index.client.cluster.name", clusterName)
                        .put("shield.audit.index.client.shield.user", ShieldSettingsSource.DEFAULT_USER_NAME + ":" + ShieldSettingsSource.DEFAULT_PASSWORD);

                if (useSSL) {
                    for (Map.Entry<String, String> entry : getClientSSLSettings().getAsMap().entrySet()) {
                        builder.put("shield.audit.index.client." + entry.getKey(), entry.getValue());
                    }
                }
                return builder.build();
            }
        };
        remoteCluster = new InternalTestCluster("network", randomLong(), createTempDir(), numNodes, numNodes, cluster2Name, cluster2SettingsSource, 0, false, SECOND_CLUSTER_NODE_PREFIX, true);
        remoteCluster.beforeTest(getRandom(), 0.5);
    }

    @After
    public void stopRemoteCluster() throws Exception {
        if (remoteCluster != null) {
            try {
                remoteCluster.wipe(Collections.<String>emptySet());
            } finally {
                remoteCluster.afterTest();
            }
            remoteCluster.close();
        }

        // stop the index audit trail so that the shards aren't locked causing the test to fail
        if (outputs.contains("index")) {
            Iterable<IndexAuditTrail> auditTrails = internalCluster().getInstances(IndexAuditTrail.class);
            for (IndexAuditTrail auditTrail : auditTrails) {
                auditTrail.close();
            }
        }
    }

    @Test
    public void testThatRemoteAuditInstancesAreStarted() throws Exception {
        Iterable<IndexAuditTrail> auditTrails = remoteCluster.getInstances(IndexAuditTrail.class);
        for (final IndexAuditTrail auditTrail : auditTrails) {
            awaitBusy(() -> auditTrail.state() == IndexAuditTrail.State.STARTED, 2L, TimeUnit.SECONDS);
            assertThat(auditTrail.state(), is(IndexAuditTrail.State.STARTED));
        }
    }
}

@@ -1190,17 +1190,211 @@
  cluster
+. The cluster%0A * started by the integration tests is indexed into by the remote cluster started before the test.%0A *%0A * The cluster started by the integrations tests may also index into itself...
 %0A
-
  */%0A@Clu
@@ -1574,21 +1574,20 @@
 ote_%22 + 
-SUI
 TE
+ST
 _CLUSTER
@@ -3442,21 +3442,20 @@
 e(Scope.
-SUI
 TE
+ST
 .name(),
@@ -3764,13 +3764,12 @@
 ope.
-SUI
 TE
+ST
 ) %7B%0A
@@ -5138,32 +5138,32 @@
 ows Exception %7B%0A
-
         if (remo
@@ -5179,24 +5179,243 @@
  != null) %7B%0A
+            Iterable%3CIndexAuditTrail%3E auditTrails = internalCluster().getInstances(IndexAuditTrail.class);%0A            for (IndexAuditTrail auditTrail : auditTrails) %7B%0A                auditTrail.close();%0A            %7D%0A%0A
             

make the remote cluster a test cluster in the RemoteIndexAuditTrailStartingTests