beacon checker

core/src/main/java/com/ctrip/xpipe/api/migration/auto/MonitorService.java

@@ -2,6 +2,7 @@
 
 import com.ctrip.xpipe.api.migration.auto.data.MonitorGroupMeta;
 
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -27,6 +28,12 @@ public interface MonitorService {
 
     void registerCluster(String system, String clusterName, Set<MonitorGroupMeta> groups);
 
+    void updateCluster(String system, String clusterName, Set<MonitorGroupMeta> groups);
+
     void unregisterCluster(String system, String clusterName);
 
+    int getBeaconClusterHash(String system, String clusterName);
+
+    Map<String,Set<String>> getAllClusterWithDc(String system);
+
 }

core/src/main/java/com/ctrip/xpipe/api/migration/auto/data/MonitorClusterMeta.java

@@ -1,8 +1,8 @@
 package com.ctrip.xpipe.api.migration.auto.data;
 
-import java.util.Collections;
-import java.util.Map;
-import java.util.Set;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+
+import java.util.*;
 
 /**
  * @author lishanglin
@@ -39,4 +39,20 @@ public void setExtra(Map<String, String> extra) {
         this.extra = extra;
     }
 
+    public int generateHashCodeForBeaconCheck () {
+        List<MonitorGroupMeta> nodeList = new ArrayList<>(nodeGroups);
+        nodeList.sort(Comparator.comparing(MonitorGroupMeta::getName, Comparator.nullsFirst(Comparator.naturalOrder()))
+                .thenComparing(MonitorGroupMeta::getIdc, Comparator.nullsFirst(Comparator.naturalOrder()))
+                .thenComparing(MonitorGroupMeta::isMasterGroup)
+                .thenComparing(monitorGroupMeta -> monitorGroupMeta.getNodes().toString()));
+        HashCodeBuilder builder = new HashCodeBuilder();
+        for(MonitorGroupMeta group : nodeList) {
+            builder.append(group.getName())
+                    .append(group.getIdc())
+                    .append(group.isMasterGroup())
+                    .append(group.getNodes());
+        }
+        return builder.toHashCode();
+    }
+
 }

core/src/main/java/com/ctrip/xpipe/migration/auto/DefaultMonitorService.java

@@ -4,6 +4,7 @@
 import com.ctrip.xpipe.api.migration.auto.data.MonitorGroupMeta;
 
 import java.util.Collections;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -57,9 +58,24 @@ public void registerCluster(String system, String clusterName, Set<MonitorGroupM
         // do nothing
     }
 
+    @Override
+    public void updateCluster(String system, String clusterName, Set<MonitorGroupMeta> groups) {
+
+    }
+
     @Override
     public void unregisterCluster(String system, String clusterName) {
         // do nothing
     }
 
+    @Override
+    public int getBeaconClusterHash(String system, String clusterName) {
+        return 0;
+    }
+
+    @Override
+    public Map<String, Set<String>> getAllClusterWithDc(String system) {
+        return Collections.emptyMap();
+    }
+
 }

core/src/test/java/com/ctrip/xpipe/api/migration/MonitorClusterMetaTest.java

@@ -0,0 +1,101 @@
+package com.ctrip.xpipe.api.migration;
+
+import com.ctrip.xpipe.api.migration.auto.data.MonitorClusterMeta;
+import com.ctrip.xpipe.api.migration.auto.data.MonitorGroupMeta;
+import com.ctrip.xpipe.endpoint.HostPort;
+import com.google.common.collect.Sets;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Set;
+
+public class MonitorClusterMetaTest {
+
+    @Test
+    public void testMonitorClusterMeta(){
+        Set<MonitorGroupMeta> group1 = getMonitorGroupMeta1();
+        MonitorClusterMeta monitorClusterMeta1 = new MonitorClusterMeta(group1);
+
+        Set<MonitorGroupMeta> group2 = getMonitorGroupMeta2();
+        MonitorClusterMeta monitorClusterMeta2 = new MonitorClusterMeta(group2);
+
+        Set<MonitorGroupMeta> group3 = getMonitorGroupMeta3();
+        MonitorClusterMeta monitorClusterMeta3 = new MonitorClusterMeta(group3);
+
+        Set<MonitorGroupMeta> group4 = getMonitorGroupMeta4();
+        MonitorClusterMeta monitorClusterMeta4 = new MonitorClusterMeta(group4);
+
+        Set<MonitorGroupMeta> group5 = getMonitorGroupMeta5();
+        MonitorClusterMeta monitorClusterMeta5 = new MonitorClusterMeta(group5);
+
+        Set<MonitorGroupMeta> group6 = getMonitorGroupMeta6();
+        MonitorClusterMeta monitorClusterMeta6 = new MonitorClusterMeta(group6);
+
+        Assert.assertEquals(monitorClusterMeta1.generateHashCodeForBeaconCheck(), monitorClusterMeta2.generateHashCodeForBeaconCheck());
+        Assert.assertNotEquals(monitorClusterMeta1.generateHashCodeForBeaconCheck(), monitorClusterMeta3.generateHashCodeForBeaconCheck());
+        Assert.assertNotEquals(monitorClusterMeta1.generateHashCodeForBeaconCheck(), monitorClusterMeta4.generateHashCodeForBeaconCheck());
+        Assert.assertNotEquals(monitorClusterMeta4.generateHashCodeForBeaconCheck(), monitorClusterMeta5.generateHashCodeForBeaconCheck());
+        Assert.assertNotEquals(monitorClusterMeta5.generateHashCodeForBeaconCheck(), monitorClusterMeta6.generateHashCodeForBeaconCheck());
+        System.out.println(monitorClusterMeta1.generateHashCodeForBeaconCheck());
+        System.out.println(monitorClusterMeta2.generateHashCodeForBeaconCheck());
+        System.out.println(monitorClusterMeta3.generateHashCodeForBeaconCheck());
+        System.out.println(monitorClusterMeta4.generateHashCodeForBeaconCheck());
+        System.out.println(monitorClusterMeta5.generateHashCodeForBeaconCheck());
+        System.out.println(monitorClusterMeta6.generateHashCodeForBeaconCheck());
+    }
+
+
+    private Set<MonitorGroupMeta> getMonitorGroupMeta1() {
+        return Sets.newHashSet(
+                new MonitorGroupMeta("shard1+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6379"), HostPort.fromString("127.0.0.1:6380")), true),
+                new MonitorGroupMeta("shard2+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6381"), HostPort.fromString("127.0.0.1:6382")), true),
+                new MonitorGroupMeta("shard1+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6383"), HostPort.fromString("127.0.0.1:6384")), false),
+                new MonitorGroupMeta("shard2+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6385"), HostPort.fromString("127.0.0.1:6386")), false)
+        );
+    }
+
+    private Set<MonitorGroupMeta> getMonitorGroupMeta2() {
+        return Sets.newHashSet(
+                new MonitorGroupMeta("shard1+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6383"), HostPort.fromString("127.0.0.1:6384")), false),
+                new MonitorGroupMeta("shard1+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6379"), HostPort.fromString("127.0.0.1:6380")), true),
+                new MonitorGroupMeta("shard2+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6381"), HostPort.fromString("127.0.0.1:6382")), true),
+                new MonitorGroupMeta("shard2+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6385"), HostPort.fromString("127.0.0.1:6386")), false)
+        );
+    }
+
+    private Set<MonitorGroupMeta> getMonitorGroupMeta3() {
+        return Sets.newHashSet(
+                new MonitorGroupMeta("shard1+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6379"), HostPort.fromString("127.0.0.1:6380")), false),
+                new MonitorGroupMeta("shard2+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6381"), HostPort.fromString("127.0.0.1:6382")), false),
+                new MonitorGroupMeta("shard1+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6383"), HostPort.fromString("127.0.0.1:6384")), true),
+                new MonitorGroupMeta("shard2+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6385"), HostPort.fromString("127.0.0.1:6386")), true)
+        );
+    }
+
+    private Set<MonitorGroupMeta> getMonitorGroupMeta4() {
+        return Sets.newHashSet(
+                new MonitorGroupMeta("shard1+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6379"), HostPort.fromString("127.0.0.1:6380")), true),
+                new MonitorGroupMeta("shard2+jq", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6381"), HostPort.fromString("127.0.0.1:6382")), true),
+                new MonitorGroupMeta("shard1+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6383"), HostPort.fromString("127.0.0.1:6384")), false),
+                new MonitorGroupMeta("shard2+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6385"), HostPort.fromString("127.0.0.1:6386")), false)
+        );
+    }
+
+    private Set<MonitorGroupMeta> getMonitorGroupMeta5() {
+        return Sets.newHashSet(
+                new MonitorGroupMeta("shard1+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6379"), HostPort.fromString("127.0.0.1:6382")), true),
+                new MonitorGroupMeta("shard2+jq", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6381"), HostPort.fromString("127.0.0.1:6380")), true),
+                new MonitorGroupMeta("shard1+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6383"), HostPort.fromString("127.0.0.1:6384")), false),
+                new MonitorGroupMeta("shard2+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6385"), HostPort.fromString("127.0.0.1:6386")), false)
+        );
+    }
+
+    private Set<MonitorGroupMeta> getMonitorGroupMeta6() {
+        return Sets.newHashSet(
+                new MonitorGroupMeta("shard1+jq", "jq", Sets.newHashSet(HostPort.fromString("127.0.0.1:6379"), HostPort.fromString("127.0.0.1:6382")), false),
+                new MonitorGroupMeta("shard2+jq", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6381"), HostPort.fromString("127.0.0.1:6380")), false),
+                new MonitorGroupMeta("shard1+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6383"), HostPort.fromString("127.0.0.1:6384")), false),
+                new MonitorGroupMeta("shard2+oy", "oy", Sets.newHashSet(HostPort.fromString("127.0.0.1:6385"), HostPort.fromString("127.0.0.1:6386")), false)
+        );
+    }
+}

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/BeaconManager.java

@@ -1,6 +1,7 @@
 package com.ctrip.xpipe.redis.checker;
 
 import com.ctrip.xpipe.cluster.ClusterType;
+import com.ctrip.xpipe.redis.checker.healthcheck.clusteractions.beacon.BeaconCheckStatus;
 
 /**
  * @author lishanglin
@@ -10,4 +11,8 @@ public interface BeaconManager {
 
     void registerCluster(String clusterId, ClusterType clusterType, int orgId);
 
+    void updateCluster(String clusterId, ClusterType clusterType, int orgId);
+
+    BeaconCheckStatus checkClusterHash(String clusterId, ClusterType clusterType, int orgId);
+
 }

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/clusteractions/beacon/BeaconCheckStatus.java

@@ -0,0 +1,21 @@
+package com.ctrip.xpipe.redis.checker.healthcheck.clusteractions.beacon;
+
+public enum BeaconCheckStatus {
+    SERVICE_NOT_FOUND("Service not found"),
+    SYSTEM_NOT_FOUND("System not found"),
+    CLUSTER_NOT_FOUND("Cluster not found"),
+    INCONSISTENCY("Inconsistency"),
+    CONSISTENCY("Consistency"),
+    UNKNOWN("Unknown"),
+    ERROR("Error");
+
+    private final String description;
+
+    BeaconCheckStatus(String description) {
+        this.description = description;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+}

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/clusteractions/beacon/BeaconConsistencyCheckAction.java

@@ -0,0 +1,105 @@
+package com.ctrip.xpipe.redis.checker.healthcheck.clusteractions.beacon;
+
+import com.ctrip.xpipe.cluster.ClusterType;
+import com.ctrip.xpipe.metric.MetricData;
+import com.ctrip.xpipe.metric.MetricProxy;
+import com.ctrip.xpipe.metric.MetricProxyException;
+import com.ctrip.xpipe.redis.checker.BeaconManager;
+import com.ctrip.xpipe.redis.checker.healthcheck.ClusterHealthCheckInstance;
+import com.ctrip.xpipe.redis.checker.healthcheck.ClusterInstanceInfo;
+import com.ctrip.xpipe.redis.checker.healthcheck.leader.AbstractLeaderAwareHealthCheckAction;
+import com.ctrip.xpipe.utils.ServicesUtil;
+import org.slf4j.Logger;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+public class BeaconConsistencyCheckAction extends AbstractLeaderAwareHealthCheckAction<ClusterHealthCheckInstance> {
+
+    private BeaconManager beaconManager;
+
+    private MetricProxy metricProxy = ServicesUtil.getMetricProxy();
+
+    private static long lastSendTime = System.currentTimeMillis();
+
+    public BeaconConsistencyCheckAction(ScheduledExecutorService scheduled, ClusterHealthCheckInstance instance, ExecutorService executors, BeaconManager beaconManager) {
+        super(scheduled, instance, executors);
+        this.beaconManager = beaconManager;
+    }
+
+    @Override
+    protected void doTask() {
+        try {
+            tryDoTask();
+        } catch (Exception e) {
+            logger.error("[CheckBeaconConsistency]", e);
+        }
+    }
+
+    private void tryDoTask() {
+        ClusterInstanceInfo info = instance.getCheckInfo();
+        String clusterId = info.getClusterId();
+        int orgId = info.getOrgId();
+        checkConsistency(clusterId, info.getClusterType(), orgId);
+    }
+
+    @Override
+    protected Logger getHealthCheckLogger() {
+        return logger;
+    }
+
+    @Override
+    protected int getBaseCheckInterval() {
+        return getActionInstance().getHealthCheckConfig().clusterCheckIntervalMilli();
+    }
+
+    private void checkConsistency(String clusterId, ClusterType clusterType, int orgId) {
+        BeaconCheckStatus status;
+        try {
+            status = beaconManager.checkClusterHash(clusterId, clusterType, orgId);
+        } catch (Throwable t) {
+            // cluster not found in beacon
+            status = BeaconCheckStatus.ERROR;
+            logger.error("[checkConsistency]{}:{}:{}", clusterType, orgId, t.getMessage());
+        }
+        handleCheckResult(status, clusterId, clusterType, orgId);
+    }
+
+    private void handleCheckResult(BeaconCheckStatus status, String clusterId, ClusterType clusterType, int orgId) {
+        try {
+            if(status == BeaconCheckStatus.CONSISTENCY) {
+                long currentTime = System.currentTimeMillis();
+                if(currentTime < lastSendTime) {
+                    lastSendTime = currentTime;
+                }
+                if(currentTime - lastSendTime <= getBaseCheckInterval()) {
+                    // avoid send many point to hickwall
+                    return;
+                }
+                lastSendTime = currentTime;
+            } else {
+                beaconManager.registerCluster(clusterId, clusterType, orgId);
+            }
+            sendMetricData(clusterId, status);
+        } catch (Throwable t) {
+            logger.error("[checkPost]{}:{}:{}", clusterType, orgId, t.getMessage());
+        }
+    }
+
+    private MetricData getMetricData(String clusterId, BeaconCheckStatus status) {
+        MetricData metricData = new MetricData("beacon.checker", null, clusterId, null);
+        metricData.setTimestampMilli(System.currentTimeMillis());
+        metricData.setValue(1);
+        metricData.addTag("consistency", status.toString());
+        return metricData;
+    }
+
+    private void sendMetricData(String trackCluster, BeaconCheckStatus status) {
+        try {
+            MetricData metricData = getMetricData(trackCluster, status);
+            metricProxy.writeBinMultiDataPoint(metricData);
+        } catch (MetricProxyException e) {
+            logger.error("[sendMetricData]", e);
+        }
+    }
+}

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/clusteractions/beacon/BeaconMetaCheckActionFactory.java → redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/clusteractions/beacon/BeaconConsistencyCheckActionFactory.java

@@ -13,12 +13,8 @@
 import java.util.Collections;
 import java.util.List;
 
-/**
- * @author lishanglin
- * date 2021/1/15
- */
 @Component
-public class BeaconMetaCheckActionFactory extends AbstractClusterLeaderAwareHealthCheckActionFactory implements OneWaySupport, BiDirectionSupport {
+public class BeaconConsistencyCheckActionFactory extends AbstractClusterLeaderAwareHealthCheckActionFactory implements OneWaySupport, BiDirectionSupport {
 
     @Autowired
     private BeaconManager beaconManager;
@@ -28,14 +24,14 @@ public class BeaconMetaCheckActionFactory extends AbstractClusterLeaderAwareHeal
 
     @Override
     public SiteLeaderAwareHealthCheckAction create(ClusterHealthCheckInstance instance) {
-        BeaconMetaCheckAction action = new BeaconMetaCheckAction(scheduled, instance, executors, beaconManager);
+        BeaconConsistencyCheckAction action = new BeaconConsistencyCheckAction(scheduled, instance, executors, beaconManager);
         action.addControllers(controllers);
         return action;
     }
 
     @Override
     public Class<? extends SiteLeaderAwareHealthCheckAction> support() {
-        return BeaconMetaCheckAction.class;
+        return BeaconConsistencyCheckAction.class;
     }
 
     @Override