diff --git a/controlplane/internet-latency-collector/internal/exporter/record.go b/controlplane/internet-latency-collector/internal/exporter/record.go index 20e216ba77..da90a3ece1 100644 --- a/controlplane/internet-latency-collector/internal/exporter/record.go +++ b/controlplane/internet-latency-collector/internal/exporter/record.go @@ -18,6 +18,7 @@ type Record struct { TargetExchangeCode string Timestamp time.Time RTT time.Duration + Cloud *CloudInfo } func (r *Record) Validate() error { @@ -32,3 +33,29 @@ func (r *Record) Validate() error { } return nil } + +// CloudInfo names the cloud each exchange code belongs to, the probe that took the sample and the +// ping packet counts behind it. It is nil on records from the onchain exchange path. +type CloudInfo struct { + SourceCloud string + TargetCloud string + ProbeID uint32 + PacketsSent uint8 + PacketsReceived uint8 +} + +func (c *CloudInfo) Validate() error { + if c == nil { + return fmt.Errorf("record has no cloud info") + } + if c.SourceCloud == "" { + return fmt.Errorf("record cloud info has no source cloud") + } + if c.TargetCloud == "" { + return fmt.Errorf("record cloud info has no target cloud") + } + if c.ProbeID == 0 { + return fmt.Errorf("record cloud info has no probe id") + } + return nil +} diff --git a/controlplane/internet-latency-collector/internal/exporter/record_test.go b/controlplane/internet-latency-collector/internal/exporter/record_test.go new file mode 100644 index 0000000000..9c48db5744 --- /dev/null +++ b/controlplane/internet-latency-collector/internal/exporter/record_test.go @@ -0,0 +1,134 @@ +package exporter_test + +import ( + "testing" + "time" + + "github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/exporter" + "github.com/stretchr/testify/require" +) + +func TestInternetLatency_Record_CloudInfoIsOptional(t *testing.T) { + t.Parallel() + + record := exporter.Record{ + DataProvider: exporter.DataProviderNameRIPEAtlas, + SourceExchangeCode: "LOC_A", + TargetExchangeCode: "LOC_B", + Timestamp: time.Unix(1757462400, 0).UTC(), + RTT: 26 * time.Millisecond, + } + + require.Nil(t, record.Cloud) + require.NoError(t, record.Validate()) + require.ErrorContains(t, record.Cloud.Validate(), "no cloud info") +} + +func TestInternetLatency_Record_ValidateIgnoresCloudInfo(t *testing.T) { + t.Parallel() + + record := exporter.Record{ + DataProvider: exporter.DataProviderNameRIPEAtlas, + SourceExchangeCode: "LOC_A", + TargetExchangeCode: "LOC_B", + Timestamp: time.Unix(1757462400, 0).UTC(), + RTT: 26 * time.Millisecond, + Cloud: &exporter.CloudInfo{}, + } + + require.NoError(t, record.Validate()) + require.Error(t, record.Cloud.Validate()) +} + +func TestInternetLatency_Record_CloudInfoValidate(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cloud exporter.CloudInfo + expectErrContains string + }{ + { + name: "complete cloud info", + cloud: exporter.CloudInfo{ + SourceCloud: "aws", + TargetCloud: "aws", + ProbeID: 1003385, + PacketsSent: 3, + PacketsReceived: 3, + }, + expectErrContains: "", + }, + { + name: "total loss validates", + cloud: exporter.CloudInfo{ + SourceCloud: "aws", + TargetCloud: "aws", + ProbeID: 1003385, + PacketsSent: 3, + PacketsReceived: 0, + }, + expectErrContains: "", + }, + { + name: "missing source cloud", + cloud: exporter.CloudInfo{ + TargetCloud: "aws", + ProbeID: 1003385, + }, + expectErrContains: "no source cloud", + }, + { + name: "missing target cloud", + cloud: exporter.CloudInfo{ + SourceCloud: "aws", + ProbeID: 1003385, + }, + expectErrContains: "no target cloud", + }, + { + name: "missing probe id", + cloud: exporter.CloudInfo{ + SourceCloud: "aws", + TargetCloud: "aws", + }, + expectErrContains: "no probe id", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.cloud.Validate() + if tt.expectErrContains != "" { + require.ErrorContains(t, err, tt.expectErrContains) + } else { + require.NoError(t, err) + } + }) + } +} + +func TestInternetLatency_Record_CarriesCloudInfo(t *testing.T) { + t.Parallel() + + record := exporter.Record{ + DataProvider: exporter.DataProviderNameRIPEAtlas, + SourceExchangeCode: "us-east-1", + TargetExchangeCode: "eu-west-1", + Timestamp: time.Unix(1757462400, 0).UTC(), + RTT: 70500 * time.Microsecond, + Cloud: &exporter.CloudInfo{ + SourceCloud: "aws", + TargetCloud: "aws", + ProbeID: 1003385, + PacketsSent: 3, + PacketsReceived: 3, + }, + } + + require.NoError(t, record.Validate()) + require.NoError(t, record.Cloud.Validate()) + require.Equal(t, uint32(1003385), record.Cloud.ProbeID) + require.Equal(t, uint8(3), record.Cloud.PacketsSent) + require.Equal(t, uint8(3), record.Cloud.PacketsReceived) +} diff --git a/telemetry/migrations/20260910000000_cloud_region_latency.sql b/telemetry/migrations/20260910000000_cloud_region_latency.sql new file mode 100644 index 0000000000..b2acbbd501 --- /dev/null +++ b/telemetry/migrations/20260910000000_cloud_region_latency.sql @@ -0,0 +1,31 @@ +-- +goose Up + +-- The export swaps source and target, so origin_region is the region that was pinged. + +-- +goose StatementBegin +CREATE TABLE IF NOT EXISTS cloud_region_latency ( + event_ts DateTime('UTC'), + ingested_at DateTime('UTC') DEFAULT now(), + origin_cloud LowCardinality(String), + origin_region LowCardinality(String), + target_cloud LowCardinality(String), + target_region LowCardinality(String), + data_provider LowCardinality(String), + probe_id UInt32, + rtt_us UInt32, + packets_sent UInt8, + packets_received UInt8 +) +ENGINE = ReplacingMergeTree(ingested_at) +PARTITION BY toYYYYMM(event_ts) +-- The key leads with time because most reads filter on time only, matching the two production latency tables. +-- probe_id is in the key so a sample stays attributable to its probe and is not merged away. +ORDER BY (event_ts, origin_cloud, origin_region, target_cloud, target_region, data_provider, probe_id) +SETTINGS index_granularity = 8192; +-- +goose StatementEnd + +-- +goose Down + +-- +goose StatementBegin +DROP TABLE IF EXISTS cloud_region_latency; +-- +goose StatementEnd diff --git a/telemetry/migrations/cloud_region_latency_test.go b/telemetry/migrations/cloud_region_latency_test.go new file mode 100644 index 0000000000..da086c84a9 --- /dev/null +++ b/telemetry/migrations/cloud_region_latency_test.go @@ -0,0 +1,119 @@ +package migrations_test + +import ( + "database/sql" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestCloudRegionLatency covers the table created by 20260910000000_cloud_region_latency.sql. +func TestCloudRegionLatency(t *testing.T) { + t.Parallel() + db := newClickHouseWithMigrations(t) + + eventTS := time.Now().UTC().Truncate(time.Second) + + t.Run("samples from different probes survive a merge", func(t *testing.T) { + mustExec(t, db, ` + INSERT INTO cloud_region_latency + (event_ts, ingested_at, origin_cloud, origin_region, target_cloud, target_region, data_provider, probe_id, rtt_us, packets_sent, packets_received) VALUES + (?, ?, 'aws', 'ap-south-1', 'aws', 'eu-west-1', 'ripeatlas', 1003385, 120500, 3, 3), + (?, ?, 'aws', 'ap-south-1', 'aws', 'eu-west-1', 'ripeatlas', 1003386, 121900, 3, 3) + `, eventTS, eventTS, eventTS, eventTS) + + mustExec(t, db, `OPTIMIZE TABLE cloud_region_latency FINAL`) + + probeIDs := selectAll(t, db, ` + SELECT probe_id + FROM cloud_region_latency + WHERE origin_region = 'ap-south-1' AND target_region = 'eu-west-1' + ORDER BY probe_id + `, func(r *sql.Rows) (uint32, error) { + var v uint32 + err := r.Scan(&v) + return v, err + }) + + require.Equal(t, []uint32{1003385, 1003386}, probeIDs, + "samples that differ only by probe must both survive the merge") + }) + + t.Run("a re-exported sample collapses to one row", func(t *testing.T) { + mustExec(t, db, ` + INSERT INTO cloud_region_latency + (event_ts, ingested_at, origin_cloud, origin_region, target_cloud, target_region, data_provider, probe_id, rtt_us, packets_sent, packets_received) VALUES + (?, ?, 'aws', 'ap-northeast-1', 'aws', 'us-west-2', 'ripeatlas', 1003390, 98000, 3, 3) + `, eventTS, eventTS) + mustExec(t, db, ` + INSERT INTO cloud_region_latency + (event_ts, ingested_at, origin_cloud, origin_region, target_cloud, target_region, data_provider, probe_id, rtt_us, packets_sent, packets_received) VALUES + (?, ?, 'aws', 'ap-northeast-1', 'aws', 'us-west-2', 'ripeatlas', 1003390, 98000, 3, 3) + `, eventTS, eventTS.Add(time.Minute)) + + mustExec(t, db, `OPTIMIZE TABLE cloud_region_latency FINAL`) + + counts := selectAll(t, db, ` + SELECT count() + FROM cloud_region_latency + WHERE origin_region = 'ap-northeast-1' AND target_region = 'us-west-2' + `, func(r *sql.Rows) (uint64, error) { + var v uint64 + err := r.Scan(&v) + return v, err + }) + + require.Equal(t, []uint64{1}, counts, "the same probe's same sample must collapse") + }) + + t.Run("a total loss row is readable", func(t *testing.T) { + mustExec(t, db, ` + INSERT INTO cloud_region_latency + (event_ts, ingested_at, origin_cloud, origin_region, target_cloud, target_region, data_provider, probe_id, rtt_us, packets_sent, packets_received) VALUES + (?, ?, 'aws', 'sa-east-1', 'aws', 'eu-north-1', 'ripeatlas', 1003391, 0, 3, 0) + `, eventTS, eventTS) + + type row struct { + RttUs uint32 + PacketsSent uint8 + PacketsReceived uint8 + } + rows := selectAll(t, db, ` + SELECT rtt_us, packets_sent, packets_received + FROM cloud_region_latency + WHERE origin_region = 'sa-east-1' AND target_region = 'eu-north-1' + `, func(r *sql.Rows) (row, error) { + var v row + err := r.Scan(&v.RttUs, &v.PacketsSent, &v.PacketsReceived) + return v, err + }) + + require.Len(t, rows, 1) + require.Equal(t, uint32(0), rows[0].RttUs) + require.Equal(t, uint8(3), rows[0].PacketsSent) + require.Equal(t, uint8(0), rows[0].PacketsReceived) + }) + + t.Run("the new table is created alongside the existing ones", func(t *testing.T) { + names := selectAll(t, db, ` + SELECT name + FROM system.tables + WHERE database = currentDatabase() + AND name IN ('cloud_region_latency', 'device_ifindex', 'flows', 'interface_state', 'isis_global_state') + ORDER BY name + `, func(r *sql.Rows) (string, error) { + var v string + err := r.Scan(&v) + return v, err + }) + + require.Equal(t, []string{ + "cloud_region_latency", + "device_ifindex", + "flows", + "interface_state", + "isis_global_state", + }, names) + }) +}