fix: address multi-backend review (collector exclude, fork env, shutdown, RPC margin)

Skip sw_grpc instrumentation for agent collector channels via thread-local
scope; move throttled reporter logs off grpc_channel so GRPC_ENABLE_FORK_SUPPORT
is set before import grpc; cancel only background tasks on async shutdown;
widen sync RPC deadline vs queue batch window.

Author: songzhendong

skywalking/agent/__init__.py

@@ -33,7 +33,7 @@
 from skywalking.profile.snapshot import TracingThreadSnapshot
 from skywalking.protocol.language_agent.Meter_pb2 import MeterData
 from skywalking.protocol.logging.Logging_pb2 import LogData
-from skywalking.utils.grpc_channel import log_dropped_throttled, log_reporter_exception_throttled
+from skywalking.utils.reporter_log import log_dropped_throttled, log_reporter_exception_throttled
 from skywalking.utils.singleton import Singleton
 
 if TYPE_CHECKING:
@@ -147,13 +147,19 @@ async def _shutdown_async_queue(q: asyncio.Queue, label: str) -> None:
         )
 
 
-async def _cancel_pending_tasks(loop) -> None:
+async def _cancel_pending_tasks(tasks) -> None:
     """
-    Cancel every task on the agent loop except the caller. The current task must be
-    excluded from the gather too, otherwise it would await itself and never return.
+    Cancel agent-owned reporter / connectivity-watch tasks only.
+
+    Must not cancel the asyncio.run root (__start_event_loop_async): that tears down
+    the Runner and cancels this fini coroutine before protocol aclose() runs.
+    The current task (fini) is also excluded so we never await ourselves.
     """
-    current = asyncio.current_task(loop)
-    pending = [task for task in asyncio.all_tasks(loop) if task is not current]
+    current = asyncio.current_task()
+    pending = [
+        task for task in tasks
+        if task is not None and task is not current and not task.done()
+    ]
     if not pending:
         return
     for task in pending:
@@ -814,8 +820,11 @@ async def __start_event_loop_async(self) -> None:
 
         await self.__bootstrap()  # gather all coroutines
 
+        # Track Tasks explicitly so shutdown cancels only reporters/watchers, not the
+        # asyncio.run root task that is awaiting this gather.
+        self.background_tasks = {asyncio.create_task(coro) for coro in self.background_coroutines}
         logger.debug('All background coroutines started')
-        await asyncio.gather(*self.background_coroutines)
+        await asyncio.gather(*self.background_tasks)
 
     def __start_event_loop(self) -> None:
         try:
@@ -885,7 +894,7 @@ async def __fini_async(self):
         if config.agent_meter_reporter_active:
             await _shutdown_async_queue(self.__meter_queue, 'meter')
 
-        await _cancel_pending_tasks(self.loop)
+        await _cancel_pending_tasks(getattr(self, 'background_tasks', set()) or set())
 
         aclose = getattr(self.__protocol, 'aclose', None)
         if callable(aclose):

skywalking/agent/protocol/grpc.py

@@ -18,7 +18,7 @@
 import logging
 import traceback
 from queue import Queue, Empty
-from time import time
+from time import monotonic
 
 import grpc
 
@@ -33,8 +33,8 @@
     create_sync_channel,
     handle_rpc_error,
     is_channel_ready,
-    log_dropped_throttled,
 )
+from skywalking.utils.reporter_log import log_dropped_throttled
 from skywalking.profile.profile_task import ProfileTask
 from skywalking.profile.snapshot import TracingThreadSnapshot
 from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
@@ -45,6 +45,25 @@
 from skywalking.trace.segment import Segment
 
 
+def _queue_get_within_batch(queue: Queue, block: bool, batch_deadline: float):
+    """
+    Get one item within an absolute batch window (monotonic deadline).
+
+    Avoids int(elapsed) truncation that could let queue waits approach
+    agent_queue_timeout + 1s and collide with a tight RPC deadline.
+    Returns None when the window is exhausted or the queue is empty.
+    """
+    remaining = batch_deadline - monotonic()
+    if remaining <= 0:
+        return None
+    try:
+        if block:
+            return queue.get(block=True, timeout=remaining)
+        return queue.get(block=False)
+    except Empty:
+        return None
+
+
 class GrpcProtocol(Protocol):
     def __init__(self):
         self.properties_sent = False
@@ -154,16 +173,11 @@ def generator():
             nonlocal start, sent
 
             while True:
-                try:
-                    timeout = config.agent_queue_timeout  # type: int
-                    if not start:  # make sure first time through queue is always checked
-                        start = time()
-                    else:
-                        timeout -= int(time() - start)
-                        if timeout <= 0:  # this is to make sure we exit eventually instead of being fed continuously
-                            return
-                    segment = queue.get(block=block, timeout=timeout)  # type: Segment
-                except Empty:
+                if start is None:
+                    start = monotonic()
+                batch_deadline = start + float(config.agent_queue_timeout)
+                segment = _queue_get_within_batch(queue, block, batch_deadline)  # type: Segment
+                if segment is None:
                     return
 
                 queue.task_done()
