forked from cerc-io/stack-orchestrator
fix: ashburn relay playbooks and document DZ tunnel ACL root cause
Playbook fixes from testing: - ashburn-relay-biscayne: insert DNAT rules at position 1 before Docker's ADDRTYPE LOCAL rule (was being swallowed at position 3+) - ashburn-relay-mia-sw01: add inbound route for 137.239.194.65 via egress-vrf vrf1 (nexthop only, no interface — EOS silently drops cross-VRF routes that specify a tunnel interface) - ashburn-relay-was-sw01: replace PBR with static route, remove Loopback101 Bug doc (bug-ashburn-tunnel-port-filtering.md): root cause is the DoubleZero agent on mia-sw01 overwrites SEC-USER-500-IN ACL, dropping outbound gossip with src 137.239.194.65. The DZ agent controls Tunnel500's lifecycle. Fix requires a separate GRE tunnel using mia-sw01's free LAN IP (209.42.167.137) to bypass DZ infrastructure. Also adds all repo docs, scripts, inventory, and remaining playbooks. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
A. F. Dudley
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
6841d5e3c3
commit
0b52fc99d7
Executable
+98
@@ -0,0 +1,98 @@
|
||||
#!/bin/bash
|
||||
# Check shred completeness at the tip of the blockstore.
|
||||
#
|
||||
# Samples the most recent N slots and reports how many are full.
|
||||
# Use this to determine when enough complete blocks have accumulated
|
||||
# to safely download a new snapshot that lands within the complete range.
|
||||
#
|
||||
# Usage: kubectl exec ... -- bash -c "$(cat check-shred-completeness.sh)"
|
||||
# Or: ssh biscayne ... 'KUBECONFIG=... kubectl exec ... -- agave-ledger-tool ...'
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
KUBECONFIG="${KUBECONFIG:-/home/rix/.kube/config}"
|
||||
NS="laconic-laconic-70ce4c4b47e23b85"
|
||||
DEPLOY="laconic-70ce4c4b47e23b85-deployment"
|
||||
SAMPLE_SIZE="${1:-200}"
|
||||
|
||||
# Get blockstore bounds
|
||||
BOUNDS=$(kubectl exec -n "$NS" deployment/"$DEPLOY" -c agave-validator -- \
|
||||
agave-ledger-tool -l /data/ledger blockstore bounds 2>&1 | grep "^Ledger")
|
||||
|
||||
HIGHEST=$(echo "$BOUNDS" | grep -oP 'to \K[0-9]+')
|
||||
START=$((HIGHEST - SAMPLE_SIZE))
|
||||
|
||||
echo "Blockstore highest slot: $HIGHEST"
|
||||
echo "Sampling slots $START to $HIGHEST ($SAMPLE_SIZE slots)"
|
||||
echo ""
|
||||
|
||||
# Get slot metadata
|
||||
OUTPUT=$(kubectl exec -n "$NS" deployment/"$DEPLOY" -c agave-validator -- \
|
||||
agave-ledger-tool -l /data/ledger blockstore print \
|
||||
--starting-slot "$START" --ending-slot "$HIGHEST" 2>&1 \
|
||||
| grep -E "^Slot|is_full")
|
||||
|
||||
TOTAL=$(echo "$OUTPUT" | grep -c "^Slot" || true)
|
||||
FULL=$(echo "$OUTPUT" | grep -c "is_full: true" || true)
|
||||
INCOMPLETE=$(echo "$OUTPUT" | grep -c "is_full: false" || true)
|
||||
|
||||
echo "Total slots with data: $TOTAL / $SAMPLE_SIZE"
|
||||
echo "Complete (is_full: true): $FULL"
|
||||
echo "Incomplete (is_full: false): $INCOMPLETE"
|
||||
|
||||
if [ "$TOTAL" -gt 0 ]; then
|
||||
PCT=$((FULL * 100 / TOTAL))
|
||||
echo "Completeness: ${PCT}%"
|
||||
else
|
||||
echo "Completeness: N/A (no data)"
|
||||
fi
|
||||
|
||||
echo ""
|
||||
|
||||
# Find the first full slot counting backward from the tip
|
||||
# This tells us where the contiguous complete run starts
|
||||
echo "--- Contiguous complete run from tip ---"
|
||||
|
||||
# Get just the slot numbers and is_full in reverse order
|
||||
REVERSED=$(echo "$OUTPUT" | paste - - | awk '{
|
||||
slot = $2;
|
||||
full = ($NF == "true") ? 1 : 0;
|
||||
print slot, full
|
||||
}' | sort -rn)
|
||||
|
||||
CONTIGUOUS=0
|
||||
FIRST_FULL=""
|
||||
while IFS=' ' read -r slot full; do
|
||||
if [ "$full" -eq 1 ]; then
|
||||
CONTIGUOUS=$((CONTIGUOUS + 1))
|
||||
FIRST_FULL="$slot"
|
||||
else
|
||||
break
|
||||
fi
|
||||
done <<< "$REVERSED"
|
||||
|
||||
if [ -n "$FIRST_FULL" ]; then
|
||||
echo "Contiguous complete slots from tip: $CONTIGUOUS"
|
||||
echo "Run starts at slot: $FIRST_FULL"
|
||||
echo "Run ends at slot: $HIGHEST"
|
||||
echo ""
|
||||
echo "A snapshot with slot >= $FIRST_FULL would replay from local blockstore."
|
||||
|
||||
# Check against mainnet
|
||||
MAINNET_SLOT=$(curl -s -X POST -H "Content-Type: application/json" \
|
||||
-d '{"jsonrpc":"2.0","id":1,"method":"getSlot","params":[{"commitment":"finalized"}]}' \
|
||||
https://api.mainnet-beta.solana.com | grep -oP '"result":\K[0-9]+')
|
||||
|
||||
GAP=$((MAINNET_SLOT - HIGHEST))
|
||||
echo "Mainnet tip: $MAINNET_SLOT (blockstore is $GAP slots behind tip)"
|
||||
|
||||
if [ "$CONTIGUOUS" -gt 100 ]; then
|
||||
echo ""
|
||||
echo ">>> READY: $CONTIGUOUS contiguous complete slots. Safe to download a snapshot."
|
||||
else
|
||||
echo ""
|
||||
echo ">>> NOT READY: Only $CONTIGUOUS contiguous complete slots. Wait for more."
|
||||
fi
|
||||
else
|
||||
echo "No contiguous complete run from tip found."
|
||||
fi
|
||||
Executable
+38
@@ -0,0 +1,38 @@
|
||||
#!/bin/bash
|
||||
# Run a command in a tmux pane and capture its output.
|
||||
# User sees it streaming in the pane; caller gets stdout back.
|
||||
#
|
||||
# Usage: pane-exec.sh <pane-id> <command...>
|
||||
# Example: pane-exec.sh %6565 ansible-playbook -i inventory/switches.yml playbooks/foo.yml
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
PANE="$1"
|
||||
shift
|
||||
CMD="$*"
|
||||
|
||||
TMPFILE=$(mktemp /tmp/pane-output.XXXXXX)
|
||||
MARKER="__PANE_EXEC_DONE_${RANDOM}_$$__"
|
||||
|
||||
cleanup() {
|
||||
tmux pipe-pane -t "$PANE" 2>/dev/null || true
|
||||
rm -f "$TMPFILE"
|
||||
}
|
||||
trap cleanup EXIT
|
||||
|
||||
# Start capturing pane output
|
||||
tmux pipe-pane -o -t "$PANE" "cat >> $TMPFILE"
|
||||
|
||||
# Send the command, then echo a marker so we know when it's done
|
||||
tmux send-keys -t "$PANE" "$CMD; echo $MARKER" Enter
|
||||
|
||||
# Wait for the marker
|
||||
while ! grep -q "$MARKER" "$TMPFILE" 2>/dev/null; do
|
||||
sleep 0.5
|
||||
done
|
||||
|
||||
# Stop capturing
|
||||
tmux pipe-pane -t "$PANE"
|
||||
|
||||
# Strip ANSI escape codes, remove the marker line, output the rest
|
||||
sed 's/\x1b\[[0-9;]*[a-zA-Z]//g; s/\x1b\[[?][0-9]*[a-zA-Z]//g' "$TMPFILE" | grep -v "$MARKER"
|
||||
@@ -0,0 +1,151 @@
|
||||
import { chromium } from 'playwright';
|
||||
import { writeFileSync, mkdirSync } from 'fs';
|
||||
import { join } from 'path';
|
||||
|
||||
const OUT_DIR = join(import.meta.dirname, '..', 'docs', 'arista-scraped');
|
||||
mkdirSync(OUT_DIR, { recursive: true });
|
||||
|
||||
const pages = [
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-static-inter-vrf-route', file: 'static-inter-vrf-route.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-inter-vrf-local-route-leaking', file: 'inter-vrf-local-route-leaking.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-policy-based-routing', file: 'policy-based-routing.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-traffic-management', file: 'traffic-management.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-policy-based-routing-pbr', file: 'pbr.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-configuring-vrf-instances', file: 'configuring-vrf.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-gre-tunnels', file: 'gre-tunnels.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-access-control-lists', file: 'access-control-lists.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-static-routes', file: 'static-routes.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-configuration-sessions', file: 'configuration-sessions.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos/eos-checkpoint-and-rollback', file: 'checkpoint-rollback.md' },
|
||||
{ url: 'https://www.arista.com/en/um-eos', file: '_index.md' },
|
||||
];
|
||||
|
||||
async function scrapePage(page, url, filename) {
|
||||
console.log(`Scraping: ${url}`);
|
||||
try {
|
||||
const resp = await page.goto(url, { waitUntil: 'domcontentloaded', timeout: 30000 });
|
||||
console.log(` Status: ${resp.status()}`);
|
||||
|
||||
// Wait for JS to render
|
||||
await page.waitForTimeout(8000);
|
||||
|
||||
// Check for CAPTCHA
|
||||
const bodyText = await page.evaluate(() => document.body.innerText.substring(0, 200));
|
||||
if (bodyText.includes('CAPTCHA') || bodyText.includes("couldn't load")) {
|
||||
console.log(` BLOCKED by CAPTCHA/anti-bot on ${filename}`);
|
||||
writeFileSync(join(OUT_DIR, filename), `# BLOCKED BY CAPTCHA\n\nURL: ${url}\nThe Arista docs site requires CAPTCHA verification for headless browsers.\n`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Extract content
|
||||
const content = await page.evaluate(() => {
|
||||
const selectors = [
|
||||
'#content', '.article-content', '.content-area', '#main-content',
|
||||
'article', '.item-page', '#sp-component', '.com-content-article',
|
||||
'main', '#sp-main-body',
|
||||
];
|
||||
|
||||
let el = null;
|
||||
for (const sel of selectors) {
|
||||
el = document.querySelector(sel);
|
||||
if (el && el.textContent.trim().length > 100) break;
|
||||
}
|
||||
if (!el) el = document.body;
|
||||
|
||||
function nodeToMd(node) {
|
||||
if (node.nodeType === Node.TEXT_NODE) return node.textContent;
|
||||
if (node.nodeType !== Node.ELEMENT_NODE) return '';
|
||||
const tag = node.tagName.toLowerCase();
|
||||
if (['nav', 'footer', 'script', 'style', 'noscript', 'iframe'].includes(tag)) return '';
|
||||
if (node.classList && (node.classList.contains('nav') || node.classList.contains('sidebar') ||
|
||||
node.classList.contains('menu') || node.classList.contains('footer') ||
|
||||
node.classList.contains('header'))) return '';
|
||||
let children = Array.from(node.childNodes).map(c => nodeToMd(c)).join('');
|
||||
switch (tag) {
|
||||
case 'h1': return `\n# ${children.trim()}\n\n`;
|
||||
case 'h2': return `\n## ${children.trim()}\n\n`;
|
||||
case 'h3': return `\n### ${children.trim()}\n\n`;
|
||||
case 'h4': return `\n#### ${children.trim()}\n\n`;
|
||||
case 'p': return `\n${children.trim()}\n\n`;
|
||||
case 'br': return '\n';
|
||||
case 'li': return `- ${children.trim()}\n`;
|
||||
case 'ul': case 'ol': return `\n${children}\n`;
|
||||
case 'pre': return `\n\`\`\`\n${children.trim()}\n\`\`\`\n\n`;
|
||||
case 'code': return `\`${children.trim()}\``;
|
||||
case 'strong': case 'b': return `**${children.trim()}**`;
|
||||
case 'em': case 'i': return `*${children.trim()}*`;
|
||||
case 'table': return `\n${children}\n`;
|
||||
case 'tr': return `${children}|\n`;
|
||||
case 'th': case 'td': return `| ${children.trim()} `;
|
||||
case 'a': {
|
||||
const href = node.getAttribute('href');
|
||||
if (href && !href.startsWith('#') && !href.startsWith('javascript'))
|
||||
return `[${children.trim()}](${href})`;
|
||||
return children;
|
||||
}
|
||||
default: return children;
|
||||
}
|
||||
}
|
||||
return nodeToMd(el);
|
||||
});
|
||||
|
||||
const cleaned = content.replace(/\n{4,}/g, '\n\n\n').replace(/[ \t]+$/gm, '').trim();
|
||||
const header = `<!-- Source: ${url} -->\n<!-- Scraped: ${new Date().toISOString()} -->\n\n`;
|
||||
writeFileSync(join(OUT_DIR, filename), header + cleaned + '\n');
|
||||
console.log(` Saved ${filename} (${cleaned.length} chars)`);
|
||||
return true;
|
||||
} catch (e) {
|
||||
console.error(` FAILED: ${e.message}`);
|
||||
writeFileSync(join(OUT_DIR, filename), `# FAILED TO LOAD\n\nURL: ${url}\nError: ${e.message}\n`);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function main() {
|
||||
// Launch with stealth-like settings
|
||||
const browser = await chromium.launch({
|
||||
headless: false, // Use headed mode via Xvfb if available, else new headless
|
||||
args: [
|
||||
'--headless=new', // New headless mode (less detectable)
|
||||
'--disable-blink-features=AutomationControlled',
|
||||
'--no-sandbox',
|
||||
],
|
||||
});
|
||||
|
||||
const context = await browser.newContext({
|
||||
userAgent: 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
|
||||
locale: 'en-US',
|
||||
timezoneId: 'America/New_York',
|
||||
viewport: { width: 1920, height: 1080 },
|
||||
});
|
||||
|
||||
// Remove webdriver property
|
||||
await context.addInitScript(() => {
|
||||
Object.defineProperty(navigator, 'webdriver', { get: () => false });
|
||||
// Override permissions
|
||||
const originalQuery = window.navigator.permissions.query;
|
||||
window.navigator.permissions.query = (parameters) =>
|
||||
parameters.name === 'notifications'
|
||||
? Promise.resolve({ state: Notification.permission })
|
||||
: originalQuery(parameters);
|
||||
});
|
||||
|
||||
const page = await context.newPage();
|
||||
|
||||
let anySuccess = false;
|
||||
for (const { url, file } of pages) {
|
||||
const ok = await scrapePage(page, url, file);
|
||||
if (ok) anySuccess = true;
|
||||
// Add delay between requests
|
||||
await page.waitForTimeout(2000);
|
||||
}
|
||||
|
||||
if (!anySuccess) {
|
||||
console.log('\nAll pages blocked by CAPTCHA. Arista docs require human verification.');
|
||||
}
|
||||
|
||||
await browser.close();
|
||||
console.log('\nDone!');
|
||||
}
|
||||
|
||||
main().catch(e => { console.error(e); process.exit(1); });
|
||||
@@ -0,0 +1,34 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Strip IP+UDP headers from mirrored packets and forward raw UDP payload."""
|
||||
import socket
|
||||
import sys
|
||||
|
||||
LISTEN_PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 9100
|
||||
FORWARD_HOST = sys.argv[2] if len(sys.argv) > 2 else "127.0.0.1"
|
||||
FORWARD_PORT = int(sys.argv[3]) if len(sys.argv) > 3 else 9000
|
||||
|
||||
sock_in = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock_in.bind(("0.0.0.0", LISTEN_PORT))
|
||||
|
||||
sock_out = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
|
||||
count = 0
|
||||
while True:
|
||||
data, addr = sock_in.recvfrom(65535)
|
||||
if len(data) < 28:
|
||||
continue
|
||||
# IP header: first nibble is version (4), second nibble is IHL (words)
|
||||
if (data[0] >> 4) != 4:
|
||||
continue
|
||||
ihl = (data[0] & 0x0F) * 4
|
||||
# Protocol should be UDP (17)
|
||||
if data[9] != 17:
|
||||
continue
|
||||
# Payload starts after IP header + 8-byte UDP header
|
||||
offset = ihl + 8
|
||||
payload = data[offset:]
|
||||
if payload:
|
||||
sock_out.sendto(payload, (FORWARD_HOST, FORWARD_PORT))
|
||||
count += 1
|
||||
if count % 10000 == 0:
|
||||
print(f"Forwarded {count} shreds", flush=True)
|
||||
Executable
+546
@@ -0,0 +1,546 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Download Solana snapshots using aria2c for parallel multi-connection downloads.
|
||||
|
||||
Discovers snapshot sources by querying getClusterNodes for all RPCs in the
|
||||
cluster, probing each for available snapshots, benchmarking download speed,
|
||||
and downloading from the fastest source using aria2c (16 connections by default).
|
||||
|
||||
Based on the discovery approach from etcusr/solana-snapshot-finder but replaces
|
||||
the single-connection wget download with aria2c parallel chunked downloads.
|
||||
|
||||
Usage:
|
||||
# Download to /srv/solana/snapshots (mainnet, 16 connections)
|
||||
./snapshot-download.py -o /srv/solana/snapshots
|
||||
|
||||
# Dry run — find best source, print URL
|
||||
./snapshot-download.py --dry-run
|
||||
|
||||
# Custom RPC for cluster node discovery + 32 connections
|
||||
./snapshot-download.py -r https://api.mainnet-beta.solana.com -n 32
|
||||
|
||||
# Testnet
|
||||
./snapshot-download.py -c testnet -o /data/snapshots
|
||||
|
||||
Requirements:
|
||||
- aria2c (apt install aria2)
|
||||
- python3 >= 3.10 (stdlib only, no pip dependencies)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import concurrent.futures
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from dataclasses import dataclass, field
|
||||
from http.client import HTTPResponse
|
||||
from pathlib import Path
|
||||
from typing import NoReturn
|
||||
from urllib.request import Request
|
||||
|
||||
log: logging.Logger = logging.getLogger("snapshot-download")
|
||||
|
||||
CLUSTER_RPC: dict[str, str] = {
|
||||
"mainnet-beta": "https://api.mainnet-beta.solana.com",
|
||||
"testnet": "https://api.testnet.solana.com",
|
||||
"devnet": "https://api.devnet.solana.com",
|
||||
}
|
||||
|
||||
# Snapshot filenames:
|
||||
# snapshot-<slot>-<hash>.tar.zst
|
||||
# incremental-snapshot-<base_slot>-<slot>-<hash>.tar.zst
|
||||
FULL_SNAP_RE: re.Pattern[str] = re.compile(
|
||||
r"^snapshot-(\d+)-([A-Za-z0-9]+)\.tar\.(zst|bz2)$"
|
||||
)
|
||||
INCR_SNAP_RE: re.Pattern[str] = re.compile(
|
||||
r"^incremental-snapshot-(\d+)-(\d+)-([A-Za-z0-9]+)\.tar\.(zst|bz2)$"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class SnapshotSource:
|
||||
"""A snapshot file available from a specific RPC node."""
|
||||
|
||||
rpc_address: str
|
||||
# Full redirect paths as returned by the server (e.g. /snapshot-123-hash.tar.zst)
|
||||
file_paths: list[str] = field(default_factory=list)
|
||||
slots_diff: int = 0
|
||||
latency_ms: float = 0.0
|
||||
download_speed: float = 0.0 # bytes/sec
|
||||
|
||||
|
||||
# -- JSON-RPC helpers ----------------------------------------------------------
|
||||
|
||||
|
||||
class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
"""Handler that captures redirect Location instead of following it."""
|
||||
|
||||
def redirect_request(
|
||||
self,
|
||||
req: Request,
|
||||
fp: HTTPResponse,
|
||||
code: int,
|
||||
msg: str,
|
||||
headers: dict[str, str], # type: ignore[override]
|
||||
newurl: str,
|
||||
) -> None:
|
||||
return None
|
||||
|
||||
|
||||
def rpc_post(url: str, method: str, params: list[object] | None = None,
|
||||
timeout: int = 25) -> object | None:
|
||||
"""JSON-RPC POST. Returns parsed 'result' field or None on error."""
|
||||
payload: bytes = json.dumps({
|
||||
"jsonrpc": "2.0", "id": 1,
|
||||
"method": method, "params": params or [],
|
||||
}).encode()
|
||||
req = Request(url, data=payload,
|
||||
headers={"Content-Type": "application/json"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
data: dict[str, object] = json.loads(resp.read())
|
||||
return data.get("result")
|
||||
except (urllib.error.URLError, json.JSONDecodeError, OSError, TimeoutError) as e:
|
||||
log.debug("rpc_post %s %s failed: %s", url, method, e)
|
||||
return None
|
||||
|
||||
|
||||
def head_no_follow(url: str, timeout: float = 3) -> tuple[str | None, float]:
|
||||
"""HEAD request without following redirects.
|
||||
|
||||
Returns (Location header value, latency_sec) if the server returned a
|
||||
3xx redirect. Returns (None, 0.0) on any error or non-redirect response.
|
||||
"""
|
||||
opener: urllib.request.OpenerDirector = urllib.request.build_opener(_NoRedirectHandler)
|
||||
req = Request(url, method="HEAD")
|
||||
try:
|
||||
start: float = time.monotonic()
|
||||
resp: HTTPResponse = opener.open(req, timeout=timeout) # type: ignore[assignment]
|
||||
latency: float = time.monotonic() - start
|
||||
# Non-redirect (2xx) — server didn't redirect, not useful for discovery
|
||||
location: str | None = resp.headers.get("Location")
|
||||
resp.close()
|
||||
return location, latency
|
||||
except urllib.error.HTTPError as e:
|
||||
# 3xx redirects raise HTTPError with the redirect info
|
||||
latency = time.monotonic() - start # type: ignore[possibly-undefined]
|
||||
location = e.headers.get("Location")
|
||||
if location and 300 <= e.code < 400:
|
||||
return location, latency
|
||||
return None, 0.0
|
||||
except (urllib.error.URLError, OSError, TimeoutError):
|
||||
return None, 0.0
|
||||
|
||||
|
||||
# -- Discovery -----------------------------------------------------------------
|
||||
|
||||
|
||||
def get_current_slot(rpc_url: str) -> int | None:
|
||||
"""Get current slot from RPC."""
|
||||
result: object | None = rpc_post(rpc_url, "getSlot")
|
||||
if isinstance(result, int):
|
||||
return result
|
||||
return None
|
||||
|
||||
|
||||
def get_cluster_rpc_nodes(rpc_url: str, version_filter: str | None = None) -> list[str]:
|
||||
"""Get all RPC node addresses from getClusterNodes."""
|
||||
result: object | None = rpc_post(rpc_url, "getClusterNodes")
|
||||
if not isinstance(result, list):
|
||||
return []
|
||||
|
||||
rpc_addrs: list[str] = []
|
||||
for node in result:
|
||||
if not isinstance(node, dict):
|
||||
continue
|
||||
if version_filter is not None:
|
||||
node_version: str | None = node.get("version")
|
||||
if node_version and not node_version.startswith(version_filter):
|
||||
continue
|
||||
rpc: str | None = node.get("rpc")
|
||||
if rpc:
|
||||
rpc_addrs.append(rpc)
|
||||
return list(set(rpc_addrs))
|
||||
|
||||
|
||||
def _parse_snapshot_filename(location: str) -> tuple[str, str | None]:
|
||||
"""Extract filename and full redirect path from Location header.
|
||||
|
||||
Returns (filename, full_path). full_path includes any path prefix
|
||||
the server returned (e.g. '/snapshots/snapshot-123-hash.tar.zst').
|
||||
"""
|
||||
# Location may be absolute URL or relative path
|
||||
if location.startswith("http://") or location.startswith("https://"):
|
||||
# Absolute URL — extract path
|
||||
from urllib.parse import urlparse
|
||||
path: str = urlparse(location).path
|
||||
else:
|
||||
path = location
|
||||
|
||||
filename: str = path.rsplit("/", 1)[-1]
|
||||
return filename, path
|
||||
|
||||
|
||||
def probe_rpc_snapshot(
|
||||
rpc_address: str,
|
||||
current_slot: int,
|
||||
max_age_slots: int,
|
||||
max_latency_ms: float,
|
||||
) -> SnapshotSource | None:
|
||||
"""Probe a single RPC node for available snapshots.
|
||||
|
||||
Probes for full snapshot first (required), then incremental. Records all
|
||||
available files. Which files to actually download is decided at download
|
||||
time based on what already exists locally — not here.
|
||||
|
||||
Based on the discovery approach from etcusr/solana-snapshot-finder.
|
||||
"""
|
||||
full_url: str = f"http://{rpc_address}/snapshot.tar.bz2"
|
||||
|
||||
# Full snapshot is required — every source must have one
|
||||
full_location, full_latency = head_no_follow(full_url, timeout=2)
|
||||
if not full_location:
|
||||
return None
|
||||
|
||||
latency_ms: float = full_latency * 1000
|
||||
if latency_ms > max_latency_ms:
|
||||
return None
|
||||
|
||||
full_filename, full_path = _parse_snapshot_filename(full_location)
|
||||
fm: re.Match[str] | None = FULL_SNAP_RE.match(full_filename)
|
||||
if not fm:
|
||||
return None
|
||||
|
||||
full_snap_slot: int = int(fm.group(1))
|
||||
slots_diff: int = current_slot - full_snap_slot
|
||||
|
||||
if slots_diff > max_age_slots or slots_diff < -100:
|
||||
return None
|
||||
|
||||
file_paths: list[str] = [full_path]
|
||||
|
||||
# Also check for incremental snapshot
|
||||
inc_url: str = f"http://{rpc_address}/incremental-snapshot.tar.bz2"
|
||||
inc_location, _ = head_no_follow(inc_url, timeout=2)
|
||||
if inc_location:
|
||||
inc_filename, inc_path = _parse_snapshot_filename(inc_location)
|
||||
m: re.Match[str] | None = INCR_SNAP_RE.match(inc_filename)
|
||||
if m:
|
||||
inc_base_slot: int = int(m.group(1))
|
||||
# Incremental must be based on this source's full snapshot
|
||||
if inc_base_slot == full_snap_slot:
|
||||
file_paths.append(inc_path)
|
||||
|
||||
return SnapshotSource(
|
||||
rpc_address=rpc_address,
|
||||
file_paths=file_paths,
|
||||
slots_diff=slots_diff,
|
||||
latency_ms=latency_ms,
|
||||
)
|
||||
|
||||
|
||||
def discover_sources(
|
||||
rpc_url: str,
|
||||
current_slot: int,
|
||||
max_age_slots: int,
|
||||
max_latency_ms: float,
|
||||
threads: int,
|
||||
version_filter: str | None,
|
||||
) -> list[SnapshotSource]:
|
||||
"""Discover all snapshot sources from the cluster."""
|
||||
rpc_nodes: list[str] = get_cluster_rpc_nodes(rpc_url, version_filter)
|
||||
if not rpc_nodes:
|
||||
log.error("No RPC nodes found via getClusterNodes")
|
||||
return []
|
||||
|
||||
log.info("Found %d RPC nodes, probing for snapshots...", len(rpc_nodes))
|
||||
|
||||
sources: list[SnapshotSource] = []
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as pool:
|
||||
futures: dict[concurrent.futures.Future[SnapshotSource | None], str] = {
|
||||
pool.submit(
|
||||
probe_rpc_snapshot, addr, current_slot,
|
||||
max_age_slots, max_latency_ms,
|
||||
): addr
|
||||
for addr in rpc_nodes
|
||||
}
|
||||
done: int = 0
|
||||
for future in concurrent.futures.as_completed(futures):
|
||||
done += 1
|
||||
if done % 200 == 0:
|
||||
log.info(" probed %d/%d nodes, %d sources found",
|
||||
done, len(rpc_nodes), len(sources))
|
||||
try:
|
||||
result: SnapshotSource | None = future.result()
|
||||
except (urllib.error.URLError, OSError, TimeoutError) as e:
|
||||
log.debug("Probe failed for %s: %s", futures[future], e)
|
||||
continue
|
||||
if result:
|
||||
sources.append(result)
|
||||
|
||||
log.info("Found %d RPC nodes with suitable snapshots", len(sources))
|
||||
return sources
|
||||
|
||||
|
||||
# -- Speed benchmark -----------------------------------------------------------
|
||||
|
||||
|
||||
def measure_speed(rpc_address: str, measure_time: int = 7) -> float:
|
||||
"""Measure download speed from an RPC node. Returns bytes/sec."""
|
||||
url: str = f"http://{rpc_address}/snapshot.tar.bz2"
|
||||
req = Request(url)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=measure_time + 5) as resp:
|
||||
start: float = time.monotonic()
|
||||
total: int = 0
|
||||
while True:
|
||||
elapsed: float = time.monotonic() - start
|
||||
if elapsed >= measure_time:
|
||||
break
|
||||
chunk: bytes = resp.read(81920)
|
||||
if not chunk:
|
||||
break
|
||||
total += len(chunk)
|
||||
elapsed = time.monotonic() - start
|
||||
if elapsed <= 0:
|
||||
return 0.0
|
||||
return total / elapsed
|
||||
except (urllib.error.URLError, OSError, TimeoutError):
|
||||
return 0.0
|
||||
|
||||
|
||||
# -- Download ------------------------------------------------------------------
|
||||
|
||||
|
||||
def download_aria2c(
|
||||
urls: list[str],
|
||||
output_dir: str,
|
||||
filename: str,
|
||||
connections: int = 16,
|
||||
) -> bool:
|
||||
"""Download a file using aria2c with parallel connections.
|
||||
|
||||
When multiple URLs are provided, aria2c treats them as mirrors of the
|
||||
same file and distributes chunks across all of them.
|
||||
"""
|
||||
num_mirrors: int = len(urls)
|
||||
total_splits: int = max(connections, connections * num_mirrors)
|
||||
cmd: list[str] = [
|
||||
"aria2c",
|
||||
"--file-allocation=none",
|
||||
"--continue=true",
|
||||
f"--max-connection-per-server={connections}",
|
||||
f"--split={total_splits}",
|
||||
"--min-split-size=50M",
|
||||
# aria2c retries individual chunk connections on transient network
|
||||
# errors (TCP reset, timeout). This is transport-level retry analogous
|
||||
# to TCP retransmit, not application-level retry of a failed operation.
|
||||
"--max-tries=5",
|
||||
"--retry-wait=5",
|
||||
"--timeout=60",
|
||||
"--connect-timeout=10",
|
||||
"--summary-interval=10",
|
||||
"--console-log-level=notice",
|
||||
f"--dir={output_dir}",
|
||||
f"--out={filename}",
|
||||
"--auto-file-renaming=false",
|
||||
"--allow-overwrite=true",
|
||||
*urls,
|
||||
]
|
||||
|
||||
log.info("Downloading %s", filename)
|
||||
log.info(" aria2c: %d connections × %d mirrors (%d splits)",
|
||||
connections, num_mirrors, total_splits)
|
||||
|
||||
start: float = time.monotonic()
|
||||
result: subprocess.CompletedProcess[bytes] = subprocess.run(cmd)
|
||||
elapsed: float = time.monotonic() - start
|
||||
|
||||
if result.returncode != 0:
|
||||
log.error("aria2c failed with exit code %d", result.returncode)
|
||||
return False
|
||||
|
||||
filepath: Path = Path(output_dir) / filename
|
||||
if not filepath.exists():
|
||||
log.error("aria2c reported success but %s does not exist", filepath)
|
||||
return False
|
||||
|
||||
size_bytes: int = filepath.stat().st_size
|
||||
size_gb: float = size_bytes / (1024 ** 3)
|
||||
avg_mb: float = size_bytes / elapsed / (1024 ** 2) if elapsed > 0 else 0
|
||||
log.info(" Done: %.1f GB in %.0fs (%.1f MiB/s avg)", size_gb, elapsed, avg_mb)
|
||||
return True
|
||||
|
||||
|
||||
# -- Main ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def main() -> int:
|
||||
p: argparse.ArgumentParser = argparse.ArgumentParser(
|
||||
description="Download Solana snapshots with aria2c parallel downloads",
|
||||
)
|
||||
p.add_argument("-o", "--output", default="/srv/solana/snapshots",
|
||||
help="Snapshot output directory (default: /srv/solana/snapshots)")
|
||||
p.add_argument("-c", "--cluster", default="mainnet-beta",
|
||||
choices=list(CLUSTER_RPC),
|
||||
help="Solana cluster (default: mainnet-beta)")
|
||||
p.add_argument("-r", "--rpc", default=None,
|
||||
help="RPC URL for cluster discovery (default: public RPC)")
|
||||
p.add_argument("-n", "--connections", type=int, default=16,
|
||||
help="aria2c connections per download (default: 16)")
|
||||
p.add_argument("-t", "--threads", type=int, default=500,
|
||||
help="Threads for parallel RPC probing (default: 500)")
|
||||
p.add_argument("--max-snapshot-age", type=int, default=1300,
|
||||
help="Max snapshot age in slots (default: 1300)")
|
||||
p.add_argument("--max-latency", type=float, default=100,
|
||||
help="Max RPC probe latency in ms (default: 100)")
|
||||
p.add_argument("--min-download-speed", type=int, default=20,
|
||||
help="Min download speed in MiB/s (default: 20)")
|
||||
p.add_argument("--measurement-time", type=int, default=7,
|
||||
help="Speed measurement duration in seconds (default: 7)")
|
||||
p.add_argument("--max-speed-checks", type=int, default=15,
|
||||
help="Max nodes to benchmark before giving up (default: 15)")
|
||||
p.add_argument("--version", default=None,
|
||||
help="Filter nodes by version prefix (e.g. '2.2')")
|
||||
p.add_argument("--full-only", action="store_true",
|
||||
help="Download only full snapshot, skip incremental")
|
||||
p.add_argument("--dry-run", action="store_true",
|
||||
help="Find best source and print URL, don't download")
|
||||
p.add_argument("-v", "--verbose", action="store_true")
|
||||
args: argparse.Namespace = p.parse_args()
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if args.verbose else logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(message)s",
|
||||
datefmt="%H:%M:%S",
|
||||
)
|
||||
|
||||
rpc_url: str = args.rpc or CLUSTER_RPC[args.cluster]
|
||||
|
||||
# aria2c is required for actual downloads (not dry-run)
|
||||
if not args.dry_run and not shutil.which("aria2c"):
|
||||
log.error("aria2c not found. Install with: apt install aria2")
|
||||
return 1
|
||||
|
||||
# Get current slot
|
||||
log.info("Cluster: %s | RPC: %s", args.cluster, rpc_url)
|
||||
current_slot: int | None = get_current_slot(rpc_url)
|
||||
if current_slot is None:
|
||||
log.error("Cannot get current slot from %s", rpc_url)
|
||||
return 1
|
||||
log.info("Current slot: %d", current_slot)
|
||||
|
||||
# Discover sources
|
||||
sources: list[SnapshotSource] = discover_sources(
|
||||
rpc_url, current_slot,
|
||||
max_age_slots=args.max_snapshot_age,
|
||||
max_latency_ms=args.max_latency,
|
||||
threads=args.threads,
|
||||
version_filter=args.version,
|
||||
)
|
||||
if not sources:
|
||||
log.error("No snapshot sources found")
|
||||
return 1
|
||||
|
||||
# Sort by latency (lowest first) for speed benchmarking
|
||||
sources.sort(key=lambda s: s.latency_ms)
|
||||
|
||||
# Benchmark top candidates — all speeds in MiB/s (binary, 1 MiB = 1048576 bytes)
|
||||
log.info("Benchmarking download speed on top %d sources...", args.max_speed_checks)
|
||||
fast_sources: list[SnapshotSource] = []
|
||||
checked: int = 0
|
||||
min_speed_bytes: int = args.min_download_speed * 1024 * 1024 # MiB to bytes
|
||||
|
||||
for source in sources:
|
||||
if checked >= args.max_speed_checks:
|
||||
break
|
||||
checked += 1
|
||||
|
||||
speed: float = measure_speed(source.rpc_address, args.measurement_time)
|
||||
source.download_speed = speed
|
||||
speed_mib: float = speed / (1024 ** 2)
|
||||
|
||||
if speed < min_speed_bytes:
|
||||
log.info(" %s: %.1f MiB/s (too slow, need >=%d MiB/s)",
|
||||
source.rpc_address, speed_mib, args.min_download_speed)
|
||||
continue
|
||||
|
||||
log.info(" %s: %.1f MiB/s (latency: %.0fms, age: %d slots)",
|
||||
source.rpc_address, speed_mib,
|
||||
source.latency_ms, source.slots_diff)
|
||||
fast_sources.append(source)
|
||||
|
||||
if not fast_sources:
|
||||
log.error("No source met minimum speed requirement (%d MiB/s)",
|
||||
args.min_download_speed)
|
||||
log.info("Try: --min-download-speed 10")
|
||||
return 1
|
||||
|
||||
# Use the fastest source as primary, collect mirrors for each file
|
||||
best: SnapshotSource = fast_sources[0]
|
||||
file_paths: list[str] = best.file_paths
|
||||
if args.full_only:
|
||||
file_paths = [fp for fp in file_paths
|
||||
if fp.rsplit("/", 1)[-1].startswith("snapshot-")]
|
||||
|
||||
# Build mirror URL lists: for each file, collect URLs from all fast sources
|
||||
# that serve the same filename
|
||||
download_plan: list[tuple[str, list[str]]] = []
|
||||
for fp in file_paths:
|
||||
filename: str = fp.rsplit("/", 1)[-1]
|
||||
mirror_urls: list[str] = [f"http://{best.rpc_address}{fp}"]
|
||||
for other in fast_sources[1:]:
|
||||
for other_fp in other.file_paths:
|
||||
if other_fp.rsplit("/", 1)[-1] == filename:
|
||||
mirror_urls.append(f"http://{other.rpc_address}{other_fp}")
|
||||
break
|
||||
download_plan.append((filename, mirror_urls))
|
||||
|
||||
speed_mib: float = best.download_speed / (1024 ** 2)
|
||||
log.info("Best source: %s (%.1f MiB/s), %d mirrors total",
|
||||
best.rpc_address, speed_mib, len(fast_sources))
|
||||
for filename, mirror_urls in download_plan:
|
||||
log.info(" %s (%d mirrors)", filename, len(mirror_urls))
|
||||
for url in mirror_urls:
|
||||
log.info(" %s", url)
|
||||
|
||||
if args.dry_run:
|
||||
for _, mirror_urls in download_plan:
|
||||
for url in mirror_urls:
|
||||
print(url)
|
||||
return 0
|
||||
|
||||
# Download — skip files that already exist locally
|
||||
os.makedirs(args.output, exist_ok=True)
|
||||
total_start: float = time.monotonic()
|
||||
|
||||
for filename, mirror_urls in download_plan:
|
||||
filepath: Path = Path(args.output) / filename
|
||||
if filepath.exists() and filepath.stat().st_size > 0:
|
||||
log.info("Skipping %s (already exists: %.1f GB)",
|
||||
filename, filepath.stat().st_size / (1024 ** 3))
|
||||
continue
|
||||
if not download_aria2c(mirror_urls, args.output, filename, args.connections):
|
||||
log.error("Failed to download %s", filename)
|
||||
return 1
|
||||
|
||||
total_elapsed: float = time.monotonic() - total_start
|
||||
log.info("All downloads complete in %.0fs", total_elapsed)
|
||||
for filename, _ in download_plan:
|
||||
fp: Path = Path(args.output) / filename
|
||||
if fp.exists():
|
||||
log.info(" %s (%.1f GB)", fp.name, fp.stat().st_size / (1024 ** 3))
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user