From 8fc27be85ae6eac8bd4ef2e30ea984af2ebdf4fe Mon Sep 17 00:00:00 2001 From: Greg Mitchell Date: Sat, 1 Aug 2026 23:39:33 +0000 Subject: [PATCH 1/2] inet-collector: stop recreating the RIPE Atlas fleet on metadata loss A failed measurement listing was substituted with an empty list, which made every live measurement look absent. Metadata was then pruned as orphaned, and the next run stopped and recreated all 29 mainnet measurements at once. - abort reconciliation when the measurement listing fails instead of reconciling against an empty view - rebuild metadata for unrecognized measurements from the description and the measurement's probe list rather than stopping and recreating them - refuse to prune orphaned metadata when it would remove every entry - write the state file atomically via temp file and rename - set the tracked-measurement gauge from state instead of incrementing and decrementing it, which had driven the value negative --- .../internal/metrics/metrics.go | 15 ++ .../internal/ripeatlas/client.go | 36 ++++ .../internal/ripeatlas/collector.go | 145 +++++++++++++- .../internal/ripeatlas/collector_test.go | 179 ++++++++++++++++++ .../internal/ripeatlas/state.go | 32 +++- .../internal/ripeatlas/state_test.go | 59 +++++- 6 files changed, 455 insertions(+), 11 deletions(-) diff --git a/controlplane/internet-latency-collector/internal/metrics/metrics.go b/controlplane/internet-latency-collector/internal/metrics/metrics.go index a809a71ae5..e54f5e5899 100644 --- a/controlplane/internet-latency-collector/internal/metrics/metrics.go +++ b/controlplane/internet-latency-collector/internal/metrics/metrics.go @@ -95,6 +95,21 @@ var ( Help: "Current RIPE Atlas credit balance", }) + RipeatlasMeasurementListFailuresTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "doublezero_internet_latency_collector_ripeatlas_measurement_list_failures_total", + Help: "Total number of times listing existing measurements failed, aborting reconciliation for that cycle", + }) + + RipeatlasMetadataRebuiltTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "doublezero_internet_latency_collector_ripeatlas_metadata_rebuilt_total", + Help: "Total number of measurements whose metadata was rebuilt from the RIPE Atlas API instead of being recreated", + }) + + RipeatlasMetadataPruneSkippedTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "doublezero_internet_latency_collector_ripeatlas_metadata_prune_skipped_total", + Help: "Total number of times orphaned-metadata pruning was skipped because it would have removed every entry", + }) + RipeatlasTotalMeasurements = promauto.NewGauge(prometheus.GaugeOpts{ Name: "doublezero_internet_latency_collector_ripeatlas_total_measurements", Help: "Total number of RIPE Atlas measurements being tracked", diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/client.go b/controlplane/internet-latency-collector/internal/ripeatlas/client.go index 430b6a5246..808f692bd3 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/client.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/client.go @@ -338,6 +338,42 @@ func (c *Client) GetAllMeasurements(ctx context.Context, env string) ([]Measurem return allMeasurements, nil } +// GetMeasurementProbes returns the probes participating in a measurement. This +// lets the collector rebuild metadata for a measurement it no longer recognizes +// instead of stopping and recreating it. +func (c *Client) GetMeasurementProbes(ctx context.Context, measurementID int) ([]Probe, error) { + endpoint := fmt.Sprintf("/measurements/%d/probes/", measurementID) + + allProbes := []Probe{} + for { + resp, err := c.makeRequest(ctx, endpoint) + if err != nil { + return nil, fmt.Errorf("failed to get probes for measurement %d: %w", measurementID, err) + } + + var response ProbesResponse + decoder := json.NewDecoder(resp.Body) + if err := decoder.Decode(&response); err != nil { + resp.Body.Close() + return nil, fmt.Errorf("failed to decode probes response for measurement %d: %w", measurementID, err) + } + resp.Body.Close() + + allProbes = append(allProbes, response.Results...) + + if response.Next == "" { + break + } + + endpoint = response.Next + if len(endpoint) > len(c.BaseURL) { + endpoint = endpoint[len(c.BaseURL):] + } + } + + return allProbes, nil +} + func (c *Client) GetMeasurementResultsIncremental(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) { endpoint := fmt.Sprintf("/measurements/%d/results/", measurementID) diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/collector.go b/controlplane/internet-latency-collector/internal/ripeatlas/collector.go index 0de30e06d0..41a26280f0 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/collector.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/collector.go @@ -6,7 +6,9 @@ import ( "log/slog" "os" "path/filepath" + "regexp" "sort" + "strconv" "strings" "sync" "time" @@ -30,6 +32,7 @@ type clientInterface interface { GetMeasurementResultsIncremental(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) StopMeasurement(ctx context.Context, measurementID int) error GetCreditBalance(ctx context.Context) (float64, error) + GetMeasurementProbes(ctx context.Context, measurementID int) ([]Probe, error) } type LocationProbeMatch struct { @@ -38,6 +41,109 @@ type LocationProbeMatch struct { ProbeCount int } +// measurementDescriptionRE matches the descriptions the collector generates for +// its own measurements, in both the environment-qualified and bare forms: +// +// DoubleZero [mainnet-beta] to osl probe 7439 +// DoubleZero to osl probe 7439 +// +// The target location and target probe are recoverable from the description +// alone, which is what makes metadata rebuilding possible. +var measurementDescriptionRE = regexp.MustCompile(`^DoubleZero(?: \[[^\]]*\])? to (\S+) probe (\d+)$`) + +// parseMeasurementDescription extracts the target location code and target probe +// ID from a collector-generated measurement description. +func parseMeasurementDescription(description string) (locationCode string, targetProbeID int, ok bool) { + m := measurementDescriptionRE.FindStringSubmatch(strings.TrimSpace(description)) + if m == nil { + return "", 0, false + } + probeID, err := strconv.Atoi(m[2]) + if err != nil { + return "", 0, false + } + return m[1], probeID, true +} + +// rehydrateMissingMetadata rebuilds state entries for live measurements the +// collector no longer has metadata for. +// +// Losing the state file used to be unrecoverable: an unrecognized measurement is +// stopped and recreated, so losing state for the whole fleet recreated every +// measurement at once and cost hours of coverage while fresh measurements ramped +// up. Everything needed to rebuild an entry is still available, though. The +// description carries the target location and target probe, and the measurement's +// probe list plus the current location-to-probe mapping gives back the sources. +func (c *Collector) rehydrateMissingMetadata(ctx context.Context, measurements []Measurement, locationMatches []LocationProbeMatch, measurementState *MeasurementState) { + // Map every probe we know about back to its location code. + probeLocations := make(map[int]string) + for _, match := range locationMatches { + for _, probe := range match.NearbyProbes { + probeLocations[probe.ID] = match.LocationCode + } + } + + recovered := 0 + for _, measurement := range measurements { + if _, hasMetadata := measurementState.GetMetadata(measurement.ID); hasMetadata { + continue + } + + targetLocation, targetProbeID, ok := parseMeasurementDescription(measurement.Description) + if !ok { + continue + } + + probes, err := c.client.GetMeasurementProbes(ctx, measurement.ID) + if err != nil { + c.log.Warn("Failed to fetch probes while rebuilding measurement metadata", + slog.Int("measurement_id", measurement.ID), + slog.String("error", err.Error())) + continue + } + + sources := make([]SourceProbeMeta, 0, len(probes)) + for _, probe := range probes { + locationCode, known := probeLocations[probe.ID] + if !known { + // A probe we can no longer place cannot be attributed to a + // metro, so the entry would be incomplete. Leave the + // measurement unrecognized and let it be recreated. + sources = nil + break + } + sources = append(sources, SourceProbeMeta{LocationCode: locationCode, ProbeID: probe.ID}) + } + if len(sources) == 0 { + c.log.Warn("Could not rebuild measurement metadata; probes are unrecognized", + slog.Int("measurement_id", measurement.ID), + slog.String("description", measurement.Description)) + continue + } + + measurementState.SetMetadata(measurement.ID, MeasurementMeta{ + TargetLocation: targetLocation, + TargetProbeID: targetProbeID, + Sources: sources, + CreatedAt: time.Now().Unix(), + }) + recovered++ + + c.log.Info("Rebuilt metadata for unrecognized measurement", + slog.Int("measurement_id", measurement.ID), + slog.String("target_location", targetLocation), + slog.Int("target_probe_id", targetProbeID), + slog.Int("sources", len(sources))) + } + + if recovered > 0 { + metrics.RipeatlasMetadataRebuiltTotal.Add(float64(recovered)) + if err := measurementState.Save(); err != nil { + c.log.Warn("Failed to save rebuilt measurement metadata", slog.String("error", err.Error())) + } + } +} + type ProbeDistance struct { Probe Probe Distance float64 @@ -731,8 +837,12 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ // Step 4: Get all existing measurements existingMeasurements, err := c.client.GetAllMeasurements(ctx, c.env) if err != nil { - c.log.Warn("Failed to get existing measurements", slog.String("error", err.Error())) - existingMeasurements = []Measurement{} + // Continuing with an empty listing would make every live measurement + // look absent: metadata gets pruned as orphaned and the next run stops + // and recreates the entire fleet. Reconciliation is only safe against a + // listing we actually retrieved, so bail out and retry next cycle. + metrics.RipeatlasMeasurementListFailuresTotal.Inc() + return fmt.Errorf("failed to get existing measurements: %w", err) } // Filter for DoubleZero measurements only @@ -743,6 +853,10 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ } } + // Rebuild metadata for any measurement we no longer recognize before making + // keep/remove decisions, so the rest of reconciliation sees a complete view. + c.rehydrateMissingMetadata(ctx, doubleZeroMeasurements, locationMatches, measurementState) + // Step 3: Build map of existing measurements by target location existingByTarget := make(map[string]Measurement) for _, m := range doubleZeroMeasurements { @@ -973,6 +1087,8 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ // Check if measurement has metadata _, hasMetadata := measurementState.GetMetadata(measurement.ID) if !hasMetadata { + // Metadata rebuilding already ran; anything still unrecognized here + // is a genuine orphan. c.log.Info("Marking measurement for removal due to missing metadata", slog.Int("measurement_id", measurement.ID), slog.String("description", measurement.Description)) @@ -1032,7 +1148,6 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ // Always remove metadata for measurements we're removing, // even if the API call fails (measurement might already be stopped) measurementState.RemoveMetadata(measurement.ID) - metrics.RipeatlasTotalMeasurements.Dec() time.Sleep(CallDelay) // Rate limiting } } @@ -1056,6 +1171,21 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ } } + // A measurement listing that came back empty, or that would orphan every + // entry we know about, is far more likely to be a degraded API response than + // a genuine fleet-wide teardown. Pruning on that signal deletes the metadata + // the collector needs to recognize its own measurements, and the next run + // then stops and recreates all of them. Keep the state and retry next cycle. + knownCount := measurementState.MetadataCount() + if len(orphanedIDs) > 0 && knownCount > 0 && len(orphanedIDs) == knownCount { + c.log.Error("Refusing to prune all measurement metadata; treating as a degraded measurement listing", + slog.Int("known_metadata", knownCount), + slog.Int("would_orphan", len(orphanedIDs)), + slog.Int("measurements_from_api", len(doubleZeroMeasurements))) + metrics.RipeatlasMetadataPruneSkippedTotal.Inc() + orphanedIDs = nil + } + if len(orphanedIDs) > 0 { c.log.Info("Cleaning up orphaned metadata entries", slog.Int("count", len(orphanedIDs))) @@ -1161,8 +1291,8 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ if err := measurementState.Save(); err != nil { c.log.Warn("Failed to save measurement metadata", slog.String("error", err.Error())) } else { - // Update metrics - metrics.RipeatlasTotalMeasurements.Inc() + // Update metrics. The tracked-measurement total is set from + // state at the end of the run rather than incremented here. metrics.RipeatlasProbesPerMeasurement.Set(float64(len(sources))) } } @@ -1215,6 +1345,11 @@ func (c *Collector) configureMeasurements(ctx context.Context, locationMatches [ metrics.RipeatlasExpectedDailyResults.Set(expectedDailyResults) metrics.RipeatlasExpectedDailyCredits.Set(expectedDailyCredits) + // Set the tracked-measurement count from state rather than incrementing and + // decrementing it per create/remove. The counter-style updates drift against + // the value written at startup and have driven this gauge negative. + metrics.RipeatlasTotalMeasurements.Set(float64(len(allMetadata))) + c.log.Info("Updated expected daily metrics", slog.Float64("expected_daily_results", expectedDailyResults), slog.Float64("expected_daily_credits", expectedDailyCredits), diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/collector_test.go b/controlplane/internet-latency-collector/internal/ripeatlas/collector_test.go index ae7685c166..6ed705073d 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/collector_test.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/collector_test.go @@ -29,6 +29,14 @@ type MockClient struct { GetMeasurementResultsIncrementalFunc func(ctx context.Context, measurementID int, startTimestamp int64) ([]any, error) StopMeasurementFunc func(ctx context.Context, measurementID int) error GetCreditBalanceFunc func(ctx context.Context) (float64, error) + GetMeasurementProbesFunc func(ctx context.Context, measurementID int) ([]Probe, error) +} + +func (m *MockClient) GetMeasurementProbes(ctx context.Context, measurementID int) ([]Probe, error) { + if m.GetMeasurementProbesFunc != nil { + return m.GetMeasurementProbesFunc(ctx, measurementID) + } + return []Probe{}, nil } func (m *MockClient) GetProbesInRadius(ctx context.Context, latitude, longitude float64, radiusKm int, anchorsOnly bool) ([]Probe, error) { @@ -1624,3 +1632,174 @@ func TestInitializeCreditBalance(t *testing.T) { require.Contains(t, err.Error(), "failed to get RIPE Atlas credit balance") }) } + +func TestInternetLatency_RIPEAtlas_ParseMeasurementDescription(t *testing.T) { + t.Parallel() + + tests := []struct { + description string + wantLoc string + wantProbe int + wantOK bool + }{ + {"DoubleZero [mainnet-beta] to osl probe 7439", "osl", 7439, true}, + {"DoubleZero to osl probe 7439", "osl", 7439, true}, + {"DoubleZero [] to ams probe 1", "ams", 1, true}, + {"DoubleZero [mainnet-beta] to slc probe 1008538", "slc", 1008538, true}, + // Legacy source-qualified descriptions cannot be rebuilt from text alone. + {"DoubleZero NYC probe 100 to LON probe 200", "", 0, false}, + {"Someone else's measurement", "", 0, false}, + {"DoubleZero [mainnet-beta] to osl probe abc", "", 0, false}, + {"", "", 0, false}, + } + + for _, tt := range tests { + loc, probe, ok := parseMeasurementDescription(tt.description) + require.Equal(t, tt.wantOK, ok, "description: %q", tt.description) + require.Equal(t, tt.wantLoc, loc, "description: %q", tt.description) + require.Equal(t, tt.wantProbe, probe, "description: %q", tt.description) + } +} + +// Losing the state file must not cost us the measurement fleet. Metadata is +// rebuilt from the description plus the measurement's probe list, so the +// measurements keep running instead of being stopped and recreated. +func TestInternetLatency_RIPEAtlas_RehydrateMissingMetadata(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + stateDir := t.TempDir() + + mockClient := &MockClient{ + GetMeasurementProbesFunc: func(ctx context.Context, measurementID int) ([]Probe, error) { + return []Probe{{ID: 600}, {ID: 700}}, nil + }, + } + c := &Collector{client: mockClient, log: log} + + locationMatches := []LocationProbeMatch{ + {LocationMatch: collector.LocationMatch{LocationCode: "ams"}, NearbyProbes: []Probe{{ID: 500}}}, + {LocationMatch: collector.LocationMatch{LocationCode: "lon"}, NearbyProbes: []Probe{{ID: 600}}}, + {LocationMatch: collector.LocationMatch{LocationCode: "fra"}, NearbyProbes: []Probe{{ID: 700}}}, + } + + ms := NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + measurements := []Measurement{ + {ID: 9001, Description: "DoubleZero [test] to ams probe 500"}, + } + + c.rehydrateMissingMetadata(t.Context(), measurements, locationMatches, ms) + + meta, ok := ms.GetMetadata(9001) + require.True(t, ok, "metadata should have been rebuilt") + require.Equal(t, "ams", meta.TargetLocation) + require.Equal(t, 500, meta.TargetProbeID) + require.ElementsMatch(t, []SourceProbeMeta{ + {LocationCode: "lon", ProbeID: 600}, + {LocationCode: "fra", ProbeID: 700}, + }, meta.Sources) + + // The rebuilt state must survive a reload. + reloaded := NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + require.NoError(t, reloaded.Load()) + require.Equal(t, 1, reloaded.MetadataCount()) +} + +func TestInternetLatency_RIPEAtlas_RehydrateMissingMetadata_Unrecoverable(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + stateDir := t.TempDir() + + probesCalled := 0 + mockClient := &MockClient{ + GetMeasurementProbesFunc: func(ctx context.Context, measurementID int) ([]Probe, error) { + probesCalled++ + // A probe we cannot place in any known metro. + return []Probe{{ID: 999999}}, nil + }, + } + c := &Collector{client: mockClient, log: log} + locationMatches := []LocationProbeMatch{ + {LocationMatch: collector.LocationMatch{LocationCode: "ams"}, NearbyProbes: []Probe{{ID: 500}}}, + } + ms := NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + + c.rehydrateMissingMetadata(t.Context(), []Measurement{ + // Unparseable description: never reaches the probes API. + {ID: 9001, Description: "DoubleZero NYC probe 1 to LON probe 2"}, + // Parseable, but its probes cannot be attributed to a metro. + {ID: 9002, Description: "DoubleZero [test] to ams probe 500"}, + }, locationMatches, ms) + + require.Equal(t, 1, probesCalled, "only the parseable description should hit the API") + require.Equal(t, 0, ms.MetadataCount(), "no metadata should be invented") +} + +// A failed measurement listing must abort reconciliation. Treating it as "no +// measurements exist" prunes all metadata as orphaned, and the next run then +// stops and recreates the entire fleet. +func TestInternetLatency_RIPEAtlas_ConfigureMeasurements_ListFailureAborts(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + stateDir := t.TempDir() + + var stopped []int + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, env string) ([]Measurement, error) { + return nil, errors.New("API request failed with status: 502") + }, + StopMeasurementFunc: func(ctx context.Context, measurementID int) error { + stopped = append(stopped, measurementID) + return nil + }, + } + + ms := NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + for i := 1; i <= 3; i++ { + ms.SetMetadata(i, MeasurementMeta{TargetLocation: "ams", TargetProbeID: i}) + } + require.NoError(t, ms.Save()) + + c := &Collector{client: mockClient, log: log, measurementState: ms} + err := c.configureMeasurements(t.Context(), []LocationProbeMatch{}, false, 1, stateDir, time.Minute) + + require.Error(t, err, "reconciliation must fail rather than proceed on a partial view") + require.Empty(t, stopped, "no measurement may be stopped when the listing failed") + require.Equal(t, 3, ms.MetadataCount(), "metadata must survive a failed listing") + + reloaded := NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + require.NoError(t, reloaded.Load()) + require.Equal(t, 3, reloaded.MetadataCount(), "persisted metadata must survive too") +} + +// If every known metadata entry looks orphaned, the listing is far more likely +// to be degraded than the fleet to have vanished. Keep the state and retry. +func TestInternetLatency_RIPEAtlas_ConfigureMeasurements_SkipsFullMetadataPrune(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + stateDir := t.TempDir() + + mockClient := &MockClient{ + GetAllMeasurementsFunc: func(ctx context.Context, env string) ([]Measurement, error) { + // A single unrelated measurement: none of our known IDs are present. + return []Measurement{ + {ID: 99, Description: "DoubleZero unparseable form"}, + }, nil + }, + } + + ms := NewMeasurementState(filepath.Join(stateDir, TimestampFileName)) + for i := 1; i <= 3; i++ { + ms.SetMetadata(i, MeasurementMeta{TargetLocation: "ams", TargetProbeID: i}) + } + require.NoError(t, ms.Save()) + + c := &Collector{client: mockClient, log: log, measurementState: ms} + err := c.configureMeasurements(t.Context(), []LocationProbeMatch{}, false, 1, stateDir, time.Minute) + require.NoError(t, err) + + require.Equal(t, 3, ms.MetadataCount(), "metadata must not be pruned wholesale") +} diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/state.go b/controlplane/internet-latency-collector/internal/ripeatlas/state.go index 647f30249b..017e4d204c 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/state.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/state.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "os" + "path/filepath" "sync" "time" ) @@ -104,22 +105,43 @@ func (ms *MeasurementState) Load() error { return nil } +// Save atomically persists the tracker by writing to a temporary file in the +// same directory and renaming it over the target. Writing in place would leave +// a truncated file behind if the process is killed mid-write, and a torn state +// file causes the collector to treat every live measurement as unrecognized. func (ms *MeasurementState) Save() error { ms.mu.Lock() defer ms.mu.Unlock() - file, err := os.Create(ms.filename) + tmp, err := os.CreateTemp(filepath.Dir(ms.filename), filepath.Base(ms.filename)+".tmp-*") if err != nil { - return fmt.Errorf("failed to create timestamp file: %w", err) + return fmt.Errorf("failed to create temporary timestamp file: %w", err) } - defer file.Close() - - encoder := json.NewEncoder(file) + tmpName := tmp.Name() + // Best-effort cleanup; a successful rename makes this a no-op. + defer func() { + tmp.Close() + os.Remove(tmpName) + }() + + encoder := json.NewEncoder(tmp) encoder.SetIndent("", " ") if err := encoder.Encode(ms.tracker); err != nil { return fmt.Errorf("failed to encode timestamp file: %w", err) } + // Flush to disk before renaming so a crash cannot leave an empty file in place. + if err := tmp.Sync(); err != nil { + return fmt.Errorf("failed to sync timestamp file: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("failed to close timestamp file: %w", err) + } + + if err := os.Rename(tmpName, ms.filename); err != nil { + return fmt.Errorf("failed to replace timestamp file: %w", err) + } + return nil } diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/state_test.go b/controlplane/internet-latency-collector/internal/ripeatlas/state_test.go index 1eaf7d39f3..478f92b502 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/state_test.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/state_test.go @@ -162,7 +162,12 @@ func TestInternetLatency_RIPEAtlas_State_FilePermissionError(t *testing.T) { err = ms.Save() require.Error(t, err, "Expected error when saving to read-only directory") - require.Contains(t, err.Error(), "failed to create timestamp file") + require.Contains(t, err.Error(), "failed to create temporary timestamp file") + + // The atomic write must not leave a partial file behind in the target directory. + entries, err := os.ReadDir(readOnlyDir) + require.NoError(t, err) + require.Empty(t, entries, "no temporary file should remain after a failed save") } func TestInternetLatency_RIPEAtlas_State_EmptyMetadataInFile(t *testing.T) { @@ -436,3 +441,55 @@ func TestInternetLatency_RIPEAtlas_State_TimestampTracker_Structure(t *testing.T require.Equal(t, tracker.Metadata, tracker2.Metadata) } + +// A save that is interrupted must never leave a truncated state file behind. +// Losing the metadata map is what causes the collector to stop recognizing its +// own measurements, which in turn triggers a full teardown and recreation. +func TestInternetLatency_RIPEAtlas_State_SaveIsAtomic(t *testing.T) { + t.Parallel() + + tempDir := t.TempDir() + filename := filepath.Join(tempDir, "timestamps.json") + + ms := NewMeasurementState(filename) + for i := 1; i <= 25; i++ { + ms.SetMetadata(i, MeasurementMeta{ + TargetLocation: "ams", + TargetProbeID: 1000 + i, + Sources: []SourceProbeMeta{{LocationCode: "lon", ProbeID: 2000 + i}}, + }) + } + require.NoError(t, ms.Save()) + + // Concurrent saves must not expose a partially written file to a reader. + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _ = ms.Save() + }() + } + for i := 0; i < 25; i++ { + wg.Add(1) + go func() { + defer wg.Done() + reloaded := NewMeasurementState(filename) + // Any read must see a complete, decodable file with every entry. + if err := reloaded.Load(); err == nil { + require.Equal(t, 25, reloaded.MetadataCount()) + } + }() + } + wg.Wait() + + // No temporary files may be left over. + entries, err := os.ReadDir(tempDir) + require.NoError(t, err) + require.Len(t, entries, 1, "only the state file should remain: %v", entries) + require.Equal(t, "timestamps.json", entries[0].Name()) + + reloaded := NewMeasurementState(filename) + require.NoError(t, reloaded.Load()) + require.Equal(t, 25, reloaded.MetadataCount()) +} From ad367b7d10ff7a41960a409314361f34c611a288 Mon Sep 17 00:00:00 2001 From: Greg Mitchell Date: Sat, 1 Aug 2026 23:52:11 +0000 Subject: [PATCH 2/2] inet-collector: add client tests for measurement probe fetching --- .../internal/ripeatlas/client_test.go | 89 +++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/controlplane/internet-latency-collector/internal/ripeatlas/client_test.go b/controlplane/internet-latency-collector/internal/ripeatlas/client_test.go index 50c3becb1c..f5e344227f 100644 --- a/controlplane/internet-latency-collector/internal/ripeatlas/client_test.go +++ b/controlplane/internet-latency-collector/internal/ripeatlas/client_test.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "net/http" + "strconv" "strings" "testing" "time" @@ -1063,3 +1064,91 @@ func TestInternetLatency_RIPEAtlas_GetCreditBalance(t *testing.T) { require.NoError(t, err, "GetCreditBalance() should not return error") require.Equal(t, 1000.0, balance, "Expected credit balance to be 1000") } + +func TestInternetLatency_RIPEAtlas_GetMeasurementProbes(t *testing.T) { + t.Parallel() + + log := logger.With("test", t.Name()) + + t.Run("single page", func(t *testing.T) { + t.Parallel() + + var requested []string + client := &Client{ + log: log, + BaseURL: "https://atlas.ripe.net/api/v2", + HTTPClient: &MockHTTPClient{ + DoFunc: func(req *http.Request) (*http.Response, error) { + requested = append(requested, req.URL.String()) + body, _ := json.Marshal(ProbesResponse{ + Count: 2, + Results: []Probe{{ID: 600}, {ID: 700}}, + }) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewReader(body)), + }, nil + }, + }, + } + + probes, err := client.GetMeasurementProbes(t.Context(), 9001) + require.NoError(t, err) + require.Len(t, probes, 2) + require.Equal(t, 600, probes[0].ID) + require.Equal(t, 700, probes[1].ID) + require.Equal(t, []string{"https://atlas.ripe.net/api/v2/measurements/9001/probes/"}, requested) + }) + + t.Run("follows pagination", func(t *testing.T) { + t.Parallel() + + page := 0 + client := &Client{ + log: log, + BaseURL: "https://atlas.ripe.net/api/v2", + HTTPClient: &MockHTTPClient{ + DoFunc: func(req *http.Request) (*http.Response, error) { + page++ + resp := ProbesResponse{Results: []Probe{{ID: 100 * page}}} + if page < 3 { + resp.Next = "https://atlas.ripe.net/api/v2/measurements/9001/probes/?page=" + strconv.Itoa(page+1) + } + body, _ := json.Marshal(resp) + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewReader(body)), + }, nil + }, + }, + } + + probes, err := client.GetMeasurementProbes(t.Context(), 9001) + require.NoError(t, err) + require.Equal(t, 3, page, "should have followed both next links") + require.Len(t, probes, 3, "results from every page should be accumulated") + require.Equal(t, []int{100, 200, 300}, []int{probes[0].ID, probes[1].ID, probes[2].ID}) + }) + + t.Run("propagates API errors", func(t *testing.T) { + t.Parallel() + + client := &Client{ + log: log, + BaseURL: "https://atlas.ripe.net/api/v2", + HTTPClient: &MockHTTPClient{ + DoFunc: func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusUnauthorized, + Body: io.NopCloser(bytes.NewReader([]byte(`{}`))), + }, nil + }, + }, + } + + probes, err := client.GetMeasurementProbes(t.Context(), 9001) + require.Error(t, err, "a non-200 must not be reported as an empty probe list") + require.Contains(t, err.Error(), "401") + require.Nil(t, probes) + }) +}