master
go 150 lines 2.91 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 package sql
4
5 import (
6 "context"
7 "database/sql"
8 _ "embed"
9 "errors"
10 "sync"
11 "time"
12
13 "github.com/netdata/netdata/go/plugins/pkg/confopt"
14 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
15 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/cloudauth"
16 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/cloudauth/sqladapter"
17 )
18
19 //go:embed "config_schema.json"
20 var configSchema string
21
22 func init() {
23 collectorapi.Register("sql", collectorapi.Creator{
24 Create: func() collectorapi.CollectorV1 { return New() },
25 JobConfigSchema: configSchema,
26 Config: func() any { return &Config{} },
27 JobMethods: sqlJobMethods,
28 MethodHandler: sqlMethodHandler,
29 })
30 }
31
32 func New() *Collector {
33 return &Collector{
34 Config: Config{
35 Driver: "mysql",
36 Timeout: confopt.Duration(time.Second * 5),
37 },
38 charts: &collectorapi.Charts{},
39 seenCharts: make(map[string]bool),
40 }
41 }
42
43 type Collector struct {
44 collectorapi.Base
45 Config `yaml:",inline" json:""`
46
47 charts *collectorapi.Charts
48
49 dbMu sync.RWMutex
50 db *sql.DB
51 dbCtx context.Context
52 dbCancel context.CancelFunc
53
54 seenCharts map[string]bool
55
56 funcTable *funcTable
57
58 azureTokenProvider *cloudauth.TokenProvider
59 }
60
61 func (c *Collector) Configuration() any {
62 return c.Config
63 }
64
65 func (c *Collector) Charts() *collectorapi.Charts {
66 if c.Config.FunctionOnly {
67 return nil
68 }
69 return c.charts
70 }
71
72 func (c *Collector) Init(context.Context) error {
73 if err := c.validateConfig(); err != nil {
74 return err
75 }
76 if c.CloudAuth.IsProvider(cloudauth.ProviderAzureAD) && c.Driver == "pgx" {
77 cred, err := c.CloudAuth.NewCredential()
78 if err != nil {
79 return err
80 }
81 provider, err := cloudauth.NewTokenProvider(
82 cred,
83 []string{sqladapter.AzurePostgreSQLAADScope},
84 cloudauth.DefaultTokenRefreshMargin,
85 )
86 if err != nil {
87 return err
88 }
89 c.azureTokenProvider = provider
90 }
91
92 c.funcTable = newFuncTable(c)
93
94 return nil
95 }
96
97 func (c *Collector) Check(ctx context.Context) error {
98 if c.db == nil {
99 if err := c.openConnection(ctx); err != nil {
100 return err
101 }
102 // Create cancellable context for function queries
103 c.dbCtx, c.dbCancel = context.WithCancel(context.Background())
104 }
105
106 if c.Config.FunctionOnly {
107 return nil
108 }
109
110 mx, err := c.collect(ctx)
111 if err != nil {
112 return err
113 }
114 if len(mx) == 0 {
115 return errors.New("no metrics collected")
116 }
117 return nil
118 }
119
120 func (c *Collector) Collect(ctx context.Context) map[string]int64 {
121 if c.Config.FunctionOnly {
122 return nil
123 }
124
125 mx, err := c.collect(ctx)
126 if err != nil {
127 c.Error(err)
128 }
129
130 if len(mx) == 0 {
131 return nil
132 }
133 return mx
134 }
135
136 func (c *Collector) Cleanup(context.Context) {
137 // Cancel context first to signal in-flight queries to abort
138 if c.dbCancel != nil {
139 c.dbCancel()
140 }
141
142 // Acquire write lock - should be quick since queries are aborting
143 c.dbMu.Lock()
144 defer c.dbMu.Unlock()
145
146 if c.db != nil {
147 _ = c.db.Close()
148 c.db = nil
149 }
150 }