internet-latency-collector: add cloud region mode, off by default

controlplane/internet-latency-collector/cmd/collector/cloud_test.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"log/slog"
+	"testing"
+
+	collector "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInternetLatency_Cloud_LoadCloudNodesCarriesEveryField(t *testing.T) {
+	t.Parallel()
+
+	log := slog.New(slog.DiscardHandler)
+
+	jsonNodes, err := collector.LoadNodesFromJSON(log, "../../config/nodes-aws.json")
+	require.NoError(t, err)
+
+	nodes, err := loadCloudNodes(log, "../../config/nodes-aws.json")
+	require.NoError(t, err)
+	require.Len(t, nodes, len(jsonNodes))
+
+	for i, node := range nodes {
+		require.Equal(t, jsonNodes[i].Code, node.Code)
+		require.Equal(t, jsonNodes[i].Cloud, node.Cloud)
+		require.Equal(t, jsonNodes[i].Latitude, node.Latitude)
+		require.Equal(t, jsonNodes[i].Longitude, node.Longitude)
+		require.Equal(t, jsonNodes[i].AtlasProbeIDs, node.AtlasProbeIDs)
+		require.Equal(t, jsonNodes[i].PingTarget, node.PingTarget)
+	}
+}
+
+func TestInternetLatency_Cloud_NodeFilePath(t *testing.T) {
+	previous := cloudNodeFile
+	defer func() { cloudNodeFile = previous }()
+
+	cloudNodeFile = ""
+	require.Empty(t, cloudNodeFilePath(), "cloud mode stays off when neither the flag nor the environment names a file")
+
+	t.Setenv(cloudNodeFileEnvVar, "from-env.json")
+	require.Equal(t, "from-env.json", cloudNodeFilePath())
+
+	cloudNodeFile = "from-flag.json"
+	require.Equal(t, "from-flag.json", cloudNodeFilePath(), "the flag wins over the environment")
+}

controlplane/internet-latency-collector/cmd/collector/main.go

