10456355969

Committed 19 Aug 2024 03:31PM UTC coverage: 27.846% (-0.4%) from 28.289%

Build # 10456355969

Build Type

Pull #504

github

Committed by

web-flow

Commit Message

Merge d04c1ad49 into 11ec15c1b

Pull Request Pull Request #504: [!] add support for application commands

Run Details

48 of 262 new or added lines in 17 files covered. (18.32%)

10 existing lines in 2 files now uncovered.

1284 of 4611 relevant lines covered (27.85%)

0.31 hits per line

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

0.0

/src/reaper/cache.go

package reaper

import (
        "context"
        "fmt"
        "maps"
        "sync"
        "time"

        "github.com/cybertec-postgresql/pgwatch3/log"
        "github.com/cybertec-postgresql/pgwatch3/metrics"
        "github.com/cybertec-postgresql/pgwatch3/sources"
)

var monitoredDbCache map[string]*sources.MonitoredDatabase
var monitoredDbCacheLock sync.RWMutex
var MonitoredDatabasesSettings = make(map[string]MonitoredDatabaseSettings)
var MonitoredDatabasesSettingsLock = sync.RWMutex{}
var MonitoredDatabasesSettingsGetLock = make(map[string]*sync.RWMutex) // synchronize initial PG version detection to 1 instance for each defined host

var unreachableDBsLock sync.RWMutex
var unreachableDB = make(map[string]time.Time)

var lastDBSizeMB = make(map[string]int64)
var lastDBSizeFetchTime = make(map[string]time.Time) // cached for DB_SIZE_CACHING_INTERVAL
var lastDBSizeCheckLock sync.RWMutex

var prevLoopMonitoredDBs sources.MonitoredDatabases // to be able to detect DBs removed from config
var undersizedDBs = make(map[string]bool)           // DBs below the --min-db-size-mb limit, if set
var undersizedDBsLock = sync.RWMutex{}
var recoveryIgnoredDBs = make(map[string]bool) // DBs in recovery state and OnlyIfMaster specified in config
var recoveryIgnoredDBsLock = sync.RWMutex{}

var hostMetricIntervalMap = make(map[string]float64) // [db1_metric] = 30

var lastSQLFetchError sync.Map

func InitPGVersionInfoFetchingLockIfNil(md *sources.MonitoredDatabase) {
        MonitoredDatabasesSettingsLock.Lock()
        if _, ok := MonitoredDatabasesSettingsGetLock[md.Name]; !ok {
                MonitoredDatabasesSettingsGetLock[md.Name] = &sync.RWMutex{}
        }
        MonitoredDatabasesSettingsLock.Unlock()
}

func UpdateMonitoredDBCache(data sources.MonitoredDatabases) {
        monitoredDbCacheNew := make(map[string]*sources.MonitoredDatabase)
        for _, row := range data {
                monitoredDbCacheNew[row.Name] = row
        }
        monitoredDbCacheLock.Lock()
        monitoredDbCache = monitoredDbCacheNew
        monitoredDbCacheLock.Unlock()
}

func GetMonitoredDatabaseByUniqueName(name string) (*sources.MonitoredDatabase, error) {
        monitoredDbCacheLock.RLock()
        defer monitoredDbCacheLock.RUnlock()
        md, exists := monitoredDbCache[name]
        if !exists || md == nil {
                return nil, fmt.Errorf("Database %s not found in cache", name)
        }
        return md, nil
}

// assumes upwards compatibility for versions
func GetMetricVersionProperties(metric string, _ MonitoredDatabaseSettings, metricDefMap *metrics.Metrics) (metrics.Metric, error) {
        mdm := new(metrics.Metrics)
        if metricDefMap != nil {
                mdm = metricDefMap
        } else {
                metricDefMapLock.RLock()
                mdm.MetricDefs = maps.Clone(metricDefinitionMap.MetricDefs) // copy of global cache
                metricDefMapLock.RUnlock()
        }

        return mdm.MetricDefs[metric], nil
}

