/*
 * 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.jmeter.visualizers.backend.graphite;

import java.text.DecimalFormat;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.jmeter.config.Arguments;
import org.apache.jmeter.samplers.SampleResult;
import org.apache.jmeter.util.JMeterUtils;
import org.apache.jmeter.visualizers.backend.AbstractBackendListenerClient;
import org.apache.jmeter.visualizers.backend.BackendListenerContext;
import org.apache.jmeter.visualizers.backend.SamplerMetric;
import org.apache.jmeter.visualizers.backend.UserMetric;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Graphite based Listener using Pickle Protocol
 * @see <a href="http://graphite.readthedocs.org/en/latest/overview.html">Graphite Overview</a>
 * @since 2.13
 */
public class GraphiteBackendListenerClient extends AbstractBackendListenerClient implements Runnable {

    //+ Argument names
    // These are stored in the JMX file, so DO NOT CHANGE ANY VALUES
    private static final String GRAPHITE_METRICS_SENDER = "graphiteMetricsSender"; //$NON-NLS-1$
    private static final String GRAPHITE_HOST = "graphiteHost"; //$NON-NLS-1$
    private static final String GRAPHITE_PORT = "graphitePort"; //$NON-NLS-1$
    private static final String ROOT_METRICS_PREFIX = "rootMetricsPrefix"; //$NON-NLS-1$
    private static final String PERCENTILES = "percentiles"; //$NON-NLS-1$
    private static final String SAMPLERS_LIST = "samplersList"; //$NON-NLS-1$
    public static final String USE_REGEXP_FOR_SAMPLERS_LIST = "useRegexpForSamplersList"; //$NON-NLS-1$
    public static final String USE_REGEXP_FOR_SAMPLERS_LIST_DEFAULT = "false";
    private static final String SUMMARY_ONLY = "summaryOnly"; //$NON-NLS-1$
    //- Argument names

    private static final int DEFAULT_PLAINTEXT_PROTOCOL_PORT = 2003;
    private static final String TEST_CONTEXT_NAME = "test";
    private static final String ALL_CONTEXT_NAME = "all";

    private static final Logger log = LoggerFactory.getLogger(GraphiteBackendListenerClient.class);
    private static final String DEFAULT_METRICS_PREFIX = "jmeter."; //$NON-NLS-1$
    private static final String CUMULATED_METRICS = "__cumulated__"; //$NON-NLS-1$
    // User Metrics
    private static final String METRIC_MAX_ACTIVE_THREADS = "maxAT"; //$NON-NLS-1$
    private static final String METRIC_MIN_ACTIVE_THREADS = "minAT"; //$NON-NLS-1$
    private static final String METRIC_MEAN_ACTIVE_THREADS = "meanAT"; //$NON-NLS-1$
    private static final String METRIC_STARTED_THREADS = "startedT"; //$NON-NLS-1$
    private static final String METRIC_FINISHED_THREADS = "endedT"; //$NON-NLS-1$

    // Response time Metrics
    private static final String METRIC_SEPARATOR = "."; //$NON-NLS-1$
    private static final String METRIC_OK_PREFIX = "ok"; //$NON-NLS-1$
    private static final String METRIC_KO_PREFIX = "ko"; //$NON-NLS-1$
    private static final String METRIC_ALL_PREFIX = "a"; //$NON-NLS-1$
    private static final String METRIC_HITS_PREFIX = "h"; //$NON-NLS-1$
    private static final String METRIC_SENT_BYTES_PREFIX = "sb"; //$NON-NLS-1$
    private static final String METRIC_RECEIVED_BYTES_PREFIX = "rb"; //$NON-NLS-1$