@@ -37,6 +37,7 @@ const (
 	defaultLedgerSubmissionInterval     = 1 * time.Minute
 	defaultWheresitupStateFile          = "wheresitup_jobs_to_process.json"
 	defaultLogLevel                     = "info"
+	cloudNodeFileEnvVar                 = "DZ_ILC_CLOUD_NODE_FILE"
 
 	// defaultLedgerRPCTimeout bounds each individual ledger RPC request. The default solana-go
 	// client uses a 5-minute timeout, which lets a request block long enough for a fetched
@@ -59,6 +60,7 @@ var (
 	stateDir                     string
 	logLevel                     string
 	locationFile                 string
+	cloudNodeFile                string
 	dryRun                       bool
 	wheresitupStateFile          string
 	ripeatlasProbesPerLocation   int
@@ -162,18 +164,34 @@ RIPE Atlas measurements hourly, and exports RIPE Atlas results periodically.`,
 			os.Exit(1)
 		}
 
-		// Create data provider collectors.
-		ripeatlasCollector := ripeatlas.NewCollector(log, exporter, env, func(ctx context.Context) []collector.LocationMatch {
-			return collector.GetLocations(ctx, log, serviceabilityClient)
-		})
-		wheresitupCollector := wheresitup.NewCollector(log, exporter, env, func(ctx context.Context) []collector.LocationMatch {
-			return collector.GetLocations(ctx, log, serviceabilityClient)
-		})
+		// Create data provider collectors. A cloud node file selects cloud mode: RIPE Atlas alone.
+		nodeFile := cloudNodeFilePath()
+		var ripeatlasCollector collector.RipeAtlasCollectorInterface
+		var wheresitupCollector collector.WheresitupCollectorInterface
+		if nodeFile != "" {
+			nodes, err := loadCloudNodes(log, nodeFile)
+			if err != nil {
+				log.Error("failed to load cloud node file", "error", err, "file", nodeFile)
+				os.Exit(1)
+			}
+			log.Info("Running in cloud mode",
+				slog.String("node_file", nodeFile),
+				slog.Int("node_count", len(nodes)))
+			ripeatlasCollector = ripeatlas.NewCloudCollector(log, exporter, env, nodes)
+		} else {
+			ripeatlasCollector = ripeatlas.NewCollector(log, exporter, env, func(ctx context.Context) []collector.LocationMatch {
+				return collector.GetLocations(ctx, log, serviceabilityClient)
+			})
+			wheresitupCollector = wheresitup.NewCollector(log, exporter, env, func(ctx context.Context) []collector.LocationMatch {
+				return collector.GetLocations(ctx, log, serviceabilityClient)
+			})
+		}
 
 		config := collector.Config{
 			Logger:     log,
 			Wheresitup: wheresitupCollector,
 			RipeAtlas:  ripeatlasCollector,
+			CloudMode:  nodeFile != "",
 
 			WheresitupSamplingInterval:   defaultWheresitupSamplingInterval,
 			RipeAtlasSamplingInterval:    defaultRipeAtlasSamplingInterval,
@@ -379,6 +397,34 @@ var wheresitupListJobsCmd = &cobra.Command{
 	},
 }
 
+func cloudNodeFilePath() string {
+	if cloudNodeFile != "" {
+		return cloudNodeFile
+	}
+	return os.Getenv(cloudNodeFileEnvVar)
+}
+
+func loadCloudNodes(logger *slog.Logger, filename string) ([]ripeatlas.CloudNode, error) {
+	jsonNodes, err := collector.LoadNodesFromJSON(logger, filename)
+	if err != nil {
+		return nil, err
+	}
+
+	nodes := make([]ripeatlas.CloudNode, 0, len(jsonNodes))
+	for _, node := range jsonNodes {
+		nodes = append(nodes, ripeatlas.CloudNode{
+			Code:          node.Code,
+			Cloud:         node.Cloud,
+			Latitude:      node.Latitude,
+			Longitude:     node.Longitude,
+			AtlasProbeIDs: node.AtlasProbeIDs,
+			PingTarget:    node.PingTarget,
+		})
+	}
+
+	return nodes, nil
+}
+
 func loadLocations(ctx context.Context, logger *slog.Logger, serviceabilityClient *serviceability.Client) []collector.LocationMatch {
 	if locationFile != "" {
 		logger.Info("Loading locations from JSON file", slog.String("file", locationFile))
@@ -431,6 +477,7 @@ func init() {
 	runCmd.Flags().DurationVar(&ripeatlasMeasurementInterval, "ripeatlas-measurement-interval", defaultRipeAtlasMeasurementInterval, "Interval at which to run RIPE Atlas measurements")
 	runCmd.Flags().DurationVar(&ledgerSubmissionInterval, "ledger-submission-interval", defaultLedgerSubmissionInterval, "Interval at which to submit metrics to the ledger")
 	runCmd.Flags().StringVar(&metricsAddr, "metrics-addr", "127.0.0.1:2113", "Address to bind the metrics server to")
+	runCmd.Flags().StringVar(&cloudNodeFile, "cloud-node-file", "", "JSON file of cloud regions to measure (code, cloud, lat, lng, atlas_probe_ids, ping_target); enables cloud mode. Falls back to "+cloudNodeFileEnvVar)
 
 	ripeatlasCreateMeasurementsCmd.Flags().IntVar(&ripeatlasProbesPerLocation, "probes-per-location", defaultAtlasProbesPerLocation, "Number of RIPE Atlas probes to associate with each DoubleZero location")
 

controlplane/internet-latency-collector/internal/collector/collector.go

@@ -33,35 +33,45 @@ type Config struct {
 	StateDir                     string
 	ProbesPerLocation            int
 	MetricsAddr                  string
+	CloudMode                    bool
 }
 
 func (cfg *Config) Validate() error {
 	if cfg.Logger == nil {
 		return errors.New("logger is required")
 	}
-	if cfg.Wheresitup == nil {
-		return errors.New("wheresitup collector is required")
+	if cfg.Wheresitup == nil && cfg.RipeAtlas == nil {
+		return errors.New("at least one of the wheresitup and ripe atlas collectors is required")
 	}
-	if cfg.RipeAtlas == nil {
-		return errors.New("ripe atlas collector is required")
+	if cfg.CloudMode {
+		if cfg.RipeAtlas == nil {
+			return errors.New("ripe atlas collector is required in cloud mode")
+		}
+		if cfg.Wheresitup != nil {
+			return errors.New("wheresitup collector is not supported in cloud mode")
+		}
 	}
-	if cfg.WheresitupSamplingInterval <= 0 {
-		return errors.New("wheresitup sampling interval must be greater than 0")
+	if cfg.Wheresitup != nil {
+		if cfg.WheresitupSamplingInterval <= 0 {
+			return errors.New("wheresitup sampling interval must be greater than 0")
+		}
+		if cfg.ProcessedJobsFile == "" {
+			return errors.New("processed jobs file is required")
+		}
 	}
-	if cfg.RipeAtlasSamplingInterval <= 0 {
-		return errors.New("ripe atlas sampling interval must be greater than 0")
-	}
-	if cfg.RipeAtlasMeasurementInterval <= 0 {
-		return errors.New("ripe atlas measurement interval must be greater than 0")
-	}
-	if cfg.RipeAtlasExportInterval <= 0 {
-		return errors.New("ripe atlas export interval must be greater than 0")
-	}
-	if cfg.ProbesPerLocation <= 0 {
-		return errors.New("probes per location must be greater than 0")
-	}
-	if cfg.ProcessedJobsFile == "" {
-		return errors.New("processed jobs file is required")
+	if cfg.RipeAtlas != nil {
+		if cfg.RipeAtlasSamplingInterval <= 0 {
+			return errors.New("ripe atlas sampling interval must be greater than 0")
+		}
+		if cfg.RipeAtlasMeasurementInterval <= 0 {
+			return errors.New("ripe atlas measurement interval must be greater than 0")
+		}
+		if cfg.RipeAtlasExportInterval <= 0 {
+			return errors.New("ripe atlas export interval must be greater than 0")
+		}
+		if cfg.ProbesPerLocation <= 0 {
+			return errors.New("probes per location must be greater than 0")
+		}
 	}
 	if cfg.StateDir == "" {
 		return errors.New("state directory is required")

controlplane/internet-latency-collector/internal/collector/collector_test.go

@@ -0,0 +1,175 @@
+package collector
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestInternetLatency_Collector_Validate(t *testing.T) {
+	t.Parallel()
+
+	log := logger.With("test", t.Name())
+
+	base := func() Config {
+		return Config{
+			Logger:    log,
+			RipeAtlas: &MockRipeAtlasCollector{},
+
+			RipeAtlasSamplingInterval:    10 * time.Minute,
+			RipeAtlasMeasurementInterval: 1 * time.Hour,
+			RipeAtlasExportInterval:      10 * time.Minute,
+			ProbesPerLocation:            1,
+			StateDir:                     t.TempDir(),
+		}
+	}
+
+	tests := []struct {
+		name    string
+		mutate  func(cfg *Config)
+		wantErr string
+	}{
+		{
+			name:   "ripe atlas only",
+			mutate: func(cfg *Config) {},
+		},
+		{
+			name: "wheresitup only",
+			mutate: func(cfg *Config) {
+				cfg.RipeAtlas = nil
+				cfg.Wheresitup = &MockWheresitupCollector{}
+				cfg.WheresitupSamplingInterval = 6 * time.Minute
+				cfg.ProcessedJobsFile = "jobs.json"
+			},
+		},
+		{
+			name: "both collectors",
+			mutate: func(cfg *Config) {
+				cfg.Wheresitup = &MockWheresitupCollector{}
+				cfg.WheresitupSamplingInterval = 6 * time.Minute
+				cfg.ProcessedJobsFile = "jobs.json"
+			},
+		},
+		{
+			name: "no collector",
+			mutate: func(cfg *Config) {
+				cfg.RipeAtlas = nil
+			},
+			wantErr: "at least one of the wheresitup and ripe atlas collectors is required",
+		},
+		{
+			name: "cloud mode with ripe atlas only",
+			mutate: func(cfg *Config) {
+				cfg.CloudMode = true
+			},
+		},
+		{
+			name: "cloud mode without ripe atlas",
+			mutate: func(cfg *Config) {
+				cfg.CloudMode = true
+				cfg.RipeAtlas = nil
+				cfg.Wheresitup = &MockWheresitupCollector{}
+				cfg.WheresitupSamplingInterval = 6 * time.Minute
+				cfg.ProcessedJobsFile = "jobs.json"
+			},
+			wantErr: "ripe atlas collector is required in cloud mode",
+		},
+		{
+			name: "cloud mode with wheresitup",
+			mutate: func(cfg *Config) {
+				cfg.CloudMode = true
+				cfg.Wheresitup = &MockWheresitupCollector{}
+				cfg.WheresitupSamplingInterval = 6 * time.Minute
+				cfg.ProcessedJobsFile = "jobs.json"
+			},
+			wantErr: "wheresitup collector is not supported in cloud mode",
+		},
+		{
+			name: "ripe atlas without sampling interval",
+			mutate: func(cfg *Config) {
+				cfg.RipeAtlasSamplingInterval = 0
+			},
+			wantErr: "ripe atlas sampling interval must be greater than 0",
+		},
+		{
+			name: "wheresitup without processed jobs file",
+			mutate: func(cfg *Config) {
+				cfg.Wheresitup = &MockWheresitupCollector{}
+				cfg.WheresitupSamplingInterval = 6 * time.Minute
+			},
+			wantErr: "processed jobs file is required",
+		},
+		{
+			name: "no state dir",
+			mutate: func(cfg *Config) {
+				cfg.StateDir = ""
+			},
+			wantErr: "state directory is required",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			cfg := base()
+			tt.mutate(&cfg)
+
+			err := cfg.Validate()
+			if tt.wantErr == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.EqualError(t, err, tt.wantErr)
+		})
+	}
+}
+
+func TestInternetLatency_Collector_Run_RipeAtlasOnly(t *testing.T) {
+	t.Parallel()
+
+	log := logger.With("test", t.Name())
+
+	mockRipe := &MockRipeAtlasCollector{}
+
+	c, err := New(Config{
+		Logger:    log,
+		RipeAtlas: mockRipe,
+
+		RipeAtlasSamplingInterval:    1 * time.Minute,
+		RipeAtlasMeasurementInterval: 1 * time.Hour,
+		RipeAtlasExportInterval:      2 * time.Minute,
+		DryRun:                       true,
+		StateDir:                     t.TempDir(),
+		ProbesPerLocation:            1,
+		MetricsAddr:                  "127.0.0.1:0",
+	})
+	require.NoError(t, err)
+
+	require.NoError(t, c.Run(t.Context()))
+	require.True(t, mockRipe.wasRunCalled(), "RIPE Atlas collector should have been called")
+}
+
+func TestInternetLatency_Collector_Run_WheresitupOnly(t *testing.T) {
+	t.Parallel()
+
+	log := logger.With("test", t.Name())
+
+	mockWheresitup := &MockWheresitupCollector{}
+
+	c, err := New(Config{
+		Logger:     log,
+		Wheresitup: mockWheresitup,
+
+		WheresitupSamplingInterval: 6 * time.Minute,
+		DryRun:                     true,
+		ProcessedJobsFile:          "jobs.json",
+		StateDir:                   t.TempDir(),
+		MetricsAddr:                "127.0.0.1:0",
+	})
+	require.NoError(t, err)
+
+	require.NoError(t, c.Run(t.Context()))
+	require.True(t, mockWheresitup.wasRunCalled(), "Wheresitup collector should have been called")
+}