/*
 *
 * Copyright (C) 2014-2015 LinkedIn Corp. All rights reserved.
 *
 * Licensed 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.
 */

package gobblin.runtime.util;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.testng.Assert;
import org.testng.annotations.Test;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;

import gobblin.configuration.State;
import gobblin.metrics.GobblinMetricsRegistry;
import gobblin.metrics.MetricContext;
import gobblin.metrics.Tag;
import gobblin.runtime.JobState;

@Test(groups = {"gobblin.runtime"})
public class JobMetricsTest {

  @Test
  public void testJobMetricsGet() {
    String jobName = "testJob";
    String jobId = "job_123";

    JobState jobState = new JobState(jobName, jobId);
    JobMetrics jobMetrics = JobMetrics.get(jobState);

    Assert.assertNotNull(jobMetrics.getMetricContext());

    List<Tag<?>> tags = jobMetrics.getMetricContext().getTags();
    Map<String, ?> tagMap = jobMetrics.getMetricContext().getTagMap();
    String contextId = tagMap.get(MetricContext.METRIC_CONTEXT_ID_TAG_NAME).toString();

    Assert.assertEquals(tagMap.size(), 4);
    Assert.assertEquals(tagMap.get("jobId"), jobId);
    Assert.assertEquals(tagMap.get("jobName"), jobName);
    Assert.assertEquals(tagMap.get(MetricContext.METRIC_CONTEXT_ID_TAG_NAME), contextId);

    // should get the original jobMetrics, can check by the id
    JobMetrics jobMetrics1 = JobMetrics.get(jobName + "_", jobId);
    Assert.assertNotNull(jobMetrics1.getMetricContext());

    tagMap = jobMetrics1.getMetricContext().getTagMap();
    Assert.assertEquals(tags.size(), 4);
    Assert.assertEquals(tagMap.get(MetricContext.METRIC_CONTEXT_ID_TAG_NAME), contextId);

    // remove original jobMetrics, should create a new one
    GobblinMetricsRegistry.getInstance().remove(jobMetrics.getId());
    JobMetrics jobMetrics2 = JobMetrics.get(jobName + "_", jobId);
    Assert.assertNotNull(jobMetrics2.getMetricContext());

    tagMap = jobMetrics2.getMetricContext().getTagMap();
    Assert.assertEquals(tags.size(), 4);
    Assert.assertNotEquals(tagMap.get(MetricContext.METRIC_CONTEXT_ID_TAG_NAME), contextId);
  }

  @Test
  public void testClusterIdentifierTags() {
    try {

      String expectedClusterIdentifier  = InetAddress.getLocalHost().getHostName();
      JobMetrics jobMetrics = buildTestJobMetrics();

      Assert.assertNotNull(jobMetrics.getMetricContext());

      Map<String, ?> tagMap = jobMetrics.getMetricContext().getTagMap();
      Assert.assertEquals(tagMap.get("clusterIdentifier"), expectedClusterIdentifier);
    } catch (UnknownHostException e) {
      //Ignore test
    }
  }

  @Test
  public void testCustomTags() {

    Properties testProperties = new Properties();
    Tag<String> expectedPropertyTag = new Tag<String>("key1", "value1");

    JobMetrics.addCustomTagToProperties(testProperties, expectedPropertyTag);
    State testState = new State(testProperties);
    List<Tag<?>> tags = JobMetrics.getCustomTagsFromState(testState);

    Assert.assertEquals(Iterables.getFirst(tags, null), expectedPropertyTag);

    Tag<String> expectedStateTag = new Tag<String>("key2", "value2");
    JobMetrics.addCustomTagToState(testState, expectedStateTag);
    tags = JobMetrics.getCustomTagsFromState(testState);

    Assert.assertTrue(tags.containsAll(ImmutableList.of(expectedPropertyTag, expectedStateTag)));

  }

  private JobMetrics buildTestJobMetrics() {
    String jobName = "testJob";
    String jobId = "job_123";

    JobState jobState = new JobState(jobName, jobId);
    return JobMetrics.get(jobState);

  }
}

@@ -546,75 +546,8 @@
 l;%0A%0A
-import java.net.InetAddress;%0Aimport java.net.UnknownHostException;%0A
 impo
@@ -561,24 +561,24 @@
 .util.List;%0A
+
 import java.
@@ -1553,33 +1553,33 @@
 (tagMap.size(), 
-4
+3
 );%0A    Assert.as
@@ -2041,33 +2041,33 @@
 ls(tags.size(), 
-4
+3
 );%0A    Assert.as
@@ -2492,17 +2492,17 @@
 size(), 
-4
+3
 );%0A    A
@@ -2598,499 +2598,8 @@
  %7D%0A%0A
-  @Test%0A  public void testClusterIdentifierTags() %7B%0A    try %7B%0A%0A      String expectedClusterIdentifier  = InetAddress.getLocalHost().getHostName();%0A      JobMetrics jobMetrics = buildTestJobMetrics();%0A%0A      Assert.assertNotNull(jobMetrics.getMetricContext());%0A%0A      Map%3CString, ?%3E tagMap = jobMetrics.getMetricContext().getTagMap();%0A      Assert.assertEquals(tagMap.get(%22clusterIdentifier%22), expectedClusterIdentifier);%0A    %7D catch (UnknownHostException e) %7B%0A      //Ignore test%0A    %7D%0A  %7D%0A%0A
   @T
@@ -3338,210 +3338,6 @@
  %7D%0A%0A
-  private JobMetrics buildTestJobMetrics() %7B%0A    String jobName = %22testJob%22;%0A    String jobId = %22job_123%22;%0A%0A    JobState jobState = new JobState(jobName, jobId);%0A    return JobMetrics.get(jobState);%0A%0A  %7D%0A
 %7D%0A

Remove clusterIdentifier tests from JobMetrics. Custom tags test should suffice