telemetry: add cloud_region_latency table and carry cloud identity on records

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
+}

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)
+}

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

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)
+	})
+}