fix(grpc): Improve reconnection logic and fix race condition

1. Restore original reconnection logic with TRANSIENT_FAILURE monitoring:
   - Keep original behavior: only force reconnect when different server selected
   - When same server selected, rely on gRPC auto-reconnect mechanism
   - Add TRANSIENT_FAILURE state monitoring to detect prolonged failures
   - Force rebuild channel if either reconnectCount or transientFailureCount exceeds threshold
   - Add keepAliveWithoutCalls(true) to detect half-open connections

2. Fix race condition between reportError() and run():
   - Wrap all state changes (reconnect flag + notifications) in synchronized blocks
   - Prevents reconnect flag and listener status from becoming inconsistent
   - Fixes production issue where reconnect=false but listeners in DISCONNECT state

3. Additional improvements:
   - Adjust keepalive default from 60s to 120s (reduces overhead by 50%)
   - Add getState() method to GRPCChannel for state monitoring

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: hzhaop

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java

@@ -216,7 +216,7 @@ public static class Collector {
          *
          * This maps to `collector.grpc_keepalive_time` in agent.config.
          */
-        public static long GRPC_KEEPALIVE_TIME = 60L;
+        public static long GRPC_KEEPALIVE_TIME = 120L;
         /**
          * The timeout in seconds to wait for a keepalive ack from the backend.
          * If the ack is not received within this time, the connection is considered dead.

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannel.java

@@ -43,7 +43,8 @@ private GRPCChannel(String host, int port, List<ChannelBuilder> channelBuilders,
 
         if (Config.Collector.GRPC_KEEPALIVE_TIME > 0) {
             channelBuilder.keepAliveTime(Config.Collector.GRPC_KEEPALIVE_TIME, TimeUnit.SECONDS)
-                          .keepAliveTimeout(Config.Collector.GRPC_KEEPALIVE_TIMEOUT, TimeUnit.SECONDS);
+                          .keepAliveTimeout(Config.Collector.GRPC_KEEPALIVE_TIMEOUT, TimeUnit.SECONDS)
+                          .keepAliveWithoutCalls(true);
         }
 
         NameResolverRegistry.getDefaultRegistry().register(new DnsNameResolverProvider());
@@ -90,6 +91,10 @@ public boolean isConnected(boolean requestConnection) {
         return originChannel.getState(requestConnection) == ConnectivityState.READY;
     }
 
+    public ConnectivityState getState(boolean requestConnection) {
+        return originChannel.getState(requestConnection);
+    }
+
     public static class Builder {
         private final String host;
         private final int port;

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannelManager.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.agent.core.remote;
 
 import io.grpc.Channel;
+import io.grpc.ConnectivityState;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 
@@ -57,6 +58,8 @@ public class GRPCChannelManager implements BootService, Runnable {
     private volatile List<String> grpcServers;
     private volatile int selectedIdx = -1;
     private volatile int reconnectCount = 0;
+    private volatile int transientFailureCount = 0;
+    private final Object statusLock = new Object();
 
     @Override
     public void prepare() {
@@ -99,7 +102,15 @@ public void shutdown() {
 
     @Override
     public void run() {
-        LOGGER.debug("Selected collector grpc service running, reconnect:{}.", reconnect);
+        if (reconnect) {
+            LOGGER.warn("Selected collector grpc service running, reconnect:{}.", reconnect);
+        } else {
+            LOGGER.debug("Selected collector grpc service running, reconnect:{}.", reconnect);
+        }
+
+        // Check channel state even when reconnect is false to detect prolonged failures
+        checkChannelStateAndTriggerReconnectIfNeeded();
+
         if (IS_RESOLVE_DNS_PERIODICALLY && reconnect) {
             grpcServers = Arrays.stream(Config.Collector.BACKEND_SERVICE.split(","))
                     .filter(StringUtil::isNotBlank)
@@ -130,26 +141,34 @@ public void run() {
                 String server = "";
                 try {
                     int index = Math.abs(random.nextInt()) % grpcServers.size();
-                    selectedIdx = index;
 
                     server = grpcServers.get(index);
                     String[] ipAndPort = server.split(":");
 
-                    LOGGER.debug("Attempting to reconnect to gRPC server {}. Shutting down existing channel if any.", server);
-                    if (managedChannel != null) {
-                        managedChannel.shutdownNow();
-                    }
+                    if (index != selectedIdx) {
+                        selectedIdx = index;
+                        LOGGER.debug("Connecting to different gRPC server {}. Shutting down existing channel if any.", server);
+                        createNewChannel(ipAndPort[0], Integer.parseInt(ipAndPort[1]));
+                    } else {
+                        // Same server, increment reconnectCount and check state
+                        reconnectCount++;
 
-                    managedChannel = GRPCChannel.newBuilder(ipAndPort[0], Integer.parseInt(ipAndPort[1]))
-                                                .addManagedChannelBuilder(new StandardChannelBuilder())
-                                                .addManagedChannelBuilder(new TLSChannelBuilder())
-                                                .addChannelDecorator(new AgentIDDecorator())
-                                                .addChannelDecorator(new AuthenticationDecorator())
-                                                .build();
-                    LOGGER.debug("Successfully reconnected to gRPC server {}.", server);
-                    reconnectCount = 0;
-                    reconnect = false;
-                    notify(GRPCChannelStatus.CONNECTED);
+                        // Force reconnect if reconnectCount or transientFailureCount exceeds threshold
+                        boolean forceReconnect = reconnectCount > Config.Agent.FORCE_RECONNECTION_PERIOD
+                                              || transientFailureCount > Config.Agent.FORCE_RECONNECTION_PERIOD;
+
+                        if (forceReconnect) {
+                            // Failed to reconnect after multiple attempts, force rebuild channel
+                            LOGGER.warn("Force rebuild channel to {} (reconnectCount={}, transientFailureCount={})",
+                                      server, reconnectCount, transientFailureCount);
+                            createNewChannel(ipAndPort[0], Integer.parseInt(ipAndPort[1]));
+                        } else if (managedChannel.isConnected(false)) {
+                            // Reconnect to the same server is automatically done by GRPC,
+                            // therefore we are responsible to check the connectivity and
+                            // set the state and notify listeners
+                            markAsConnected();
+                        }
+                    }
 
                     return;
                 } catch (Throwable t) {
@@ -177,17 +196,85 @@ public Channel getChannel() {
      */
     public void reportError(Throwable throwable) {
         if (isNetworkError(throwable)) {
+            triggerReconnect();
+        }
+    }
+
+    private void notify(GRPCChannelStatus status) {
+        synchronized (listeners) {
+            for (GRPCChannelListener listener : listeners) {
+                try {
+                    listener.statusChanged(status);
+                } catch (Throwable t) {
+                    LOGGER.error(t, "Fail to notify {} about channel connected.", listener.getClass().getName());
+                }
+            }
+        }
+    }
+
+    /**
+     * Create a new gRPC channel to the specified server and reset connection state.
+     */
+    private void createNewChannel(String host, int port) throws Exception {
+        if (managedChannel != null) {
+            managedChannel.shutdownNow();
+        }
+
+        managedChannel = GRPCChannel.newBuilder(host, port)
+                                    .addManagedChannelBuilder(new StandardChannelBuilder())
+                                    .addManagedChannelBuilder(new TLSChannelBuilder())
+                                    .addChannelDecorator(new AgentIDDecorator())
+                                    .addChannelDecorator(new AuthenticationDecorator())
+                                    .build();
+
+        markAsConnected();
+    }
+
+    /**
+     * Trigger reconnection by setting reconnect flag and notifying listeners.
+     */
+    private void triggerReconnect() {
+        synchronized (statusLock) {
             reconnect = true;
             notify(GRPCChannelStatus.DISCONNECT);
         }
     }
 
-    private void notify(GRPCChannelStatus status) {
-        for (GRPCChannelListener listener : listeners) {
+    /**
+     * Mark connection as successful and reset connection state.
+     */
+    private void markAsConnected() {
+        synchronized (statusLock) {
+            reconnectCount = 0;
+            reconnect = false;
+            notify(GRPCChannelStatus.CONNECTED);
+        }
+    }
+
+    /**
+     * Check the connectivity state of existing channel and trigger reconnect if needed.
+     * This method monitors TRANSIENT_FAILURE state and triggers reconnect if the failure persists too long.
+     */
+    private void checkChannelStateAndTriggerReconnectIfNeeded() {
+        if (managedChannel != null) {
             try {
-                listener.statusChanged(status);
+                ConnectivityState state = managedChannel.getState(false);
+                LOGGER.debug("Current channel state: {}", state);
+
+                if (state == ConnectivityState.TRANSIENT_FAILURE) {
+                    transientFailureCount++;
+                    LOGGER.warn("Channel in TRANSIENT_FAILURE state, count: {}", transientFailureCount);
+                } else if (state == ConnectivityState.SHUTDOWN) {
+                    LOGGER.warn("Channel is SHUTDOWN");
+                    if (!reconnect) {
+                        triggerReconnect();
+                    }
+                } else {
+                    // IDLE, READY, CONNECTING are all normal states
+                    transientFailureCount = 0;
+                }
             } catch (Throwable t) {
-                LOGGER.error(t, "Fail to notify {} about channel connected.", listener.getClass().getName());
+                LOGGER.error(t, "Error checking channel state");
             }
         }
     }

apm-sniffer/config/agent.config

@@ -102,7 +102,7 @@ collector.backend_service=${SW_AGENT_COLLECTOR_BACKEND_SERVICES:127.0.0.1:11800}
 collector.grpc_upstream_timeout=${SW_AGENT_COLLECTOR_GRPC_UPSTREAM_TIMEOUT:30}
 # The interval in seconds to send a keepalive ping to the backend.
 # If this is less than or equal to 0, the keepalive is disabled.
-#collector.grpc_keepalive_time=${SW_AGENT_COLLECTOR_GRPC_KEEPALIVE_TIME:60}
+#collector.grpc_keepalive_time=${SW_AGENT_COLLECTOR_GRPC_KEEPALIVE_TIME:120}
 # The timeout in seconds to wait for a keepalive ack from the backend.
 # If the ack is not received within this time, the connection is considered dead.
 #collector.grpc_keepalive_timeout=${SW_AGENT_COLLECTOR_GRPC_KEEPALIVE_TIMEOUT:30}