From dd4db15afc8f228165042594287715580f70cb17 Mon Sep 17 00:00:00 2001 From: Thijs van Emmerik Date: Fri, 11 Sep 2026 09:12:43 +0000 Subject: [PATCH] internet-latency-collector: separate cloud mode measurement identity Cloud mode gets its own measurement description prefix, its own RIPE tag, its own tag filter and its own state file. Sharing any of them with the exchange collector lets one process reconcile against the other's measurements and stop all of them, because a measurement with no metadata in the local state file is treated as unwanted. Cloud mode has no target probe, so a source is blamed for delivering nothing only while its own measurement is still exporting, and a source that has never delivered is judged against its measurement's creation time. Recreation is now decided from the create list in one pass instead of a second copy of the same checks, and a changed target probe is reported before a changed target address. --- .../internal/ripeatlas/client.go | 8 +- .../internal/ripeatlas/cloud.go | 57 +++ .../internal/ripeatlas/cloud_blame_test.go | 335 ++++++++++++++++++ .../internal/ripeatlas/cloud_identity_test.go | 116 ++++++ .../internal/ripeatlas/cloud_target_test.go | 81 +++++ .../internal/ripeatlas/collector.go | 223 ++++++------ 6 files changed, 702 insertions(+), 118 deletions(-) create mode 100644 controlplane/internet-latency-collector/internal/ripeatlas/cloud_blame_test.go create mode 100644 controlplane/internet-latency-collector/internal/ripeatlas/cloud_identity_test.go diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/client.go b/controlplane/internet-latency-collector/internal/ripeatlas/client.go index ef3867ee0f..e5afb11129 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/client.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/client.go @@ -300,15 +300,15 @@ func (c *Client) CreateMeasurement(ctx context.Context, request MeasurementReque return &measurementResponse, nil } -func (c *Client) GetAllMeasurements(ctx context.Context, env string) ([]Measurement, error) { - if env == "" { - return nil, fmt.Errorf("env parameter is required") +func (c *Client) GetAllMeasurements(ctx context.Context, tag string) ([]Measurement, error) { + if tag == "" { + return nil, fmt.Errorf("tag parameter is required") } allMeasurements := []Measurement{} // Include both Ongoing and Scheduled statuses to catch newly created measurements // Status values: 1=Scheduled, 2=Ongoing - endpoint := fmt.Sprintf("/measurements/my/?status=Ongoing,Scheduled&tags=%s", env) + endpoint := fmt.Sprintf("/measurements/my/?status=Ongoing,Scheduled&tags=%s", tag) for { resp, err := c.makeRequest(ctx, endpoint) diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/cloud.go b/controlplane/internet-latency-collector/internal/ripeatlas/cloud.go index 7d12bcb5a1..5c80f19434 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/cloud.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/cloud.go @@ -3,11 +3,22 @@ package ripeatlas import ( "context" "log/slog" + "strings" "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/exporter" ) +const ( + CloudTimestampFileName = "ripe_atlas_cloud_timestamps.json" + + exchangeDescriptionPrefix = "DoubleZero " + cloudDescriptionPrefix = "DoubleZero Cloud " + + cloudTagSuffix = "-cloud" + cloudMeasurementTag = "doublezero-cloud" +) + type CloudNode struct { Code string Cloud string @@ -44,3 +55,49 @@ func NewCloudCollector(logger *slog.Logger, exporter exporter.Exporter, env stri }, } } + +func (c *Collector) timestampFileName() string { + if c.cloudMode { + return CloudTimestampFileName + } + return TimestampFileName +} + +func (c *Collector) descriptionPrefix() string { + if c.cloudMode { + return cloudDescriptionPrefix + } + return exchangeDescriptionPrefix +} + +func (c *Collector) measurementTag() string { + if c.cloudMode { + return c.env + cloudTagSuffix + } + return c.env +} + +// The cloud prefix extends the exchange prefix, so exchange mode has to reject it explicitly. +func (c *Collector) ownsDescription(description string) bool { + if !strings.HasPrefix(description, c.descriptionPrefix()) { + return false + } + return c.cloudMode || !strings.HasPrefix(description, cloudDescriptionPrefix) +} + +// Exchange descriptions end "to probe ", cloud descriptions "to target
". +func (c *Collector) targetLocationFromDescription(description string) (string, bool) { + parts := strings.Split(description, " to ") + if len(parts) != 2 { + return "", false + } + marker := " probe" + if c.cloudMode { + marker = " target" + } + idx := strings.Index(parts[1], marker) + if idx == -1 { + return "", false + } + return parts[1][:idx], true +} diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/cloud_blame_test.go b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_blame_test.go new file mode 100644 index 0000000000..0a57c07f20 --- /dev/null +++ b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_blame_test.go @@ -0,0 +1,335 @@ +package ripeatlas + +import ( + "context" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" + "github.com/stretchr/testify/require" +) + +// cloudMeasurement is the eu-west-1 measurement cloud mode creates for cloudTestNodes. +func cloudMeasurement() Measurement { + return Measurement{ + ID: 8001, + Description: "DoubleZero Cloud [mainnet-beta] to eu-west-1 target 3.248.0.0", + Target: "3.248.0.0", + Status: struct { + Name string `json:"name"` + ID int `json:"id"` + }{Name: "Ongoing"}, + Type: "ping", + } +} + +func TestInternetLatency_RIPEAtlas_CloudBlame_StaleMeasurementBlamesNoProbe(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + var stoppedMeasurements []int + var mu sync.Mutex + + twoHoursAgo := time.Now().Unix() - 7200 + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, tag string) ([]Measurement, error) { + return []Measurement{cloudMeasurement()}, 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 := newCloudTestCollector(t, log, mockClient, "mainnet-beta", cloudTestNodes()) + + // The measurement has exported nothing for two hours, and its stored target probe is + // also the live source probe for us-east-1 in this same run. + c.measurementState = NewMeasurementState(filepath.Join(stateDir, CloudTimestampFileName)) + c.measurementState.SetMetadata(8001, MeasurementMeta{ + TargetLocation: "eu-west-1", + TargetProbeID: 1000731, + TargetAddress: "3.248.0.0", + Sources: []SourceProbeMeta{ + {LocationCode: "us-east-1", ProbeID: 1000731, LastResponseAt: time.Now().Unix()}, + }, + CreatedAt: twoHoursAgo - 3600, + LastExportAt: twoHoursAgo, + }) + + err := c.configureMeasurements(t.Context(), cloudTestLocationMatches(), false, 1, stateDir, 10*time.Minute) + require.NoError(t, err) + + require.False(t, c.measurementState.IsProbeUnresponsive(1000731), + "cloud mode must not blame the stored target probe, it is a source elsewhere") + require.Empty(t, c.measurementState.GetUnresponsiveProbes(), + "a quiet cloud measurement must mark no probe unresponsive") + + mu.Lock() + defer mu.Unlock() + require.Empty(t, stoppedMeasurements, "the measurement still matches what is wanted") +} + +func TestInternetLatency_RIPEAtlas_CloudBlame_NeverStartedSourceIsMarkedAfterGrace(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, tag string) ([]Measurement, error) { + return []Measurement{cloudMeasurement()}, nil + }, + GetMeasurementResultsIncrementalFunc: func(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) { + return []any{}, nil + }, + } + + stateDir := t.TempDir() + c := newCloudTestCollector(t, log, mockClient, "mainnet-beta", cloudTestNodes()) + + // The measurement is exporting, but its one source has never produced a sample and was + // created well past the grace period. + c.measurementState = NewMeasurementState(filepath.Join(stateDir, CloudTimestampFileName)) + c.measurementState.SetMetadata(8001, MeasurementMeta{ + TargetLocation: "eu-west-1", + TargetProbeID: 1000731, + TargetAddress: "3.248.0.0", + Sources: []SourceProbeMeta{ + {LocationCode: "us-east-1", ProbeID: 1000731, LastResponseAt: 0}, + }, + CreatedAt: time.Now().Unix() - 10800, + LastExportAt: time.Now().Unix(), + }) + + err := c.configureMeasurements(t.Context(), cloudTestLocationMatches(), false, 1, stateDir, 10*time.Minute) + require.NoError(t, err) + + require.True(t, c.measurementState.IsProbeUnresponsive(1000731), + "a cloud source that has delivered nothing since creation must be blacklisted") +} + +func TestInternetLatency_RIPEAtlas_CloudBlame_NeverStartedSourceKeptInsideGrace(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, tag string) ([]Measurement, error) { + return []Measurement{cloudMeasurement()}, nil + }, + GetMeasurementResultsIncrementalFunc: func(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) { + return []any{}, nil + }, + } + + stateDir := t.TempDir() + c := newCloudTestCollector(t, log, mockClient, "mainnet-beta", cloudTestNodes()) + + // Created 90 minutes ago, inside the two-hour grace, so the source is still warming up. + c.measurementState = NewMeasurementState(filepath.Join(stateDir, CloudTimestampFileName)) + c.measurementState.SetMetadata(8001, MeasurementMeta{ + TargetLocation: "eu-west-1", + TargetProbeID: 1000731, + TargetAddress: "3.248.0.0", + Sources: []SourceProbeMeta{ + {LocationCode: "us-east-1", ProbeID: 1000731, LastResponseAt: 0}, + }, + CreatedAt: time.Now().Unix() - 5400, + LastExportAt: time.Now().Unix(), + }) + + err := c.configureMeasurements(t.Context(), cloudTestLocationMatches(), false, 1, stateDir, 10*time.Minute) + require.NoError(t, err) + + require.Empty(t, c.measurementState.GetUnresponsiveProbes(), + "a cloud source inside the grace period must not be blacklisted") +} + +func TestInternetLatency_RIPEAtlas_CloudBlame_ExchangeKeepsNeverStartedSource(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, tag string) ([]Measurement, error) { + return []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", + }}, nil + }, + GetMeasurementResultsIncrementalFunc: func(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) { + return []any{}, nil + }, + } + + stateDir := t.TempDir() + c := NewCollector(log, nil, "testnet", func(ctx context.Context) []collector.LocationMatch { + return []collector.LocationMatch{} + }) + c.client = mockClient + + c.measurementState = NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + c.measurementState.SetMetadata(1001, MeasurementMeta{ + TargetLocation: "ams", + TargetProbeID: 300, + TargetAddress: "3.3.3.1", + Sources: []SourceProbeMeta{ + {LocationCode: "lon", ProbeID: 200, LastResponseAt: 0}, + {LocationCode: "nyc", ProbeID: 100, LastResponseAt: 0}, + }, + CreatedAt: time.Now().Unix() - 10800, + LastExportAt: time.Now().Unix(), + }) + + err := c.configureMeasurements(t.Context(), exchangeTestLocations(), false, 1, stateDir, 10*time.Minute) + require.NoError(t, err) + + require.Empty(t, c.measurementState.GetUnresponsiveProbes(), + "exchange mode still waits for an export cycle to populate a source's last response") +} + +func TestInternetLatency_RIPEAtlas_CloudBlame_TargetRegionNeedsNoResponsiveProbe(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + c := newCloudTestCollector(t, log, &MockClient{}, "mainnet-beta", cloudTestNodes()) + + measurementState := NewMeasurementState(filepath.Join(t.TempDir(), CloudTimestampFileName)) + measurementState.AddUnresponsiveProbe(1000441) + + wanted := c.generateWantedMeasurements(cloudTestLocationMatches(), 1, measurementState) + + require.Len(t, wanted, 1, "nothing pings the target region, so its own probe need not be live") + require.Equal(t, "eu-west-1", wanted[0].TargetLocationCode) + require.Equal(t, "3.248.0.0", wanted[0].TargetAddress) + require.Len(t, wanted[0].SourceSpecs, 1) + require.Equal(t, 1000731, wanted[0].SourceSpecs[0].Probe.ID) +} + +func TestInternetLatency_RIPEAtlas_CloudBlame_ExchangeTargetStillNeedsAResponsiveProbe(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)) + measurementState.AddUnresponsiveProbe(300) + measurementState.AddUnresponsiveProbe(301) + + wanted := c.generateWantedMeasurements(exchangeTestLocations(), 1, measurementState) + + require.Len(t, wanted, 1, "ams has no live probe, so it drops out as a target") + require.Equal(t, "lon", wanted[0].TargetLocationCode) +} + +// cloudFleetNodes gives three regions, each with a spare probe to rotate to. +func cloudFleetNodes() []CloudNode { + return []CloudNode{ + {Code: "eu-west-1", Cloud: "aws", Latitude: 53.3498, Longitude: -6.2603, + AtlasProbeIDs: []int{1000441, 1000442}, 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"}, + {Code: "us-west-2", Cloud: "aws", Latitude: 45.8399, Longitude: -119.7006, + AtlasProbeIDs: []int{1000901, 1000902}, PingTarget: "52.32.0.0"}, + } +} + +func cloudFleetLocations() []LocationProbeMatch { + return []LocationProbeMatch{ + {LocationMatch: collector.LocationMatch{LocationCode: "eu-west-1", Latitude: 53.3498, Longitude: -6.2603}}, + {LocationMatch: collector.LocationMatch{LocationCode: "us-east-1", Latitude: 39.0438, Longitude: -77.4874}}, + {LocationMatch: collector.LocationMatch{LocationCode: "us-west-2", Latitude: 45.8399, Longitude: -119.7006}}, + } +} + +// cloudFleetMeasurements is every measurement three regions produce: eu-west-1 sourced from the +// other two, us-east-1 sourced from us-west-2. +func cloudFleetMeasurements() []Measurement { + ongoing := struct { + Name string `json:"name"` + ID int `json:"id"` + }{Name: "Ongoing"} + + return []Measurement{ + { + ID: 8001, + Description: "DoubleZero Cloud [mainnet-beta] to eu-west-1 target 3.248.0.0", + Target: "3.248.0.0", + Status: ongoing, + Type: "ping", + }, + { + ID: 8002, + Description: "DoubleZero Cloud [mainnet-beta] to us-east-1 target 34.192.0.54", + Target: "34.192.0.54", + Status: ongoing, + Type: "ping", + }, + } +} + +func TestInternetLatency_RIPEAtlas_CloudBlame_LiveMeasurementBlamesItsDarkSource(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, tag string) ([]Measurement, error) { + return cloudFleetMeasurements(), nil + }, + } + + stateDir := t.TempDir() + c := newCloudTestCollector(t, log, mockClient, "mainnet-beta", cloudFleetNodes()) + + // eu-west-1 is exporting on us-east-1's samples while us-west-2 has delivered nothing to + // it. us-east-1 has only us-west-2 to draw on, so it exports nothing at all. + now := time.Now().Unix() + c.measurementState = NewMeasurementState(filepath.Join(stateDir, CloudTimestampFileName)) + c.measurementState.SetMetadata(8001, MeasurementMeta{ + TargetLocation: "eu-west-1", + TargetAddress: "3.248.0.0", + Sources: []SourceProbeMeta{ + {LocationCode: "us-east-1", ProbeID: 1000731, LastResponseAt: now}, + {LocationCode: "us-west-2", ProbeID: 1000901, LastResponseAt: 0}, + }, + CreatedAt: now - 10800, + LastExportAt: now, + }) + c.measurementState.SetMetadata(8002, MeasurementMeta{ + TargetLocation: "us-east-1", + TargetAddress: "34.192.0.54", + Sources: []SourceProbeMeta{ + {LocationCode: "us-west-2", ProbeID: 1000901, LastResponseAt: 0}, + }, + CreatedAt: now - 10800, + }) + + err := c.configureMeasurements(t.Context(), cloudFleetLocations(), false, 1, stateDir, 10*time.Minute) + require.NoError(t, err) + + require.Equal(t, []int{1000901}, c.measurementState.GetUnresponsiveProbes(), + "the one source that has delivered nothing to a measurement that is exporting must be blamed") + require.False(t, c.measurementState.IsProbeUnresponsive(1000731), + "a source that is delivering must not be blamed") +} diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/cloud_identity_test.go b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_identity_test.go new file mode 100644 index 0000000000..e05c0386a3 --- /dev/null +++ b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_identity_test.go @@ -0,0 +1,116 @@ +package ripeatlas + +import ( + "context" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector" + "github.com/stretchr/testify/require" +) + +func TestInternetLatency_RIPEAtlas_CloudIdentity_ModeNames(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + exchange := NewCollector(log, nil, "mainnet-beta", func(ctx context.Context) []collector.LocationMatch { + return []collector.LocationMatch{} + }) + cloud := newCloudTestCollector(t, log, &MockClient{}, "mainnet-beta", cloudTestNodes()) + + require.Equal(t, TimestampFileName, exchange.timestampFileName()) + require.Equal(t, "mainnet-beta", exchange.measurementTag()) + require.Equal(t, "DoubleZero ", exchange.descriptionPrefix()) + + require.Equal(t, CloudTimestampFileName, cloud.timestampFileName()) + require.Equal(t, "mainnet-beta-cloud", cloud.measurementTag()) + require.Equal(t, "DoubleZero Cloud ", cloud.descriptionPrefix()) + + const exchangeDescription = "DoubleZero [mainnet-beta] to xams probe 6626" + const cloudDescription = "DoubleZero Cloud [mainnet-beta] to eu-west-1 target 3.248.0.0" + + require.True(t, exchange.ownsDescription(exchangeDescription)) + require.False(t, exchange.ownsDescription(cloudDescription)) + require.True(t, cloud.ownsDescription(cloudDescription)) + require.False(t, cloud.ownsDescription(exchangeDescription)) + + location, ok := exchange.targetLocationFromDescription(exchangeDescription) + require.True(t, ok) + require.Equal(t, "xams", location) + + location, ok = cloud.targetLocationFromDescription(cloudDescription) + require.True(t, ok) + require.Equal(t, "eu-west-1", location) +} + +func TestInternetLatency_RIPEAtlas_CloudIdentity_IgnoresExchangeMeasurements(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + var requestedTags []string + var createdMeasurements []MeasurementRequest + var stoppedMeasurements []int + var mu sync.Mutex + + exchangeMeasurement := Measurement{ + ID: 1001, + Description: "DoubleZero [mainnet-beta] to xams probe 6626", + Target: "84.38.236.1", + Status: struct { + Name string `json:"name"` + ID int `json:"id"` + }{Name: "Ongoing"}, + Type: "ping", + } + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, tag string) ([]Measurement, error) { + mu.Lock() + requestedTags = append(requestedTags, tag) + mu.Unlock() + return []Measurement{exchangeMeasurement}, nil + }, + CreateMeasurementFunc: func(ctx context.Context, request MeasurementRequest) (*MeasurementResponse, error) { + mu.Lock() + createdMeasurements = append(createdMeasurements, request) + measurementID := 7000 + 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 + }, + } + + stateDir := t.TempDir() + c := newCloudTestCollector(t, log, mockClient, "mainnet-beta", cloudTestNodes()) + + err := c.configureMeasurements(t.Context(), cloudTestLocationMatches(), false, 1, stateDir, 10*time.Minute) + require.NoError(t, err) + + mu.Lock() + defer mu.Unlock() + + require.NotEmpty(t, requestedTags) + for _, tag := range requestedTags { + require.Equal(t, "mainnet-beta-cloud", tag, "cloud mode must filter on its own RIPE Atlas tag") + } + + require.Empty(t, stoppedMeasurements, "an exchange measurement must never be stopped by cloud mode") + + require.Len(t, createdMeasurements, 1) + require.Equal(t, "DoubleZero Cloud [mainnet-beta] to eu-west-1 target 3.248.0.0", + createdMeasurements[0].Definitions[0].Description) + require.ElementsMatch(t, []string{"mainnet-beta-cloud", "doublezero-cloud"}, + createdMeasurements[0].Definitions[0].Tags) + + require.FileExists(t, filepath.Join(stateDir, CloudTimestampFileName)) + require.NoFileExists(t, filepath.Join(stateDir, TimestampFileName)) +} diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/cloud_target_test.go b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_target_test.go index ae7032103a..47e9481f6b 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/cloud_target_test.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/cloud_target_test.go @@ -247,3 +247,84 @@ func TestInternetLatency_RIPEAtlas_TargetAddress_EmptyStoredValueIsNotAChange(t require.Empty(t, stoppedMeasurements, "an unset stored target address must not force a recreation") } + +func TestInternetLatency_RIPEAtlas_TargetAddress_ProbeChangeStillRecreatesMeasurement(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 301", + Target: "3.3.3.2", + Status: struct { + Name string `json:"name"` + ID int `json:"id"` + }{Name: "Ongoing"}, + Type: "ping", + }, + } + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, tag string) ([]Measurement, error) { + return existing, nil + }, + CreateMeasurementFunc: func(ctx context.Context, request MeasurementRequest) (*MeasurementResponse, error) { + mu.Lock() + createdMeasurements = append(createdMeasurements, request) + measurementID := 5000 + 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{} + }} + + // State written before the address field existed, so only the target probe can differ: + // 301 is stored, 300 is the nearest responsive probe for ams. + c.measurementState = NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + c.measurementState.SetMetadata(1001, MeasurementMeta{ + TargetLocation: "ams", + TargetProbeID: 301, + 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 probe must recreate the measurement") + + var amsCreated []string + for _, m := range createdMeasurements { + if m.Definitions[0].Target == "3.3.3.1" { + amsCreated = append(amsCreated, m.Definitions[0].Description) + } + } + require.Equal(t, []string{"DoubleZero [testnet] to ams probe 300"}, amsCreated, + "ams is recreated exactly once, against its current probe") +} diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/collector.go b/controlplane/internet-latency-collector/internal/ripeatlas/collector.go index 91e9d9cc73..8f949e751f 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/collector.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/collector.go @@ -7,7 +7,6 @@ import ( "os" "path/filepath" "sort" - "strings" "sync" "time" @@ -33,6 +32,10 @@ const ( // tuning this reporting window cannot perturb the marking path. sourceSampleGracePeriod = 2 * time.Hour + // neverStartedRotationGrace is how long after its measurement was created a cloud source + // probe may deliver nothing before it is marked unresponsive and replaced. + neverStartedRotationGrace = 2 * time.Hour + // maxSourcesWithoutSamplesLogged caps the per-cycle sample of sources named in the // log. The full count is on the metric; a widespread outage should not emit hundreds // of identifiers every hour. @@ -45,7 +48,7 @@ type clientInterface interface { GetProbesInRadius(ctx context.Context, latitude, longitude float64, radiusKm int, anchorsOnly bool) ([]Probe, error) GetProbesForLocations(ctx context.Context, locations []LocationProbeMatch) ([]LocationProbeMatch, error) CreateMeasurement(ctx context.Context, request MeasurementRequest) (*MeasurementResponse, error) - GetAllMeasurements(ctx context.Context, env string) ([]Measurement, error) + GetAllMeasurements(ctx context.Context, tag string) ([]Measurement, error) GetMeasurementResultsIncremental(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) StopMeasurement(ctx context.Context, measurementID int) error GetCreditBalance(ctx context.Context) (float64, error) @@ -123,7 +126,7 @@ func (c *Collector) InitializeCreditBalance(ctx context.Context) error { } func (c *Collector) InitializeMeasurementMetrics(stateDir string) error { - timestampFile := filepath.Join(stateDir, TimestampFileName) + timestampFile := filepath.Join(stateDir, c.timestampFileName()) measurementState := NewMeasurementState(timestampFile) if err := measurementState.Load(); err != nil { @@ -240,7 +243,7 @@ func (c *Collector) parseLatencyFromResult(result any) (time.Duration, time.Time func (c *Collector) ClearAllMeasurements(ctx context.Context) error { c.log.Info("Retrieving all measurements") - measurements, err := c.client.GetAllMeasurements(ctx, c.env) + measurements, err := c.client.GetAllMeasurements(ctx, c.measurementTag()) if err != nil { return collector.NewAPIError("get_measurements", "failed to get measurements", err) } @@ -262,8 +265,8 @@ func (c *Collector) ClearAllMeasurements(ctx context.Context) error { continue } - // Only clear DoubleZero measurements to avoid affecting other measurements - if !strings.Contains(measurement.Description, "DoubleZero") { + // Only clear this mode's measurements to avoid affecting other measurements + if !c.ownsDescription(measurement.Description) { c.log.Debug("Skipping measurement - not a DoubleZero measurement", slog.Int("measurement_id", measurement.ID), slog.String("description", measurement.Description)) @@ -302,7 +305,7 @@ func (c *Collector) ClearAllMeasurements(ctx context.Context) error { } func (c *Collector) ListMeasurements(ctx context.Context) error { - measurements, err := c.client.GetAllMeasurements(ctx, c.env) + measurements, err := c.client.GetAllMeasurements(ctx, c.measurementTag()) if err != nil { return collector.NewAPIError("get_measurements", "failed to get measurements", err) } @@ -394,14 +397,14 @@ func (c *Collector) ExportMeasurementResults(ctx context.Context, stateDir strin measurementState := c.measurementState if measurementState == nil { // Fallback for standalone/test usage without Run() - timestampFile := filepath.Join(stateDir, TimestampFileName) + timestampFile := filepath.Join(stateDir, c.timestampFileName()) measurementState = NewMeasurementState(timestampFile) if err := measurementState.Load(); err != nil { return err } } - measurements, err := c.client.GetAllMeasurements(ctx, c.env) + measurements, err := c.client.GetAllMeasurements(ctx, c.measurementTag()) if err != nil { return collector.NewAPIError("get_measurements", "failed to get measurements", err) } @@ -414,7 +417,7 @@ func (c *Collector) ExportMeasurementResults(ctx context.Context, stateDir strin // Filter for active DoubleZero measurements var activeMeasurements []Measurement for _, measurement := range measurements { - if strings.Contains(measurement.Description, "DoubleZero") && measurement.Status.Name != "Stopped" { + if c.ownsDescription(measurement.Description) && measurement.Status.Name != "Stopped" { activeMeasurements = append(activeMeasurements, measurement) } } @@ -794,7 +797,7 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ // Step 1: Get measurement state (shared instance from Run(), or fallback for tests) measurementState := c.measurementState if measurementState == nil { - timestampFile := filepath.Join(stateDir, TimestampFileName) + timestampFile := filepath.Join(stateDir, c.timestampFileName()) measurementState = NewMeasurementState(timestampFile) if err := measurementState.Load(); err != nil { c.log.Warn("Failed to load measurement state", slog.String("error", err.Error())) @@ -817,16 +820,16 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ wantedMeasurements := c.generateWantedMeasurements(locationMatches, probesPerLocation, measurementState) // Step 4: Get all existing measurements - existingMeasurements, err := c.client.GetAllMeasurements(ctx, c.env) + existingMeasurements, err := c.client.GetAllMeasurements(ctx, c.measurementTag()) if err != nil { c.log.Warn("Failed to get existing measurements", slog.String("error", err.Error())) existingMeasurements = []Measurement{} } - // Filter for DoubleZero measurements only + // Filter for this mode's measurements only var doubleZeroMeasurements []Measurement for _, m := range existingMeasurements { - if strings.HasPrefix(m.Description, "DoubleZero ") && m.Status.Name != "Stopped" { + if c.ownsDescription(m.Description) && m.Status.Name != "Stopped" { doubleZeroMeasurements = append(doubleZeroMeasurements, m) } } @@ -834,15 +837,8 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ // Step 3: Build map of existing measurements by target location existingByTarget := make(map[string]Measurement) for _, m := range doubleZeroMeasurements { - // Format: "DoubleZero [env] to TARGET probe Y" - parts := strings.Split(m.Description, " to ") - if len(parts) == 2 { - targetPart := parts[1] - // Extract location code (before " probe") - if idx := strings.Index(targetPart, " probe"); idx != -1 { - targetLocation := targetPart[:idx] - existingByTarget[targetLocation] = m - } + if targetLocation, ok := c.targetLocationFromDescription(m.Description); ok { + existingByTarget[targetLocation] = m } } @@ -873,6 +869,17 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ } if isStale { + if c.cloudMode { + // The target is an address, not a probe, so nothing here names a probe to blame. + c.log.Warn("Measurement has no recent exports", + slog.Int("measurement_id", measurement.ID), + slog.String("target_location", meta.TargetLocation), + slog.String("target_address", meta.TargetAddress), + slog.String("reason", reason), + slog.Time("created_at", time.Unix(meta.CreatedAt, 0)), + slog.Time("last_export_at", time.Unix(meta.LastExportAt, 0))) + continue + } c.log.Warn("Marking probe as unresponsive - no exports after 1 hour", slog.Int("measurement_id", measurement.ID), slog.Int("probe_id", meta.TargetProbeID), @@ -893,7 +900,12 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ // Skip measurements whose target is already marked unresponsive — // source probes in these measurements will have stale LastResponseAt // because the target isn't replying, not because the sources are broken - if measurementState.IsProbeUnresponsive(meta.TargetProbeID) { + if !c.cloudMode && measurementState.IsProbeUnresponsive(meta.TargetProbeID) { + continue + } + // Cloud mode has no target probe carrying that signal. A measurement that is + // not exporting makes every source of it look dark, and step 4 reported it. + if c.cloudMode && meta.LastExportAt < probeTimeout { continue } for _, source := range meta.Sources { @@ -907,14 +919,22 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ } // Skip probes where LastResponseAt hasn't been populated yet — // on first deploy, all existing source probes have 0 and need - // at least one export cycle to populate the field - if source.LastResponseAt == 0 { + // at least one export cycle to populate the field. Cloud mode instead + // judges a source that never delivers against its measurement's creation time. + neverStarted := source.LastResponseAt == 0 + pastGrace := meta.CreatedAt < currentTime-int64(neverStartedRotationGrace.Seconds()) + if neverStarted && !(c.cloudMode && pastGrace) { continue } // Last response was > 1 hour ago - if source.LastResponseAt < probeTimeout { + if neverStarted || source.LastResponseAt < probeTimeout { + message := "Marking source probe as unresponsive - no results after 1 hour" reason := "no_recent_responses" - c.log.Warn("Marking source probe as unresponsive - no results after 1 hour", + if neverStarted { + message = "Marking source probe as unresponsive - no results since creation" + reason = "never_responded" + } + c.log.Warn(message, slog.Int("measurement_id", measurement.ID), slog.Int("probe_id", source.ProbeID), slog.String("source_location", source.LocationCode), @@ -996,25 +1016,26 @@ 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", + // Check if target probe has changed. Cloud mode pings an address, so its + // measurement identity is the target address, not a probe. + targetProbeChanged := !c.cloudMode && meta.TargetProbeID != wanted.TargetProbe.ID + if targetProbeChanged { + c.log.Info("Measurement has outdated target probe, 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)) + slog.Int("existing_probe_id", meta.TargetProbeID), + slog.Int("wanted_probe_id", wanted.TargetProbe.ID)) toCreate = append(toCreate, wanted) continue } - // Check if target probe has changed - targetProbeChanged := meta.TargetProbeID != wanted.TargetProbe.ID - if targetProbeChanged { - c.log.Info("Measurement has outdated target probe, marking for recreation", + // 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.Int("existing_probe_id", meta.TargetProbeID), - slog.Int("wanted_probe_id", wanted.TargetProbe.ID)) + slog.String("existing_target_address", meta.TargetAddress), + slog.String("wanted_target_address", wanted.TargetAddress)) toCreate = append(toCreate, wanted) continue } @@ -1057,47 +1078,11 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ toRemove := []Measurement{} measurementsToRecreate := make(map[string]bool) - // First, identify measurements that need recreation due to outdated target or source probes - for _, wanted := range wantedMeasurements { - 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 - continue - } - - // Check if sources match (both location and probe ID) - existingSources := make(map[string]int) - for _, source := range meta.Sources { - existingSources[source.LocationCode] = source.ProbeID - } - - wantedSources := make(map[string]int) - for _, source := range wanted.SourceSpecs { - wantedSources[source.LocationCode] = source.Probe.ID - } - - sourcesMatch := len(existingSources) == len(wantedSources) - if sourcesMatch { - for loc, wantedProbeID := range wantedSources { - if existingProbeID, ok := existingSources[loc]; !ok || existingProbeID != wantedProbeID { - sourcesMatch = false - break - } - } - } - - if !sourcesMatch { - measurementsToRecreate[wanted.TargetLocationCode] = true - } - } + // A wanted measurement being created while one already exists for its target is a + // replacement, so the measurement it replaces has to be stopped. + for _, wanted := range toCreate { + if _, exists := existingByTarget[wanted.TargetLocationCode]; exists { + measurementsToRecreate[wanted.TargetLocationCode] = true } } @@ -1112,17 +1097,12 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ continue } - parts := strings.Split(measurement.Description, " to ") - if len(parts) == 2 { - targetPart := parts[1] - if idx := strings.Index(targetPart, " probe"); idx != -1 { - targetLocation := targetPart[:idx] - if !wantedTargets[targetLocation] { - toRemove = append(toRemove, measurement) - } else if measurementsToRecreate[targetLocation] { - // This measurement needs to be recreated due to outdated target or source probes - toRemove = append(toRemove, measurement) - } + if targetLocation, ok := c.targetLocationFromDescription(measurement.Description); ok { + if !wantedTargets[targetLocation] { + toRemove = append(toRemove, measurement) + } else if measurementsToRecreate[targetLocation] { + // This measurement needs to be recreated due to outdated target or source probes + toRemove = append(toRemove, measurement) } } } @@ -1222,7 +1202,15 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ } else { // Use simplified description without source list var description string - if c.env != "" { + if c.cloudMode { + if c.env != "" { + description = fmt.Sprintf("%s[%s] to %s target %s", + cloudDescriptionPrefix, c.env, spec.TargetLocationCode, spec.TargetAddress) + } else { + description = fmt.Sprintf("%sto %s target %s", + cloudDescriptionPrefix, spec.TargetLocationCode, spec.TargetAddress) + } + } else if c.env != "" { description = fmt.Sprintf("DoubleZero [%s] to %s probe %d", c.env, spec.TargetLocationCode, spec.TargetProbe.ID) } else { @@ -1233,9 +1221,13 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ // Build tags including environment if set var tags []string if c.env != "" { - tags = append(tags, c.env) + tags = append(tags, c.measurementTag()) + } + if c.cloudMode { + tags = append(tags, cloudMeasurementTag) + } else { + tags = append(tags, "doublezero") } - tags = append(tags, "doublezero") var probes []MeasurementProbe for _, source := range spec.SourceSpecs { @@ -1427,25 +1419,9 @@ func (c *Collector) generateWantedMeasurements(locationMatches []LocationProbeMa // Create one measurement per target location // Each measurement will ping from all other locations' probes to this target for targetIdx, targetLocation := range sortedLocations { - if len(targetLocation.NearbyProbes) == 0 { - continue - } - - responsiveProbes := filterResponsiveProbes(targetLocation.NearbyProbes, measurementState) - if len(responsiveProbes) == 0 { - c.log.Warn("No responsive probes found for location", - slog.String("location", targetLocation.LocationCode)) - continue - } - - targetProbes := getNearestProbesSorted(responsiveProbes, - targetLocation.Latitude, targetLocation.Longitude, probesPerLocation) - if len(targetProbes) == 0 { - continue - } - targetProbe := targetProbes[0] - - targetAddress := targetProbe.Address + // Cloud mode pings a fixed address, so the target location needs no probe of its own. + var targetProbe Probe + var targetAddress string if c.cloudMode { node, ok := c.cloudNodes[targetLocation.LocationCode] if !ok || node.PingTarget == "" { @@ -1454,6 +1430,25 @@ func (c *Collector) generateWantedMeasurements(locationMatches []LocationProbeMa continue } targetAddress = node.PingTarget + } else { + if len(targetLocation.NearbyProbes) == 0 { + continue + } + + responsiveProbes := filterResponsiveProbes(targetLocation.NearbyProbes, measurementState) + if len(responsiveProbes) == 0 { + c.log.Warn("No responsive probes found for location", + slog.String("location", targetLocation.LocationCode)) + continue + } + + targetProbes := getNearestProbesSorted(responsiveProbes, + targetLocation.Latitude, targetLocation.Longitude, probesPerLocation) + if len(targetProbes) == 0 { + continue + } + targetProbe = targetProbes[0] + targetAddress = targetProbe.Address } // Collect source probes from all other locations @@ -1518,7 +1513,7 @@ func (c *Collector) Run(ctx context.Context, dryRun bool, probesPerLocation int, } // Initialize shared measurement state once, used by both goroutines - timestampFile := filepath.Join(stateDir, TimestampFileName) + timestampFile := filepath.Join(stateDir, c.timestampFileName()) c.measurementState = NewMeasurementState(timestampFile) if err := c.measurementState.Load(); err != nil { c.log.Warn("Failed to load measurement state at startup", slog.String("error", err.Error()))