Cherry-pick #21166 to 7.9: Fix timestamp handling in remote_write (#2…

…1237)

CHANGELOG.next.asciidoc

@@ -95,6 +95,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix storage metricset to allow config without region/zone. {issue}17623[17623] {pull}17624[17624]
 - Fix overflow on Prometheus rates when new buckets are added on the go. {pull}17753[17753]
 - Add a switch to the driver definition on SQL module to use pretty names {pull}17378[17378]
+- Fix remote_write flaky test. {pull}21173[21173]
 
 *Packetbeat*
 

metricbeat/module/prometheus/remote_write/data.go

@@ -44,13 +44,14 @@ func samplesToEvents(metrics model.Samples) map[string]mb.Event {
 
 		val := float64(metric.Value)
 		if !math.IsNaN(val) && !math.IsInf(val, 0) {
-			// join metrics with same labels in a single event
-			labelsHash := labels.String()
+			// join metrics with same labels and same timestamp in a single event
+			labelsHash := labels.String() + metric.Timestamp.Time().String()
 			if _, ok := eventList[labelsHash]; !ok {
 				eventList[labelsHash] = mb.Event{
 					ModuleFields: common.MapStr{
 						"metrics": common.MapStr{},
 					},
+					Timestamp: metric.Timestamp.Time(),
 				}
 
 				// Add labels
@@ -61,7 +62,6 @@ func samplesToEvents(metrics model.Samples) map[string]mb.Event {
 
 			// Not checking anything here because we create these maps some lines before
 			e := eventList[labelsHash]
-			e.Timestamp = metric.Timestamp.Time()
 			data := common.MapStr{
 				name: val,
 			}

metricbeat/module/prometheus/remote_write/remote_write_test.go

@@ -0,0 +1,79 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package remote_write
+
+import (
+	"testing"
+
+	"github.com/prometheus/common/model"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+// TestGenerateEventsCounter tests counter simple cases
+func TestGenerateEventsCounter(t *testing.T) {
+
+	timestamp := model.Time(424242)
+	timestamp1 := model.Time(424243)
+	labels := common.MapStr{
+		"listener_name": model.LabelValue("http"),
+	}
+
+	// first fetch
+	metrics := model.Samples{
+		&model.Sample{
+			Metric: map[model.LabelName]model.LabelValue{
+				"__name__":      "net_conntrack_listener_conn_closed_total",
+				"listener_name": "http",
+			},
+			Value:     model.SampleValue(42),
+			Timestamp: timestamp,
+		},
+		&model.Sample{
+			Metric: map[model.LabelName]model.LabelValue{
+				"__name__":      "net_conntrack_listener_conn_closed_total",
+				"listener_name": "http",
+			},
+			Value:     model.SampleValue(43),
+			Timestamp: timestamp1,
+		},
+	}
+	events := samplesToEvents(metrics)
+
+	expected := common.MapStr{
+		"metrics": common.MapStr{
+			"net_conntrack_listener_conn_closed_total": float64(42),
+		},
+		"labels": labels,
+	}
+	expected1 := common.MapStr{
+		"metrics": common.MapStr{
+			"net_conntrack_listener_conn_closed_total": float64(43),
+		},
+		"labels": labels,
+	}
+
+	assert.Equal(t, len(events), 2)
+	e := events[labels.String()+timestamp.Time().String()]
+	assert.EqualValues(t, e.ModuleFields, expected)
+	assert.EqualValues(t, e.Timestamp, timestamp.Time())
+	e = events[labels.String()+timestamp1.Time().String()]
+	assert.EqualValues(t, e.ModuleFields, expected1)
+	assert.EqualValues(t, e.Timestamp, timestamp1.Time())
+}