db-o11y: change the server_id calculation for mysql (#4943)

* change the server_id calculation for mysql

* changelog

Author: gaantunes

CHANGELOG.md

@@ -28,7 +28,11 @@ Main (unreleased)
 
 
 - (_Public Preview_) Additions to `database_observability.postgres` component:
-    - fixes collection of Postgres schema details for mixed case table names (@fridgepoet)
+  - `schema_details`
+    - fixes collection of schema details for mixed case table names (@fridgepoet)
+
+- (_Public Preview_) Additions to `database_observability.mysql` component:
+  - replace the internal `server_id` label attribution in favor of a hash composed from `@@server_uuid` and `@@hostname`
 
 v1.12.0-rc.3
 -----------------

internal/component/database_observability/mysql/component.go

@@ -2,6 +2,7 @@ package mysql
 
 import (
 	"context"
+	"crypto/sha256"
 	"database/sql"
 	"fmt"
 	"net/http"
@@ -34,7 +35,7 @@ import (
 
 const name = "database_observability.mysql"
 
-const selectServerInfo = `SELECT @@server_uuid, VERSION()`
+const selectServerInfo = `SELECT @@server_uuid, @@hostname, VERSION()`
 
 func init() {
 	component.Register(component.Registration{
@@ -314,12 +315,15 @@ func (c *Component) Update(args component.Arguments) error {
 		return nil
 	}
 
-	var serverUUID, engineVersion string
-	if err := rs.Scan(&serverUUID, &engineVersion); err != nil {
+	var serverUUID, hostname, engineVersion string
+	if err := rs.Scan(&serverUUID, &hostname, &engineVersion); err != nil {
 		c.reportError("failed to scan engine version", err)
 		return nil
 	}
 
+	// Generate server_id hash from server_uuid and hostname, similar to Postgres collector
+	generatedServerID := fmt.Sprintf("%x", sha256.Sum256([]byte(fmt.Sprintf("%s:%s", serverUUID, hostname))))
+
 	var parsedEngineVersion semver.Version
 	matches := versionRegex.FindStringSubmatch(engineVersion)
 	if len(matches) > 1 {
@@ -334,7 +338,7 @@ func (c *Component) Update(args component.Arguments) error {
 	targets := make([]discovery.Target, 0, len(c.args.Targets)+1)
 	for _, t := range c.args.Targets {
 		builder := discovery.NewTargetBuilderFrom(t)
-		if relabel.ProcessBuilder(builder, database_observability.GetRelabelingRules(serverUUID)...) {
+		if relabel.ProcessBuilder(builder, database_observability.GetRelabelingRules(generatedServerID)...) {
 			targets = append(targets, builder.Target())
 		}
 	}
@@ -348,7 +352,7 @@ func (c *Component) Update(args component.Arguments) error {
 	}
 	c.collectors = nil
 
-	if err := c.startCollectors(serverUUID, engineVersion, parsedEngineVersion); err != nil {
+	if err := c.startCollectors(generatedServerID, engineVersion, parsedEngineVersion); err != nil {
 		c.reportError("failed to start collectors", err)
 		return nil
 	}
@@ -383,7 +387,7 @@ func enableOrDisableCollectors(a Arguments) map[string]bool {
 }
 
 // startCollectors attempts to start all of the enabled collectors. If one or more collectors fail to start, their errors are reported
-func (c *Component) startCollectors(serverUUID string, engineVersion string, parsedEngineVersion semver.Version) error {
+func (c *Component) startCollectors(serverID string, engineVersion string, parsedEngineVersion semver.Version) error {
 	var startErrors []string
 
 	logStartError := func(collectorName, action string, err error) {
@@ -405,7 +409,7 @@ func (c *Component) startCollectors(serverUUID string, engineVersion string, par
 		}
 	}
 
-	entryHandler := addLokiLabels(loki.NewEntryHandler(c.handler.Chan(), func() {}), c.instanceKey, serverUUID)
+	entryHandler := addLokiLabels(loki.NewEntryHandler(c.handler.Chan(), func() {}), c.instanceKey, serverID)
 
 	collectors := enableOrDisableCollectors(c.args)
 
@@ -616,11 +620,11 @@ func formatDSN(dsn string, params ...string) string {
 	return dsn + strings.Join(params, "&")
 }
 
-func addLokiLabels(entryHandler loki.EntryHandler, instanceKey string, serverUUID string) loki.EntryHandler {
+func addLokiLabels(entryHandler loki.EntryHandler, instanceKey string, serverID string) loki.EntryHandler {
 	entryHandler = loki.AddLabelsMiddleware(model.LabelSet{
 		"job":       database_observability.JobName,
 		"instance":  model.LabelValue(instanceKey),
-		"server_id": model.LabelValue(serverUUID),
+		"server_id": model.LabelValue(serverID),
 	}).Wrap(entryHandler)
 
 	return entryHandler

internal/component/database_observability/mysql/component_test.go

@@ -259,7 +259,7 @@ func Test_addLokiLabels(t *testing.T) {
 	t.Run("add required labels to loki entries", func(t *testing.T) {
 		lokiClient := loki.NewCollectingHandler()
 		defer lokiClient.Stop()
-		entryHandler := addLokiLabels(lokiClient, "some-instance-key", "some-server-uuid")
+		entryHandler := addLokiLabels(lokiClient, "some-instance-key", "some-server-id-hash")
 
 		go func() {
 			ts := time.Now().UnixNano()
@@ -279,7 +279,7 @@ func Test_addLokiLabels(t *testing.T) {
 		assert.Equal(t, model.LabelSet{
 			"job":       database_observability.JobName,
 			"instance":  model.LabelValue("some-instance-key"),
-			"server_id": model.LabelValue("some-server-uuid"),
+			"server_id": model.LabelValue("some-server-id-hash"),
 		}, lokiClient.Received()[0].Labels)
 		assert.Equal(t, "some-message", lokiClient.Received()[0].Line)
 	})
@@ -336,7 +336,7 @@ func TestMySQL_StartCollectors_ReportsUnhealthy_StackedErrors(t *testing.T) {
 	// First ping to the database succeeds, so we can start collectors
 	mock.ExpectPing()
 	// Engine info succeeds (if reached)
-	mock.ExpectQuery(`SELECT @@server_uuid, VERSION\(\)`).WillReturnRows(sqlmock.NewRows([]string{"server_uuid", "version"}).AddRow("uuid-1", "8.0.0"))
+	mock.ExpectQuery(`SELECT @@server_uuid, @@hostname, VERSION\(\)`).WillReturnRows(sqlmock.NewRows([]string{"server_uuid", "hostname", "version"}).AddRow("uuid-1", "test-hostname", "8.0.0"))
 	// QuerySample constructor queries uptime and fails
 	mock.ExpectQuery(regexp.QuoteMeta("SELECT variable_value FROM performance_schema.global_status WHERE variable_name = 'UPTIME'")).
 		WillReturnRows(sqlmock.NewRows([]string{"variable_value"}).AddRow(1))