dbo11y: stop tracking alloy's own queries in mysql (#4978)

dbo11y: stop tracking alloy's own queries in mysql and postgres

This PR introduces a change for excluding "own queries" in mysql: we add a new `setup_actors` collector that checks (and optionally updates) settings for the currently connected user. By default it just checks that `enabled=NO` and logs out if that's not the case. Optionally, the component can auto-update this setting itself. This behaviour is disabled by default, though: it follows the same pattern of `setup_consumers` where two boolean settings are required to enable the auto-update. Additionally, auto-update requires `INSERT` and `UPDATE` permissions (docs on the website).

Author: cristiangreco

CHANGELOG.md

@@ -30,13 +30,13 @@ Main (unreleased)
 
 ### Bugfixes
 
-
 - (_Public Preview_) Additions to `database_observability.postgres` component:
   - `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`
+  - add `setup_actors` collector that checks and optionally updates settings to avoid tracking queries for the currently connected user (@cristiangreco)
 
 v1.12.0
 -----------------

docs/sources/reference/components/database_observability/database_observability.mysql.md

@@ -37,14 +37,15 @@ You can use the following arguments with `database_observability.mysql`:
 
 The following collectors are configurable:
 
-| Name              | Description                                              | Enabled by default |
-|-------------------|----------------------------------------------------------|--------------------|
-| `query_details`   | Collect queries information.                             | yes                |
-| `schema_details`  | Collect schemas and tables from `information_schema`.    | yes                |
-| `query_samples`   | Collect query samples.                                   | yes                |
-| `setup_consumers` | Collect enabled `performance_schema.setup_consumers`.    | yes                |
-| `locks`           | Collect queries that are waiting/blocking other queries. | no                 |
-| `explain_plans`   | Collect explain plans information.                       | yes                |
+| Name              | Description                                                  | Enabled by default |
+|-------------------|--------------------------------------------------------------|--------------------|
+| `query_details`   | Collect queries information.                                 | yes                |
+| `schema_details`  | Collect schemas and tables from `information_schema`.        | yes                |
+| `query_samples`   | Collect query samples.                                       | yes                |
+| `setup_consumers` | Collect enabled `performance_schema.setup_consumers`.        | yes                |
+| `setup_actors`    | Check and update `performance_schema.setup_actors` settings. | yes                |
+| `locks`           | Collect queries that are waiting/blocking other queries.     | no                 |
+| `explain_plans`   | Collect explain plans information.                           | yes                |
 
 ## Blocks
 
@@ -55,6 +56,7 @@ You can use the following blocks with `database_observability.mysql`:
 | [`cloud_provider`][cloud_provider]   | Provide Cloud Provider information.               | no       |
 | `cloud_provider` > [`aws`][aws]      | Provide AWS database host information.            | no       |
 | [`setup_consumers`][setup_consumers] | Configure the `setup_consumers` collector.        | no       |
+| [`setup_actors`][setup_actors]       | Configure the `setup_actors` collector.           | no       |
 | [`query_details`][query_details]     | Configure the queries collector.                  | no       |
 | [`schema_details`][schema_details]   | Configure the schema and table details collector. | no       |
 | [`explain_plans`][explain_plans]     | Configure the explain plans collector.            | no       |
@@ -72,6 +74,7 @@ For example, `cloud_provider` > `aws` refers to a `aws` block defined inside an
 [explain_plans]: #explain_plans
 [locks]: #locks
 [query_samples]: #query_samples
+[setup_actors]: #setup_actors
 
 ### `cloud_provider`
 
