internet-latency-collector: add node file generate and verify commands

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
 

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() {

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
+}

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)
+}