feat: native gRPC multi-backend failover via pick_first

[songzhendong]

Summary

• Native gRPC multi-backend failover for SW_AGENT_COLLECTOR_BACKEND_SERVICES, aligned with skywalking-nodejs #144: one channel for the process lifetime with C-core pick_first
• Single address stays plain host:port; multi uses ipv4: / ipv6: (mixed families via IPv4-mapped). shuffleAddressList is on; :authority / TLS SNI stay on the first configured endpoint
• Channel options: grpc.enable_http_proxy=0, no gRPC keepalive, reconnect backoff capped at 30s, service_config retries only ManagementService.reportInstanceProperties. READY gate before dequeue; failed send batches discarded with throttled drop logs. Unary / sync streaming RPCs use a deadline; aio client-streaming collect omits timeout= because generators await empty queues. Bounded shutdown flush so atexit cannot hang

Docs: docs/en/setup/Intrusive.md and Configuration.md.

Test plan

• Unit: channel parse/encode, READY gate, drop counting, keepAlive isolation, aio streaming without deadline, shutdown helpers
• E2E: tests/e2e/case/grpc/failover/ (two OAPs; stop a backend that has traffic; sorted service list)
• Fork CI green (full matrix + dedicated e2e-failover on 3.10 / 3.12 / 3.14)

[wu-sheng]

Thanks for the comprehensive failover work. I found four issues that should be addressed before merging:

1. [P1] Keep multi-backend collector channels excluded from gRPC instrumentation

skywalking/utils/grpc_channel.py:480

The existing gRPC plugin excludes the agent's collector channel by comparing target with config.agent_collector_backend_services. For multiple backends, this code passes an encoded ipv4:/ipv6: target, so the comparison no longer matches and the agent instruments its own reporting RPCs.

I reproduced an agent-channel RPC producing a SkyWalking span. In the sync reporter, completing a reporting RPC then enqueues its own span, which can sustain a continuous self-reporting loop.

Please adapt the multi-address sync and aio paths so agent-owned channels remain excluded from tracing. Prefer an explicit agent-channel marker over resolving/comparing the target again, and add regression tests with the gRPC plugin installed.

2. [P1] Set gRPC fork support before importing grpc

skywalking/agent/__init__.py:36

This top-level import loads skywalking.utils.grpc_channel, which imports grpc. That happens before start() and start_prefork_master() set GRPC_ENABLE_FORK_SUPPORT=true.

grpcio reads this setting during module initialization. I reproduced grpc._cython.cygrpc.is_fork_support_enabled() remaining False after the later environment assignment. Explicit fork support can therefore run without gRPC's required fork handlers, causing hangs or silent reporting failures.

Please keep the generic logging helpers in a gRPC-neutral module or otherwise defer importing grpc until after the environment is configured. A subprocess import-order test would cover this reliably.

3. [P2] Do not cancel the task owned by asyncio.run during shutdown

skywalking/agent/__init__.py:155

_cancel_pending_tasks() cancels every task except the current shutdown task. In production this includes the root __start_event_loop_async task owned by asyncio.run.

Once that root task is cancelled, asyncio.Runner begins teardown and cancels the still-running __fini_async task. I reproduced the submitted shutdown future ending with CancelledError; execution never reached the protocol aclose() block.

The current unit test uses loop.run_forever, so it has no asyncio.run root task and misses this behavior. Please cancel only the tracked reporter/watch tasks and add a test using the actual async-agent lifecycle.

4. [P2] Leave sufficient RPC headroom beyond the queue window

skywalking/utils/grpc_channel.py:102

For larger queue windows this returns agent_queue_timeout + 1. However, the sync generators subtract integer-truncated elapsed time and can spend almost queue_timeout + 1 seconds waiting for their final item.

The RPC deadline also has to cover protobuf conversion, transport, and the server response. A healthy call can therefore reach DEADLINE_EXCEEDED, after which the consumed batch is discarded.

Please either enforce an absolute batching deadline below the RPC deadline or add a meaningful fixed margin for RPC completion. The test should exercise the worst-case final queue.get(), rather than only asserting the arithmetic.

