package com.tscloud.agent.flume.source.dataprovider;

/**
 * 通過HTTP作為HDFS Master監控資訊的source
 * 明細可參考http://zookeeper.apache.org/doc/r3.4.6/zookeeperAdmin.html#sc_zkCommands
 * @author aihua.sun
 * @date 2015/4/6
 * @since V1.0
 */

import com.eric.agent.flume.model.ZookeeperRoleInfo;
import com.eric.agent.flume.source.base.IClusterServiceRoleDataProvider;
import com.eric.agent.utils.AgentConstants;
import com.eric.agent.utils.TelnetTools;
import com.eric.common.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.*;

public class ZookeeperDataProvider{
    protected final Logger LOGGER = LoggerFactory.getLogger(getClass());
    private static final String zk_avg_latency = "zk_avg_latency";
    private static final String zk_max_latency = "zk_max_latency";
    private static final String zk_min_latency = "zk_min_latency";
    private static final String zk_packets_received = "zk_packets_received";
    private static final String zk_packets_sent = "zk_packets_sent";
    private static final String zk_server_state = "zk_server_state";
    private static final String zk_znode_count = "zk_znode_count";
    private static final String zk_followers = "zk_followers";
    private static final String zk_open_file_descriptor_count = "zk_open_file_descriptor_count";

    public String extractMonitorData() {
        //TODO 通過呼叫API獲得IP以及引數
        ZookeeperRoleInfo monitorDataPoint = new ZookeeperRoleInfo();
        String IP = "192.168.40.242";
        int port = 2181;
        TelnetTools telnet=null;
        try {
            telnet = new TelnetTools();
            telnet.login(IP, port);
            String rs = telnet.sendCommand("mntr");
            Map<String, String> telnetResultMap = parseTelnetResult(rs);
            monitorDataPoint.setZkAvgLatency(translateStrToLong(telnetResultMap.get(zk_avg_latency)));
            monitorDataPoint.setZkMaxLatency(translateStrToLong(telnetResultMap.get(zk_max_latency)));
            monitorDataPoint.setZkMinLatency(translateStrToLong(telnetResultMap.get(zk_min_latency)));
            monitorDataPoint.setZkPacketsReceived(translateStrToLong(telnetResultMap.get(zk_packets_received)));
            monitorDataPoint.setZkPacketsSent(translateStrToLong(telnetResultMap.get(zk_packets_sent)));
            monitorDataPoint.setZkServerState(telnetResultMap.get(zk_server_state));
            monitorDataPoint.setZkZnodeCount(translateStrToLong(telnetResultMap.get(zk_znode_count)));
            monitorDataPoint.setZkFollowers(translateStrToLong(telnetResultMap.get(zk_followers)));
            monitorDataPoint.setZkOpenFileDescriptorCount(translateStrToLong(telnetResultMap.get(zk_open_file_descriptor_count)));
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            telnet.distinct();
        }

        return monitorDataPoint.toString();
    }

    private Long translateStrToLong(String value) {
        if (org.apache.commons.lang.StringUtils.isAlphanumeric(value)) {
            return Long.valueOf(value);
        }
        return 0L;
    }

    private Map<String, String> parseTelnetResult(String rs) {
        //The output contains multiple lines with the following format:
        //key \t value
        String[] resultArray = rs.split("\n");
        Map<String, String> resultMap = new HashMap<String, String>();
        for (String recordLine : resultArray) {
            String[] recordKeyValue = recordLine.split("\t");
            LOGGER.debug("############recordKeyValue.size:" + recordKeyValue.length + " recordKeyValue:" + Arrays.toString(recordKeyValue));
            if (recordKeyValue != null && recordKeyValue.length == 2) {
                resultMap.put(recordKeyValue[0], recordKeyValue[1]);
            }
        }
        return resultMap;
    }


    public static void main(String[] args) {
        System.out.println(new ZookeeperDataProvider().extractMonitorData());
    }

}