/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.master.metrics;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import alluxio.AlluxioURI;
import alluxio.clock.SystemClock;
import alluxio.grpc.MetricType;
import alluxio.metrics.Metric;
import alluxio.metrics.MetricInfo;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;

import com.google.common.collect.Lists;
import org.junit.Before;
import org.junit.Test;

import java.util.List;

public class MetricsStoreTest {
  private MetricsStore mMetricStore;

  @Before
  public void before() {
    MetricsSystem.resetAllMetrics();
    mMetricStore = new MetricsStore(new SystemClock());
    mMetricStore.initMetricKeys();
  }

  @Test
  public void putWorkerMetrics() {
    String workerHost1 = "192_1_1_1";
    List<Metric> metrics1 = Lists.newArrayList(
        Metric.from(MetricKey.WORKER_BYTES_READ_REMOTE.getName() + "." + workerHost1,
            10, MetricType.COUNTER),
        Metric.from(MetricKey.WORKER_BYTES_READ_DOMAIN.getName() + "." + workerHost1,
            20, MetricType.COUNTER));
    mMetricStore.putWorkerMetrics(workerHost1, metrics1);

    String workerHost2 = "192_1_1_2";
    List<Metric> metrics2 = Lists.newArrayList(
        Metric.from(MetricKey.WORKER_BYTES_READ_REMOTE.getName() + "." + workerHost2,
            1, MetricType.COUNTER));
    mMetricStore.putWorkerMetrics(workerHost2, metrics2);
    assertEquals(11,
        MetricsSystem.counter(MetricKey.CLUSTER_BYTES_READ_REMOTE.getName()).getCount());
  }

  @Test
  public void putWorkerUfsMetrics() {
    String readBytes = MetricKey.WORKER_BYTES_READ_UFS.getName();
    String writtenBytes = MetricKey.WORKER_BYTES_WRITTEN_UFS.getName();

    String ufsName1 = MetricsSystem.escape(new AlluxioURI("/my/local/folder"));
    String readBytesUfs1 = Metric.getMetricNameWithTags(readBytes, MetricInfo.TAG_UFS, ufsName1);
    String writtenBytesUfs1 = Metric
        .getMetricNameWithTags(writtenBytes, MetricInfo.TAG_UFS, ufsName1);

    String ufsName2 = MetricsSystem.escape(new AlluxioURI("s3://my/s3/bucket/"));
    String readBytesUfs2 = Metric.getMetricNameWithTags(readBytes, MetricInfo.TAG_UFS, ufsName2);
    String writtenBytesUfs2 = Metric
        .getMetricNameWithTags(writtenBytes, MetricInfo.TAG_UFS, ufsName2);

    String host1 = "192_1_1_1";
    List<Metric> metrics1 = Lists.newArrayList(
        Metric.from(readBytesUfs1 + "." + host1, 10, MetricType.COUNTER),
        Metric.from(writtenBytesUfs1 + "." + host1, 20, MetricType.COUNTER),
        Metric.from(writtenBytesUfs2 + "." + host1, 7, MetricType.COUNTER));
    mMetricStore.putWorkerMetrics(host1, metrics1);

    String host2 = "192_1_1_2";
    List<Metric> metrics2 = Lists.newArrayList(
        Metric.from(readBytesUfs1 + "." + host2, 5, MetricType.COUNTER),
        Metric.from(writtenBytesUfs1 + "." + host2, 12, MetricType.COUNTER),
        Metric.from(readBytesUfs2 + "." + host2, 33, MetricType.COUNTER));
    mMetricStore.putWorkerMetrics(host2, metrics2);

    assertEquals(15, MetricsSystem.counter(
        Metric.getMetricNameWithTags(MetricKey.CLUSTER_BYTES_READ_UFS.getName(),
            MetricInfo.TAG_UFS, ufsName1)).getCount());
    assertEquals(33, MetricsSystem.counter(
        Metric.getMetricNameWithTags(MetricKey.CLUSTER_BYTES_READ_UFS.getName(),
            MetricInfo.TAG_UFS, ufsName2)).getCount());
    assertEquals(48,
        MetricsSystem.counter(MetricKey.CLUSTER_BYTES_READ_UFS_ALL.getName()).getCount());

    assertEquals(32, MetricsSystem.counter(
        Metric.getMetricNameWithTags(MetricKey.CLUSTER_BYTES_WRITTEN_UFS.getName(),
            MetricInfo.TAG_UFS, ufsName1)).getCount());
    assertEquals(7, MetricsSystem.counter(
        Metric.getMetricNameWithTags(MetricKey.CLUSTER_BYTES_WRITTEN_UFS.getName(),
            MetricInfo.TAG_UFS, ufsName2)).getCount());
    assertEquals(39,
        MetricsSystem.counter(MetricKey.CLUSTER_BYTES_WRITTEN_UFS_ALL.getName()).getCount());
  }

  @Test
  public void clearAndGetClearTime() throws Exception {
    long clearTime = mMetricStore.getLastClearTime();
    String workerHost1 = "192_1_1_1";
    List<Metric> metrics1 = Lists.newArrayList(
        Metric.from(MetricKey.WORKER_BYTES_WRITTEN_REMOTE.getName() + "." + workerHost1,
            10, MetricType.COUNTER),
        Metric.from(MetricKey.WORKER_BYTES_WRITTEN_DOMAIN.getName() + "." + workerHost1,
            20, MetricType.COUNTER));
    mMetricStore.putWorkerMetrics(workerHost1, metrics1);

    assertEquals(10,
        MetricsSystem.counter(MetricKey.CLUSTER_BYTES_WRITTEN_REMOTE.getName()).getCount());
    // Make sure the test does not execute too fast
    // so that new clear time will be definitely different from the previous one
    Thread.sleep(10);
    mMetricStore.clear();
    assertEquals(0,
        MetricsSystem.counter(MetricKey.CLUSTER_BYTES_WRITTEN_REMOTE.getName()).getCount());
    long newClearTime = mMetricStore.getLastClearTime();
    assertTrue(newClearTime > clearTime);
  }
}
