<commit_before>/*
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.eagle.alert.metric;

import com.codahale.metrics.*;
import com.codahale.metrics.jvm.FileDescriptorRatioGauge;
import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import kafka.admin.AdminUtils;
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.OffsetRequest;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;
import kafka.utils.ZKStringSerializer$;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.io.FileUtils;
import org.apache.eagle.alert.metric.entity.MetricEvent;
import org.apache.eagle.alert.metric.sink.ConsoleSink;
import org.apache.eagle.alert.metric.sink.Slf4jSink;
import org.apache.eagle.alert.metric.source.JVMMetricSource;
import org.apache.eagle.alert.metric.source.MetricSource;
import org.apache.eagle.alert.utils.JsonUtils;
import org.apache.eagle.alert.utils.KafkaEmbedded;
import org.junit.*;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Scanner;

public class MetricSystemTest {

    public static final String END_LINE = System.getProperty("line.separator");
    @Rule
    public TemporaryFolder tempFolder = new TemporaryFolder();
    private static final int DATA_BEGIN_INDEX = 55;
    private static final String TOPIC = "alert_metric_test";
    private static final String clientName = "test";

    @Test
    public void testMetricEvent() {
        MetricEvent metricEvent = MetricEvent.of("test").build();
        Assert.assertEquals(metricEvent.get("name"), "test");
        Assert.assertNotNull(metricEvent.get("timestamp"));

        metricEvent = MetricEvent.of("test1").build();
        metricEvent.put("timestamp", 1);
        Assert.assertEquals(metricEvent.get("name"), "test1");
        Assert.assertEquals(metricEvent.get("timestamp"), 1);

        Counter counter = new Counter();
        counter.inc(10);
        metricEvent = MetricEvent.of("testcount").from(counter).build();
        Assert.assertEquals(metricEvent.get("count"), 10l);

        Gauge gauge = Mockito.mock(FileDescriptorRatioGauge.class);
        Mockito.when(gauge.getValue()).thenReturn(new Double("0.4"));
        metricEvent = MetricEvent.of("testGauge").from(gauge).build();
        Assert.assertEquals(metricEvent.get("value"), 0.4);

        //Histogram
        Histogram histogram = Mockito.mock(Histogram.class);
        Snapshot snapshot = Mockito.mock(Snapshot.class);
        Mockito.when(histogram.getCount()).thenReturn(11l);
        Mockito.when(histogram.getSnapshot()).thenReturn(snapshot);
        Mockito.when(snapshot.getMin()).thenReturn(1l);
        Mockito.when(snapshot.getMax()).thenReturn(2l);
        Mockito.when(snapshot.getMean()).thenReturn(3d);
        Mockito.when(snapshot.getStdDev()).thenReturn(4d);
        Mockito.when(snapshot.getMedian()).thenReturn(5d);
        Mockito.when(snapshot.get75thPercentile()).thenReturn(6d);
        Mockito.when(snapshot.get95thPercentile()).thenReturn(7d);
        Mockito.when(snapshot.get98thPercentile()).thenReturn(8d);
        Mockito.when(snapshot.get99thPercentile()).thenReturn(9d);
        Mockito.when(snapshot.get999thPercentile()).thenReturn(10d);
        metricEvent = MetricEvent.of("testHistogram").from(histogram).build();

        Assert.assertEquals(metricEvent.get("count"), 11l);
        Assert.assertEquals(metricEvent.get("min"), 1l);
        Assert.assertEquals(metricEvent.get("max"), 2l);
        Assert.assertEquals(metricEvent.get("mean"), 3d);
        Assert.assertEquals(metricEvent.get("stddev"), 4d);
        Assert.assertEquals(metricEvent.get("median"), 5d);
        Assert.assertEquals(metricEvent.get("75thPercentile"), 6d);
        Assert.assertEquals(metricEvent.get("95thPercentile"), 7d);
        Assert.assertEquals(metricEvent.get("98thPercentile"), 8d);
        Assert.assertEquals(metricEvent.get("99thPercentile"), 9d);
        Assert.assertEquals(metricEvent.get("999thPercentile"), 10d);

        //Meter
        Meter meter = Mockito.mock(Meter.class);
        Mockito.when(meter.getCount()).thenReturn(1l);
        Mockito.when(meter.getOneMinuteRate()).thenReturn(2d);
        Mockito.when(meter.getFiveMinuteRate()).thenReturn(3d);
        Mockito.when(meter.getFifteenMinuteRate()).thenReturn(4d);
        Mockito.when(meter.getMeanRate()).thenReturn(5d);
        metricEvent = MetricEvent.of("testMeter").from(meter).build();

        Assert.assertEquals(metricEvent.get("value"), 1l);
        Assert.assertEquals(metricEvent.get("1MinRate"), 2d);
        Assert.assertEquals(metricEvent.get("5MinRate"), 3d);
        Assert.assertEquals(metricEvent.get("15MinRate"), 4d);
        Assert.assertEquals(metricEvent.get("mean"), 5d);

        //Timer
        Timer value = Mockito.mock(Timer.class);
        Mockito.when(value.getCount()).thenReturn(1l);
        Mockito.when(value.getOneMinuteRate()).thenReturn(2d);
        Mockito.when(value.getFiveMinuteRate()).thenReturn(3d);
        Mockito.when(value.getFifteenMinuteRate()).thenReturn(4d);
        Mockito.when(value.getMeanRate()).thenReturn(5d);
        metricEvent = MetricEvent.of("testTimer").from(value).build();

        Assert.assertEquals(metricEvent.get("value"), 1l);
        Assert.assertEquals(metricEvent.get("1MinRate"), 2d);
        Assert.assertEquals(metricEvent.get("5MinRate"), 3d);
        Assert.assertEquals(metricEvent.get("15MinRate"), 4d);
        Assert.assertEquals(metricEvent.get("mean"), 5d);

    }

    @Test
    public void testMerticSystemWithKafkaSink() throws IOException {

        JVMMetricSource jvmMetricSource = mockMetricRegistry();
        //setup kafka
        KafkaEmbedded kafkaEmbedded = new KafkaEmbedded();
        makeSureTopic(kafkaEmbedded.getZkConnectionString());
        //setup metric system
        File file = genKafkaSinkConfig(kafkaEmbedded.getBrokerConnectionString());
        Config config = ConfigFactory.parseFile(file);
        MetricSystem system = MetricSystem.load(config);
        system.register(jvmMetricSource);
        system.start();
        system.report();

        SimpleConsumer consumer = assertMsgFromKafka(kafkaEmbedded);
        system.stop();
        consumer.close();
        kafkaEmbedded.shutdown();
    }

    @Test
    public void testConsoleSink() throws IOException {
        PrintStream console = System.out;
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        System.setOut(new PrintStream(bytes));

        ConsoleSink sink = new ConsoleSink();
        MetricRegistry registry = new MetricRegistry();
        JvmAttributeGaugeSet jvm = Mockito.mock(JvmAttributeGaugeSet.class);
        Map<String, Metric> metrics = new HashMap<>();
        metrics.put("name", (Gauge) () -> "testname");
        metrics.put("uptime", (Gauge) () -> "testuptime");
        metrics.put("vendor", (Gauge) () -> "testvendor");
        Mockito.when(jvm.getMetrics()).thenReturn(metrics);
        registry.registerAll(jvm);
        File file = genConsoleSinkConfig();
        Config config = ConfigFactory.parseFile(file);
        sink.prepare(config, registry);
        sink.report();
        sink.stop();
        String result = bytes.toString();
        result = result.substring(result.indexOf(END_LINE) + 1);//remove first line
        Assert.assertEquals("" + END_LINE + "" +
                "-- Gauges ----------------------------------------------------------------------" + END_LINE + "" +<commit_msg>[MINOR] Fix test fail on windows<commit_after>/*
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.eagle.alert.metric;

import com.codahale.metrics.*;
import com.codahale.metrics.jvm.FileDescriptorRatioGauge;
import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import kafka.admin.AdminUtils;
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.OffsetRequest;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;
import kafka.utils.ZKStringSerializer$;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.io.FileUtils;
import org.apache.eagle.alert.metric.entity.MetricEvent;
import org.apache.eagle.alert.metric.sink.ConsoleSink;
import org.apache.eagle.alert.metric.sink.Slf4jSink;
import org.apache.eagle.alert.metric.source.JVMMetricSource;
import org.apache.eagle.alert.metric.source.MetricSource;
import org.apache.eagle.alert.utils.JsonUtils;
import org.apache.eagle.alert.utils.KafkaEmbedded;
import org.junit.*;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Scanner;

public class MetricSystemTest {

    public static final String END_LINE = System.getProperty("line.separator");
    @Rule
    public TemporaryFolder tempFolder = new TemporaryFolder();
    private static final int DATA_BEGIN_INDEX = 55;
    private static final String TOPIC = "alert_metric_test";
    private static final String clientName = "test";

    @Test
    public void testMetricEvent() {
        MetricEvent metricEvent = MetricEvent.of("test").build();
        Assert.assertEquals(metricEvent.get("name"), "test");
        Assert.assertNotNull(metricEvent.get("timestamp"));

        metricEvent = MetricEvent.of("test1").build();
        metricEvent.put("timestamp", 1);
        Assert.assertEquals(metricEvent.get("name"), "test1");
        Assert.assertEquals(metricEvent.get("timestamp"), 1);

        Counter counter = new Counter();
        counter.inc(10);
        metricEvent = MetricEvent.of("testcount").from(counter).build();
        Assert.assertEquals(metricEvent.get("count"), 10l);

        Gauge gauge = Mockito.mock(FileDescriptorRatioGauge.class);
        Mockito.when(gauge.getValue()).thenReturn(new Double("0.4"));
        metricEvent = MetricEvent.of("testGauge").from(gauge).build();
        Assert.assertEquals(metricEvent.get("value"), 0.4);

        //Histogram
        Histogram histogram = Mockito.mock(Histogram.class);
        Snapshot snapshot = Mockito.mock(Snapshot.class);
        Mockito.when(histogram.getCount()).thenReturn(11l);
        Mockito.when(histogram.getSnapshot()).thenReturn(snapshot);
        Mockito.when(snapshot.getMin()).thenReturn(1l);
        Mockito.when(snapshot.getMax()).thenReturn(2l);
        Mockito.when(snapshot.getMean()).thenReturn(3d);
        Mockito.when(snapshot.getStdDev()).thenReturn(4d);
        Mockito.when(snapshot.getMedian()).thenReturn(5d);
        Mockito.when(snapshot.get75thPercentile()).thenReturn(6d);
        Mockito.when(snapshot.get95thPercentile()).thenReturn(7d);
        Mockito.when(snapshot.get98thPercentile()).thenReturn(8d);
        Mockito.when(snapshot.get99thPercentile()).thenReturn(9d);
        Mockito.when(snapshot.get999thPercentile()).thenReturn(10d);
        metricEvent = MetricEvent.of("testHistogram").from(histogram).build();

        Assert.assertEquals(metricEvent.get("count"), 11l);
        Assert.assertEquals(metricEvent.get("min"), 1l);
        Assert.assertEquals(metricEvent.get("max"), 2l);
        Assert.assertEquals(metricEvent.get("mean"), 3d);
        Assert.assertEquals(metricEvent.get("stddev"), 4d);
        Assert.assertEquals(metricEvent.get("median"), 5d);
        Assert.assertEquals(metricEvent.get("75thPercentile"), 6d);
        Assert.assertEquals(metricEvent.get("95thPercentile"), 7d);
        Assert.assertEquals(metricEvent.get("98thPercentile"), 8d);
        Assert.assertEquals(metricEvent.get("99thPercentile"), 9d);
        Assert.assertEquals(metricEvent.get("999thPercentile"), 10d);

        //Meter
        Meter meter = Mockito.mock(Meter.class);
        Mockito.when(meter.getCount()).thenReturn(1l);
        Mockito.when(meter.getOneMinuteRate()).thenReturn(2d);
        Mockito.when(meter.getFiveMinuteRate()).thenReturn(3d);
        Mockito.when(meter.getFifteenMinuteRate()).thenReturn(4d);
        Mockito.when(meter.getMeanRate()).thenReturn(5d);
        metricEvent = MetricEvent.of("testMeter").from(meter).build();

        Assert.assertEquals(metricEvent.get("value"), 1l);
        Assert.assertEquals(metricEvent.get("1MinRate"), 2d);
        Assert.assertEquals(metricEvent.get("5MinRate"), 3d);
        Assert.assertEquals(metricEvent.get("15MinRate"), 4d);
        Assert.assertEquals(metricEvent.get("mean"), 5d);

        //Timer
        Timer value = Mockito.mock(Timer.class);
        Mockito.when(value.getCount()).thenReturn(1l);
        Mockito.when(value.getOneMinuteRate()).thenReturn(2d);
        Mockito.when(value.getFiveMinuteRate()).thenReturn(3d);
        Mockito.when(value.getFifteenMinuteRate()).thenReturn(4d);
        Mockito.when(value.getMeanRate()).thenReturn(5d);
        metricEvent = MetricEvent.of("testTimer").from(value).build();

        Assert.assertEquals(metricEvent.get("value"), 1l);
        Assert.assertEquals(metricEvent.get("1MinRate"), 2d);
        Assert.assertEquals(metricEvent.get("5MinRate"), 3d);
        Assert.assertEquals(metricEvent.get("15MinRate"), 4d);
        Assert.assertEquals(metricEvent.get("mean"), 5d);

    }

    @Test
    public void testMerticSystemWithKafkaSink() throws IOException {

        JVMMetricSource jvmMetricSource = mockMetricRegistry();
        //setup kafka
        KafkaEmbedded kafkaEmbedded = new KafkaEmbedded();
        makeSureTopic(kafkaEmbedded.getZkConnectionString());
        //setup metric system
        File file = genKafkaSinkConfig(kafkaEmbedded.getBrokerConnectionString());
        Config config = ConfigFactory.parseFile(file);
        MetricSystem system = MetricSystem.load(config);
        system.register(jvmMetricSource);
        system.start();
        system.report();

        SimpleConsumer consumer = assertMsgFromKafka(kafkaEmbedded);
        system.stop();
        consumer.close();
        kafkaEmbedded.shutdown();
    }

    @Test
    public void testConsoleSink() throws IOException {
        PrintStream console = System.out;
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        System.setOut(new PrintStream(bytes));

        ConsoleSink sink = new ConsoleSink();
        MetricRegistry registry = new MetricRegistry();
        JvmAttributeGaugeSet jvm = Mockito.mock(JvmAttributeGaugeSet.class);
        Map<String, Metric> metrics = new HashMap<>();
        metrics.put("name", (Gauge) () -> "testname");
        metrics.put("uptime", (Gauge) () -> "testuptime");
        metrics.put("vendor", (Gauge) () -> "testvendor");
        Mockito.when(jvm.getMetrics()).thenReturn(metrics);
        registry.registerAll(jvm);
        File file = genConsoleSinkConfig();
        Config config = ConfigFactory.parseFile(file);
        sink.prepare(config, registry);
        sink.report();
        sink.stop();
        String result = bytes.toString();
        result = result.substring(result.indexOf(END_LINE) + END_LINE.length());//remove first line
        Assert.assertEquals("" + END_LINE + "" +
                "-- Gauges ----------------------------------------------------------------------" + END_LINE + "" +
{'fraction_non_alphanumeric': 0.13523722314427974, 'fraction_numerical': 0.011045613231843415, 'mean_word_length': 3.6935267257587965, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4829920', 'n_tokens_mistral': 5100, 'n_tokens_neox': 4941, 'n_words': 897}