diff --git a/controlplane/internet-latency-collector/cmd/collector/cloud_test.go b/controlplane/internet-latency-collector/cmd/collector/cloud_test.go new file mode 100644 index 0000000000..3babc7737d --- /dev/null +++ b/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") +} diff --git a/controlplane/internet-latency-collector/cmd/collector/main.go b/controlplane/internet-latency-collector/cmd/collector/main.go index 1a676b5b63..4f5bf2f1f7 100644 --- a/controlplane/internet-latency-collector/cmd/collector/main.go +++ b/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") diff --git a/controlplane/internet-latency-collector/internal/collector/collector.go b/controlplane/internet-latency-collector/internal/collector/collector.go index e4f50f8fcd..8ded474c8c 100644 --- a/controlplane/internet-latency-collector/internal/collector/collector.go +++ b/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") diff --git a/controlplane/internet-latency-collector/internal/collector/collector_test.go b/controlplane/internet-latency-collector/internal/collector/collector_test.go new file mode 100644 index 0000000000..95a019591a --- /dev/null +++ b/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") +} diff --git a/controlplane/internet-latency-collector/internal/collector/run.go b/controlplane/internet-latency-collector/internal/collector/run.go index 9bb863a649..e5bf8e2907 100644 --- a/controlplane/internet-latency-collector/internal/collector/run.go +++ b/controlplane/internet-latency-collector/internal/collector/run.go @@ -12,6 +12,7 @@ import ( func (c *Collector) Run(ctx context.Context) error { c.log.Info("Starting continuous collector", + slog.Bool("cloud_mode", c.cfg.CloudMode), slog.String("wheresitup_sampling_interval", c.cfg.WheresitupSamplingInterval.String()), slog.String("ripe_atlas_sampling_interval", c.cfg.RipeAtlasSamplingInterval.String()), slog.String("ripe_atlas_measurement_interval", c.cfg.RipeAtlasMeasurementInterval.String()), @@ -24,16 +25,20 @@ func (c *Collector) Run(ctx context.Context) error { c.log.Info("Initializing metrics") - if err := c.cfg.Wheresitup.InitializeCreditBalance(ctx); err != nil { - c.log.Warn("Failed to initialize Wheresitup credit balance metric", slog.String("error", err.Error())) + if c.cfg.Wheresitup != nil { + if err := c.cfg.Wheresitup.InitializeCreditBalance(ctx); err != nil { + c.log.Warn("Failed to initialize Wheresitup credit balance metric", slog.String("error", err.Error())) + } } - if err := c.cfg.RipeAtlas.InitializeCreditBalance(ctx); err != nil { - c.log.Warn("Failed to initialize RIPE Atlas credit balance metric", slog.String("error", err.Error())) - } + if c.cfg.RipeAtlas != nil { + if err := c.cfg.RipeAtlas.InitializeCreditBalance(ctx); err != nil { + c.log.Warn("Failed to initialize RIPE Atlas credit balance metric", slog.String("error", err.Error())) + } - if err := c.cfg.RipeAtlas.InitializeMeasurementMetrics(c.cfg.StateDir); err != nil { - c.log.Warn("Failed to initialize RIPE Atlas measurement metrics", slog.String("error", err.Error())) + if err := c.cfg.RipeAtlas.InitializeMeasurementMetrics(c.cfg.StateDir); err != nil { + c.log.Warn("Failed to initialize RIPE Atlas measurement metrics", slog.String("error", err.Error())) + } } // Start Prometheus metrics endpoint @@ -54,24 +59,28 @@ func (c *Collector) Run(ctx context.Context) error { errChan := make(chan error, 2) // Wheresitup job creation and export - wg.Add(1) - go func() { - defer wg.Done() - if err := c.cfg.Wheresitup.Run(ctx, c.cfg.WheresitupSamplingInterval, c.cfg.DryRun, c.cfg.ProcessedJobsFile, c.cfg.StateDir); err != nil { - errChan <- fmt.Errorf("wheresitup collector error: %w", err) - cancel() // Cancel other goroutines on error - } - }() + if c.cfg.Wheresitup != nil { + wg.Add(1) + go func() { + defer wg.Done() + if err := c.cfg.Wheresitup.Run(ctx, c.cfg.WheresitupSamplingInterval, c.cfg.DryRun, c.cfg.ProcessedJobsFile, c.cfg.StateDir); err != nil { + errChan <- fmt.Errorf("wheresitup collector error: %w", err) + cancel() // Cancel other goroutines on error + } + }() + } // Ripe Atlas measurement creation and export - wg.Add(1) - go func() { - defer wg.Done() - if err := c.cfg.RipeAtlas.Run(ctx, c.cfg.DryRun, c.cfg.ProbesPerLocation, c.cfg.StateDir, c.cfg.RipeAtlasSamplingInterval, c.cfg.RipeAtlasMeasurementInterval, c.cfg.RipeAtlasExportInterval); err != nil { - errChan <- fmt.Errorf("ripe atlas collector error: %w", err) - cancel() // Cancel other goroutines on error - } - }() + if c.cfg.RipeAtlas != nil { + wg.Add(1) + go func() { + defer wg.Done() + if err := c.cfg.RipeAtlas.Run(ctx, c.cfg.DryRun, c.cfg.ProbesPerLocation, c.cfg.StateDir, c.cfg.RipeAtlasSamplingInterval, c.cfg.RipeAtlasMeasurementInterval, c.cfg.RipeAtlasExportInterval); err != nil { + errChan <- fmt.Errorf("ripe atlas collector error: %w", err) + cancel() // Cancel other goroutines on error + } + }() + } // Wait for all goroutines to complete wg.Wait() diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/cloud.go b/controlplane/internet-latency-collector/internal/ripeatlas/cloud.go new file mode 100644 index 0000000000..7d12bcb5a1 --- /dev/null +++ b/controlplane/internet-latency-collector/internal/ripeatlas/cloud.go @@ -0,0 +1,46 @@ +package ripeatlas + +import ( + "context" + "log/slog" + + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/exporter" +) + +type CloudNode struct { + Code string + Cloud string + Latitude float64 + Longitude float64 + AtlasProbeIDs []int + PingTarget string +} + +func NewCloudCollector(logger *slog.Logger, exporter exporter.Exporter, env string, nodes []CloudNode) *Collector { + byCode := make(map[string]CloudNode, len(nodes)) + for _, node := range nodes { + byCode[node.Code] = node + } + + return &Collector{ + client: NewClient(logger), + log: logger, + exporter: exporter, + env: env, + probeToLocation: make(map[int]string), + cloudMode: true, + cloudNodes: byCode, + getLocationsFunc: func(_ context.Context) []collector.LocationMatch { + locations := make([]collector.LocationMatch, 0, len(nodes)) + for _, node := range nodes { + locations = append(locations, collector.LocationMatch{ + LocationCode: node.Code, + Latitude: node.Latitude, + Longitude: node.Longitude, + }) + } + return locations + }, + } +} diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/cloud_test.go b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_test.go new file mode 100644 index 0000000000..54000c4a45 --- /dev/null +++ b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_test.go @@ -0,0 +1,124 @@ +package ripeatlas + +import ( + "context" + "log/slog" + "path/filepath" + "testing" + + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" + "github.com/stretchr/testify/require" +) + +// cloudTestNodes returns two AWS regions; eu-west-1 sorts first, so it is the measurement target. +func cloudTestNodes() []CloudNode { + return []CloudNode{ + { + Code: "eu-west-1", + Cloud: "aws", + Latitude: 53.3498, + Longitude: -6.2603, + AtlasProbeIDs: []int{1000441}, + PingTarget: "3.248.0.0", + }, + { + Code: "us-east-1", + Cloud: "aws", + Latitude: 39.0438, + Longitude: -77.4874, + AtlasProbeIDs: []int{1000731, 1000732}, + PingTarget: "34.192.0.54", + }, + } +} + +func newCloudTestCollector(t *testing.T, log *slog.Logger, client clientInterface, env string, nodes []CloudNode) *Collector { + t.Helper() + + c := NewCloudCollector(log, nil, env, nodes) + c.client = client + return c +} + +// exchangeTestLocations returns three exchanges with two probes each, at distinct distances so +// probe selection is deterministic. +func exchangeTestLocations() []LocationProbeMatch { + return []LocationProbeMatch{ + { + LocationMatch: collector.LocationMatch{LocationCode: "nyc", Latitude: 40.7128, Longitude: -74.0060}, + NearbyProbes: []Probe{ + {ID: 100, Address: "1.1.1.1", Latitude: 40.7128, Longitude: -74.0060}, + {ID: 101, Address: "1.1.1.2", Latitude: 41.0000, Longitude: -74.0060}, + }, + ProbeCount: 2, + }, + { + LocationMatch: collector.LocationMatch{LocationCode: "lon", Latitude: 51.5074, Longitude: -0.1278}, + NearbyProbes: []Probe{ + {ID: 200, Address: "2.2.2.1", Latitude: 51.5074, Longitude: -0.1278}, + {ID: 201, Address: "2.2.2.2", Latitude: 52.0000, Longitude: -0.1278}, + }, + ProbeCount: 2, + }, + { + LocationMatch: collector.LocationMatch{LocationCode: "ams", Latitude: 52.3676, Longitude: 4.9041}, + NearbyProbes: []Probe{ + {ID: 300, Address: "3.3.3.1", Latitude: 52.3676, Longitude: 4.9041}, + {ID: 301, Address: "3.3.3.2", Latitude: 53.0000, Longitude: 4.9041}, + }, + ProbeCount: 2, + }, + } +} + +func TestInternetLatency_RIPEAtlas_Cloud_DisabledByDefault(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + c := NewCollector(log, nil, "testnet", func(ctx context.Context) []collector.LocationMatch { + return []collector.LocationMatch{} + }) + + require.False(t, c.cloudMode, "NewCollector must not enable cloud mode") + require.Empty(t, c.cloudNodes, "NewCollector must not carry node file entries") + + measurementState := NewMeasurementState(filepath.Join(t.TempDir(), TimestampFileName)) + wanted := c.generateWantedMeasurements(exchangeTestLocations(), 1, measurementState) + + require.Len(t, wanted, 2, "three locations give two measurements, one per unordered pair") + + require.Equal(t, "ams", wanted[0].TargetLocationCode) + require.Equal(t, 300, wanted[0].TargetProbe.ID) + require.Equal(t, "3.3.3.1", wanted[0].TargetProbe.Address) + require.Len(t, wanted[0].SourceSpecs, 2) + require.Equal(t, "lon", wanted[0].SourceSpecs[0].LocationCode) + require.Equal(t, 200, wanted[0].SourceSpecs[0].Probe.ID) + require.Equal(t, "nyc", wanted[0].SourceSpecs[1].LocationCode) + require.Equal(t, 100, wanted[0].SourceSpecs[1].Probe.ID) + + require.Equal(t, "lon", wanted[1].TargetLocationCode) + require.Equal(t, 200, wanted[1].TargetProbe.ID) + require.Len(t, wanted[1].SourceSpecs, 1) + require.Equal(t, "nyc", wanted[1].SourceSpecs[0].LocationCode) + require.Equal(t, 100, wanted[1].SourceSpecs[0].Probe.ID) +} + +func TestInternetLatency_RIPEAtlas_Cloud_ConstructorEnablesMode(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + c := newCloudTestCollector(t, log, &MockClient{}, "mainnet-beta", cloudTestNodes()) + + require.True(t, c.cloudMode, "NewCloudCollector must enable cloud mode") + require.Len(t, c.cloudNodes, 2) + require.Equal(t, "3.248.0.0", c.cloudNodes["eu-west-1"].PingTarget) + require.Equal(t, []int{1000731, 1000732}, c.cloudNodes["us-east-1"].AtlasProbeIDs) + + locations := c.getLocationsFunc(t.Context()) + require.Len(t, locations, 2) + require.Equal(t, "eu-west-1", locations[0].LocationCode) + require.Equal(t, 53.3498, locations[0].Latitude) + require.Equal(t, "us-east-1", locations[1].LocationCode) +} diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/collector.go b/controlplane/internet-latency-collector/internal/ripeatlas/collector.go index ad95eab1f9..fdad083e98 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/collector.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/collector.go @@ -82,6 +82,8 @@ type Collector struct { probeToLocation map[int]string // Maps probe IDs to location codes mu sync.RWMutex // Protects probeToLocation map measurementState *MeasurementState // Shared state; initialized in Run() + cloudMode bool // Measures cloud regions from cloudNodes instead of exchanges + cloudNodes map[string]CloudNode } type MeasurementSpec struct {