internet-latency-collector: separate cloud mode measurement identity

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)

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 <code> probe <id>", cloud descriptions "to <code> target <address>".
+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
+}