From 1f7d6de052f67eb279cd13afeb1a4069244e049e Mon Sep 17 00:00:00 2001 From: Thijs van Emmerik Date: Fri, 11 Sep 2026 09:08:35 +0000 Subject: [PATCH] internet-latency-collector: add aws reachability package Fetches the AWS published address list, keys the prefixes by exact region name so a Local Zone such as us-west-2-lax-1 keeps its own key, and pings candidate addresses in order to find the first that answers. Self-contained; no collector changes. --- .../internal/awsreach/awsreach.go | 129 ++++++++++++++ .../internal/awsreach/awsreach_test.go | 167 ++++++++++++++++++ 2 files changed, 296 insertions(+) create mode 100644 controlplane/internet-latency-collector/internal/awsreach/awsreach.go create mode 100644 controlplane/internet-latency-collector/internal/awsreach/awsreach_test.go diff --git a/controlplane/internet-latency-collector/internal/awsreach/awsreach.go b/controlplane/internet-latency-collector/internal/awsreach/awsreach.go new file mode 100644 index 0000000000..7cecb9fb31 --- /dev/null +++ b/controlplane/internet-latency-collector/internal/awsreach/awsreach.go @@ -0,0 +1,129 @@ +package awsreach + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "os/exec" + "slices" + "sort" + "time" + + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" +) + +// PrefixesURL is served over plain HTTP; the https URL redirects back to http. +const PrefixesURL = "http://ec2-reachability.amazonaws.com/prefixes-ipv4.json" + +const pingTimeout = 3 * time.Second + +type RegionTargets map[string][]string + +// ParsePrefixes keys targets by exact region name. Local Zones such as us-west-2-lax-1 carry their own key. +func ParsePrefixes(r io.Reader) (RegionTargets, error) { + var raw []map[string]map[string]string + if err := json.NewDecoder(r).Decode(&raw); err != nil { + return nil, fmt.Errorf("failed to decode prefixes document: %w", err) + } + if len(raw) == 0 { + return nil, errors.New("prefixes document lists no regions") + } + + parsed := make(map[string]map[string]net.IP) + for _, entry := range raw { + for region, cidrToAddress := range entry { + for _, candidate := range cidrToAddress { + ip := net.ParseIP(candidate) + if ip == nil || ip.To4() == nil { + continue + } + if parsed[region] == nil { + parsed[region] = make(map[string]net.IP) + } + parsed[region][ip.String()] = ip.To4() + } + } + } + + if len(parsed) == 0 { + return nil, errors.New("prefixes document lists no regions with IPv4 addresses") + } + + targets := make(RegionTargets, len(parsed)) + for region, byAddress := range parsed { + ips := make([]net.IP, 0, len(byAddress)) + for _, ip := range byAddress { + ips = append(ips, ip) + } + sort.Slice(ips, func(i, j int) bool { return bytes.Compare(ips[i], ips[j]) < 0 }) + + addresses := make([]string, len(ips)) + for i, ip := range ips { + addresses[i] = ip.String() + } + targets[region] = addresses + } + + return targets, nil +} + +func FetchPrefixes(ctx context.Context, client collector.HTTPClient, url string) (RegionTargets, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to fetch prefixes: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("prefixes request failed with status: %d", resp.StatusCode) + } + + return ParsePrefixes(resp.Body) +} + +type PingFunc func(ctx context.Context, address string) bool + +func SystemPing(ctx context.Context, address string) bool { + ctx, cancel := context.WithTimeout(ctx, pingTimeout) + defer cancel() + + return exec.CommandContext(ctx, "ping", "-n", "-c", "1", "-W", "2", address).Run() == nil +} + +func PreferFirst(candidates []string, preferred string) []string { + index := slices.Index(candidates, preferred) + if index <= 0 { + return candidates + } + + reordered := make([]string, 0, len(candidates)) + reordered = append(reordered, candidates[index]) + reordered = append(reordered, candidates[:index]...) + reordered = append(reordered, candidates[index+1:]...) + return reordered +} + +func FirstAnswering(ctx context.Context, candidates []string, ping PingFunc, maxAttempts int) (string, error) { + attempts := 0 + for _, candidate := range candidates { + if attempts >= maxAttempts { + break + } + attempts++ + if ping(ctx, candidate) { + return candidate, nil + } + } + + return "", fmt.Errorf("no address answered after %d attempts", attempts) +} diff --git a/controlplane/internet-latency-collector/internal/awsreach/awsreach_test.go b/controlplane/internet-latency-collector/internal/awsreach/awsreach_test.go new file mode 100644 index 0000000000..0b168813fa --- /dev/null +++ b/controlplane/internet-latency-collector/internal/awsreach/awsreach_test.go @@ -0,0 +1,167 @@ +package awsreach + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +const testPrefixes = `[ + {"us-west-2": {"35.95.0.0/16": "35.95.2.254", "34.208.0.0/12": "34.208.0.0", "35.100.0.0/16": "35.100.0.0"}}, + {"us-west-2-lax-1": {"70.224.192.0/18": "70.224.192.0"}}, + {"eu-central-1": {"3.64.0.0/12": "3.64.0.0", "18.153.0.0/16": "not-an-address"}} +]` + +type mockHTTPClient struct { + DoFunc func(req *http.Request) (*http.Response, error) +} + +func (m *mockHTTPClient) Do(req *http.Request) (*http.Response, error) { + if m.DoFunc != nil { + return m.DoFunc(req) + } + return nil, errors.New("mock not configured") +} + +func TestInternetLatency_AWSReach_ParsePrefixes(t *testing.T) { + t.Parallel() + + targets, err := ParsePrefixes(strings.NewReader(testPrefixes)) + require.NoError(t, err) + + require.Len(t, targets, 3, "every region key stands on its own") + require.Equal(t, []string{"34.208.0.0", "35.95.2.254", "35.100.0.0"}, targets["us-west-2"], + "addresses must be sorted numerically so output is stable") + require.Equal(t, []string{"70.224.192.0"}, targets["us-west-2-lax-1"], + "a Local Zone must not fold into its parent region") + require.Equal(t, []string{"3.64.0.0"}, targets["eu-central-1"], + "values that are not IPv4 addresses are dropped") +} + +func TestInternetLatency_AWSReach_ParsePrefixes_Empty(t *testing.T) { + t.Parallel() + + _, err := ParsePrefixes(strings.NewReader(`[]`)) + require.Error(t, err) + require.ErrorContains(t, err, "no regions") +} + +func TestInternetLatency_AWSReach_ParsePrefixes_NoIPv4(t *testing.T) { + t.Parallel() + + _, err := ParsePrefixes(strings.NewReader(`[{"eu-west-1": {"18.153.0.0/16": "not-an-address"}}]`)) + require.Error(t, err) + require.ErrorContains(t, err, "no regions with IPv4 addresses") +} + +func TestInternetLatency_AWSReach_FetchPrefixes(t *testing.T) { + t.Parallel() + + var requestedURL string + client := &mockHTTPClient{ + DoFunc: func(req *http.Request) (*http.Response, error) { + requestedURL = req.URL.String() + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewBufferString(testPrefixes)), + }, nil + }, + } + + targets, err := FetchPrefixes(t.Context(), client, PrefixesURL) + require.NoError(t, err) + require.Equal(t, "http://ec2-reachability.amazonaws.com/prefixes-ipv4.json", requestedURL, + "the list is served over plain HTTP only") + require.Contains(t, targets, "eu-central-1") +} + +func TestInternetLatency_AWSReach_FetchPrefixes_BadStatus(t *testing.T) { + t.Parallel() + + client := &mockHTTPClient{ + DoFunc: func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusServiceUnavailable, + Body: io.NopCloser(bytes.NewBufferString("")), + }, nil + }, + } + + _, err := FetchPrefixes(t.Context(), client, PrefixesURL) + require.Error(t, err) + require.ErrorContains(t, err, "status: 503") +} + +func TestInternetLatency_AWSReach_PreferFirst(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + candidates []string + preferred string + want []string + }{ + { + name: "Preferred moves to the front", + candidates: []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"}, + preferred: "3.3.3.3", + want: []string{"3.3.3.3", "1.1.1.1", "2.2.2.2"}, + }, + { + name: "Preferred already first", + candidates: []string{"1.1.1.1", "2.2.2.2"}, + preferred: "1.1.1.1", + want: []string{"1.1.1.1", "2.2.2.2"}, + }, + { + name: "Preferred absent", + candidates: []string{"1.1.1.1", "2.2.2.2"}, + preferred: "9.9.9.9", + want: []string{"1.1.1.1", "2.2.2.2"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, tt.want, PreferFirst(tt.candidates, tt.preferred)) + }) + } +} + +func TestInternetLatency_AWSReach_FirstAnswering(t *testing.T) { + t.Parallel() + + var tried []string + ping := func(ctx context.Context, address string) bool { + tried = append(tried, address) + return address == "3.3.3.3" + } + + got, err := FirstAnswering(t.Context(), []string{"1.1.1.1", "2.2.2.2", "3.3.3.3", "4.4.4.4"}, ping, 8) + require.NoError(t, err) + require.Equal(t, "3.3.3.3", got) + require.Equal(t, []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"}, tried, + "it stops at the first address that answers") +} + +func TestInternetLatency_AWSReach_FirstAnswering_AttemptCap(t *testing.T) { + t.Parallel() + + attempts := 0 + ping := func(ctx context.Context, address string) bool { + attempts++ + return false + } + + _, err := FirstAnswering(t.Context(), []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"}, ping, 2) + require.Error(t, err) + require.ErrorContains(t, err, "no address answered after 2 attempts") + require.Equal(t, 2, attempts) +}