@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
go 364 lines 9.33 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 //go:build cgo
4
5 package db2
6
7 import (
8 "context"
9 "fmt"
10 "strconv"
11 "strings"
12 )
13
14 // collectMemoryPoolInstances collects memory pool metrics
15 func (c *Collector) collectMemoryPoolInstances(ctx context.Context) error {
16 // Initialize maps if needed
17 if c.mx.memoryPools == nil {
18 c.mx.memoryPools = make(map[string]memoryPoolInstanceMetrics)
19 }
20
21 // Mark all pools as not updated
22 for _, pool := range c.memoryPools {
23 pool.updated = false
24 }
25
26 var currentPoolType string
27 var currentMetrics memoryPoolInstanceMetrics
28
29 err := c.doQuery(ctx, queryMonGetMemoryPool, func(column, value string, lineEnd bool) {
30 switch column {
31 case "MEMORY_POOL_TYPE":
32 currentPoolType = cleanName(value)
33
34 // Create pool metadata if new
35 if _, exists := c.memoryPools[currentPoolType]; !exists {
36 c.memoryPools[currentPoolType] = &memoryPoolMetrics{
37 poolType: value,
38 }
39 }
40 c.memoryPools[currentPoolType].updated = true
41
42 case "MEMORY_POOL_USED":
43 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
44 currentMetrics.PoolUsed = v
45 }
46
47 case "MEMORY_POOL_USED_HWM":
48 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
49 currentMetrics.PoolUsedHWM = v
50 }
51 }
52
53 // At end of row, save metrics
54 if lineEnd && currentPoolType != "" {
55 c.mx.memoryPools[currentPoolType] = currentMetrics
56 currentPoolType = ""
57 currentMetrics = memoryPoolInstanceMetrics{}
58 }
59 })
60
61 if err != nil {
62 return err
63 }
64
65 // Remove stale pools
66 for poolType, pool := range c.memoryPools {
67 if !pool.updated {
68 delete(c.memoryPools, poolType)
69 delete(c.mx.memoryPools, poolType)
70 }
71 }
72
73 return nil
74 }
75
76 // collectConnectionWaits collects enhanced wait metrics for connections
77 func (c *Collector) collectConnectionWaits(ctx context.Context) error {
78 // This enhances existing connection metrics with wait time details
79 // We add wait metrics to the existing connection instance metrics
80
81 if !c.CollectWaitMetrics || c.MaxConnections <= 0 {
82 return nil
83 }
84
85 // Build query to get wait metrics from MON_GET_CONNECTION
86 query := `
87 SELECT
88 APPLICATION_ID,
89 TOTAL_WAIT_TIME,
90 LOCK_WAIT_TIME,
91 LOG_DISK_WAIT_TIME,
92 LOG_BUFFER_WAIT_TIME,
93 POOL_READ_TIME,
94 POOL_WRITE_TIME,
95 DIRECT_READ_TIME,
96 DIRECT_WRITE_TIME,
97 FCM_RECV_WAIT_TIME,
98 FCM_SEND_WAIT_TIME,
99 TOTAL_ROUTINE_TIME,
100 TOTAL_COMPILE_TIME,
101 TOTAL_SECTION_TIME,
102 TOTAL_COMMIT_TIME,
103 TOTAL_ROLLBACK_TIME
104 FROM TABLE(MON_GET_CONNECTION(NULL,-2)) AS T
105 -- WHERE APPLICATION_HANDLE IS NOT NULL
106 ORDER BY TOTAL_WAIT_TIME DESC
107 `
108
109 var currentAppID string
110 var waitMetrics struct {
111 TotalWaitTime int64
112 LockWaitTime int64
113 LogDiskWaitTime int64
114 LogBufferWaitTime int64
115 PoolReadTime int64
116 PoolWriteTime int64
117 DirectReadTime int64
118 DirectWriteTime int64
119 FCMRecvWaitTime int64
120 FCMSendWaitTime int64
121 TotalRoutineTime int64
122 TotalCompileTime int64
123 TotalSectionTime int64
124 TotalCommitTime int64
125 TotalRollbackTime int64
126 }
127
128 err := c.doQuery(ctx, query, func(column, value string, lineEnd bool) {
129 switch column {
130 case "APPLICATION_ID":
131 currentAppID = strings.TrimSpace(value)
132
133 case "TOTAL_WAIT_TIME":
134 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
135 waitMetrics.TotalWaitTime = v
136 }
137
138 case "LOCK_WAIT_TIME":
139 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
140 waitMetrics.LockWaitTime = v
141 }
142
143 case "LOG_DISK_WAIT_TIME":
144 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
145 waitMetrics.LogDiskWaitTime = v
146 }
147
148 case "LOG_BUFFER_WAIT_TIME":
149 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
150 waitMetrics.LogBufferWaitTime = v
151 }
152
153 case "POOL_READ_TIME":
154 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
155 waitMetrics.PoolReadTime = v
156 }
157
158 case "POOL_WRITE_TIME":
159 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
160 waitMetrics.PoolWriteTime = v
161 }
162
163 case "DIRECT_READ_TIME":
164 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
165 waitMetrics.DirectReadTime = v
166 }
167
168 case "DIRECT_WRITE_TIME":
169 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
170 waitMetrics.DirectWriteTime = v
171 }
172
173 case "FCM_RECV_WAIT_TIME":
174 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
175 waitMetrics.FCMRecvWaitTime = v
176 }
177
178 case "FCM_SEND_WAIT_TIME":
179 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
180 waitMetrics.FCMSendWaitTime = v
181 }
182
183 case "TOTAL_ROUTINE_TIME":
184 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
185 waitMetrics.TotalRoutineTime = v
186 }
187
188 case "TOTAL_COMPILE_TIME":
189 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
190 waitMetrics.TotalCompileTime = v
191 }
192
193 case "TOTAL_SECTION_TIME":
194 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
195 waitMetrics.TotalSectionTime = v
196 }
197
198 case "TOTAL_COMMIT_TIME":
199 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
200 waitMetrics.TotalCommitTime = v
201 }
202
203 case "TOTAL_ROLLBACK_TIME":
204 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
205 waitMetrics.TotalRollbackTime = v
206 }
207 }
208
209 // At end of row, add wait metrics to existing connection metrics
210 if lineEnd && currentAppID != "" {
211 // Only update if we already track this connection
212 if metrics, exists := c.mx.connections[currentAppID]; exists {
213 // Add wait time metrics to existing connection metrics
214 metrics.TotalWaitTime = waitMetrics.TotalWaitTime
215 metrics.LockWaitTime = waitMetrics.LockWaitTime
216 metrics.LogDiskWaitTime = waitMetrics.LogDiskWaitTime
217 metrics.LogBufferWaitTime = waitMetrics.LogBufferWaitTime
218 metrics.PoolReadTime = waitMetrics.PoolReadTime
219 metrics.PoolWriteTime = waitMetrics.PoolWriteTime
220 metrics.DirectReadTime = waitMetrics.DirectReadTime
221 metrics.DirectWriteTime = waitMetrics.DirectWriteTime
222 metrics.FCMRecvWaitTime = waitMetrics.FCMRecvWaitTime
223 metrics.FCMSendWaitTime = waitMetrics.FCMSendWaitTime
224 metrics.TotalRoutineTime = waitMetrics.TotalRoutineTime
225 metrics.TotalCompileTime = waitMetrics.TotalCompileTime
226 metrics.TotalSectionTime = waitMetrics.TotalSectionTime
227 metrics.TotalCommitTime = waitMetrics.TotalCommitTime
228 metrics.TotalRollbackTime = waitMetrics.TotalRollbackTime
229
230 c.mx.connections[currentAppID] = metrics
231 }
232
233 // Reset for next row
234 currentAppID = ""
235 waitMetrics = struct {
236 TotalWaitTime int64
237 LockWaitTime int64
238 LogDiskWaitTime int64
239 LogBufferWaitTime int64
240 PoolReadTime int64
241 PoolWriteTime int64
242 DirectReadTime int64
243 DirectWriteTime int64
244 FCMRecvWaitTime int64
245 FCMSendWaitTime int64
246 TotalRoutineTime int64
247 TotalCompileTime int64
248 TotalSectionTime int64
249 TotalCommitTime int64
250 TotalRollbackTime int64
251 }{}
252 }
253 })
254
255 if err != nil {
256 return fmt.Errorf("failed to collect connection wait metrics: %w", err)
257 }
258
259 return nil
260 }
261
262 // collectTableIOInstances collects table I/O statistics
263 func (c *Collector) collectTableIOInstances(ctx context.Context) error {
264 if c.MaxTables <= 0 {
265 return nil
266 }
267
268 // Initialize maps if needed
269 if c.mx.tableIOs == nil {
270 c.mx.tableIOs = make(map[string]tableIOInstanceMetrics)
271 }
272
273 c.Debugf("collectTableIOInstances: starting collection, MaxTables=%d", c.MaxTables)
274
275 // Mark all tables as not updated for I/O metrics
276 for _, table := range c.tables {
277 table.ioUpdated = false
278 }
279
280 query := queryMonGetTable
281
282 var currentTableName string
283 var currentMetrics tableIOInstanceMetrics
284 collected := 0
285
286 err := c.doQuery(ctx, query, func(column, value string, lineEnd bool) {
287 switch column {
288 case "TABSCHEMA":
289 if value != "" {
290 currentTableName = strings.TrimSpace(value) + "."
291 }
292
293 case "TABNAME":
294 currentTableName += strings.TrimSpace(value)
295 cleanName := cleanName(currentTableName)
296
297 // Create table metadata if new
298 if _, exists := c.tables[cleanName]; !exists {
299 c.tables[cleanName] = &tableMetrics{
300 name: currentTableName,
301 }
302 }
303 c.tables[cleanName].ioUpdated = true
304 collected++
305
306 case "TABLE_SCANS":
307 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
308 currentMetrics.TableScans = v
309 }
310
311 case "ROWS_READ":
312 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
313 currentMetrics.RowsRead = v
314 }
315
316 case "ROWS_INSERTED":
317 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
318 currentMetrics.RowsInserted = v
319 }
320
321 case "ROWS_UPDATED":
322 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
323 currentMetrics.RowsUpdated = v
324 }
325
326 case "ROWS_DELETED":
327 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
328 currentMetrics.RowsDeleted = v
329 }
330
331 case "OVERFLOW_ACCESSES":
332 if v, err := strconv.ParseInt(value, 10, 64); err == nil {
333 currentMetrics.OverflowAccesses = v
334 }
335 }
336
337 // At end of row, save metrics
338 if lineEnd && currentTableName != "" {
339 cleanName := cleanName(currentTableName)
340 c.mx.tableIOs[cleanName] = currentMetrics
341 currentTableName = ""
342 currentMetrics = tableIOInstanceMetrics{}
343 }
344 })
345
346 if err != nil {
347 return err
348 }
349
350 // Remove stale table I/O metrics
351 for name, table := range c.tables {
352 if !table.ioUpdated {
353 delete(c.mx.tableIOs, name)
354 }
355 }
356
357 if collected == c.MaxTables {
358 c.Debugf("reached max_tables limit (%d) for I/O metrics, some tables may not be collected", c.MaxTables)
359 }
360
361 c.Debugf("collectTableIOInstances: completed, collected=%d tables, tableIOs map size=%d", collected, len(c.mx.tableIOs))
362
363 return nil
364 }