Merge pull request #152 from codex-team/master

Update prod

Author: pavelzotikov

pkg/redis/client.go

@@ -346,15 +346,15 @@ func (r *RedisClient) TSAdd(
 	return res.Err()
 }
 
-// SafeTSAdd ensures that a TS key exists and adds a sample safely
+// SafeTSAdd ensures that a TS key exists and adds a sample safely.
+// timestamp is the bucket start time in milliseconds.
 func (r *RedisClient) SafeTSAdd(
 	key string,
 	value int64,
 	labels map[string]string,
 	retention time.Duration,
+	timestamp int64,
 ) error {
-	timestamp := time.Now().UnixNano() / int64(time.Millisecond)
-
 	err := r.TSAdd(key, value, timestamp, labels)
 	if err != nil && strings.Contains(err.Error(), "TSDB: key does not exist") {
 		log.Warnf("TS key %s does not exist, creating it...", key)

pkg/server/errorshandler/handler.go

@@ -144,6 +144,24 @@ func getTimeSeriesKey(projectId, metricType, granularity string, isSystemMetric
 	return fmt.Sprintf("ts:project-%s:%s:%s", metricType, projectId, granularity)
 }
 
+// bucketTimestampMs returns the current UTC time truncated to the start of the
+// given granularity bucket, in milliseconds. Truncating ensures that all events
+// within the same bucket share one timestamp so ON_DUPLICATE SUM accumulates
+// them into a single sample instead of creating a separate sample per event.
+func bucketTimestampMs(granularity string) int64 {
+	now := time.Now().UTC()
+	var t time.Time
+	switch granularity {
+	case "hourly":
+		t = now.Truncate(time.Hour)
+	case "daily":
+		t = now.Truncate(24 * time.Hour)
+	default: // minutely
+		t = now.Truncate(time.Minute)
+	}
+	return t.UnixNano() / int64(time.Millisecond)
+}
+
 // recordProjectMetrics records project metrics to Redis TimeSeries
 // metricType can be: "events-accepted", "events-rate-limited", etc.
 func (handler *Handler) recordProjectMetrics(projectId, metricType string, isSystemMetric bool) {
@@ -158,18 +176,17 @@ func (handler *Handler) recordProjectMetrics(projectId, metricType string, isSys
 	}
 
 	// minutely: store for 24 hours
-	// Use TS.ADD with ON_DUPLICATE SUM to accumulate events within the same timestamp
-	if err := handler.RedisClient.SafeTSAdd(minutelyKey, 1, labels, 24*time.Hour); err != nil {
+	if err := handler.RedisClient.SafeTSAdd(minutelyKey, 1, labels, 24*time.Hour, bucketTimestampMs("minutely")); err != nil {
 		log.Errorf("failed to add minutely TS for %s: %v", metricType, err)
 	}
 
 	// hourly: store for 7 days
-	if err := handler.RedisClient.SafeTSAdd(hourlyKey, 1, labels, 7*24*time.Hour); err != nil {
+	if err := handler.RedisClient.SafeTSAdd(hourlyKey, 1, labels, 7*24*time.Hour, bucketTimestampMs("hourly")); err != nil {
 		log.Errorf("failed to add hourly TS for %s: %v", metricType, err)
 	}
 
 	// daily: store for 90 days
-	if err := handler.RedisClient.SafeTSAdd(dailyKey, 1, labels, 90*24*time.Hour); err != nil {
+	if err := handler.RedisClient.SafeTSAdd(dailyKey, 1, labels, 90*24*time.Hour, bucketTimestampMs("daily")); err != nil {
 		log.Errorf("failed to add daily TS for %s: %v", metricType, err)
 	}
 }