package com.mesosphere.metrics.consumer;

import java.io.IOException;
import java.net.URISyntaxException;
import org.kairosdb.client.HttpClient;
import org.kairosdb.client.builder.Metric;
import org.kairosdb.client.builder.MetricBuilder;

import com.mesosphere.metrics.consumer.common.ConsumerRunner;
import com.mesosphere.metrics.consumer.common.MetricOutput;

import dcos.metrics.Datapoint;
import dcos.metrics.MetricList;
import dcos.metrics.Tag;

public class KairosMain {
  private static class KairosOutput implements MetricOutput {
    private final HttpClient client;
    private final boolean exitOnConnectFailure;
    private MetricBuilder metricBuilder;

    public KairosOutput(String host, int port, boolean exitOnConnectFailure)
        throws IllegalStateException, IOException {
      client = new HttpClient(String.format("http://%s:%s", host, port));
      this.exitOnConnectFailure = exitOnConnectFailure;
      metricBuilder = MetricBuilder.getInstance();
    }

    @Override
    public void append(MetricList list) {
      // note: we COULD optimize here by grouping the datapoints by metric name across multiple appends (ie Map<String, Metric>).
      // however, we would also need to ensure that we don't have mismatched tags within these groupings.
      // and in practice I suspect we won't see a ton of duplicate metric names between flush()es anyway
      for (Datapoint d : list.getDatapoints()) {
        Metric metric = metricBuilder.addMetric(d.getName()).addDataPoint(d.getValue());
        for (Tag t : list.getTags()) {
          metric.addTag(t.getKey(), t.getValue());
        }
      }
    }

    @Override
    public void flush() {
      if (metricBuilder.getMetrics().isEmpty()) {
        return;
      }
      try {
        client.pushMetrics(metricBuilder);
      } catch (URISyntaxException | IOException e) {
        e.printStackTrace();
        if (exitOnConnectFailure) {
          System.exit(1);
        }
      }
      metricBuilder = MetricBuilder.getInstance();
    }
  }

  private static boolean parseBool(String str, boolean defaultVal) {
    if (str == null || str.isEmpty()) {
      return defaultVal;
    }
    switch (str.charAt(0)) {
    case 't':
    case 'T':
    case '1':
      return true;
    default:
      return false;
    }
  }

  public static void main(String[] args) throws IOException {
    String hostFlag = System.getenv("OUTPUT_HOST");
    String portFlag = System.getenv("OUTPUT_PORT");
    if (hostFlag == null || portFlag == null) {
      throw new IllegalArgumentException("OUTPUT_HOST and OUTPUT_PORT are required");
    }
    boolean exitOnConnectFailure = parseBool(System.getenv("EXIT_ON_CONNECT_FAILURE"), true);
    ConsumerRunner.run(new KairosOutput(
        hostFlag, Integer.parseInt(portFlag), exitOnConnectFailure));
  }
}

@@ -230,16 +230,73 @@
 Builder;
+%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0Aimport
@@ -589,16 +589,103 @@
 utput %7B%0A
+    private static final Logger LOGGER = LoggerFactory.getLogger(KairosOutput.class);%0A%0A
     priv
@@ -1562,24 +1562,221 @@
 points()) %7B%0A
+        Double val = d.getValue();%0A        if (Double.isNaN(val)) %7B%0A          LOGGER.warn(%22Skipping value %7B%7D = NaN, it won't encode for Kairos anyway.%22, d.getName());%0A          continue;%0A        %7D%0A
         Metr
@@ -2093,16 +2093,106 @@
   try %7B%0A
+        LOGGER.info(%22Writing %7B%7D metrics to KairosDB%22, metricBuilder.getMetrics().size());%0A
         
@@ -2245,40 +2245,53 @@
 ch (
-URISyntaxException %7C IOE
+Throwable e) %7B // may throw runtime e
 xception
  e) 
@@ -2286,21 +2286,16 @@
 xception
- e) %7B
 %0A       

Add explicit check/filter for NaN values