internet-latency-collector: ping a fixed target address per measurement

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

@@ -0,0 +1,249 @@
+package ripeatlas
+
+import (
+	"context"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector"
+	"github.com/stretchr/testify/require"
+)
+
+// cloudTestLocationMatches returns the two cloud regions as location matches carrying their
+// pinned probes.
+func cloudTestLocationMatches() []LocationProbeMatch {
+	return []LocationProbeMatch{
+		{
+			LocationMatch: collector.LocationMatch{LocationCode: "eu-west-1", Latitude: 53.3498, Longitude: -6.2603},
+			NearbyProbes: []Probe{
+				{ID: 1000441, Address: "10.0.0.1", Latitude: 53.3498, Longitude: -6.2603},
+			},
+			ProbeCount: 1,
+		},
+		{
+			LocationMatch: collector.LocationMatch{LocationCode: "us-east-1", Latitude: 39.0438, Longitude: -77.4874},
+			NearbyProbes: []Probe{
+				{ID: 1000731, Address: "10.0.0.2", Latitude: 39.0438, Longitude: -77.4874},
+			},
+			ProbeCount: 1,
+		},
+	}
+}
+
+func TestInternetLatency_RIPEAtlas_TargetAddress_ExchangeUsesTargetProbeAddress(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{}
+	})
+
+	measurementState := NewMeasurementState(filepath.Join(t.TempDir(), TimestampFileName))
+	wanted := c.generateWantedMeasurements(exchangeTestLocations(), 1, measurementState)
+
+	require.Len(t, wanted, 2)
+	require.Equal(t, "3.3.3.1", wanted[0].TargetAddress, "exchange mode pings the target probe address")
+	require.Equal(t, wanted[0].TargetProbe.Address, wanted[0].TargetAddress)
+	require.Equal(t, "2.2.2.1", wanted[1].TargetAddress)
+	require.Equal(t, wanted[1].TargetProbe.Address, wanted[1].TargetAddress)
+}
+
+func TestInternetLatency_RIPEAtlas_TargetAddress_CloudUsesNodeFilePingTarget(t *testing.T) {
+	t.Parallel()
+
+	log := logger.With("test", t.Name())
+
+	var createdMeasurements []MeasurementRequest
+	var mu sync.Mutex
+
+	mockClient := &MockClient{
+		GetAllMeasurementsFunc: func(ctx context.Context, env string) ([]Measurement, error) {
+			return []Measurement{}, nil
+		},
+		CreateMeasurementFunc: func(ctx context.Context, request MeasurementRequest) (*MeasurementResponse, error) {
+			mu.Lock()
+			createdMeasurements = append(createdMeasurements, request)
+			measurementID := 3000 + len(createdMeasurements)
+			mu.Unlock()
+			return &MeasurementResponse{Measurements: []int{measurementID}}, nil
+		},
+	}
+
+	stateDir := t.TempDir()
+	c := newCloudTestCollector(t, log, mockClient, "mainnet-beta", cloudTestNodes())
+
+	measurementState := NewMeasurementState(filepath.Join(stateDir, TimestampFileName))
+	wanted := c.generateWantedMeasurements(cloudTestLocationMatches(), 1, measurementState)
+
+	require.Len(t, wanted, 1, "two regions give one measurement")
+	require.Equal(t, "eu-west-1", wanted[0].TargetLocationCode)
+	require.Equal(t, "3.248.0.0", wanted[0].TargetAddress, "cloud mode pings the node file ping_target")
+	require.NotEqual(t, wanted[0].TargetProbe.Address, wanted[0].TargetAddress,
+		"the far-end probe address must not be the ping target")
+
+	err := c.configureMeasurements(t.Context(), cloudTestLocationMatches(), false, 1, stateDir, 10*time.Minute)
+	require.NoError(t, err)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Len(t, createdMeasurements, 1)
+	require.Equal(t, "3.248.0.0", createdMeasurements[0].Definitions[0].Target,
+		"the created measurement must ping the fixed region address")
+}
+
+func TestInternetLatency_RIPEAtlas_TargetAddress_CloudSkipsLocationWithoutPingTarget(t *testing.T) {
+	t.Parallel()
+
+	log := logger.With("test", t.Name())
+
+	nodes := cloudTestNodes()
+	nodes[0].PingTarget = ""
+
+	c := newCloudTestCollector(t, log, &MockClient{}, "mainnet-beta", nodes)
+
+	measurementState := NewMeasurementState(filepath.Join(t.TempDir(), TimestampFileName))
+	wanted := c.generateWantedMeasurements(cloudTestLocationMatches(), 1, measurementState)
+
+	require.Empty(t, wanted, "a target region with no ping target yields no measurement")
+}
+
+func TestInternetLatency_RIPEAtlas_TargetAddress_ChangeRecreatesMeasurement(t *testing.T) {
+	t.Parallel()
+
+	log := logger.With("test", t.Name())
+
+	var createdMeasurements []MeasurementRequest
+	var stoppedMeasurements []int
+	var mu sync.Mutex
+
+	existing := []Measurement{
+		{
+			ID:          1001,
+			Description: "DoubleZero [testnet] to ams probe 300",
+			Target:      "3.3.3.1",
+			Status: struct {
+				Name string `json:"name"`
+				ID   int    `json:"id"`
+			}{Name: "Ongoing"},
+			Type: "ping",
+		},
+	}
+
+	mockClient := &MockClient{
+		GetAllMeasurementsFunc: func(ctx context.Context, env string) ([]Measurement, error) {
+			return existing, nil
+		},
+		CreateMeasurementFunc: func(ctx context.Context, request MeasurementRequest) (*MeasurementResponse, error) {
+			mu.Lock()
+			createdMeasurements = append(createdMeasurements, request)
+			measurementID := 4000 + len(createdMeasurements)
+			mu.Unlock()
+			return &MeasurementResponse{Measurements: []int{measurementID}}, nil
+		},
+		StopMeasurementFunc: func(ctx context.Context, measurementID int) error {
+			mu.Lock()
+			stoppedMeasurements = append(stoppedMeasurements, measurementID)
+			mu.Unlock()
+			return nil
+		},
+		GetMeasurementResultsIncrementalFunc: func(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) {
+			return []any{}, nil
+		},
+	}
+
+	stateDir := t.TempDir()
+	c := &Collector{client: mockClient, log: log, env: "testnet", getLocationsFunc: func(ctx context.Context) []collector.LocationMatch {
+		return []collector.LocationMatch{}
+	}}
+
+	// The stored address differs from the probe's current address, so ams is stale.
+	c.measurementState = NewMeasurementState(filepath.Join(stateDir, TimestampFileName))
+	c.measurementState.SetMetadata(1001, MeasurementMeta{
+		TargetLocation: "ams",
+		TargetProbeID:  300,
+		TargetAddress:  "3.3.3.9",
+		Sources: []SourceProbeMeta{
+			{LocationCode: "lon", ProbeID: 200, LastResponseAt: time.Now().Unix()},
+			{LocationCode: "nyc", ProbeID: 100, LastResponseAt: time.Now().Unix()},
+		},
+		CreatedAt:    time.Now().Unix() - 60,
+		LastExportAt: time.Now().Unix(),
+	})
+
+	err := c.configureMeasurements(t.Context(), exchangeTestLocations(), false, 1, stateDir, 10*time.Minute)
+	require.NoError(t, err)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Contains(t, stoppedMeasurements, 1001, "a changed target address must recreate the measurement")
+	require.NotEmpty(t, createdMeasurements)
+	require.Equal(t, "3.3.3.1", createdMeasurements[0].Definitions[0].Target)
+}
+
+func TestInternetLatency_RIPEAtlas_TargetAddress_EmptyStoredValueIsNotAChange(t *testing.T) {
+	t.Parallel()
+
+	log := logger.With("test", t.Name())
+
+	var stoppedMeasurements []int
+	var mu sync.Mutex
+
+	existing := []Measurement{
+		{
+			ID:          1001,
+			Description: "DoubleZero [testnet] to ams probe 300",
+			Target:      "3.3.3.1",
+			Status: struct {
+				Name string `json:"name"`
+				ID   int    `json:"id"`
+			}{Name: "Ongoing"},
+			Type: "ping",
+		},
+	}
+
+	mockClient := &MockClient{
+		GetAllMeasurementsFunc: func(ctx context.Context, env string) ([]Measurement, error) {
+			return existing, nil
+		},
+		StopMeasurementFunc: func(ctx context.Context, measurementID int) error {
+			mu.Lock()
+			stoppedMeasurements = append(stoppedMeasurements, measurementID)
+			mu.Unlock()
+			return nil
+		},
+		GetMeasurementResultsIncrementalFunc: func(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) {
+			return []any{}, nil
+		},
+	}
+
+	stateDir := t.TempDir()
+	c := &Collector{client: mockClient, log: log, env: "testnet", getLocationsFunc: func(ctx context.Context) []collector.LocationMatch {
+		return []collector.LocationMatch{}
+	}}
+
+	// State written before the field existed carries no address.
+	c.measurementState = NewMeasurementState(filepath.Join(stateDir, TimestampFileName))
+	c.measurementState.SetMetadata(1001, MeasurementMeta{
+		TargetLocation: "ams",
+		TargetProbeID:  300,
+		Sources: []SourceProbeMeta{
+			{LocationCode: "lon", ProbeID: 200, LastResponseAt: time.Now().Unix()},
+			{LocationCode: "nyc", ProbeID: 100, LastResponseAt: time.Now().Unix()},
+		},
+		CreatedAt:    time.Now().Unix() - 60,
+		LastExportAt: time.Now().Unix(),
+	})
+
+	err := c.configureMeasurements(t.Context(), exchangeTestLocations(), false, 1, stateDir, 10*time.Minute)
+	require.NoError(t, err)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	require.Empty(t, stoppedMeasurements, "an unset stored target address must not force a recreation")
+}

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