    private static final String METRIC_BYTES = "bytes"; //$NON-NLS-1$
    private static final String METRIC_COUNT = "count"; //$NON-NLS-1$
    private static final String METRIC_MIN_RESPONSE_TIME = "min"; //$NON-NLS-1$
    private static final String METRIC_MAX_RESPONSE_TIME = "max"; //$NON-NLS-1$
    private static final String METRIC_AVG_RESPONSE_TIME = "avg"; //$NON-NLS-1$
    private static final String METRIC_PERCENTILE = "pct"; //$NON-NLS-1$

    private static final String METRIC_OK_COUNT             = METRIC_OK_PREFIX+METRIC_SEPARATOR+METRIC_COUNT;
    private static final String METRIC_OK_MIN_RESPONSE_TIME = METRIC_OK_PREFIX+METRIC_SEPARATOR+METRIC_MIN_RESPONSE_TIME;
    private static final String METRIC_OK_MAX_RESPONSE_TIME = METRIC_OK_PREFIX+METRIC_SEPARATOR+METRIC_MAX_RESPONSE_TIME;
    private static final String METRIC_OK_AVG_RESPONSE_TIME = METRIC_OK_PREFIX+METRIC_SEPARATOR+METRIC_AVG_RESPONSE_TIME;
    private static final String METRIC_OK_PERCENTILE_PREFIX = METRIC_OK_PREFIX+METRIC_SEPARATOR+METRIC_PERCENTILE;

    private static final String METRIC_KO_COUNT             = METRIC_KO_PREFIX+METRIC_SEPARATOR+METRIC_COUNT;
    private static final String METRIC_KO_MIN_RESPONSE_TIME = METRIC_KO_PREFIX+METRIC_SEPARATOR+METRIC_MIN_RESPONSE_TIME;
    private static final String METRIC_KO_MAX_RESPONSE_TIME = METRIC_KO_PREFIX+METRIC_SEPARATOR+METRIC_MAX_RESPONSE_TIME;
    private static final String METRIC_KO_AVG_RESPONSE_TIME = METRIC_KO_PREFIX+METRIC_SEPARATOR+METRIC_AVG_RESPONSE_TIME;
    private static final String METRIC_KO_PERCENTILE_PREFIX = METRIC_KO_PREFIX+METRIC_SEPARATOR+METRIC_PERCENTILE;

    private static final String METRIC_ALL_COUNT             = METRIC_ALL_PREFIX+METRIC_SEPARATOR+METRIC_COUNT;
    private static final String METRIC_ALL_MIN_RESPONSE_TIME = METRIC_ALL_PREFIX+METRIC_SEPARATOR+METRIC_MIN_RESPONSE_TIME;
    private static final String METRIC_ALL_MAX_RESPONSE_TIME = METRIC_ALL_PREFIX+METRIC_SEPARATOR+METRIC_MAX_RESPONSE_TIME;
    private static final String METRIC_ALL_AVG_RESPONSE_TIME = METRIC_ALL_PREFIX+METRIC_SEPARATOR+METRIC_AVG_RESPONSE_TIME;
    private static final String METRIC_ALL_PERCENTILE_PREFIX = METRIC_ALL_PREFIX+METRIC_SEPARATOR+METRIC_PERCENTILE;

    private static final String METRIC_ALL_HITS_COUNT        = METRIC_HITS_PREFIX+METRIC_SEPARATOR+METRIC_COUNT;
    private static final String METRIC_ALL_SENT_BYTES        = METRIC_SENT_BYTES_PREFIX+METRIC_SEPARATOR+METRIC_BYTES;
    private static final String METRIC_ALL_RECEIVED_BYTES    = METRIC_RECEIVED_BYTES_PREFIX+METRIC_SEPARATOR+METRIC_BYTES;

    private static final long SEND_INTERVAL = JMeterUtils.getPropDefault("backend_graphite.send_interval", 1);
    private static final int MAX_POOL_SIZE = 1;
    private static final String DEFAULT_PERCENTILES = "90;95;99";
    private static final String SEPARATOR = ";"; //$NON-NLS-1$
    private static final Object LOCK = new Object();

