feat(collector): always use UTC timestamps

Author: tothandras

collector/benthos/input/prometheus.go

@@ -40,17 +40,17 @@ func prometheusInputConfig() *service.ConfigSpec {
 			).Description("List of PromQL queries to execute."),
 			service.NewStringField(fieldPrometheusSchedule).
 				Description("The cron expression to use for the scrape job.").
-				Examples("*/30 * * * * *", "@every 30s").
-				Default("*/30 * * * * *"),
-		).Example("Basic Configuration", "Collect Prometheus metrics with a scrape interval of 30 seconds.", `
+				Examples("0 * * * * *", "@every 1m").
+				Default("0 * * * * *"),
+		).Example("Basic Configuration", "Collect Prometheus metrics with a scrape interval of 1 minute.", `
 input:
   prometheus:
     url: "${PROMETHEUS_URL:http://localhost:9090}"
-    schedule: "* * * * *"
+    schedule: "0 * * * * *"
     queries:
       - query:
           name: "node_cpu_usage"
-          promql: "sum(rate(node_cpu_seconds_total{mode!='idle'}[1m])) by (instance)"
+          promql: "sum(increase(node_cpu_seconds_total{mode!='idle'}[1m])) by (instance)"
 `)
 }
 
@@ -173,6 +173,9 @@ func newPrometheusInput(conf *service.ParsedConfig, logger *service.Logger) (*pr
 
 // scrape executes the PromQL queries and stores the results.
 func (in *prometheusInput) scrape(ctx context.Context, t time.Time) error {
+	// Convert time to UTC
+	t = t.UTC()
+
 	in.logger.Debugf("executing PromQL queries at %s", t.Format(time.RFC3339))
 
 	results := make([]QueryResult, 0, len(in.queries))

collector/benthos/input/run_ai.go

@@ -183,8 +183,8 @@ func newRunAIInput(conf *service.ParsedConfig, logger *service.Logger) (*runAIIn
 			return nil, err
 		}
 
-		// Get current time
-		now := time.Now()
+		// Get current time in UTC
+		now := time.Now().UTC()
 
 		// Get next two occurrences
 		nextRun := cronSchedule.Next(now)
@@ -291,7 +291,8 @@ func (in *runAIInput) Connect(ctx context.Context) error {
 		gocron.CronJob(in.schedule, true),
 		gocron.NewTask(
 			func(ctx context.Context) error {
-				t := time.Now()
+				// Get current time in UTC
+				t := time.Now().UTC()
 				err := in.scrape(ctx, t)
 				if err != nil {
 					in.logger.Errorf("error scraping metrics: %v", err)

collector/benthos/input/runai/metrics.go

@@ -152,22 +152,22 @@ func (s *Service) GetAllWorkloadWithMetrics(ctx context.Context, params Measurem
 	workloadsWithMetrics := make([]WorkloadWithMetrics, len(workloads))
 	for i, workload := range workloads {
 		metrics := Metrics{
-			Timestamp: params.StartTime,
+			Timestamp: params.StartTime.UTC(),
 			Values:    make(map[MetricType]float64),
 		}
 
 		// We chunk the metric types to not exceed the max number of metrics per request
 		for _, metricTypes := range lo.Chunk(params.MetricType, 9) {
 			m, err := s.GetWorkloadMetrics(ctx, workload.ID, MeasurementParams{
 				MetricType: metricTypes,
-				StartTime:  params.StartTime,
-				EndTime:    params.EndTime,
+				StartTime:  params.StartTime.UTC(),
+				EndTime:    params.EndTime.UTC(),
 			})
 			if err != nil {
 				return nil, err
 			}
 
-			metrics.Timestamp = m.Timestamp
+			metrics.Timestamp = m.Timestamp.UTC()
 			for mt, v := range m.Values {
 				metrics.Values[mt] = v
 			}
@@ -289,22 +289,22 @@ func (s *Service) GetAllPodWithMetrics(ctx context.Context, params MeasurementPa
 	podsWithMetrics := make([]PodWithMetrics, len(pods))
 	for i, pod := range pods {
 		metrics := Metrics{
-			Timestamp: params.StartTime,
+			Timestamp: params.StartTime.UTC(),
 			Values:    make(map[MetricType]float64),
 		}
 
 		// We chunk the metric types to not exceed the max number of metrics per request
 		for _, metricTypes := range lo.Chunk(params.MetricType, 9) {
 			m, err := s.GetPodMetrics(ctx, pod.WorkloadID, pod.ID, MeasurementParams{
 				MetricType: metricTypes,
-				StartTime:  params.StartTime,
-				EndTime:    params.EndTime,
+				StartTime:  params.StartTime.UTC(),
+				EndTime:    params.EndTime.UTC(),
 			})
 			if err != nil {
 				return nil, err
 			}
 
-			metrics.Timestamp = m.Timestamp
+			metrics.Timestamp = m.Timestamp.UTC()
 			for mt, v := range m.Values {
 				metrics.Values[mt] = v
 			}

collector/benthos/input/schedule.go

@@ -66,7 +66,7 @@ func (in *scheduleInput) ReadBatch(ctx context.Context) (service.MessageBatch, s
 
 	select {
 	case v := <-in.timer.C:
-		t = v
+		t = v.UTC()
 	default:
 		return nil, func(context.Context, error) error { return nil }, nil
 	}