Try to find out what kind of connection work for the docker container.

find_connectable.py

import contextlib
import os
import socket
import sys
import urllib.parse
from collections.abc import Callable, Generator
from functools import cache
from itertools import product
from pathlib import Path
from time import perf_counter, sleep, time
from typing import Final
from uuid import uuid4

from testcontainers.core.config import testcontainers_config as c
from testcontainers.core.container import DockerContainer
from testcontainers.core.utils import default_gateway_ip, inside_container, os_name
from testcontainers.core.waiting_utils import wait_for_logs


BASE_DIR: Final = Path(os.environ.get("BASE_DIR", Path(__file__).resolve().parent))
OUTFILE: Final = BASE_DIR / "output.txt"


def get_running_in_container_id() -> str | None:
    """
    Get the id of the currently running container
    """
    cgroup_file = Path("/proc/self/cgroup")
    if not cgroup_file.is_file():
        return None
    cgroup = cgroup_file.read_text()
    for line in cgroup.splitlines(keepends=False):
        path = line.rpartition(":")[2]
        if path.startswith("/docker"):
            return path.removeprefix("/docker/")
    return None


def output(msg: str) -> None:
    """Write to text file and print, in order to extract all msgs"""
    with OUTFILE.open("a") as f:
        f.write(msg + "\n")
    print(msg, file=sys.stderr)


@contextlib.contextmanager
def time_echo() -> Generator[Callable[[str], None], None, None]:
    def echo(msg: str) -> None:
        print(f"🌈  {perf_counter()-start:.2f} {msg}")

    start = perf_counter()
    yield echo


def create_ryuk(use_network: str = "") -> DockerContainer:
    _container = (
        DockerContainer(c.ryuk_image)
        .with_name(f"ryuk-defined-port-{str(uuid4())}")
        .with_exposed_ports(8080)
        .with_volume_mapping("/var/run/docker.sock", "/var/run/docker.sock", "rw")
        .with_kwargs(privileged=c.ryuk_privileged, auto_remove=True, remove=True)
        .with_env("RYUK_RECONNECTION_TIMEOUT", "200s")
    )
    if use_network:
        _container.with_kwargs(network=use_network)

    return _container.start()


def test_ryuk(_container: DockerContainer, ip: str, use_exposed_port: bool) -> bool:
    container_port = 8080

    wait_for_logs(_container, r".* Started!")

    if use_exposed_port:
        container_port = int(_container.get_exposed_port(container_port))

    print(f"▶️ Start Ryuk, connect to {ip}:{container_port}")
    for i in range(3):
        _socket: None | socket.socket = socket.socket()
        _socket.settimeout(2)
        start = time()
        try:
            _socket.connect((ip, container_port))
            return True
        except (ConnectionRefusedError, OSError):
            if _socket is not None:
                with contextlib.suppress(Exception):
                    _socket.close()
                _socket = None
        elapsed = time() - start
        if elapsed < 2:
            sleep(2 - elapsed)

    return False


@cache
def verbose_host(name: str) -> str:
    try:
        return f"{name} ({socket.gethostbyname(name)})"
    except OSError:
        return name


def try_extract_host(url: str) -> str:
    host: str | None = None
    with contextlib.suppress(Exception):
        host = urllib.parse.urlparse(url).hostname
    if host:
        return f" ({verbose_host(host)})"
    return ""


def test_ryuk_connections(ryuk: DockerContainer, network: str) -> None:
    docker_client = ryuk.get_docker_client()
    gateway_ip = docker_client.gateway_ip(ryuk._container.id)
    bridge_ip = docker_client.bridge_ip(ryuk._container.id)
    container_host_ip = ryuk.get_container_host_ip()
    docker_host = docker_client.host()
    client_base_url = docker_client.client.api.base_url

    ryuk_ips: dict[str, list[str]] = {}
    ryuk_ips.setdefault(container_host_ip, []).append("container_host_ip")
    ryuk_ips.setdefault(gateway_ip, []).append("gateway_ip")
    ryuk_ips.setdefault(bridge_ip, []).append("brigde_ip")
    ryuk_ips.setdefault(docker_host, []).append("docker_host")
    ryuk_ips.setdefault(default_gateway_ip() or "?", []).append("default_gateway_ip")

    oks = []
    for (addr, labels), use_port in sorted(product(ryuk_ips.items(), [True, False])):
        if addr == "?":
            continue
        with time_echo() as echo:
            l = ", ".join(labels)
            if test_ryuk(ryuk, addr, use_port):
                echo(f"  → ✅ ({addr, use_port}) ryuk ok ({l})\n")
                oks.append(
                    f" ✅  '{addr}' ({'mapped' if use_port else 'original'} port) → {l}"
                )
            else:
                echo(f"   🔥 ({addr, use_port}) ryuk failed ({l})\n")

    output(f"### Run {network}")
    for addr, labels in sorted(ryuk_ips.items()):
        output(f" - {', '.join(labels)}: '{verbose_host(addr)}'")

    find_host_network = ryuk.get_docker_client().find_host_network()
    output(f"{find_host_network=}")
    output(
        f"Docker Client points to {client_base_url=}{try_extract_host(client_base_url)}"
    )

    try:
        network_name = ryuk.get_docker_client().network_name(ryuk._container.id)
    except Exception as e:
        output(f"Could not determine network_name: {type(e)}: {e}")
    else:
        output(f"{network_name=}")

    output(f"Successfully connections for {network or 'default network'}:")
    if oks:
        output("\n".join(oks))
    else:
        output("☠️ none")
    ryuk.stop(force=True)


def run() -> None:
    # first check ryuk without
    output(f"{inside_container()=}")
    output(f"{os_name()=}")
    network = ""

    ryuk = create_ryuk("")
    test_ryuk_connections(ryuk, network)

    docker_client = ryuk.get_docker_client()

    # okay we are within a container try to guess the network
    # we need to use to properly work
    in_container_id = get_running_in_container_id()
    if in_container_id:
        output(f"Found {in_container_id=}")
        with contextlib.suppress(Exception):
            network = docker_client.network_name(in_container_id)

    if network:
        ryuk = create_ryuk(network)
        test_ryuk_connections(ryuk, network)
    else:
        output(f"Could not determine network running in")


if __name__ == "__main__":
    run()