@@ -90,6 +90,7 @@ type MeasurementSpec struct {
 	TargetLocation     string
 	TargetLocationCode string
 	TargetProbe        Probe
+	TargetAddress      string
 	SourceSpecs        []SourceSpec
 }
 
@@ -995,6 +996,17 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [
 				continue
 			}
 
+			// An empty stored target address predates the field and is not a mismatch.
+			if meta.TargetAddress != "" && meta.TargetAddress != wanted.TargetAddress {
+				c.log.Info("Measurement has outdated target address, marking for recreation",
+					slog.Int("measurement_id", existing.ID),
+					slog.String("target", wanted.TargetLocationCode),
+					slog.String("existing_target_address", meta.TargetAddress),
+					slog.String("wanted_target_address", wanted.TargetAddress))
+				toCreate = append(toCreate, wanted)
+				continue
+			}
+
 			// Check if target probe has changed
 			targetProbeChanged := meta.TargetProbeID != wanted.TargetProbe.ID
 			if targetProbeChanged {
@@ -1050,6 +1062,11 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [
 		if existing, exists := existingByTarget[wanted.TargetLocationCode]; exists {
 			meta, hasMeta := measurementState.GetMetadata(existing.ID)
 			if hasMeta {
+				if meta.TargetAddress != "" && meta.TargetAddress != wanted.TargetAddress {
+					measurementsToRecreate[wanted.TargetLocationCode] = true
+					continue
+				}
+
 				// Check if target probe has changed
 				if meta.TargetProbeID != wanted.TargetProbe.ID {
 					measurementsToRecreate[wanted.TargetLocationCode] = true
@@ -1200,6 +1217,7 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [
 			c.log.Info("Would create measurement (dry run)",
 				slog.String("target_location", spec.TargetLocation),
 				slog.Int("target_probe", spec.TargetProbe.ID),
+				slog.String("target_address", spec.TargetAddress),
 				slog.Int("source_count", len(spec.SourceSpecs)))
 		} else {
 			// Use simplified description without source list
@@ -1236,7 +1254,7 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [
 						Interval:       int(samplingInterval.Seconds()),
 						Packets:        1,
 						PacketInterval: 1000, // Delay between packets; only matters when Packets > 1
-						Target:         spec.TargetProbe.Address,
+						Target:         spec.TargetAddress,
 						Description:    description,
 						Tags:           tags,
 					},
@@ -1268,6 +1286,7 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [
 				meta := MeasurementMeta{
 					TargetLocation: spec.TargetLocationCode,
 					TargetProbeID:  spec.TargetProbe.ID,
+					TargetAddress:  spec.TargetAddress,
 					Sources:        sources,
 					CreatedAt:      time.Now().Unix(),
 				}
@@ -1426,6 +1445,17 @@ func (c *Collector) generateWantedMeasurements(locationMatches []LocationProbeMa
 		}
 		targetProbe := targetProbes[0]
 
+		targetAddress := targetProbe.Address
+		if c.cloudMode {
+			node, ok := c.cloudNodes[targetLocation.LocationCode]
+			if !ok || node.PingTarget == "" {
+				c.log.Warn("No ping target for location, skipping measurement",
+					slog.String("location", targetLocation.LocationCode))
+				continue
+			}
+			targetAddress = node.PingTarget
+		}
+
 		// Collect source probes from all other locations
 		// Since we're iterating in alphabetical order and only need to measure once between any pair,
 		// we only include sources from locations that come after this target in the alphabet
@@ -1466,6 +1496,7 @@ func (c *Collector) generateWantedMeasurements(locationMatches []LocationProbeMa
 				TargetLocation:     targetLocation.LocationCode,
 				TargetLocationCode: targetLocation.LocationCode,
 				TargetProbe:        targetProbe,
+				TargetAddress:      targetAddress,
 				SourceSpecs:        sourceSpecs,
 			})
 		}

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

@@ -32,6 +32,7 @@ type UnresponsiveProbeEntry struct {
 type MeasurementMeta struct {
 	TargetLocation string            `json:"target_location"`
 	TargetProbeID  int               `json:"target_probe_id"`
+	TargetAddress  string            `json:"target_address,omitempty"`
 	Sources        []SourceProbeMeta `json:"sources"`
 	CreatedAt      int64             `json:"created_at"`
 	LastExportAt   int64             `json:"last_export_at,omitempty"`