// LoadMetricDefs loads metric definitions from the reader
func LoadMetricDefs(r metrics.Reader) (err error) {
        var metricDefs *metrics.Metrics
        if metricDefs, err = r.GetMetrics(); err != nil {
                return
        }
        metricDefMapLock.Lock()
        metricDefinitionMap.MetricDefs = maps.Clone(metricDefs.MetricDefs)
        metricDefinitionMap.PresetDefs = maps.Clone(metricDefs.PresetDefs)
        metricDefMapLock.Unlock()
        return
}

const metricDefinitionRefreshInterval time.Duration = time.Minute * 2 // min time before checking for new/changed metric definitions

// SyncMetricDefs refreshes metric definitions at regular intervals
func SyncMetricDefs(ctx context.Context, r metrics.Reader) {
        for {
                select {
                case <-ctx.Done():
                        return
                case <-time.After(metricDefinitionRefreshInterval):
                        if err := LoadMetricDefs(r); err != nil {
                                log.GetLogger(ctx).Errorf("Could not refresh metric definitions: %w", err)
                        }
                }
        }
}

func GetFromInstanceCacheIfNotOlderThanSeconds(msg MetricFetchConfig, maxAgeSeconds int64) metrics.Measurements {
        var clonedData metrics.Measurements
        instanceMetricCacheTimestampLock.RLock()
        instanceMetricTS, ok := instanceMetricCacheTimestamp[msg.DBUniqueNameOrig+msg.MetricName]
        instanceMetricCacheTimestampLock.RUnlock()
        if !ok || time.Now().Unix()-instanceMetricTS.Unix() > maxAgeSeconds {
                return nil
        }

        instanceMetricCacheLock.RLock()
        instanceMetricData, ok := instanceMetricCache[msg.DBUniqueNameOrig+msg.MetricName]
        if !ok {
                instanceMetricCacheLock.RUnlock()
                return nil
        }
        clonedData = deepCopyMetricData(instanceMetricData)
        instanceMetricCacheLock.RUnlock()

        return clonedData
}

