Skip to content

Commit 2824162

Browse files
msakrejdamsakrejda
authored
Miscellaneous refactors and cleanup (#686)
* Rename reloadRun to reload * Stop using deprecated ioutil package * Drop unnecessary fmt.Sprintf * Drop unnecessary fmt.Errorf calls * Drop obsolete debugging output * Drop unused function * Rename globalCollectionOpts to just opts
1 parent e3a07dd commit 2824162

26 files changed

+242
-260
lines changed

Diff for: ‎input/collector.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -42,11 +42,11 @@ func getCollectorStats() state.CollectorStats {
4242
}
4343
}
4444

45-
func getCollectorPlatform(server *state.Server, globalCollectionOpts state.CollectionOpts, logger *util.Logger) state.CollectorPlatform {
45+
func getCollectorPlatform(server *state.Server, opts state.CollectionOpts, logger *util.Logger) state.CollectorPlatform {
4646
hostInfo, err := host.Info()
4747
if err != nil {
4848
server.SelfTest.MarkCollectionAspectError(state.CollectionAspectTelemetry, "could not get collector host information: %s", err)
49-
if globalCollectionOpts.TestRun {
49+
if opts.TestRun {
5050
logger.PrintVerbose("Could not get collector host information: %s", err)
5151
}
5252
return state.CollectorPlatform{}
@@ -58,7 +58,7 @@ func getCollectorPlatform(server *state.Server, globalCollectionOpts state.Colle
5858
virtSystem = hostInfo.VirtualizationSystem
5959
}
6060
return state.CollectorPlatform{
61-
StartedAt: globalCollectionOpts.StartedAt,
61+
StartedAt: opts.StartedAt,
6262
Architecture: runtime.GOARCH,
6363
Hostname: hostInfo.Hostname,
6464
OperatingSystem: hostInfo.OS,

Diff for: ‎input/full.go

+8-8
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,10 @@ import (
1818
)
1919

2020
// CollectFull - Collects a "full" snapshot of all data we need on a regular interval
21-
func CollectFull(ctx context.Context, server *state.Server, connection *sql.DB, globalCollectionOpts state.CollectionOpts, logger *util.Logger) (ps state.PersistedState, ts state.TransientState, err error) {
21+
func CollectFull(ctx context.Context, server *state.Server, connection *sql.DB, opts state.CollectionOpts, logger *util.Logger) (ps state.PersistedState, ts state.TransientState, err error) {
2222
ps.CollectedAt = time.Now()
2323

24-
c, err := postgres.NewCollection(ctx, logger, server, globalCollectionOpts, connection)
24+
c, err := postgres.NewCollection(ctx, logger, server, opts, connection)
2525
if err != nil {
2626
logger.PrintError("Error setting up collection info: %s", err)
2727
return
@@ -37,7 +37,7 @@ func CollectFull(ctx context.Context, server *state.Server, connection *sql.DB,
3737

3838
bufferCacheReady := make(chan state.BufferCache)
3939
go func() {
40-
postgres.GetBufferCache(ctx, c, server, globalCollectionOpts, bufferCacheReady)
40+
postgres.GetBufferCache(ctx, c, server, opts, bufferCacheReady)
4141
}()
4242

4343
ps.LastStatementStatsAt = time.Now()
@@ -60,7 +60,7 @@ func CollectFull(ctx context.Context, server *state.Server, connection *sql.DB,
6060
// CollectorErrors information.
6161
logger.PrintError("Error collecting pg_stat_statements: %s", err)
6262
var e *pq.Error
63-
if errors.As(err, &e) && e.Code == "55000" && globalCollectionOpts.TestRun { // object_not_in_prerequisite_state
63+
if errors.As(err, &e) && e.Code == "55000" && opts.TestRun { // object_not_in_prerequisite_state
6464
shared_preload_libraries, _ := postgres.GetPostgresSetting(ctx, connection, "shared_preload_libraries")
6565
logger.PrintInfo("HINT - Current shared_preload_libraries setting: '%s'. Your Postgres server may need to be restarted for changes to take effect.", shared_preload_libraries)
6666
server.SelfTest.HintCollectionAspect(state.CollectionAspectPgStatStatements, "Current shared_preload_libraries setting: '%s'. Your Postgres server may need to be restarted for changes to take effect.", shared_preload_libraries)
@@ -99,7 +99,7 @@ func CollectFull(ctx context.Context, server *state.Server, connection *sql.DB,
9999
}
100100
}
101101

102-
if globalCollectionOpts.CollectPostgresSettings {
102+
if opts.CollectPostgresSettings {
103103
ts.Settings, err = postgres.GetSettings(ctx, connection)
104104
if err != nil {
105105
logger.PrintError("Error collecting config settings: %s", err)
@@ -157,15 +157,15 @@ func CollectFull(ctx context.Context, server *state.Server, connection *sql.DB,
157157
ps.Relations = filteredRelations
158158
}
159159

160-
if globalCollectionOpts.CollectSystemInformation {
161-
ps.System = system.GetSystemState(ctx, server, logger, globalCollectionOpts)
160+
if opts.CollectSystemInformation {
161+
ps.System = system.GetSystemState(ctx, server, logger, opts)
162162
}
163163

164164
logs.SyncLogParser(server, ts.Settings)
165165

166166
ps.CollectorStats = getCollectorStats()
167167
ts.CollectorConfig = getCollectorConfig(server.Config)
168-
ts.CollectorPlatform = getCollectorPlatform(server, globalCollectionOpts, logger)
168+
ts.CollectorPlatform = getCollectorPlatform(server, opts, logger)
169169

170170
return
171171
}

Diff for: ‎input/postgres/buffer_cache.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -31,10 +31,10 @@ WHERE reldatabase IS NOT NULL -- filters out unused pages
3131
GROUP BY 1, 2
3232
`
3333

34-
func GetBufferCache(ctx context.Context, c *Collection, server *state.Server, globalCollectionOpts state.CollectionOpts, channel chan state.BufferCache) {
34+
func GetBufferCache(ctx context.Context, c *Collection, server *state.Server, opts state.CollectionOpts, channel chan state.BufferCache) {
3535
start := time.Now()
3636
bufferCache := make(state.BufferCache)
37-
db, err := EstablishConnection(ctx, server, c.Logger, globalCollectionOpts, "")
37+
db, err := EstablishConnection(ctx, server, c.Logger, opts, "")
3838
if err != nil {
3939
c.Logger.PrintError("GetBufferCache: %s", err)
4040
channel <- bufferCache
@@ -52,7 +52,7 @@ func GetBufferCache(ctx context.Context, c *Collection, server *state.Server, gl
5252
sizeGB := 0
5353
db.QueryRowContext(ctx, QueryMarkerSQL+bufferCacheSizeSQL).Scan(&sizeGB)
5454
if sizeGB > server.Config.MaxBufferCacheMonitoringGB {
55-
if globalCollectionOpts.TestRun {
55+
if opts.TestRun {
5656
c.Logger.PrintWarning("GetBufferCache: skipping collection. To enable, set max_buffer_cache_monitoring_gb to a value over %d", sizeGB)
5757
}
5858
channel <- bufferCache

Diff for: ‎input/postgres/databases.go

+1-2
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ package postgres
33
import (
44
"context"
55
"database/sql"
6-
"fmt"
76

87
"github.com/pganalyze/collector/state"
98
)
@@ -31,7 +30,7 @@ FROM pg_catalog.pg_database d
3130
ON d.oid = sd.datid`
3231

3332
func GetDatabases(ctx context.Context, db *sql.DB) ([]state.PostgresDatabase, state.PostgresDatabaseStatsMap, error) {
34-
stmt, err := db.PrepareContext(ctx, QueryMarkerSQL+fmt.Sprintf(databasesSQL))
33+
stmt, err := db.PrepareContext(ctx, QueryMarkerSQL+databasesSQL)
3534
if err != nil {
3635
return nil, nil, err
3736
}

Diff for: ‎input/postgres/establish_connection.go

+11-12
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package postgres
33
import (
44
"context"
55
"database/sql"
6+
"errors"
67
"fmt"
78
"strings"
89

@@ -13,20 +14,20 @@ import (
1314
"github.com/pganalyze/collector/util/awsutil"
1415
)
1516

16-
func EstablishConnection(ctx context.Context, server *state.Server, logger *util.Logger, globalCollectionOpts state.CollectionOpts, databaseName string) (connection *sql.DB, err error) {
17-
connection, err = connectToDb(ctx, server.Config, logger, globalCollectionOpts, databaseName)
17+
func EstablishConnection(ctx context.Context, server *state.Server, logger *util.Logger, opts state.CollectionOpts, databaseName string) (connection *sql.DB, err error) {
18+
connection, err = connectToDb(ctx, server.Config, logger, opts, databaseName)
1819
if err != nil {
1920
if err.Error() == "pq: SSL is not enabled on the server" && (server.Config.DbSslMode == "prefer" || server.Config.DbSslMode == "") {
2021
server.Config.DbSslModePreferFailed = true
21-
connection, err = connectToDb(ctx, server.Config, logger, globalCollectionOpts, databaseName)
22+
connection, err = connectToDb(ctx, server.Config, logger, opts, databaseName)
2223
}
2324
}
2425

2526
if err != nil {
2627
return
2728
}
2829

29-
err = validateConnectionCount(ctx, connection, logger, server.Config.MaxCollectorConnections, globalCollectionOpts)
30+
err = validateConnectionCount(ctx, connection, logger, server.Config.MaxCollectorConnections, opts)
3031
if err != nil {
3132
connection.Close()
3233
return
@@ -41,7 +42,7 @@ func EstablishConnection(ctx context.Context, server *state.Server, logger *util
4142
return
4243
}
4344

44-
func connectToDb(ctx context.Context, config config.ServerConfig, logger *util.Logger, globalCollectionOpts state.CollectionOpts, databaseName string) (*sql.DB, error) {
45+
func connectToDb(ctx context.Context, config config.ServerConfig, logger *util.Logger, opts state.CollectionOpts, databaseName string) (*sql.DB, error) {
4546
var dbPasswordOverride string
4647
var hostOverride string
4748
var sslmodeOverride string
@@ -66,25 +67,23 @@ func connectToDb(ctx context.Context, config config.ServerConfig, logger *util.L
6667
}
6768
} else if config.SystemType == "google_cloudsql" {
6869
if config.GcpProjectID == "" || config.GcpRegion == "" || config.GcpCloudSQLInstanceID == "" {
69-
return nil, fmt.Errorf("To use IAM auth with Google Cloud SQL, you must specify project ID, region, and instance ID")
70+
return nil, errors.New("To use IAM auth with Google Cloud SQL, you must specify project ID, region, and instance ID")
7071
}
7172
hostOverride = strings.Join([]string{config.GcpProjectID, config.GcpRegion, config.GcpCloudSQLInstanceID}, ":")
7273
// When using cloud-sql-go-connector, this needs to be set as disable
7374
// https://github.com/GoogleCloudPlatform/cloud-sql-go-connector/issues/889
7475
sslmodeOverride = "disable"
7576
driverName = "cloudsql-postgres"
7677
} else {
77-
return nil, fmt.Errorf("IAM auth is only supported for Amazon RDS, Aurora, and Google Cloud SQL - turn off IAM auth setting to use password-based authentication")
78+
return nil, errors.New("IAM auth is only supported for Amazon RDS, Aurora, and Google Cloud SQL - turn off IAM auth setting to use password-based authentication")
7879
}
7980
}
8081

8182
connectString, err := config.GetPqOpenString(databaseName, dbPasswordOverride, hostOverride, sslmodeOverride)
8283
if err != nil {
8384
return nil, err
8485
}
85-
connectString += " application_name=" + globalCollectionOpts.CollectorApplicationName
86-
87-
// logger.PrintVerbose("sql.Open(\"postgres\", \"%s\")", connectString)
86+
connectString += " application_name=" + opts.CollectorApplicationName
8887

8988
db, err = sql.Open(driverName, connectString)
9089
if err != nil {
@@ -102,10 +101,10 @@ func connectToDb(ctx context.Context, config config.ServerConfig, logger *util.L
102101
return db, nil
103102
}
104103

105-
func validateConnectionCount(ctx context.Context, connection *sql.DB, logger *util.Logger, maxCollectorConnections int, globalCollectionOpts state.CollectionOpts) error {
104+
func validateConnectionCount(ctx context.Context, connection *sql.DB, logger *util.Logger, maxCollectorConnections int, opts state.CollectionOpts) error {
106105
var connectionCount int
107106

108-
err := connection.QueryRowContext(ctx, QueryMarkerSQL+"SELECT pg_catalog.count(*) FROM pg_catalog.pg_stat_activity WHERE application_name = '"+globalCollectionOpts.CollectorApplicationName+"'").Scan(&connectionCount)
107+
err := connection.QueryRowContext(ctx, QueryMarkerSQL+"SELECT pg_catalog.count(*) FROM pg_catalog.pg_stat_activity WHERE application_name = '"+opts.CollectorApplicationName+"'").Scan(&connectionCount)
109108
if err != nil {
110109
return err
111110
}

Diff for: ‎input/postgres/helpers.go

-13
Original file line numberDiff line numberDiff line change
@@ -11,19 +11,6 @@ import (
1111
"github.com/pganalyze/collector/state"
1212
)
1313

14-
func unpackPostgresInt32Array(input null.String) (result []int32) {
15-
if !input.Valid {
16-
return
17-
}
18-
19-
for _, cstr := range strings.Split(strings.Trim(input.String, "{}"), ",") {
20-
cint, _ := strconv.ParseInt(cstr, 10, 32)
21-
result = append(result, int32(cint))
22-
}
23-
24-
return
25-
}
26-
2714
func unpackPostgresOidArray(input null.String) (result []state.Oid) {
2815
if !input.Valid {
2916
return

Diff for: ‎input/postgres/log_pg_read_file.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -33,22 +33,22 @@ SELECT (SELECT size FROM pg_catalog.pg_ls_logdir() WHERE name = $1),
3333
`
3434

3535
// LogPgReadFile - Gets log files using the pg_read_file function
36-
func LogPgReadFile(ctx context.Context, server *state.Server, globalCollectionOpts state.CollectionOpts, logger *util.Logger) (state.PersistedLogState, []state.LogFile, []state.PostgresQuerySample, error) {
36+
func LogPgReadFile(ctx context.Context, server *state.Server, opts state.CollectionOpts, logger *util.Logger) (state.PersistedLogState, []state.LogFile, []state.PostgresQuerySample, error) {
3737
var err error
3838
var psl state.PersistedLogState = server.LogPrevState
3939
var logFiles []state.LogFile
4040
var samples []state.PostgresQuerySample
4141

4242
linesNewerThan := time.Now().Add(-2 * time.Minute)
4343

44-
db, err := EstablishConnection(ctx, server, logger, globalCollectionOpts, "")
44+
db, err := EstablishConnection(ctx, server, logger, opts, "")
4545
if err != nil {
4646
logger.PrintWarning("Could not connect to fetch logs: %s", err)
4747
return server.LogPrevState, nil, nil, err
4848
}
4949
defer db.Close()
5050

51-
h, err := NewCollection(ctx, logger, server, globalCollectionOpts, db)
51+
h, err := NewCollection(ctx, logger, server, opts, db)
5252
if err != nil {
5353
logger.PrintError("Error setting up collection helper: %s", err)
5454
return server.LogPrevState, nil, nil, err

Diff for: ‎input/system/azure/logs.go

+6-6
Original file line numberDiff line numberDiff line change
@@ -217,9 +217,9 @@ func setupEventHubReceiver(ctx context.Context, wg *sync.WaitGroup, logger *util
217217
return nil
218218
}
219219

220-
func SetupLogSubscriber(ctx context.Context, wg *sync.WaitGroup, globalCollectionOpts state.CollectionOpts, logger *util.Logger, servers []*state.Server, parsedLogStream chan state.ParsedLogStreamItem) error {
220+
func SetupLogSubscriber(ctx context.Context, wg *sync.WaitGroup, opts state.CollectionOpts, logger *util.Logger, servers []*state.Server, parsedLogStream chan state.ParsedLogStreamItem) error {
221221
azureLogStream := make(chan AzurePostgresLogRecord, state.LogStreamBufferLen)
222-
setupLogTransformer(ctx, wg, servers, azureLogStream, parsedLogStream, globalCollectionOpts, logger)
222+
setupLogTransformer(ctx, wg, servers, azureLogStream, parsedLogStream, opts, logger)
223223

224224
// This map is used to avoid duplicate receivers to the same Azure Event Hub
225225
eventHubReceivers := make(map[string]bool)
@@ -232,7 +232,7 @@ func SetupLogSubscriber(ctx context.Context, wg *sync.WaitGroup, globalCollectio
232232
}
233233
err := setupEventHubReceiver(ctx, wg, prefixedLogger, server.Config, azureLogStream)
234234
if err != nil {
235-
if globalCollectionOpts.TestRun {
235+
if opts.TestRun {
236236
return err
237237
}
238238

@@ -306,7 +306,7 @@ func ParseRecordToLogLines(in AzurePostgresLogRecord, parser state.LogParser) ([
306306
return logLines, nil
307307
}
308308

309-
func setupLogTransformer(ctx context.Context, wg *sync.WaitGroup, servers []*state.Server, in <-chan AzurePostgresLogRecord, out chan state.ParsedLogStreamItem, globalCollectionOpts state.CollectionOpts, logger *util.Logger) {
309+
func setupLogTransformer(ctx context.Context, wg *sync.WaitGroup, servers []*state.Server, in <-chan AzurePostgresLogRecord, out chan state.ParsedLogStreamItem, opts state.CollectionOpts, logger *util.Logger) {
310310
wg.Add(1)
311311
go func() {
312312
defer wg.Done()
@@ -332,7 +332,7 @@ func setupLogTransformer(ctx context.Context, wg *sync.WaitGroup, servers []*sta
332332
}
333333
}
334334
if server == nil {
335-
if globalCollectionOpts.TestRun {
335+
if opts.TestRun {
336336
logger.PrintVerbose("Discarding log line because of unknown server (did you set the correct azure_db_server_name?): %s", azureDbServerName)
337337
}
338338
continue
@@ -351,7 +351,7 @@ func setupLogTransformer(ctx context.Context, wg *sync.WaitGroup, servers []*sta
351351
}
352352

353353
// Ignore loglines which are outside our time window (except in test runs)
354-
if !logLines[0].OccurredAt.IsZero() && logLines[0].OccurredAt.Before(linesNewerThan) && !globalCollectionOpts.TestRun {
354+
if !logLines[0].OccurredAt.IsZero() && logLines[0].OccurredAt.Before(linesNewerThan) && !opts.TestRun {
355355
continue
356356
}
357357

Diff for: ‎input/system/google_cloudsql/logs.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ func setupPubSubSubscriber(ctx context.Context, wg *sync.WaitGroup, logger *util
138138
return nil
139139
}
140140

141-
func SetupLogSubscriber(ctx context.Context, wg *sync.WaitGroup, globalCollectionOpts state.CollectionOpts, logger *util.Logger, servers []*state.Server, parsedLogStream chan state.ParsedLogStreamItem) error {
141+
func SetupLogSubscriber(ctx context.Context, wg *sync.WaitGroup, opts state.CollectionOpts, logger *util.Logger, servers []*state.Server, parsedLogStream chan state.ParsedLogStreamItem) error {
142142
gcpLogStream := make(chan LogStreamItem, state.LogStreamBufferLen)
143143
setupLogTransformer(ctx, wg, servers, gcpLogStream, parsedLogStream, logger)
144144

@@ -154,7 +154,7 @@ func SetupLogSubscriber(ctx context.Context, wg *sync.WaitGroup, globalCollectio
154154
}
155155
err := setupPubSubSubscriber(ctx, wg, prefixedLogger, server.Config, gcpLogStream)
156156
if err != nil {
157-
if globalCollectionOpts.TestRun {
157+
if opts.TestRun {
158158
return err
159159
}
160160

Diff for: ‎input/system/heroku/http_handler.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,9 @@ import (
1111
"github.com/pganalyze/collector/util"
1212
)
1313

14-
func SetupHttpHandlerLogs(ctx context.Context, wg *sync.WaitGroup, globalCollectionOpts state.CollectionOpts, logger *util.Logger, servers []*state.Server, parsedLogStream chan state.ParsedLogStreamItem) {
14+
func SetupHttpHandlerLogs(ctx context.Context, wg *sync.WaitGroup, opts state.CollectionOpts, logger *util.Logger, servers []*state.Server, parsedLogStream chan state.ParsedLogStreamItem) {
1515
herokuLogStream := make(chan HttpSyslogMessage, state.LogStreamBufferLen)
16-
setupLogTransformer(ctx, wg, servers, herokuLogStream, parsedLogStream, globalCollectionOpts, logger)
16+
setupLogTransformer(ctx, wg, servers, herokuLogStream, parsedLogStream, opts, logger)
1717

1818
go func() {
1919
http.HandleFunc("/", util.HttpRedirectToApp)

0 commit comments

Comments
 (0)