public class Constants {
    private Constants() {
    }

    ;

    public static final int ALL_SLOTS_SIZE = 16384;

    public static final String ERR_CODE_INVALID_PARAMETER = "400_000";
    public static final String ERR_CODE_REDIS_SET_FAILED = "500_000";
    public static final String ERR_CODE_ALL_NODE_DOWN = "500_001";
    public static final String ERR_CODE_CLUSTER_NOT_AGREE_CONFIG = "500_002";
    public static final String ERR_CODE_CLUSTER_HAS_OPEN_SLOTS = "500_003";
    public static final String ERR_CODE_CLUSTER_HAS_ERRORS = "500_004";
    public static final String ERR_CODE_UNKNOWN = "500_999";

    public static String getUsersRedisKey(String prefixKey) {
        return Joiner.on(".").join(prefixKey, "users").toString();
    }

    public static String getClustersRedisKey(String prefixKey) {
        return Joiner.on(".").join(prefixKey, "clusters").toString();
    }

    public static String getClusterRedisKey(String prefixKey, String clusterName) {
        return Joiner.on(".").join(prefixKey, "cluster", clusterName).toString();
    }

    public static String getClusterNoticeRedisKey(String prefixKey, String clusterName) {
        return Joiner.on(".").join(prefixKey, "cluster", clusterName, "notice").toString();
    }

    public static String getNodeRedisKey(String prefixKey, String clusterName, String nodeId) {
        return Joiner.on(".").join(prefixKey, "cluster", clusterName, "node", nodeId).toString();
    }

    public static String getNodeStaticsInfoRedisKey(String prefixKey, String clusterName, String nodeId) {
        return Joiner.on(".").join(prefixKey, "cluster", clusterName, "node", nodeId, "staticsInfo").toString();
    }

    public static String getClusterSlowLogRedisKey(String prefixKey, String clusterName) {
        return Joiner.on(".").join(prefixKey, "cluster", clusterName, "slowLog").toString();
    }
} @Data
public class SchedulerConfig {
    public static final String DEFAULT_REFRESH_CLUSTERS_INTERVAL_MILLIS = "120000";
    public static final String DEFAULT_COLLECT_STATICS_INFO_INTERVAL_MILLIS = "120000";
    public static final String DEFAULT_COLLECT_STATICS_INFO_MAX_COUNT = "1500";
    public static final String DEFAULT_COLLECT_SLOW_LOG_MAX_COUNT = "1500";

    private String refreshClustersIntervalMillis;

    private String collectStaticsInfoIntervalMillis;
    private String collectStaticsInfoMaxCount;
    private String collectSlowLogMaxCount;
} public class ApiException extends RuntimeException {
    private static final long serialVersionUID = 3285456520279227122L;

    @Getter
    @Setter
    private ErrorResponse errorResponse;

    @Getter
    @Setter
    private HttpStatus httpStatus;

    public ApiException(ErrorResponse errorResponse, HttpStatus httpStatus) {
        super(errorResponse.getError().getMessage());
        this.errorResponse = errorResponse;
        this.httpStatus = httpStatus;
    }

    public ApiException(String code, String message, HttpStatus httpStatus) {
        super(message);
        this.errorResponse = new ErrorResponse(code, message);
        this.httpStatus = httpStatus;
    }
} public enum NoticeOperator {
    EQ("eq", " == "),
    NE("ne", " != "),
    GT("gt", " > "),
    GE("ge", " >= "),
    LT("lt", " < "),
    LE("le", " <= ");

    @Getter
    private String value;
    @Getter
    private String label;

    NoticeOperator(String value, String label) {
        this.value = value;
        this.label = label;
    }

    public static NoticeOperator getNoticeOperator(String value) {
        return Arrays.stream(values())
                .filter(v -> StringUtils.equalsIgnoreCase(v.value, value))
                .findFirst()
                .orElse(null);
    }
} public enum NoticeType {
    CLUSTER_INFO("cluster_info"),
    NODE_INFO("node_info");

    @Getter
    private String value;

    NoticeType(String value) {
        this.value = value;
    }

    public static NoticeType getNoticeType(String value) {
        return Arrays.stream(values())
                .filter(v -> StringUtils.equalsIgnoreCase(v.value, value))
                .findFirst()
                .orElse(null);
    }
} public enum NoticeValueType {
    NUMBER("number"),
    STRING("string");

    @Getter
    private String value;

    NoticeValueType(String value) {
        this.value = value;
    }

    public static NoticeValueType getNoticeValueType(String value) {
        return Arrays.stream(values())
                .filter(v -> StringUtils.equalsIgnoreCase(v.value, value))
                .findFirst()
                .orElse(null);
    }
} @Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public static class ResultValue {
    private String nodeId;
    private String hostAndPort;
    private String value;
} public interface ClusterService {
    Set<String> getClusters();

    Cluster getCluster(String clusterName) throws IOException;

    Cluster getClusterByHostAndPort(String hostAndPort) throws IOException;

    boolean existsClusterName(String clusterName);

    void setCluster(String clusterName, String node) throws IOException;

    void changeClusterName(String clusterName, String newClusterName) throws IOException;

    Notice getClusterNotice(String clusterName) throws IOException;

    void setClusterNotice(String clusterName, Notice notice) throws IOException;

    void deleteCluster(String clusterName);

    void refreshClusters();

    ClusterNode getActiveClusterNode(String clusterName) throws IOException;

    ClusterNode getActiveClusterNodeWithExcludeNodeId(String clusterName, String nodeId) throws IOException;

    ClusterNode getActiveClusterNodeWithExcludeHostAndPort(String clusterName, String hostAndPort) throws IOException;

    Map<String, Map<String, List<List<Object>>>> getClusterStaticsInfoHistory(String clusterName, List<String> nodes,
                                                                              List<String> fields, long start, long end);

    PagerData<SlowLog> getClusterSlowLogHistory(String clusterName, long offset, long limit);
} public interface NodeService {
    Map<String, String> getStaticsInfo(ClusterNode clusterNode);

    Map<String, List<List<Object>>> getStaticsInfoHistory(String clusterName, String nodeId, List<String> fields,
                                                          long start, long end);

    List<SlowLog> getSlowLog(ClusterNode clusterNode);

    void shutdown(String hostAndPort);
} public interface NotifyService {
    void notify(Cluster cluster, Notice notice, List<NoticeJob> jobs);
}
import com.behase.relumin.Constants;
import com.behase.relumin.config.SchedulerConfig;
import com.behase.relumin.exception.ApiException;
import com.behase.relumin.model.*;
import com.behase.relumin.model.NoticeItem.NoticeOperator;
import com.behase.relumin.model.NoticeItem.NoticeType;
import com.behase.relumin.model.NoticeItem.NoticeValueType;
import com.behase.relumin.model.NoticeJob.ResultValue;
import com.behase.relumin.service.ClusterService;
import com.behase.relumin.service.NodeService;
import com.behase.relumin.service.NotifyService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.fluentd.logger.FluentLogger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
package com.behase.relumin.scheduler;



@Slf4j
@Component
@Profile(value = "!test")
public class NodeScheduler {
    @Autowired
    private ClusterService clusterService;

    @Autowired
    private NodeService nodeService;

    @Autowired
    private JedisPool datastoreJedisPool;

    @Autowired
    private NotifyService notifyService;

    @Autowired
    private ObjectMapper mapper;

    @Autowired
    private FluentLogger fluentLogger;

    @Value("${scheduler.collectStaticsInfoMaxCount:" + SchedulerConfig.DEFAULT_COLLECT_STATICS_INFO_MAX_COUNT +