|
4 | 4 | "context" |
5 | 5 | "errors" |
6 | 6 | "fmt" |
7 | | - "strconv" |
8 | 7 | "strings" |
9 | 8 | "time" |
10 | 9 |
|
@@ -38,10 +37,13 @@ func QueryMeasurements(ctx context.Context, dbUnique string, sql string, args .. |
38 | 37 | return nil, err |
39 | 38 | } |
40 | 39 | conn = tx |
| 40 | + } else { |
| 41 | + // we want simple protocol for non-postgres connections, e.g. pgpool |
| 42 | + args = append([]any{pgx.QueryExecModeSimpleProtocol}, args...) |
41 | 43 | } |
42 | 44 | rows, err := conn.Query(ctx, sql, args...) |
43 | 45 | if err == nil { |
44 | | - return pgx.CollectRows(rows, pgx.RowToMap) |
| 46 | + return pgx.CollectRows(rows, metrics.RowToMeasurement) |
45 | 47 | } |
46 | 48 | return nil, err |
47 | 49 | } |
@@ -461,103 +463,11 @@ func (r *Reaper) CheckForPGObjectChangesAndStore(ctx context.Context, dbUnique s |
461 | 463 | } |
462 | 464 | } |
463 | 465 |
|
464 | | -// some extra work needed as pgpool SHOW commands don't specify the return data types for some reason |
465 | | -func FetchMetricsPgpool(ctx context.Context, msg MetricFetchConfig, md *sources.SourceConn, mvp metrics.Metric) (metrics.Measurements, error) { |
466 | | - var retData = make(metrics.Measurements, 0) |
467 | | - epochNs := time.Now().UnixNano() |
468 | | - |
469 | | - sqlLines := strings.Split(strings.ToUpper(mvp.GetSQL(int(md.Version))), "\n") |
470 | | - |
471 | | - for _, sql := range sqlLines { |
472 | | - if strings.HasPrefix(sql, "SHOW POOL_NODES") { |
473 | | - data, err := QueryMeasurements(ctx, msg.DBUniqueName, sql) |
474 | | - if err != nil { |
475 | | - log.GetLogger(ctx).Errorf("[%s][%s] Could not fetch PgPool statistics: %v", msg.DBUniqueName, msg.MetricName, err) |
476 | | - return data, err |
477 | | - } |
478 | | - |
479 | | - for _, row := range data { |
480 | | - retRow := metrics.NewMeasurement(epochNs) |
481 | | - for k, v := range row { |
482 | | - vs := string(v.([]byte)) |
483 | | - // need 1 tag so that Influx would not merge rows |
484 | | - if k == "node_id" { |
485 | | - retRow["tag_node_id"] = vs |
486 | | - continue |
487 | | - } |
488 | | - |
489 | | - retRow[k] = vs |
490 | | - if k == "status" { // was changed from numeric to string at some pgpool version so leave the string |
491 | | - // but also add "status_num" field |
492 | | - switch vs { |
493 | | - case "up": |
494 | | - retRow["status_num"] = 1 |
495 | | - case "down": |
496 | | - retRow["status_num"] = 0 |
497 | | - default: |
498 | | - i, err := strconv.ParseInt(vs, 10, 64) |
499 | | - if err == nil { |
500 | | - retRow["status_num"] = i |
501 | | - } |
502 | | - } |
503 | | - continue |
504 | | - } |
505 | | - // everything is returned as text, so try to convert all numerics into ints / floats |
506 | | - if k != "lb_weight" { |
507 | | - i, err := strconv.ParseInt(vs, 10, 64) |
508 | | - if err == nil { |
509 | | - retRow[k] = i |
510 | | - continue |
511 | | - } |
512 | | - } |
513 | | - f, err := strconv.ParseFloat(vs, 64) |
514 | | - if err == nil { |
515 | | - retRow[k] = f |
516 | | - continue |
517 | | - } |
518 | | - } |
519 | | - retData = append(retData, retRow) |
520 | | - } |
521 | | - } else if strings.HasPrefix(sql, "SHOW POOL_PROCESSES") { |
522 | | - if len(retData) == 0 { |
523 | | - log.GetLogger(ctx).Warningf("[%s][%s] SHOW POOL_NODES needs to be placed before SHOW POOL_PROCESSES. ignoring SHOW POOL_PROCESSES", msg.DBUniqueName, msg.MetricName) |
524 | | - continue |
525 | | - } |
526 | | - |
527 | | - data, err := QueryMeasurements(ctx, msg.DBUniqueName, sql) |
528 | | - if err != nil { |
529 | | - log.GetLogger(ctx).Errorf("[%s][%s] Could not fetch PgPool statistics: %v", msg.DBUniqueName, msg.MetricName, err) |
530 | | - continue |
531 | | - } |
532 | | - |
533 | | - // summarize processesTotal / processes_active over all rows |
534 | | - processesTotal := 0 |
535 | | - processesActive := 0 |
536 | | - for _, row := range data { |
537 | | - processesTotal++ |
538 | | - v, ok := row["database"] |
539 | | - if !ok { |
540 | | - log.GetLogger(ctx).Infof("[%s][%s] column 'database' not found from data returned by SHOW POOL_PROCESSES, check pool version / SQL definition", msg.DBUniqueName, msg.MetricName) |
541 | | - continue |
542 | | - } |
543 | | - if len(v.([]byte)) > 0 { |
544 | | - processesActive++ |
545 | | - } |
546 | | - } |
547 | | - |
548 | | - for _, retRow := range retData { |
549 | | - retRow["processes_total"] = processesTotal |
550 | | - retRow["processes_active"] = processesActive |
551 | | - } |
552 | | - } |
553 | | - } |
554 | | - return retData, nil |
555 | | -} |
556 | | - |
557 | 466 | // Called once on daemon startup if some commonly wanted extension (most notably pg_stat_statements) is missing. |
558 | 467 | // With newer Postgres version can even succeed if the user is not a real superuser due to some cloud-specific |
559 | 468 | // whitelisting or "trusted extensions" (a feature from v13). Ignores errors. |
560 | 469 | func TryCreateMissingExtensions(ctx context.Context, dbUnique string, extensionNames []string, existingExtensions map[string]int) []string { |
| 470 | + // TODO: move to sources package and use direct pgx connection |
561 | 471 | sqlAvailable := `select name::text from pg_available_extensions` |
562 | 472 | extsCreated := make([]string, 0) |
563 | 473 |
|
@@ -595,6 +505,7 @@ func TryCreateMissingExtensions(ctx context.Context, dbUnique string, extensionN |
595 | 505 |
|
596 | 506 | // Called once on daemon startup to try to create "metric fething helper" functions automatically |
597 | 507 | func TryCreateMetricsFetchingHelpers(ctx context.Context, md *sources.SourceConn) (err error) { |
| 508 | + // TODO: replace with md.GetMetricDefs() and move to sources package |
598 | 509 | metricConfig := func() map[string]float64 { |
599 | 510 | if len(md.Metrics) > 0 { |
600 | 511 | return md.Metrics |
|
0 commit comments