/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.tools;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * MetricsReporter that aggregates metrics data and reports it via HTTP requests to a configurable
 * webhook endpoint in JSON format.
 *
 * This is an internal class used for system tests and does not provide any compatibility guarantees.
 */
public class PushHttpMetricsReporter implements MetricsReporter {
    private static final Logger log = LoggerFactory.getLogger(PushHttpMetricsReporter.class);

    private static final String METRICS_PREFIX = "metrics.";
    static final String METRICS_URL_CONFIG = METRICS_PREFIX + "url";
    static final String METRICS_PERIOD_CONFIG = METRICS_PREFIX + "period";
    static final String METRICS_HOST_CONFIG = METRICS_PREFIX + "host";
    static final String CLIENT_ID_CONFIG = ProducerConfig.CLIENT_ID_CONFIG;

    private static final Map<String, String> HEADERS = new LinkedHashMap<>();
    static {
        HEADERS.put("Content-Type", "application/json");
    }

    private final Object lock = new Object();
    private final Time time;
    private final ScheduledExecutorService executor;
    // The set of metrics are updated in init/metricChange/metricRemoval
    private final Map<MetricName, KafkaMetric> metrics = new LinkedHashMap<>();
    private final ObjectMapper json = new ObjectMapper();

    // Non-final because these are set via configure()
    private URL url;
    private String host;
    private String clientId;

    private static final ConfigDef CONFIG_DEF = new ConfigDef()
            .define(METRICS_URL_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH,
                    "The URL to report metrics to")
            .define(METRICS_PERIOD_CONFIG, ConfigDef.Type.INT, ConfigDef.Importance.HIGH,
                    "The frequency at which metrics should be reported, in second")
            .define(METRICS_HOST_CONFIG, ConfigDef.Type.STRING, "", ConfigDef.Importance.LOW,
                    "The hostname to report with each metric; if empty, defaults to the FQDN that can be automatically" +
                            "determined")
            .define(CLIENT_ID_CONFIG, ConfigDef.Type.STRING, "", ConfigDef.Importance.LOW,
                    "Client ID to identify the application, generally inherited from the " +
                            "producer/consumer/streams/connect instance");

    public PushHttpMetricsReporter() {
        time = new SystemTime();
        executor = Executors.newSingleThreadScheduledExecutor();
    }

    PushHttpMetricsReporter(Time mockTime, ScheduledExecutorService mockExecutor) {
        time = mockTime;
        executor = mockExecutor;
    }

    @Override
    public void configure(Map<String, ?> configs) {
        PushHttpMetricsReporterConfig config = new PushHttpMetricsReporterConfig(CONFIG_DEF, configs);
        try {
            url = new URL(config.getString(METRICS_URL_CONFIG));
        } catch (MalformedURLException e) {
            throw new ConfigException("Malformed metrics.url", e);
        }
        int period = config.getInteger(METRICS_PERIOD_CONFIG);
        clientId = config.getString(CLIENT_ID_CONFIG);

        host = config.getString(METRICS_HOST_CONFIG);
        if (host == null || host.isEmpty()) {
            try {
                host = InetAddress.getLocalHost().getCanonicalHostName();
            } catch (UnknownHostException e) {
                throw new ConfigException("Failed to get canonical hostname", e);
            }
        }

        executor.scheduleAtFixedRate(new HttpReporter(), period, period, TimeUnit.SECONDS);

        log.info("Configured PushHttpMetricsReporter for {} to report every {} seconds", url, period);
    }

    @Override
    public void init(List<KafkaMetric> initMetrics) {
        synchronized (lock) {
            for (KafkaMetric metric : initMetrics) {
                log.debug("Adding metric {}", metric.metricName());
                metrics.put(metric.metricName(), metric);
            }
        }
    }

    @Override
    public void metricChange(KafkaMetric metric) {
        synchronized (lock) {
            log.debug("Updating metric {}", metric.metricName());
            metrics.put(metric.metricName(), metric);
        }
    }

    @Override
    public void metricRemoval(KafkaMetric metric) {
        synchronized (lock) {
            log.debug("Removing metric {}", metric.metricName());
            metrics.remove(metric.metricName());
        }
    }

