database_observability: add function to build loki entry with timestamp (#3469)

Author: fridgepoet

CHANGELOG.md

@@ -34,6 +34,7 @@ Main (unreleased)
   - make tidbparser the default choice (@cristiangreco)
   - `query_sample`: better handling of timer overflows (@fridgepoet)
   - collect metrics on enabled `performance_schema.setup_consumers` (@fridgepoet)
+  - `query_sample`: base log entries on calculated timestamp from rows, not now() (@fridgepoet)
 
 - Mixin dashboards improvements: added minimum cluster size to Cluster Overview dashboard, fixed units in OpenTelemetry dashboard, fixed slow components evaluation time units in Controller dashboard and updated Prometheus dashboard to correctly aggregate across instances. (@thampiotr)
 

internal/component/database_observability/mysql/collector/loki_entry.go

@@ -12,16 +12,20 @@ import (
 	"github.com/grafana/alloy/internal/runtime/logging"
 )
 
-func buildLokiEntry(level logging.Level, op, instanceKey, line string) loki.Entry {
+func buildLokiEntryWithTimestamp(level logging.Level, op, instanceKey, line string, timestamp int64) loki.Entry {
 	return loki.Entry{
 		Labels: model.LabelSet{
 			"job":      database_observability.JobName,
 			"op":       model.LabelValue(op),
 			"instance": model.LabelValue(instanceKey),
 		},
 		Entry: logproto.Entry{
-			Timestamp: time.Unix(0, time.Now().UnixNano()),
+			Timestamp: time.Unix(0, timestamp),
 			Line:      fmt.Sprintf(`level="%s" %s`, level, line),
 		},
 	}
 }
+
+func buildLokiEntry(level logging.Level, op, instanceKey, line string) loki.Entry {
+	return buildLokiEntryWithTimestamp(level, op, instanceKey, line, time.Now().UnixNano())
+}

internal/component/database_observability/mysql/collector/loki_entry_test.go

@@ -2,6 +2,7 @@ package collector
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -18,3 +19,10 @@ func TestBuildLokiEntry(t *testing.T) {
 	require.Equal(t, "test-instance", string(entry.Labels["instance"]))
 	require.Equal(t, `level="debug" This is a test log line`, entry.Line)
 }
+
+func TestBuildLokiEntryWithTimestamp(t *testing.T) {
+	entry := buildLokiEntryWithTimestamp(logging.LevelInfo, "test-operation", "test-instance", "This is a test log line", 5)
+
+	require.Equal(t, int64(5), entry.Entry.Timestamp.UnixNano())
+	require.Equal(t, time.Unix(0, 5), entry.Entry.Timestamp)
+}

internal/component/database_observability/mysql/collector/query_sample.go

@@ -204,7 +204,7 @@ func (c *QuerySample) fetchQuerySamples(ctx context.Context) error {
 
 			// sample time
 			TimerEndPicoseconds    sql.NullFloat64
-			TimestampMilliseconds  uint64
+			TimestampMilliseconds  float64
 			ElapsedTimePicoseconds sql.NullFloat64
 			CPUTime                float64
 
@@ -286,11 +286,12 @@ func (c *QuerySample) fetchQuerySamples(ctx context.Context) error {
 			logMessage += fmt.Sprintf(` sql_text="%s"`, row.SQLText.String)
 		}
 
-		c.entryHandler.Chan() <- buildLokiEntry(
+		c.entryHandler.Chan() <- buildLokiEntryWithTimestamp(
 			logging.LevelInfo,
 			OP_QUERY_SAMPLE,
 			c.instanceKey,
 			logMessage,
+			int64(millisecondsToNanoseconds(row.TimestampMilliseconds)),
 		)
 	}
 
@@ -302,7 +303,7 @@ func (c *QuerySample) fetchQuerySamples(ctx context.Context) error {
 	return nil
 }
 
-func (c *QuerySample) calculateWallTime(serverStartTime, timer float64) uint64 {
+func (c *QuerySample) calculateWallTime(serverStartTime, timer float64) float64 {
 	// timer indicates event timing since server startup.
 	// The timer value is in picoseconds with a column type of bigint unsigned. This value can overflow after about ~213 days.
 	// We need to account for this overflow when calculating the timestamp.
@@ -358,6 +359,7 @@ const (
 	picosecondsPerSecond      float64 = 1e12
 	millisecondsPerSecond     float64 = 1e3
 	millisecondsPerPicosecond float64 = 1e9
+	nanosecondsPerMillisecond float64 = 1e6
 )
 
 func picosecondsToSeconds(picoseconds float64) float64 {
@@ -368,10 +370,14 @@ func picosecondsToMilliseconds(picoseconds float64) float64 {
 	return picoseconds / millisecondsPerPicosecond
 }
 
+func millisecondsToNanoseconds(milliseconds float64) float64 {
+	return milliseconds * nanosecondsPerMillisecond
+}
+
 func secondsToPicoseconds(seconds float64) float64 {
 	return seconds * picosecondsPerSecond
 }
 
-func secondsToMilliseconds(seconds float64) uint64 {
-	return uint64(seconds * millisecondsPerSecond)
+func secondsToMilliseconds(seconds float64) float64 {
+	return seconds * millisecondsPerSecond
 }

internal/component/database_observability/mysql/collector/query_sample_test.go

@@ -1482,7 +1482,7 @@ func TestQuerySample_calculateWallTime(t *testing.T) {
 		timer := 2e12 // Timer indicates event timing, counted since server startup. 2 seconds in picoseconds
 
 		result := c.calculateWallTime(serverStartTime, timer)
-		assert.Equalf(t, uint64(4000), result, "got %d, want 4000", result)
+		assert.Equalf(t, float64(4000), result, "got %f, want 4000", result)
 	})
 
 	t.Run("calculates the timestamp, taking into account the overflows", func(t *testing.T) {
@@ -1492,7 +1492,7 @@ func TestQuerySample_calculateWallTime(t *testing.T) {
 
 		result := c.calculateWallTime(serverStartTime, timer)
 
-		assert.Equalf(t, uint64(18446749073), result, "got %d, want 18446749073", result)
+		assert.Equalf(t, 18446749073.709553, result, "got %f, want 18446749073.709553", result)
 	})
 
 	t.Run("calculates another timestamp when timer approaches overflow", func(t *testing.T) {
@@ -1502,7 +1502,7 @@ func TestQuerySample_calculateWallTime(t *testing.T) {
 
 		result := c.calculateWallTime(serverStartTime, timer)
 
-		assert.Equalf(t, uint64(36893491147), result, "got %d, want 36893491147", result)
+		assert.Equalf(t, 3.6893491147419106e+10, result, "got %f, want 3.6893491147419106e+10", result)
 	})
 }