/*
 * 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.exporter.http;

import org.elasticsearch.Version;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.env.Environment;
import org.elasticsearch.license.XPackLicenseState;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.ESIntegTestCase.Scope;
import org.elasticsearch.test.http.MockRequest;
import org.elasticsearch.test.http.MockResponse;
import org.elasticsearch.test.http.MockWebServer;
import org.elasticsearch.xpack.monitoring.MonitoringService;
import org.elasticsearch.xpack.monitoring.MonitoringTestUtils;
import org.elasticsearch.xpack.monitoring.collector.indices.IndexRecoveryMonitoringDoc;
import org.elasticsearch.xpack.monitoring.exporter.ClusterAlertsUtil;
import org.elasticsearch.xpack.monitoring.exporter.Exporter;
import org.elasticsearch.xpack.monitoring.exporter.MonitoringDoc;
import org.elasticsearch.xpack.monitoring.exporter.MonitoringTemplateUtils;
import org.elasticsearch.xpack.monitoring.test.MonitoringIntegTestCase;
import org.elasticsearch.xpack.ssl.SSLService;
import org.joda.time.format.DateTimeFormat;
import org.junit.After;
import org.junit.Before;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.xpack.monitoring.exporter.MonitoringTemplateUtils.LAST_UPDATED_VERSION;
import static org.elasticsearch.xpack.monitoring.exporter.MonitoringTemplateUtils.TEMPLATE_VERSION;
import static org.elasticsearch.xpack.monitoring.exporter.MonitoringTemplateUtils.indexName;
import static org.elasticsearch.xpack.monitoring.exporter.http.ClusterAlertHttpResource.CLUSTER_ALERT_VERSION_PARAMETERS;
import static org.elasticsearch.xpack.monitoring.exporter.http.PublishableHttpResource.FILTER_PATH_RESOURCE_VERSION;
import static org.elasticsearch.xpack.monitoring.exporter.http.WatcherExistsHttpResource.WATCHER_CHECK_PARAMETERS;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.notNullValue;

@ESIntegTestCase.ClusterScope(scope = Scope.TEST,
                              numDataNodes = 1, numClientNodes = 0, transportClientRatio = 0.0, supportsDedicatedMasters = false)
public class HttpExporterIT extends MonitoringIntegTestCase {

    private final boolean templatesExistsAlready = randomBoolean();
    private final boolean includeOldTemplates = randomBoolean();
    private final boolean pipelineExistsAlready = randomBoolean();
    private final boolean remoteClusterAllowsWatcher = randomBoolean();
    private final boolean currentLicenseAllowsWatcher = true;
    private final boolean watcherAlreadyExists = randomBoolean();
    private final Environment environment = new Environment(Settings.builder().put("path.home", createTempDir()).build());

    private MockWebServer webServer;

    @Before
    public void startWebServer() throws IOException {
        webServer = createMockWebServer();
    }

    @After
    public void stopWebServer() throws Exception {
        if (webServer != null) {
            webServer.close();
        }
    }

    @Override
    protected boolean ignoreExternalCluster() {
        return true;
    }

    @Override
    protected Settings nodeSettings(int nodeOrdinal) {
        // we create and disable the exporter to avoid the cluster actually using it (thus speeding up tests)
        // we make an exporter on demand per test
        return Settings.builder()
                       .put(super.nodeSettings(nodeOrdinal))
                       .put(MonitoringService.INTERVAL.getKey(), "-1")
                       .put("xpack.monitoring.exporters._http.type", "http")
                       .put("xpack.monitoring.exporters._http.enabled", false)
                       .build();
    }

    public void testExport() throws Exception {
        final Settings settings = Settings.builder()
                .put("xpack.monitoring.exporters._http.type", "http")
                .put("xpack.monitoring.exporters._http.host", getFormattedAddress(webServer))
                .put("xpack.monitoring.exporters._http.index.template.create_legacy_templates", includeOldTemplates)
                .build();

        enqueueGetClusterVersionResponse(Version.CURRENT);
        enqueueSetupResponses(webServer,
                              templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                              remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        enqueueResponse(200, "{\"errors\": false, \"msg\": \"successful bulk request\"}");

        final int nbDocs = randomIntBetween(1, 25);
        export(settings, newRandomMonitoringDocs(nbDocs));

        assertMonitorResources(webServer,
                               templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                               remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        assertBulk(webServer, nbDocs);
    }

    public void testExportWithHeaders() throws Exception {
        final String headerValue = randomAlphaOfLengthBetween(3, 9);
        final String[] array = generateRandomStringArray(2, 4, false, false);

        final Map<String, String[]> headers = new HashMap<>();

        headers.put("X-Cloud-Cluster", new String[] { headerValue });
        headers.put("X-Found-Cluster", new String[] { headerValue });
        headers.put("Array-Check", array);

        Settings settings = Settings.builder()
                .put("xpack.monitoring.exporters._http.type", "http")
                .put("xpack.monitoring.exporters._http.host", getFormattedAddress(webServer))
                .put("xpack.monitoring.exporters._http.index.template.create_legacy_templates", includeOldTemplates)
                .put("xpack.monitoring.exporters._http.headers.X-Cloud-Cluster", headerValue)
                .put("xpack.monitoring.exporters._http.headers.X-Found-Cluster", headerValue)
                .putList("xpack.monitoring.exporters._http.headers.Array-Check", array)
                .build();

        enqueueGetClusterVersionResponse(Version.CURRENT);
        enqueueSetupResponses(webServer,
                              templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                              remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        enqueueResponse(200, "{\"errors\": false, \"msg\": \"successful bulk request\"}");

        final int nbDocs = randomIntBetween(1, 25);
        export(settings, newRandomMonitoringDocs(nbDocs));

        assertMonitorResources(webServer,
                               templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                               remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists,
                               headers, null);
        assertBulk(webServer, nbDocs, headers, null);
    }

    public void testExportWithBasePath() throws Exception {
        final boolean useHeaders = randomBoolean();

        final String headerValue = randomAlphaOfLengthBetween(3, 9);
        final String[] array = generateRandomStringArray(2, 4, false, false);

        final Map<String, String[]> headers = new HashMap<>();

        if (useHeaders) {
            headers.put("X-Cloud-Cluster", new String[] { headerValue });
            headers.put("X-Found-Cluster", new String[] { headerValue });
            headers.put("Array-Check", array);
        }

        String basePath = "path/to";

        if (randomBoolean()) {
            basePath += "/something";

            if (rarely()) {
                basePath += "/proxied";
            }
        }

        if (randomBoolean()) {
            basePath = "/" + basePath;
        }

        final Settings.Builder builder = Settings.builder()
                .put("xpack.monitoring.exporters._http.type", "http")
                .put("xpack.monitoring.exporters._http.host", getFormattedAddress(webServer))
                .put("xpack.monitoring.exporters._http.proxy.base_path", basePath + (randomBoolean() ? "/" : ""));

        if (includeOldTemplates == false) {
            builder.put("xpack.monitoring.exporters._http.index.template.create_legacy_templates", false);
        }

        if (useHeaders) {
            builder.put("xpack.monitoring.exporters._http.headers.X-Cloud-Cluster", headerValue)
                    .put("xpack.monitoring.exporters._http.headers.X-Found-Cluster", headerValue)
                    .putList("xpack.monitoring.exporters._http.headers.Array-Check", array);
        }

        enqueueGetClusterVersionResponse(Version.CURRENT);
        enqueueSetupResponses(webServer,
                              templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                              remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        enqueueResponse(200, "{\"errors\": false}");

        final int nbDocs = randomIntBetween(1, 25);
        export(builder.build(), newRandomMonitoringDocs(nbDocs));

        assertMonitorResources(webServer,
                               templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                               remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists,
                               headers, basePath);
        assertBulk(webServer, nbDocs, headers, basePath);
    }

    public void testHostChangeReChecksTemplate() throws Exception {
        Settings settings = Settings.builder()
                .put("xpack.monitoring.exporters._http.type", "http")
                .put("xpack.monitoring.exporters._http.host", getFormattedAddress(webServer))
                .put("xpack.monitoring.exporters._http.index.template.create_legacy_templates", includeOldTemplates)
                .build();

        enqueueGetClusterVersionResponse(Version.CURRENT);
        enqueueSetupResponses(webServer,
                              templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                              remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        enqueueResponse(200, "{\"errors\": false}");

        export(settings, Collections.singletonList(newRandomMonitoringDoc()));

        assertMonitorResources(webServer,
                               templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                               remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        assertBulk(webServer);

        try (MockWebServer secondWebServer = createMockWebServer()) {
            String missingTemplate = null;

            final Settings newSettings = Settings.builder()
                    .put(settings)
                    .putList("xpack.monitoring.exporters._http.host", getFormattedAddress(secondWebServer))
                    .build();

            enqueueGetClusterVersionResponse(secondWebServer, Version.CURRENT);
            // pretend that one of the templates is missing
            for (Tuple<String, String> template : monitoringTemplates(includeOldTemplates)) {
                if (missingTemplate != null) {
                    enqueueResponse(secondWebServer, 200, "{\"" + template.v1() + "\":{\"version\":" + LAST_UPDATED_VERSION + "}}");
                } else {
                    missingTemplate = template.v1();

                    enqueueResponse(secondWebServer, 404, "template [" + template.v1() + "] does not exist");
                    enqueueResponse(secondWebServer, 201, "template [" + template.v1() + "] created");
                }
            }
            // opposite of if it existed before
            enqueuePipelineResponses(secondWebServer, !pipelineExistsAlready);
            enqueueWatcherResponses(secondWebServer, remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
            enqueueResponse(secondWebServer, 200, "{\"errors\": false}");

            // second event
            export(newSettings, Collections.singletonList(newRandomMonitoringDoc()));

            assertMonitorVersion(secondWebServer);

            for (Tuple<String, String> template : monitoringTemplates(includeOldTemplates)) {
                MockRequest recordedRequest = secondWebServer.takeRequest();
                assertThat(recordedRequest.getMethod(), equalTo("GET"));
                assertThat(recordedRequest.getUri().getPath(), equalTo("/_template/" + template.v1()));
                assertThat(recordedRequest.getUri().getQuery(), equalTo(resourceVersionQueryString()));

                if (missingTemplate.equals(template.v1())) {
                    recordedRequest = secondWebServer.takeRequest();
                    assertThat(recordedRequest.getMethod(), equalTo("PUT"));
                    assertThat(recordedRequest.getUri().getPath(), equalTo("/_template/" + template.v1()));
                    assertThat(recordedRequest.getUri().getQuery(), equalTo(resourceVersionQueryString()));
                    assertThat(recordedRequest.getBody(), equalTo(template.v2()));
                }
            }
            assertMonitorPipelines(secondWebServer, !pipelineExistsAlready, null, null);
            assertMonitorWatches(secondWebServer, remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists,
                                 null, null);
            assertBulk(secondWebServer);
        }
    }

    public void testUnsupportedClusterVersion() throws Exception {
        Settings settings = Settings.builder()
                .put("xpack.monitoring.exporters._http.type", "http")
                .put("xpack.monitoring.exporters._http.host", getFormattedAddress(webServer))
                .build();

        // returning an unsupported cluster version
        enqueueGetClusterVersionResponse(
                randomFrom(Version.fromString("0.18.0"),
                           Version.fromString("1.0.0"),
                           Version.fromString("1.4.0"),
                           Version.fromString("2.4.0"),
                           Version.fromString("5.0.0"),
                           Version.fromString("5.4.0")));

        // ensure that the exporter is not able to be used
        try (HttpExporter exporter = createHttpExporter(settings)) {
            assertThat(exporter.isExporterReady(), is(false));
        }

        assertThat(webServer.requests(), hasSize(1));

        assertMonitorVersion(webServer);
    }

    public void testDynamicIndexFormatChange() throws Exception {
        final Settings settings = Settings.builder()
                .put("xpack.monitoring.exporters._http.type", "http")
                .put("xpack.monitoring.exporters._http.host", getFormattedAddress(webServer))
                .put("xpack.monitoring.exporters._http.index.template.create_legacy_templates", includeOldTemplates)
                .build();

        enqueueGetClusterVersionResponse(Version.CURRENT);
        enqueueSetupResponses(webServer,
                              templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                              remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        enqueueResponse(200, "{\"errors\": false, \"msg\": \"successful bulk request\"}");

        MonitoringDoc doc = newRandomMonitoringDoc();
        export(settings, Collections.singletonList(doc));

        assertMonitorResources(webServer,
                               templatesExistsAlready, includeOldTemplates, pipelineExistsAlready,
                               remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
        MockRequest recordedRequest = assertBulk(webServer);

        String indexName = indexName(DateTimeFormat.forPattern("YYYY.MM.dd").withZoneUTC(), doc.getSystem(), doc.getTimestamp());

        byte[] bytes = recordedRequest.getBody().getBytes(StandardCharsets.UTF_8);
        Map<String, Object> data = XContentHelper.convertToMap(new BytesArray(bytes), false, XContentType.JSON).v2();
        @SuppressWarnings("unchecked")
        Map<String, Object> index = (Map<String, Object>) data.get("index");
        assertThat(index.get("_index"), equalTo(indexName));

        String newTimeFormat = randomFrom("YY", "YYYY", "YYYY.MM", "YYYY-MM", "MM.YYYY", "MM");

        final Settings newSettings = Settings.builder()
                .put(settings)
                .put("xpack.monitoring.exporters._http.index.name.time_format", newTimeFormat)
                .build();

        enqueueGetClusterVersionResponse(Version.CURRENT);
        enqueueSetupResponses(webServer, true, includeOldTemplates, true,
                              true, true, true);
        enqueueResponse(200, "{\"errors\": false, \"msg\": \"successful bulk request\"}");

        doc = newRandomMonitoringDoc();
        export(newSettings, Collections.singletonList(doc));

        String expectedMonitoringIndex = ".monitoring-es-" + TEMPLATE_VERSION + "-"
                + DateTimeFormat.forPattern(newTimeFormat).withZoneUTC().print(doc.getTimestamp());

        assertMonitorResources(webServer, true, includeOldTemplates, true,
                               true, true, true);
        recordedRequest = assertBulk(webServer);

        bytes = recordedRequest.getBody().getBytes(StandardCharsets.UTF_8);
        data = XContentHelper.convertToMap(new BytesArray(bytes), false, XContentType.JSON).v2();
        @SuppressWarnings("unchecked")
        final Map<String, Object> newIndex = (Map<String, Object>) data.get("index");
        assertThat(newIndex.get("_index"), equalTo(expectedMonitoringIndex));
    }

    private void assertMonitorVersion(final MockWebServer webServer) throws Exception {
        assertMonitorVersion(webServer, null, null);
    }

    private void assertMonitorVersion(final MockWebServer webServer, @Nullable final Map<String, String[]> customHeaders,
            @Nullable final String basePath) throws Exception {
        final MockRequest request = webServer.takeRequest();

        assertThat(request.getMethod(), equalTo("GET"));
        final String pathPrefix = basePathToAssertablePrefix(basePath);
        if (Strings.isEmpty(pathPrefix) == false) {
            assertThat(request.getUri().getPath(), equalTo(pathPrefix + "/"));
        }
        assertThat(request.getUri().getQuery(), equalTo("filter_path=version.number"));
        assertHeaders(request, customHeaders);
    }

    private void assertMonitorResources(final MockWebServer webServer,
                                        final boolean templateAlreadyExists, final boolean includeOldTemplates,
                                        final boolean pipelineAlreadyExists,
                                        final boolean remoteClusterAllowsWatcher, final boolean currentLicenseAllowsWatcher,
                                        final boolean watcherAlreadyExists) throws Exception {
        assertMonitorResources(webServer, templateAlreadyExists, includeOldTemplates, pipelineAlreadyExists,
                               remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists,
                               null, null);
    }

    private void assertMonitorResources(final MockWebServer webServer,
                                        final boolean templateAlreadyExists, final boolean includeOldTemplates,
                                        final boolean pipelineAlreadyExists,
                                        final boolean remoteClusterAllowsWatcher, final boolean currentLicenseAllowsWatcher,
                                        final boolean watcherAlreadyExists,
                                        @Nullable final Map<String, String[]> customHeaders,
                                        @Nullable final String basePath) throws Exception {
        assertMonitorVersion(webServer, customHeaders, basePath);
        assertMonitorTemplates(webServer, templateAlreadyExists, includeOldTemplates, customHeaders, basePath);
        assertMonitorPipelines(webServer, pipelineAlreadyExists, customHeaders, basePath);
        assertMonitorWatches(webServer, remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists,
                             customHeaders, basePath);
    }

    private void assertMonitorTemplates(final MockWebServer webServer,
                                        final boolean alreadyExists,
                                        final boolean includeOldTemplates,
                                        @Nullable final Map<String, String[]> customHeaders,
                                        @Nullable final String basePath) throws Exception {
        final List<Tuple<String, String>> templates = monitoringTemplates(includeOldTemplates);

        assertMonitorVersionResource(webServer, alreadyExists, "/_template/", templates, customHeaders, basePath);
    }

    private void assertMonitorPipelines(final MockWebServer webServer,
                                        final boolean alreadyExists,
                                        @Nullable final Map<String, String[]> customHeaders,
                                        @Nullable final String basePath) throws Exception {
        assertMonitorVersionResource(webServer, alreadyExists, "/_ingest/pipeline/", monitoringPipelines(),
                                     customHeaders, basePath);
    }

    private void assertMonitorVersionResource(final MockWebServer webServer, final boolean alreadyExists,
                                              final String resourcePrefix, final List<Tuple<String, String>> resources,
                                              @Nullable final Map<String, String[]> customHeaders,
                                              @Nullable final String basePath) throws Exception {
        final String pathPrefix = basePathToAssertablePrefix(basePath);

        for (Tuple<String, String> resource : resources) {
            final MockRequest getRequest = webServer.takeRequest();

            assertThat(getRequest.getMethod(), equalTo("GET"));
            assertThat(getRequest.getUri().getPath(), equalTo(pathPrefix + resourcePrefix + resource.v1()));
            assertThat(getRequest.getUri().getQuery(), equalTo(resourceVersionQueryString()));
            assertHeaders(getRequest, customHeaders);

            if (alreadyExists == false) {
                final MockRequest putRequest = webServer.takeRequest();

                assertThat(putRequest.getMethod(), equalTo("PUT"));
                assertThat(putRequest.getUri().getPath(), equalTo(pathPrefix + resourcePrefix + resource.v1()));
                assertThat(putRequest.getUri().getQuery(), equalTo(resourceVersionQueryString()));
                assertThat(putRequest.getBody(), equalTo(resource.v2()));
                assertHeaders(putRequest, customHeaders);
            }
        }
    }

    private void assertMonitorWatches(final MockWebServer webServer,
                                      final boolean remoteClusterAllowsWatcher, final boolean currentLicenseAllowsWatcher,
                                      final boolean alreadyExists,
                                      @Nullable final Map<String, String[]> customHeaders,
                                      @Nullable final String basePath) throws Exception {
        final String pathPrefix = basePathToAssertablePrefix(basePath);
        MockRequest request;

        request = webServer.takeRequest();

        // GET /_xpack
        assertThat(request.getMethod(), equalTo("GET"));
        assertThat(request.getUri().getPath(), equalTo(pathPrefix + "/_xpack"));
        assertThat(request.getUri().getQuery(), equalTo(watcherCheckQueryString()));
        assertHeaders(request, customHeaders);

        if (remoteClusterAllowsWatcher) {
            for (Tuple<String, String> watch : monitoringWatches()) {
                request = webServer.takeRequest();

                // GET / PUT if we are allowed to use it
                if (currentLicenseAllowsWatcher) {
                    assertThat(request.getMethod(), equalTo("GET"));
                    assertThat(request.getUri().getPath(), equalTo(pathPrefix + "/_xpack/watcher/watch/" + watch.v1()));
                    assertThat(request.getUri().getQuery(), equalTo(resourceClusterAlertQueryString()));
                    assertHeaders(request, customHeaders);

                    if (alreadyExists == false) {
                        request = webServer.takeRequest();

                        assertThat(request.getMethod(), equalTo("PUT"));
                        assertThat(request.getUri().getPath(), equalTo(pathPrefix + "/_xpack/watcher/watch/" + watch.v1()));
                        assertThat(request.getUri().getQuery(), equalTo(resourceClusterAlertQueryString()));
                        assertThat(request.getBody(), equalTo(watch.v2()));
                        assertHeaders(request, customHeaders);
                    }
                // DELETE if we're not allowed to use it
                } else {
                    assertThat(request.getMethod(), equalTo("DELETE"));
                    assertThat(request.getUri().getPath(), equalTo(pathPrefix + "/_xpack/watcher/watch/" + watch.v1()));
                    assertThat(request.getUri().getQuery(), equalTo(resourceClusterAlertQueryString()));
                    assertHeaders(request, customHeaders);
                }
            }
        }
    }

    private MockRequest assertBulk(final MockWebServer webServer) throws Exception {
        return assertBulk(webServer, -1);
    }

    private MockRequest assertBulk(final MockWebServer webServer, final int docs) throws Exception {
        return assertBulk(webServer, docs, null, null);
    }

    private MockRequest assertBulk(final MockWebServer webServer, final int docs,
                                   @Nullable final Map<String, String[]> customHeaders, @Nullable final String basePath)
            throws Exception {
        final String pathPrefix = basePathToAssertablePrefix(basePath);
        final MockRequest request = webServer.takeRequest();

        assertThat(request.getMethod(), equalTo("POST"));
        assertThat(request.getUri().getPath(), equalTo(pathPrefix + "/_bulk"));
        assertThat(request.getUri().getQuery(), equalTo(bulkQueryString()));
        assertHeaders(request, customHeaders);

        if (docs != -1) {
            assertBulkRequest(request.getBody(), docs);
        }

        return request;
    }

    private void assertHeaders(final MockRequest request, final Map<String, String[]> customHeaders) {
        if (customHeaders != null) {
            for (final Map.Entry<String, String[]> entry : customHeaders.entrySet()) {
                final String header = entry.getKey();
                final String[] values = entry.getValue();

                final List<String> headerValues = request.getHeaders().get(header);

                if (values.length > 0) {
                    assertThat(headerValues, hasSize(values.length));
                    assertThat(headerValues, containsInAnyOrder(values));
                }
            }
        }
    }

    private HttpExporter createHttpExporter(final Settings settings) throws Exception {
        final Exporter.Config config =
                new Exporter.Config("_http", "http",
                                    settings, settings.getAsSettings("xpack.monitoring.exporters._http"),
                                    clusterService(), new XPackLicenseState());

        return new HttpExporter(config, new SSLService(settings, environment), new ThreadContext(settings));
    }

    private void export(final Settings settings, final Collection<MonitoringDoc> docs) throws Exception {
        // wait until the cluster is ready (this is done at the "Exporters" level)
        assertBusy(() -> assertThat(clusterService().state().version(), not(ClusterState.UNKNOWN_VERSION)));

        try (HttpExporter exporter = createHttpExporter(settings)) {
            // the readiness check happens synchronously, so we don't need to busy-wait for it
            assertThat("Exporter is not ready", exporter.isExporterReady(), is(true));

            final HttpExportBulk bulk = exporter.openBulk();

            assertThat("Bulk should never be null after the exporter is ready", bulk, notNullValue());

            final CountDownLatch awaitResponseAndClose = new CountDownLatch(2);
            final ActionListener<Void> listener = new ActionListener<Void>() {
                @Override
                public void onResponse(Void response) {
                    awaitResponseAndClose.countDown();
                }

                @Override
                public void onFailure(Exception e) {
                    fail(e.getMessage());

                    awaitResponseAndClose.countDown();
                }
            };

            bulk.doAdd(docs);
            bulk.doFlush(listener);
            bulk.doClose(listener);

            // block until the bulk responds
            awaitResponseAndClose.await(15, TimeUnit.SECONDS);
        }
    }

    private MonitoringDoc newRandomMonitoringDoc() {
        String clusterUUID = internalCluster().getClusterName();
        long timestamp = System.currentTimeMillis();
        long intervalMillis = randomNonNegativeLong();
        MonitoringDoc.Node sourceNode = MonitoringTestUtils.randomMonitoringNode(random());

        return new IndexRecoveryMonitoringDoc(clusterUUID, timestamp, intervalMillis, sourceNode, new RecoveryResponse());
    }

    private List<MonitoringDoc> newRandomMonitoringDocs(int nb) {
        List<MonitoringDoc> docs = new ArrayList<>(nb);
        for (int i = 0; i < nb; i++) {
            docs.add(newRandomMonitoringDoc());
        }
        return docs;
    }

    private String basePathToAssertablePrefix(@Nullable String basePath) {
        if (basePath == null) {
            return "";
        }
        basePath = basePath.startsWith("/")? basePath : "/" + basePath;
        return basePath;
    }

    private String resourceClusterAlertQueryString() {
        return "filter_path=" + CLUSTER_ALERT_VERSION_PARAMETERS.get("filter_path");
    }

    private String resourceVersionQueryString() {
        return "filter_path=" + FILTER_PATH_RESOURCE_VERSION;
    }

    private String watcherCheckQueryString() {
        return "filter_path=" + WATCHER_CHECK_PARAMETERS.get("filter_path");
    }

    private String bulkQueryString() {
        final String pipelineName = MonitoringTemplateUtils.pipelineName(TEMPLATE_VERSION);

        return "pipeline=" + pipelineName + "&filter_path=" + "errors,items.*.error";
    }

    private void enqueueGetClusterVersionResponse(Version v) throws IOException {
        enqueueGetClusterVersionResponse(webServer, v);
    }

    private void enqueueGetClusterVersionResponse(MockWebServer mockWebServer, Version v) throws IOException {
        mockWebServer.enqueue(new MockResponse().setResponseCode(200).setBody(
                jsonBuilder().startObject().startObject("version")
                    .field("number", v.toString()).endObject().endObject().bytes().utf8ToString()));
    }

    private void enqueueSetupResponses(final MockWebServer webServer,
                                       final boolean templatesAlreadyExists, final boolean includeOldTemplates,
                                       final boolean pipelineAlreadyExists,
                                       final boolean remoteClusterAllowsWatcher, final boolean currentLicenseAllowsWatcher,
                                       final boolean watcherAlreadyExists) throws IOException {
        enqueueTemplateResponses(webServer, templatesAlreadyExists, includeOldTemplates);
        enqueuePipelineResponses(webServer, pipelineAlreadyExists);
        enqueueWatcherResponses(webServer, remoteClusterAllowsWatcher, currentLicenseAllowsWatcher, watcherAlreadyExists);
    }

    private void enqueueTemplateResponses(final MockWebServer webServer,
                                          final boolean alreadyExists, final boolean includeOldTemplates)
            throws IOException {
        if (alreadyExists) {
            enqueueTemplateResponsesExistsAlready(webServer, includeOldTemplates);
        } else {
            enqueueTemplateResponsesDoesNotExistYet(webServer, includeOldTemplates);
        }
    }

    private void enqueueTemplateResponsesDoesNotExistYet(final MockWebServer webServer,
                                                         final boolean includeOldTemplates)
            throws IOException {
        enqueueVersionedResourceResponsesDoesNotExistYet(monitoringTemplateNames(includeOldTemplates), webServer);
    }

    private void enqueueTemplateResponsesExistsAlready(final MockWebServer webServer,
                                                       final boolean includeOldTemplates)
            throws IOException {
        enqueueVersionedResourceResponsesExistsAlready(monitoringTemplateNames(includeOldTemplates), webServer);
    }

    private void enqueuePipelineResponses(final MockWebServer webServer, final boolean alreadyExists) throws IOException {
        if (alreadyExists) {
            enqueuePipelineResponsesExistsAlready(webServer);
        } else {
            enqueuePipelineResponsesDoesNotExistYet(webServer);
        }
    }

    private void enqueuePipelineResponsesDoesNotExistYet(final MockWebServer webServer) throws IOException {
        enqueueVersionedResourceResponsesDoesNotExistYet(monitoringPipelineNames(), webServer);
    }

    private void enqueuePipelineResponsesExistsAlready(final MockWebServer webServer) throws IOException {
        enqueueVersionedResourceResponsesExistsAlready(monitoringPipelineNames(), webServer);
    }

    private void enqueueVersionedResourceResponsesDoesNotExistYet(final List<String> names, final MockWebServer webServer)
            throws IOException {
        for (String resource : names) {
            if (randomBoolean()) {
                enqueueResponse(webServer, 404, "[" + resource + "] does not exist");
            } else if (randomBoolean()) {
                final int version = LAST_UPDATED_VERSION - randomIntBetween(1, 1000000);

                // it DOES exist, but it's an older version
                enqueueResponse(webServer, 200, "{\"" + resource + "\":{\"version\":" + version + "}}");
            } else {
                // no version specified
                enqueueResponse(webServer, 200, "{\"" + resource + "\":{}}");
            }
            enqueueResponse(webServer, 201, "[" + resource + "] created");
        }
    }

    private void enqueueVersionedResourceResponsesExistsAlready(final List<String> names, final MockWebServer webServer)
            throws IOException {
        for (String resource : names) {
            if (randomBoolean()) {
                final int newerVersion = randomFrom(Version.CURRENT.id, LAST_UPDATED_VERSION) + randomIntBetween(1, 1000000);

                // it's a NEWER resource (template / pipeline)
                enqueueResponse(webServer, 200, "{\"" + resource + "\":{\"version\":" + newerVersion + "}}");
            } else {
                // we already put it
                enqueueResponse(webServer, 200, "{\"" + resource + "\":{\"version\":" + LAST_UPDATED_VERSION + "}}");
            }
        }
    }

    private void enqueueWatcherResponses(final MockWebServer webServer,
                                         final boolean remoteClusterAllowsWatcher, final boolean currentLicenseAllowsWatcher,
                                         final boolean alreadyExists) throws IOException {
        // if the remote cluster doesn't allow watcher, then we only check for it and we're done
        if (remoteClusterAllowsWatcher) {
            // X-Pack exists and Watcher can be used
            enqueueResponse(webServer, 200, "{\"features\":{\"watcher\":{\"available\":true,\"enabled\":true}}}");

            // if we have an active license that's not Basic, then we should add watches
            if (currentLicenseAllowsWatcher) {
                if (alreadyExists) {
                    enqueueClusterAlertResponsesExistsAlready(webServer);
                } else {
                    enqueueClusterAlertResponsesDoesNotExistYet(webServer);
                }
            // otherwise we need to delete them from the remote cluster
            } else {
                enqueueDeleteClusterAlertResponses(webServer);
            }
        } else {
            // X-Pack exists but Watcher just cannot be used
            if (randomBoolean()) {
                final String responseBody = randomFrom(
                    "{\"features\":{\"watcher\":{\"available\":false,\"enabled\":true}}}",
                    "{\"features\":{\"watcher\":{\"available\":true,\"enabled\":false}}}",
                    "{}"
                );

                enqueueResponse(webServer, 200, responseBody);
            // X-Pack is not installed
            } else {
                enqueueResponse(webServer, 404, "{}");
            }
        }
    }

    private void enqueueClusterAlertResponsesDoesNotExistYet(final MockWebServer webServer)
            throws IOException {
        for (final String watchId : monitoringWatchIds()) {
            if (randomBoolean()) {
                enqueueResponse(webServer, 404, "watch [" + watchId + "] does not exist");
            } else if (randomBoolean()) {
                final int version = ClusterAlertsUtil.LAST_UPDATED_VERSION - randomIntBetween(1, 1000000);

                // it DOES exist, but it's an older version
                enqueueResponse(webServer, 200, "{\"metadata\":{\"xpack\":{\"version_created\":" + version + "}}}");
            } else {
                // no version specified
                enqueueResponse(webServer, 200, "{\"metadata\":{\"xpack\":{}}}");
            }
            enqueueResponse(webServer, 201, "[" + watchId + "] created");
        }
    }

    private void enqueueClusterAlertResponsesExistsAlready(final MockWebServer webServer) throws IOException {
        final int count = monitoringWatchIds().size();
        for (int i = 0; i < count; ++i) {
            final int existsVersion;

            if (randomBoolean()) {
                // it's a NEWER cluster alert
                existsVersion = randomFrom(Version.CURRENT.id, ClusterAlertsUtil.LAST_UPDATED_VERSION) + randomIntBetween(1, 1000000);
            } else {
                // we already put it
                existsVersion = ClusterAlertsUtil.LAST_UPDATED_VERSION;
            }

            enqueueResponse(webServer, 200, "{\"metadata\":{\"xpack\":{\"version_created\":" + existsVersion + "}}}");
        }
    }

    private void enqueueDeleteClusterAlertResponses(final MockWebServer webServer) throws IOException {
        for (final String watchId : monitoringWatchIds()) {
            if (randomBoolean()) {
                enqueueResponse(webServer, 404, "watch [" + watchId + "] did not exist");
            } else {
                enqueueResponse(webServer, 200, "watch [" + watchId + "] deleted");
            }
        }
    }

    private void writeIndex(XContentBuilder response, String index, String alias) throws IOException {
        response.startObject(index);
        {
            response.startObject("aliases");
            {
                response.startObject(alias).endObject();
            }
            response.endObject();
        }
        response.endObject();
    }

    private void enqueueResponse(int responseCode, String body) throws IOException {
        enqueueResponse(webServer, responseCode, body);
    }

    private void enqueueResponse(MockWebServer mockWebServer, int responseCode, String body) throws IOException {
        mockWebServer.enqueue(new MockResponse().setResponseCode(responseCode).setBody(body));
    }

    private void assertBulkRequest(String requestBody, int numberOfActions) throws Exception {
        BulkRequest bulkRequest = Requests.bulkRequest()
                .add(new BytesArray(requestBody.getBytes(StandardCharsets.UTF_8)), null, null, XContentType.JSON);
        assertThat(bulkRequest.numberOfActions(), equalTo(numberOfActions));
        for (DocWriteRequest actionRequest : bulkRequest.requests()) {
            assertThat(actionRequest, instanceOf(IndexRequest.class));
        }
    }

    private String getFormattedAddress(MockWebServer server) {
        return server.getHostName() + ":" + server.getPort();
    }

    private MockWebServer createMockWebServer() throws IOException {
        MockWebServer server = new MockWebServer();
        server.start();
        return server;
    }

    private List<Tuple<String, String>> monitoringTemplates(final boolean includeOldTemplates) {
        return includeOldTemplates ? monitoringTemplatesWithOldTemplates() : monitoringTemplates();
    }

    // this can be removed in 7.0
    private List<Tuple<String, String>> monitoringTemplatesWithOldTemplates() {
        final List<Tuple<String, String>> expectedTemplates = monitoringTemplates();

        expectedTemplates.addAll(
                Arrays.stream(MonitoringTemplateUtils.OLD_TEMPLATE_IDS)
                      .map(id -> new Tuple<>(MonitoringTemplateUtils.oldTemplateName(id), MonitoringTemplateUtils.createEmptyTemplate(id)))
                      .collect(Collectors.toList()));

        return expectedTemplates;
    }

    private List<String> monitoringTemplateNames(final boolean includeOldTemplates) {
        return includeOldTemplates ? monitoringTemplateNamesWithOldTemplates() : monitoringTemplateNames();
    }

    // this can be removed in 7.0
    protected List<String> monitoringTemplateNamesWithOldTemplates() {
        final List<String> expectedTemplateNames = monitoringTemplateNames();

        expectedTemplateNames.addAll(
                Arrays.stream(MonitoringTemplateUtils.OLD_TEMPLATE_IDS)
                      .map(MonitoringTemplateUtils::oldTemplateName)
                      .collect(Collectors.toList()));

        return expectedTemplateNames;
    }

}
