Skip to content

Commit 941c4f7

Browse files
cristiangrecocristiangreco
committed
database_observability: add auto-enable setup consumers to query_sample
Introduce a check in the `query_sample` collector to automatically enable the `events_statements_cpu` consumer in the `performance_schema.setup_consumers` table. This is guarded by two config options: - `allow_update_performance_schema_settings`: to enable "updates" to performance_schema settings altogether. - `auto_enable_setup_consumers`: to start auto-enabling of the setting specific to the `query_sample` collector.
1 parent 203f56e commit 941c4f7

File tree

2 files changed

+69
-33
lines changed

2 files changed

+69
-33
lines changed

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

Lines changed: 50 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -71,23 +71,30 @@ WHERE
7171
AND statements.CURRENT_SCHEMA NOT IN ('mysql', 'performance_schema', 'sys', 'information_schema')
7272
%s %s`
7373

74+
const updateSetupConsumers = `
75+
UPDATE performance_schema.setup_consumers
76+
SET enabled = 'yes'
77+
WHERE name = 'events_statements_cpu'`
78+
7479
type QuerySampleArguments struct {
75-
DB *sql.DB
76-
InstanceKey string
77-
CollectInterval time.Duration
78-
EntryHandler loki.EntryHandler
79-
DisableQueryRedaction bool
80+
DB *sql.DB
81+
InstanceKey string
82+
CollectInterval time.Duration
83+
EntryHandler loki.EntryHandler
84+
DisableQueryRedaction bool
85+
AutoEnableSetupConsumers bool
8086

8187
Logger log.Logger
8288
}
8389

8490
type QuerySample struct {
85-
dbConnection *sql.DB
86-
instanceKey string
87-
collectInterval time.Duration
88-
entryHandler loki.EntryHandler
89-
sqlParser parser.Parser
90-
disableQueryRedaction bool
91+
dbConnection *sql.DB
92+
instanceKey string
93+
collectInterval time.Duration
94+
entryHandler loki.EntryHandler
95+
sqlParser parser.Parser
96+
disableQueryRedaction bool
97+
autoEnableSetupConsumers bool
9198

9299
logger log.Logger
93100
running *atomic.Bool
@@ -100,14 +107,15 @@ type QuerySample struct {
100107

101108
func NewQuerySample(args QuerySampleArguments) (*QuerySample, error) {
102109
c := &QuerySample{
103-
dbConnection: args.DB,
104-
instanceKey: args.InstanceKey,
105-
collectInterval: args.CollectInterval,
106-
entryHandler: args.EntryHandler,
107-
sqlParser: parser.NewTiDBSqlParser(),
108-
disableQueryRedaction: args.DisableQueryRedaction,
109-
logger: log.With(args.Logger, "collector", QuerySampleName),
110-
running: &atomic.Bool{},
110+
dbConnection: args.DB,
111+
instanceKey: args.InstanceKey,
112+
collectInterval: args.CollectInterval,
113+
entryHandler: args.EntryHandler,
114+
sqlParser: parser.NewTiDBSqlParser(),
115+
disableQueryRedaction: args.DisableQueryRedaction,
116+
autoEnableSetupConsumers: args.AutoEnableSetupConsumers,
117+
logger: log.With(args.Logger, "collector", QuerySampleName),
118+
running: &atomic.Bool{},
111119
}
112120

113121
return c, nil
@@ -185,6 +193,12 @@ func (c *QuerySample) initializeBookmark(ctx context.Context) error {
185193
}
186194

187195
func (c *QuerySample) fetchQuerySamples(ctx context.Context) error {
196+
if c.autoEnableSetupConsumers {
197+
if err := c.checkSetupConsumersSettings(ctx); err != nil {
198+
return err
199+
}
200+
}
201+
188202
timeRow := c.dbConnection.QueryRowContext(ctx, selectNowAndUptime)
189203

190204
var now, uptime float64
@@ -382,6 +396,23 @@ func (c *QuerySample) fetchQuerySamples(ctx context.Context) error {
382396
return nil
383397
}
384398

399+
func (c *QuerySample) checkSetupConsumersSettings(ctx context.Context) error {
400+
rs, err := c.dbConnection.ExecContext(ctx, updateSetupConsumers)
401+
if err != nil {
402+
level.Error(c.logger).Log("msg", "failed to update performance_schema.setup_consumers", "err", err)
403+
return err
404+
}
405+
406+
rowsAffected, err := rs.RowsAffected()
407+
if err != nil {
408+
level.Error(c.logger).Log("msg", "failed to get rows affected from performance_schema.setup_consumers", "err", err)
409+
return err
410+
}
411+
level.Debug(c.logger).Log("msg", "updated performance_schema.setup_consumers", "rows_affected", rowsAffected)
412+
413+
return nil
414+
}
415+
385416
func (c *QuerySample) calculateWallTime(serverStartTime, timer float64) float64 {
386417
// timer indicates event timing since server startup.
387418
// The timer value is in picoseconds with a column type of bigint unsigned. This value can overflow after about ~213 days.

‎internal/component/database_observability/mysql/component.go

Lines changed: 19 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -50,11 +50,12 @@ var (
5050
)
5151

5252
type Arguments struct {
53-
DataSourceName alloytypes.Secret `alloy:"data_source_name,attr"`
54-
CollectInterval time.Duration `alloy:"collect_interval,attr,optional"`
55-
ForwardTo []loki.LogsReceiver `alloy:"forward_to,attr"`
56-
EnableCollectors []string `alloy:"enable_collectors,attr,optional"`
57-
DisableCollectors []string `alloy:"disable_collectors,attr,optional"`
53+
DataSourceName alloytypes.Secret `alloy:"data_source_name,attr"`
54+
CollectInterval time.Duration `alloy:"collect_interval,attr,optional"`
55+
ForwardTo []loki.LogsReceiver `alloy:"forward_to,attr"`
56+
EnableCollectors []string `alloy:"enable_collectors,attr,optional"`
57+
DisableCollectors []string `alloy:"disable_collectors,attr,optional"`
58+
AllowUpdatePerfSchemaSettings bool `alloy:"allow_update_performance_schema_settings,attr,optional"`
5859

5960
// collector: 'setup_consumers'
6061
SetupConsumersCollectInterval time.Duration `alloy:"setup_consumers_collect_interval,attr,optional"`
@@ -69,11 +70,13 @@ type Arguments struct {
6970
LocksThreshold time.Duration `alloy:"locks_threshold,attr,optional"`
7071

7172
// collector: 'query_sample'
72-
DisableQueryRedaction bool `alloy:"disable_query_redaction,attr,optional"`
73+
DisableQueryRedaction bool `alloy:"disable_query_redaction,attr,optional"`
74+
AutoEnableSetupConsumers bool `alloy:"auto_enable_setup_consumers,attr,optional"`
7375
}
7476

7577
var DefaultArguments = Arguments{
76-
CollectInterval: 1 * time.Minute,
78+
CollectInterval: 1 * time.Minute,
79+
AllowUpdatePerfSchemaSettings: false,
7780

7881
// collector: 'setup_consumers'
7982
SetupConsumersCollectInterval: 1 * time.Hour,
@@ -88,7 +91,8 @@ var DefaultArguments = Arguments{
8891
LocksThreshold: 1 * time.Second,
8992

9093
// collector: 'query_sample'
91-
DisableQueryRedaction: false,
94+
DisableQueryRedaction: false,
95+
AutoEnableSetupConsumers: false,
9296
}
9397

9498
func (a *Arguments) SetToDefault() {
@@ -322,12 +326,13 @@ func (c *Component) startCollectors() error {
322326

323327
if collectors[collector.QuerySampleName] {
324328
qsCollector, err := collector.NewQuerySample(collector.QuerySampleArguments{
325-
DB: dbConnection,
326-
InstanceKey: c.instanceKey,
327-
CollectInterval: c.args.CollectInterval,
328-
EntryHandler: entryHandler,
329-
Logger: c.opts.Logger,
330-
DisableQueryRedaction: c.args.DisableQueryRedaction,
329+
DB: dbConnection,
330+
InstanceKey: c.instanceKey,
331+
CollectInterval: c.args.CollectInterval,
332+
EntryHandler: entryHandler,
333+
Logger: c.opts.Logger,
334+
DisableQueryRedaction: c.args.DisableQueryRedaction,
335+
AutoEnableSetupConsumers: c.args.AllowUpdatePerfSchemaSettings && c.args.AutoEnableSetupConsumers,
331336
})
332337
if err != nil {
333338
level.Error(c.opts.Logger).Log("msg", "failed to create QuerySample collector", "err", err)

0 commit comments

Comments
 (0)