Latency test against all Oracle Cloud regions

oracle_latency_test.py

#!/usr/bin/env python3

import concurrent.futures
import socket
import timeit
import argparse
from typing import Tuple, Optional

REGIONS = [  # https://docs.oracle.com/en-us/iaas/Content/General/Concepts/regions.htm
    {"region": "Australia East", "area": "Sydney", "code": "ap-sydney-1"},
    {"region": "Australia Southeast", "area": "Melbourne", "code": "ap-melbourne-1"},
    {"region": "Brazil East", "area": "Sao Paulo", "code": "sa-saopaulo-1"},
    {"region": "Brazil Southeast", "area": "Vinhedo", "code": "sa-vinhedo-1"},
    {"region": "Canada Southeast", "area": "Montreal", "code": "ca-montreal-1"},
    {"region": "Canada Southeast", "area": "Toronto", "code": "ca-toronto-1"},
    {"region": "Chile", "area": "Santiago", "code": "sa-santiago-1"},
    {"region": "France Central", "area": "Paris", "code": "eu-paris-1"},
    {"region": "France South", "area": "Marseille", "code": "eu-marseille-1"},
    {"region": "Germany Central", "area": "Frankfurt", "code": "eu-frankfurt-1"},
    {"region": "India South", "area": "Hyderabad", "code": "ap-hyderabad-1"},
    {"region": "India West", "area": "Mumbai", "code": "ap-mumbai-1"},
    {"region": "Israel Central", "area": "Jerusalem", "code": "il-jerusalem-1"},
    {"region": "Italy Northwest", "area": "Milan", "code": "eu-milan-1"},
    {"region": "Japan Central", "area": "Osaka", "code": "ap-osaka-1"},
    {"region": "Japan East", "area": "Tokyo", "code": "ap-tokyo-1"},
    {"region": "Mexico Central", "area": "Queretaro", "code": "mx-queretaro-1"},
    {"region": "Mexico Northeast", "area": "Monterrey", "code": "mx-monterrey-1"},
    {"region": "Netherlands Northwest", "area": "Amsterdam", "code": "eu-amsterdam-1"},
    {"region": "Saudi Arabia West", "area": "Jeddah", "code": "me-jeddah-1"},
    # {"region": "Serbia Central", "area": "Jovanovac", "code": "eu-jovanovac-1"},
    {"region": "Singapore", "area": "Singapore", "code": "ap-singapore-1"},
    {
        "region": "South Africa Central",
        "area": "Johannesburg",
        "code": "af-johannesburg-1",
    },
    {"region": "South Korea Central", "area": "Seoul", "code": "ap-seoul-1"},
    {"region": "South Korea North", "area": "Chuncheon", "code": "ap-chuncheon-1"},
    {"region": "Spain Central", "area": "Madrid", "code": "eu-madrid-1"},
    {"region": "Sweden Central", "area": "Stockholm", "code": "eu-stockholm-1"},
    {"region": "Switzerland North", "area": "Zurich", "code": "eu-zurich-1"},
    {"region": "UAE Central", "area": "Abu Dhabi", "code": "me-abudhabi-1"},
    {"region": "UAE East", "area": "Dubai", "code": "me-dubai-1"},
    {"region": "UK South", "area": "London", "code": "uk-london-1"},
    {"region": "UK West", "area": "Newport", "code": "uk-cardiff-1"},
    {"region": "US East", "area": "Ashburn", "code": "us-ashburn-1"},
    {"region": "US Midwest", "area": "Chicago", "code": "us-chicago-1"},
    {"region": "US West", "area": "Phoenix", "code": "us-phoenix-1"},
    {"region": "US West", "area": "San Jose", "code": "us-sanjose-1"},
]


class Ping:
    def __init__(self, host: str, port: int = 443, timeout: float = 10.0, bind: Optional[str] = None):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.bind = bind

    def _ping_once(self) -> None:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(self.timeout)
        if self.bind:
            sock.bind((self.bind, 0))
        sock.connect((self.host, self.port))
        sock.close()

    def ping(self) -> float:
        try:
            return timeit.timeit(self._ping_once, number=1)
        except socket.timeout:
            raise TimeoutError(f"Connection to {self.host}:{self.port} timed out after {self.timeout}s")
        except socket.error as e:
            raise ConnectionError(f"Failed to connect to {self.host}:{self.port}: {e}")
        except Exception as e:
            raise RuntimeError(f"Unexpected error pinging {self.host}:{self.port}: {e}")


def min_avg_max(results: list[float]) -> Tuple[float, float, float]:
    if not results:
        return float('inf'), float('inf'), float('inf')

    min_val = float("inf")
    max_val = float("-inf")
    total = 0.0

    for res in results:
        if res < min_val:
            min_val = res
        if res > max_val:
            max_val = res
        total += res

    avg_val = total / len(results)
    return min_val, avg_val, max_val


