package com.yingxin.prms.service.asyncTask;

import com.yingxin.prms.config.Constant;
import com.yingxin.prms.domain.AlarmLog;
import com.yingxin.prms.domain.Server_Host_Info;
import com.yingxin.prms.service.dao.CassandraDaoImpl;
import com.yingxin.prms.utils.TimeUtil;
import org.kairosdb.client.HttpClient;
import org.kairosdb.client.builder.*;
import org.kairosdb.client.response.QueryResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.TimeZone;
import java.util.UUID;

@Component
public class AlarmTask {

    private static Logger logger = LoggerFactory.getLogger(AlarmTask.class);

    private final HttpClient client;


    @Autowired
    private CassandraDaoImpl cassandraDao;

    @Autowired
    public AlarmTask(HttpClient client) {
        this.client = client;
    }

    @Async("asyncTaskPool")
    public void doTask(Server_Host_Info hostInfo) {

        String tag = hostInfo.getHost_ip() + ":" + hostInfo.getHost_port();
        Aggregator aggregator5Min = AggregatorFactory.createSumAggregator(5, TimeUnit.MINUTES);
        Aggregator aggregator1Min = AggregatorFactory.createSumAggregator(1, TimeUnit.MINUTES);
        QueryBuilder query5Min = createQuery(5, TimeUnit.MINUTES, "heartbeat", tag, aggregator5Min);
        QueryBuilder query1Min = createQuery(1, TimeUnit.MINUTES, "heartbeat", tag, aggregator1Min);
        try {
            QueryResponse heartbeat5Min = client.query(query5Min);
            QueryResponse heartbeat1Min = client.query(query1Min);

            List<DataPoint> dataPoints5Min = heartbeat5Min.getQueries().get(0).getResults().get(0).getDataPoints();
            List<DataPoint> dataPoints1Min = heartbeat1Min.getQueries().get(0).getResults().get(0).getDataPoints();
            long faultCount5Min = 0;
            long faultCount1Min = 0;
            if(dataPoints5Min.size() != 0){
                faultCount5Min = dataPoints5Min.get(0).longValue();
            }
            if (dataPoints1Min.size() != 0) {
                faultCount1Min = dataPoints1Min.get(0).longValue();
            }
            if (dataPoints5Min.size() > 1) {
                faultCount5Min = dataPoints5Min.get(dataPoints5Min.size() - 1).longValue();
            }
            if (dataPoints1Min.size() > 1) {
                faultCount1Min = dataPoints1Min.get(dataPoints1Min.size() - 1).longValue() + dataPoints1Min.get(dataPoints1Min.size() - 2).longValue();
            }
            int correctCount1Min = 60 / 10 - 1;

            String uuid = UUID.randomUUID().toString();
            AlarmLog alarmLog  = new AlarmLog();
            alarmLog.setSerialnum(uuid);
            alarmLog.setClusterName(hostInfo.getCluster_name());
            alarmLog.setIp(hostInfo.getHost_ip());
            alarmLog.setPort(String.valueOf(hostInfo.getHost_port()));
            AlarmLog alarmCheck = cassandraDao.checkAlarm(hostInfo.getCluster_name(),hostInfo.getHost_ip(), String.valueOf(hostInfo.getHost_port()));
            String msg = "";
            boolean flag = true;
            if (faultCount5Min != 0) {
                /* 不稳定告警 */
                alarmLog.setAlatime(TimeUtil.getDate());
                alarmLog.setState(Constant.ALARM_ING);
                alarmLog.setAlarmLevel(Constant.ALARM_UNSTABLE);

                msg ="系统不稳定"+" " + hostInfo.getCluster_name() + " " + hostInfo.getHost_name()
                        + " " + hostInfo.getHost_ip() + " 5分钟内服务存在调用失败";
                alarmLog.setAlarmMessage(msg);
                flag = false;

            }
            if(faultCount1Min >= correctCount1Min) {
                /* 宕机告警 */
                alarmLog.setAlatime(TimeUtil.getDate());
                alarmLog.setState(Constant.ALARM_ING);
                alarmLog.setAlarmLevel(Constant.ALARM_DOWN);

                msg ="系统宕机"+ " " + hostInfo.getCluster_name() + " " + hostInfo.getHost_name()
                        + " " + hostInfo.getHost_ip() + " 连续1分钟服务调用失败";
                alarmLog.setAlarmMessage(msg);
                flag = false;
            }
            if(flag){
                if(alarmCheck != null){
                    /* 如果存在告警  则恢复 */
                    cassandraDao.updateAlarm(alarmCheck.getSerialnum(),TimeUtil.getDate());
                }
            }else{
                if(alarmCheck == null){
                    /* 如果不存在告警  新增数据 */
                    cassandraDao.insertAlarm(alarmLog);
                }else {
                    /* 如果告警发生变化 回复时间变成告警时间 */
                   if(!alarmCheck.getAlarmLevel().equals(alarmLog.getAlarmLevel())) {
                       cassandraDao.updateAlarm(alarmCheck.getSerialnum(),"------");
                       cassandraDao.insertAlarm(alarmLog);
                    }
                }
            }
        } catch (IOException | DataFormatException e) {
            logger.error("告警校验发生错误。inMonitoring返回false");
            logger.error(e.getMessage());
        }
    }

    private QueryBuilder createQuery(int start, TimeUnit timeUnit, String metric, String tag, Aggregator aggregator) {
        QueryBuilder queryBuilder = QueryBuilder.getInstance();
        queryBuilder.setStart(start, timeUnit).setTimeZone(TimeZone.getDefault())
                .addMetric(metric)
                .addTag("host", tag)
                .addAggregator(aggregator);
        return queryBuilder;
    }

}
