| 1 | //go:build cgo |
| 2 | |
| 3 | package as400 |
| 4 | |
| 5 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 6 | |
| 7 | import ( |
| 8 | "context" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "maps" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | as400proto "github.com/netdata/netdata/go/plugins/plugin/ibm.d/protocols/as400" |
| 16 | ) |
| 17 | |
| 18 | const ( |
| 19 | queryNameMessageQueueTotals = "message_queue_totals" |
| 20 | queryNameJobQueueTotals = "job_queue_totals" |
| 21 | queryNameOutputQueueTotals = "output_queue_totals" |
| 22 | ) |
| 23 | |
| 24 | type batchPathConfig struct { |
| 25 | enabled bool |
| 26 | interval time.Duration |
| 27 | maxConnections int |
| 28 | } |
| 29 | |
| 30 | type queueTotalsSnapshot struct { |
| 31 | timestamp time.Time |
| 32 | err error |
| 33 | queues map[string]int64 |
| 34 | items map[string]int64 |
| 35 | } |
| 36 | |
| 37 | type batchCache struct { |
| 38 | mu sync.RWMutex |
| 39 | totals queueTotalsSnapshot |
| 40 | latency latencyCache |
| 41 | } |
| 42 | |
| 43 | func (c *Collector) batchTotalsEnabled() bool { |
| 44 | return c != nil && (c.CollectMessageQueueTotals.IsEnabled() || c.CollectJobQueueTotals.IsEnabled() || c.CollectOutputQueueTotals.IsEnabled()) |
| 45 | } |
| 46 | |
| 47 | func (c *Collector) startBatchPath() error { |
| 48 | c.stopBatchPath() |
| 49 | |
| 50 | totalsEnabled := c.batchTotalsEnabled() |
| 51 | if !c.BatchPath && totalsEnabled { |
| 52 | c.Infof("batch path not started: batch_path is disabled while totals are enabled (message=%s job=%s output=%s)", |
| 53 | c.CollectMessageQueueTotals.String(), c.CollectJobQueueTotals.String(), c.CollectOutputQueueTotals.String()) |
| 54 | } |
| 55 | |
| 56 | cfg := batchPathConfig{ |
| 57 | enabled: c.BatchPath && totalsEnabled, |
| 58 | interval: time.Duration(c.BatchPathUpdateEvery), |
| 59 | maxConnections: c.BatchPathMaxConnections, |
| 60 | } |
| 61 | |
| 62 | if cfg.interval <= 0 { |
| 63 | cfg.interval = time.Minute |
| 64 | } |
| 65 | if cfg.interval < time.Minute { |
| 66 | c.Warningf("batch path update every %s is shorter than 1m; using 1m", cfg.interval) |
| 67 | cfg.interval = time.Minute |
| 68 | } |
| 69 | if cfg.maxConnections <= 0 { |
| 70 | cfg.maxConnections = 1 |
| 71 | } |
| 72 | |
| 73 | c.batch.config = cfg |
| 74 | |
| 75 | if !cfg.enabled { |
| 76 | c.Infof("batch path not started: totals-enabled=%t (message=%s job=%s output=%s) batch_path=%t", |
| 77 | totalsEnabled, |
| 78 | c.CollectMessageQueueTotals.String(), c.CollectJobQueueTotals.String(), c.CollectOutputQueueTotals.String(), |
| 79 | c.BatchPath) |
| 80 | return nil |
| 81 | } |
| 82 | |
| 83 | clientCfg := as400proto.Config{ |
| 84 | DSN: c.DSN, |
| 85 | Timeout: time.Duration(c.Timeout), |
| 86 | MaxOpenConns: cfg.maxConnections, |
| 87 | } |
| 88 | |
| 89 | client := as400proto.NewClient(clientCfg) |
| 90 | ctx := context.Background() |
| 91 | if err := client.Connect(ctx); err != nil { |
| 92 | return fmt.Errorf("batch path: connect failed: %w", err) |
| 93 | } |
| 94 | if err := client.Ping(ctx); err != nil { |
| 95 | _ = client.Close() |
| 96 | return fmt.Errorf("batch path: ping failed: %w", err) |
| 97 | } |
| 98 | |
| 99 | runCtx, cancel := context.WithCancel(context.Background()) |
| 100 | c.batch.client = client |
| 101 | c.batch.cancel = cancel |
| 102 | c.batch.wg.Add(1) |
| 103 | go c.runBatchPath(runCtx) |
| 104 | c.Infof("batch path worker started (interval=%s, max_conns=%d)", cfg.interval, cfg.maxConnections) |
| 105 | return nil |
| 106 | } |
| 107 | |
| 108 | func (c *Collector) stopBatchPath() { |
| 109 | if c.batch.cancel != nil { |
| 110 | c.batch.cancel() |
| 111 | } |
| 112 | c.batch.wg.Wait() |
| 113 | if c.batch.client != nil { |
| 114 | if err := c.batch.client.Close(); err != nil { |
| 115 | c.Errorf("batch path: closing client failed: %v", err) |
| 116 | } |
| 117 | } |
| 118 | c.batch.cancel = nil |
| 119 | c.batch.client = nil |
| 120 | c.batch.config = batchPathConfig{} |
| 121 | } |
| 122 | |
| 123 | func (c *Collector) runBatchPath(ctx context.Context) { |
| 124 | defer c.batch.wg.Done() |
| 125 | |
| 126 | interval := c.batch.config.interval |
| 127 | if interval <= 0 { |
| 128 | interval = time.Minute |
| 129 | } |
| 130 | |
| 131 | now := time.Now() |
| 132 | beat := now |
| 133 | c.runBatchCollectors(ctx, beat) |
| 134 | nextBeat := beat.Add(interval) |
| 135 | |
| 136 | for { |
| 137 | sleep := time.Until(nextBeat) |
| 138 | if sleep > 0 { |
| 139 | timer := time.NewTimer(sleep) |
| 140 | select { |
| 141 | case <-ctx.Done(): |
| 142 | timer.Stop() |
| 143 | return |
| 144 | case <-timer.C: |
| 145 | } |
| 146 | } else { |
| 147 | select { |
| 148 | case <-ctx.Done(): |
| 149 | return |
| 150 | default: |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | beat = nextBeat |
| 155 | c.runBatchCollectors(ctx, beat) |
| 156 | |
| 157 | nextBeat = nextBeat.Add(interval) |
| 158 | now = time.Now() |
| 159 | for nextBeat.Before(now) { |
| 160 | nextBeat = nextBeat.Add(interval) |
| 161 | } |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | func (c *Collector) runBatchCollectors(ctx context.Context, beat time.Time) { |
| 166 | if ctx.Err() != nil { |
| 167 | return |
| 168 | } |
| 169 | |
| 170 | c.batch.cache.beginLatencyCycle(beat) |
| 171 | |
| 172 | if !c.batchTotalsEnabled() { |
| 173 | c.batch.cache.setTotals(queueTotalsSnapshot{timestamp: beat}) |
| 174 | return |
| 175 | } |
| 176 | |
| 177 | snapshot, err := c.fetchQueueTotals(ctx, beat, c.batchDoQueryRow) |
| 178 | c.batch.cache.setTotals(snapshot) |
| 179 | if err != nil && !errors.Is(err, context.Canceled) { |
| 180 | c.logErrorOnce("batch_path_error", "batch path: %s", trimDriverMessage(err)) |
| 181 | } else if err == nil { |
| 182 | c.clearErrorOnce("batch_path_error") |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | func (c *Collector) fetchQueueTotals(ctx context.Context, beat time.Time, do queryRowFunc) (queueTotalsSnapshot, error) { |
| 187 | snapshot := queueTotalsSnapshot{ |
| 188 | timestamp: beat, |
| 189 | queues: make(map[string]int64), |
| 190 | items: make(map[string]int64), |
| 191 | } |
| 192 | |
| 193 | var firstErr error |
| 194 | |
| 195 | if c.CollectMessageQueueTotals.IsEnabled() { |
| 196 | var messageCount, queueCount int64 |
| 197 | err := do(ctx, queryNameMessageQueueTotals, queryMessageQueueTotals, func(column, value string) { |
| 198 | switch column { |
| 199 | case "MESSAGE_COUNT": |
| 200 | messageCount = parseInt64OrZero(value) |
| 201 | case "QUEUE_COUNT": |
| 202 | queueCount = parseInt64OrZero(value) |
| 203 | } |
| 204 | }) |
| 205 | if err != nil { |
| 206 | c.logQueryErrorOnce("batch_message_queue_totals", queryMessageQueueTotals, err) |
| 207 | if firstErr == nil { |
| 208 | firstErr = fmt.Errorf("message queue totals: %w", err) |
| 209 | } |
| 210 | } else { |
| 211 | c.clearErrorOnce("batch_message_queue_totals") |
| 212 | snapshot.queues["message_queue"] = queueCount |
| 213 | snapshot.items["message_queue"] = messageCount |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | if c.CollectJobQueueTotals.IsEnabled() { |
| 218 | var queueCount, jobCount int64 |
| 219 | err := do(ctx, queryNameJobQueueTotals, queryJobQueueTotals, func(column, value string) { |
| 220 | switch column { |
| 221 | case "QUEUE_COUNT": |
| 222 | queueCount = parseInt64OrZero(value) |
| 223 | case "JOB_COUNT": |
| 224 | jobCount = parseInt64OrZero(value) |
| 225 | } |
| 226 | }) |
| 227 | if err != nil { |
| 228 | c.logQueryErrorOnce("batch_job_queue_totals", queryJobQueueTotals, err) |
| 229 | if firstErr == nil { |
| 230 | firstErr = fmt.Errorf("job queue totals: %w", err) |
| 231 | } |
| 232 | } else { |
| 233 | c.clearErrorOnce("batch_job_queue_totals") |
| 234 | snapshot.queues["job_queue"] = queueCount |
| 235 | snapshot.items["job_queue"] = jobCount |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | if c.CollectOutputQueueTotals.IsEnabled() { |
| 240 | var queueCount, fileCount int64 |
| 241 | err := do(ctx, queryNameOutputQueueTotals, queryOutputQueueTotals, func(column, value string) { |
| 242 | switch column { |
| 243 | case "QUEUE_COUNT": |
| 244 | queueCount = parseInt64OrZero(value) |
| 245 | case "FILE_COUNT": |
| 246 | fileCount = parseInt64OrZero(value) |
| 247 | } |
| 248 | }) |
| 249 | if err != nil { |
| 250 | c.logQueryErrorOnce("batch_output_queue_totals", queryOutputQueueTotals, err) |
| 251 | if firstErr == nil { |
| 252 | firstErr = fmt.Errorf("output queue totals: %w", err) |
| 253 | } |
| 254 | } else { |
| 255 | c.clearErrorOnce("batch_output_queue_totals") |
| 256 | snapshot.queues["output_queue"] = queueCount |
| 257 | snapshot.items["output_queue"] = fileCount |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | snapshot.err = firstErr |
| 262 | return snapshot, snapshot.err |
| 263 | } |
| 264 | |
| 265 | func (c *Collector) batchDoQueryRow(ctx context.Context, queryName, query string, assign func(column, value string)) error { |
| 266 | if c.batch.client == nil { |
| 267 | return errors.New("batch path client not initialised") |
| 268 | } |
| 269 | |
| 270 | start := time.Now() |
| 271 | err := c.queryRowWithClient(ctx, c.batch.client, queryName, query, assign) |
| 272 | elapsed := time.Since(start) |
| 273 | latency := elapsed.Microseconds() |
| 274 | if latency == 0 { |
| 275 | latency = 1 |
| 276 | } |
| 277 | c.batch.cache.addLatency(queryName, latency) |
| 278 | c.Debugf("batch recorded %s=%dµs", queryName, latency) |
| 279 | return err |
| 280 | } |
| 281 | |
| 282 | func (c *Collector) batchPathActive() bool { |
| 283 | return c != nil && c.batch.config.enabled && c.batch.client != nil |
| 284 | } |
| 285 | |
| 286 | func (c *Collector) batchPathIntervalSeconds() int { |
| 287 | if c == nil { |
| 288 | return 0 |
| 289 | } |
| 290 | if !c.batchPathActive() { |
| 291 | return 0 |
| 292 | } |
| 293 | interval := int(c.batch.config.interval / time.Second) |
| 294 | if interval < 1 { |
| 295 | interval = 60 |
| 296 | } |
| 297 | return interval |
| 298 | } |
| 299 | |
| 300 | func (c *batchCache) beginLatencyCycle(ts time.Time) { |
| 301 | c.latency.beginCycle(ts) |
| 302 | } |
| 303 | |
| 304 | func (c *batchCache) addLatency(name string, value int64) { |
| 305 | c.latency.add(name, value) |
| 306 | } |
| 307 | |
| 308 | func (c *batchCache) setTotals(snapshot queueTotalsSnapshot) { |
| 309 | c.mu.Lock() |
| 310 | c.totals = snapshot |
| 311 | c.mu.Unlock() |
| 312 | } |
| 313 | |
| 314 | func (c *batchCache) getTotals() queueTotalsSnapshot { |
| 315 | c.mu.RLock() |
| 316 | defer c.mu.RUnlock() |
| 317 | return cloneQueueTotalsSnapshot(c.totals) |
| 318 | } |
| 319 | |
| 320 | func (c *batchCache) getLatencies() (map[string]int64, time.Time) { |
| 321 | return c.latency.snapshot() |
| 322 | } |
| 323 | |
| 324 | func cloneQueueTotalsSnapshot(src queueTotalsSnapshot) queueTotalsSnapshot { |
| 325 | dst := queueTotalsSnapshot{ |
| 326 | timestamp: src.timestamp, |
| 327 | err: src.err, |
| 328 | } |
| 329 | if src.queues != nil { |
| 330 | dst.queues = make(map[string]int64, len(src.queues)) |
| 331 | maps.Copy(dst.queues, src.queues) |
| 332 | } |
| 333 | if src.items != nil { |
| 334 | dst.items = make(map[string]int64, len(src.items)) |
| 335 | maps.Copy(dst.items, src.items) |
| 336 | } |
| 337 | return dst |
| 338 | } |