/*
 * 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 org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.Exit;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.lang.management.ManagementFactory;
import java.net.ServerSocket;
import java.rmi.registry.LocateRegistry;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.management.remote.JMXConnectorServer;
import javax.management.remote.JMXConnectorServerFactory;
import javax.management.remote.JMXServiceURL;

import static java.lang.String.format;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class JmxToolTest {
    private final ToolsTestUtils.MockExitProcedure exitProcedure = new ToolsTestUtils.MockExitProcedure();
    private static JMXConnectorServer jmxAgent;
    private static String jmxUrl;

    @BeforeAll
    public static void beforeAll() throws Exception {
        int port = findRandomOpenPortOnAllLocalInterfaces();
        jmxUrl = format("service:jmx:rmi:///jndi/rmi://:%d/jmxrmi", port);
        // explicitly set the hostname returned to the the clients in the remote stub object
        // when connecting to a multi-homed machine using RMI, the wrong address may be returned
        // by the RMI registry to the client, causing the connection to the RMI server to timeout
        System.setProperty("java.rmi.server.hostname", "localhost");
        LocateRegistry.createRegistry(port);
        Map<String, Object> env = new HashMap<>();
        env.put("com.sun.management.jmxremote.authenticate", "false");
        env.put("com.sun.management.jmxremote.ssl", "false");
        JMXServiceURL url = new JMXServiceURL(jmxUrl);
        MBeanServer server = ManagementFactory.getPlatformMBeanServer();
        server.registerMBean(new Metrics(),
            new ObjectName("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec"));
        jmxAgent = JMXConnectorServerFactory.newJMXConnectorServer(url, env, server);
        jmxAgent.start();
    }

    @AfterAll
    public static void afterAll() throws Exception {
        jmxAgent.stop();
    }

    @BeforeEach
    public void beforeEach() {
        Exit.setExitProcedure(exitProcedure);
    }

    @AfterEach
    public void afterEach() {
        Exit.resetExitProcedure();
    }

    @Test
    public void kafkaVersion() {
        String out = executeAndGetOut("--version");
        assertNormalExit();
        assertTrue(out.contains(AppInfoParser.getVersion()));
    }

    @Test
    public void unrecognizedOption() {
        String err = executeAndGetErr("--foo");
        assertCommandFailure();
        assertTrue(err.contains("UnrecognizedOptionException"));
        assertTrue(err.contains("foo"));
    }

    @Test
    public void missingRequired() {
        String err = executeAndGetErr("--reporting-interval");
        assertCommandFailure();
        assertTrue(err.contains("OptionMissingRequiredArgumentException"));
        assertTrue(err.contains("reporting-interval"));
    }

    @Test
    public void malformedURL() {
        String err = executeAndGetErr("--jmx-url", "localhost:9999");
        assertCommandFailure();
        assertTrue(err.contains("MalformedURLException"));
    }

    @Test
    public void helpOptions() {
        String[] expectedOptions = new String[]{
            "--attributes", "--date-format", "--help", "--jmx-auth-prop",
            "--jmx-ssl-enable", "--jmx-url", "--object-name", "--one-time",
            "--report-format", "--reporting-interval", "--version", "--wait"
        };
        String err = executeAndGetErr("--help");
        assertCommandFailure();
        for (String option : expectedOptions) {
            assertTrue(err.contains(option), option);
        }
    }

    @Test
    public void csvFormat() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        Arrays.stream(out.split("\\r?\\n")).forEach(line -> {
            assertTrue(line.matches("([a-zA-Z0-9=:,.]+),\"([ -~]+)\""), line);
        });
    }

    @Test
    public void tsvFormat() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec",
            "--report-format", "tsv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        Arrays.stream(out.split("\\r?\\n")).forEach(line -> {
            assertTrue(line.matches("([a-zA-Z0-9=:,.]+)\\t([ -~]+)"), line);
        });
    }

    @Test
    public void allMetrics() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--report-format", "csv",
            "--reporting-interval", "-1"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertFalse(csv.isEmpty());
    }

    @Test
    public void filteredMetrics() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec",
            "--attributes", "FifteenMinuteRate,FiveMinuteRate",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    public void testDomainNamePattern() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.serve?:*",
            "--attributes", "FifteenMinuteRate,FiveMinuteRate",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    public void testDomainNamePatternWithNoAttributes() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.serve?:*",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    public void testPropertyListPattern() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=BrokerTopicMetrics,*",
            "--attributes", "FifteenMinuteRate,FiveMinuteRate",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    public void testPropertyListPatternWithNoAttributes() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=BrokerTopicMetrics,*",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    public void testPropertyValuePattern() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=BrokerTopicMetrics,name=*InPerSec",
            "--attributes", "FifteenMinuteRate,FiveMinuteRate",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    public void testPropertyValuePatternWithNoAttributes() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=BrokerTopicMetrics,name=*InPerSec",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    // Combination of property-list and property-value patterns
    public void testPropertyPattern() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=*,*",
            "--attributes", "FifteenMinuteRate,FiveMinuteRate",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    // Combination of property-list and property-value patterns
    public void testPropertyPatternWithNoAttributes() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=*,*",
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertEquals("1.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FifteenMinuteRate"));
        assertEquals("3.0", csv.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec:FiveMinuteRate"));
    }

    @Test
    public void dateFormat() {
        String dateFormat = "yyyyMMdd-hh:mm:ss";
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--date-format", dateFormat,
            "--report-format", "csv",
            "--one-time"
        };
        String out = executeAndGetOut(args);
        assertNormalExit();

        Map<String, String> csv = parseCsv(out);
        assertTrue(validDateFormat(dateFormat, csv.get("time")));
    }

    @Test
    public void unknownObjectName() {
        String[] args = new String[]{
            "--jmx-url", jmxUrl,
            "--object-name", "kafka.server:type=DummyMetrics,name=MessagesInPerSec",
            "--wait"
        };

        String err = executeAndGetErr(args);
        assertCommandFailure();
        assertTrue(err.contains("Could not find all requested object names after 10000 ms"));
    }

    private static int findRandomOpenPortOnAllLocalInterfaces() throws Exception {
        try (ServerSocket socket = new ServerSocket(0)) {
            return socket.getLocalPort();
        }
    }

    private String executeAndGetOut(String... args) {
        return execute(args, false);
    }

    private String executeAndGetErr(String... args) {
        return execute(args, true);
    }

    private String execute(String[] args, boolean err) {
        Runnable runnable = () -> {
            try {
                JmxTool.main(args);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        };
        return err ? ToolsTestUtils.captureStandardErr(runnable)
                    : ToolsTestUtils.captureStandardOut(runnable);
    }

    private void assertNormalExit() {
        assertTrue(exitProcedure.hasExited());
        assertEquals(0, exitProcedure.statusCode());
    }

    private void assertCommandFailure() {
        assertTrue(exitProcedure.hasExited());
        assertEquals(1, exitProcedure.statusCode());
    }

    private Map<String, String> parseCsv(String value) {
        Map<String, String> result = new HashMap<>();
        Arrays.stream(value.split("\\r?\\n")).forEach(line -> {
            String[] cells = line.split(",\"");
            if (cells.length == 2) {
                result.put(cells[0], cells[1].replaceAll("\"", ""));
            }
        });
        return result;
    }

    private boolean validDateFormat(String format, String value) {
        DateFormat formatter = new SimpleDateFormat(format);
        formatter.setLenient(false);
        try {
            formatter.parse(value);
            return true;
        } catch (ParseException e) {
            return false;
        }
    }

    public interface MetricsMBean {
        double getFifteenMinuteRate();
        double getFiveMinuteRate();
    }

    public static class Metrics implements MetricsMBean {
        @Override
        public double getFifteenMinuteRate() {
            return 1.0;
        }

        @Override
        public double getFiveMinuteRate() {
            return 3.0;
        }
    }
}