    @Override
    public void close() {
        executor.shutdown();
        try {
            executor.awaitTermination(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            throw new KafkaException("Interrupted when shutting down PushHttpMetricsReporter", e);
        }
    }

    private class HttpReporter implements Runnable {
        @Override
        public void run() {
            long now = time.milliseconds();
            final List<MetricValue> samples;
            synchronized (lock) {
                samples = new ArrayList<>(metrics.size());
                for (KafkaMetric metric : metrics.values()) {
                    MetricName name = metric.metricName();
                    samples.add(new MetricValue(name.name(), name.group(), name.tags(), metric.metricValue()));
                }
            }

            MetricsReport report = new MetricsReport(new MetricClientInfo(host, clientId, now), samples);

            log.trace("Reporting {} metrics to {}", samples.size(), url);
            HttpURLConnection connection = null;
            try {
                connection = newHttpConnection(url);
                connection.setRequestMethod("POST");
                // connection.getResponseCode() implicitly calls getInputStream, so always set to true.
                // On the other hand, leaving this out breaks nothing.
                connection.setDoInput(true);
                connection.setRequestProperty("Content-Type", "application/json");
                byte[] data = json.writeValueAsBytes(report);
                connection.setRequestProperty("Content-Length", Integer.toString(data.length));
                connection.setRequestProperty("Accept", "*/*");
                connection.setUseCaches(false);

                connection.setDoOutput(true);

                try (OutputStream os = connection.getOutputStream()) {
                    os.write(data);
                    os.flush();
                }

                int responseCode = connection.getResponseCode();
                if (responseCode >= 400) {
                    InputStream is = connection.getErrorStream();
                    String msg = readResponse(is);
                    log.error("Error reporting metrics, {}: {}", responseCode, msg);
                } else if (responseCode >= 300) {
                    log.error("PushHttpMetricsReporter does not currently support redirects, saw {}", responseCode);
                } else {
                    log.info("Finished reporting metrics with response code {}", responseCode);
                }
            } catch (Throwable t) {
                log.error("Error reporting metrics", t);
            } finally {
                if (connection != null) {
                    connection.disconnect();
                }
            }
        }
    }

    // Static package-private so unit tests can use a mock connection
    static HttpURLConnection newHttpConnection(URL url) throws IOException {
        return (HttpURLConnection) url.openConnection();
    }

    // Static package-private so unit tests can mock reading response
    static String readResponse(InputStream is) {
        try (Scanner s = new Scanner(is, StandardCharsets.UTF_8.name()).useDelimiter("\\A")) {
            return s.hasNext() ? s.next() : "";
        }
    }

    private static class MetricsReport {
        private final MetricClientInfo client;
        private final Collection<MetricValue> metrics;

        MetricsReport(MetricClientInfo client, Collection<MetricValue> metrics) {
            this.client = client;
            this.metrics = metrics;
        }

        @JsonProperty
        public MetricClientInfo client() {
            return client;
        }

        @JsonProperty
        public Collection<MetricValue> metrics() {
            return metrics;
        }
    }

    private static class MetricClientInfo {
        private final String host;
        private final String clientId;
        private final long time;

        MetricClientInfo(String host, String clientId, long time) {
            this.host = host;
            this.clientId = clientId;
            this.time = time;
        }

        @JsonProperty
        public String host() {
            return host;
        }

        @JsonProperty("client_id")
        public String clientId() {
            return clientId;
        }

        @JsonProperty
        public long time() {
            return time;
        }
    }

    private static class MetricValue {

        private final String name;
        private final String group;
        private final Map<String, String> tags;
        private final Object value;

        MetricValue(String name, String group, Map<String, String> tags, Object value) {
            this.name = name;
            this.group = group;
            this.tags = tags;
            this.value = value;
        }

        @JsonProperty
        public String name() {
            return name;
        }

        @JsonProperty
        public String group() {
            return group;
        }

        @JsonProperty
        public Map<String, String> tags() {
            return tags;
        }

        @JsonProperty
        public Object value() {
            return value;
        }
    }

    // The signature for getInt changed from returning int to Integer so to remain compatible with 0.8.2.2 jars
    // for system tests we replace it with a custom version that works for all versions.
    private static class PushHttpMetricsReporterConfig extends AbstractConfig {
        public PushHttpMetricsReporterConfig(ConfigDef definition, Map<?, ?> originals) {
            super(definition, originals);
        }

        public Integer getInteger(String key) {
            return (Integer) get(key);
        }

    }
}
