9208786291

Committed 23 May 2024 01:19PM UTC coverage: 12.682% (+3.0%) from 9.667%

Build # 9208786291

Build Type

push

github

Committed by

web-flow

Commit Message

[+] add tests for `sinks` (#449)

* [+] add tests for `multiwriter`
* [*] prepare postgres writer for tests
* [+] add `TestNewWriterFromPostgresConn()`
* [+] add tests for `JSONWriter`
* [+] add `TestSyncMetric` and `TestWrite`
* [*] remove obsolete `EnsureMetric()`
* [*] prettify code

Run Details

38 of 84 new or added lines in 5 files covered. (45.24%)

156 existing lines in 5 files now uncovered.

584 of 4605 relevant lines covered (12.68%)

0.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

12.07

/src/db/bootstrap.go

package db

import (
        "context"
        "time"

        "github.com/cybertec-postgresql/pgwatch3/log"
        "github.com/jackc/pgx/v5"
        "github.com/jackc/pgx/v5/pgconn"
        "github.com/jackc/pgx/v5/pgxpool"
        "github.com/jackc/pgx/v5/tracelog"
        retry "github.com/sethvargo/go-retry"
)

const (
        pgConnRecycleSeconds = 1800       // applies for monitored nodes
        applicationName      = "pgwatch3" // will be set on all opened PG connections for informative purposes
)

func Ping(ctx context.Context, connStr string) error {
        c, err := pgx.Connect(ctx, connStr)
        if c != nil {
                _ = c.Close(ctx)
        }
        return err
}

type ConnConfigCallback = func(*pgxpool.Config) error

// New create a new pool
func New(ctx context.Context, connStr string, callbacks ...ConnConfigCallback) (PgxPoolIface, error) {
        connConfig, err := pgxpool.ParseConfig(connStr)
        if err != nil {
                return nil, err
        }
        logger := log.GetLogger(ctx)
        if connConfig.ConnConfig.ConnectTimeout == 0 {
                connConfig.ConnConfig.ConnectTimeout = time.Second * 5
        }
        connConfig.MaxConnIdleTime = 15 * time.Second
        connConfig.MaxConnLifetime = pgConnRecycleSeconds * time.Second
        connConfig.ConnConfig.RuntimeParams["application_name"] = applicationName
        connConfig.ConnConfig.OnNotice = func(_ *pgconn.PgConn, n *pgconn.Notice) {
                logger.WithField("severity", n.Severity).WithField("notice", n.Message).Info("Notice received")
        }
        tracelogger := &tracelog.TraceLog{
                Logger:   log.NewPgxLogger(logger),
                LogLevel: tracelog.LogLevelDebug, //map[bool]tracelog.LogLevel{false: tracelog.LogLevelWarn, true: tracelog.LogLevelDebug}[true],
        }
        connConfig.ConnConfig.Tracer = tracelogger
        for _, f := range callbacks {
                if err = f(connConfig); err != nil {
                        return nil, err
                }
        }
        return pgxpool.NewWithConfig(ctx, connConfig)
}

type ConnInitCallback = func(context.Context, PgxIface) error

// Init creates a new pool, check connection is establised. If not retries connection 3 times with delay 1s
func Init(ctx context.Context, db PgxPoolIface, init ConnInitCallback) error {
        var backoff = retry.WithMaxRetries(3, retry.NewConstant(1*time.Second))
        logger := log.GetLogger(ctx)
        if err := retry.Do(ctx, backoff, func(ctx context.Context) error {
                if err := db.Ping(ctx); err != nil {
                        logger.WithError(err).Error("connection failed")
                        logger.Info("sleeping before reconnecting...")
                        return retry.RetryableError(err)
                }
                return nil
        }); err != nil {
                return err
        }
        return init(ctx, db)
}

// DoesSchemaExist checks if schema exists
func DoesSchemaExist(ctx context.Context, conn PgxIface, schema string) (bool, error) {
        var exists bool
        sqlSchemaExists := "SELECT EXISTS(SELECT 1 FROM pg_namespace WHERE nspname = $1)"
        err := conn.QueryRow(ctx, sqlSchemaExists, schema).Scan(&exists)
        return exists, err
}

// GetTableColumns returns the list of columns for a given table
func GetTableColumns(ctx context.Context, conn PgxIface, table string) (cols []string, err error) {
        sql := `SELECT attname FROM pg_attribute a WHERE a.attrelid = to_regclass($1) and a.attnum > 0 and not a.attisdropped`
        r, err := conn.Query(ctx, sql, table)
        if err != nil {
                return
        }
        return pgx.CollectRows(r, pgx.RowTo[string])
}

1	package db
2
3	import (
4	"context"
5	"time"
6
7	"github.com/cybertec-postgresql/pgwatch3/log"
8	"github.com/jackc/pgx/v5"
9	"github.com/jackc/pgx/v5/pgconn"
10	"github.com/jackc/pgx/v5/pgxpool"
11	"github.com/jackc/pgx/v5/tracelog"
12	retry "github.com/sethvargo/go-retry"
13	)
14
15	const (
16	pgConnRecycleSeconds = 1800 // applies for monitored nodes
17	applicationName = "pgwatch3" // will be set on all opened PG connections for informative purposes
18	)
19
20	func Ping(ctx context.Context, connStr string) error {	×
21	c, err := pgx.Connect(ctx, connStr)	×
22	if c != nil {	×
23	_ = c.Close(ctx)	×
24	}	×
UNCOV 25	return err	×
26	}
27
28	type ConnConfigCallback = func(*pgxpool.Config) error
29
30	// New create a new pool
31	func New(ctx context.Context, connStr string, callbacks ...ConnConfigCallback) (PgxPoolIface, error) {	×
32	connConfig, err := pgxpool.ParseConfig(connStr)	×
33	if err != nil {	×
34	return nil, err	×
35	}	×
36	logger := log.GetLogger(ctx)	×
37	if connConfig.ConnConfig.ConnectTimeout == 0 {	×
38	connConfig.ConnConfig.ConnectTimeout = time.Second * 5	×
39	}	×
40	connConfig.MaxConnIdleTime = 15 * time.Second	×
41	connConfig.MaxConnLifetime = pgConnRecycleSeconds * time.Second	×
42	connConfig.ConnConfig.RuntimeParams["application_name"] = applicationName	×
43	connConfig.ConnConfig.OnNotice = func(_ pgconn.PgConn, n pgconn.Notice) {	×
44	logger.WithField("severity", n.Severity).WithField("notice", n.Message).Info("Notice received")	×
45	}	×
46	tracelogger := &tracelog.TraceLog{	×
47	Logger: log.NewPgxLogger(logger),	×
48	LogLevel: tracelog.LogLevelDebug, //map[bool]tracelog.LogLevel{false: tracelog.LogLevelWarn, true: tracelog.LogLevelDebug}[true],	×
UNCOV 49	}	×
50	connConfig.ConnConfig.Tracer = tracelogger	×
UNCOV 51	for _, f := range callbacks {	×
UNCOV 52	if err = f(connConfig); err != nil {	×
UNCOV 53	return nil, err	×
UNCOV 54	}	×
55	}
UNCOV 56	return pgxpool.NewWithConfig(ctx, connConfig)	×
57	}
58
59	type ConnInitCallback = func(context.Context, PgxIface) error
60
61	// Init creates a new pool, check connection is establised. If not retries connection 3 times with delay 1s
62	func Init(ctx context.Context, db PgxPoolIface, init ConnInitCallback) error {	×
63	var backoff = retry.WithMaxRetries(3, retry.NewConstant(1*time.Second))	×
64	logger := log.GetLogger(ctx)	×
UNCOV 65	if err := retry.Do(ctx, backoff, func(ctx context.Context) error {	×
66	if err := db.Ping(ctx); err != nil {	×
67	logger.WithError(err).Error("connection failed")	×
68	logger.Info("sleeping before reconnecting...")	×
UNCOV 69	return retry.RetryableError(err)	×
UNCOV 70	}	×
UNCOV 71	return nil	×
UNCOV 72	}); err != nil {	×
UNCOV 73	return err	×
UNCOV 74	}	×
UNCOV 75	return init(ctx, db)	×
76	}
77
78	// DoesSchemaExist checks if schema exists
UNCOV 79	func DoesSchemaExist(ctx context.Context, conn PgxIface, schema string) (bool, error) {	×
UNCOV 80	var exists bool	×
UNCOV 81	sqlSchemaExists := "SELECT EXISTS(SELECT 1 FROM pg_namespace WHERE nspname = $1)"	×
UNCOV 82	err := conn.QueryRow(ctx, sqlSchemaExists, schema).Scan(&exists)	×
UNCOV 83	return exists, err	×
UNCOV 84	}	×
85
86	// GetTableColumns returns the list of columns for a given table
87	func GetTableColumns(ctx context.Context, conn PgxIface, table string) (cols []string, err error) {	1✔
88	sql := `SELECT attname FROM pg_attribute a WHERE a.attrelid = to_regclass($1) and a.attnum > 0 and not a.attisdropped`	1✔
89	r, err := conn.Query(ctx, sql, table)	1✔
90	if err != nil {	2✔
91	return	1✔
92	}	1✔
93	return pgx.CollectRows(r, pgx.RowTo[string])	1✔
94	}

cybertec-postgresql / pgwatch3 / 9208786291

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous