feat: native gRPC multi-backend failover via pick_first (#415)
diff --git a/.github/workflows/CI.yaml b/.github/workflows/CI.yaml
index 37fce06..2f2e956 100644
--- a/.github/workflows/CI.yaml
+++ b/.github/workflows/CI.yaml
@@ -305,13 +305,55 @@
name: e2e_logs_${{ matrix.case.name }}_${{ matrix.python-image-variant }}
path: "${{ env.SW_INFRA_E2E_LOG_DIR }}"
+ # Separate from the main E2E matrix so the 20m job timeout stays tight for
+ # ordinary cases; failover pulls two OAPs and needs its own headroom.
+ # Span oldest/mid/newest CPython in the support window — pick_first / IPv6
+ # encoding / default_authority are grpcio C-core sensitive.
+ e2e-failover:
+ name: E2E Failover
+ needs: [ license-and-lint, changes, plugin-doc-check, docker-e2e ]
+ if: |
+ ( always() && ! cancelled() ) &&
+ ((github.event_name == 'schedule' && github.repository == 'apache/skywalking-python') || needs.changes.outputs.agent == 'true')
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ strategy:
+ matrix:
+ python-image-variant: [ "3.10-slim", "3.12-slim", "3.14-slim" ]
+ fail-fast: false
+ steps:
+ - name: Checkout source codes
+ uses: actions/checkout@v4
+ with:
+ submodules: true
+ persist-credentials: false
+ - name: Pull SkyWalking Python agent base image
+ uses: actions/download-artifact@v4
+ with:
+ name: docker-images-skywalking-python-e2e-${{ matrix.python-image-variant }}
+ path: docker-images
+ - name: Load docker images
+ run: find docker-images -name "*.tar" -exec docker load -i {} \;
+ - name: Run Failover E2E
+ uses: apache/skywalking-infra-e2e@cf589b4a0b9f8e6f436f78e9cfd94a1ee5494180
+ with:
+ log-dir: /tmp/e2e-logs
+ e2e-file: tests/e2e/case/grpc/failover/e2e.yaml
+ - name: Upload Logs
+ uses: actions/upload-artifact@v4
+ if: ${{ failure() }}
+ with:
+ name: e2e_logs_gRPC-failover_${{ matrix.python-image-variant }}
+ path: "${{ env.SW_INFRA_E2E_LOG_DIR }}"
+
+
CheckStatus:
# a required check that must pass before merging a PR
runs-on: ubuntu-latest
timeout-minutes: 60
# wait upon them regardless of success or skipped or failure
if: ${{ always() }}
- needs: [ license-and-lint, changes, plugin-and-unit-tests, plugin-doc-check, e2e-tests ]
+ needs: [ license-and-lint, changes, plugin-and-unit-tests, plugin-doc-check, e2e-tests, e2e-failover ]
steps:
- name: Merge Requirement
# check license, lint, plugin and e2e tests, then naturally exits 0
@@ -320,9 +362,11 @@
lintResults=${{ needs.license-and-lint.result }}
pluginResults=${{ needs.plugin-and-unit-tests.result }};
e2eResults=${{ needs.e2e-tests.result }};
+ e2eFailoverResults=${{ needs.e2e-failover.result }};
docCheckResults=${{ needs.plugin-doc-check.result }};
[[ ${lintResults} == 'success' ]] || exit 2;
[[ ${docCheckResults} == 'success' ]] || exit 3;
[[ ${pluginResults} == 'success' ]] || [[ ${execute} != 'true' && ${pluginResults} == 'skipped' ]] || exit 4;
[[ ${e2eResults} == 'success' ]] || [[ ${execute} != 'true' && ${e2eResults} == 'skipped' ]] || exit 5;
+ [[ ${e2eFailoverResults} == 'success' ]] || [[ ${execute} != 'true' && ${e2eFailoverResults} == 'skipped' ]] || exit 6;
exit 0;
diff --git a/docs/en/setup/Configuration.md b/docs/en/setup/Configuration.md
index bfefb89..7d521e4 100644
--- a/docs/en/setup/Configuration.md
+++ b/docs/en/setup/Configuration.md
@@ -17,7 +17,7 @@
### Agent Core Configuration Options
| Configuration | Environment Variable | Type | Default Value | Description |
| :------------ | :------------ | :------------ | :------------ | :------------ |
-| agent_collector_backend_services | SW_AGENT_COLLECTOR_BACKEND_SERVICES | <class 'str'> | oap_host:oap_port | The backend OAP server address, 11800 is default OAP gRPC port, 12800 is HTTP, Kafka ignores this option and uses kafka_bootstrap_servers option. **This option should be changed accordingly with selected protocol** |
+| agent_collector_backend_services | SW_AGENT_COLLECTOR_BACKEND_SERVICES | <class 'str'> | oap_host:oap_port | The backend OAP server address(es). 11800 is default OAP gRPC port, 12800 is HTTP, Kafka ignores this option and uses kafka_bootstrap_servers option. For gRPC, a comma-separated list enables native pick_first failover (per-process shuffle of the preferred backend; :authority stays the first configured endpoint). See Intrusive.md for encoding, DNS, TLS authority, proxy, and READY-gate details. **This option should be changed accordingly with selected protocol** |
| agent_protocol | SW_AGENT_PROTOCOL | <class 'str'> | grpc | The protocol to communicate with the backend OAP, `http`, `grpc` or `kafka`, **we highly suggest using `grpc` in production as it's well optimized than `http`**. The `kafka` protocol provides an alternative way to submit data to the backend. |
| agent_name | SW_AGENT_NAME | <class 'str'> | Python Service Name | The name of your awesome Python service |
| agent_instance_name | SW_AGENT_INSTANCE_NAME | <class 'str'> | str(uuid.uuid1()).replace('-', '') | The name of this particular awesome Python service instance |
@@ -29,14 +29,14 @@
| kafka_topic_log | SW_KAFKA_TOPIC_LOG | <class 'str'> | skywalking-logs | Specifying Kafka topic name for Log data, this should be in sync with OAP |
| kafka_topic_meter | SW_KAFKA_TOPIC_METER | <class 'str'> | skywalking-meters | Specifying Kafka topic name for Meter data, this should be in sync with OAP |
| kafka_reporter_custom_configurations | SW_KAFKA_REPORTER_CUSTOM_CONFIGURATIONS | <class 'str'> | | The configs to init KafkaProducer, supports the basic arguments (whose type is either `str`, `bool`, or `int`) listed [here](https://kafka-python.readthedocs.io/en/master/apidoc/KafkaProducer.html#kafka.KafkaProducer) This config only works from env variables, each one should be passed in `SW_KAFKA_REPORTER_CONFIG_<KEY_NAME>` |
-| agent_force_tls | SW_AGENT_FORCE_TLS | <class 'bool'> | False | Use TLS for communication with SkyWalking OAP (no cert required) |
+| agent_force_tls | SW_AGENT_FORCE_TLS | <class 'bool'> | False | Use TLS for gRPC/HTTP with the OAP (no client cert required). For gRPC, ensure the server certificate SAN matches the first usable backend in agent_collector_backend_services (used as grpc.default_authority). |
| agent_authentication | SW_AGENT_AUTHENTICATION | <class 'str'> | | The authentication token to verify that the agent is trusted by the backend OAP, as for how to configure the backend, refer to [the yaml](https://github.com/apache/skywalking/blob/4f0f39ffccdc9b41049903cc540b8904f7c9728e/oap-server/server-bootstrap/src/main/resources/application.yml#L155-L158). |
| agent_logging_level | SW_AGENT_LOGGING_LEVEL | <class 'str'> | INFO | The level of agent self-logs, could be one of `CRITICAL`, `FATAL`, `ERROR`, `WARN`(`WARNING`), `INFO`, `DEBUG`. Please turn on debug if an issue is encountered to find out what's going on |
### Agent Core Danger Zone
| Configuration | Environment Variable | Type | Default Value | Description |
| :------------ | :------------ | :------------ | :------------ | :------------ |
| agent_collector_heartbeat_period | SW_AGENT_COLLECTOR_HEARTBEAT_PERIOD | <class 'int'> | 30 | The agent will exchange heartbeat message with SkyWalking OAP backend every `period` seconds |
-| agent_collector_properties_report_period_factor | SW_AGENT_COLLECTOR_PROPERTIES_REPORT_PERIOD_FACTOR | <class 'int'> | 10 | The agent will report service instance properties every `factor * heartbeat period` seconds default: 10*30 = 300 seconds |
+| agent_collector_properties_report_period_factor | SW_AGENT_COLLECTOR_PROPERTIES_REPORT_PERIOD_FACTOR | <class 'int'> | 10 | The agent will report service instance properties every `factor * heartbeat period` seconds default: 10*30 = 300 seconds (Java/Node cadence). Also covers gRPC pick_first silent backend switches that stay READY without a disconnect event. |
| agent_instance_properties_json | SW_AGENT_INSTANCE_PROPERTIES_JSON | <class 'str'> | | A custom JSON string to be reported as service instance properties, e.g. `{"key": "value"}` |
| agent_experimental_fork_support | SW_AGENT_EXPERIMENTAL_FORK_SUPPORT | <class 'bool'> | False | The agent will restart itself in any os.fork()-ed child process. Important Note: it's not suitable for short-lived processes as each one will create a new instance in SkyWalking dashboard in format of `service_instance-child(pid)`. When the sw-python CLI detects a pre-forking server (Gunicorn), only worker processes run a full agent; the master installs instrumentation only. |
| agent_queue_timeout | SW_AGENT_QUEUE_TIMEOUT | <class 'int'> | 1 | DANGEROUS - This option controls the interval of each bulk report from telemetry data queues Do not modify unless you have evaluated its impact given your service load. |
diff --git a/docs/en/setup/Intrusive.md b/docs/en/setup/Intrusive.md
index 089579e..d006889 100644
--- a/docs/en/setup/Intrusive.md
+++ b/docs/en/setup/Intrusive.md
@@ -27,6 +27,39 @@
agent.start()
```
+#### gRPC multi-address (failover)
+
+Pass a comma-separated list. The agent opens **one** gRPC channel and lets C-core `pick_first` fail over (same idea as Node `sw-static`). Each process shuffles the preferred backend at channel build; `:authority` / TLS SNI still use the **first configured** endpoint.
+
+```python
+config.init(
+ agent_collector_backend_services='oap-a:11800,oap-b:11800',
+ agent_name='your awesome service',
+)
+agent.start()
+```
+
+Implementation notes (maintainers / operators):
+
+- Mixed IPv4/IPv6 stays in one list; IPv4 is encoded as IPv4-mapped IPv6 for grpcio so `pick_first` can try both families.
+- Multi-hostname lists are DNS-expanded once at channel build (about 5s lookup budget per name); there is no periodic re-resolve — prefer a single address or stable IPs when DNS changes.
+- Channel `:authority` / TLS SAN uses `grpc.default_authority` = the first configured endpoint (before shuffle). With `agent_force_tls`, every backend cert must cover that authority.
+- Reporters wait until the channel is READY; non-READY skips the RPC rather than failing fast into a black hole.
+- After a silent backend switch that stays READY, instance properties are re-reported on the normal properties period so the new OAP learns the instance.
+- Reconnect backoff caps at 30s.
+
+#### gRPC HTTP proxy (behavior change)
+
+`grpc.enable_http_proxy=0` is set on **every** gRPC channel, including a single-address config. Host `http_proxy` / `https_proxy` / `no_proxy` are ignored for OAP traffic.
+
+If you currently reach OAP only through an HTTP CONNECT proxy, upgrading this agent will lose that path. The agent disables the proxy so application proxy env vars cannot silently black-hole telemetry, and because HTTP CONNECT happens before name resolution — it cannot express a comma-separated backend list.
+
+#### Best-effort reporting and buffers
+
+Send failures are discarded and never retried (a failed batch is dropped). On process shutdown the agent flushes only while the channel is READY, with a short time budget, then **abandons** whatever is still queued.
+
+When a reporter queue is full (`SW_AGENT_TRACE_REPORTER_MAX_BUFFER_SIZE` / log / meter / snapshot equivalents), new items are dropped. The agent logs drops at most once per 30 seconds, with the increment since the last line and the process total. Raise the buffer if you routinely see these under load; a full queue during an outage is expected because the READY gate holds data until the backend returns.
+
### Report data via HTTP protocol
However, if you want to use HTTP protocol to report data, configure `agent_collector_backend_services`
diff --git a/skywalking/agent/__init__.py b/skywalking/agent/__init__.py
index 4e208eb..7a1d5a4 100644
--- a/skywalking/agent/__init__.py
+++ b/skywalking/agent/__init__.py
@@ -20,9 +20,10 @@
import functools
import os
import sys
-from queue import Full, Queue
+import time
+from queue import Empty, Full, Queue
from threading import Event, Thread
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Callable, Optional
from skywalking import config, loggings, meter, plugins, profile, sampling
from skywalking.agent.protocol import Protocol, ProtocolAsync
@@ -32,6 +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.reporter_log import log_dropped_throttled, log_reporter_exception_throttled
from skywalking.utils.singleton import Singleton
if TYPE_CHECKING:
@@ -41,6 +43,281 @@
import uvloop
uvloop.install()
+# Shutdown must never block the host indefinitely (Node flush budget parity).
+_SHUTDOWN_FLUSH_TIMEOUT_SEC = 2.0
+_SHUTDOWN_JOIN_TIMEOUT_SEC = 2.0
+_SHUTDOWN_LOOP_JOIN_TIMEOUT_SEC = 3.0
+
+
+def _abandon_sync_queue(q: Queue) -> int:
+ """Drop queued items and balance unfinished_tasks so Queue.join() can finish."""
+ abandoned = 0
+ while True:
+ try:
+ q.get_nowait()
+ q.task_done()
+ abandoned += 1
+ except Empty:
+ break
+ return abandoned
+
+
+def _join_sync_queue(q: Queue, timeout: float) -> bool:
+ """Return True if join completed within timeout."""
+ done = Event()
+
+ def _wait():
+ q.join()
+ done.set()
+
+ Thread(target=_wait, name='sw-queue-join', daemon=True).start()
+ return done.wait(timeout)
+
+
+def _shutdown_sync_queue(
+ report_fn: Optional[Callable[[], None]],
+ q: Queue,
+ label: str,
+ *,
+ may_send: bool,
+) -> None:
+ """
+ Best-effort shutdown drain: optional timed flush when READY, then abandon + timed join.
+ Never blocks longer than flush+join budgets (prevents atexit hang when not READY).
+ """
+ if may_send and report_fn is not None and not q.empty():
+ done = Event()
+
+ def _flush():
+ try:
+ report_fn()
+ except Exception: # noqa: BLE001 - shutdown must continue
+ logger.exception('shutdown flush failed for %s queue', label)
+ finally:
+ done.set()
+
+ Thread(target=_flush, name=f'sw-shutdown-flush-{label}', daemon=True).start()
+ if not done.wait(_SHUTDOWN_FLUSH_TIMEOUT_SEC):
+ logger.warning(
+ 'shutdown flush timed out after %.1fs for %s queue; abandoning remainder',
+ _SHUTDOWN_FLUSH_TIMEOUT_SEC,
+ label,
+ )
+
+ abandoned = _abandon_sync_queue(q)
+ if abandoned:
+ log_dropped_throttled(label, abandoned, force=True)
+
+ if not _join_sync_queue(q, _SHUTDOWN_JOIN_TIMEOUT_SEC):
+ logger.warning(
+ 'shutdown join timed out after %.1fs for %s queue (unfinished_tasks=%s)',
+ _SHUTDOWN_JOIN_TIMEOUT_SEC,
+ label,
+ getattr(q, 'unfinished_tasks', '?'),
+ )
+
+
+async def _abandon_async_queue(q: asyncio.Queue) -> int:
+ abandoned = 0
+ while True:
+ try:
+ q.get_nowait()
+ q.task_done()
+ abandoned += 1
+ except asyncio.QueueEmpty:
+ break
+ return abandoned
+
+
+async def _shutdown_async_queue(q: asyncio.Queue, label: str) -> None:
+ """
+ Async reporters use unbounded queue.get() generators — do not await report() on
+ shutdown (can hang forever). Abandon + timed join, then caller cancels tasks.
+ """
+ abandoned = await _abandon_async_queue(q)
+ if abandoned:
+ log_dropped_throttled(label, abandoned, force=True)
+ try:
+ await asyncio.wait_for(q.join(), timeout=_SHUTDOWN_JOIN_TIMEOUT_SEC)
+ except asyncio.TimeoutError:
+ logger.warning(
+ 'shutdown join timed out after %.1fs for async %s queue',
+ _SHUTDOWN_JOIN_TIMEOUT_SEC,
+ label,
+ )
+
+
+def _retrieve_background_task_outcome(
+ task: asyncio.Task,
+ *,
+ report_unexpected_completion: bool = True,
+ report_unexpected_cancellation: bool = False,
+):
+ """
+ Return a completed background task's exception when it should be reported.
+
+ Retrieves the exception so asyncio does not emit "never retrieved" warnings.
+ Successful completion / cancellation are only converted to errors when the
+ corresponding report_* flag is set (supervisor path before shutdown).
+ """
+ if task is None or not task.done():
+ return None
+ if task.cancelled():
+ if report_unexpected_cancellation:
+ return RuntimeError(
+ 'Python agent asyncio background task was cancelled unexpectedly'
+ )
+ return None
+ exc = task.exception()
+ if exc is not None:
+ return exc
+ if report_unexpected_completion:
+ return RuntimeError('Python agent asyncio background task finished unexpectedly')
+ return None
+
+
+def _log_background_task_outcome(
+ task: asyncio.Task,
+ *,
+ report_unexpected_completion: bool = True,
+ report_unexpected_cancellation: bool = False,
+) -> bool:
+ """Log and retrieve a completed background task outcome. Returns True if logged."""
+ if getattr(task, '_sw_outcome_handled', False):
+ return False
+ exc = _retrieve_background_task_outcome(
+ task,
+ report_unexpected_completion=report_unexpected_completion,
+ report_unexpected_cancellation=report_unexpected_cancellation,
+ )
+ if exc is None:
+ return False
+ # Mark before logging so cleanup never re-logs a supervisor-handled outcome.
+ task._sw_outcome_handled = True
+ logger.error('Error in Python agent asyncio event loop: %s', exc, exc_info=exc)
+ return True
+
+
+async def _await_shutdown_or_background_failure(
+ finished: asyncio.Event,
+ background_tasks,
+) -> None:
+ """
+ Wait for shutdown or the first unexpected background-task completion.
+
+ Background tasks are expected to run until shutdown. If one finishes or is
+ cancelled early, retrieve/log its outcome and signal shutdown so the root
+ can clean up.
+ """
+ shutdown_waiter = asyncio.create_task(finished.wait())
+ pending = {task for task in background_tasks if task is not None}
+ try:
+ while not finished.is_set():
+ if not pending:
+ await finished.wait()
+ return
+ done, _ = await asyncio.wait(
+ pending | {shutdown_waiter},
+ return_when=asyncio.FIRST_COMPLETED,
+ )
+ if shutdown_waiter in done or finished.is_set():
+ return
+ for task in done:
+ pending.discard(task)
+ # Before shutdown, both early success and unexpected cancel are errors.
+ if _log_background_task_outcome(
+ task,
+ report_unexpected_completion=True,
+ report_unexpected_cancellation=True,
+ ):
+ finished.set()
+ return
+ finally:
+ if not shutdown_waiter.done():
+ shutdown_waiter.cancel()
+ try:
+ await shutdown_waiter
+ except asyncio.CancelledError:
+ pass
+
+
+async def _cancel_pending_tasks(tasks) -> None:
+ """
+ Cancel agent-owned reporter / connectivity-watch tasks only.
+
+ Cancel only the given reporter / watch tasks; never the asyncio.run root.
+ The current task is excluded so we never await ourselves.
+
+ During cleanup, normal completion and intentional cancellation are silent;
+ only real exceptions that were not already handled by the supervisor are logged.
+ """
+ current = asyncio.current_task()
+ for task in tasks:
+ if task is None or task is current:
+ continue
+ if task.done():
+ _log_background_task_outcome(
+ task,
+ report_unexpected_completion=False,
+ report_unexpected_cancellation=False,
+ )
+ 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:
+ task.cancel()
+ try:
+ await asyncio.wait_for(
+ asyncio.gather(*pending, return_exceptions=True),
+ timeout=_SHUTDOWN_JOIN_TIMEOUT_SEC,
+ )
+ except asyncio.TimeoutError:
+ logger.warning(
+ 'shutdown task cancellation timed out after %.1fs; %d task(s) still pending',
+ _SHUTDOWN_JOIN_TIMEOUT_SEC,
+ sum(1 for task in pending if not task.done()),
+ )
+ for task in pending:
+ _log_background_task_outcome(
+ task,
+ report_unexpected_completion=False,
+ report_unexpected_cancellation=False,
+ )
+
+
+def _close_previous_protocol(protocol) -> None:
+ """Close a replaced protocol channel (fork re-bootstrap / defensive re-open)."""
+ if protocol is None:
+ return
+ close = getattr(protocol, 'close', None)
+ if callable(close):
+ try:
+ close()
+ except Exception: # noqa: BLE001
+ logger.exception('failed to close previous protocol channel')
+
+
+async def _aclose_previous_protocol(protocol) -> None:
+ """Await aclose on the running loop. Never run_coroutine_threadsafe().result() here."""
+ if protocol is None:
+ return
+ aclose = getattr(protocol, 'aclose', None)
+ if callable(aclose):
+ try:
+ await aclose()
+ return
+ except Exception: # noqa: BLE001
+ logger.exception('failed to aclose previous protocol channel')
+ close = getattr(protocol, 'close', None)
+ if callable(close):
+ try:
+ close()
+ except Exception: # noqa: BLE001
+ logger.exception('failed to close previous protocol channel')
+
def report_with_backoff(reporter_name, init_wait):
"""
@@ -59,8 +336,7 @@
wait = 0 if flag else base
except Exception: # noqa
wait = min(60, wait * 2 or 1) # double wait time with each consecutive error up to a maximum
- logger.exception(f'Exception in {reporter_name} service in pid {os.getpid()}, '
- f'retry in {wait} seconds')
+ log_reporter_exception_throttled(reporter_name, wait)
self._finished.wait(wait)
logger.info('finished reporter thread')
@@ -86,9 +362,16 @@
wait = 0 if flag else base
except Exception: # noqa
wait = min(60, wait * 2 or 1) # double wait time with each consecutive error up to a maximum
- logger.exception(f'Exception in {reporter_name} service in pid {os.getpid()}, '
- f'retry in {wait} seconds')
- await asyncio.sleep(wait)
+ log_reporter_exception_throttled(reporter_name, wait)
+ # Prefer Event.wait so shutdown (_finished.set) wakes immediately;
+ # plain sleep(wait) would otherwise block up to heartbeat/backoff period.
+ if wait:
+ try:
+ await asyncio.wait_for(self._finished.wait(), timeout=wait)
+ except asyncio.TimeoutError:
+ pass
+ else:
+ await asyncio.sleep(0)
logger.info('finished reporter coroutine')
return backoff_wrapper
@@ -126,12 +409,15 @@
if config.agent_protocol == 'grpc':
from skywalking.agent.protocol.grpc import GrpcProtocol
+ _close_previous_protocol(self.__protocol)
self.__protocol = GrpcProtocol()
elif config.agent_protocol == 'http':
from skywalking.agent.protocol.http import HttpProtocol
+ _close_previous_protocol(self.__protocol)
self.__protocol = HttpProtocol()
elif config.agent_protocol == 'kafka':
from skywalking.agent.protocol.kafka import KafkaProtocol
+ _close_previous_protocol(self.__protocol)
self.__protocol = KafkaProtocol()
# Start reporter threads and register queues
@@ -373,25 +659,51 @@
"""
This method is called when the agent is shutting down.
Clean up all the queues and threads.
+
+ Stop reporter loops first, then best-effort flush (timed) when READY, else
+ abandon queued items so Queue.join() cannot hang forever (READY-gate gap).
"""
if not self.__reporting: # never bootstrapped in this process (e.g. pre-fork master)
return
- self.__protocol.report_segment(self.__segment_queue, False)
- self.__segment_queue.join()
+
+ # Wake backoff loops immediately; do not leave finished.set() until after joins.
+ self._finished.set()
+ may_send = self.__protocol.is_ready()
+
+ _shutdown_sync_queue(
+ (lambda: self.__protocol.report_segment(self.__segment_queue, False)) if may_send else None,
+ self.__segment_queue,
+ 'segment',
+ may_send=may_send,
+ )
if config.agent_log_reporter_active:
- self.__protocol.report_log(self.__log_queue, False)
- self.__log_queue.join()
+ _shutdown_sync_queue(
+ (lambda: self.__protocol.report_log(self.__log_queue, False)) if may_send else None,
+ self.__log_queue,
+ 'log',
+ may_send=may_send,
+ )
if config.agent_profile_active:
- self.__protocol.report_snapshot(self.__snapshot_queue, False)
- self.__snapshot_queue.join()
+ _shutdown_sync_queue(
+ (lambda: self.__protocol.report_snapshot(self.__snapshot_queue, False)) if may_send else None,
+ self.__snapshot_queue,
+ 'snapshot',
+ may_send=may_send,
+ )
if config.agent_meter_reporter_active:
- self.__protocol.report_meter(self.__meter_queue, False)
- self.__meter_queue.join()
+ _shutdown_sync_queue(
+ (lambda: self.__protocol.report_meter(self.__meter_queue, False)) if may_send else None,
+ self.__meter_queue,
+ 'meter',
+ may_send=may_send,
+ )
- self._finished.set()
+ close = getattr(self.__protocol, 'close', None)
+ if callable(close):
+ close()
def stop(self) -> None:
"""
@@ -404,15 +716,24 @@
self.__started = False
@report_with_backoff(reporter_name='heartbeat', init_wait=config.agent_collector_heartbeat_period)
- def __heartbeat(self) -> None:
+ def __heartbeat(self) -> bool:
+ # Until subscribe reports READY, retry soon (do not wait a full heartbeat period).
+ if not self.__protocol.is_ready():
+ time.sleep(0.5)
+ return True
self.__protocol.heartbeat()
+ return False
# segment/log init_wait is set to 0.02 to prevent threads from hogging the cpu too much
# The value of 0.02(20 ms) is set to be consistent with the queue delay of the Java agent
@report_with_backoff(reporter_name='segment', init_wait=0.02)
def __report_segment(self) -> bool:
- """Returns True if the queue is not empty"""
+ """Returns True if the queue is not empty and a send was attempted."""
+ # Not READY: sleep then wait=0 — avoid 20ms busy-loop + IDLE nudge thrash.
+ if not self.__protocol.is_ready():
+ time.sleep(0.5)
+ return True
queue_not_empty_flag = not self.__segment_queue.empty()
if queue_not_empty_flag:
self.__protocol.report_segment(self.__segment_queue)
@@ -420,7 +741,10 @@
@report_with_backoff(reporter_name='log', init_wait=0.02)
def __report_log(self) -> bool:
- """Returns True if the queue is not empty"""
+ """Returns True if the queue is not empty and a send was attempted."""
+ if not self.__protocol.is_ready():
+ time.sleep(0.5)
+ return True
queue_not_empty_flag = not self.__log_queue.empty()
if queue_not_empty_flag:
self.__protocol.report_log(self.__log_queue)
@@ -428,11 +752,17 @@
@report_with_backoff(reporter_name='meter', init_wait=config.agent_meter_reporter_period)
def __report_meter(self) -> None:
+ if not self.__protocol.is_ready():
+ time.sleep(0.5)
+ return
if not self.__meter_queue.empty():
self.__protocol.report_meter(self.__meter_queue)
@report_with_backoff(reporter_name='profile_snapshot', init_wait=0.5)
def __send_profile_snapshot(self) -> None:
+ if not self.__protocol.is_ready():
+ time.sleep(0.5)
+ return
if not self.__snapshot_queue.empty():
self.__protocol.report_snapshot(self.__snapshot_queue)
@@ -464,7 +794,7 @@
try: # unlike checking __queue.full() then inserting, this is atomic
self.__segment_queue.put(segment, block=False)
except Full:
- logger.warning('the queue is full, the segment will be abandoned')
+ log_dropped_throttled('segment')
def archive_log(self, log_data: 'LogData'):
if not self.__reporting:
@@ -472,7 +802,7 @@
try:
self.__log_queue.put(log_data, block=False)
except Full:
- logger.warning('the queue is full, the log will be abandoned')
+ log_dropped_throttled('log')
def archive_meter(self, meter_data: 'MeterData'):
if not self.__reporting:
@@ -480,15 +810,15 @@
try:
self.__meter_queue.put(meter_data, block=False)
except Full:
- logger.warning('the queue is full, the meter will be abandoned')
+ log_dropped_throttled('meter')
def add_profiling_snapshot(self, snapshot: TracingThreadSnapshot):
if not self.__reporting:
return
try:
- self.__snapshot_queue.put(snapshot)
+ self.__snapshot_queue.put_nowait(snapshot)
except Full:
- logger.warning('the snapshot queue is full, the snapshot will be abandoned')
+ log_dropped_throttled('snapshot')
def notify_profile_finish(self, task: ProfileTask):
try:
@@ -516,7 +846,8 @@
self.event_loop_thread: Optional[Thread] = None
- def __bootstrap(self):
+ async def __bootstrap(self):
+ await _aclose_previous_protocol(self.__protocol)
if config.agent_protocol == 'grpc':
from skywalking.agent.protocol.grpc_aio import GrpcProtocolAsync
self.__protocol = GrpcProtocolAsync()
@@ -544,6 +875,10 @@
self.background_coroutines = set()
+ watch = getattr(self.__protocol, 'watch_connectivity', None)
+ if callable(watch):
+ self.background_coroutines.add(watch())
+
self.background_coroutines.add(self.__heartbeat())
self.background_coroutines.add(self.__report_segment())
@@ -594,10 +929,15 @@
if config.sample_n_per_3_secs > 0:
await sampling.init_async()
- self.__bootstrap() # gather all coroutines
+ await self.__bootstrap() # gather all coroutines
+ 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)
+ # Wait for shutdown or unexpected background-task completion inside the
+ # asyncio.run root, then clean up here so the Runner stays alive through
+ # protocol aclose() before asyncio.run returns.
+ await _await_shutdown_or_background_failure(self._finished, self.background_tasks)
+ await self.__async_shutdown_cleanup()
def __start_event_loop(self) -> None:
try:
@@ -645,35 +985,52 @@
self.event_loop_thread = Thread(name='event_loop_thread', target=self.__start_event_loop, daemon=True)
self.event_loop_thread.start()
- async def __fini_async(self):
+ async def __async_shutdown_cleanup(self) -> None:
"""
- This method is called when the agent is shutting down.
- Clean up all the queues and stop all the asyncio tasks.
+ Async shutdown body that must run on the asyncio.run root task.
+
+ Do not await report_* here: aio generators use unbounded queue.get() and can
+ hang forever. Abandon + timed join, then cancel tasks.
"""
- if self._finished is not None:
- self._finished.set()
- queue_join_coroutine_list = [self.__segment_queue.join()]
+ await _shutdown_async_queue(self.__segment_queue, 'segment')
if config.agent_log_reporter_active:
- queue_join_coroutine_list.append(self.__log_queue.join())
+ await _shutdown_async_queue(self.__log_queue, 'log')
if config.agent_profile_active:
- queue_join_coroutine_list.append(self.__snapshot_queue.join())
+ await _shutdown_async_queue(self.__snapshot_queue, 'snapshot')
if config.agent_meter_reporter_active:
- queue_join_coroutine_list.append(self.__meter_queue.join())
+ await _shutdown_async_queue(self.__meter_queue, 'meter')
- await asyncio.gather(*queue_join_coroutine_list, return_exceptions=True) # clean queues
- # cancel all tasks
- all_tasks = asyncio.all_tasks(self.loop)
- for task in all_tasks:
- task.cancel()
+ await _cancel_pending_tasks(getattr(self, 'background_tasks', ()))
+
+ aclose = getattr(self.__protocol, 'aclose', None)
+ if callable(aclose):
+ try:
+ await aclose()
+ except Exception: # noqa: BLE001
+ pass
+ else:
+ close = getattr(self.__protocol, 'close', None)
+ if callable(close):
+ close()
def __fini(self):
- if not self.loop.is_closed():
- asyncio.run_coroutine_threadsafe(self.__fini_async(), self.loop)
- self.event_loop_thread.join()
- logger.info('Finished Python agent event_loop thread')
+ loop = getattr(self, 'loop', None)
+ if loop is not None and not loop.is_closed() and self._finished is not None:
+ loop.call_soon_threadsafe(self._finished.set)
+ if self.event_loop_thread is not None:
+ self.event_loop_thread.join(
+ timeout=_SHUTDOWN_LOOP_JOIN_TIMEOUT_SEC + _SHUTDOWN_JOIN_TIMEOUT_SEC * 4,
+ )
+ if self.event_loop_thread.is_alive():
+ logger.warning(
+ 'Python agent event_loop thread still alive after %.1fs shutdown budget',
+ _SHUTDOWN_LOOP_JOIN_TIMEOUT_SEC,
+ )
+ else:
+ logger.info('Finished Python agent event_loop thread')
# TODO: Unhandled error in sys.excepthook https://github.com/pytest-dev/execnet/issues/30
def stop(self) -> None:
@@ -685,12 +1042,19 @@
self.__started = False
@report_with_backoff_async(reporter_name='heartbeat', init_wait=config.agent_collector_heartbeat_period)
- async def __heartbeat(self) -> None:
+ async def __heartbeat(self) -> bool:
+ if not self.__protocol.is_ready():
+ await asyncio.sleep(0.5)
+ return True
await self.__protocol.heartbeat()
+ return False
@report_with_backoff_async(reporter_name='segment', init_wait=0.02)
async def __report_segment(self) -> bool:
- """Returns True if the queue is not empty"""
+ """Returns True if the queue is not empty and a send was attempted."""
+ if not self.__protocol.is_ready():
+ await asyncio.sleep(0.5)
+ return True
queue_not_empty_flag = not self.__segment_queue.empty()
if queue_not_empty_flag:
await self.__protocol.report_segment(self.__segment_queue)
@@ -698,7 +1062,10 @@
@report_with_backoff_async(reporter_name='log', init_wait=0.02)
async def __report_log(self) -> bool:
- """Returns True if the queue is not empty"""
+ """Returns True if the queue is not empty and a send was attempted."""
+ if not self.__protocol.is_ready():
+ await asyncio.sleep(0.5)
+ return True
queue_not_empty_flag = not self.__log_queue.empty()
if queue_not_empty_flag:
await self.__protocol.report_log(self.__log_queue)
@@ -706,11 +1073,17 @@
@report_with_backoff_async(reporter_name='meter', init_wait=config.agent_meter_reporter_period)
async def __report_meter(self) -> None:
+ if not self.__protocol.is_ready():
+ await asyncio.sleep(0.5)
+ return
if not self.__meter_queue.empty():
await self.__protocol.report_meter(self.__meter_queue)
@report_with_backoff_async(reporter_name='profile_snapshot', init_wait=0.5)
async def __send_profile_snapshot(self) -> None:
+ if not self.__protocol.is_ready():
+ await asyncio.sleep(0.5)
+ return
if not self.__snapshot_queue.empty():
await self.__protocol.report_snapshot(self.__snapshot_queue)
@@ -729,7 +1102,7 @@
try:
q.put_nowait(item)
except asyncio.QueueFull:
- logger.warning(f'the {queue_name} queue is full, the item will be abandoned')
+ log_dropped_throttled(queue_name)
def is_segment_queue_full(self):
return self.__segment_queue.full()
@@ -750,7 +1123,7 @@
try:
self.__meter_queue.put_nowait(meter_data)
except asyncio.QueueFull:
- logger.warning('the meter queue is full, the item will be abandoned')
+ log_dropped_throttled('meter')
def add_profiling_snapshot(self, snapshot: TracingThreadSnapshot):
self.loop.call_soon_threadsafe(self.__asyncio_queue_put_nowait, self.__snapshot_queue, 'snapshot', snapshot)
diff --git a/skywalking/agent/protocol/__init__.py b/skywalking/agent/protocol/__init__.py
index 556c9d6..2744c07 100644
--- a/skywalking/agent/protocol/__init__.py
+++ b/skywalking/agent/protocol/__init__.py
@@ -21,6 +21,13 @@
class Protocol(ABC):
+ def is_ready(self) -> bool:
+ """
+ Whether the reporter may send RPCs this tick.
+ gRPC overrides with channel READY gate; HTTP/Kafka stay always-ready.
+ """
+ return True
+
@abstractmethod
def heartbeat(self):
raise NotImplementedError()
@@ -51,6 +58,10 @@
class ProtocolAsync(ABC):
+ def is_ready(self) -> bool:
+ """See Protocol.is_ready — gRPC aio overrides with channel READY gate."""
+ return True
+
@abstractmethod
async def heartbeat(self):
raise NotImplementedError()
diff --git a/skywalking/agent/protocol/grpc.py b/skywalking/agent/protocol/grpc.py
index bd669d7..0e7b7d4 100644
--- a/skywalking/agent/protocol/grpc.py
+++ b/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
@@ -28,6 +28,13 @@
from skywalking.client.grpc import GrpcServiceManagementClient, GrpcTraceSegmentReportService, \
GrpcProfileTaskChannelService, GrpcLogDataReportService, GrpcMeterReportService
from skywalking.loggings import logger, logger_debug_enabled
+from skywalking.utils.grpc_channel import (
+ apply_connectivity_transition,
+ create_sync_channel,
+ handle_rpc_error,
+ is_channel_ready,
+)
+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
@@ -38,78 +45,146 @@
from skywalking.trace.segment import Segment
+def _queue_get_within_batch(queue: Queue, block: bool, batch_deadline: float, *, allow_immediate: bool = False):
+ """
+ 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.
+ When allow_immediate is True (first generator iteration), still attempt
+ Queue.get once so SW_AGENT_QUEUE_TIMEOUT=0 can drain an immediately
+ available item via get(timeout=0).
+ Returns None when the window is exhausted or the queue is empty.
+ """
+ remaining = batch_deadline - monotonic()
+ if remaining <= 0 and not allow_immediate:
+ return None
+ try:
+ if block:
+ timeout = remaining if remaining > 0 else 0
+ return queue.get(block=True, timeout=timeout)
+ return queue.get(block=False)
+ except Empty:
+ return None
+
+
class GrpcProtocol(Protocol):
def __init__(self):
self.properties_sent = False
self.state = None
- if config.agent_force_tls:
- self.channel = grpc.secure_channel(config.agent_collector_backend_services, grpc.ssl_channel_credentials())
- else:
- self.channel = grpc.insecure_channel(config.agent_collector_backend_services)
+ # One channel for process lifetime; multi-address failover via gRPC pick_first.
+ self.channel = create_sync_channel()
if config.agent_authentication:
self.channel = grpc.intercept_channel(
self.channel, header_adder_interceptor('authentication', config.agent_authentication)
)
- self.channel.subscribe(self._cb, try_to_connect=True)
self.service_management = GrpcServiceManagementClient(self.channel)
self.traces_reporter = GrpcTraceSegmentReportService(self.channel)
self.profile_channel = GrpcProfileTaskChannelService(self.channel)
self.log_reporter = GrpcLogDataReportService(self.channel)
self.meter_reporter = GrpcMeterReportService(self.channel)
+ # Subscribe last: _cb runs on a grpc thread and touches service_management.
+ self.channel.subscribe(self._cb, try_to_connect=True)
+
+ def is_ready(self) -> bool:
+ """
+ Node CONNECTED ≈ subscribe-watched gRPC READY.
+
+ Sync grpcio has no Channel.get_state(); check_connectivity_state can disagree
+ with subscribe callbacks on some builds and permanently skipped all RPCs in
+ E2E (channel already READY via subscribe, reporters still gated). Use the
+ watched state as source of truth; only nudge C-core when IDLE.
+ """
+ if self.state == grpc.ChannelConnectivity.READY:
+ return True
+ if self.state == grpc.ChannelConnectivity.IDLE:
+ # Side-effect nudge (ignore return); subscribe callback updates self.state.
+ is_channel_ready(self.channel)
+ return self.state == grpc.ChannelConnectivity.READY
+
def _cb(self, state):
+ prev = self.state
if logger_debug_enabled:
- logger.debug('grpc channel connectivity changed, [%s -> %s]', self.state, state)
+ logger.debug('grpc channel connectivity changed, [%s -> %s]', prev, state)
+ try:
+ apply_connectivity_transition(prev, state)
+ # Independent OAPs need properties re-registered after failover.
+ # Immediate send via properties_sent; periodic refresh covers silent READY switches.
+ if prev == grpc.ChannelConnectivity.READY and state != grpc.ChannelConnectivity.READY:
+ self.properties_sent = False
+ self.service_management.sent_properties_counter = 0
+ except Exception: # noqa: BLE001 - never let grpc's connectivity thread die on us
+ logger.exception('failed to handle grpc connectivity transition')
self.state = state
def query_profile_commands(self):
+ if not self.is_ready():
+ return
if logger_debug_enabled:
logger.debug('query profile commands')
self.profile_channel.do_query()
def notify_profile_task_finish(self, task: ProfileTask):
+ if not self.is_ready():
+ return
self.profile_channel.finish(task)
def heartbeat(self):
- try:
- if not self.properties_sent:
+ if not self.is_ready():
+ return
+ if not self.properties_sent:
+ try:
self.service_management.send_instance_props()
self.properties_sent = True
-
+ except grpc.RpcError as e:
+ handle_rpc_error(e, self.on_error)
+ try:
self.service_management.send_heart_beat()
-
- except grpc.RpcError:
- self.on_error()
+ except grpc.RpcError as e:
+ handle_rpc_error(e, self.on_error)
raise
def on_error(self):
+ # Re-subscribe the same channel only — never rebuild or rotate backends here.
+ # DEADLINE_EXCEEDED on READY is not a connectivity failure; see handle_rpc_error.
traceback.print_exc() if logger.isEnabledFor(logging.DEBUG) else None
self.channel.unsubscribe(self._cb)
self.channel.subscribe(self._cb, try_to_connect=True)
+ def close(self):
+ """Best-effort channel teardown on agent stop (Node shutdownNow parity)."""
+ try:
+ self.channel.unsubscribe(self._cb)
+ except Exception: # noqa: BLE001
+ pass
+ try:
+ self.channel.close()
+ except Exception: # noqa: BLE001
+ pass
+
def report_segment(self, queue: Queue, block: bool = True):
- start = None
+ # Gate before dequeue so disconnect windows keep segments in the queue (Node buffer parity).
+ if not self.is_ready():
+ return
+ sent = 0
def generator():
- nonlocal start
+ nonlocal sent
+ batch_deadline = monotonic() + float(config.agent_queue_timeout)
+ first_get = True
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:
+ segment = _queue_get_within_batch(queue, block, batch_deadline, allow_immediate=first_get) # type: Segment
+ first_get = False
+ if segment is None:
return
queue.task_done()
+ sent += 1
if logger_debug_enabled:
logger.debug('reporting segment %s', segment)
@@ -156,30 +231,30 @@
try:
self.traces_reporter.report(generator())
- except grpc.RpcError:
- self.on_error()
- raise # reraise so that incremental reconnect wait can process
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('segment', sent)
+ handle_rpc_error(e, self.on_error)
+ raise # reraise so that incremental reconnect wait can process; failed batch discarded
def report_log(self, queue: Queue, block: bool = True):
- start = None
+ if not self.is_ready():
+ return
+ sent = 0
def generator():
- nonlocal start
+ nonlocal sent
+ batch_deadline = monotonic() + float(config.agent_queue_timeout)
+ first_get = True
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:
+ log_data = _queue_get_within_batch(queue, block, batch_deadline, allow_immediate=first_get) # type: LogData
+ first_get = False
+ if log_data is None:
return
queue.task_done()
+ sent += 1
if logger_debug_enabled:
logger.debug('Reporting Log')
@@ -188,30 +263,30 @@
try:
self.log_reporter.report(generator())
- except grpc.RpcError:
- self.on_error()
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('log', sent)
+ handle_rpc_error(e, self.on_error)
raise
def report_meter(self, queue: Queue, block: bool = True):
- start = None
+ if not self.is_ready():
+ return
+ sent = 0
def generator():
- nonlocal start
+ nonlocal sent
+ batch_deadline = monotonic() + float(config.agent_queue_timeout)
+ first_get = True
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:
+ meter_data = _queue_get_within_batch(queue, block, batch_deadline, allow_immediate=first_get) # type: MeterData
+ first_get = False
+ if meter_data is None:
return
queue.task_done()
+ sent += 1
yield meter_data
@@ -219,30 +294,30 @@
if logger_debug_enabled:
logger.debug('Reporting Meter')
self.meter_reporter.report(generator())
- except grpc.RpcError:
- self.on_error()
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('meter', sent)
+ handle_rpc_error(e, self.on_error)
raise
def report_snapshot(self, queue: Queue, block: bool = True):
- start = None
+ if not self.is_ready():
+ return
+ sent = 0
def generator():
- nonlocal start
+ nonlocal sent
+ batch_deadline = monotonic() + float(config.agent_queue_timeout)
+ first_get = True
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:
+ snapshot = _queue_get_within_batch(queue, block, batch_deadline, allow_immediate=first_get) # type: TracingThreadSnapshot
+ first_get = False
+ if snapshot is None:
return
queue.task_done()
+ sent += 1
transform_snapshot = ThreadSnapshot(
taskId=str(snapshot.task_id),
@@ -256,6 +331,8 @@
try:
self.profile_channel.report(generator())
- except grpc.RpcError:
- self.on_error()
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('snapshot', sent)
+ handle_rpc_error(e, self.on_error)
raise
diff --git a/skywalking/agent/protocol/grpc_aio.py b/skywalking/agent/protocol/grpc_aio.py
index 54198c4..f50106f 100644
--- a/skywalking/agent/protocol/grpc_aio.py
+++ b/skywalking/agent/protocol/grpc_aio.py
@@ -17,6 +17,7 @@
import logging
import traceback
+import asyncio
from asyncio import Queue, Event
import grpc
@@ -27,6 +28,13 @@
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,
+)
from skywalking.profile.profile_task import ProfileTask
from skywalking.profile.snapshot import TracingThreadSnapshot
from skywalking.protocol.common.Common_pb2 import KeyStringValuePair
@@ -43,23 +51,17 @@
"""
def __init__(self):
self.properties_sent = Event()
+ self.state = None
- # grpc.aio.channel do not have subscribe() method to set a callback when channel state changed
- # instead, it has wait_for_state_change()/get_state() method to get the current state of the channel
- # since here is an inherent race between the invocation of `wait_for_state_change` and `get_state`,
- # and the channel state is only used for debug, the cost of monitoring this value is too high to support.
- # self.state = None
+ # grpc.aio has no Channel.subscribe(); watch_connectivity() mirrors Node
+ # watchConnectivityState via wait_for_state_change (started by the agent loop).
interceptors = None
if config.agent_authentication:
interceptors = [header_adder_interceptor_async('authentication', config.agent_authentication)]
- if config.agent_force_tls:
- self.channel = grpc.aio.secure_channel(config.agent_collector_backend_services,
- grpc.ssl_channel_credentials(), interceptors=interceptors)
- else:
- self.channel = grpc.aio.insecure_channel(config.agent_collector_backend_services,
- interceptors=interceptors)
+ # One channel for process lifetime; multi-address failover via gRPC pick_first.
+ self.channel = create_aio_channel(interceptors=interceptors)
self.service_management = GrpcServiceManagementClientAsync(self.channel)
self.traces_reporter = GrpcTraceSegmentReportServiceAsync(self.channel)
@@ -67,38 +69,118 @@
self.meter_reporter = GrpcMeterReportServiceAsync(self.channel)
self.profile_channel = GrpcProfileTaskChannelServiceAsync(self.channel)
+ def is_ready(self) -> bool:
+ """Prefer watch-maintained state; peek+nudge when IDLE/None before watch catches up."""
+ if self.state == grpc.ChannelConnectivity.READY:
+ return True
+ if self.state in (None, grpc.ChannelConnectivity.IDLE):
+ try:
+ peeked = self.channel.get_state(True)
+ if peeked is not None:
+ self._on_connectivity(peeked)
+ except Exception: # noqa: BLE001
+ is_channel_ready(self.channel)
+ return self.state == grpc.ChannelConnectivity.READY
+
+ def _on_connectivity(self, state) -> None:
+ prev = self.state
+ if logger_debug_enabled:
+ logger.debug('grpc aio channel connectivity changed, [%s -> %s]', prev, state)
+ apply_connectivity_transition(prev, state)
+ if prev == grpc.ChannelConnectivity.READY and state != grpc.ChannelConnectivity.READY:
+ self.properties_sent.clear()
+ self.service_management.sent_properties_counter = 0
+ self.state = state
+
+ async def watch_connectivity(self):
+ """
+ Background watch: aio equivalent of sync Channel.subscribe.
+ get_state(True) nudges IDLE; wait_for_state_change blocks until transition.
+ """
+ while True:
+ try:
+ state = self.channel.get_state(try_to_connect=True)
+ self._on_connectivity(state)
+ await self.channel.wait_for_state_change(state)
+ except asyncio.CancelledError:
+ raise
+ except Exception: # noqa: BLE001 - keep watch alive across transient errors
+ if logger_debug_enabled:
+ logger.debug('aio connectivity watch error', exc_info=True)
+ await asyncio.sleep(1.0)
+
async def query_profile_commands(self):
+ if not self.is_ready():
+ return
if logger_debug_enabled:
logger.debug('query profile commands')
await self.profile_channel.do_query()
async def notify_profile_task_finish(self, task: ProfileTask):
+ if not self.is_ready():
+ return
await self.profile_channel.finish(task)
async def heartbeat(self):
- try:
- if not self.properties_sent.is_set():
+ if not self.is_ready():
+ return
+ if not self.properties_sent.is_set():
+ try:
await self.service_management.send_instance_props()
self.properties_sent.set()
-
+ except grpc.aio.AioRpcError as e:
+ handle_rpc_error(e, self.on_error)
+ try:
await self.service_management.send_heart_beat()
-
- except grpc.aio.AioRpcError:
- self.on_error()
+ except grpc.aio.AioRpcError as e:
+ handle_rpc_error(e, self.on_error)
raise
def on_error(self):
if logger_debug_enabled:
logger.debug('error occurred in grpc protocol (Async)')
+ # Never rebuild / rotate the channel on RPC errors (auth or otherwise).
+ # DEADLINE_EXCEEDED on READY is not a connectivity failure; see handle_rpc_error.
traceback.print_exc() if logger.isEnabledFor(logging.DEBUG) else None
+ def close(self):
+ """Best-effort channel teardown on agent stop (Node shutdownNow parity)."""
+ # grpc.aio.Channel.close is async; schedule on the running loop when possible.
+ try:
+ result = self.channel.close()
+ if asyncio.iscoroutine(result):
+ try:
+ loop = asyncio.get_running_loop()
+ except RuntimeError:
+ # Called off-loop (should not happen from __fini_async); drop.
+ result.close()
+ return
+ loop.create_task(result)
+ except Exception: # noqa: BLE001
+ pass
+
+ async def aclose(self):
+ """Await channel close from the agent event loop."""
+ try:
+ await self.channel.close()
+ except Exception: # noqa: BLE001
+ pass
+
async def report_segment(self, queue: Queue):
+ # Gate before dequeue so disconnect windows keep segments in the queue.
+ if not self.is_ready():
+ return
+
+ sent = 0
+
async def generator():
+ nonlocal sent
while True:
# Let eventloop schedule blocking instead of user configuration: `config.agent_queue_timeout`
segment = await queue.get() # type: Segment
queue.task_done()
+ sent += 1
if logger_debug_enabled:
logger.debug('reporting segment %s', segment)
@@ -145,17 +227,26 @@
try:
await self.traces_reporter.report(generator())
- except grpc.aio.AioRpcError:
- self.on_error()
- raise # reraise so that incremental reconnect wait can process
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('segment', sent)
+ handle_rpc_error(e, self.on_error)
+ raise # reraise so that incremental reconnect wait can process; failed batch discarded
async def report_log(self, queue: Queue):
+ if not self.is_ready():
+ return
+
+ sent = 0
+
async def generator():
+ nonlocal sent
while True:
# Let eventloop schedule blocking instead of user configuration: `config.agent_queue_timeout`
log_data = await queue.get() # type: LogData
queue.task_done()
+ sent += 1
if logger_debug_enabled:
logger.debug('Reporting Log %s', log_data.timestamp)
@@ -164,17 +255,26 @@
try:
await self.log_reporter.report(generator())
- except grpc.aio.AioRpcError:
- self.on_error()
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('log', sent)
+ handle_rpc_error(e, self.on_error)
raise
async def report_meter(self, queue: Queue):
+ if not self.is_ready():
+ return
+
+ sent = 0
+
async def generator():
+ nonlocal sent
while True:
# Let eventloop schedule blocking instead of user configuration: `config.agent_queue_timeout`
meter_data = await queue.get() # type: MeterData
queue.task_done()
+ sent += 1
if logger_debug_enabled:
logger.debug('Reporting Meter %s', meter_data.timestamp)
@@ -183,17 +283,26 @@
try:
await self.meter_reporter.report(generator())
- except grpc.aio.AioRpcError:
- self.on_error()
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('meter', sent)
+ handle_rpc_error(e, self.on_error)
raise
async def report_snapshot(self, queue: Queue):
+ if not self.is_ready():
+ return
+
+ sent = 0
+
async def generator():
+ nonlocal sent
while True:
# Let eventloop schedule blocking instead of user configuration: `config.agent_queue_timeout`
snapshot = await queue.get() # type: TracingThreadSnapshot
queue.task_done()
+ sent += 1
transform_snapshot = ThreadSnapshot(
taskId=str(snapshot.task_id),
@@ -207,6 +316,8 @@
try:
await self.profile_channel.report(generator())
- except grpc.aio.AioRpcError:
- self.on_error()
+ except grpc.RpcError as e:
+ if sent:
+ log_dropped_throttled('snapshot', sent)
+ handle_rpc_error(e, self.on_error)
raise
diff --git a/skywalking/bootstrap/hooks/uwsgi_hook.py b/skywalking/bootstrap/hooks/uwsgi_hook.py
index fb77171..1331372 100644
--- a/skywalking/bootstrap/hooks/uwsgi_hook.py
+++ b/skywalking/bootstrap/hooks/uwsgi_hook.py
@@ -55,7 +55,11 @@
"""
config.agent_instance_name = f'{config.agent_instance_name}-child({os.getpid()})'
- agent.start()
+ try:
+ agent.start()
+ except Exception: # noqa: BLE001 - never crash the uWSGI worker (sitecustomize parity)
+ logger.exception('SkyWalking agent failed to start in uWSGI worker PID-%s', os.getpid())
+ return
# append pid-suffix to instance name
logger.info(f'Apache SkyWalking Python agent started in pre-forked worker process PID-{os.getpid()}. '
f'Service {config.agent_name}, instance name: {config.agent_instance_name}')
diff --git a/skywalking/client/grpc.py b/skywalking/client/grpc.py
index 7be0d0a..6ceb23b 100644
--- a/skywalking/client/grpc.py
+++ b/skywalking/client/grpc.py
@@ -20,6 +20,7 @@
from skywalking import config
from skywalking.client import ServiceManagementClient, TraceSegmentReportService, ProfileTaskChannelService, \
LogDataReportService, MeterReportService
+from skywalking.utils.grpc_channel import grpc_call_timeout
from skywalking.command import command_service
from skywalking.loggings import logger, logger_debug_enabled
from skywalking.profile import profile_task_execution_service
@@ -40,19 +41,29 @@
self.service_stub = ManagementServiceStub(channel)
def send_instance_props(self):
- self.service_stub.reportInstanceProperties(InstanceProperties(
- service=config.agent_name,
- serviceInstance=config.agent_instance_name,
- properties=self.instance_properties,
- ))
+ self.service_stub.reportInstanceProperties(
+ InstanceProperties(
+ service=config.agent_name,
+ serviceInstance=config.agent_instance_name,
+ properties=self.instance_properties,
+ ),
+ timeout=grpc_call_timeout(),
+ )
def send_heart_beat(self):
- self.refresh_instance_props()
+ # Periodic properties refresh must not block keepAlive (oversized JSON, etc.).
+ try:
+ self.refresh_instance_props()
+ except grpc.RpcError:
+ logger.exception('reportInstanceProperties during keepAlive failed; sending ping anyway')
- self.service_stub.keepAlive(InstancePingPkg(
- service=config.agent_name,
- serviceInstance=config.agent_instance_name,
- ))
+ self.service_stub.keepAlive(
+ InstancePingPkg(
+ service=config.agent_name,
+ serviceInstance=config.agent_instance_name,
+ ),
+ timeout=grpc_call_timeout(),
+ )
if logger_debug_enabled:
logger.debug(
@@ -67,7 +78,7 @@
self.report_stub = TraceSegmentReportServiceStub(channel)
def report(self, generator):
- self.report_stub.collect(generator)
+ self.report_stub.collect(generator, timeout=grpc_call_timeout())
class GrpcMeterReportService(MeterReportService):
@@ -75,10 +86,10 @@
self.report_stub = MeterReportServiceStub(channel)
def report_batch(self, generator):
- self.report_stub.collectBatch(generator)
+ self.report_stub.collectBatch(generator, timeout=grpc_call_timeout())
def report(self, generator):
- self.report_stub.collect(generator)
+ self.report_stub.collect(generator, timeout=grpc_call_timeout())
class GrpcLogDataReportService(LogDataReportService):
@@ -86,7 +97,7 @@
self.report_stub = LogReportServiceStub(channel)
def report(self, generator):
- self.report_stub.collect(generator)
+ self.report_stub.collect(generator, timeout=grpc_call_timeout())
class GrpcProfileTaskChannelService(ProfileTaskChannelService):
@@ -100,11 +111,11 @@
lastCommandTime=profile_task_execution_service.get_last_command_create_time()
)
- commands = self.profile_stub.getProfileTaskCommands(query)
+ commands = self.profile_stub.getProfileTaskCommands(query, timeout=grpc_call_timeout())
command_service.receive_command(commands)
def report(self, generator):
- self.profile_stub.collectSnapshot(generator)
+ self.profile_stub.collectSnapshot(generator, timeout=grpc_call_timeout())
def finish(self, task: ProfileTask):
finish_report = ProfileTaskFinishReport(
@@ -112,4 +123,4 @@
serviceInstance=config.agent_instance_name,
taskId=task.task_id
)
- self.profile_stub.reportTaskFinish(finish_report)
+ self.profile_stub.reportTaskFinish(finish_report, timeout=grpc_call_timeout())
diff --git a/skywalking/client/grpc_aio.py b/skywalking/client/grpc_aio.py
index 038e5c5..a3b8bd5 100644
--- a/skywalking/client/grpc_aio.py
+++ b/skywalking/client/grpc_aio.py
@@ -20,6 +20,7 @@
from skywalking import config
from skywalking.client import ServiceManagementClientAsync, TraceSegmentReportServiceAsync, \
ProfileTaskChannelServiceAsync, LogDataReportServiceAsync, MeterReportServiceAsync
+from skywalking.utils.grpc_channel import grpc_call_timeout
from skywalking.command import command_service_async
from skywalking.loggings import logger, logger_debug_enabled
from skywalking.profile import profile_task_execution_service
@@ -40,19 +41,28 @@
self.service_stub = ManagementServiceStub(channel)
async def send_instance_props(self):
- await self.service_stub.reportInstanceProperties(InstanceProperties(
- service=config.agent_name,
- serviceInstance=config.agent_instance_name,
- properties=self.instance_properties,
- ))
+ await self.service_stub.reportInstanceProperties(
+ InstanceProperties(
+ service=config.agent_name,
+ serviceInstance=config.agent_instance_name,
+ properties=self.instance_properties,
+ ),
+ timeout=grpc_call_timeout(),
+ )
async def send_heart_beat(self):
- await self.refresh_instance_props()
+ try:
+ await self.refresh_instance_props()
+ except grpc.RpcError:
+ logger.exception('reportInstanceProperties during keepAlive failed; sending ping anyway')
- await self.service_stub.keepAlive(InstancePingPkg(
- service=config.agent_name,
- serviceInstance=config.agent_instance_name,
- ))
+ await self.service_stub.keepAlive(
+ InstancePingPkg(
+ service=config.agent_name,
+ serviceInstance=config.agent_instance_name,
+ ),
+ timeout=grpc_call_timeout(),
+ )
if logger_debug_enabled:
logger.debug(
@@ -68,6 +78,8 @@
self.report_stub = TraceSegmentReportServiceStub(channel)
async def report(self, generator):
+ # Aio generators await queue.get() with no idle stop; a deadline would
+ # DEADLINE_EXCEEDED healthy long-lived streams and drop already-sent items.
await self.report_stub.collect(generator)
@@ -103,7 +115,7 @@
lastCommandTime=profile_task_execution_service.get_last_command_create_time()
)
- commands = await self.profile_stub.getProfileTaskCommands(query)
+ commands = await self.profile_stub.getProfileTaskCommands(query, timeout=grpc_call_timeout())
command_service_async.receive_command(commands) # put_nowait() not need to be awaited
async def report(self, generator):
@@ -115,4 +127,4 @@
serviceInstance=config.agent_instance_name,
taskId=task.task_id
)
- await self.profile_stub.reportTaskFinish(finish_report)
+ await self.profile_stub.reportTaskFinish(finish_report, timeout=grpc_call_timeout())
diff --git a/skywalking/config.py b/skywalking/config.py
index 461ccd5..9756e30 100644
--- a/skywalking/config.py
+++ b/skywalking/config.py
@@ -44,8 +44,11 @@
# THIS MUST PRECEDE DIRECTLY BEFORE LIST OF CONFIG OPTIONS!
# BEGIN: Agent Core Configuration Options
-# The backend OAP server address, 11800 is default OAP gRPC port, 12800 is HTTP, Kafka ignores this option
-# and uses kafka_bootstrap_servers option. **This option should be changed accordingly with selected protocol**
+# The backend OAP server address(es). 11800 is default OAP gRPC port, 12800 is HTTP, Kafka ignores this option
+# and uses kafka_bootstrap_servers option. For gRPC, a comma-separated list enables native pick_first failover
+# (per-process shuffle of the preferred backend; :authority stays the first configured endpoint).
+# See Intrusive.md for encoding, DNS, TLS authority, proxy, and READY-gate details.
+# **This option should be changed accordingly with selected protocol**
agent_collector_backend_services: str = os.getenv('SW_AGENT_COLLECTOR_BACKEND_SERVICES', 'oap_host:oap_port')
# The protocol to communicate with the backend OAP, `http`, `grpc` or `kafka`, **we highly suggest using `grpc` in
# production as it's well optimized than `http`**. The `kafka` protocol provides an alternative way to submit data to
@@ -74,7 +77,9 @@
# [here](https://kafka-python.readthedocs.io/en/master/apidoc/KafkaProducer.html#kafka.KafkaProducer)
# This config only works from env variables, each one should be passed in `SW_KAFKA_REPORTER_CONFIG_<KEY_NAME>`
kafka_reporter_custom_configurations: str = os.getenv('SW_KAFKA_REPORTER_CUSTOM_CONFIGURATIONS', '')
-# Use TLS for communication with SkyWalking OAP (no cert required)
+# Use TLS for gRPC/HTTP with the OAP (no client cert required). For gRPC, ensure the server
+# certificate SAN matches the first usable backend in agent_collector_backend_services
+# (used as grpc.default_authority).
agent_force_tls: bool = os.getenv('SW_AGENT_FORCE_TLS', '').lower() == 'true'
# The authentication token to verify that the agent is trusted by the backend OAP, as for how to configure the
# backend, refer to [the yaml](https://github.com/apache/skywalking/blob/4f0f39ffccdc9b41049903cc540b8904f7c9728e/
@@ -88,7 +93,8 @@
# The agent will exchange heartbeat message with SkyWalking OAP backend every `period` seconds
agent_collector_heartbeat_period: int = int(os.getenv('SW_AGENT_COLLECTOR_HEARTBEAT_PERIOD', '30'))
# The agent will report service instance properties every
-# `factor * heartbeat period` seconds default: 10*30 = 300 seconds
+# `factor * heartbeat period` seconds default: 10*30 = 300 seconds (Java/Node cadence).
+# Also covers gRPC pick_first silent backend switches that stay READY without a disconnect event.
agent_collector_properties_report_period_factor = int(
os.getenv('SW_AGENT_COLLECTOR_PROPERTIES_REPORT_PERIOD_FACTOR', '10'))
# A custom JSON string to be reported as service instance properties, e.g. `{"key": "value"}`
diff --git a/skywalking/plugins/sw_grpc.py b/skywalking/plugins/sw_grpc.py
index d2d671d..32a4bd7 100644
--- a/skywalking/plugins/sw_grpc.py
+++ b/skywalking/plugins/sw_grpc.py
@@ -236,8 +236,15 @@
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_building_agent_collector_channel
+
c = _grpc_channel(target, *args, **kwargs)
- if target == config.agent_collector_backend_services:
+ # Prefer agent→OAP build scope: multi-address targets are rewritten (ipv4:/ipv6:)
+ # and no longer equal agent_collector_backend_services.
+ if (
+ is_building_agent_collector_channel()
+ or target == config.agent_collector_backend_services
+ ):
return c
return grpc.intercept_channel(c, _ClientInterceptor(target))
@@ -463,7 +470,14 @@
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),
diff --git a/skywalking/utils/grpc_channel.py b/skywalking/utils/grpc_channel.py
new file mode 100644
index 0000000..6d68a9d
--- /dev/null
+++ b/skywalking/utils/grpc_channel.py
@@ -0,0 +1,677 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Shared gRPC channel target / options helpers for sync and aio reporters.
+
+Multi-backend design (aligned with skywalking-nodejs native failover):
+- One channel for the process lifetime; no hand-rolled poll/reselect manager.
+- Single address → plain host:port (DNS for hostnames, with re-resolve).
+- Multiple addresses are assembled like Node sw-static endpoints
+ ({host, port} list, IPv4 / IPv6 / hostname can coexist).
+- grpcio cannot register a custom scheme; the endpoint list is encoded for
+ C-core: homogeneous ipv4:/ipv6:, mixed families via ipv6: + IPv4-mapped
+ (::ffff:a.b.c.d) so pick_first can try both families.
+- Hostnames in a multi list are resolved once at channel build (grpcio cannot
+ keep a literal hostname in ipv4:/ipv6:). No periodic DNS re-resolve for multi.
+- pick_first shuffleAddressList is on (per-process random preferred backend).
+ Channel target / grpc.default_authority still follow config order (TLS SAN).
+- Invalid entries are logged and dropped; never silently ignored without a log.
+- HTTP proxy disabled; keepalive channel options intentionally omitted (OAP conflict).
+- Unary and sync streaming RPCs use a deadline (Node 10s floor, always >
+ agent_queue_timeout + margin so sync collect is not cut off by the batching window).
+ Aio client-streaming collect/collectBatch/collectSnapshot omit timeout=
+ (generators await empty queues). DEADLINE_EXCEEDED / RESOURCE_EXHAUSTED on a
+ READY backend do not rotate or rebuild; failover is for unreachable backends,
+ not a slow but connected one.
+- READY gate (application-level): skip report RPCs unless channel connectivity is READY;
+ nudge IDLE via get_state(True) so gating does not starve reconnect (Node watch parity).
+- Reconnect backoff max 30s (Node multi-backend parity).
+- service_config retries only ManagementService.reportInstanceProperties on UNAVAILABLE
+ (max 3); never retry client-streaming collect.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import json
+import socket
+import threading
+import time
+from contextlib import contextmanager
+from dataclasses import dataclass
+from enum import Enum
+from typing import Dict, List, Optional, Sequence, Tuple
+
+import grpc
+
+from skywalking.loggings import logger
+
+# Retry only unary idempotent reportInstanceProperties (Node service_config parity).
+# Client-streaming collect must NOT be retried — replay would duplicate segments.
+# keepAlive relies on the next heartbeat tick instead.
+_PROPERTIES_RETRY_SERVICE_CONFIG = json.dumps({
+ # Shuffle is LB-layer (Node parity): target string stays config-order so
+ # grpc.default_authority / TLS SNI remain the first configured endpoint.
+ 'loadBalancingConfig': [{'pick_first': {'shuffleAddressList': True}}],
+ 'methodConfig': [{
+ 'name': [{
+ 'service': 'skywalking.v3.ManagementService',
+ 'method': 'reportInstanceProperties',
+ }],
+ 'retryPolicy': {
+ 'maxAttempts': 3,
+ 'initialBackoff': '1s',
+ 'maxBackoff': '10s',
+ 'backoffMultiplier': 2,
+ 'retryableStatusCodes': ['UNAVAILABLE'],
+ },
+ }],
+})
+
+# Channel options shared by sync + aio. Do NOT add keepalive_* options here.
+# pick_first + shuffle lives in grpc.service_config (not grpc.lb_policy_name).
+GRPC_CHANNEL_OPTIONS: Tuple[Tuple[str, int | str], ...] = (
+ ('grpc.enable_http_proxy', 0),
+ ('grpc.enable_retries', 1),
+ ('grpc.service_config', _PROPERTIES_RETRY_SERVICE_CONFIG),
+ ('grpc.initial_reconnect_backoff_ms', 1000),
+ ('grpc.min_reconnect_backoff_ms', 1000),
+ # Cap aligns with Node multi-backend (~30s); shorter caps reconnect too aggressively.
+ ('grpc.max_reconnect_backoff_ms', 30000),
+)
+
+# Node default RPC deadline is 10s. Sync streaming collect must outlive the queue
+# batch window with room for protobuf encode + RTT + server handling.
+# Sync generators may spend nearly the full queue window on the final queue.get
+# (absolute batch deadline); margin keeps healthy sends off DEADLINE_EXCEEDED.
+# Do not apply this to aio client-streaming: those generators await queue.get() forever.
+_GRPC_RPC_TIMEOUT_FLOOR_SEC = 10.0
+_GRPC_RPC_TIMEOUT_MARGIN_SEC = 5.0
+
+
+def grpc_call_timeout() -> float:
+ """Seconds for unary / sync-streaming stub timeout=. Always > agent_queue_timeout + margin."""
+ from skywalking import config
+
+ return max(
+ _GRPC_RPC_TIMEOUT_FLOOR_SEC,
+ float(config.agent_queue_timeout) + _GRPC_RPC_TIMEOUT_MARGIN_SEC,
+ )
+
+
+_AUTH_LOG_INTERVAL_SEC = 60.0
+_last_auth_log_at = 0.0
+_CONNECTIVITY_LOG_INTERVAL_SEC = 30.0
+_last_connectivity_log_at: Dict[str, float] = {}
+_DNS_LOOKUP_TIMEOUT_SEC = 5.0
+
+# Thread-local: set while create_*_channel builds the agent→OAP channel so sw_grpc
+# does not attach client interceptors (multi-address targets no longer match config).
+_building_agent_collector = threading.local()
+_SW_AGENT_COLLECTOR_ATTR = '_sw_agent_collector_channel'
+
+
+@contextmanager
+def agent_collector_channel_scope():
+ _building_agent_collector.active = True
+ try:
+ yield
+ finally:
+ _building_agent_collector.active = False
+
+
+def is_building_agent_collector_channel() -> bool:
+ return bool(getattr(_building_agent_collector, 'active', False))
+
+
+def mark_agent_collector_channel(channel):
+ try:
+ setattr(channel, _SW_AGENT_COLLECTOR_ATTR, True)
+ except Exception: # noqa: BLE001 - exotic channel wrappers
+ pass
+ return channel
+
+
+def is_agent_collector_channel(channel) -> bool:
+ return bool(getattr(channel, _SW_AGENT_COLLECTOR_ATTR, False))
+
+
+class AddressKind(Enum):
+ IPV4 = 'ipv4'
+ IPV6 = 'ipv6'
+ HOSTNAME = 'hostname'
+
+
+@dataclass(frozen=True)
+class BackendAddress:
+ host: str
+ port: int
+ kind: AddressKind
+
+ def endpoint(self) -> str:
+ if self.kind == AddressKind.IPV6:
+ return f'[{self.host}]:{self.port}'
+ return f'{self.host}:{self.port}'
+
+
+def _classify_host(host: str) -> Optional[AddressKind]:
+ if not host or any(c.isspace() or ord(c) < 32 for c in host) or '/' in host:
+ return None
+ # Zone indices (fe80::1%eth0) are not usable in static ipv6: targets.
+ if '%' in host:
+ return None
+ try:
+ ip = ipaddress.ip_address(host)
+ except ValueError:
+ return AddressKind.HOSTNAME
+ if isinstance(ip, ipaddress.IPv4Address):
+ return AddressKind.IPV4
+ return AddressKind.IPV6
+
+
+def parse_backend_address(raw: str) -> Optional[BackendAddress]:
+ """Parse a single host:port (IPv6 requires [host]:port). Returns None if invalid."""
+ text = (raw or '').strip()
+ if not text:
+ return None
+
+ host: str
+ port_str: str
+ if text.startswith('['):
+ # [ipv6]:port
+ closing = text.find(']')
+ if closing <= 1 or closing + 1 >= len(text) or text[closing + 1] != ':':
+ return None
+ host = text[1:closing]
+ port_str = text[closing + 2:]
+ else:
+ if text.count(':') != 1:
+ # Ambiguous IPv6 without brackets, or missing port.
+ return None
+ host, port_str = text.rsplit(':', 1)
+
+ host = host.strip()
+ port_str = port_str.strip()
+ if not host or not port_str:
+ return None
+ try:
+ port = int(port_str)
+ except ValueError:
+ return None
+ if port < 1 or port > 65535:
+ return None
+
+ kind = _classify_host(host)
+ if kind is None:
+ return None
+ return BackendAddress(host=host, port=port, kind=kind)
+
+
+def parse_backend_addresses(services: str) -> List[BackendAddress]:
+ """
+ Split SW_AGENT_COLLECTOR_BACKEND_SERVICES on commas.
+ Invalid entries are skipped with an error log (never silent).
+ """
+ parts = [p.strip() for p in (services or '').split(',') if p.strip()]
+ addresses: List[BackendAddress] = []
+ for part in parts:
+ addr = parse_backend_address(part)
+ if addr is None:
+ logger.error(
+ 'Invalid collector backend address %r in SW_AGENT_COLLECTOR_BACKEND_SERVICES; '
+ 'expected host:port or [ipv6]:port',
+ part,
+ )
+ continue
+ addresses.append(addr)
+ return addresses
+
+
+def sw_static_endpoints(addresses: Sequence[BackendAddress]) -> List[Dict]:
+ """
+ Node sw-static resolver output shape: a list of endpoints, each with
+ addresses: [{host, port}]. IPv4, IPv6, and hostnames can share one list.
+ """
+ return [
+ {'addresses': [{'host': addr.host, 'port': addr.port}]}
+ for addr in addresses
+ ]
+
+
+def _lookup_hostname(host: str, port: int) -> List[BackendAddress]:
+ """
+ Resolve hostname to BackendAddress IPs (order preserved, duplicates dropped).
+
+ Multi-address targets require literal IPs for ipv4:/ipv6:. A hung DNS lookup
+ must not block agent startup or process exit — bound the wait and run the
+ lookup on a daemon thread (ThreadPoolExecutor workers are non-daemon and
+ would keep the process alive after timeout).
+ """
+ import threading
+ from concurrent.futures import Future, TimeoutError as FuturesTimeout
+
+ fut: Future = Future()
+
+ def _run() -> None:
+ try:
+ fut.set_result(socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM))
+ except Exception as exc: # noqa: BLE001 - forward any lookup failure to waiter
+ if not fut.done():
+ fut.set_exception(exc)
+
+ threading.Thread(target=_run, name=f'sw-dns-{host}', daemon=True).start()
+ try:
+ infos = fut.result(timeout=_DNS_LOOKUP_TIMEOUT_SEC)
+ except FuturesTimeout:
+ logger.error(
+ 'Timed out resolving collector hostname %r:%s after %.1fs; skipping this backend',
+ host,
+ port,
+ _DNS_LOOKUP_TIMEOUT_SEC,
+ )
+ return []
+ except socket.gaierror as exc:
+ logger.error(
+ 'Failed to resolve collector hostname %r:%s (%s); skipping this backend',
+ host,
+ port,
+ exc,
+ )
+ return []
+ except Exception as exc: # noqa: BLE001 - never crash agent init on DNS oddities
+ logger.error(
+ 'Unexpected error resolving collector hostname %r:%s (%s); skipping this backend',
+ host,
+ port,
+ exc,
+ )
+ return []
+
+ resolved: List[BackendAddress] = []
+ seen = set()
+ for family, _type, _proto, _canon, sockaddr in infos:
+ if family == socket.AF_INET:
+ ip = sockaddr[0]
+ kind = AddressKind.IPV4
+ elif family == socket.AF_INET6:
+ ip = sockaddr[0]
+ if '%' in ip:
+ ip = ip.split('%', 1)[0]
+ # Skip IPv4-mapped IPv6; the AF_INET result already covers that backend.
+ try:
+ packed = ipaddress.ip_address(ip)
+ except ValueError:
+ continue
+ if packed.ipv4_mapped is not None:
+ continue
+ kind = AddressKind.IPV6
+ else:
+ continue
+ key = (kind, ip, port)
+ if key in seen:
+ continue
+ seen.add(key)
+ resolved.append(BackendAddress(host=ip, port=port, kind=kind))
+ if not resolved:
+ logger.error(
+ 'Collector hostname %r:%s resolved to no usable IPv4/IPv6 address; skipping',
+ host,
+ port,
+ )
+ return resolved
+
+
+def expand_backend_addresses(addresses: Sequence[BackendAddress]) -> List[BackendAddress]:
+ """
+ Expand hostnames to literal IPs for C-core static targets.
+ Literal IP entries are kept as-is. Failed hostname lookups are skipped with error logs.
+ """
+ expanded: List[BackendAddress] = []
+ seen = set()
+ for addr in addresses:
+ candidates: Sequence[BackendAddress]
+ if addr.kind == AddressKind.HOSTNAME:
+ candidates = _lookup_hostname(addr.host, addr.port)
+ else:
+ candidates = (addr,)
+ for item in candidates:
+ key = (item.kind, item.host, item.port)
+ if key in seen:
+ continue
+ seen.add(key)
+ expanded.append(item)
+ return expanded
+
+
+def _ipv4_mapped_v6(ipv4: str) -> str:
+ return f'::ffff:{ipv4}'
+
+
+def encode_sw_static_for_c_core(addresses: Sequence[BackendAddress]) -> str:
+ """
+ Encode a Node-style mixed endpoint list for grpcio/C-core.
+
+ Homogeneous lists use ipv4:/ipv6:. Mixed IPv4+IPv6 uses the ipv6 resolver
+ with IPv4-mapped addresses so pick_first can try both families.
+ """
+ if not addresses:
+ raise ValueError(
+ 'No valid collector backend address in SW_AGENT_COLLECTOR_BACKEND_SERVICES'
+ )
+ if any(a.kind == AddressKind.HOSTNAME for a in addresses):
+ raise ValueError('encode_sw_static_for_c_core requires literal IP endpoints')
+
+ kinds = {a.kind for a in addresses}
+ if kinds == {AddressKind.IPV4}:
+ return 'ipv4:' + ','.join(f'{a.host}:{a.port}' for a in addresses)
+ if kinds == {AddressKind.IPV6}:
+ return 'ipv6:' + ','.join(f'[{a.host}]:{a.port}' for a in addresses)
+
+ parts = []
+ for addr in addresses:
+ if addr.kind == AddressKind.IPV4:
+ parts.append(f'[{_ipv4_mapped_v6(addr.host)}]:{addr.port}')
+ else:
+ parts.append(f'[{addr.host}]:{addr.port}')
+ logger.info(
+ 'Encoding mixed-family sw-static endpoints for grpcio via ipv6 IPv4-mapped list: %s',
+ parts,
+ )
+ return 'ipv6:' + ','.join(parts)
+
+
+def prepare_grpc_channel_endpoints(
+ addresses: Sequence[BackendAddress],
+) -> Tuple[str, str]:
+ """
+ Build (channel_target, default_authority) from parsed backends.
+
+ Authority prefers the first *usable* original entry's host:port (hostname kept
+ for TLS SAN when that name resolved). Never points at a hostname that DNS skipped.
+ """
+ if not addresses:
+ raise ValueError(
+ 'No valid collector backend address in SW_AGENT_COLLECTOR_BACKEND_SERVICES'
+ )
+
+ if len(addresses) == 1:
+ ep = addresses[0].endpoint()
+ return ep, ep
+
+ to_encode: List[BackendAddress] = []
+ seen = set()
+ authority: Optional[str] = None
+
+ for orig in addresses:
+ if orig.kind == AddressKind.HOSTNAME:
+ candidates = _lookup_hostname(orig.host, orig.port)
+ if not candidates:
+ continue
+ if authority is None:
+ # Prefer original hostname for :authority / SNI (Node sw-static style).
+ authority = orig.endpoint()
+ else:
+ candidates = (orig,)
+ if authority is None:
+ authority = orig.endpoint()
+ for item in candidates:
+ key = (item.kind, item.host, item.port)
+ if key in seen:
+ continue
+ seen.add(key)
+ to_encode.append(item)
+
+ if not to_encode:
+ raise ValueError(
+ 'No usable collector backend address after DNS expansion of '
+ 'SW_AGENT_COLLECTOR_BACKEND_SERVICES'
+ )
+ if authority is None:
+ authority = to_encode[0].endpoint()
+
+ if any(a.kind == AddressKind.HOSTNAME for a in addresses):
+ logger.info(
+ 'Expanded multi-backend collector addresses %s -> %s (authority=%s)',
+ [a.endpoint() for a in addresses],
+ [a.endpoint() for a in to_encode],
+ authority,
+ )
+ return encode_sw_static_for_c_core(to_encode), authority
+
+
+def _resolve_channel_target_and_authority() -> Tuple[str, str]:
+ """
+ Never raise into the host app. prepare_grpc_channel_endpoints stays strict;
+ factories degrade so the agent can idle behind the READY gate.
+ """
+ from skywalking import config
+
+ raw = config.agent_collector_backend_services
+ addresses = parse_backend_addresses(raw)
+ try:
+ return prepare_grpc_channel_endpoints(addresses)
+ except ValueError:
+ if addresses:
+ target = addresses[0].endpoint()
+ logger.error(
+ 'No usable collector backend after DNS expansion of %r; '
+ 'falling back to plain target %s so grpcio can re-resolve',
+ raw,
+ target,
+ )
+ return target, target
+ fallback = (raw or '').strip() or 'localhost:1'
+ logger.error(
+ 'No valid collector backend address in %r; opening a channel to %s '
+ '(agent stays up; READY gate skips reports)',
+ raw,
+ fallback,
+ )
+ return fallback, fallback
+
+
+def build_grpc_target(addresses: Sequence[BackendAddress]) -> str:
+ """Build a gRPC channel target string (see prepare_grpc_channel_endpoints)."""
+ target, _authority = prepare_grpc_channel_endpoints(addresses)
+ return target
+
+
+def resolve_grpc_target(services: Optional[str] = None) -> str:
+ from skywalking import config
+
+ raw = config.agent_collector_backend_services if services is None else services
+ return build_grpc_target(parse_backend_addresses(raw))
+
+
+def _channel_options(default_authority: str) -> Tuple[Tuple[str, int | str], ...]:
+ options = list(GRPC_CHANNEL_OPTIONS)
+ # Align with Node sw-static getDefaultAuthority (first usable backend).
+ options.append(('grpc.default_authority', default_authority))
+ return tuple(options)
+
+
+def create_sync_channel():
+ """Create one sync gRPC channel (caller may wrap with auth interceptor)."""
+ from skywalking import config
+
+ target, authority = _resolve_channel_target_and_authority()
+ options = _channel_options(authority)
+ logger.info('Creating gRPC channel to collector target %s (authority=%s)', target, authority)
+ with agent_collector_channel_scope():
+ try:
+ if config.agent_force_tls:
+ channel = grpc.secure_channel(target, grpc.ssl_channel_credentials(), options=options)
+ else:
+ channel = grpc.insecure_channel(target, options=options)
+ except Exception: # noqa: BLE001 - never fail host process start
+ logger.exception(
+ 'Failed to create gRPC channel to %s; using localhost:1 placeholder',
+ target,
+ )
+ channel = grpc.insecure_channel('localhost:1', options=options)
+ return mark_agent_collector_channel(channel)
+
+
+def create_aio_channel(interceptors=None):
+ """Create one aio gRPC channel with optional interceptors."""
+ from skywalking import config
+
+ target, authority = _resolve_channel_target_and_authority()
+ options = _channel_options(authority)
+ logger.info('Creating aio gRPC channel to collector target %s (authority=%s)', target, authority)
+ with agent_collector_channel_scope():
+ try:
+ if config.agent_force_tls:
+ channel = grpc.aio.secure_channel(
+ target,
+ grpc.ssl_channel_credentials(),
+ options=options,
+ interceptors=interceptors,
+ )
+ else:
+ channel = grpc.aio.insecure_channel(target, options=options, interceptors=interceptors)
+ except Exception: # noqa: BLE001 - never fail host process start
+ logger.exception(
+ 'Failed to create aio gRPC channel to %s; using localhost:1 placeholder',
+ target,
+ )
+ channel = grpc.aio.insecure_channel('localhost:1', options=options, interceptors=interceptors)
+ return mark_agent_collector_channel(channel)
+
+
+def _unwrap_connectivity_state(channel, try_to_connect: bool):
+ """
+ Read channel connectivity for sync, aio, and intercept_channel wrappers.
+
+ grpc.aio.Channel exposes get_state(). Sync grpc._channel.Channel does not
+ (subscribe only); use C-core check_connectivity_state on the cython channel.
+ Intercepted channels nest the real Channel on ``_channel``.
+ """
+ get_state = getattr(channel, 'get_state', None)
+ if callable(get_state):
+ return get_state(try_to_connect)
+
+ candidate = channel
+ for _ in range(4):
+ inner = getattr(candidate, '_channel', None)
+ if inner is None:
+ break
+ check = getattr(inner, 'check_connectivity_state', None)
+ if callable(check):
+ code = check(try_to_connect)
+ for state in grpc.ChannelConnectivity:
+ if state.value[0] == code:
+ return state
+ return None
+ candidate = inner
+ return None
+
+
+def is_channel_ready(channel) -> bool:
+ """
+ Application-level READY gate (gRPC only exposes connectivity state).
+
+ Returns True only when connectivity is READY. Uses try_to_connect=True so an
+ IDLE channel is nudged into CONNECTING — otherwise skipping all RPCs would
+ leave the channel idle forever (same role as Node watchConnectivityState
+ with requestConnection=true). CONNECTING / TRANSIENT_FAILURE still return
+ False so reporters skip until READY.
+
+ If connectivity cannot be read, returns True (fail-open) so a missing API
+ cannot permanently silence reporting.
+ """
+ try:
+ state = _unwrap_connectivity_state(channel, True)
+ except Exception: # noqa: BLE001 - defensive for closed / exotic channels
+ return True
+ if state is None:
+ return True
+ return state == grpc.ChannelConnectivity.READY
+
+
+def is_auth_rpc_error(error: BaseException) -> bool:
+ code = getattr(error, 'code', None)
+ if not callable(code):
+ return False
+ try:
+ status = code()
+ except Exception: # noqa: BLE001 - defensive for non-grpc exceptions
+ return False
+ return status in (grpc.StatusCode.UNAUTHENTICATED, grpc.StatusCode.PERMISSION_DENIED)
+
+
+def log_auth_failure_throttled(error: BaseException) -> None:
+ """Auth failures must not rotate backends; same cluster shares one token."""
+ global _last_auth_log_at
+ now = time.monotonic()
+ if now - _last_auth_log_at < _AUTH_LOG_INTERVAL_SEC:
+ return
+ _last_auth_log_at = now
+ logger.error(
+ 'Collector rejected authentication (%s). Check SW_AGENT_AUTHENTICATION; '
+ 'the agent will not rotate backends for auth failures.',
+ error,
+ )
+
+
+def log_connectivity_event(kind: str, message: str, *args) -> None:
+ """Throttle disconnect warnings; recovery stays informative but rate-limited."""
+ now = time.monotonic()
+ last = _last_connectivity_log_at.get(kind, 0.0)
+ if now - last < _CONNECTIVITY_LOG_INTERVAL_SEC:
+ return
+ _last_connectivity_log_at[kind] = now
+ if kind == 'recovered':
+ logger.info(message, *args)
+ else:
+ logger.warning(message, *args)
+
+
+def apply_connectivity_transition(prev, state) -> None:
+ """Shared INFO/WARN side-effects for sync subscribe and aio watch."""
+ if state == grpc.ChannelConnectivity.TRANSIENT_FAILURE:
+ log_connectivity_event(
+ 'transient_failure',
+ 'gRPC collector channel disconnected (TRANSIENT_FAILURE)',
+ )
+ elif state == grpc.ChannelConnectivity.IDLE and prev == grpc.ChannelConnectivity.READY:
+ log_connectivity_event('idle', 'gRPC collector channel disconnected (IDLE)')
+ elif state == grpc.ChannelConnectivity.READY and prev in (
+ grpc.ChannelConnectivity.TRANSIENT_FAILURE,
+ grpc.ChannelConnectivity.IDLE,
+ grpc.ChannelConnectivity.CONNECTING,
+ None,
+ ):
+ log_connectivity_event('recovered', 'gRPC collector channel recovered (READY)')
+
+
+def handle_rpc_error(error: BaseException, on_connectivity_error) -> None:
+ """
+ Shared RpcError side-effects for sync/aio reporters.
+ Auth: throttle log only (no channel rebuild / backend rotate).
+ Other: invoke connectivity recovery hook (resubscribe / debug), never rebuild channel.
+
+ Failover is for an unreachable backend, not functional health of a connected one.
+ DEADLINE_EXCEEDED and RESOURCE_EXHAUSTED on a READY channel are intentionally
+ left to the call site: pick_first will not move off a slow-but-READY backend,
+ and this agent does not rotate or rebuild the channel for those codes.
+ """
+ if is_auth_rpc_error(error):
+ log_auth_failure_throttled(error)
+ return
+ on_connectivity_error()
diff --git a/skywalking/utils/reporter_log.py b/skywalking/utils/reporter_log.py
new file mode 100644
index 0000000..37f5b46
--- /dev/null
+++ b/skywalking/utils/reporter_log.py
@@ -0,0 +1,80 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Throttled reporter logs that must not import grpc.
+
+agent/__init__.py imports these helpers at module load. Keeping them out of
+grpc_channel.py ensures GRPC_ENABLE_FORK_SUPPORT can be set before the first
+`import grpc` (see start / start_prefork_master).
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from typing import Dict
+
+from skywalking.loggings import logger
+
+_REPORTER_LOG_INTERVAL_SEC = 30.0
+_last_reporter_log_at: Dict[str, float] = {}
+_DROP_LOG_INTERVAL_SEC = 30.0
+_last_drop_log_at: Dict[str, float] = {}
+_drop_totals: Dict[str, int] = {}
+_drop_logged_totals: Dict[str, int] = {}
+
+
+def log_reporter_exception_throttled(reporter_name: str, wait: float) -> None:
+ """
+ Throttle reporter exception stacks during outages (otherwise every backoff tick floods logs).
+ """
+ now = time.monotonic()
+ last = _last_reporter_log_at.get(reporter_name, 0.0)
+ if now - last < _REPORTER_LOG_INTERVAL_SEC:
+ return
+ _last_reporter_log_at[reporter_name] = now
+ logger.exception(
+ 'Exception in %s service in pid %s, retry in %s seconds',
+ reporter_name,
+ os.getpid(),
+ wait,
+ )
+
+
+def log_dropped_throttled(kind: str, count: int = 1, *, force: bool = False) -> None:
+ """
+ Throttle drop warnings. ``count`` is added to a process-wide total for ``kind``.
+ At most one line per 30s unless ``force`` (shutdown). Message includes this
+ window's increment and the process total.
+ """
+ if count <= 0:
+ return
+ now = time.monotonic()
+ total = _drop_totals.get(kind, 0) + count
+ _drop_totals[kind] = total
+ last = _last_drop_log_at.get(kind, 0.0)
+ if not force and now - last < _DROP_LOG_INTERVAL_SEC:
+ return
+ logged = _drop_logged_totals.get(kind, 0)
+ delta = total - logged
+ _drop_logged_totals[kind] = total
+ _last_drop_log_at[kind] = now
+ logger.warning(
+ 'dropped %s: +%d since last log, %d total this process (best-effort; never retried)',
+ kind,
+ delta,
+ total,
+ )
diff --git a/tests/e2e/case/expected/service-sorted.yml b/tests/e2e/case/expected/service-sorted.yml
new file mode 100644
index 0000000..073f348
--- /dev/null
+++ b/tests/e2e/case/expected/service-sorted.yml
@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Same services as service.yml, sorted by name. Used when registration order
+# is non-deterministic (e.g. cold standby after pick_first failover).
+- id: {{ b64enc "e2e-service-consumer|namespace" }}.1
+ name: "e2e-service-consumer|namespace"
+ shortname: "e2e-service-consumer|namespace"
+ normal: true
+ layers:
+ - GENERAL
+ group: ""
+- id: {{ b64enc "e2e-service-provider" }}.1
+ name: e2e-service-provider
+ shortname: e2e-service-provider
+ normal: true
+ layers:
+ - GENERAL
+ group: ""
diff --git a/tests/e2e/case/grpc/failover/docker-compose.grpc.failover.yaml b/tests/e2e/case/grpc/failover/docker-compose.grpc.failover.yaml
new file mode 100644
index 0000000..b99acf0
--- /dev/null
+++ b/tests/e2e/case/grpc/failover/docker-compose.grpc.failover.yaml
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+services:
+ # Two independent OAPs (not a cluster). Service names must stay bash-safe
+ # (oap_a / oap_b) so infra-e2e can export ${oap_a_host}/${oap_a_12800}.
+ # Do NOT set container_name: infra-e2e looks up {project}_oap_b_1 /
+ # {project}-oap_b-1; a fixed name breaks setup before verify runs.
+ oap_a:
+ extends:
+ file: ../../../base/docker-compose.base.yml
+ service: oap
+ ports:
+ - "12800"
+
+ oap_b:
+ extends:
+ file: ../../../base/docker-compose.base.yml
+ service: oap
+ ports:
+ - "12800"
+
+ provider:
+ extends:
+ file: ../../../base/docker-compose.base.yml
+ service: fastapi-provider
+ environment:
+ SW_AGENT_COLLECTOR_BACKEND_SERVICES: oap_a:11800,oap_b:11800
+ depends_on:
+ oap_a:
+ condition: service_healthy
+ oap_b:
+ condition: service_healthy
+ ports:
+ - "9090"
+
+ consumer:
+ extends:
+ file: ../../../base/docker-compose.base.yml
+ service: flask-consumer
+ environment:
+ SW_AGENT_COLLECTOR_BACKEND_SERVICES: oap_a:11800,oap_b:11800
+ depends_on:
+ provider:
+ condition: service_healthy
+ ports:
+ - "9090"
+
+networks:
+ e2e:
diff --git a/tests/e2e/case/grpc/failover/e2e.yaml b/tests/e2e/case/grpc/failover/e2e.yaml
new file mode 100644
index 0000000..f26667a
--- /dev/null
+++ b/tests/e2e/case/grpc/failover/e2e.yaml
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Multi-OAP gRPC pick_first failover (Node static-failover analogue).
+# Preferred backend is shuffled per process, so provider and consumer may
+# already sit on different OAPs. Assert the union of both service lists,
+# stop one OAP that has data (prefer oap_a), then require the other list.
+
+setup:
+ env: compose
+ file: docker-compose.grpc.failover.yaml
+ timeout: 20m
+ init-system-environment: ../../../script/env
+ steps:
+ - name: install yq
+ command: bash tests/e2e/script/prepare/install.sh yq
+ - name: install swctl
+ command: bash tests/e2e/script/prepare/install.sh swctl
+
+trigger:
+ action: http
+ interval: 1s
+ times: 10
+ url: http://${consumer_host}:${consumer_9090}/artist-consumer
+ method: POST
+ headers:
+ "Content-Type": "application/json"
+ body: '{"song": "Despacito"}'
+
+verify:
+ retry:
+ count: 20
+ interval: 10s
+ cases:
+ - name: both agents registered (union across OAPs)
+ query: bash tests/e2e/case/grpc/failover/verify-active.sh
+ expected: ../../expected/service-sorted.yml
+ - name: remaining OAP has both services after one backend stop
+ query: bash tests/e2e/case/grpc/failover/verify-standby.sh
+ expected: ../../expected/service-sorted.yml
diff --git a/tests/e2e/case/grpc/failover/lib.sh b/tests/e2e/case/grpc/failover/lib.sh
new file mode 100755
index 0000000..f5ba18c
--- /dev/null
+++ b/tests/e2e/case/grpc/failover/lib.sh
@@ -0,0 +1,143 @@
+#!/usr/bin/env bash
+
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# ----------------------------------------------------------------------------
+
+# Shared helpers for gRPC failover E2E (sourced). stdout of callers must stay YAML.
+
+set -euo pipefail
+
+: "${oap_a_host:?}"
+: "${oap_a_12800:?}"
+: "${oap_b_host:?}"
+: "${oap_b_12800:?}"
+
+_FAILOVER_STICKY_DIR="${TMPDIR:-/tmp}/sw-python-e2e-failover"
+mkdir -p "${_FAILOVER_STICKY_DIR}"
+_FAILOVER_KEY="${oap_a_host}_${oap_a_12800}_${oap_b_host}_${oap_b_12800}"
+FAILOVER_STOPPED="${_FAILOVER_STICKY_DIR}/${_FAILOVER_KEY}.stopped"
+FAILOVER_STANDBY="${_FAILOVER_STICKY_DIR}/${_FAILOVER_KEY}.standby"
+
+oap_service_count() {
+ local host="$1" port="$2" yaml count
+ if ! yaml="$(swctl --display yaml --base-url="http://${host}:${port}/graphql" service ls)"; then
+ echo "FATAL: could not read service list from ${host}:${port}" >&2
+ return 1
+ fi
+ case "${yaml}" in
+ ''|'[]'|'null')
+ echo 0
+ return 0
+ ;;
+ esac
+ if ! count="$(printf '%s\n' "${yaml}" | yq e 'length' -)"; then
+ echo "FATAL: could not parse service list from ${host}:${port}" >&2
+ return 1
+ fi
+ case "${count}" in
+ ''|'null')
+ echo 0
+ return 0
+ ;;
+ *[!0-9]*)
+ echo "FATAL: could not read service list from ${host}:${port} (got ${count})" >&2
+ return 1
+ ;;
+ esac
+ echo "${count}"
+}
+
+oap_service_ls_yaml() {
+ local host="$1" port="$2" yaml
+ if ! yaml="$(swctl --display yaml --base-url="http://${host}:${port}/graphql" service ls)"; then
+ echo "FATAL: could not read service list from ${host}:${port}" >&2
+ return 1
+ fi
+ case "${yaml}" in
+ ''|'[]'|'null')
+ echo '[]'
+ ;;
+ *)
+ printf '%s\n' "${yaml}"
+ ;;
+ esac
+}
+
+oap_service_ls_sorted() {
+ local host="$1" port="$2"
+ oap_service_ls_yaml "${host}" "${port}" | yq e 'sort_by(.name)' -
+}
+
+# Union of two service-list YAML documents, unique by name, sorted.
+oap_service_union_sorted() {
+ local ya yb
+ ya="$(oap_service_ls_yaml "$1" "$2")" || return 1
+ yb="$(oap_service_ls_yaml "$3" "$4")" || return 1
+ yq ea 'select(fileIndex == 0) + select(fileIndex == 1) | unique_by(.name) | sort_by(.name)' \
+ <(printf '%s\n' "${ya}") <(printf '%s\n' "${yb}")
+}
+
+compose_project_from_oap_b() {
+ local project="${COMPOSE_PROJECT_NAME:-}" cid name
+ if [ -n "${project}" ]; then
+ echo "${project}"
+ return 0
+ fi
+ while read -r cid; do
+ [ -z "${cid}" ] && continue
+ if docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}{{range $p, $c := .NetworkSettings.Ports}}{{range $c}}{{.HostIp}} {{end}}{{end}}' "${cid}" 2>/dev/null \
+ | tr ' ' '\n' | grep -qx "${oap_b_host}"; then
+ docker inspect -f '{{index .Config.Labels "com.docker.compose.project"}}' "${cid}"
+ return 0
+ fi
+ name="$(docker inspect -f '{{.Name}}' "${cid}" | sed 's#^/##')"
+ if [ "${name}" = "${oap_b_host}" ] || echo "${name}" | grep -q "${oap_b_host}"; then
+ docker inspect -f '{{index .Config.Labels "com.docker.compose.project"}}' "${cid}"
+ return 0
+ fi
+ done < <(docker ps -q --filter 'label=com.docker.compose.service=oap_b' || true)
+ docker ps \
+ --filter 'label=com.docker.compose.service=oap_b' \
+ --format '{{.Label "com.docker.compose.project"}}' \
+ | head -n1 || true
+}
+
+stop_compose_service() {
+ local service="$1" project filters running any
+ project="$(compose_project_from_oap_b)"
+ filters="--filter label=com.docker.compose.service=${service}"
+ if [ -n "${project}" ]; then
+ filters="${filters} --filter label=com.docker.compose.project=${project}"
+ fi
+ # shellcheck disable=SC2086
+ running="$(docker ps -q ${filters} | head -n1 || true)"
+ # shellcheck disable=SC2086
+ any="$(docker ps -aq ${filters} | head -n1 || true)"
+ if [ -z "${any}" ]; then
+ echo "FATAL: no ${service} container found (project=${project:-unknown})" >&2
+ docker ps -a --format '{{.Names}} {{.Label "com.docker.compose.project"}}/{{.Label "com.docker.compose.service"}} {{.State}}' >&2 || true
+ exit 1
+ fi
+ if [ -n "${running}" ]; then
+ echo "stopping ${service} ${running} (project=${project:-unknown})" >&2
+ docker stop "${running}" >/dev/null
+ else
+ echo "${service} already stopped by an earlier attempt" >&2
+ fi
+}
diff --git a/tests/e2e/case/grpc/failover/verify-active.sh b/tests/e2e/case/grpc/failover/verify-active.sh
new file mode 100755
index 0000000..9e25149
--- /dev/null
+++ b/tests/e2e/case/grpc/failover/verify-active.sh
@@ -0,0 +1,39 @@
+#!/usr/bin/env bash
+
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# ----------------------------------------------------------------------------
+
+# Print the union of both OAPs' service lists (shuffle-safe). Provider and
+# consumer shuffle independently, so both backends may already have data.
+
+# shellcheck source=lib.sh
+. "$(dirname "$0")/lib.sh"
+
+count_a="$(oap_service_count "${oap_a_host}" "${oap_a_12800}")"
+count_b="$(oap_service_count "${oap_b_host}" "${oap_b_12800}")"
+
+if [ "${count_a}" = "0" ] && [ "${count_b}" = "0" ]; then
+ echo "FATAL: neither OAP has services yet" >&2
+ exit 1
+fi
+
+echo "pre-failover union: oap_a=${count_a} oap_b=${count_b}" >&2
+oap_service_union_sorted \
+ "${oap_a_host}" "${oap_a_12800}" \
+ "${oap_b_host}" "${oap_b_12800}"
diff --git a/tests/e2e/case/grpc/failover/verify-standby.sh b/tests/e2e/case/grpc/failover/verify-standby.sh
new file mode 100755
index 0000000..1a578ed
--- /dev/null
+++ b/tests/e2e/case/grpc/failover/verify-standby.sh
@@ -0,0 +1,78 @@
+#!/usr/bin/env bash
+
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# ----------------------------------------------------------------------------
+
+# Stop one OAP that currently has services (prefer oap_a), generate traffic,
+# print the other OAP's sorted list. Provider/consumer may already be split
+# across backends; stopping A still forces the A-side agent onto B.
+
+# shellcheck source=lib.sh
+. "$(dirname "$0")/lib.sh"
+
+: "${consumer_host:?}"
+: "${consumer_9090:?}"
+
+if [ ! -f "${FAILOVER_STOPPED}" ]; then
+ count_a="$(oap_service_count "${oap_a_host}" "${oap_a_12800}")"
+ count_b="$(oap_service_count "${oap_b_host}" "${oap_b_12800}")"
+
+ if [ "${count_a}" = "0" ] && [ "${count_b}" = "0" ]; then
+ echo "FATAL: neither OAP has services; cannot choose a backend to stop" >&2
+ exit 1
+ fi
+
+ if [ "${count_a}" != "0" ]; then
+ active=oap_a
+ standby_host="${oap_b_host}"
+ standby_port="${oap_b_12800}"
+ else
+ active=oap_b
+ standby_host="${oap_a_host}"
+ standby_port="${oap_a_12800}"
+ fi
+
+ echo "stopping ${active} (oap_a=${count_a} oap_b=${count_b}); remaining=${standby_host}:${standby_port}" >&2
+ stop_compose_service "${active}"
+ printf '%s\n' "${active}" > "${FAILOVER_STOPPED}"
+ printf '%s\n' "${standby_host}:${standby_port}" > "${FAILOVER_STANDBY}"
+else
+ echo "a backend already stopped by an earlier attempt; skipping stop" >&2
+fi
+
+sent=0
+for _ in $(seq 1 5); do
+ if curl -fsS -X POST \
+ -H 'Content-Type: application/json' \
+ -d '{"song": "Despacito"}' \
+ "http://${consumer_host}:${consumer_9090}/artist-consumer" >/dev/null; then
+ sent=$((sent + 1))
+ fi
+ sleep 1
+done
+
+if [ "${sent}" -eq 0 ]; then
+ echo "FATAL: no post-failover request reached the consumer at" >&2
+ echo " http://${consumer_host}:${consumer_9090}/artist-consumer" >&2
+ exit 1
+fi
+echo "generated ${sent}/5 post-failover requests" >&2
+
+IFS=':' read -r standby_host standby_port < "${FAILOVER_STANDBY}"
+oap_service_ls_sorted "${standby_host}" "${standby_port}"
diff --git a/tests/unit/test_fork_support_env.py b/tests/unit/test_fork_support_env.py
new file mode 100644
index 0000000..5378004
--- /dev/null
+++ b/tests/unit/test_fork_support_env.py
@@ -0,0 +1,80 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+import subprocess
+import sys
+import unittest
+
+
+class TestForkSupportEnvBeforeGrpcImport(unittest.TestCase):
+
+ def test_import_agent_does_not_load_grpc(self):
+ code = r"""
+import sys
+import os
+os.environ["SW_AGENT_ASYNCIO_ENHANCEMENT"] = "false"
+os.environ["SW_AGENT_EXPERIMENTAL_FORK_SUPPORT"] = "true"
+os.environ["SW_AGENT_PROTOCOL"] = "grpc"
+os.environ["SW_AGENT_COLLECTOR_BACKEND_SERVICES"] = "127.0.0.1:11800"
+import skywalking.agent # noqa: F401
+assert "grpc" not in sys.modules, sorted(m for m in sys.modules if m == "grpc" or m.startswith("grpc."))
+print("OK_NO_GRPC")
+"""
+ r = subprocess.run(
+ [sys.executable, '-c', code],
+ cwd=os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ self.assertEqual(r.returncode, 0, msg=r.stdout + r.stderr)
+ self.assertIn('OK_NO_GRPC', r.stdout)
+
+ def test_prefork_master_enables_fork_support(self):
+ code = r"""
+import sys
+import os
+os.environ["SW_AGENT_ASYNCIO_ENHANCEMENT"] = "false"
+os.environ["SW_AGENT_EXPERIMENTAL_FORK_SUPPORT"] = "true"
+os.environ["SW_AGENT_PROTOCOL"] = "grpc"
+os.environ["SW_AGENT_COLLECTOR_BACKEND_SERVICES"] = "127.0.0.1:11800"
+os.environ["SW_AGENT_NAME"] = "fork-env-test"
+os.environ["SW_AGENT_INSTANCE_NAME"] = "i1"
+# Avoid plugin side effects that need a live collector.
+os.environ["SW_AGENT_DISABLE_PLUGINS"] = ".*"
+from skywalking.agent import agent
+assert "grpc" not in sys.modules
+agent.start_prefork_master()
+import grpc # noqa: F401
+from grpc._cython import cygrpc
+assert cygrpc.is_fork_support_enabled(), "GRPC_ENABLE_FORK_SUPPORT must be set before import grpc"
+print("OK_FORK_SUPPORT")
+"""
+ r = subprocess.run(
+ [sys.executable, '-c', code],
+ cwd=os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ self.assertEqual(r.returncode, 0, msg=r.stdout + r.stderr)
+ self.assertIn('OK_FORK_SUPPORT', r.stdout)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tests/unit/test_grpc_channel.py b/tests/unit/test_grpc_channel.py
new file mode 100644
index 0000000..bd73fc5
--- /dev/null
+++ b/tests/unit/test_grpc_channel.py
@@ -0,0 +1,732 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import socket
+import json
+import unittest
+import asyncio
+from queue import Queue
+from time import monotonic
+from unittest.mock import MagicMock, patch
+
+import grpc
+
+from skywalking.utils.grpc_channel import (
+ GRPC_CHANNEL_OPTIONS,
+ _GRPC_RPC_TIMEOUT_MARGIN_SEC,
+ AddressKind,
+ BackendAddress,
+ build_grpc_target,
+ encode_sw_static_for_c_core,
+ expand_backend_addresses,
+ grpc_call_timeout,
+ handle_rpc_error,
+ is_auth_rpc_error,
+ is_channel_ready,
+ parse_backend_address,
+ parse_backend_addresses,
+ prepare_grpc_channel_endpoints,
+ resolve_grpc_target,
+ sw_static_endpoints,
+)
+
+
+class TestGrpcBackendAddress(unittest.TestCase):
+
+ def test_parse_ipv4_and_hostname(self):
+ v4 = parse_backend_address('127.0.0.1:11800')
+ self.assertEqual(v4.host, '127.0.0.1')
+ self.assertEqual(v4.port, 11800)
+ self.assertEqual(v4.kind.value, 'ipv4')
+
+ host = parse_backend_address('oap.example.com:11800')
+ self.assertEqual(host.host, 'oap.example.com')
+ self.assertEqual(host.kind.value, 'hostname')
+
+ def test_parse_ipv6_requires_brackets(self):
+ v6 = parse_backend_address('[::1]:11800')
+ self.assertEqual(v6.host, '::1')
+ self.assertEqual(v6.port, 11800)
+ self.assertEqual(v6.kind.value, 'ipv6')
+ self.assertIsNone(parse_backend_address('::1:11800'))
+
+ def test_parse_invalid_logged_and_skipped(self):
+ with self.assertLogs('skywalking', level='ERROR') as cm:
+ addrs = parse_backend_addresses('127.0.0.1:11800,bad-entry,10.0.0.2:11800')
+ self.assertEqual(len(addrs), 2)
+ self.assertTrue(any('bad-entry' in line for line in cm.output))
+
+ def test_single_target_plain(self):
+ self.assertEqual(
+ build_grpc_target(parse_backend_addresses('oap.svc:11800')),
+ 'oap.svc:11800',
+ )
+ self.assertEqual(
+ build_grpc_target(parse_backend_addresses('127.0.0.1:11800')),
+ '127.0.0.1:11800',
+ )
+
+ def test_multi_ipv4_static_target(self):
+ target = build_grpc_target(parse_backend_addresses('10.0.0.1:11800,10.0.0.2:11800'))
+ self.assertEqual(target, 'ipv4:10.0.0.1:11800,10.0.0.2:11800')
+
+ def test_multi_ipv6_static_target(self):
+ target = build_grpc_target(parse_backend_addresses('[::1]:11800,[::2]:11800'))
+ self.assertEqual(target, 'ipv6:[::1]:11800,[::2]:11800')
+
+ def test_multi_hostname_expands_to_ipv4_static(self):
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ mapping = {
+ 'oap-a': [('10.0.0.1', port)],
+ 'oap-b': [('10.0.0.2', port)],
+ }
+ return [
+ (socket.AF_INET, socket.SOCK_STREAM, 6, '', (ip, p))
+ for ip, p in mapping[host]
+ ]
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo):
+ target = build_grpc_target(parse_backend_addresses('oap-a:11800,oap-b:11800'))
+ self.assertEqual(target, 'ipv4:10.0.0.1:11800,10.0.0.2:11800')
+
+ def test_mixed_hostname_and_ip_expands(self):
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.9', port))]
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo):
+ target = build_grpc_target(
+ parse_backend_addresses('10.0.0.1:11800,oap-b:11800')
+ )
+ self.assertEqual(target, 'ipv4:10.0.0.1:11800,10.0.0.9:11800')
+
+ def test_mixed_families_encoded_as_ipv4_mapped(self):
+ addrs = [
+ BackendAddress('10.0.0.1', 11800, AddressKind.IPV4),
+ BackendAddress('::1', 11800, AddressKind.IPV6),
+ ]
+ target = build_grpc_target(addrs)
+ self.assertEqual(target, 'ipv6:[::ffff:10.0.0.1]:11800,[::1]:11800')
+ self.assertEqual(
+ sw_static_endpoints(addrs),
+ [
+ {'addresses': [{'host': '10.0.0.1', 'port': 11800}]},
+ {'addresses': [{'host': '::1', 'port': 11800}]},
+ ],
+ )
+
+ def test_hostname_dual_stack_keeps_both_families(self):
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ if host == 'oap-a':
+ return [
+ (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.1', port)),
+ (socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('2001:db8::1', port)),
+ ]
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.2', port))]
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo):
+ target = build_grpc_target(parse_backend_addresses('oap-a:11800,oap-b:11800'))
+ self.assertEqual(
+ target,
+ 'ipv6:[::ffff:10.0.0.1]:11800,[2001:db8::1]:11800,[::ffff:10.0.0.2]:11800',
+ )
+
+ def test_encode_rejects_hostname(self):
+ with self.assertRaises(ValueError):
+ encode_sw_static_for_c_core([
+ BackendAddress('oap.svc', 11800, AddressKind.HOSTNAME),
+ ])
+
+ def test_expand_skips_failed_hostname(self):
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ if host == 'bad.host':
+ raise socket.gaierror(socket.EAI_NONAME, 'Name or service not known')
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.3', port))]
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo):
+ with self.assertLogs('skywalking', level='ERROR'):
+ expanded = expand_backend_addresses(
+ parse_backend_addresses('bad.host:11800,ok.host:11800')
+ )
+ self.assertEqual([a.endpoint() for a in expanded], ['10.0.0.3:11800'])
+
+ def test_authority_skips_failed_first_hostname(self):
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ if host == 'bad.host':
+ raise socket.gaierror(socket.EAI_NONAME, 'Name or service not known')
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.2', port))]
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo):
+ with self.assertLogs('skywalking', level='ERROR'):
+ target, authority = prepare_grpc_channel_endpoints(
+ parse_backend_addresses('bad.host:11800,good.host:11800')
+ )
+ self.assertEqual(target, 'ipv4:10.0.0.2:11800')
+ self.assertEqual(authority, 'good.host:11800')
+
+ def test_rejects_ipv6_zone_and_control_chars(self):
+ self.assertIsNone(parse_backend_address('[fe80::1%eth0]:11800'))
+ self.assertIsNone(parse_backend_address('bad\nhost:11800'))
+ self.assertIsNone(parse_backend_address('has space:11800'))
+
+ def test_dns_timeout_returns_quickly_without_joining_worker(self):
+ import threading
+ import time
+ from skywalking.utils import grpc_channel as mod
+
+ def hang_getaddrinfo(*_args, **_kwargs):
+ time.sleep(30)
+ return []
+
+ previous = mod._DNS_LOOKUP_TIMEOUT_SEC
+ try:
+ mod._DNS_LOOKUP_TIMEOUT_SEC = 0.3
+ before = {t.ident for t in threading.enumerate()}
+ t0 = time.monotonic()
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=hang_getaddrinfo):
+ with self.assertLogs('skywalking', level='ERROR'):
+ result = mod._lookup_hostname('slow.host', 11800)
+ elapsed = time.monotonic() - t0
+ leftover = [
+ t for t in threading.enumerate()
+ if t.ident not in before and t.is_alive()
+ ]
+ finally:
+ mod._DNS_LOOKUP_TIMEOUT_SEC = previous
+
+ self.assertEqual(result, [])
+ self.assertLess(elapsed, 2.0)
+ # Hung lookup may still be running, but must be daemon so exit is not blocked.
+ for t in leftover:
+ self.assertTrue(t.daemon, msg=f'non-daemon leftover thread: {t.name}')
+
+ def test_all_hostname_resolve_fail_raises(self):
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ raise socket.gaierror(socket.EAI_NONAME, 'Name or service not known')
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo):
+ with self.assertLogs('skywalking', level='ERROR'):
+ with self.assertRaises(ValueError):
+ build_grpc_target(parse_backend_addresses('a.host:11800,b.host:11800'))
+
+ def test_empty_raises(self):
+ with self.assertRaises(ValueError):
+ build_grpc_target([])
+
+ def test_channel_options_disable_proxy_no_keepalive(self):
+ keys = {k for k, _ in GRPC_CHANNEL_OPTIONS}
+ self.assertIn('grpc.enable_http_proxy', keys)
+ self.assertNotIn('grpc.lb_policy_name', keys)
+ self.assertEqual(dict(GRPC_CHANNEL_OPTIONS)['grpc.enable_http_proxy'], 0)
+ self.assertEqual(dict(GRPC_CHANNEL_OPTIONS)['grpc.max_reconnect_backoff_ms'], 30000)
+ self.assertFalse(any('keepalive' in k for k in keys))
+
+ def test_channel_options_properties_retry_service_config(self):
+ opts = dict(GRPC_CHANNEL_OPTIONS)
+ self.assertEqual(opts['grpc.enable_retries'], 1)
+ cfg = json.loads(opts['grpc.service_config'])
+ methods = cfg['methodConfig']
+ self.assertEqual(len(methods), 1)
+ names = methods[0]['name']
+ self.assertEqual(names, [{
+ 'service': 'skywalking.v3.ManagementService',
+ 'method': 'reportInstanceProperties',
+ }])
+ policy = methods[0]['retryPolicy']
+ self.assertEqual(policy['maxAttempts'], 3)
+ self.assertEqual(policy['retryableStatusCodes'], ['UNAVAILABLE'])
+ # Streaming collect must not appear — retries would duplicate segments.
+ blob = opts['grpc.service_config']
+ self.assertNotIn('collect', blob)
+ self.assertNotIn('keepAlive', blob)
+ lb = cfg['loadBalancingConfig']
+ self.assertEqual(lb, [{'pick_first': {'shuffleAddressList': True}}])
+
+ def test_resolve_uses_config(self):
+ from skywalking import config
+
+ previous = config.agent_collector_backend_services
+ try:
+ config.agent_collector_backend_services = '1.1.1.1:11800,1.1.1.2:11800'
+ self.assertEqual(resolve_grpc_target(), 'ipv4:1.1.1.1:11800,1.1.1.2:11800')
+ finally:
+ config.agent_collector_backend_services = previous
+
+ def test_create_sync_channel_tls_passes_authority(self):
+ from skywalking import config
+ from skywalking.utils.grpc_channel import create_sync_channel
+
+ previous = config.agent_collector_backend_services
+ previous_tls = config.agent_force_tls
+ try:
+ config.agent_collector_backend_services = 'oap.example:11800,10.0.0.2:11800'
+ config.agent_force_tls = True
+
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.1', port))]
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo), \
+ patch('skywalking.utils.grpc_channel.grpc.secure_channel') as secure, \
+ patch('skywalking.utils.grpc_channel.grpc.ssl_channel_credentials', return_value='creds'):
+ create_sync_channel()
+ args, kwargs = secure.call_args
+ self.assertEqual(args[0], 'ipv4:10.0.0.1:11800,10.0.0.2:11800')
+ opts = dict(kwargs['options'])
+ self.assertEqual(opts['grpc.default_authority'], 'oap.example:11800')
+ finally:
+ config.agent_collector_backend_services = previous
+ config.agent_force_tls = previous_tls
+
+
+class TestAuthRpcHandling(unittest.TestCase):
+
+ def _rpc_error(self, status):
+ err = MagicMock()
+ err.code = MagicMock(return_value=status)
+ return err
+
+ def test_auth_errors_detected(self):
+ self.assertTrue(is_auth_rpc_error(self._rpc_error(grpc.StatusCode.UNAUTHENTICATED)))
+ self.assertTrue(is_auth_rpc_error(self._rpc_error(grpc.StatusCode.PERMISSION_DENIED)))
+ self.assertFalse(is_auth_rpc_error(self._rpc_error(grpc.StatusCode.UNAVAILABLE)))
+
+ def test_auth_does_not_invoke_connectivity_hook(self):
+ hook = MagicMock()
+ with patch('skywalking.utils.grpc_channel._last_auth_log_at', 0):
+ handle_rpc_error(self._rpc_error(grpc.StatusCode.UNAUTHENTICATED), hook)
+ hook.assert_not_called()
+
+ def test_unavailable_invokes_connectivity_hook(self):
+ hook = MagicMock()
+ handle_rpc_error(self._rpc_error(grpc.StatusCode.UNAVAILABLE), hook)
+ hook.assert_called_once()
+
+
+class TestReadyGate(unittest.TestCase):
+
+ def test_ready_true_only_for_ready_state(self):
+ channel = MagicMock()
+ channel.get_state.return_value = grpc.ChannelConnectivity.READY
+ # Ensure unwrap prefers public get_state (aio path).
+ channel._channel = MagicMock()
+ self.assertTrue(is_channel_ready(channel))
+ channel.get_state.assert_called_with(True)
+
+ def test_non_ready_states_skip(self):
+ channel = MagicMock()
+ for state in (
+ grpc.ChannelConnectivity.IDLE,
+ grpc.ChannelConnectivity.CONNECTING,
+ grpc.ChannelConnectivity.TRANSIENT_FAILURE,
+ grpc.ChannelConnectivity.SHUTDOWN,
+ ):
+ channel.get_state.return_value = state
+ self.assertFalse(is_channel_ready(channel), msg=str(state))
+
+ def test_sync_channel_without_get_state_uses_cython_check(self):
+ # grpcio sync Channel has subscribe but no get_state — must not fail-closed forever.
+ class SyncLikeChannel:
+ pass
+
+ channel = SyncLikeChannel()
+ cython = MagicMock()
+ cython.check_connectivity_state.return_value = grpc.ChannelConnectivity.READY.value[0]
+ channel._channel = cython
+ self.assertTrue(is_channel_ready(channel))
+ cython.check_connectivity_state.assert_called_with(True)
+
+ cython.check_connectivity_state.return_value = grpc.ChannelConnectivity.IDLE.value[0]
+ self.assertFalse(is_channel_ready(channel))
+
+ def test_intercept_channel_unwraps_to_cython_check(self):
+ class InterceptLike:
+ pass
+
+ class SyncLike:
+ pass
+
+ intercept = InterceptLike()
+ sync = SyncLike()
+ cython = MagicMock()
+ cython.check_connectivity_state.return_value = grpc.ChannelConnectivity.READY.value[0]
+ sync._channel = cython
+ intercept._channel = sync
+ self.assertTrue(is_channel_ready(intercept))
+ cython.check_connectivity_state.assert_called_with(True)
+
+ def test_unknown_channel_fail_open(self):
+ # Cannot read connectivity → do not permanently silence reporters.
+ self.assertTrue(is_channel_ready(object()))
+
+
+class TestLogThrottle(unittest.TestCase):
+
+ def test_reporter_exception_throttled(self):
+ from skywalking.utils import reporter_log as mod
+
+ # patch replaces the module dict for this test only (auto-restored);
+ # do not .clear() the shared throttle state — that leaks across tests.
+ with patch.object(mod, '_last_reporter_log_at', {}):
+ with self.assertLogs('skywalking', level='ERROR') as cm:
+ try:
+ raise RuntimeError('boom')
+ except RuntimeError:
+ mod.log_reporter_exception_throttled('segment', 1)
+ mod.log_reporter_exception_throttled('segment', 2)
+ self.assertEqual(len(cm.records), 1)
+
+ def test_connectivity_event_throttled(self):
+ from skywalking.utils import grpc_channel as mod
+
+ with patch.object(mod, '_last_connectivity_log_at', {}):
+ with self.assertLogs('skywalking', level='WARNING') as cm:
+ mod.log_connectivity_event('transient_failure', 'down1')
+ mod.log_connectivity_event('transient_failure', 'down2')
+ self.assertEqual(len(cm.records), 1)
+
+ def test_dropped_throttled_includes_delta_and_total(self):
+ from skywalking.utils import reporter_log as mod
+
+ with patch.object(mod, '_last_drop_log_at', {}), \
+ patch.object(mod, '_drop_totals', {}), \
+ patch.object(mod, '_drop_logged_totals', {}):
+ with self.assertLogs('skywalking', level='WARNING') as cm:
+ mod.log_dropped_throttled('segment', 2)
+ mod.log_dropped_throttled('segment', 3)
+ self.assertEqual(len(cm.records), 1)
+ self.assertIn('+2 since last log', cm.records[0].getMessage())
+ self.assertIn('2 total', cm.records[0].getMessage())
+
+
+class TestCreateChannelDoesNotRaise(unittest.TestCase):
+
+ def _assert_factory_degrades(self, services: str):
+ from skywalking import config
+ from skywalking.utils.grpc_channel import create_sync_channel
+
+ previous = config.agent_collector_backend_services
+ channel = MagicMock()
+ channel.get_state.return_value = grpc.ChannelConnectivity.IDLE
+ try:
+ config.agent_collector_backend_services = services
+ with patch('skywalking.utils.grpc_channel.grpc.insecure_channel', return_value=channel) as insecure:
+ with self.assertLogs('skywalking', level='ERROR'):
+ got = create_sync_channel()
+ self.assertIs(got, channel)
+ insecure.assert_called()
+ self.assertNotEqual(got.get_state(), grpc.ChannelConnectivity.READY)
+ finally:
+ config.agent_collector_backend_services = previous
+
+ def test_empty_config_does_not_raise(self):
+ self._assert_factory_degrades('')
+
+ def test_garbage_config_does_not_raise(self):
+ self._assert_factory_degrades('not-an-address,also bad')
+
+ def test_unresolvable_hostnames_do_not_raise(self):
+ def fake_getaddrinfo(host, port, type=0, *args, **kwargs):
+ raise socket.gaierror(socket.EAI_NONAME, 'Name or service not known')
+
+ with patch('skywalking.utils.grpc_channel.socket.getaddrinfo', side_effect=fake_getaddrinfo):
+ self._assert_factory_degrades('no.such.host.invalid:11800,also.invalid:11800')
+
+
+class TestProfilingSnapshotNonBlocking(unittest.TestCase):
+
+ def test_full_snapshot_queue_does_not_block(self):
+ from queue import Queue
+ from threading import Event, Thread
+
+ from skywalking.agent import SkyWalkingAgent
+
+ agent = SkyWalkingAgent.__new__(SkyWalkingAgent)
+ agent._SkyWalkingAgent__reporting = True
+ q = Queue(maxsize=1)
+ q.put('full')
+ agent._SkyWalkingAgent__snapshot_queue = q
+
+ done = Event()
+
+ def _put():
+ agent.add_profiling_snapshot('next')
+ done.set()
+
+ Thread(target=_put, daemon=True).start()
+ self.assertTrue(done.wait(1.0), 'add_profiling_snapshot blocked on a full queue')
+ self.assertEqual(q.qsize(), 1)
+
+
+class TestGrpcCallTimeoutAndKeepAlive(unittest.TestCase):
+
+ def test_rpc_timeout_exceeds_queue_window(self):
+ from skywalking import config
+
+ prev = config.agent_queue_timeout
+ try:
+ config.agent_queue_timeout = 1
+ self.assertEqual(grpc_call_timeout(), 10.0)
+ config.agent_queue_timeout = 20
+ self.assertEqual(grpc_call_timeout(), 25.0)
+ finally:
+ config.agent_queue_timeout = prev
+
+ def test_sync_collect_passes_timeout(self):
+ from skywalking.client.grpc import GrpcTraceSegmentReportService
+
+ stub = MagicMock()
+ svc = GrpcTraceSegmentReportService.__new__(GrpcTraceSegmentReportService)
+ svc.report_stub = stub
+ svc.report(iter(()))
+ self.assertEqual(stub.collect.call_args.kwargs.get('timeout'), grpc_call_timeout())
+
+ def test_keep_alive_after_properties_refresh_failure(self):
+ from skywalking.client.grpc import GrpcServiceManagementClient
+
+ class FakeRpcError(grpc.RpcError):
+ def code(self):
+ return grpc.StatusCode.UNAVAILABLE
+
+ def details(self):
+ return 'props failed'
+
+ client = GrpcServiceManagementClient.__new__(GrpcServiceManagementClient)
+ client.service_stub = MagicMock()
+ client.refresh_instance_props = MagicMock(side_effect=FakeRpcError())
+ client.send_heart_beat()
+ client.service_stub.keepAlive.assert_called_once()
+ self.assertEqual(
+ client.service_stub.keepAlive.call_args.kwargs.get('timeout'),
+ grpc_call_timeout(),
+ )
+
+
+class TestAioStreamingOmitsDeadline(unittest.IsolatedAsyncioTestCase):
+
+ async def test_aio_collect_omits_timeout(self):
+ from unittest.mock import AsyncMock
+
+ from skywalking.client.grpc_aio import (
+ GrpcLogReportServiceAsync,
+ GrpcMeterReportServiceAsync,
+ GrpcProfileTaskChannelServiceAsync,
+ GrpcTraceSegmentReportServiceAsync,
+ )
+
+ traces = MagicMock()
+ traces.collect = AsyncMock()
+ svc = GrpcTraceSegmentReportServiceAsync.__new__(GrpcTraceSegmentReportServiceAsync)
+ svc.report_stub = traces
+ await svc.report(object())
+ self.assertNotIn('timeout', traces.collect.call_args.kwargs)
+
+ meters = MagicMock()
+ meters.collect = AsyncMock()
+ meters.collectBatch = AsyncMock()
+ meter_svc = GrpcMeterReportServiceAsync.__new__(GrpcMeterReportServiceAsync)
+ meter_svc.report_stub = meters
+ await meter_svc.report(object())
+ await meter_svc.report_batch(object())
+ self.assertNotIn('timeout', meters.collect.call_args.kwargs)
+ self.assertNotIn('timeout', meters.collectBatch.call_args.kwargs)
+
+ logs = MagicMock()
+ logs.collect = AsyncMock()
+ log_svc = GrpcLogReportServiceAsync.__new__(GrpcLogReportServiceAsync)
+ log_svc.report_stub = logs
+ await log_svc.report(object())
+ self.assertNotIn('timeout', logs.collect.call_args.kwargs)
+
+ profile = MagicMock()
+ profile.collectSnapshot = AsyncMock()
+ profile_svc = GrpcProfileTaskChannelServiceAsync.__new__(GrpcProfileTaskChannelServiceAsync)
+ profile_svc.profile_stub = profile
+ await profile_svc.report(object())
+ self.assertNotIn('timeout', profile.collectSnapshot.call_args.kwargs)
+
+ async def test_aio_unary_keeps_timeout(self):
+ from unittest.mock import AsyncMock
+
+ from skywalking.client.grpc_aio import GrpcServiceManagementClientAsync
+
+ client = GrpcServiceManagementClientAsync.__new__(GrpcServiceManagementClientAsync)
+ client.service_stub = MagicMock()
+ client.service_stub.keepAlive = AsyncMock()
+ client.refresh_instance_props = AsyncMock()
+ await client.send_heart_beat()
+ self.assertEqual(
+ client.service_stub.keepAlive.call_args.kwargs.get('timeout'),
+ grpc_call_timeout(),
+ )
+
+
+class TestClosePreviousProtocol(unittest.IsolatedAsyncioTestCase):
+
+ def test_sync_close_never_blocks_on_aclose(self):
+ from skywalking.agent import _close_previous_protocol
+
+ proto = MagicMock()
+ proto.close = MagicMock()
+ proto.aclose = MagicMock()
+ _close_previous_protocol(proto)
+ proto.close.assert_called_once()
+ proto.aclose.assert_not_called()
+ _close_previous_protocol(None)
+
+ async def test_aclose_awaited_on_running_loop(self):
+ from skywalking.agent import _aclose_previous_protocol
+
+ called = []
+ loop = asyncio.get_running_loop()
+
+ class _Proto:
+ async def aclose(self):
+ called.append(loop)
+
+ def close(self):
+ called.append('close')
+
+ await _aclose_previous_protocol(_Proto())
+ self.assertEqual(called, [loop])
+ await _aclose_previous_protocol(None)
+
+ async def test_aclose_falls_back_to_close(self):
+ from skywalking.agent import _aclose_previous_protocol
+
+ proto = MagicMock()
+ proto.aclose = None
+ proto.close = MagicMock()
+ await _aclose_previous_protocol(proto)
+ proto.close.assert_called_once()
+
+
+
+class TestRpcTimeoutVsQueueWindow(unittest.TestCase):
+
+ def test_timeout_has_margin_over_worst_case_batch(self):
+ """RPC timeout must exceed absolute batch window + encode/RTT margin."""
+ from skywalking import config
+
+ prev = config.agent_queue_timeout
+ try:
+ config.agent_queue_timeout = 20
+ timeout = grpc_call_timeout()
+ self.assertEqual(timeout, 20 + _GRPC_RPC_TIMEOUT_MARGIN_SEC)
+ self.assertGreater(timeout, float(config.agent_queue_timeout) + 1.0)
+ finally:
+ config.agent_queue_timeout = prev
+
+ def test_sync_report_uses_timeout_with_margin(self):
+ from skywalking import config
+ from skywalking.client.grpc import GrpcTraceSegmentReportService
+
+ prev = config.agent_queue_timeout
+ try:
+ config.agent_queue_timeout = 20
+ stub = MagicMock()
+ svc = GrpcTraceSegmentReportService.__new__(GrpcTraceSegmentReportService)
+ svc.report_stub = stub
+ svc.report(iter(()))
+ self.assertEqual(
+ stub.collect.call_args.kwargs.get('timeout'),
+ 20 + _GRPC_RPC_TIMEOUT_MARGIN_SEC,
+ )
+ finally:
+ config.agent_queue_timeout = prev
+
+
+class TestQueueGetWithinBatch(unittest.TestCase):
+
+ def test_queue_timeout_zero_drains_immediately_available_item(self):
+ from skywalking.agent.protocol.grpc import _queue_get_within_batch
+
+ q = Queue()
+ q.put('segment')
+ batch_deadline = monotonic()
+ item = _queue_get_within_batch(q, True, batch_deadline, allow_immediate=True)
+ self.assertEqual(item, 'segment')
+ self.assertTrue(q.empty())
+
+ def test_queue_timeout_zero_skips_when_empty(self):
+ from skywalking.agent.protocol.grpc import _queue_get_within_batch
+
+ q = Queue()
+ batch_deadline = monotonic()
+ self.assertIsNone(
+ _queue_get_within_batch(q, True, batch_deadline, allow_immediate=True),
+ )
+
+
+class TestCollectorChannelNotInstrumented(unittest.TestCase):
+
+ def test_multi_address_collector_channel_skips_sw_interceptor(self):
+ """Regression: ipv4: multi targets must not get sw_grpc client interceptors."""
+ import grpc
+
+ from skywalking import config
+ from skywalking.plugins import sw_grpc
+ from skywalking.utils.grpc_channel import (
+ create_sync_channel,
+ is_agent_collector_channel,
+ )
+
+ prev = config.agent_collector_backend_services
+ sw_grpc.install_sync()
+ try:
+ config.agent_collector_backend_services = '10.0.0.1:11800,10.0.0.2:11800'
+ with patch('grpc.intercept_channel') as intercept:
+ channel = create_sync_channel()
+ intercept.assert_not_called()
+ self.assertTrue(is_agent_collector_channel(channel))
+ with patch('grpc.intercept_channel',
+ side_effect=lambda c, *a, **k: c) as intercept:
+ grpc.insecure_channel('business.example:50051')
+ intercept.assert_called()
+ finally:
+ config.agent_collector_backend_services = prev
+
+ def test_aio_multi_address_uses_collector_scope(self):
+ from skywalking import config
+ from skywalking.plugins import sw_grpc
+ from skywalking.utils.grpc_channel import (
+ create_aio_channel,
+ is_agent_collector_channel,
+ is_building_agent_collector_channel,
+ )
+
+ prev = config.agent_collector_backend_services
+ sw_grpc.install_async()
+ seen_building = []
+
+ class _Probe:
+ def __init__(self, *args, **kwargs):
+ seen_building.append(is_building_agent_collector_channel())
+ # Minimal stand-in; create_aio_channel only needs a return object.
+ self._sw_agent_collector_channel = False
+
+ try:
+ config.agent_collector_backend_services = '10.0.0.1:11800,10.0.0.2:11800'
+ with patch('skywalking.utils.grpc_channel.grpc.aio.insecure_channel', side_effect=_Probe):
+ channel = create_aio_channel()
+ self.assertEqual(seen_building, [True])
+ self.assertTrue(is_agent_collector_channel(channel))
+ self.assertFalse(is_building_agent_collector_channel())
+ finally:
+ config.agent_collector_backend_services = prev
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tests/unit/test_grpc_ready_gate.py b/tests/unit/test_grpc_ready_gate.py
new file mode 100644
index 0000000..795574c
--- /dev/null
+++ b/tests/unit/test_grpc_ready_gate.py
@@ -0,0 +1,167 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+import grpc
+
+from skywalking.agent.protocol.grpc import GrpcProtocol
+
+
+class TestSyncGrpcReadyGate(unittest.TestCase):
+
+ def _protocol(self) -> GrpcProtocol:
+ channel = MagicMock()
+ with patch('skywalking.agent.protocol.grpc.create_sync_channel', return_value=channel), \
+ patch('skywalking.agent.protocol.grpc.GrpcServiceManagementClient'), \
+ patch('skywalking.agent.protocol.grpc.GrpcTraceSegmentReportService'), \
+ patch('skywalking.agent.protocol.grpc.GrpcProfileTaskChannelService'), \
+ patch('skywalking.agent.protocol.grpc.GrpcLogDataReportService'), \
+ patch('skywalking.agent.protocol.grpc.GrpcMeterReportService'):
+ return GrpcProtocol()
+
+ def test_is_ready_follows_subscribe_state(self):
+ protocol = self._protocol()
+ self.assertFalse(protocol.is_ready())
+
+ protocol.state = grpc.ChannelConnectivity.CONNECTING
+ self.assertFalse(protocol.is_ready())
+
+ protocol.state = grpc.ChannelConnectivity.READY
+ self.assertTrue(protocol.is_ready())
+
+ protocol.properties_sent = True
+ protocol.service_management.sent_properties_counter = 7
+ protocol._cb(grpc.ChannelConnectivity.TRANSIENT_FAILURE)
+ self.assertFalse(protocol.properties_sent)
+ self.assertEqual(protocol.service_management.sent_properties_counter, 0)
+
+ protocol.state = grpc.ChannelConnectivity.IDLE
+ with patch('skywalking.agent.protocol.grpc.is_channel_ready') as nudge:
+ nudge.return_value = False
+ self.assertFalse(protocol.is_ready())
+ nudge.assert_called_once_with(protocol.channel)
+
+ def test_heartbeat_keep_alive_after_instance_props_failure(self):
+ class FakeRpcError(grpc.RpcError):
+ def code(self):
+ return grpc.StatusCode.UNAVAILABLE
+
+ def details(self):
+ return 'props failed'
+
+ protocol = self._protocol()
+ protocol.state = grpc.ChannelConnectivity.READY
+ protocol.properties_sent = False
+ protocol.service_management.send_instance_props = MagicMock(side_effect=FakeRpcError())
+ protocol.service_management.send_heart_beat = MagicMock()
+ protocol.heartbeat()
+ protocol.service_management.send_heart_beat.assert_called_once()
+ self.assertFalse(protocol.properties_sent)
+
+ def test_failed_segment_batch_counts_drops(self):
+ from queue import Queue
+
+ class FakeRpcError(grpc.RpcError):
+ def code(self):
+ return grpc.StatusCode.UNAVAILABLE
+
+ def details(self):
+ return 'collect failed'
+
+ protocol = self._protocol()
+ protocol.state = grpc.ChannelConnectivity.READY
+ protocol.on_error = MagicMock()
+
+ segment = MagicMock()
+ segment.related_traces = ['trace']
+ segment.segment_id = 'seg'
+ segment.is_size_limited = False
+ segment.spans = []
+
+ queue = Queue()
+ queue.put(segment)
+
+ def _report(generator):
+ list(generator)
+ raise FakeRpcError()
+
+ protocol.traces_reporter.report = _report
+ with patch('skywalking.agent.protocol.grpc.log_dropped_throttled') as dropped, \
+ patch('skywalking.agent.protocol.grpc.SegmentObject', return_value=object()), \
+ patch('skywalking.agent.protocol.grpc.handle_rpc_error'):
+ with self.assertRaises(FakeRpcError):
+ protocol.report_segment(queue, block=False)
+ dropped.assert_called_with('segment', 1)
+
+ def test_properties_refresh_every_factor_heartbeats(self):
+ """Java/Node cadence: reportInstanceProperties every N keepAlive ticks."""
+ from skywalking import config
+ from skywalking.client import ServiceManagementClient
+
+ class _Client(ServiceManagementClient):
+ def send_instance_props(self) -> None:
+ pass
+
+ client = _Client()
+ client.send_instance_props = MagicMock()
+
+ prev = config.agent_collector_properties_report_period_factor
+ try:
+ config.agent_collector_properties_report_period_factor = 3
+ client.refresh_instance_props() # 1
+ client.refresh_instance_props() # 2
+ self.assertEqual(client.send_instance_props.call_count, 0)
+ client.refresh_instance_props() # 3
+ self.assertEqual(client.send_instance_props.call_count, 1)
+ client.refresh_instance_props() # 4
+ client.refresh_instance_props() # 5
+ client.refresh_instance_props() # 6
+ self.assertEqual(client.send_instance_props.call_count, 2)
+ finally:
+ config.agent_collector_properties_report_period_factor = prev
+
+
+class TestAsyncGrpcReadyGate(unittest.TestCase):
+
+ def test_aio_is_ready_follows_watched_state(self):
+ from skywalking.agent.protocol.grpc_aio import GrpcProtocolAsync
+
+ channel = MagicMock()
+ channel.get_state.return_value = grpc.ChannelConnectivity.CONNECTING
+ with patch('skywalking.agent.protocol.grpc_aio.create_aio_channel', return_value=channel), \
+ patch('skywalking.agent.protocol.grpc_aio.GrpcServiceManagementClientAsync'), \
+ patch('skywalking.agent.protocol.grpc_aio.GrpcTraceSegmentReportServiceAsync'), \
+ patch('skywalking.agent.protocol.grpc_aio.GrpcProfileTaskChannelServiceAsync'), \
+ patch('skywalking.agent.protocol.grpc_aio.GrpcLogReportServiceAsync'), \
+ patch('skywalking.agent.protocol.grpc_aio.GrpcMeterReportServiceAsync'):
+ protocol = GrpcProtocolAsync()
+
+ protocol.state = grpc.ChannelConnectivity.READY
+ self.assertTrue(protocol.is_ready())
+ protocol.properties_sent.set()
+ protocol.service_management.sent_properties_counter = 4
+ protocol._on_connectivity(grpc.ChannelConnectivity.TRANSIENT_FAILURE)
+ self.assertFalse(protocol.properties_sent.is_set())
+ self.assertEqual(protocol.service_management.sent_properties_counter, 0)
+ protocol.state = grpc.ChannelConnectivity.TRANSIENT_FAILURE
+ self.assertFalse(protocol.is_ready())
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/tests/unit/test_shutdown_queue.py b/tests/unit/test_shutdown_queue.py
new file mode 100644
index 0000000..5258fe9
--- /dev/null
+++ b/tests/unit/test_shutdown_queue.py
@@ -0,0 +1,377 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import asyncio
+import logging
+import time
+import unittest
+from queue import Queue
+from threading import Event, Thread
+
+from skywalking.agent import (
+ _abandon_async_queue,
+ _abandon_sync_queue,
+ _await_shutdown_or_background_failure,
+ _cancel_pending_tasks,
+ _join_sync_queue,
+ _shutdown_async_queue,
+ _shutdown_sync_queue,
+)
+
+
+class TestShutdownQueueHelpers(unittest.TestCase):
+
+ def test_abandon_sync_queue_unblocks_join(self):
+ q = Queue()
+ q.put('a')
+ q.put('b')
+ self.assertEqual(q.unfinished_tasks, 2)
+
+ hung = Event()
+
+ def _join():
+ q.join()
+ hung.set()
+
+ Thread(target=_join, daemon=True).start()
+ time.sleep(0.05)
+ self.assertFalse(hung.is_set())
+
+ abandoned = _abandon_sync_queue(q)
+ self.assertEqual(abandoned, 2)
+ self.assertTrue(hung.wait(1.0))
+ self.assertTrue(_join_sync_queue(q, 0.5))
+
+ def test_shutdown_sync_skips_flush_when_not_ready(self):
+ q = Queue()
+ q.put('x')
+ called = []
+
+ def report():
+ called.append(1)
+ raise AssertionError('must not flush when may_send is False')
+
+ _shutdown_sync_queue(report, q, 'test', may_send=False)
+ self.assertEqual(called, [])
+ self.assertTrue(q.empty())
+ self.assertEqual(q.unfinished_tasks, 0)
+
+ def test_shutdown_sync_flush_timeout_then_abandon(self):
+ q = Queue()
+ q.put('x')
+ started = Event()
+
+ def report():
+ started.set()
+ time.sleep(10) # longer than flush budget
+
+ # Temporarily shrink budget via monkeypatch on module constant
+ import skywalking.agent as agent_mod
+
+ previous = agent_mod._SHUTDOWN_FLUSH_TIMEOUT_SEC
+ try:
+ agent_mod._SHUTDOWN_FLUSH_TIMEOUT_SEC = 0.2
+ t0 = time.monotonic()
+ _shutdown_sync_queue(report, q, 'test', may_send=True)
+ elapsed = time.monotonic() - t0
+ finally:
+ agent_mod._SHUTDOWN_FLUSH_TIMEOUT_SEC = previous
+
+ self.assertTrue(started.wait(1.0))
+ self.assertLess(elapsed, 2.0)
+ self.assertTrue(q.empty())
+ self.assertEqual(q.unfinished_tasks, 0)
+
+ def test_abandon_async_queue_unblocks_join(self):
+ async def _run():
+ q = asyncio.Queue()
+ await q.put('a')
+ await q.put('b')
+ join_task = asyncio.create_task(q.join())
+ await asyncio.sleep(0.05)
+ self.assertFalse(join_task.done())
+ abandoned = await _abandon_async_queue(q)
+ self.assertEqual(abandoned, 2)
+ await asyncio.wait_for(join_task, timeout=1.0)
+
+ asyncio.run(_run())
+
+ def test_cancel_pending_tasks_excludes_caller(self):
+ """Regression: gathering the caller's own task made shutdown hang until the outer budget."""
+ loop = asyncio.new_event_loop()
+ thread = Thread(target=loop.run_forever, daemon=True)
+ thread.start()
+
+ async def _forever():
+ while True:
+ await asyncio.sleep(0.05)
+
+ try:
+ async def _spawn():
+ return [asyncio.create_task(_forever()) for _ in range(2)]
+
+ reporter_tasks = asyncio.run_coroutine_threadsafe(_spawn(), loop).result(timeout=2.0)
+ time.sleep(0.1)
+
+ t0 = time.monotonic()
+ future = asyncio.run_coroutine_threadsafe(_cancel_pending_tasks(reporter_tasks), loop)
+ future.result(timeout=2.0)
+ self.assertLess(time.monotonic() - t0, 2.0)
+ for task in reporter_tasks:
+ self.assertTrue(task.cancelled() or task.done())
+ finally:
+ loop.call_soon_threadsafe(loop.stop)
+ thread.join(timeout=2.0)
+ loop.close()
+
+ def test_async_shutdown_cleanup_runs_inside_asyncio_run_root(self):
+ """
+ Production topology: root waits on _finished, then cancels reporters and
+ awaits a yielding protocol aclose() before asyncio.run returns.
+ """
+ holder = {}
+ aclose_entered = Event()
+ aclose_done = Event()
+ loop_ready = Event()
+
+ async def yielding_aclose():
+ aclose_entered.set()
+ await asyncio.sleep(0.05)
+ aclose_done.set()
+
+ async def root():
+ finished = asyncio.Event()
+ holder['finished'] = finished
+ holder['loop'] = asyncio.get_running_loop()
+
+ async def reporter():
+ while not finished.is_set():
+ await asyncio.sleep(0.02)
+
+ tasks = {asyncio.create_task(reporter()) for _ in range(2)}
+ loop_ready.set()
+ await finished.wait()
+ await _cancel_pending_tasks(tasks)
+ await yielding_aclose()
+ holder['root_finished'] = True
+
+ thread = Thread(target=lambda: asyncio.run(root()), daemon=True)
+ thread.start()
+ self.assertTrue(loop_ready.wait(3.0))
+ holder['loop'].call_soon_threadsafe(holder['finished'].set)
+ thread.join(timeout=5.0)
+ self.assertTrue(aclose_entered.wait(2.0))
+ self.assertTrue(aclose_done.wait(2.0))
+ self.assertTrue(holder.get('root_finished', False))
+
+ def test_background_task_failure_is_logged_not_silenced(self):
+ """Regression: failing background tasks must be observed and logged exactly once."""
+ holder = {'errors': []}
+ error_logged = Event()
+ aclose_done = Event()
+
+ class _Handler(logging.Handler):
+ def emit(self, record):
+ msg = record.getMessage()
+ if 'Error in Python agent asyncio event loop' in msg:
+ holder['errors'].append(msg)
+ error_logged.set()
+
+ agent_logger = logging.getLogger('skywalking')
+ handler = _Handler()
+ agent_logger.addHandler(handler)
+ previous_level = agent_logger.level
+ agent_logger.setLevel(logging.ERROR)
+
+ async def failing_background():
+ await asyncio.sleep(0.02)
+ raise ValueError('command dispatch failed')
+
+ async def yielding_aclose():
+ await asyncio.sleep(0.02)
+ aclose_done.set()
+
+ async def root():
+ finished = asyncio.Event()
+ holder['finished'] = finished
+ failing_task = asyncio.create_task(failing_background())
+
+ async def reporter():
+ while not finished.is_set():
+ await asyncio.sleep(0.05)
+
+ reporter_task = asyncio.create_task(reporter())
+ tasks = {failing_task, reporter_task}
+ await _await_shutdown_or_background_failure(finished, tasks)
+ await _cancel_pending_tasks(tasks)
+ await yielding_aclose()
+ holder['after_wait'] = True
+ holder['failing_task'] = failing_task
+
+ try:
+ asyncio.run(root())
+ self.assertTrue(holder.get('after_wait'))
+ self.assertTrue(error_logged.wait(2.0))
+ self.assertTrue(aclose_done.wait(2.0))
+ self.assertEqual(len(holder['errors']), 1)
+ self.assertIn('command dispatch failed', holder['errors'][0])
+ self.assertTrue(holder['finished'].is_set())
+ self.assertTrue(holder['failing_task'].done())
+ self.assertIsInstance(holder['failing_task'].exception(), ValueError)
+ finally:
+ agent_logger.removeHandler(handler)
+ agent_logger.setLevel(previous_level)
+
+ def test_clean_shutdown_does_not_log_normal_background_completion(self):
+ """Clean shutdown: reporters finish after _finished; cleanup must stay silent."""
+ holder = {'errors': []}
+
+ class _Handler(logging.Handler):
+ def emit(self, record):
+ if 'Error in Python agent asyncio event loop' in record.getMessage():
+ holder['errors'].append(record.getMessage())
+
+ agent_logger = logging.getLogger('skywalking')
+ handler = _Handler()
+ agent_logger.addHandler(handler)
+ previous_level = agent_logger.level
+ agent_logger.setLevel(logging.ERROR)
+
+ async def root():
+ finished = asyncio.Event()
+
+ async def reporter():
+ while not finished.is_set():
+ await asyncio.sleep(0.01)
+
+ tasks = {asyncio.create_task(reporter()) for _ in range(2)}
+ finished.set()
+ await _await_shutdown_or_background_failure(finished, tasks)
+ # Give reporters a turn to observe _finished and return normally.
+ await asyncio.sleep(0.05)
+ await _cancel_pending_tasks(tasks)
+ holder['all_done'] = all(task.done() for task in tasks)
+
+ try:
+ asyncio.run(root())
+ self.assertTrue(holder.get('all_done'))
+ self.assertEqual(holder['errors'], [])
+ finally:
+ agent_logger.removeHandler(handler)
+ agent_logger.setLevel(previous_level)
+
+ def test_unexpected_pre_shutdown_cancellation_is_logged(self):
+ """Cancellation before shutdown must be logged and trigger orderly cleanup."""
+ holder = {'errors': []}
+ error_logged = Event()
+ aclose_done = Event()
+
+ class _Handler(logging.Handler):
+ def emit(self, record):
+ msg = record.getMessage()
+ if 'Error in Python agent asyncio event loop' in msg:
+ holder['errors'].append(msg)
+ error_logged.set()
+
+ agent_logger = logging.getLogger('skywalking')
+ handler = _Handler()
+ agent_logger.addHandler(handler)
+ previous_level = agent_logger.level
+ agent_logger.setLevel(logging.ERROR)
+
+ async def cancelled_background():
+ asyncio.current_task().cancel()
+ await asyncio.sleep(0)
+
+ async def yielding_aclose():
+ await asyncio.sleep(0.02)
+ aclose_done.set()
+
+ async def root():
+ finished = asyncio.Event()
+ holder['finished'] = finished
+
+ async def reporter():
+ while not finished.is_set():
+ await asyncio.sleep(0.05)
+
+ cancelled_task = asyncio.create_task(cancelled_background())
+ reporter_task = asyncio.create_task(reporter())
+ tasks = {cancelled_task, reporter_task}
+ await _await_shutdown_or_background_failure(finished, tasks)
+ await _cancel_pending_tasks(tasks)
+ await yielding_aclose()
+ holder['after_wait'] = True
+
+ try:
+ asyncio.run(root())
+ self.assertTrue(holder.get('after_wait'))
+ self.assertTrue(error_logged.wait(2.0))
+ self.assertTrue(aclose_done.wait(2.0))
+ self.assertEqual(len(holder['errors']), 1)
+ self.assertIn('cancelled unexpectedly', holder['errors'][0])
+ self.assertTrue(holder['finished'].is_set())
+ finally:
+ agent_logger.removeHandler(handler)
+ agent_logger.setLevel(previous_level)
+
+ def test_intentional_cleanup_cancellation_is_silent(self):
+ """Tasks cancelled by _cancel_pending_tasks during cleanup must not ERROR."""
+ holder = {'errors': []}
+
+ class _Handler(logging.Handler):
+ def emit(self, record):
+ if 'Error in Python agent asyncio event loop' in record.getMessage():
+ holder['errors'].append(record.getMessage())
+
+ agent_logger = logging.getLogger('skywalking')
+ handler = _Handler()
+ agent_logger.addHandler(handler)
+ previous_level = agent_logger.level
+ agent_logger.setLevel(logging.ERROR)
+
+ async def forever():
+ while True:
+ await asyncio.sleep(0.05)
+
+ async def root():
+ tasks = {asyncio.create_task(forever()) for _ in range(2)}
+ await _cancel_pending_tasks(tasks)
+ holder['all_done'] = all(task.done() for task in tasks)
+
+ try:
+ asyncio.run(root())
+ self.assertTrue(holder.get('all_done'))
+ self.assertEqual(holder['errors'], [])
+ finally:
+ agent_logger.removeHandler(handler)
+ agent_logger.setLevel(previous_level)
+
+ def test_shutdown_async_queue_bounded(self):
+ async def _run():
+ q = asyncio.Queue()
+ await q.put('a')
+ t0 = time.monotonic()
+ await _shutdown_async_queue(q, 'test')
+ self.assertLess(time.monotonic() - t0, 2.0)
+ self.assertTrue(q.empty())
+
+ asyncio.run(_run())
+
+
+if __name__ == '__main__':
+ unittest.main()
