/**
 * 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.hadoop.metrics2.sink.timeline;

import junit.framework.Assert;
import org.apache.hadoop.metrics2.sink.timeline.availability.MetricCollectorUnavailableException;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.easymock.PowerMock;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;

import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.TreeMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.easymock.EasyMock.anyString;
import static org.easymock.EasyMock.expect;
import static org.powermock.api.easymock.PowerMock.expectNew;
import static org.powermock.api.easymock.PowerMock.replayAll;
import static org.powermock.api.easymock.PowerMock.verifyAll;

@RunWith(PowerMockRunner.class)
@PrepareForTest({AbstractTimelineMetricsSink.class, HttpURLConnection.class})
public class AbstractTimelineMetricSinkTest {

  @Test
  public void testParseHostsStringIntoCollection() {
    AbstractTimelineMetricsSink sink = new TestTimelineMetricsSink();
    Collection<String> hosts;

    hosts = sink.parseHostsStringIntoCollection("");
    Assert.assertTrue(hosts.isEmpty());

    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local");
    Assert.assertTrue(hosts.size() == 1);
    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));

    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local ");
    Assert.assertTrue(hosts.size() == 1);
    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));

    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local,test1.456.abc.def.local");
    Assert.assertTrue(hosts.size() == 2);

    hosts = sink.parseHostsStringIntoCollection("test1.123.abc.def.local, test1.456.abc.def.local");
    Assert.assertTrue(hosts.size() == 2);
    Assert.assertTrue(hosts.contains("test1.123.abc.def.local"));
    Assert.assertTrue(hosts.contains("test1.456.abc.def.local"));
  }

  @Test
  @PrepareForTest({URL.class, OutputStream.class, AbstractTimelineMetricsSink.class, HttpURLConnection.class, TimelineMetric.class})
  public void testEmitMetrics() throws Exception {
    HttpURLConnection connection = PowerMock.createNiceMock(HttpURLConnection.class);
    URL url = PowerMock.createNiceMock(URL.class);
    expectNew(URL.class, anyString()).andReturn(url).anyTimes();
    expect(url.openConnection()).andReturn(connection).anyTimes();
    expect(connection.getResponseCode()).andReturn(200).anyTimes();
    OutputStream os = PowerMock.createNiceMock(OutputStream.class);
    expect(connection.getOutputStream()).andReturn(os).anyTimes();


    TestTimelineMetricsSink sink = new TestTimelineMetricsSink();
    TimelineMetrics timelineMetrics = new TimelineMetrics();
    long startTime = System.currentTimeMillis() / 60000 * 60000;

    long seconds = 1000;
    TreeMap<Long, Double> metricValues = new TreeMap<>();
    /*

    0        +30s      +60s
    |         |         |
      (1)(2)(3) (4)(5)   (6)  m1

    */
    // (6) should be cached, the rest - posted

    metricValues.put(startTime + 4*seconds, 1.0);
    metricValues.put(startTime + 14*seconds, 2.0);
    metricValues.put(startTime + 24*seconds, 3.0);
    metricValues.put(startTime + 34*seconds, 4.0);
    metricValues.put(startTime + 44*seconds, 5.0);
    metricValues.put(startTime + 64*seconds, 6.0);

    TimelineMetric timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
    timelineMetric.setStartTime(metricValues.firstKey());
    timelineMetric.addMetricValues(metricValues);

    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);

    replayAll();
    sink.emitMetrics(timelineMetrics);
    Assert.assertEquals(1, sink.getMetricsPostCache().size());
    metricValues = new TreeMap<>();
    metricValues.put(startTime + 64*seconds, 6.0);
    Assert.assertEquals(metricValues, sink.getMetricsPostCache().getIfPresent("metric1").getMetricValues());

    timelineMetrics = new TimelineMetrics();
    metricValues = new TreeMap<>();
    /*

    +60      +90s     +120s     +150s     +180s
    |         |         |         |         |
       (7)      (8)       (9)           (10)   (11)   m1

    */
    // (6) from previous post should be merged with current data
    // (6),(7),(8),(9),(10) - should be posted, (11) - cached
    metricValues.put(startTime + 74*seconds, 7.0);
    metricValues.put(startTime + 94*seconds, 8.0);
    metricValues.put(startTime + 124*seconds, 9.0);
    metricValues.put(startTime + 154*seconds, 10.0);
    metricValues.put(startTime + 184*seconds, 11.0);

    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
    timelineMetric.setStartTime(metricValues.firstKey());
    timelineMetric.addMetricValues(metricValues);

    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);
    sink.emitMetrics(timelineMetrics);

    Assert.assertEquals(1, sink.getMetricsPostCache().size());
    metricValues = new TreeMap<>();
    metricValues.put(startTime + 184*seconds, 11.0);
    Assert.assertEquals(metricValues, sink.getMetricsPostCache().getIfPresent("metric1").getMetricValues());timelineMetrics = new TimelineMetrics();

    metricValues = new TreeMap<>();
    /*

    +180s   +210s   +240s
    |         |       |
       (12)        (13)

    */
    // (11) from previous post should be merged with current data
    // (11),(12),(13) - should be posted, cache should be empty
    metricValues.put(startTime + 194*seconds, 12.0);
    metricValues.put(startTime + 239*seconds, 13.0);

    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
    timelineMetric.setStartTime(metricValues.firstKey());
    timelineMetric.addMetricValues(metricValues);

    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);
    sink.emitMetrics(timelineMetrics);

    Assert.assertEquals(0, sink.getMetricsPostCache().size());

    metricValues = new TreeMap<>();
    /*

    +240s   +270s   +300s   +330s
    |         |       |       |
       (14)        (15)   (16)

    */
    // since postAllCachedMetrics in emitMetrics call is true (14),(15),(16) - should be posted, cache should be empty
    metricValues.put(startTime + 245*seconds, 14.0);
    metricValues.put(startTime + 294*seconds, 15.0);
    metricValues.put(startTime + 315*seconds, 16.0);

    timelineMetric = new TimelineMetric("metric1", "host1", "app1", "instance1");
    timelineMetric.setStartTime(metricValues.firstKey());
    timelineMetric.addMetricValues(metricValues);

    timelineMetrics.addOrMergeTimelineMetric(timelineMetric);
    sink.emitMetrics(timelineMetrics, true);

    Assert.assertEquals(0, sink.getMetricsPostCache().size());
  }

  @Test(expected = MetricCollectorUnavailableException.class)
  @PrepareForTest({URL.class, AbstractTimelineMetricsSink.class, HttpURLConnection.class, TimelineMetric.class})
  public void testFindLiveCollectorHostsFromKnownCollector() throws Exception {
    HttpURLConnection connection = PowerMock.createNiceMock(HttpURLConnection.class);
    URL url = PowerMock.createNiceMock(URL.class);
    expectNew(URL.class, anyString()).andReturn(url).anyTimes();
    expect(url.openConnection()).andReturn(connection).anyTimes();
    expect(connection.getResponseCode()).andReturn(500).anyTimes();
    replayAll();

    TestTimelineMetricsSink sink = new TestTimelineMetricsSink();
    sink.findLiveCollectorHostsFromKnownCollector("host", "1234");

    verifyAll();
  }

  @Test
  public void testGetCurrentCollectorHostMultiThreaded() {
    final int threadPoolSize = 32;
    final int numberOfThreads = threadPoolSize * 8;
    final AtomicBoolean stop = new AtomicBoolean(false);
    final TestTimelineMetricsThreadingSink sink = new TestTimelineMetricsThreadingSink();
    final ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);
    int threadCount = 0;

    try {
      while (!stop.get() && threadCount < numberOfThreads) {
        executorService.execute(() -> {
          try {
            String host = sink.getCurrentCollectorHost();
          } catch (Exception e) {
            e.printStackTrace();
            stop.set(true);
            executorService.shutdownNow();
          }
        });

        threadCount++;
      }

      executorService.awaitTermination(30, TimeUnit.SECONDS);
      if(stop.get()) {
        Assert.fail("Unexpected exception(s) has been thrown! See the stack traces above!");
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      executorService.shutdown();
    }
  }

  private class TestTimelineMetricsSink extends AbstractTimelineMetricsSink {
    @Override
    protected String getCollectorUri(String host) {
      return "";
    }

    @Override
    protected String getCollectorProtocol() {
      return "http";
    }

    @Override
    protected String getCollectorPort() {
      return "2181";
    }

    @Override
    protected int getTimeoutSeconds() {
      return 10;
    }

    @Override
    protected String getZookeeperQuorum() {
      return "localhost:2181";
    }

    @Override
    protected Collection<String> getConfiguredCollectorHosts() {
      return Arrays.asList("localhost");
    }

    @Override
    protected String getHostname() {
      return "h1";
    }

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

    @Override
    protected int getHostInMemoryAggregationPort() {
      return 61888;
    }

    @Override
    protected String getHostInMemoryAggregationProtocol() {
      return "http";
    }
  }

  private class TestTimelineMetricsThreadingSink extends TestTimelineMetricsSink {
    private final AtomicInteger counter = new AtomicInteger(0);
    private final Collection<String> hosts = Collections.singletonList("host1");

    @Override
    protected Collection<String> findLiveCollectorHostsFromKnownCollector(String host, String port) throws MetricCollectorUnavailableException {
      failedCollectorConnectionsCounter.set(4);
      int c = counter.getAndIncrement();
      if (c % 2 == 0 || c % 3 == 0) {
        throw new MetricCollectorUnavailableException("MetricCollectorUnavailable");
      }
      return hosts;
    }

    @Override
    protected Collection<String> getConfiguredCollectorHosts() {
      return hosts;
    }

    @Override
    protected SupplierExpiry getSupplierExpiry() {
      return new SupplierExpiry(128, TimeUnit.MILLISECONDS);
    }
  }
}