[songzhendong]

Thanks for the thorough review, All four points are addressed in the latest commit

1. [P1] Keep multi-backend collector channels excluded from gRPC instrumentation

Agent-owned collector channels are now marked explicitly instead of relying on target == config.agent_collector_backend_services. create_sync_channel() and create_aio_channel() run inside agent_collector_channel_scope() (thread-local) and mark the returned channel with _sw_agent_collector_channel. The sw_grpc sync and aio channel factories check is_building_agent_collector_channel() first, so rewritten multi-address targets (ipv4: / ipv6:) are not instrumented. Single-address behavior is unchanged; the old string-equality check remains as a fallback.

Regression tests with sw_grpc installed verify that a multi-address collector channel does not call grpc.intercept_channel, while a business channel still gets the interceptor. The aio path also confirms the build scope is active during collector channel creation.

2. [P1] Set gRPC fork support before importing grpc

Throttled reporter logging helpers were moved out of grpc_channel.py into a gRPC-neutral module skywalking/utils/reporter_log.py. skywalking/agent/__init__.py no longer imports grpc_channel at module load. start() and start_prefork_master() still set GRPC_ENABLE_FORK_SUPPORT=true before the first import grpc.

Subprocess-based import-order tests confirm that import skywalking.agent does not load grpc, and that after start_prefork_master(), grpc._cython.cygrpc.is_fork_support_enabled() is True.

3. [P2] Do not cancel the task owned by asyncio.run during shutdown

_cancel_pending_tasks() now cancels only tracked reporter and connectivity-watch tasks, not every task on the loop. SkyWalkingAgentAsync stores background_tasks explicitly, and __fini_async() cancels only that set. This preserves the asyncio.run root task so protocol.aclose() can complete.

A shutdown test using the real asyncio.run root-task lifecycle verifies that fini completes and is not cancelled by Runner teardown.

4. [P2] Leave sufficient RPC headroom beyond the queue window

Two changes were applied. First, the RPC deadline margin was increased from queue_timeout + 1 to queue_timeout + 5 (_GRPC_RPC_TIMEOUT_MARGIN_SEC). Second, sync report generators now use an absolute monotonic batch_deadline via _queue_get_within_batch(), avoiding int(elapsed) truncation that could let the final queue.get() consume almost the full RPC window.

Timeout assertions were updated for the widened deadline, and coverage was added to confirm sync collect() uses the new timeout value.

[wu-sheng]

Thanks for the update. I rechecked head d6c3761.

The collector-channel exclusion now works for both sync and aio, the fork-support import ordering is fixed, and positive queue timeouts now have sufficient RPC headroom. I still see two P2 issues:

1. Keep the asyncio.run root alive until async cleanup completes.

   Production still awaits asyncio.gather(*self.background_tasks) at skywalking/agent/__init__.py:827. When __fini_async() cancels those tasks at line 897, their cancellation finishes the root task. asyncio.run then tears down the Runner and can cancel __fini_async() while it is awaiting protocol aclose() at lines 899–903.

   I reproduced this with the actual lifecycle: aclose() was entered, yielded once, and never completed; both the shutdown future and root ended with CancelledError. The new test does not model production because its root uses gather(..., return_exceptions=True) and its post-cancellation marker is synchronous rather than a yielding aclose(). Adding return_exceptions=True alone is also insufficient—the root can still finish before aclose() does. Please coordinate root completion with shutdown completion, or perform cleanup inside the root task, and test the production topology with a yielding close.

2. Preserve the initial dequeue when SW_AGENT_QUEUE_TIMEOUT=0, or reject that value.

   _queue_get_within_batch() now returns as soon as remaining <= 0, before attempting Queue.get(). Zero is currently accepted with no positive-value validation. The previous implementation deliberately checked the queue once on the first iteration, and Queue.get(timeout=0) consumed an immediately available item. With the new code, every sync gRPC generator remains empty, queued telemetry is never drained, and a reporter can repeatedly issue empty streaming RPCs with zero backoff.

   Please either attempt one initial nonblocking dequeue or validate agent_queue_timeout as strictly positive, with a regression test for the chosen behavior.

