6341578582

Committed 28 Sep 2023 04:19PM UTC coverage: 7.887% (+1.9%) from 5.978%

Build # 6341578582

Build Type

push

github

Committed by

web-flow

Commit Message

[+] add bootstrap tests (#277)

* [+] add config and database bootstrap tests

Run Details

4 of 4 new or added lines in 2 files covered. (100.0%)

433 of 5490 relevant lines covered (7.89%)

0.09 hits per line

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

0.0

/src/api.go

package main

import (
        "context"
        "encoding/json"
        "fmt"
        "reflect"
        "strings"
        "sync/atomic"
        "time"

        "github.com/cybertec-postgresql/pgwatch3/db"
        "golang.org/x/exp/slices"
)

type uiapihandler struct{}

var uiapi uiapihandler

func (uiapi uiapihandler) TryConnectToDB(params []byte) (err error) {
        return db.TryDatabaseConnection(context.TODO(), string(params))
}

// AddPreset adds the preset to the list of available presets
func (uiapi uiapihandler) AddPreset(params []byte) error {
        sql := `INSERT INTO pgwatch3.preset_config(pc_name, pc_description, pc_config) VALUES ($1, $2, $3)`
        var m map[string]any
        err := json.Unmarshal(params, &m)
        if err == nil {
                config, _ := json.Marshal(m["pc_config"])
                _, err = configDb.Exec(context.TODO(), sql, m["pc_name"], m["pc_description"], config)
        }
        return err
}

// UpdatePreset updates the stored preset
func (uiapi uiapihandler) UpdatePreset(id string, params []byte) error {
        fields, values, err := paramsToFieldValues("pgwatch3.preset_config", params)
        if err != nil {
                return err
        }
        sql := fmt.Sprintf(`UPDATE pgwatch3.preset_config SET %s WHERE pc_name = $1`,
                strings.Join(fields, ","))
        values = append([]any{id}, values...)
        _, err = configDb.Exec(context.TODO(), sql, values...)
        return err
}

// GetPresets ret        urns the list of available presets
func (uiapi uiapihandler) GetPresets() (res string, err error) {
        sql := `select coalesce(jsonb_agg(to_jsonb(p)), '[]') from pgwatch3.preset_config p`
        err = configDb.QueryRow(context.TODO(), sql).Scan(&res)
        return
}

// DeletePreset removes the preset from the configuration
func (uiapi uiapihandler) DeletePreset(name string) error {
        _, err := configDb.Exec(context.TODO(), "DELETE FROM pgwatch3.preset_config WHERE pc_name = $1", name)
        return err
}

// GetMetrics returns the list of metrics
func (uiapi uiapihandler) GetMetrics() (res string, err error) {
        sql := `select coalesce(jsonb_agg(to_jsonb(m)), '[]') from metric m`
        err = configDb.QueryRow(context.TODO(), sql).Scan(&res)
        return
}

// DeleteMetric removes the metric from the configuration
func (uiapi uiapihandler) DeleteMetric(id int) error {
        _, err := configDb.Exec(context.TODO(), "DELETE FROM pgwatch3.metric WHERE m_id = $1", id)
        return err
}

// AddMetric adds the metric to the list of stored metrics
func (uiapi uiapihandler) AddMetric(params []byte) error {
        sql := `INSERT INTO pgwatch3.metric(
m_name, m_pg_version_from, m_sql, m_comment, m_is_active, m_is_helper, 
m_master_only, m_standby_only, m_column_attrs, m_sql_su)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`
        var m map[string]any
        err := json.Unmarshal(params, &m)
        if err == nil {
                _, err = configDb.Exec(context.TODO(), sql, m["m_name"], m["m_pg_version_from"],
                        m["m_sql"], m["m_comment"], m["m_is_active"],
                        m["m_is_helper"], m["m_master_only"], m["m_standby_only"],
                        m["m_column_attrs"], m["m_sql_su"])
        }
        return err
}

// GetDatabases returns the list of monitored databases
func (uiapi uiapihandler) GetDatabases() (res string, err error) {
        sql := `select coalesce(jsonb_agg(to_jsonb(db)), '[]') from monitored_db db`
        err = configDb.QueryRow(context.TODO(), sql).Scan(&res)
        return
}

// DeleteDatabase removes the database from the list of monitored databases
func (uiapi uiapihandler) DeleteDatabase(database string) error {
        _, err := configDb.Exec(context.TODO(), "DELETE FROM pgwatch3.monitored_db WHERE md_unique_name = $1", database)
        return err
}

// AddDatabase adds the database to the list of monitored databases
func (uiapi uiapihandler) AddDatabase(params []byte) error {
        sql := `INSERT INTO pgwatch3.monitored_db(
md_unique_name, md_preset_config_name, md_config, md_hostname, 
md_port, md_dbname, md_user, md_password, md_is_superuser, md_is_enabled)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`
        var m map[string]any
        err := json.Unmarshal(params, &m)
        if err == nil {
                _, err = configDb.Exec(context.TODO(), sql, m["md_unique_name"], m["md_preset_config_name"],
                        m["md_config"], m["md_hostname"], m["md_port"],
                        m["md_dbname"], m["md_user"], m["md_password"],
                        m["md_is_superuser"], m["md_is_enabled"])
        }
        return err
}

// UpdateMetric updates the stored metric information
func (uiapi uiapihandler) UpdateMetric(id int, params []byte) error {
        fields, values, err := paramsToFieldValues("pgwatch3.metric", params)
        if err != nil {
                return err
        }
        sql := fmt.Sprintf(`UPDATE pgwatch3.metric SET %s WHERE m_id = $1`, strings.Join(fields, ","))
        values = append([]any{id}, values...)
        _, err = configDb.Exec(context.TODO(), sql, values...)
        return err
}

// UpdateDatabase updates the monitored database information
func (uiapi uiapihandler) UpdateDatabase(database string, params []byte) error {
        fields, values, err := paramsToFieldValues("pgwatch3.monitored_db", params)
        if err != nil {
                return err
        }
        sql := fmt.Sprintf(`UPDATE pgwatch3.monitored_db SET %s WHERE md_unique_name = $1`, strings.Join(fields, ","))
        values = append([]any{database}, values...)
        _, err = configDb.Exec(context.TODO(), sql, values...)
        return err
}

// paramsToFieldValues tranforms JSON body into arrays to be used in UPDATE statement
func paramsToFieldValues(table string, params []byte) (fields []string, values []any, err error) {
        var (
                paramsMap map[string]any
                cols      []string
        )
        if err = json.Unmarshal(params, &paramsMap); err != nil {
                return
        }
        if cols, err = db.GetTableColumns(context.TODO(), configDb, table); err != nil {
                return
        }
        i := 2 // start with the second parameter number, first is reserved for WHERE key value
        for k, v := range paramsMap {
                if slices.Index(cols, k) == -1 {
                        continue
                }
                fields = append(fields, fmt.Sprintf("%s = $%d", quoteIdent(k), i))
                if reflect.ValueOf(v).Kind() == reflect.Map { // transform into json key-value
                        v, _ = json.Marshal(v)
                }
                values = append(values, v)
                i++
        }
        return
}

func quoteIdent(s string) string {
        return `"` + strings.Replace(s, `"`, `""`, -1) + `"`
}

// GetStats
func (uiapi uiapihandler) GetStats() string {
        jsonResponseTemplate := `{
                "main": {
                        "version": "%s",
                        "dbSchema": "%s",
                        "commit": "%s",
                        "built": "%s"
                },
                "metrics": {
                        "totalMetricsFetchedCounter": %d,
                        "totalMetricsReusedFromCacheCounter": %d,
                        "metricPointsPerMinuteLast5MinAvg": %v,
                        "metricsDropped": %d,
                        "totalMetricFetchFailuresCounter": %d
                },
                "datastore": {
                        "secondsFromLastSuccessfulDatastoreWrite": %d,
                        "datastoreWriteFailuresCounter": %d,
                        "datastoreSuccessfulWritesCounter": %d,
                        "datastoreAvgSuccessfulWriteTimeMillis": %.1f
                },
                "general": {
                        "totalDatasetsFetchedCounter": %d,
                        "databasesMonitored": %d,
                        "databasesConfigured": %d,
                        "unreachableDBs": %d,
                        "gathererUptimeSeconds": %d
                }
        }`

        secondsFromLastSuccessfulDatastoreWrite := atomic.LoadInt64(&lastSuccessfulDatastoreWriteTimeEpoch)
        totalMetrics := atomic.LoadUint64(&totalMetricsFetchedCounter)
        cacheMetrics := atomic.LoadUint64(&totalMetricsReusedFromCacheCounter)
        totalDatasets := atomic.LoadUint64(&totalDatasetsFetchedCounter)
        metricsDropped := atomic.LoadUint64(&totalMetricsDroppedCounter)
        metricFetchFailuresCounter := atomic.LoadUint64(&totalMetricFetchFailuresCounter)
        datastoreFailures := atomic.LoadUint64(&datastoreWriteFailuresCounter)
        datastoreSuccess := atomic.LoadUint64(&datastoreWriteSuccessCounter)
        datastoreTotalTimeMicros := atomic.LoadUint64(&datastoreTotalWriteTimeMicroseconds) // successful writes only
        datastoreAvgSuccessfulWriteTimeMillis := float64(datastoreTotalTimeMicros) / float64(datastoreSuccess) / 1000.0
        gathererUptimeSeconds := uint64(time.Since(gathererStartTime).Seconds())
        metricPointsPerMinute := atomic.LoadInt64(&metricPointsPerMinuteLast5MinAvg)
        if metricPointsPerMinute == -1 { // calculate avg. on the fly if 1st summarization hasn't happened yet
                metricPointsPerMinute = int64((totalMetrics * 60) / gathererUptimeSeconds)
        }
        monitoredDbs := getMonitoredDatabasesSnapshot()
        databasesConfigured := len(monitoredDbs) // including replicas
        databasesMonitored := 0
        for _, md := range monitoredDbs {
                if shouldDbBeMonitoredBasedOnCurrentState(md) {
                        databasesMonitored++
                }
        }
        unreachableDBsLock.RLock()
        unreachableDBs := len(unreachableDB)
        unreachableDBsLock.RUnlock()
        return fmt.Sprintf(jsonResponseTemplate, version, dbapi, commit, date,
                totalMetrics, cacheMetrics, metricPointsPerMinute, metricsDropped,
                metricFetchFailuresCounter, time.Now().Unix()-secondsFromLastSuccessfulDatastoreWrite,
                datastoreFailures, datastoreSuccess, datastoreAvgSuccessfulWriteTimeMillis,
                totalDatasets, databasesMonitored, databasesConfigured, unreachableDBs,
                gathererUptimeSeconds)
}

1	package main
2
3	import (
4	"context"
5	"encoding/json"
6	"fmt"
7	"reflect"
8	"strings"
9	"sync/atomic"
10	"time"
11
12	"github.com/cybertec-postgresql/pgwatch3/db"
13	"golang.org/x/exp/slices"
14	)
15
16	type uiapihandler struct{}
17
18	var uiapi uiapihandler
19
20	func (uiapi uiapihandler) TryConnectToDB(params []byte) (err error) {	×
21	return db.TryDatabaseConnection(context.TODO(), string(params))	×
22	}	×
23
24	// AddPreset adds the preset to the list of available presets
25	func (uiapi uiapihandler) AddPreset(params []byte) error {	×
26	sql := `INSERT INTO pgwatch3.preset_config(pc_name, pc_description, pc_config) VALUES ($1, $2, $3)`	×
27	var m map[string]any	×
28	err := json.Unmarshal(params, &m)	×
29	if err == nil {	×
30	config, _ := json.Marshal(m["pc_config"])	×
31	_, err = configDb.Exec(context.TODO(), sql, m["pc_name"], m["pc_description"], config)	×
32	}	×
33	return err	×
34	}
35
36	// UpdatePreset updates the stored preset
37	func (uiapi uiapihandler) UpdatePreset(id string, params []byte) error {	×
38	fields, values, err := paramsToFieldValues("pgwatch3.preset_config", params)	×
39	if err != nil {	×
40	return err	×
41	}	×
42	sql := fmt.Sprintf(`UPDATE pgwatch3.preset_config SET %s WHERE pc_name = $1`,	×
43	strings.Join(fields, ","))	×
44	values = append([]any{id}, values...)	×
45	_, err = configDb.Exec(context.TODO(), sql, values...)	×
46	return err	×
47	}
48
49	// GetPresets ret urns the list of available presets
50	func (uiapi uiapihandler) GetPresets() (res string, err error) {	×
51	sql := `select coalesce(jsonb_agg(to_jsonb(p)), '[]') from pgwatch3.preset_config p`	×
52	err = configDb.QueryRow(context.TODO(), sql).Scan(&res)	×
53	return	×
54	}	×
55
56	// DeletePreset removes the preset from the configuration
57	func (uiapi uiapihandler) DeletePreset(name string) error {	×
58	_, err := configDb.Exec(context.TODO(), "DELETE FROM pgwatch3.preset_config WHERE pc_name = $1", name)	×
59	return err	×
60	}	×
61
62	// GetMetrics returns the list of metrics
63	func (uiapi uiapihandler) GetMetrics() (res string, err error) {	×
64	sql := `select coalesce(jsonb_agg(to_jsonb(m)), '[]') from metric m`	×
65	err = configDb.QueryRow(context.TODO(), sql).Scan(&res)	×
66	return	×
67	}	×
68
69	// DeleteMetric removes the metric from the configuration
70	func (uiapi uiapihandler) DeleteMetric(id int) error {	×
71	_, err := configDb.Exec(context.TODO(), "DELETE FROM pgwatch3.metric WHERE m_id = $1", id)	×
72	return err	×
73	}	×
74
75	// AddMetric adds the metric to the list of stored metrics
76	func (uiapi uiapihandler) AddMetric(params []byte) error {	×
77	sql := `INSERT INTO pgwatch3.metric(	×
78	m_name, m_pg_version_from, m_sql, m_comment, m_is_active, m_is_helper,	×
79	m_master_only, m_standby_only, m_column_attrs, m_sql_su)	×
80	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`	×
81	var m map[string]any	×
82	err := json.Unmarshal(params, &m)	×
83	if err == nil {	×
84	_, err = configDb.Exec(context.TODO(), sql, m["m_name"], m["m_pg_version_from"],	×
85	m["m_sql"], m["m_comment"], m["m_is_active"],	×
86	m["m_is_helper"], m["m_master_only"], m["m_standby_only"],	×
87	m["m_column_attrs"], m["m_sql_su"])	×
88	}	×
89	return err	×
90	}
91
92	// GetDatabases returns the list of monitored databases
93	func (uiapi uiapihandler) GetDatabases() (res string, err error) {	×
94	sql := `select coalesce(jsonb_agg(to_jsonb(db)), '[]') from monitored_db db`	×
95	err = configDb.QueryRow(context.TODO(), sql).Scan(&res)	×
96	return	×
97	}	×
98
99	// DeleteDatabase removes the database from the list of monitored databases
100	func (uiapi uiapihandler) DeleteDatabase(database string) error {	×
101	_, err := configDb.Exec(context.TODO(), "DELETE FROM pgwatch3.monitored_db WHERE md_unique_name = $1", database)	×
102	return err	×
103	}	×
104
105	// AddDatabase adds the database to the list of monitored databases
106	func (uiapi uiapihandler) AddDatabase(params []byte) error {	×
107	sql := `INSERT INTO pgwatch3.monitored_db(	×
108	md_unique_name, md_preset_config_name, md_config, md_hostname,	×
109	md_port, md_dbname, md_user, md_password, md_is_superuser, md_is_enabled)	×
110	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`	×
111	var m map[string]any	×
112	err := json.Unmarshal(params, &m)	×
113	if err == nil {	×
114	_, err = configDb.Exec(context.TODO(), sql, m["md_unique_name"], m["md_preset_config_name"],	×
115	m["md_config"], m["md_hostname"], m["md_port"],	×
116	m["md_dbname"], m["md_user"], m["md_password"],	×
117	m["md_is_superuser"], m["md_is_enabled"])	×
118	}	×
119	return err	×
120	}
121
122	// UpdateMetric updates the stored metric information
123	func (uiapi uiapihandler) UpdateMetric(id int, params []byte) error {	×
124	fields, values, err := paramsToFieldValues("pgwatch3.metric", params)	×
125	if err != nil {	×
126	return err	×
127	}	×
128	sql := fmt.Sprintf(`UPDATE pgwatch3.metric SET %s WHERE m_id = $1`, strings.Join(fields, ","))	×
129	values = append([]any{id}, values...)	×
130	_, err = configDb.Exec(context.TODO(), sql, values...)	×
131	return err	×
132	}
133
134	// UpdateDatabase updates the monitored database information
135	func (uiapi uiapihandler) UpdateDatabase(database string, params []byte) error {	×
136	fields, values, err := paramsToFieldValues("pgwatch3.monitored_db", params)	×
137	if err != nil {	×
138	return err	×
139	}	×
140	sql := fmt.Sprintf(`UPDATE pgwatch3.monitored_db SET %s WHERE md_unique_name = $1`, strings.Join(fields, ","))	×
141	values = append([]any{database}, values...)	×
142	_, err = configDb.Exec(context.TODO(), sql, values...)	×
143	return err	×
144	}
145
146	// paramsToFieldValues tranforms JSON body into arrays to be used in UPDATE statement
147	func paramsToFieldValues(table string, params []byte) (fields []string, values []any, err error) {	×
148	var (	×
149	paramsMap map[string]any	×
150	cols []string	×
151	)	×
152	if err = json.Unmarshal(params, &paramsMap); err != nil {	×
153	return	×
154	}	×
155	if cols, err = db.GetTableColumns(context.TODO(), configDb, table); err != nil {	×
156	return	×
157	}	×
158	i := 2 // start with the second parameter number, first is reserved for WHERE key value	×
159	for k, v := range paramsMap {	×
160	if slices.Index(cols, k) == -1 {	×
161	continue	×
162	}
163	fields = append(fields, fmt.Sprintf("%s = $%d", quoteIdent(k), i))	×
164	if reflect.ValueOf(v).Kind() == reflect.Map { // transform into json key-value	×
165	v, _ = json.Marshal(v)	×
166	}	×
167	values = append(values, v)	×
168	i++	×
169	}
170	return	×
171	}
172
173	func quoteIdent(s string) string {	×
174	return `"` + strings.Replace(s, `"`, `""`, -1) + `"`	×
175	}	×
176
177	// GetStats
178	func (uiapi uiapihandler) GetStats() string {	×
179	jsonResponseTemplate := `{	×
180	"main": {	×
181	"version": "%s",	×
182	"dbSchema": "%s",	×
183	"commit": "%s",	×
184	"built": "%s"	×
185	},	×
186	"metrics": {	×
187	"totalMetricsFetchedCounter": %d,	×
188	"totalMetricsReusedFromCacheCounter": %d,	×
189	"metricPointsPerMinuteLast5MinAvg": %v,	×
190	"metricsDropped": %d,	×
191	"totalMetricFetchFailuresCounter": %d	×
192	},	×
193	"datastore": {	×
194	"secondsFromLastSuccessfulDatastoreWrite": %d,	×
195	"datastoreWriteFailuresCounter": %d,	×
196	"datastoreSuccessfulWritesCounter": %d,	×
197	"datastoreAvgSuccessfulWriteTimeMillis": %.1f	×
198	},	×
199	"general": {	×
200	"totalDatasetsFetchedCounter": %d,	×
201	"databasesMonitored": %d,	×
202	"databasesConfigured": %d,	×
203	"unreachableDBs": %d,	×
204	"gathererUptimeSeconds": %d	×
205	}	×
206	}`	×
207		×
208	secondsFromLastSuccessfulDatastoreWrite := atomic.LoadInt64(&lastSuccessfulDatastoreWriteTimeEpoch)	×
209	totalMetrics := atomic.LoadUint64(&totalMetricsFetchedCounter)	×
210	cacheMetrics := atomic.LoadUint64(&totalMetricsReusedFromCacheCounter)	×
211	totalDatasets := atomic.LoadUint64(&totalDatasetsFetchedCounter)	×
212	metricsDropped := atomic.LoadUint64(&totalMetricsDroppedCounter)	×
213	metricFetchFailuresCounter := atomic.LoadUint64(&totalMetricFetchFailuresCounter)	×
214	datastoreFailures := atomic.LoadUint64(&datastoreWriteFailuresCounter)	×
215	datastoreSuccess := atomic.LoadUint64(&datastoreWriteSuccessCounter)	×
216	datastoreTotalTimeMicros := atomic.LoadUint64(&datastoreTotalWriteTimeMicroseconds) // successful writes only	×
217	datastoreAvgSuccessfulWriteTimeMillis := float64(datastoreTotalTimeMicros) / float64(datastoreSuccess) / 1000.0	×
218	gathererUptimeSeconds := uint64(time.Since(gathererStartTime).Seconds())	×
219	metricPointsPerMinute := atomic.LoadInt64(&metricPointsPerMinuteLast5MinAvg)	×
220	if metricPointsPerMinute == -1 { // calculate avg. on the fly if 1st summarization hasn't happened yet	×
221	metricPointsPerMinute = int64((totalMetrics * 60) / gathererUptimeSeconds)	×
222	}	×
223	monitoredDbs := getMonitoredDatabasesSnapshot()	×
224	databasesConfigured := len(monitoredDbs) // including replicas	×
225	databasesMonitored := 0	×
226	for _, md := range monitoredDbs {	×
227	if shouldDbBeMonitoredBasedOnCurrentState(md) {	×
228	databasesMonitored++	×
229	}	×
230	}
231	unreachableDBsLock.RLock()	×
232	unreachableDBs := len(unreachableDB)	×
233	unreachableDBsLock.RUnlock()	×
234	return fmt.Sprintf(jsonResponseTemplate, version, dbapi, commit, date,	×
235	totalMetrics, cacheMetrics, metricPointsPerMinute, metricsDropped,	×
236	metricFetchFailuresCounter, time.Now().Unix()-secondsFromLastSuccessfulDatastoreWrite,	×
237	datastoreFailures, datastoreSuccess, datastoreAvgSuccessfulWriteTimeMillis,	×
238	totalDatasets, databasesMonitored, databasesConfigured, unreachableDBs,	×
239	gathererUptimeSeconds)	×
240	}

cybertec-postgresql / pgwatch3 / 6341578582

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