1	package reaper
2
3	import (
4	"context"
5	"fmt"
6	"maps"
7	"sync"
8	"time"
9
10	"github.com/cybertec-postgresql/pgwatch3/log"
11	"github.com/cybertec-postgresql/pgwatch3/metrics"
12	"github.com/cybertec-postgresql/pgwatch3/sources"
13	)
14
15	var monitoredDbCache map[string]*sources.MonitoredDatabase
16	var monitoredDbCacheLock sync.RWMutex
17	var MonitoredDatabasesSettings = make(map[string]MonitoredDatabaseSettings)
18	var MonitoredDatabasesSettingsLock = sync.RWMutex{}
19	var MonitoredDatabasesSettingsGetLock = make(map[string]*sync.RWMutex) // synchronize initial PG version detection to 1 instance for each defined host
20
21	var unreachableDBsLock sync.RWMutex
22	var unreachableDB = make(map[string]time.Time)
23
24	var lastDBSizeMB = make(map[string]int64)
25	var lastDBSizeFetchTime = make(map[string]time.Time) // cached for DB_SIZE_CACHING_INTERVAL
26	var lastDBSizeCheckLock sync.RWMutex
27
28	var prevLoopMonitoredDBs sources.MonitoredDatabases // to be able to detect DBs removed from config
29	var undersizedDBs = make(map[string]bool) // DBs below the --min-db-size-mb limit, if set
30	var undersizedDBsLock = sync.RWMutex{}
31	var recoveryIgnoredDBs = make(map[string]bool) // DBs in recovery state and OnlyIfMaster specified in config
32	var recoveryIgnoredDBsLock = sync.RWMutex{}
33
34	var hostMetricIntervalMap = make(map[string]float64) // [db1_metric] = 30
35
36	var lastSQLFetchError sync.Map
37
38	func InitPGVersionInfoFetchingLockIfNil(md *sources.MonitoredDatabase) {	×
39	MonitoredDatabasesSettingsLock.Lock()	×
NEW 40	if _, ok := MonitoredDatabasesSettingsGetLock[md.Name]; !ok {	×
NEW 41	MonitoredDatabasesSettingsGetLock[md.Name] = &sync.RWMutex{}	×
42	}	×
43	MonitoredDatabasesSettingsLock.Unlock()	×
44	}
45
46	func UpdateMonitoredDBCache(data sources.MonitoredDatabases) {	×
47	monitoredDbCacheNew := make(map[string]*sources.MonitoredDatabase)	×
48	for _, row := range data {	×
NEW 49	monitoredDbCacheNew[row.Name] = row	×
50	}	×
51	monitoredDbCacheLock.Lock()	×
52	monitoredDbCache = monitoredDbCacheNew	×
53	monitoredDbCacheLock.Unlock()	×
54	}
55
56	func GetMonitoredDatabaseByUniqueName(name string) (*sources.MonitoredDatabase, error) {	×
57	monitoredDbCacheLock.RLock()	×
58	defer monitoredDbCacheLock.RUnlock()	×
59	md, exists := monitoredDbCache[name]	×
60	if !exists \|\| md == nil {	×
61	return nil, fmt.Errorf("Database %s not found in cache", name)	×
62	}	×
63	return md, nil	×
64	}
65
66	// assumes upwards compatibility for versions
67	func GetMetricVersionProperties(metric string, _ MonitoredDatabaseSettings, metricDefMap *metrics.Metrics) (metrics.Metric, error) {	×
68	mdm := new(metrics.Metrics)	×
69	if metricDefMap != nil {	×
70	mdm = metricDefMap	×
71	} else {	×
72	metricDefMapLock.RLock()	×
73	mdm.MetricDefs = maps.Clone(metricDefinitionMap.MetricDefs) // copy of global cache	×
74	metricDefMapLock.RUnlock()	×
75	}	×
76
77	return mdm.MetricDefs[metric], nil	×
78	}
79
80	// LoadMetricDefs loads metric definitions from the reader
81	func LoadMetricDefs(r metrics.Reader) (err error) {	×
82	var metricDefs *metrics.Metrics	×
83	if metricDefs, err = r.GetMetrics(); err != nil {	×
84	return	×
85	}	×
86	metricDefMapLock.Lock()	×
87	metricDefinitionMap.MetricDefs = maps.Clone(metricDefs.MetricDefs)	×
88	metricDefinitionMap.PresetDefs = maps.Clone(metricDefs.PresetDefs)	×
89	metricDefMapLock.Unlock()	×
90	return	×
91	}
92
93	const metricDefinitionRefreshInterval time.Duration = time.Minute * 2 // min time before checking for new/changed metric definitions
94
95	// SyncMetricDefs refreshes metric definitions at regular intervals
96	func SyncMetricDefs(ctx context.Context, r metrics.Reader) {	×
97	for {	×
98	select {	×
99	case <-ctx.Done():	×
100	return	×
101	case <-time.After(metricDefinitionRefreshInterval):	×
102	if err := LoadMetricDefs(r); err != nil {	×
103	log.GetLogger(ctx).Errorf("Could not refresh metric definitions: %w", err)	×
104	}	×
105	}
106	}
107	}
108
109	func GetFromInstanceCacheIfNotOlderThanSeconds(msg MetricFetchConfig, maxAgeSeconds int64) metrics.Measurements {	×
110	var clonedData metrics.Measurements	×
111	instanceMetricCacheTimestampLock.RLock()	×
112	instanceMetricTS, ok := instanceMetricCacheTimestamp[msg.DBUniqueNameOrig+msg.MetricName]	×
113	instanceMetricCacheTimestampLock.RUnlock()	×
114	if !ok \|\| time.Now().Unix()-instanceMetricTS.Unix() > maxAgeSeconds {	×
115	return nil	×
116	}	×
117
118	instanceMetricCacheLock.RLock()	×
119	instanceMetricData, ok := instanceMetricCache[msg.DBUniqueNameOrig+msg.MetricName]	×
120	if !ok {	×
121	instanceMetricCacheLock.RUnlock()	×
122	return nil	×
123	}	×
124	clonedData = deepCopyMetricData(instanceMetricData)	×
125	instanceMetricCacheLock.RUnlock()	×
126		×
127	return clonedData	×
128	}

cybertec-postgresql / pgwatch3 / 10456355969

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