@@ -230,16 +244,11 @@ def generator():
             nonlocal start, sent
 
             while True:
-                try:
-                    timeout = config.agent_queue_timeout  # type: int
-                    if not start:  # make sure first time through queue is always checked
-                        start = time()
-                    else:
-                        timeout -= int(time() - start)
-                        if timeout <= 0:  # this is to make sure we exit eventually instead of being fed continuously
-                            return
-                    log_data = queue.get(block=block, timeout=timeout)  # type: LogData
-                except Empty:
+                if start is None:
+                    start = monotonic()
+                batch_deadline = start + float(config.agent_queue_timeout)
+                log_data = _queue_get_within_batch(queue, block, batch_deadline)  # type: LogData
+                if log_data is None:
                     return
 
                 queue.task_done()
@@ -268,16 +277,11 @@ def generator():
             nonlocal start, sent
 
             while True:
-                try:
-                    timeout = config.agent_queue_timeout  # type: int
-                    if not start:  # make sure first time through queue is always checked
-                        start = time()
-                    else:
-                        timeout -= int(time() - start)
-                        if timeout <= 0:  # this is to make sure we exit eventually instead of being fed continuously
-                            return
-                    meter_data = queue.get(block=block, timeout=timeout)  # type: MeterData
-                except Empty:
+                if start is None:
+                    start = monotonic()
+                batch_deadline = start + float(config.agent_queue_timeout)
+                meter_data = _queue_get_within_batch(queue, block, batch_deadline)  # type: MeterData
+                if meter_data is None:
                     return
 
                 queue.task_done()
@@ -305,16 +309,11 @@ def generator():
             nonlocal start, sent
 
             while True:
-                try:
-                    timeout = config.agent_queue_timeout  # type: int
-                    if not start:  # make sure first time through queue is always checked
-                        start = time()
-                    else:
-                        timeout -= int(time() - start)
-                        if timeout <= 0:  # this is to make sure we exit eventually instead of being fed continuously
-                            return
-                    snapshot = queue.get(block=block, timeout=timeout)  # type: TracingThreadSnapshot
-                except Empty:
+                if start is None:
+                    start = monotonic()
+                batch_deadline = start + float(config.agent_queue_timeout)
+                snapshot = _queue_get_within_batch(queue, block, batch_deadline)  # type: TracingThreadSnapshot
+                if snapshot is None:
                     return
 
                 queue.task_done()

skywalking/agent/protocol/grpc_aio.py

@@ -28,12 +28,12 @@
 from skywalking.client.grpc_aio import GrpcServiceManagementClientAsync, GrpcTraceSegmentReportServiceAsync, \
     GrpcProfileTaskChannelServiceAsync, GrpcLogReportServiceAsync, GrpcMeterReportServiceAsync
 from skywalking.loggings import logger, logger_debug_enabled
+from skywalking.utils.reporter_log import log_dropped_throttled
 from skywalking.utils.grpc_channel import (
     apply_connectivity_transition,
     create_aio_channel,
     handle_rpc_error,
     is_channel_ready,
-    log_dropped_throttled,
 )
 from skywalking.profile.profile_task import ProfileTask
 from skywalking.profile.snapshot import TracingThreadSnapshot

skywalking/plugins/sw_grpc.py

@@ -236,8 +236,19 @@ def intercept_stream_stream(self, continuation, client_call_details, request_ite
                 return self._intercept(continuation, client_call_details, request_iterator)
 
         def _sw_grpc_channel_factory(target: str, *args: Any, **kwargs: Any):
+            from skywalking.utils.grpc_channel import (
+                is_agent_collector_channel,
+                is_building_agent_collector_channel,
+            )
+
             c = _grpc_channel(target, *args, **kwargs)
-            if target == config.agent_collector_backend_services:
+            # Prefer explicit agent→OAP marker/scope: multi-address targets are rewritten
+            # (ipv4:/ipv6:) and no longer equal agent_collector_backend_services.
+            if (
+                is_building_agent_collector_channel()
+                or is_agent_collector_channel(c)
+                or target == config.agent_collector_backend_services
+            ):
                 return c
             return grpc.intercept_channel(c, _ClientInterceptor(target))
 
@@ -463,7 +474,14 @@ def __init__(
                 compression: Optional[grpc.Compression],
                 interceptors: Optional[Sequence[grpc.aio.ClientInterceptor]],
             ):
-                if target != config.agent_collector_backend_services:
+                from skywalking.utils.grpc_channel import is_building_agent_collector_channel
+
+                # Multi-address collector targets are rewritten; do not rely on string equality alone.
+                skip_sw = (
+                    is_building_agent_collector_channel()
+                    or target == config.agent_collector_backend_services
+                )
+                if not skip_sw:
                     _sw_interceptors: List[grpc.aio.ClientInterceptor] = [
                         _AioClientUnaryUnaryInterceptor(target),
                         _AioClientUnaryStreamInterceptor(target),