Public Access
emil
ef3b882e5b
Post-merge / detect-type (push) Successful in 13s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 28s
Post-merge / vikunja (push) Successful in 44s
Post-merge / sync-wiki (push) Successful in 58s
Post-merge / release (push) Successful in 1m7s
Post-merge / publish (push) Successful in 44s
Post-merge / badges (push) Successful in 1m6s
103 lines
3.0 KiB
Python
103 lines
3.0 KiB
Python
"""Network connectivity helpers.
|
|
|
|
Provides retry-aware HTTP connectivity checks and SSH availability
|
|
checks for deployment workflows. Uses ``tenacity`` for exponential
|
|
backoff retry logic.
|
|
|
|
Usage::
|
|
|
|
from devx.utils.network import check_http_connectivity, wait_for_ssh
|
|
|
|
check_http_connectivity("https://auth.example.com")
|
|
wait_for_ssh("178.105.254.83")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import socket
|
|
import time
|
|
from collections.abc import Callable
|
|
|
|
import requests
|
|
from tenacity import (
|
|
Retrying,
|
|
before_sleep_log,
|
|
retry_if_exception_type,
|
|
stop_after_attempt,
|
|
wait_exponential,
|
|
)
|
|
|
|
|
|
def check_http_connectivity(
|
|
base_url: str,
|
|
max_attempts: int = 30,
|
|
*,
|
|
verify: bool = True,
|
|
sleep: Callable[[float], None] | None = None,
|
|
) -> None:
|
|
"""Verify HTTP reachability of *base_url* with retry.
|
|
|
|
Uses tenacity for retry with exponential backoff (2 s min, 10 s max).
|
|
|
|
Args:
|
|
base_url: URL to check via GET request.
|
|
max_attempts: Maximum retry attempts.
|
|
verify: Whether to verify TLS certificates.
|
|
sleep: Custom sleep function for testing (defaults to ``time.sleep``).
|
|
|
|
Raises:
|
|
requests.exceptions.ConnectionError: If the URL is not reachable
|
|
after *max_attempts*.
|
|
"""
|
|
retrying = Retrying(
|
|
stop=stop_after_attempt(max_attempts),
|
|
wait=wait_exponential(multiplier=2, min=2, max=10),
|
|
retry=retry_if_exception_type(requests.exceptions.ConnectionError),
|
|
before_sleep=before_sleep_log(logging.getLogger("devx.utils.network"), logging.WARNING),
|
|
sleep=sleep if sleep is not None else time.sleep,
|
|
reraise=True,
|
|
)
|
|
|
|
def _check() -> None:
|
|
requests.get(base_url, timeout=10, verify=verify) # nosec B501
|
|
|
|
retrying(_check)
|
|
|
|
|
|
def wait_for_ssh(
|
|
host: str,
|
|
port: int = 22,
|
|
max_attempts: int = 30,
|
|
interval: int = 10,
|
|
*,
|
|
sleep: Callable[[float], None] | None = None,
|
|
) -> None:
|
|
"""Wait for SSH to be available on a host using a pure-Python socket check.
|
|
|
|
Uses socket instead of ``nc(1)`` so it works on CI runners without
|
|
netcat. Uses exponential backoff: starts at 2 s, doubles each
|
|
attempt up to 10 s max.
|
|
|
|
Args:
|
|
host: VM IP address or hostname.
|
|
port: SSH port (default 22).
|
|
max_attempts: Maximum number of connection attempts.
|
|
interval: Base interval for backoff calculation (seconds).
|
|
sleep: Custom sleep function for testing (defaults to ``time.sleep``).
|
|
|
|
Raises:
|
|
RuntimeError: If SSH is not available after *max_attempts*.
|
|
"""
|
|
_sleep = sleep if sleep is not None else time.sleep
|
|
for i in range(max_attempts):
|
|
try:
|
|
with socket.create_connection((host, port), timeout=5):
|
|
return
|
|
except OSError:
|
|
pass
|
|
if i < max_attempts - 1:
|
|
wait = min(2 * (2**i), 10)
|
|
_sleep(wait)
|
|
raise RuntimeError(f"SSH not available on {host}:{port} after {max_attempts} attempts")
|