fix bugs below

core/src/main/java/com/ctrip/xpipe/endpoint/ClusterShardHostPort.java

@@ -1,6 +1,8 @@
 package com.ctrip.xpipe.endpoint;
 
-import com.ctrip.xpipe.utils.ObjectUtils;
+import org.springframework.lang.Nullable;
+
+import java.util.Objects;
 
 /**
  * @author wenchao.meng
@@ -15,20 +17,26 @@
 
     private HostPort hostPort;
 
+    @Nullable
+    private String activeDc;
 
-    public ClusterShardHostPort(String clusterName, String shardName, HostPort hostPort){
+    public ClusterShardHostPort(String clusterName, String shardName, String activeDc, HostPort hostPort){
         this.clusterName = clusterName;
         this.shardName = shardName;
         this.hostPort = hostPort;
+        this.activeDc = activeDc;
+    }
+
+    public ClusterShardHostPort(String clusterName, String shardName, HostPort hostPort){
+        this(clusterName, shardName, null, hostPort);
     }
 
     public ClusterShardHostPort(String clusterName, String shardName){
-        this.clusterName = clusterName;
-        this.shardName = shardName;
+        this(clusterName, shardName, null, null);
     }
 
     public ClusterShardHostPort(HostPort hostPort){
-        this.hostPort = hostPort;
+        this(null, null, null, hostPort);
     }
 
     public HostPort getHostPort() {
@@ -55,38 +63,33 @@
         this.shardName = shardName;
     }
 
+    @Nullable
+    public String getActiveDc() {
+        return activeDc;
+    }
+
+    public void setActiveDc(@Nullable String activeDc) {
+        this.activeDc = activeDc;
+    }
+
     @Override
-    public boolean equals(Object obj) {
-
-        if(this == obj){
-            return true;
-        }
-
-        if(!(obj instanceof ClusterShardHostPort)){
-            return false;
-        }
-
-        ClusterShardHostPort other = (ClusterShardHostPort) obj;
-        if(!(ObjectUtils.equals(this.clusterName, other.clusterName))){
-            return false;
-        }
-        if(!(ObjectUtils.equals(this.shardName, other.shardName))){
-            return false;
-        }
-        if(!(ObjectUtils.equals(this.hostPort, other.hostPort))){
-            return false;
-        }
-
-        return true;
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ClusterShardHostPort that = (ClusterShardHostPort) o;
+        return Objects.equals(clusterName, that.clusterName) &&
+                Objects.equals(shardName, that.shardName) &&
+                Objects.equals(hostPort, that.hostPort) &&
+                Objects.equals(activeDc, that.activeDc);
     }
 
     @Override
     public int hashCode() {
-        return ObjectUtils.hashCode(clusterName, shardName, hostPort);
+        return Objects.hash(clusterName, shardName, hostPort, activeDc);
     }
 
     @Override
     public String toString() {
-        return String.format("%s,%s[%s]", clusterName, shardName, hostPort);
+        return String.format("%s,%s[%s][%s]", clusterName, shardName, activeDc, hostPort);
     }
 }

redis/package/redis-console-package/src/main/scripts/startup.sh

@@ -97,12 +97,19 @@ function tryRemoveHeapTrace() {
     find "$logdir" -type f -name "heap_trace_$suffix*.txt" -delete
 }
 
+function tryRemoveJarLog() {
+    logdir=$1
+    appname=$2
+    find "$logdir" -type f -name "$appname_*.log" -delete
+}
+
 #VARS
 FULL_DIR=`getCurrentRealPath`
 SERVICE_NAME=redis-console
 LOG_DIR=/opt/logs/100004374
 `trySaveHeapTrace ${LOG_DIR}`
 `tryRemoveHeapTrace ${LOG_DIR}`
+`tryRemoveJarLog ${LOG_DIR} ${SERVICE_NAME}`
 SERVER_PORT=`getPortFromPathOrDefault $FULL_DIR 8080`
 JMX_PORT=` expr $SERVER_PORT + 10000 `
 IP=`ifconfig | grep "inet.10" | awk '{print $2}; NR == 1 {exit}'`

redis/package/redis-keeper-package/src/main/scripts/startup.sh

@@ -105,6 +105,11 @@ function getRole(){
     fi
     echo `toUpper $ENV`
 }
+function tryRemoveJarLog() {
+    logdir=$1
+    appname=$2
+    find "$logdir" -type f -name "$appname_*.log" -delete
+}
 
 #VARS
 FULL_DIR=`getCurrentRealPath`
@@ -113,6 +118,7 @@ SERVER_PORT=`getPortFromPathOrDefault $FULL_DIR 8080`
 JMX_PORT=` expr $SERVER_PORT + 10000 `
 IP=`ifconfig | grep "inet.10" | awk '{print $2}; NR == 1 {exit}'`
 LOG_DIR=/opt/logs/100004376
+`tryRemoveJarLog ${LOG_DIR} ${SERVICE_NAME}`
 
 if [ ! $SERVER_PORT -eq 8080 ];then
     LOG_DIR=${LOG_DIR}_$SERVER_PORT

redis/package/redis-metaserver-package/src/main/scripts/startup.sh

@@ -116,12 +116,19 @@ function tryRemoveHeapTrace() {
     find "$logdir" -type f -name "heap_trace_$suffix*.txt" -delete
 }
 
+function tryRemoveJarLog() {
+    logdir=$1
+    appname=$2
+    find "$logdir" -type f -name "$appname_*.log" -delete
+}
+
 #VARS
 FULL_DIR=`getCurrentRealPath`
 SERVICE_NAME=redis-meta
 LOG_DIR=/opt/logs/100004375
 `trySaveHeapTrace ${LOG_DIR}`
 `tryRemoveHeapTrace ${LOG_DIR}`
+`tryRemoveJarLog ${LOG_DIR} ${SERVICE_NAME}`
 SERVER_PORT=`getPortFromPathOrDefault $FULL_DIR 8080`
 JMX_PORT=` expr $SERVER_PORT + 10000 `
 IP=`ifconfig | grep "inet.10" | awk '{print $2}; NR == 1 {exit}'`
@@ -175,7 +182,7 @@ else
     fi
     JAVA_OPTS="$JAVA_OPTS -Xms${USED_MEM}m -Xmx${USED_MEM}m -Xmn${XMN}m -XX:+AlwaysPreTouch  -XX:MaxDirectMemorySize=${MAX_DIRECT}m -XX:MetaspaceSize=${META_SPACE}m -XX:MaxMetaspaceSize=${MAX_META_SPACE}m"
 fi
-export JAVA_OPTS="$JAVA_OPTS -XX:+UseParNewGC -XX:MaxTenuringThreshold=5 -XX:+UseConcMarkSweepGC -XX:+UseCMSInitiatingOccupancyOnly -XX:+ScavengeBeforeFullGC -XX:+UseCMSCompactAtFullCollection -XX:+CMSParallelRemarkEnabled -XX:CMSFullGCsBeforeCompaction=9 -XX:CMSInitiatingOccupancyFraction=60 -XX:-CMSClassUnloadingEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:-ReduceInitialCardMarks -XX:+CMSPermGenSweepingEnabled -XX:CMSInitiatingPermOccupancyFraction=70 -XX:+ExplicitGCInvokesConcurrent -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationConcurrentTime -XX:+PrintHeapAtGC -XX:+HeapDumpOnOutOfMemoryError -XX:-OmitStackTraceInFastThrow -Duser.timezone=Asia/Shanghai -Dclient.encoding.override=UTF-8 -Dfile.encoding=UTF-8 -Xloggc:$LOG_DIR/heap_trace.txt -XX:HeapDumpPath=$LOG_DIR/HeapDumpOnOutOfMemoryError/  -Dcom.sun.management.jmxremote.port=$JMX_PORT -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=${IP} -XX:+UnlockCommercialFeatures -XX:+FlightRecorder -Dio.netty.allocator.useCacheForAllThreads=false -Djava.security.egd=file:/dev/./urandom"
+export JAVA_OPTS="$JAVA_OPTS -XX:+UseParNewGC -XX:MaxTenuringThreshold=5 -XX:+UseConcMarkSweepGC -XX:+UseCMSInitiatingOccupancyOnly -XX:+ScavengeBeforeFullGC -XX:+UseCMSCompactAtFullCollection -XX:+CMSParallelRemarkEnabled -XX:CMSFullGCsBeforeCompaction=9 -XX:CMSInitiatingOccupancyFraction=60 -XX:-CMSClassUnloadingEnabled -XX:SoftRefLRUPolicyMSPerMB=1000 -XX:-ReduceInitialCardMarks -XX:+CMSPermGenSweepingEnabled -XX:CMSInitiatingPermOccupancyFraction=70 -XX:+ExplicitGCInvokesConcurrent -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationConcurrentTime -XX:+PrintHeapAtGC -XX:+HeapDumpOnOutOfMemoryError -XX:-OmitStackTraceInFastThrow -Duser.timezone=Asia/Shanghai -Dclient.encoding.override=UTF-8 -Dfile.encoding=UTF-8 -Xloggc:$LOG_DIR/heap_trace.txt -XX:HeapDumpPath=$LOG_DIR/HeapDumpOnOutOfMemoryError/  -Dcom.sun.management.jmxremote.port=$JMX_PORT -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=${IP} -XX:+UnlockCommercialFeatures -XX:+FlightRecorder -Dio.netty.allocator.useCacheForAllThreads=false -Djava.security.egd=file:/dev/./urandom"
 
 PATH_TO_JAR=$SERVICE_NAME".jar"
 SERVER_URL="http://localhost:$SERVER_PORT"

redis/package/redis-proxy-package/src/main/scripts/startup.sh

@@ -110,6 +110,12 @@ function tryRemoveHeapTrace() {
     find "$logdir" -type f -name "heap_trace_$suffix*.txt" -delete
 }
 
+function tryRemoveJarLog() {
+    logdir=$1
+    appname=$2
+    find "$logdir" -type f -name "$appname_*.log" -delete
+}
+
 #VARS
 FULL_DIR=`getCurrentRealPath`
 SERVICE_NAME=redis-proxy
@@ -119,6 +125,7 @@ IP=`ifconfig | grep "inet.10" | awk '{print $2}; NR == 1 {exit}'`
 LOG_DIR=/opt/logs/100013684
 `trySaveHeapTrace ${LOG_DIR}`
 `tryRemoveHeapTrace ${LOG_DIR}`
+`tryRemoveJarLog ${LOG_DIR} ${SERVICE_NAME}`
 
 if [ ! $SERVER_PORT -eq 8080 ];then
     LOG_DIR=${LOG_DIR}_$SERVER_PORT

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/interaction/handler/AbstractHealthEventHandler.java

@@ -66,8 +66,9 @@ protected List<HEALTH_STATE> getSatisfiedStates(){
     }
 
     protected void tryMarkDown(AbstractInstanceEvent event) {
-        if (siteStability.isSiteStable()) {
+        if (!siteStability.isSiteStable()) {
             logger.warn("[onEvent][site down, skip] {}", event);
+            return;
         }
         if(!masterUp(event)) {
             logger.info("[onEvent][master down, do not call client service]{}", event);

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/sentinel/NoRedisToSubContext.java

@@ -0,0 +1,31 @@
+package com.ctrip.xpipe.redis.checker.healthcheck.actions.sentinel;
+
+import com.ctrip.xpipe.exception.XpipeRuntimeException;
+import com.ctrip.xpipe.redis.checker.healthcheck.impl.DefaultRedisHealthCheckInstance;
+
+/**
+ * @author lishanglin
+ * date 2024/2/5
+ */
+public class NoRedisToSubContext extends SentinelActionContext {
+
+    private static final Throwable th = new XpipeRuntimeException("No Redis to sub.");
+
+    private String cluster;
+
+    private String shard;
+
+    public NoRedisToSubContext(String cluster, String shard) {
+        super(new DefaultRedisHealthCheckInstance(), th);
+        this.cluster = cluster;
+        this.shard = shard;
+    }
+
+    public String getCluster() {
+        return cluster;
+    }
+
+    public String getShard() {
+        return shard;
+    }
+}

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/sentinel/SentinelHelloCheckAction.java