All CI checks are green, including the failover jobs, and the focused unit suite passes; these two lifecycle/configuration cases are not covered by the current tests.

[songzhendong]

Thanks for the recheck. Both follow-up items from the latest review are addressed in feature/python-multi-backend (77bbfb9).

1. Keep the asyncio.run root alive until async cleanup completes

The async agent no longer runs shutdown cleanup from a separate coroutine submitted via run_coroutine_threadsafe. The asyncio.run root task now waits on _finished, then runs __async_shutdown_cleanup() (queue abandon, cancel tracked reporter/watch tasks, and protocol.aclose()) before returning. __fini() only signals shutdown with call_soon_threadsafe(self._finished.set) and joins the event-loop thread.

A regression test models the production topology: root waits on _finished, cancels background reporters, and awaits a yielding aclose() before asyncio.run returns.

2. Preserve the initial dequeue when SW_AGENT_QUEUE_TIMEOUT=0

_queue_get_within_batch() now accepts allow_immediate=True on the first generator iteration. When SW_AGENT_QUEUE_TIMEOUT=0, the first call still attempts Queue.get(timeout=0) so an immediately available item is drained; an empty queue returns without issuing an empty streaming RPC. This matches the previous first-iteration behavior.

Regression tests cover both the non-empty and empty queue cases.

CI

Fork CI is green at 77bbfb9, including License/Lint, unit tests, plugin tests, full E2E matrix, and failover jobs.

Please let me know if you'd like any of these adjusted before merge. Thanks again.

[wu-sheng]

Thanks for the update. The two follow-up issues are fixed, but the root-lifecycle change introduces one new P2 regression.

[P2] Observe unexpected background-task completion while waiting for shutdown

At skywalking/agent/__init__.py:822-827, the root creates background_tasks and then waits only on _finished. It no longer observes task completion. In addition, _cancel_pending_tasks() filters out tasks that are already done at lines 158-161, so their exceptions are not retrieved during shutdown either.

The concrete path is the unwrapped profile command dispatcher (__command_dispatch() at lines 983-986). If profile command conversion or execution raises, that task terminates permanently while the root, heartbeat, and telemetry reporters continue running. Profile commands then stop being processed with no agent error log, so the agent appears healthy.

I reproduced this on 77bbfb9 with the actual agent lifecycle:

thread_alive=True
_finished=False
exception_unretrieved=True
agent_error_logged=False

The underlying possibility of __command_dispatch() raising already existed, but this silent behavior is new in 77bbfb9. On d6c3761, the root's gather() observed the same failure, logged Error in Python agent asyncio event loop: ..., and exited. That behavior was not ideal either, but the failure was not silently ignored.

Please keep cleanup inside the root, while also observing both the shutdown event and unexpected background-task completion. Any completed task should have its result/exception retrieved and logged (and then be restarted or trigger orderly root cleanup). A regression test should verify that a failing background task cannot disappear silently.

[songzhendong]

Thanks for the recheck. The remaining P2 from the latest review is addressed in feature/python-multi-backend (662fd92).

Observe unexpected background-task completion while waiting for shutdown

The asyncio.run root still performs cleanup in-place, but no longer waits only on _finished. It now waits on both the shutdown event and the tracked background_tasks via _await_shutdown_or_background_failure(). If a background task finishes early (for example the unwrapped __command_dispatch()), its result/exception is retrieved and logged with Error in Python agent asyncio event loop: ..., then _finished is set so the root can run orderly cleanup (__async_shutdown_cleanup()).

_cancel_pending_tasks() also retrieves outcomes for tasks that are already done before cancellation, so exceptions are not skipped during shutdown.

A regression test verifies that a failing background task is logged and cannot disappear silently while the root continues waiting.

CI

Fork CI is green at 662fd92, including License/Lint, unit tests, plugin tests, full E2E matrix, and failover jobs.

[wu-sheng]

Thanks for the update. The original silent-exception path is now observed, root-owned cleanup runs, and a yielding aclose() completes. Two P2 cases remain because supervision and shutdown cleanup currently classify task outcomes the same way.

