/**
 * Copyright 2016 Pinterest, Inc.
 *
 * 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.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.pinterest.arcee.metrics;

import com.google.gson.GsonBuilder;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.URL;
import java.util.*;
import com.google.gson.reflect.TypeToken;
import com.pinterest.arcee.bean.MetricDatumBean;
import com.pinterest.deployservice.common.HTTPClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;

public class TSDBMetricSource implements MetricSource {
    private static final Logger LOG = LoggerFactory.getLogger(com.pinterest.arcee.metrics.TSDBMetricSource.class);
    private String writePath;
    private String readPath;
    private Integer port;

    public TSDBMetricSource(String urlpath, String readUrl) throws Exception {
        URL url = new URL(urlpath);
        writePath = url.getHost();
        port = url.getPort();
        readPath = readUrl;
        LOG.info(String.format("set up tsdb server as: %s : %d, read path: %s", writePath, port, readPath));
}

    @Override
    public void export(String metricName, Map<String, String> tags, Double value, Long timestamp) throws Exception {
        if (tags.isEmpty()) {
            tags.put("host", InetAddress.getLocalHost().getHostName());
        }
        StringBuilder sb = new StringBuilder();
        String prefix = "";
        for (Map.Entry<String, String> entry : tags.entrySet()) {
            sb.append(prefix);
            prefix = " ";
            sb.append(String.format("%s=%s", entry.getKey(), entry.getValue()));
        }

        byte[] message = String.format("put %s %f %f %s\n", metricName, (double)timestamp/1000, value, sb.toString()).getBytes();
        LOG.info("exporting metric to tsdb " + new String(message));
        sendMessage(message, 3);
    }

    @Override
    public Collection<MetricDatumBean> getMetrics(String metricName, String start) throws Exception {
        HTTPClient httpClient = new HTTPClient();
        Map<String, String> params = new HashMap<>();
        params.put("m", metricName);
        params.put("start", start);
        HashMap<String, String> headers = new HashMap<>();

        String url = String.format("%s/api/query", readPath);
        String jsonPayload = httpClient.get(url, params, headers, 1);
        LOG.debug(String.format("Get metric data from source: %s", jsonPayload));

        GsonBuilder builder = new GsonBuilder();
        Map<String, Object>[] jsonMaps = builder.create().fromJson(jsonPayload, new TypeToken<HashMap<String, Object>[]>() {
        }.getType());

        List<MetricDatumBean> dataPoints = new ArrayList<>();
        for (Map<String, Object> jsonMap : jsonMaps) {
            Map<String, Double> metrics = (Map<String, Double>) jsonMap.get("dps");
            if (metrics != null) {
                for (Map.Entry<String, Double> metric : metrics.entrySet()) {
                    MetricDatumBean dataPoint = new MetricDatumBean();
                    dataPoint.setTimestamp(1000 * Long.parseLong(metric.getKey()));
                    double value = metric.getValue() == Double.MIN_VALUE ? 0 : metric.getValue();
                    dataPoint.setValue(value);
                    dataPoints.add(dataPoint);
                }
            }
        }
        return dataPoints;
    }

    private boolean sendMessage(byte[] message, int times) throws Exception {
        Exception lastException = null;
        for (int i = 0; i < times; ++i) {
            DatagramSocket socket = new DatagramSocket();
            try {
                InetAddress address = InetAddress.getByName(writePath);
                DatagramPacket packet = new DatagramPacket(message, message.length, address, port);
                socket.send(packet);
                socket.close();
                return true;
            } catch (Exception ex) {
                LOG.error("Failed to send message to tsd server", ex);
                lastException = ex;
            } finally {
                if (!socket.isClosed()) {
                    socket.close();
                }
            }
        }

        throw lastException;
    }
}

@@ -674,16 +674,17 @@
 ilder;%0A%0A
+%0A
 import j
@@ -695,75 +695,134 @@
 net.
-DatagramPacket;%0Aimport java.net.DatagramSocket;%0Aimport java.net.URL
+InetAddress;%0Aimport java.net.InetSocketAddress;%0Aimport java.net.URL;%0Aimport java.net.Socket;%0Aimport java.io.OutputStreamWriter
 ;%0Aim
@@ -1046,37 +1046,8 @@
 y;%0A%0A
-import java.net.InetAddress;%0A
 %0Apub
@@ -2162,22 +2162,22 @@
         
-byte%5B%5D
+String
  message
@@ -2202,17 +2202,17 @@
 put %25s %25
-f
+d
  %25f %25s%5Cn
@@ -2230,16 +2230,8 @@
 me, 
-(double)
 time
@@ -2239,13 +2239,8 @@
 tamp
-/1000
 , va
@@ -2262,19 +2262,8 @@
 g())
-.getBytes()
 ;%0A  
@@ -2306,32 +2306,23 @@
 tsdb
+: 
  %22 + 
-new String(
 message)
 );%0A 
@@ -2317,17 +2317,16 @@
 message)
-)
 ;%0A      
@@ -3897,14 +3897,14 @@
 age(
-byte%5B%5D
+String
  mes
@@ -4033,24 +4033,16 @@
         
-Datagram
 Socket s
@@ -4053,24 +4053,16 @@
 t = new 
-Datagram
 Socket()
@@ -4101,51 +4101,44 @@
     
-InetAddress address = InetAddress.getByName
+socket.connect(new InetSocketAddress
 (wri
@@ -4139,24 +4139,31 @@
 ss(writePath
+, port)
 );%0A         
@@ -4173,126 +4173,122 @@
     
-DatagramPacket packet = new DatagramPacket(message, message.length, address, port);%0A                socket.send(packet
+OutputStreamWriter writer = new OutputStreamWriter(socket.getOutputStream());%0A                writer.write(message
 );%0A 
@@ -4298,36 +4298,36 @@
             
-socket.close
+writer.flush
 ();%0A        

Update on metrics write path