    private boolean summaryOnly;
    private String samplersList = ""; //$NON-NLS-1$
    private boolean useRegexpForSamplersList;
    private Set<String> samplersToFilter;
    private Map<String, Float> okPercentiles;
    private Map<String, Float> koPercentiles;
    private Map<String, Float> allPercentiles;

    private GraphiteMetricsSender graphiteMetricsManager;

    private ScheduledExecutorService scheduler;
    private ScheduledFuture<?> timerHandle;

    private Pattern pattern;

    public GraphiteBackendListenerClient() {
        super();
    }

    @Override
    public void run() {
        sendMetrics();
    }

    /** Send metrics to Graphite */
    protected void sendMetrics() {
        // Need to convert millis to seconds for Graphite
        long timestampInSeconds = TimeUnit.SECONDS.convert(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        synchronized (LOCK) {
            for (Map.Entry<String, SamplerMetric> entry : getMetricsPerSampler().entrySet()) {
                final String key = entry.getKey();
                final SamplerMetric metric = entry.getValue();
                if (key.equals(CUMULATED_METRICS)) {
                    addMetrics(timestampInSeconds, ALL_CONTEXT_NAME, metric);
                } else {
                    addMetrics(timestampInSeconds, AbstractGraphiteMetricsSender.sanitizeString(key), metric);
                }
                // We are computing on interval basis so cleanup
                metric.resetForTimeInterval();
            }
        }
        UserMetric userMetric = getUserMetrics();
        graphiteMetricsManager.addMetric(timestampInSeconds, TEST_CONTEXT_NAME,
                METRIC_MIN_ACTIVE_THREADS,
                Integer.toString(userMetric.getMinActiveThreads()));
        graphiteMetricsManager.addMetric(timestampInSeconds, TEST_CONTEXT_NAME,
                METRIC_MAX_ACTIVE_THREADS,
                Integer.toString(userMetric.getMaxActiveThreads()));
        graphiteMetricsManager.addMetric(timestampInSeconds, TEST_CONTEXT_NAME,
                METRIC_MEAN_ACTIVE_THREADS,
                Integer.toString(userMetric.getMeanActiveThreads()));
        graphiteMetricsManager.addMetric(timestampInSeconds, TEST_CONTEXT_NAME,
                METRIC_STARTED_THREADS,
                Integer.toString(userMetric.getStartedThreads()));
        graphiteMetricsManager.addMetric(timestampInSeconds, TEST_CONTEXT_NAME,
                METRIC_FINISHED_THREADS,
                Integer.toString(userMetric.getFinishedThreads()));

        graphiteMetricsManager.writeAndSendMetrics();
    }

    /**
     * Add request metrics to metrics manager.
     * Note if total number of requests is 0, no response time metrics are sent.
     *
     * @param timestampInSeconds long
     * @param contextName        String
     * @param metric             {@link SamplerMetric}
     */
    private void addMetrics(long timestampInSeconds, String contextName, SamplerMetric metric) {

        // See https://bz.apache.org/bugzilla/show_bug.cgi?id=57350
        if (metric.getTotal() <= 0) {
            return;
        }

        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_OK_COUNT, Integer.toString(metric.getSuccesses()));
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_KO_COUNT, Integer.toString(metric.getFailures()));
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_ALL_COUNT, Integer.toString(metric.getTotal()));
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_ALL_HITS_COUNT, Integer.toString(metric.getHits()));
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_ALL_SENT_BYTES, Long.toString(metric.getSentBytes()));
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_ALL_RECEIVED_BYTES, Long.toString(metric.getReceivedBytes()));

        if (metric.getSuccesses() > 0) {
            graphiteMetricsManager.addMetric(timestampInSeconds,
                    contextName, METRIC_OK_MIN_RESPONSE_TIME,
                    Double.toString(metric.getOkMinTime()));
            graphiteMetricsManager.addMetric(timestampInSeconds,
                    contextName, METRIC_OK_MAX_RESPONSE_TIME,
                    Double.toString(metric.getOkMaxTime()));
            graphiteMetricsManager.addMetric(timestampInSeconds,
                    contextName, METRIC_OK_AVG_RESPONSE_TIME,
                    Double.toString(metric.getOkMean()));
            for (Map.Entry<String, Float> entry : okPercentiles.entrySet()) {
                graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                        entry.getKey(),
                        Double.toString(metric.getOkPercentile(entry.getValue())));
            }
        }
        if (metric.getFailures() > 0) {
            graphiteMetricsManager.addMetric(timestampInSeconds,
                    contextName, METRIC_KO_MIN_RESPONSE_TIME,
                    Double.toString(metric.getKoMinTime()));
            graphiteMetricsManager.addMetric(timestampInSeconds,
                    contextName, METRIC_KO_MAX_RESPONSE_TIME,
                    Double.toString(metric.getKoMaxTime()));
            graphiteMetricsManager.addMetric(timestampInSeconds,
                    contextName, METRIC_KO_AVG_RESPONSE_TIME,
                    Double.toString(metric.getKoMean()));
            for (Map.Entry<String, Float> entry : koPercentiles.entrySet()) {
                graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                        entry.getKey(),
                        Double.toString(metric.getKoPercentile(entry.getValue())));
            }
        }
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_ALL_MIN_RESPONSE_TIME,
                Double.toString(metric.getAllMinTime()));
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_ALL_MAX_RESPONSE_TIME,
                Double.toString(metric.getAllMaxTime()));
        graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                METRIC_ALL_AVG_RESPONSE_TIME,
                Double.toString(metric.getAllMean()));
        for (Map.Entry<String, Float> entry : allPercentiles.entrySet()) {
            graphiteMetricsManager.addMetric(timestampInSeconds, contextName,
                    entry.getKey(),
                    Double.toString(metric.getAllPercentile(entry.getValue())));
        }
    }

    /** @return the samplersList */
    public String getSamplersList() {
        return samplersList;
    }

    /** @param samplersList the samplersList to set */
    public void setSamplersList(String samplersList) {
        this.samplersList = samplersList;
    }

    @Override
    public void handleSampleResults(List<SampleResult> sampleResults, BackendListenerContext context) {
        boolean samplersToFilterMatch;
        synchronized (LOCK) {
            UserMetric userMetrics = getUserMetrics();
            for (SampleResult sampleResult : sampleResults) {
                userMetrics.add(sampleResult);

                if (!summaryOnly) {
                    if (useRegexpForSamplersList) {
                        Matcher matcher = pattern.matcher(sampleResult.getSampleLabel());
                        samplersToFilterMatch = matcher.matches();
                    } else {
                        samplersToFilterMatch = samplersToFilter.contains(sampleResult.getSampleLabel());
                    }
                    if (samplersToFilterMatch) {
                        SamplerMetric samplerMetric = getSamplerMetric(sampleResult.getSampleLabel());
                        samplerMetric.add(sampleResult);
                    }
                }
                getSamplerMetric(CUMULATED_METRICS).add(sampleResult);
            }
        }
    }

    @Override
    public void setupTest(BackendListenerContext context) throws Exception {
        String graphiteMetricsSenderClass = context.getParameter(GRAPHITE_METRICS_SENDER);

        String graphiteHost = context.getParameter(GRAPHITE_HOST);
        int graphitePort = context.getIntParameter(GRAPHITE_PORT, DEFAULT_PLAINTEXT_PROTOCOL_PORT);
        summaryOnly = context.getBooleanParameter(SUMMARY_ONLY, true);
        samplersList = context.getParameter(SAMPLERS_LIST, "");
        useRegexpForSamplersList = context.getBooleanParameter(USE_REGEXP_FOR_SAMPLERS_LIST, false);
        String rootMetricsPrefix = context.getParameter(ROOT_METRICS_PREFIX, DEFAULT_METRICS_PREFIX);
        String[] percentilesStringArray = context.getParameter(PERCENTILES, DEFAULT_METRICS_PREFIX).split(SEPARATOR);
        okPercentiles = new HashMap<>(percentilesStringArray.length);
        koPercentiles = new HashMap<>(percentilesStringArray.length);
        allPercentiles = new HashMap<>(percentilesStringArray.length);
        Arrays.stream(percentilesStringArray)
                .map(String::trim)
                .filter(StringUtils::isNotEmpty)
                .forEach(this::initPercentileMaps);
        Class<?> clazz = Class.forName(graphiteMetricsSenderClass);
        this.graphiteMetricsManager = (GraphiteMetricsSender) clazz.getDeclaredConstructor().newInstance();
        graphiteMetricsManager.setup(graphiteHost, graphitePort, rootMetricsPrefix);
        if (useRegexpForSamplersList) {
            pattern = Pattern.compile(samplersList);
        } else {
            String[] samplers = samplersList.split(SEPARATOR);
            samplersToFilter = new HashSet<>();
            Collections.addAll(samplersToFilter, samplers);
        }
        scheduler = Executors.newScheduledThreadPool(MAX_POOL_SIZE);
        // Don't change this as metrics are per second
        this.timerHandle = scheduler.scheduleAtFixedRate(this, SEND_INTERVAL, SEND_INTERVAL, TimeUnit.SECONDS);
    }

    private void initPercentileMaps(String percentilesString) {
        DecimalFormat decimalFormat = new DecimalFormat("0.##");
        try {
            Float percentileValue = Float.valueOf(percentilesString.trim());
            String sanitizedFormattedPercentile =
                    AbstractGraphiteMetricsSender.sanitizeString(
                            decimalFormat.format(percentileValue));
            okPercentiles.put(
                    METRIC_OK_PERCENTILE_PREFIX + sanitizedFormattedPercentile,
                    percentileValue);
            koPercentiles.put(
                    METRIC_KO_PERCENTILE_PREFIX + sanitizedFormattedPercentile,
                    percentileValue);
            allPercentiles.put(
                    METRIC_ALL_PERCENTILE_PREFIX + sanitizedFormattedPercentile,
                    percentileValue);

        } catch (Exception e) {
            log.error("Error parsing percentile: '{}'", percentilesString, e);
        }
    }

    @Override
    public void teardownTest(BackendListenerContext context) throws Exception {
        boolean cancelState = timerHandle.cancel(false);
        log.debug("Canceled state: {}", cancelState);
        scheduler.shutdown();
        try {
            scheduler.awaitTermination(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("Error waiting for end of scheduler");
            Thread.currentThread().interrupt();
        }
        // Send last set of data before ending
        sendMetrics();

        if (samplersToFilter != null) {
            samplersToFilter.clear();
        }
        graphiteMetricsManager.destroy();
        super.teardownTest(context);
    }

    @Override
    public Arguments getDefaultParameters() {
        Arguments arguments = new Arguments();
        arguments.addArgument(GRAPHITE_METRICS_SENDER, TextGraphiteMetricsSender.class.getName());
        arguments.addArgument(GRAPHITE_HOST, "");
        arguments.addArgument(GRAPHITE_PORT, Integer.toString(DEFAULT_PLAINTEXT_PROTOCOL_PORT));
        arguments.addArgument(ROOT_METRICS_PREFIX, DEFAULT_METRICS_PREFIX);
        arguments.addArgument(SUMMARY_ONLY, "true");
        arguments.addArgument(SAMPLERS_LIST, "");
        arguments.addArgument(USE_REGEXP_FOR_SAMPLERS_LIST, USE_REGEXP_FOR_SAMPLERS_LIST_DEFAULT);
        arguments.addArgument(PERCENTILES, DEFAULT_PERCENTILES);
        return arguments;
    }
}