1. [P2] Clean shutdown logs normal reporter completion as an error

_cancel_pending_tasks() calls _log_background_task_outcome() for every task that is already done (skywalking/agent/__init__.py:218-243). _retrieve_background_task_outcome() converts a successful completion into RuntimeError('Python agent asyncio background task finished unexpectedly').

Reproduction scenario using the actual SkyWalkingAgentAsync root lifecycle:

1. Start the async root with two normal background reporters and a protocol whose aclose() yields.
2. Trigger normal shutdown through __fini(), which sets _finished.
3. Both reporters wake and return normally because _finished is set.
4. Root-owned cleanup reaches _cancel_pending_tasks() and finds both tasks already done.
5. Each successful completion is logged as an unexpected agent error.

Result, reproduced 30/30 runs:

thread exited: true
aclose completed: true
ERROR: Error in Python agent asyncio event loop:
       Python agent asyncio background task finished unexpectedly
ERROR: Error in Python agent asyncio event loop:
       Python agent asyncio background task finished unexpectedly

The same cleanup path also logs a real task exception twice: once when _await_shutdown_or_background_failure() observes it, and again when _cancel_pending_tasks() processes the already-done task.

During shutdown, a task that returned normally after _finished was set is expected. Cleanup should retrieve real exceptions without synthesizing an error for successful completion, and it should not re-log an outcome already handled by the supervisor.

2. [P2] Unexpected cancellation before shutdown remains silent

_retrieve_background_task_outcome() returns None for task.cancelled() at lines 150-157. The supervisor then discards that task at lines 196-200, does not log anything, and does not set _finished.

Reproduction scenario using the actual root lifecycle:

1. Start the root with one live reporter and one background task that cancels itself before shutdown:

   async def cancelled_background():
       asyncio.current_task().cancel()
       await asyncio.sleep(0)

2. Do not set _finished; this cancellation is not issued by shutdown cleanup.

3. The cancelled task completes and is returned by asyncio.wait().

4. The supervisor removes it, treats the outcome as absent, and continues waiting on the remaining reporter.

Result, reproduced 30/30 runs:

error logs: 0
_finished: false
aclose called: false
root thread alive after 200 ms: true

Therefore a cancelled reporter, command dispatcher, or connectivity watcher can disappear while the agent appears healthy. Cancellation observed by the supervisor before shutdown should be treated as unexpected and should trigger logging plus orderly root cleanup. Cancellation deliberately issued later by _cancel_pending_tasks() is expected and should remain silent.

Suggested regression coverage:

• clean shutdown completes with no background-task ERROR logs;
• one failed task is logged exactly once and aclose() completes;
• unexpected pre-shutdown cancellation is logged and triggers aclose();
• intentional cancellation during cleanup is not logged as an error.

The focused local suite passes (60 passed), but the added test exercises only the wait helper and does not run the subsequent cleanup path, so it does not catch these cases.

[songzhendong]

Thanks for the recheck. Both P2 follow-ups from the latest review are addressed in feature/python-multi-backend (97309ba).

1. Clean shutdown no longer logs normal reporter completion as an error

_cancel_pending_tasks() no longer treats successful completion as RuntimeError('... finished unexpectedly'). Cleanup only reports real exceptions (report_unexpected_completion=False), and outcomes already handled by the supervisor are marked so they are not logged again.

2. Unexpected cancellation before shutdown is observed

The shutdown supervisor now treats pre-shutdown task.cancelled() as unexpected (report_unexpected_cancellation=True): it logs Error in Python agent asyncio event loop: ..., sets _finished, and proceeds with root-owned cleanup. Cancellation issued later by _cancel_pending_tasks() remains silent.

Regression coverage now exercises the wait helper and the subsequent cleanup / aclose() path:

• clean shutdown completes with no background-task ERROR logs;
• one failed task is logged exactly once and aclose() completes;
• unexpected pre-shutdown cancellation is logged and triggers aclose();
• intentional cancellation during cleanup is not logged as an error.

CI

Fork CI is green at 2b699e5 (same tree as 97309ba), including License/Lint, unit tests, plugin tests, full E2E matrix, and failover jobs.