database_observability: make collectors configurable

CHANGELOG.md

@@ -34,6 +34,7 @@ Main (unreleased)
   - Better error handling for components (@cristiangreco)
   - Add namespace to `connection_info` metric (@cristiangreco)
   - Added table columns parsing (@cristiagreco)
+  - Add enable/disable collector configurability to `database_observability.mysql`. This removes the `query_samples_enabled` argument, now configurable via enable/disable collector. (@fridgepoet)
 
 ### Bugfixes
 
@@ -49,13 +50,6 @@ Main (unreleased)
 
 - Add support for pushv1.PusherService Connect API in `pyroscope.receive_http`. (@simonswine)
 
-- Add support for path prefixes in `pyroscope.scrape` to allow scraping targets behind a proxy or with custom URL paths. (@korniltsev)
-
-- Fixes godeltaprof hiding (renaming `godeltaprof_*` profile names to regular ones). (@korniltsev)
-
-- Change profile handling in `pyroscope.receive_http` and `pyroscope.write` components to use in-memory processing instead of pipes. (@marcsanmi)
-
-
 v1.6.1
 -----------------
 

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

@@ -23,12 +23,20 @@ database_observability.mysql "<LABEL>" {
 
 You can use the following arguments with `database_observability.mysql`:
 
-Name                    | Type                 | Description                                              | Default | Required
-------------------------|----------------------|----------------------------------------------------------|---------|---------
-`data_source_name`      | `secret`             | [Data Source Name][] for the MySQL server to connect to. |         | yes
-`forward_to`            | `list(LogsReceiver)` | Where to forward log entries after processing.           |         | yes
-`collect_interval`      | `duration`           | How frequently to collect information from database.     | `"1m"`  | no
-`query_samples_enabled` | `bool`               | Whether to enable collection of query samples.           | `true`  | no
+ Name                 | Type                 | Description                                               | Default | Required 
+----------------------|----------------------|-----------------------------------------------------------|---------|----------
+ `data_source_name`   | `secret`             | [Data Source Name][] for the MySQL server to connect to.  |         | yes      
+ `forward_to`         | `list(LogsReceiver)` | Where to forward log entries after processing.            |         | yes      
+ `collect_interval`   | `duration`           | How frequently to collect information from database.      | `"1m"`  | no       
+ `disable_collectors` | `list(string)`       | A list of collectors to disable from the default set.     |         | no       
+ `enable_collectors`  | `list(string)`       | A list of collectors to enable on top of the default set. |         | no       
+
+The following collectors are enabled by default:
+
+ Name           | Description                                           
+----------------|-------------------------------------------------------
+ `query_sample` | Collect query samples.                                
+ `schema_table` | Collect schemas and tables from `information_schema`. 
 
 ## Blocks
 

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

@@ -11,6 +11,8 @@ import (
 	"go.uber.org/atomic"
 )
 
+const ConnectionInfoName = "connection_info"
+
 var rdsRegex = regexp.MustCompile(`(?P<identifier>[^\.]+)\.([^\.]+)\.(?P<region>[^\.]+)\.rds\.amazonaws\.com`)
 
 type ConnectionInfoArguments struct {
@@ -44,7 +46,7 @@ func NewConnectionInfo(args ConnectionInfoArguments) (*ConnectionInfo, error) {
 }
 
 func (c *ConnectionInfo) Name() string {
-	return "ConnectionInfo"
+	return ConnectionInfoName
 }
 
 func (c *ConnectionInfo) Start(ctx context.Context) error {

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

@@ -21,6 +21,7 @@ import (
 const (
 	OP_QUERY_SAMPLE            = "query_sample"
 	OP_QUERY_PARSED_TABLE_NAME = "query_parsed_table_name"
+	QuerySampleName            = "query_sample"
 )
 
 const selectQuerySamples = `
@@ -67,7 +68,7 @@ func NewQuerySample(args QuerySampleArguments) (*QuerySample, error) {
 }
 
 func (c *QuerySample) Name() string {
-	return "QuerySample"
+	return QuerySampleName
 }
 
 func (c *QuerySample) Start(ctx context.Context) error {

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

@@ -24,6 +24,7 @@ const (
 	OP_SCHEMA_DETECTION = "schema_detection"
 	OP_TABLE_DETECTION  = "table_detection"
 	OP_CREATE_STATEMENT = "create_statement"
+	SchemaTableName     = "schema_table"
 )
 
 const (
@@ -127,7 +128,7 @@ func NewSchemaTable(args SchemaTableArguments) (*SchemaTable, error) {
 }
 
 func (c *SchemaTable) Name() string {
-	return "SchemaTable"
+	return SchemaTableName
 }
 
 func (c *SchemaTable) Start(ctx context.Context) error {

internal/component/database_observability/mysql/component.go

@@ -49,18 +49,16 @@ var (
 	_ syntax.Validator = (*Arguments)(nil)
 )
 
-// TODO(cristian) consider using something like "enabled_collectors"
-// to allow users to enable/disable collectors.
 type Arguments struct {
-	DataSourceName      alloytypes.Secret   `alloy:"data_source_name,attr"`
-	CollectInterval     time.Duration       `alloy:"collect_interval,attr,optional"`
-	QuerySamplesEnabled bool                `alloy:"query_samples_enabled,attr,optional"`
-	ForwardTo           []loki.LogsReceiver `alloy:"forward_to,attr"`
+	DataSourceName    alloytypes.Secret   `alloy:"data_source_name,attr"`
+	CollectInterval   time.Duration       `alloy:"collect_interval,attr,optional"`
+	ForwardTo         []loki.LogsReceiver `alloy:"forward_to,attr"`
+	EnableCollectors  []string            `alloy:"enable_collectors,attr,optional"`
+	DisableCollectors []string            `alloy:"disable_collectors,attr,optional"`
 }
 
 var DefaultArguments = Arguments{
-	CollectInterval:     1 * time.Minute,
-	QuerySamplesEnabled: true,
+	CollectInterval: 1 * time.Minute,
 }
 
 func (a *Arguments) SetToDefault() {
@@ -206,6 +204,27 @@ func (c *Component) Update(args component.Arguments) error {
 	return nil
 }
 
+func enableOrDisableCollectors(a Arguments) map[string]bool {
+	// configurable collectors and their default enabled/disabled value
+	collectors := map[string]bool{
+		collector.QuerySampleName: true,
+		collector.SchemaTableName: true,
+	}
+
+	for _, disabled := range a.DisableCollectors {
+		if _, ok := collectors[disabled]; ok {
+			collectors[disabled] = false
+		}
+	}
+	for _, enabled := range a.EnableCollectors {
+		if _, ok := collectors[enabled]; ok {
+			collectors[enabled] = true
+		}
+	}
+
+	return collectors
+}
+
 func (c *Component) startCollectors() error {
 	dbConnection, err := sql.Open("mysql", formatDSN(string(c.args.DataSourceName), "parseTime=true"))
 	if err != nil {
@@ -222,7 +241,9 @@ func (c *Component) startCollectors() error {
 
 	entryHandler := loki.NewEntryHandler(c.handler.Chan(), func() {})
 
-	if c.args.QuerySamplesEnabled {
+	collectors := enableOrDisableCollectors(c.args)
+
+	if collectors[collector.QuerySampleName] {
 		qsCollector, err := collector.NewQuerySample(collector.QuerySampleArguments{
 			DB:              dbConnection,
 			InstanceKey:     c.instanceKey,
@@ -241,23 +262,26 @@ func (c *Component) startCollectors() error {
 		c.collectors = append(c.collectors, qsCollector)
 	}
 
-	stCollector, err := collector.NewSchemaTable(collector.SchemaTableArguments{
-		DB:              dbConnection,
-		InstanceKey:     c.instanceKey,
-		CollectInterval: c.args.CollectInterval,
-		EntryHandler:    entryHandler,
-		Logger:          c.opts.Logger,
-	})
-	if err != nil {
-		level.Error(c.opts.Logger).Log("msg", "failed to create SchemaTable collector", "err", err)
-		return err
-	}
-	if err := stCollector.Start(context.Background()); err != nil {
-		level.Error(c.opts.Logger).Log("msg", "failed to start SchemaTable collector", "err", err)
-		return err
+	if collectors[collector.QuerySampleName] {
+		stCollector, err := collector.NewSchemaTable(collector.SchemaTableArguments{
+			DB:              dbConnection,
+			InstanceKey:     c.instanceKey,
+			CollectInterval: c.args.CollectInterval,
+			EntryHandler:    entryHandler,
+			Logger:          c.opts.Logger,
+		})
+		if err != nil {
+			level.Error(c.opts.Logger).Log("msg", "failed to create SchemaTable collector", "err", err)
+			return err
+		}
+		if err := stCollector.Start(context.Background()); err != nil {
+			level.Error(c.opts.Logger).Log("msg", "failed to start SchemaTable collector", "err", err)
+			return err
+		}
+		c.collectors = append(c.collectors, stCollector)
 	}
-	c.collectors = append(c.collectors, stCollector)
 
+	// Connection Info collector is always enabled
 	ciCollector, err := collector.NewConnectionInfo(collector.ConnectionInfoArguments{
 		DSN:      string(c.args.DataSourceName),
 		Registry: c.registry,

internal/component/database_observability/mysql/component_test.go

@@ -0,0 +1,130 @@
+package mysql
+
+import (
+	"testing"
+
+	"github.com/grafana/alloy/internal/component/database_observability/mysql/collector"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/alloy/syntax"
+)
+
+func Test_enableOrDisableCollectors(t *testing.T) {
+	t.Run("nothing specified (default behavior)", func(t *testing.T) {
+		var exampleDBO11yAlloyConfig = `
+		data_source_name = ""
+		forward_to = []
+	`
+
+		var args Arguments
+		err := syntax.Unmarshal([]byte(exampleDBO11yAlloyConfig), &args)
+		require.NoError(t, err)
+
+		actualCollectors := enableOrDisableCollectors(args)
+
+		assert.Equal(t, map[string]bool{
+			collector.QuerySampleName: true,
+			collector.SchemaTableName: true,
+		}, actualCollectors)
+	})
+
+	t.Run("enable collectors", func(t *testing.T) {
+		var exampleDBO11yAlloyConfig = `
+		data_source_name = ""
+		forward_to = []
+		enable_collectors = ["query_sample", "schema_table"]
+	`
+
+		var args Arguments
+		err := syntax.Unmarshal([]byte(exampleDBO11yAlloyConfig), &args)
+		require.NoError(t, err)
+
+		actualCollectors := enableOrDisableCollectors(args)
+
+		assert.Equal(t, map[string]bool{
+			collector.QuerySampleName: true,
+			collector.SchemaTableName: true,
+		}, actualCollectors)
+	})
+
+	t.Run("disable collectors", func(t *testing.T) {
+		var exampleDBO11yAlloyConfig = `
+		data_source_name = ""
+		forward_to = []
+		disable_collectors = ["query_sample", "schema_table"]
+	`
+
+		var args Arguments
+		err := syntax.Unmarshal([]byte(exampleDBO11yAlloyConfig), &args)
+		require.NoError(t, err)
+
+		actualCollectors := enableOrDisableCollectors(args)
+
+		assert.Equal(t, map[string]bool{
+			collector.QuerySampleName: false,
+			collector.SchemaTableName: false,
+		}, actualCollectors)
+	})
+
+	t.Run("enable collectors takes precedence over disable collectors", func(t *testing.T) {
+		var exampleDBO11yAlloyConfig = `
+		data_source_name = ""
+		forward_to = []
+		disable_collectors = ["query_sample", "schema_table"]
+		enable_collectors = ["query_sample", "schema_table"]
+	`
+
+		var args Arguments
+		err := syntax.Unmarshal([]byte(exampleDBO11yAlloyConfig), &args)
+		require.NoError(t, err)
+
+		actualCollectors := enableOrDisableCollectors(args)
+
+		assert.Equal(t, map[string]bool{
+			collector.QuerySampleName: true,
+			collector.SchemaTableName: true,
+		}, actualCollectors)
+	})
+
+	t.Run("enabling one and disabling one", func(t *testing.T) {
+		var exampleDBO11yAlloyConfig = `
+		data_source_name = ""
+		forward_to = []
+		disable_collectors = ["schema_table"]
+		enable_collectors = ["query_sample"]
+	`
+
+		var args Arguments
+		err := syntax.Unmarshal([]byte(exampleDBO11yAlloyConfig), &args)
+		require.NoError(t, err)
+
+		actualCollectors := enableOrDisableCollectors(args)
+
+		assert.Equal(t, map[string]bool{
+			collector.QuerySampleName: true,
+			collector.SchemaTableName: false,
+		}, actualCollectors)
+	})
+
+	t.Run("unknown collectors are ignored", func(t *testing.T) {
+		var exampleDBO11yAlloyConfig = `
+		data_source_name = ""
+		forward_to = []
+		enable_collectors = ["some_string"]
+		disable_collectors = ["another_string"]
+	`
+
+		var args Arguments
+		err := syntax.Unmarshal([]byte(exampleDBO11yAlloyConfig), &args)
+		require.NoError(t, err)
+
+		actualCollectors := enableOrDisableCollectors(args)
+
+		assert.Equal(t, map[string]bool{
+			collector.QuerySampleName: true,
+			collector.SchemaTableName: true,
+		}, actualCollectors)
+	})
+
+}