@@ -50,7 +50,7 @@
 
     private HealthCheckInstanceManager instanceManager;
 
-    private Set<RedisHealthCheckInstance> redisInstancesToCheck = new HashSet<>();
+    private Map<String, Set<RedisHealthCheckInstance>> redisInstancesToCheck = new HashMap<>();
 
     private volatile boolean collecting = false;
 
@@ -109,16 +109,20 @@
         return logger;
     }
 
-    Set<RedisHealthCheckInstance> redisInstancesToCheck() {
-        Set<RedisHealthCheckInstance> redisHealthCheckInstances = new HashSet<>();
+    Map<String, Set<RedisHealthCheckInstance>> redisInstancesToCheck() {
+        Map<String, Set<RedisHealthCheckInstance>> redisHealthCheckInstances = new HashMap<>();
         try {
             metaCache.getXpipeMeta().getDcs().forEach((dc, dcMeta) -> {
                 ClusterMeta clusterMeta = dcMeta.getClusters().get(getActionInstance().getCheckInfo().getClusterId());
-                logger.debug("[{}-{}][{}]found in MetaCache", LOG_TITLE, instance.getCheckInfo().getClusterId(), dc);
                 if (clusterMeta != null) {
+                    logger.debug("[{}-{}][{}]found in MetaCache", LOG_TITLE, instance.getCheckInfo().getClusterId(), dc);
                     Map<String, ShardMeta> clusterShards = clusterMeta.getShards();
                     logger.debug("[{}-{}][{}]shards num:{}, detail info:{}", LOG_TITLE, instance.getCheckInfo().getClusterId(), dc, clusterShards.size(), clusterShards);
                     clusterShards.forEach((shardId, shardMeta) -> {
+                        if (!redisHealthCheckInstances.containsKey(shardId)) {
+                            redisHealthCheckInstances.put(shardId, new HashSet<>());
+                        }
+
                         try {
                             List<RedisMeta> redisMetas = shardMeta.getRedises();
                             logger.debug("[{}-{}+{}][{}]redis num:{}, detail info:{}", LOG_TITLE, instance.getCheckInfo().getClusterId(), shardId, dc, redisMetas.size(), redisMetas);
@@ -130,7 +134,7 @@
                                         continue;
                                     }
                                     if (super.shouldCheckInstance(redisInstance)) {
-                                        redisHealthCheckInstances.add(redisInstance);
+                                        redisHealthCheckInstances.get(shardId).add(redisInstance);
                                         hellos.put(redisInstance, new SentinelHellos());
                                     } else {
                                         redisInstance.getRedisSession().closeSubscribedChannel(HELLO_CHANNEL);
@@ -151,42 +155,49 @@
         return redisHealthCheckInstances;
     }
 
-    void subAllRedisInstances(Set<RedisHealthCheckInstance> redisInstancesToCheck) {
-        redisInstancesToCheck.forEach(redisInstanceToCheck -> {
-            try {
-                RedisInstanceInfo info = redisInstanceToCheck.getCheckInfo();
-                if (redisInstanceToCheck.getCheckInfo().isInActiveDc()) {
-                    logger.debug("[{}-{}+{}]{} instance {} in active dc", LOG_TITLE, info.getClusterShardHostport().getClusterName(),
-                            info.getShardId(), info.getDcId(), redisInstanceToCheck.getCheckInfo().getHostPort());
-                }
-                redisInstanceToCheck.getRedisSession().subscribeIfAbsent(new RedisSession.SubscribeCallback() {
-                    @Override
-                    public void message(String channel, String message) {
-                        if (!collecting)
-                            return;
-
-                        SentinelHello hello = SentinelHello.fromString(message);
-                        SentinelHellos sentinelHellos = hellos.get(redisInstanceToCheck);
-                        if (sentinelHellos == null) {
-                            logger.warn("[{}-{}+{}]{} redisHealthCheckInstance {} not found", LOG_TITLE, info.getClusterShardHostport().getClusterName(),
+    void subAllRedisInstances(Map<String, Set<RedisHealthCheckInstance>> redisInstancesToCheck) {
+        redisInstancesToCheck.forEach((shardId, shardRedisInstances) -> {
+            if (shardRedisInstances.isEmpty()) {
+                logger.debug("[{}-{}+{}] no instances to sub", LOG_TITLE, instance.getCheckInfo().getClusterId(), shardId);
+                this.notifyListeners(new NoRedisToSubContext(instance.getCheckInfo().getClusterId(), shardId));
+            } else {
+                shardRedisInstances.forEach(redisInstanceToCheck -> {
+                    try {
+                        RedisInstanceInfo info = redisInstanceToCheck.getCheckInfo();
+                        if (redisInstanceToCheck.getCheckInfo().isInActiveDc()) {
+                            logger.debug("[{}-{}+{}]{} instance {} in active dc", LOG_TITLE, info.getClusterShardHostport().getClusterName(),
                                     info.getShardId(), info.getDcId(), redisInstanceToCheck.getCheckInfo().getHostPort());
-                        } else {
-                            sentinelHellos.addSentinelHello(hello);
                         }
-                    }
+                        redisInstanceToCheck.getRedisSession().subscribeIfAbsent(new RedisSession.SubscribeCallback() {
+                            @Override
+                            public void message(String channel, String message) {
+                                if (!collecting)
+                                    return;
+
+                                SentinelHello hello = SentinelHello.fromString(message);
+                                SentinelHellos sentinelHellos = hellos.get(redisInstanceToCheck);
+                                if (sentinelHellos == null) {
+                                    logger.warn("[{}-{}+{}]{} redisHealthCheckInstance {} not found", LOG_TITLE, info.getClusterShardHostport().getClusterName(),
+                                            info.getShardId(), info.getDcId(), redisInstanceToCheck.getCheckInfo().getHostPort());
+                                } else {
+                                    sentinelHellos.addSentinelHello(hello);
+                                }
+                            }
 
-                    @Override
-                    public void fail(Throwable e) {
-                        if (!collecting) {
-                            return;
-                        }
+                            @Override
+                            public void fail(Throwable e) {
+                                if (!collecting) {
+                                    return;
+                                }
 
-                        logger.warn("[{}-{}+{}]{} instance {} sub-failed, reason:{}", LOG_TITLE, info.getClusterShardHostport().getClusterName(), info.getShardId(), info.getDcId(), info.getHostPort(), e.getMessage());
-                        errors.put(redisInstanceToCheck, e);
+                                logger.warn("[{}-{}+{}]{} instance {} sub-failed, reason:{}", LOG_TITLE, info.getClusterShardHostport().getClusterName(), info.getShardId(), info.getDcId(), info.getHostPort(), e.getMessage());
+                                errors.put(redisInstanceToCheck, e);
+                            }
+                        }, HELLO_CHANNEL);
+                    } catch (Exception e) {
+                        logger.warn("[{}-{}]subscribe redis instance {}:{} failed", LOG_TITLE,instance.getCheckInfo().getClusterId(), redisInstanceToCheck.getEndpoint().getHost(), redisInstanceToCheck.getEndpoint().getPort(), e);
                     }
-                }, HELLO_CHANNEL);
-            } catch (Exception e) {
-                logger.warn("[{}-{}]subscribe redis instance {}:{} failed", LOG_TITLE,instance.getCheckInfo().getClusterId(), redisInstanceToCheck.getEndpoint().getHost(), redisInstanceToCheck.getEndpoint().getPort(), e);
+                });
             }
         });
     }
@@ -250,9 +261,9 @@
 
     @Override
     public void doStop() {
-        redisInstancesToCheck.forEach(redisInstance -> {
+        redisInstancesToCheck.values().forEach(instances -> instances.forEach(redisInstance -> {
             redisInstance.getRedisSession().closeSubscribedChannel(HELLO_CHANNEL);
-        });
+        }));
         resetResults();
         super.doStop();
     }

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/sentinel/SentinelHelloCollector.java

@@ -13,6 +13,6 @@ public interface SentinelHelloCollector extends HealthCheckActionListener<Sentin
 
     @Override
     default boolean worksfor(ActionContext t) {
-        return t instanceof SentinelActionContext;
+        return !(t instanceof NoRedisToSubContext) && t instanceof SentinelActionContext;
     }
 }