From 4b6a0987d26ead0e96f9ac93d62033c29b36249e Mon Sep 17 00:00:00 2001 From: Thijs van Emmerik Date: Fri, 11 Sep 2026 09:09:28 +0000 Subject: [PATCH] internet-latency-collector: add node file generate and verify commands nodefile generate fills each region's ping target from the AWS published address list, keeping the current target when it still answers. nodefile verify checks a committed file against the same list and reports every problem at once rather than stopping at the first. Adds config/nodes-aws.json, seventeen AWS regions with up to four RIPE Atlas probe IDs each, ordered with the preferred probe first. A signal during either command reports cancellation instead of naming live addresses as dead. The collector's main package now spans more than one file, so the make and release builds name the package instead of main.go. --- .../internet-latency-collector/Makefile | 2 +- .../cmd/collector/main.go | 7 + .../cmd/collector/nodefile.go | 206 +++++++++++++++++ .../cmd/collector/nodefile_test.go | 99 +++++++++ .../config/nodes-aws.json | 209 ++++++++++++++++++ ...easer.base.internet-latency-collector.yaml | 2 +- 6 files changed, 523 insertions(+), 2 deletions(-) create mode 100644 controlplane/internet-latency-collector/cmd/collector/nodefile.go create mode 100644 controlplane/internet-latency-collector/cmd/collector/nodefile_test.go create mode 100644 controlplane/internet-latency-collector/config/nodes-aws.json diff --git a/controlplane/internet-latency-collector/Makefile b/controlplane/internet-latency-collector/Makefile index 841cca9791..a6164fe803 100644 --- a/controlplane/internet-latency-collector/Makefile +++ b/controlplane/internet-latency-collector/Makefile @@ -12,5 +12,5 @@ lint: .PHONY: build build: - CGO_ENABLED=0 go build -v $(LDFLAGS) -o bin/doublezero-internet-latency-collector cmd/collector/main.go + CGO_ENABLED=0 go build -v $(LDFLAGS) -o bin/doublezero-internet-latency-collector ./cmd/collector diff --git a/controlplane/internet-latency-collector/cmd/collector/main.go b/controlplane/internet-latency-collector/cmd/collector/main.go index 5b110e8571..1a676b5b63 100644 --- a/controlplane/internet-latency-collector/cmd/collector/main.go +++ b/controlplane/internet-latency-collector/cmd/collector/main.go @@ -434,11 +434,15 @@ func init() { ripeatlasCreateMeasurementsCmd.Flags().IntVar(&ripeatlasProbesPerLocation, "probes-per-location", defaultAtlasProbesPerLocation, "Number of RIPE Atlas probes to associate with each DoubleZero location") + nodefileCmd.PersistentFlags().StringVar(&nodeFilePath, "node-file", defaultNodeFilePath, "Path to the cloud region node file") + nodefileCmd.PersistentFlags().BoolVar(&nodeFileSkipPing, "skip-ping", false, "Check only that each ping target is still published, without sending a ping") + cobra.EnableCommandSorting = false rootCmd.AddCommand(ripeatlasCmd) rootCmd.AddCommand(wheresitupCmd) rootCmd.AddCommand(runCmd) + rootCmd.AddCommand(nodefileCmd) ripeatlasCmd.AddCommand(ripeatlasListProbesCmd) ripeatlasCmd.AddCommand(ripeatlasListMeasurementsCmd) @@ -447,6 +451,9 @@ func init() { wheresitupCmd.AddCommand(wheresitupListSourcesCmd) wheresitupCmd.AddCommand(wheresitupListJobsCmd) + + nodefileCmd.AddCommand(nodefileGenerateCmd) + nodefileCmd.AddCommand(nodefileVerifyCmd) } func main() { diff --git a/controlplane/internet-latency-collector/cmd/collector/nodefile.go b/controlplane/internet-latency-collector/cmd/collector/nodefile.go new file mode 100644 index 0000000000..b07f9c3c2b --- /dev/null +++ b/controlplane/internet-latency-collector/cmd/collector/nodefile.go @@ -0,0 +1,206 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "os" + "os/exec" + "os/signal" + "slices" + "sort" + "syscall" + "time" + + "github.com/spf13/cobra" + + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/awsreach" + collector "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" +) + +const ( + defaultNodeFilePath = "config/nodes-aws.json" + nodeFileHTTPTimeout = 30 * time.Second + nodeFilePingAttempts = 8 +) + +var ( + nodeFilePath string + nodeFileSkipPing bool +) + +var nodefileCmd = &cobra.Command{ + Use: "nodefile", + Short: "Generate and verify the cloud region node file", + Long: `Commands for maintaining the node file that lists cloud regions, their pinned +RIPE Atlas probe IDs, and the address other regions ping to reach them.`, + // These commands read no ledger state, so they replace the root command's network config setup. + PersistentPreRun: func(cmd *cobra.Command, args []string) {}, +} + +var nodefileGenerateCmd = &cobra.Command{ + Use: "generate", + Short: "Resolve a ping target for every node from the AWS published address list", + Run: func(cmd *cobra.Command, args []string) { + log := collector.NewLogger(collector.LogLevel(logLevel)) + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + nodes, targets, ok := loadNodesAndTargets(ctx, log) + if !ok { + os.Exit(1) + } + + var problems []string + for i, node := range nodes { + candidates, published := targets[node.Code] + if !published { + problems = append(problems, fmt.Sprintf("%s: region absent from the AWS address list", node.Code)) + continue + } + + chosen, err := chooseNodeTarget(ctx, candidates, node.PingTarget) + if err != nil { + if ctx.Err() != nil { + log.Info("Operation cancelled by signal") + return + } + problems = append(problems, fmt.Sprintf("%s: no published address answered: %s", node.Code, err.Error())) + continue + } + + if chosen != node.PingTarget { + log.Warn("Ping target changed", + slog.String("code", node.Code), + slog.String("old_target", node.PingTarget), + slog.String("new_target", chosen)) + } + nodes[i].PingTarget = chosen + } + + if len(problems) > 0 { + for _, problem := range problems { + log.Error("Node file generation failed", slog.String("problem", problem)) + } + os.Exit(1) + } + + if err := writeNodeFile(nodeFilePath, nodes); err != nil { + log.Error("Operation failed: write_node_file", slog.String("error", err.Error())) + os.Exit(1) + } + + log.Info("Operation completed: generate_node_file", + slog.String("file", nodeFilePath), + slog.Int("nodes", len(nodes)), + slog.Bool("skipped_ping", nodeFileSkipPing)) + }, +} + +var nodefileVerifyCmd = &cobra.Command{ + Use: "verify", + Short: "Check every ping target is still published by AWS and still answers", + Run: func(cmd *cobra.Command, args []string) { + log := collector.NewLogger(collector.LogLevel(logLevel)) + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + nodes, targets, ok := loadNodesAndTargets(ctx, log) + if !ok { + os.Exit(1) + } + + var problems []string + for _, node := range nodes { + candidates, published := targets[node.Code] + if !published { + problems = append(problems, fmt.Sprintf("%s: region absent from the AWS address list", node.Code)) + continue + } + if !slices.Contains(candidates, node.PingTarget) { + problems = append(problems, fmt.Sprintf("%s: %s is no longer published", node.Code, node.PingTarget)) + continue + } + if nodeFileSkipPing { + continue + } + if !awsreach.SystemPing(ctx, node.PingTarget) { + problems = append(problems, fmt.Sprintf("%s: %s did not answer", node.Code, node.PingTarget)) + } + } + + if ctx.Err() != nil { + log.Info("Operation cancelled by signal") + return + } + + if len(problems) > 0 { + for _, problem := range problems { + log.Error("Node file check failed", slog.String("problem", problem)) + } + os.Exit(1) + } + + log.Info("Operation completed: verify_node_file", + slog.String("file", nodeFilePath), + slog.Int("nodes", len(nodes)), + slog.Bool("skipped_ping", nodeFileSkipPing)) + }, +} + +func loadNodesAndTargets(ctx context.Context, log *slog.Logger) ([]collector.JSONNode, awsreach.RegionTargets, bool) { + if !nodeFileSkipPing { + if _, err := exec.LookPath("ping"); err != nil { + log.Error("Operation failed: ping_binary_missing", slog.String("error", err.Error())) + return nil, nil, false + } + } + + nodes, err := collector.LoadNodesFromJSON(log, nodeFilePath) + if err != nil { + log.Error("Operation failed: load_node_file", + slog.String("file", nodeFilePath), + slog.String("error", err.Error())) + return nil, nil, false + } + + targets, err := awsreach.FetchPrefixes(ctx, &http.Client{Timeout: nodeFileHTTPTimeout}, awsreach.PrefixesURL) + if err != nil { + log.Error("Operation failed: fetch_aws_prefixes", slog.String("error", err.Error())) + return nil, nil, false + } + + return nodes, targets, true +} + +func chooseNodeTarget(ctx context.Context, candidates []string, current string) (string, error) { + ordered := awsreach.PreferFirst(candidates, current) + if len(ordered) == 0 { + return "", fmt.Errorf("no published addresses to choose from") + } + if nodeFileSkipPing { + return ordered[0], nil + } + return awsreach.FirstAnswering(ctx, ordered, awsreach.SystemPing, nodeFilePingAttempts) +} + +// writeNodeFile sorts nodes by code so a regenerated file differs only where a value changed. +func writeNodeFile(path string, nodes []collector.JSONNode) error { + sort.Slice(nodes, func(i, j int) bool { return nodes[i].Code < nodes[j].Code }) + + data, err := json.MarshalIndent(nodes, "", " ") + if err != nil { + return fmt.Errorf("failed to encode node file: %w", err) + } + data = append(data, '\n') + + if err := os.WriteFile(path, data, 0644); err != nil { + return fmt.Errorf("failed to write node file: %w", err) + } + + return nil +} diff --git a/controlplane/internet-latency-collector/cmd/collector/nodefile_test.go b/controlplane/internet-latency-collector/cmd/collector/nodefile_test.go new file mode 100644 index 0000000000..49a7dedae7 --- /dev/null +++ b/controlplane/internet-latency-collector/cmd/collector/nodefile_test.go @@ -0,0 +1,99 @@ +package main + +import ( + "log/slog" + "net" + "os" + "path/filepath" + "slices" + "testing" + + collector "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" + "github.com/stretchr/testify/require" +) + +func TestInternetLatency_NodeFile_CommittedAWSNodes(t *testing.T) { + t.Parallel() + + log := slog.New(slog.DiscardHandler) + + nodes, err := collector.LoadNodesFromJSON(log, "../../config/nodes-aws.json") + require.NoError(t, err) + require.Len(t, nodes, 17, "17 regions give 136 pairs") + + expected := map[string]struct { + probeIDs []int + target string + }{ + "us-east-1": {[]int{1003385, 1009925, 1010723, 1012092}, "34.192.0.54"}, + "us-east-2": {[]int{1003386, 1000074, 1005330, 1015708}, "3.130.0.254"}, + "us-west-1": {[]int{1003387, 1013400}, "13.52.0.0"}, + "us-west-2": {[]int{1003388, 1005331, 1007744, 1012207}, "35.95.2.254"}, + "ca-central-1": {[]int{1003389, 1005332, 1015534}, "3.98.0.0"}, + "sa-east-1": {[]int{1000709, 1002617, 1015704}, "15.228.0.0"}, + "eu-west-1": {[]int{1003378, 1002616, 1010727, 1012211}, "3.248.0.0"}, + "eu-west-2": {[]int{1003377, 1005333, 1009922, 1015778}, "3.8.0.0"}, + "eu-west-3": {[]int{1003375, 1016689}, "13.36.0.0"}, + "eu-central-1": {[]int{1000566, 1005334, 1015777, 1016525}, "3.64.0.0"}, + "eu-north-1": {[]int{1003374, 1005867}, "13.50.0.254"}, + "eu-south-2": {[]int{1004991, 1016435}, "15.216.0.0"}, + "ap-northeast-1": {[]int{1003384, 1010741, 1012762, 1013401}, "3.112.0.0"}, + "ap-northeast-2": {[]int{1002619, 1015781, 1017320}, "13.209.0.0"}, + "ap-east-1": {[]int{1012347, 1012349, 1012350, 1012351}, "16.162.0.253"}, + "ap-southeast-1": {[]int{1003382, 1002618, 1012208, 1015779}, "3.0.0.9"}, + "ap-south-1": {[]int{1003379}, "3.6.0.0"}, + } + + seen := map[string]bool{} + for i, node := range nodes { + want, ok := expected[node.Code] + require.True(t, ok, "unexpected region %s", node.Code) + + require.Equal(t, "aws", node.Cloud, "%s cloud", node.Code) + require.Equal(t, want.probeIDs, node.AtlasProbeIDs, "%s probe ids", node.Code) + require.Equal(t, want.target, node.PingTarget, "%s ping target", node.Code) + require.NotNil(t, net.ParseIP(node.PingTarget).To4(), "%s ping target must be IPv4", node.Code) + require.NotZero(t, node.Latitude, "%s latitude", node.Code) + require.NotZero(t, node.Longitude, "%s longitude", node.Code) + + if i > 0 { + require.Less(t, nodes[i-1].Code, node.Code, "the node file must be sorted by code") + } + + seen[node.Code] = true + } + + require.Len(t, seen, len(expected), "every expected region must be present exactly once") +} + +func TestInternetLatency_NodeFile_WriteMatchesCommittedFile(t *testing.T) { + t.Parallel() + + committed, err := os.ReadFile("../../config/nodes-aws.json") + require.NoError(t, err) + + nodes, err := collector.LoadNodesFromJSON(slog.New(slog.DiscardHandler), "../../config/nodes-aws.json") + require.NoError(t, err) + + path := filepath.Join(t.TempDir(), "nodes.json") + slices.Reverse(nodes) + require.NoError(t, writeNodeFile(path, nodes)) + + written, err := os.ReadFile(path) + require.NoError(t, err) + require.Equal(t, string(committed), string(written), + "generate must rewrite an unchanged node file byte for byte") +} + +func TestInternetLatency_NodeFile_ChooseNodeTargetSkipPing(t *testing.T) { + previous := nodeFileSkipPing + nodeFileSkipPing = true + defer func() { nodeFileSkipPing = previous }() + + chosen, err := chooseNodeTarget(t.Context(), []string{"1.1.1.1", "2.2.2.2"}, "2.2.2.2") + require.NoError(t, err) + require.Equal(t, "2.2.2.2", chosen, "the address already in the file is kept when it is still published") + + _, err = chooseNodeTarget(t.Context(), nil, "2.2.2.2") + require.Error(t, err) +} diff --git a/controlplane/internet-latency-collector/config/nodes-aws.json b/controlplane/internet-latency-collector/config/nodes-aws.json new file mode 100644 index 0000000000..5f82c45b28 --- /dev/null +++ b/controlplane/internet-latency-collector/config/nodes-aws.json @@ -0,0 +1,209 @@ +[ + { + "code": "ap-east-1", + "cloud": "aws", + "lat": 22.3193, + "lng": 114.1694, + "atlas_probe_ids": [ + 1012347, + 1012349, + 1012350, + 1012351 + ], + "ping_target": "16.162.0.253" + }, + { + "code": "ap-northeast-1", + "cloud": "aws", + "lat": 35.6762, + "lng": 139.6503, + "atlas_probe_ids": [ + 1003384, + 1010741, + 1012762, + 1013401 + ], + "ping_target": "3.112.0.0" + }, + { + "code": "ap-northeast-2", + "cloud": "aws", + "lat": 37.5665, + "lng": 126.978, + "atlas_probe_ids": [ + 1002619, + 1015781, + 1017320 + ], + "ping_target": "13.209.0.0" + }, + { + "code": "ap-south-1", + "cloud": "aws", + "lat": 19.076, + "lng": 72.8777, + "atlas_probe_ids": [ + 1003379 + ], + "ping_target": "3.6.0.0" + }, + { + "code": "ap-southeast-1", + "cloud": "aws", + "lat": 1.3521, + "lng": 103.8198, + "atlas_probe_ids": [ + 1003382, + 1002618, + 1012208, + 1015779 + ], + "ping_target": "3.0.0.9" + }, + { + "code": "ca-central-1", + "cloud": "aws", + "lat": 45.5019, + "lng": -73.5674, + "atlas_probe_ids": [ + 1003389, + 1005332, + 1015534 + ], + "ping_target": "3.98.0.0" + }, + { + "code": "eu-central-1", + "cloud": "aws", + "lat": 50.1109, + "lng": 8.6821, + "atlas_probe_ids": [ + 1000566, + 1005334, + 1015777, + 1016525 + ], + "ping_target": "3.64.0.0" + }, + { + "code": "eu-north-1", + "cloud": "aws", + "lat": 59.3293, + "lng": 18.0686, + "atlas_probe_ids": [ + 1003374, + 1005867 + ], + "ping_target": "13.50.0.254" + }, + { + "code": "eu-south-2", + "cloud": "aws", + "lat": 41.6488, + "lng": -0.8891, + "atlas_probe_ids": [ + 1004991, + 1016435 + ], + "ping_target": "15.216.0.0" + }, + { + "code": "eu-west-1", + "cloud": "aws", + "lat": 53.3498, + "lng": -6.2603, + "atlas_probe_ids": [ + 1003378, + 1002616, + 1010727, + 1012211 + ], + "ping_target": "3.248.0.0" + }, + { + "code": "eu-west-2", + "cloud": "aws", + "lat": 51.5074, + "lng": -0.1278, + "atlas_probe_ids": [ + 1003377, + 1005333, + 1009922, + 1015778 + ], + "ping_target": "3.8.0.0" + }, + { + "code": "eu-west-3", + "cloud": "aws", + "lat": 48.8566, + "lng": 2.3522, + "atlas_probe_ids": [ + 1003375, + 1016689 + ], + "ping_target": "13.36.0.0" + }, + { + "code": "sa-east-1", + "cloud": "aws", + "lat": -23.5505, + "lng": -46.6333, + "atlas_probe_ids": [ + 1000709, + 1002617, + 1015704 + ], + "ping_target": "15.228.0.0" + }, + { + "code": "us-east-1", + "cloud": "aws", + "lat": 39.0438, + "lng": -77.4874, + "atlas_probe_ids": [ + 1003385, + 1009925, + 1010723, + 1012092 + ], + "ping_target": "34.192.0.54" + }, + { + "code": "us-east-2", + "cloud": "aws", + "lat": 39.9612, + "lng": -82.9988, + "atlas_probe_ids": [ + 1003386, + 1000074, + 1005330, + 1015708 + ], + "ping_target": "3.130.0.254" + }, + { + "code": "us-west-1", + "cloud": "aws", + "lat": 37.3382, + "lng": -121.8863, + "atlas_probe_ids": [ + 1003387, + 1013400 + ], + "ping_target": "13.52.0.0" + }, + { + "code": "us-west-2", + "cloud": "aws", + "lat": 45.8399, + "lng": -119.7006, + "atlas_probe_ids": [ + 1003388, + 1005331, + 1007744, + 1012207 + ], + "ping_target": "35.95.2.254" + } +] diff --git a/release/.goreleaser.base.internet-latency-collector.yaml b/release/.goreleaser.base.internet-latency-collector.yaml index d53319f0ad..41e0115482 100644 --- a/release/.goreleaser.base.internet-latency-collector.yaml +++ b/release/.goreleaser.base.internet-latency-collector.yaml @@ -11,7 +11,7 @@ monorepo: builds: - id: doublezero-internet-latency-collector - main: cmd/collector/main.go + main: ./cmd/collector binary: doublezero-internet-latency-collector env: - CGO_ENABLED=0