@@ -135,6 +138,14 @@ The `aws` block supplies the [ARN](https://docs.aws.amazon.com/IAM/latest/UserGu
 | `auto_enable_setup_consumers`    | `boolean`  | Whether to enable some specific `performance_schema.setup_consumers` settings. | `false` | no       |
 | `setup_consumers_check_interval` | `duration` | How frequently to check if `setup_consumers` are correctly enabled.            | `"1h"`  | no       |
 
+### `setup_actors`
+
+| Name                       | Type       | Description                                                            | Default | Required |
+| -------------------------- | ---------- | ---------------------------------------------------------------------- | ------- | -------- |
+| `auto_update_setup_actors` | `boolean`  | Whether to enable updating `performance_schema.setup_actors` settings. | `false` | no       |
+| `collect_interval`         | `duration` | How frequently to check if `setup_actors` are configured correctly.    | `"1h"`  | no       |
+
+
 ## Example
 
 ```alloy

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

@@ -0,0 +1,174 @@
+package collector
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/go-kit/log"
+	"go.uber.org/atomic"
+
+	"github.com/grafana/alloy/internal/runtime/logging/level"
+)
+
+const (
+	SetupActorsCollector = "setup_actors"
+
+	selectUserQuery = `SELECT substring_index(current_user(), '@', 1)`
+
+	selectQuery = `SELECT enabled, history
+		FROM performance_schema.setup_actors
+		WHERE user = ?`
+
+	updateQuery = `UPDATE performance_schema.setup_actors
+		SET enabled='NO', history='NO'
+		WHERE user = ?`
+
+	insertQuery = `INSERT INTO performance_schema.setup_actors
+		(host, user, role, enabled, history)
+		VALUES ('%', ?, '%', 'NO', 'NO')`
+)
+
+type SetupActorsArguments struct {
+	DB                    *sql.DB
+	Logger                log.Logger
+	CollectInterval       time.Duration
+	AutoUpdateSetupActors bool
+}
+
+type SetupActors struct {
+	dbConnection          *sql.DB
+	collectInterval       time.Duration
+	autoUpdateSetupActors bool
+
+	logger  log.Logger
+	running *atomic.Bool
+	ctx     context.Context
+	cancel  context.CancelFunc
+}
+
+func NewSetupActors(args SetupActorsArguments) (*SetupActors, error) {
+	return &SetupActors{
+		dbConnection:          args.DB,
+		running:               &atomic.Bool{},
+		logger:                log.With(args.Logger, "collector", SetupActorsCollector),
+		collectInterval:       args.CollectInterval,
+		autoUpdateSetupActors: args.AutoUpdateSetupActors,
+	}, nil
+}
+
+func (c *SetupActors) Name() string {
+	return SetupActorsCollector
+}
+
+func (c *SetupActors) Start(ctx context.Context) error {
+	level.Debug(c.logger).Log("msg", "collector started")
+	c.running.Store(true)
+
+	ctx, cancel := context.WithCancel(ctx)
+	c.ctx = ctx
+	c.cancel = cancel
+
+	var user string
+	if err := c.dbConnection.QueryRowContext(ctx, selectUserQuery).Scan(&user); err != nil {
+		level.Error(c.logger).Log("msg", "failed to get current user", "err", err)
+		c.running.Store(false)
+		cancel()
+		return err
+	}
+
+	go func() {
+		defer func() {
+			c.Stop()
+			c.running.Store(false)
+		}()
+
+		ticker := time.NewTicker(c.collectInterval)
+
+		for {
+			if err := c.checkSetupActors(c.ctx, user); err != nil {
+				level.Error(c.logger).Log("msg", "collector error", "err", err)
+			}
+
+			select {
+			case <-c.ctx.Done():
+				return
+			case <-ticker.C:
+				// continue loop
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (c *SetupActors) Stopped() bool {
+	return !c.running.Load()
+}
+
+func (c *SetupActors) Stop() {
+	c.cancel()
+	c.running.Store(false)
+}
+
+func (c *SetupActors) checkSetupActors(ctx context.Context, user string) error {
+	var enabled, history string
+	err := c.dbConnection.QueryRowContext(ctx, selectQuery, user).Scan(&enabled, &history)
+	if errors.Is(err, sql.ErrNoRows) {
+		if c.autoUpdateSetupActors {
+			return c.insertSetupActors(ctx, user)
+		} else {
+			level.Info(c.logger).Log("msg", "setup_actors configuration missing, but auto-update is disabled")
+			return nil
+		}
+	} else if err != nil {
+		level.Error(c.logger).Log("msg", "failed to query setup_actors table", "err", err)
+		return err
+	}
+
+	if strings.ToUpper(enabled) != "NO" || strings.ToUpper(history) != "NO" {
+		if c.autoUpdateSetupActors {
+			return c.updateSetupActors(ctx, user, enabled, history)
+		} else {
+			level.Info(c.logger).Log("msg", "setup_actors configuration is not correct, but auto-update is disabled")
+			return nil
+		}
+	}
+
+	return nil
+}
+
+func (c *SetupActors) insertSetupActors(ctx context.Context, user string) error {
+	_, err := c.dbConnection.ExecContext(ctx, insertQuery, user)
+	if err != nil {
+		level.Error(c.logger).Log("msg", "failed to insert setup_actors row", "err", err, "user", user)
+		return err
+	}
+
+	level.Debug(c.logger).Log("msg", "inserted new setup_actors row", "user", user)
+	return nil
+}
+
+func (c *SetupActors) updateSetupActors(ctx context.Context, user string, enabled string, history string) error {
+	r, err := c.dbConnection.ExecContext(ctx, updateQuery, user)
+	if err != nil {
+		level.Error(c.logger).Log("msg", "failed to update setup_actors row", "err", err, "user", user)
+		return err
+	}
+
+	rowsAffected, err := r.RowsAffected()
+	if err != nil {
+		level.Error(c.logger).Log("msg", "failed to get rows affected from setup_actors update", "err", err)
+		return err
+	}
+	if rowsAffected == 0 {
+		level.Error(c.logger).Log("msg", "no rows affected from setup_actors update", "user", user)
+		return fmt.Errorf("no rows affected from setup_actors update")
+	}
+
+	level.Debug(c.logger).Log("msg", "updated setup_actors row", "rows_affected", rowsAffected, "previous_enabled", enabled, "previous_history", history, "user", user)
+	return nil
+}