def ping_check(addr: str, count: int = 5, timeout: float = 10.0, bind: Optional[str] = None) -> Tuple[float, float, float]:
    results = []

    with concurrent.futures.ThreadPoolExecutor(max_workers=count) as executor:
        futures = []
        for _ in range(count):
            future = executor.submit(Ping(addr, timeout=timeout, bind=bind).ping)
            futures.append(future)

        for future in concurrent.futures.as_completed(futures):
            try:
                res = future.result()
                results.append(res * 1000)  # Convert to milliseconds
            except (TimeoutError, ConnectionError, RuntimeError) as e:
                print(f"  Warning: {e}")
                continue  # Skip failed attempts

    if not results:
        print(f"  Error: All {count} attempts to {addr} failed")
        return float('inf'), float('inf'), float('inf')

    return min_avg_max(results)


class PrettyTable:
    def __init__(self, *args):
        self.table = []
        self.header = args

    def add_row(self, *args):
        self.table.append(args)

    def get_string(self, sortby=None, reversesort=False):
        if sortby and sortby in self.header:
            self.table.sort(
                key=lambda x: x[self.header.index(sortby)], reverse=reversesort
            )

        max_len = [0] * len(self.header)
        for row in self.table:
            for i, col in enumerate(row):
                if len(str(col)) > max_len[i]:
                    max_len[i] = len(str(col))

        lpadding = 1
        rpadding = 2
        border = (
            "+"
            + "+".join(
                "-" * (max_len[i] + lpadding + rpadding)
                for i in range(len(self.header))
            )
            + "+\n"
        )
        table = border
        for i, col in enumerate(self.header):
            table += "|" + lpadding * " " + str(col).ljust(max_len[i] + rpadding)
        table += "|\n"
        table += border
        for row in self.table:
            for i, col in enumerate(row):
                table += "|" + lpadding * " " + str(col).ljust(max_len[i] + rpadding)
            table += "|\n"
        table += border
        return table.rstrip("\n")


def main(count: int, timeout: float, bind: Optional[str]):
    print(f"Testing Oracle Cloud regions with {count} pings per region, timeout={timeout}s...")

    x = PrettyTable("Region", "Area", "Host", "Minimum", "Average", "Maximum")

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = []
        for ls in REGIONS:
            region = ls.get("region")
            area = ls.get("area")
            code = ls.get("code")
            addr = f"objectstorage.{code}.oraclecloud.com"

            print(f"Testing {region} ({area})...")
            future = executor.submit(
                ping_check, addr, count=count, timeout=timeout, bind=bind
            )
            futures.append((future, region, area, code))

        for future, region, area, code in futures:
            try:
                rtt_min, rtt_avg, rtt_max = future.result()
                if rtt_min == float('inf'):
                    # All attempts failed
                    x.add_row(region, area, code, "FAILED", "FAILED", "FAILED")
                else:
                    x.add_row(
                        region,
                        area,
                        code,
                        f"{rtt_min:.2f}",
                        f"{rtt_avg:.2f}",
                        f"{rtt_max:.2f}",
                    )
            except Exception as e:
                print(f"Error processing {region}: {e}")
                x.add_row(region, area, code, "ERROR", "ERROR", "ERROR")

    print("\nResults (sorted by Minimum latency):")
    print(x.get_string(sortby="Minimum", reversesort=False))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Test latency to Oracle Cloud regions",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    parser.add_argument(
        "-c", "--count", type=int, default=3, help="Number of pings to send per region"
    )
    parser.add_argument(
        "-t", "--timeout", type=float, default=15.0, help="Timeout in seconds"
    )
    parser.add_argument(
        "-b", "--bind", type=str, default=None, help="Bind to a specific IP address"
    )
    parser.add_argument(
        "-w", "--workers", type=int, default=10, help="Maximum concurrent region tests"
    )

    args = parser.parse_args()

    # Test a single region first to verify connectivity
    print("Testing connectivity to Oracle Cloud...")
    test_addr = "objectstorage.us-ashburn-1.oraclecloud.com"
    try:
        test_result = ping_check(test_addr, count=1, timeout=args.timeout)
        if test_result[0] == float('inf'):
            print(f"Warning: Could not connect to test region {test_addr}")
            print("You may have network connectivity issues or firewall restrictions.")
            response = input("Continue anyway? (y/N): ")
            if response.lower() != 'y':
                exit(1)
    except Exception as e:
        print(f"Error during connectivity test: {e}")
        exit(1)

    main(args.count, args.timeout, args.bind)

thank you so much for this! great script

Edited to save space

@fragtion Thanks, I updated the gist to your version for visibility.