Preliminary PostgreSQL 9.2, 9.3 and 9.6 support.

.gitignore

@@ -1,3 +1,4 @@
 collector
 pganalyze-collector
 packages/tmp
+test/*.json

input/postgres/backends.go

@@ -2,21 +2,36 @@ package postgres
 
 import (
 	"database/sql"
+	"fmt"
 
 	"github.com/guregu/null"
 	"github.com/lfittl/pg_query_go"
 	"github.com/pganalyze/collector/state"
 	"github.com/pganalyze/collector/util"
 )
 
+const activitySQLDefaultOptionalFields = "waiting, NULL, NULL"
+const activitySQLpg94OptionalFields = "waiting, backend_xid, backend_xmin"
+const activitySQLpg96OptionalFields = "wait_event IS NOT NULL, backend_xid, backend_xmin"
+
 // https://www.postgresql.org/docs/9.5/static/monitoring-stats.html#PG-STAT-ACTIVITY-VIEW
 const activitySQL string = `SELECT datid, usesysid, pid, application_name, client_addr::text, client_port,
-				backend_start, xact_start, query_start, state_change, waiting, backend_xid, backend_xmin, state, query
+				backend_start, xact_start, query_start, state_change, %s, state, query
 	 FROM pg_stat_activity
 	WHERE pid <> pg_backend_pid() AND datname = current_database()`
 
 func GetBackends(logger *util.Logger, db *sql.DB, postgresVersion state.PostgresVersion) ([]state.PostgresBackend, error) {
-	stmt, err := db.Prepare(QueryMarkerSQL + activitySQL)
+	var optionalFields string
+
+	if postgresVersion.Numeric >= state.PostgresVersion96 {
+		optionalFields = activitySQLpg96OptionalFields
+	} else if postgresVersion.Numeric >= state.PostgresVersion94 {
+		optionalFields = activitySQLpg94OptionalFields
+	} else {
+		optionalFields = activitySQLDefaultOptionalFields
+	}
+
+	stmt, err := db.Prepare(QueryMarkerSQL + fmt.Sprintf(activitySQL, optionalFields))
 	if err != nil {
 		return nil, err
 	}

input/postgres/relations.go

@@ -10,6 +10,9 @@ import (
 	"github.com/pganalyze/collector/state"
 )
 
+const relationsSQLDefaultOptionalFields = "0"
+const relationsSQLpg93OptionalFields = "c.relminmxid"
+
 const relationsSQL string = `
 	 WITH locked_relids AS (SELECT DISTINCT relation relid FROM pg_locks WHERE mode = 'AccessExclusiveLock')
  SELECT c.oid,
@@ -22,7 +25,7 @@ const relationsSQL string = `
 				c.relhassubclass AS relation_has_inheritance_children,
 				c.reltoastrelid IS NULL AS relation_has_toast,
 				c.relfrozenxid AS relation_frozen_xid,
-				c.relminmxid AS relation_min_mxid,
+				%s,
 				locked_relids.relid IS NOT NULL
 	 FROM pg_catalog.pg_class c
 	 LEFT JOIN pg_catalog.pg_namespace n ON (n.oid = c.relnamespace)
@@ -116,7 +119,15 @@ func GetRelations(db *sql.DB, postgresVersion state.PostgresVersion, currentData
 	relations := make(map[state.Oid]state.PostgresRelation, 0)
 
 	// Relations
-	rows, err := db.Query(QueryMarkerSQL + relationsSQL)
+	var optionalFields string
+
+	if postgresVersion.Numeric >= state.PostgresVersion93 {
+		optionalFields = relationsSQLpg93OptionalFields
+	} else {
+		optionalFields = relationsSQLDefaultOptionalFields
+	}
+
+	rows, err := db.Query(QueryMarkerSQL + fmt.Sprintf(relationsSQL, optionalFields))
 	if err != nil {
 		err = fmt.Errorf("Relations/Query: %s", err)
 		return nil, err

input/postgres/statements.go

@@ -3,8 +3,10 @@ package postgres
 import (
 	"database/sql"
 	"fmt"
+	"hash/fnv"
 	"strings"
 
+	"github.com/guregu/null"
 	"github.com/lib/pq"
 	"github.com/pganalyze/collector/state"
 	"github.com/pganalyze/collector/util"
@@ -99,18 +101,28 @@ func GetStatements(logger *util.Logger, db *sql.DB, postgresVersion state.Postgr
 
 	for rows.Next() {
 		var key state.PostgresStatementKey
+		var queryID null.Int
 		var statement state.PostgresStatement
 		var stats state.PostgresStatementStats
 
 		err = rows.Scan(&key.DatabaseOid, &key.UserOid, &statement.NormalizedQuery, &stats.Calls, &stats.TotalTime, &stats.Rows,
 			&stats.SharedBlksHit, &stats.SharedBlksRead, &stats.SharedBlksDirtied, &stats.SharedBlksWritten,
 			&stats.LocalBlksHit, &stats.LocalBlksRead, &stats.LocalBlksDirtied, &stats.LocalBlksWritten,
 			&stats.TempBlksRead, &stats.TempBlksWritten, &stats.BlkReadTime, &stats.BlkWriteTime,
-			&key.QueryID, &stats.MinTime, &stats.MaxTime, &stats.MeanTime, &stats.StddevTime)
+			&queryID, &stats.MinTime, &stats.MaxTime, &stats.MeanTime, &stats.StddevTime)
 		if err != nil {
 			return nil, nil, err
 		}
 
+		if queryID.Valid {
+			key.QueryID = queryID.Int64
+		} else {
+			// Note: This is a heuristic for old Postgres versions and will not work for duplicate queries (e.g. when tables are dropped and recreated)
+			h := fnv.New64a()
+			h.Write([]byte(statement.NormalizedQuery))
+			key.QueryID = int64(h.Sum64())
+		}
+
 		statements[key] = statement
 		statementStats[key] = stats
 	}

main.go

@@ -74,6 +74,7 @@ const defaultConfigFile = "/etc/pganalyze-collector.conf"
 func main() {
 	var dryRun bool
 	var testRun bool
+	var forceStateUpdate bool
 	var configFilename string
 	var stateFilename string
 	var pidFilename string
@@ -91,6 +92,7 @@ func main() {
 	flag.BoolVar(&logToSyslog, "syslog", false, "Write all log output to syslog instead of stderr (disabled by default)")
 	flag.BoolVar(&logNoTimestamps, "no-log-timestamps", false, "Disable timestamps in the log output (automatically done when syslog is enabled)")
 	flag.BoolVar(&dryRun, "dry-run", false, "Print JSON data that would get sent to web service (without actually sending) and exit afterwards.")
+	flag.BoolVar(&forceStateUpdate, "force-state-update", false, "Updates the state file even if other options would have prevented it (intended to be used together with --dry-run for debugging)")
 	flag.BoolVar(&noPostgresRelations, "no-postgres-relations", false, "Don't collect any Postgres relation information (not recommended)")
 	flag.BoolVar(&noPostgresSettings, "no-postgres-settings", false, "Don't collect Postgres configuration settings")
 	flag.BoolVar(&noPostgresLocks, "no-postgres-locks", false, "Don't collect Postgres lock information (NOTE: This is always enabled right now, i.e. no lock data is gathered)")
@@ -146,6 +148,7 @@ func main() {
 		CollectSystemInformation: !noSystemInformation,
 		DiffStatements:           diffStatements,
 		StateFilename:            stateFilename,
+		WriteStateUpdate:         !testRun || forceStateUpdate,
 		StatementTimeoutMs:       10000,
 	}
 

output/full.go

@@ -14,6 +14,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/golang/protobuf/jsonpb"
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/pganalyze/collector/output/pganalyze_collector"
@@ -84,10 +85,15 @@ func debugOutputAsJSON(logger *util.Logger, compressedData bytes.Buffer) {
 	}
 
 	var out bytes.Buffer
-	dataJSON, _ := json.Marshal(s)
-	json.Indent(&out, dataJSON, "", "\t")
+	var marshaler jsonpb.Marshaler
+	dataJSON, err := marshaler.MarshalToString(s)
+	if err != nil {
+		logger.PrintError("Failed to transform protocol buffers to JSON: %s", err)
+		return
+	}
+	json.Indent(&out, []byte(dataJSON), "", "\t")
 	logger.PrintInfo("Dry run - data that would have been sent will be output on stdout:\n")
-	logger.PrintInfo(out.String())
+	fmt.Printf("%s\n", out.String())
 }
 
 type s3UploadResponse struct {

runner/full.go

@@ -15,16 +15,21 @@ import (
 )
 
 func processDatabase(server state.Server, globalCollectionOpts state.CollectionOpts, logger *util.Logger) (state.PersistedState, state.Grant, error) {
-	// Note: In case of server errors, we should reuse the old grant if its still recent (i.e. less than 50 minutes ago)
-	grant, err := getSnapshotGrant(server, globalCollectionOpts, logger)
-	if err != nil {
-		if server.Grant.Valid {
-			logger.PrintVerbose("Could not acquire snapshot grant, reusing previous grant: %s", err)
+	var grant state.Grant
+	var err error
+
+	if globalCollectionOpts.SubmitCollectedData {
+		// Note: In case of server errors, we should reuse the old grant if its still recent (i.e. less than 50 minutes ago)
+		grant, err = getSnapshotGrant(server, globalCollectionOpts, logger)
+		if err != nil {
+			if server.Grant.Valid {
+				logger.PrintVerbose("Could not acquire snapshot grant, reusing previous grant: %s", err)
+			} else {
+				return state.PersistedState{}, state.Grant{}, err
+			}
 		} else {
-			return state.PersistedState{}, state.Grant{}, err
+			server.Grant = grant
 		}
-	} else {
-		server.Grant = grant
 	}
 
 	newState, transientState, err := input.CollectFull(server, globalCollectionOpts, logger)
@@ -151,7 +156,7 @@ func CollectAllServers(servers []state.Server, globalCollectionOpts state.Collec
 		server.Connection = nil
 	}
 
-	if !globalCollectionOpts.TestRun {
+	if globalCollectionOpts.WriteStateUpdate {
 		writeStateFile(servers, globalCollectionOpts, logger)
 	}
 }

state/state.go

@@ -83,7 +83,8 @@ type CollectionOpts struct {
 	SubmitCollectedData bool
 	TestRun             bool
 
-	StateFilename string
+	StateFilename    string
+	WriteStateUpdate bool
 }
 
 type GrantConfig struct {

state/system.go

@@ -281,16 +281,24 @@ func (curr NetworkStats) DiffSince(prev NetworkStats, collectedIntervalSecs uint
 
 // DiffSince - Calculate the diff between two disk stats runs
 func (curr DiskStats) DiffSince(prev DiskStats, collectedIntervalSecs uint32) DiffedDiskStats {
-	return DiffedDiskStats{
-		ReadOperationsPerSecond:  float64(curr.ReadsCompleted-prev.ReadsCompleted) / float64(collectedIntervalSecs),
+	readsPerSecond := float64(curr.ReadsCompleted - prev.ReadsCompleted)
+	writePerSecond := float64(curr.WritesCompleted - prev.WritesCompleted)
+
+	diffed := DiffedDiskStats{
+		ReadOperationsPerSecond:  readsPerSecond / float64(collectedIntervalSecs),
 		ReadsMergedPerSecond:     float64(curr.ReadsMerged-prev.ReadsMerged) / float64(collectedIntervalSecs),
 		BytesReadPerSecond:       float64(curr.BytesRead-prev.BytesRead) / float64(collectedIntervalSecs),
-		AvgReadLatency:           float64(curr.ReadTimeMs-prev.ReadTimeMs) / float64(curr.ReadsCompleted-prev.ReadsCompleted),
-		WriteOperationsPerSecond: float64(curr.WritesCompleted-prev.WritesCompleted) / float64(collectedIntervalSecs),
+		WriteOperationsPerSecond: writePerSecond / float64(collectedIntervalSecs),
 		WritesMergedPerSecond:    float64(curr.WritesMerged-prev.WritesMerged) / float64(collectedIntervalSecs),
 		BytesWrittenPerSecond:    float64(curr.BytesWritten-prev.BytesWritten) / float64(collectedIntervalSecs),
-		AvgWriteLatency:          float64(curr.WriteTimeMs-prev.WriteTimeMs) / float64(curr.WritesCompleted-prev.WritesCompleted),
 		AvgQueueSize:             curr.AvgQueueSize,
 		UtilizationPercent:       float64(curr.IoTime-prev.IoTime) / float64(collectedIntervalSecs) * 100.0,
 	}
+
+	if readsPerSecond > 0 {
+		diffed.AvgReadLatency = float64(curr.ReadTimeMs-prev.ReadTimeMs) / readsPerSecond
+		diffed.AvgWriteLatency = float64(curr.WriteTimeMs-prev.WriteTimeMs) / writePerSecond
+	}
+
+	return diffed
 }

test/Dockerfile.test-pg92

@@ -0,0 +1,18 @@
+FROM postgres:9.2
+ADD test/postgres-pgss.sh /docker-entrypoint-initdb.d/
+
+ENV GOPATH /go
+ENV CODE_DIR $GOPATH/src/github.com/pganalyze/collector
+ENV PATH $PATH:/usr/local/go/bin
+
+# Packages required for both building and packaging
+RUN apt-get update -qq && apt-get install -y -q build-essential git curl
+
+# Golang
+RUN curl -o go.tar.gz -sSL "https://storage.googleapis.com/golang/go1.6.2.linux-amd64.tar.gz"
+RUN tar -C /usr/local -xzf go.tar.gz
+
+# Build the collector
+COPY . $CODE_DIR
+WORKDIR $CODE_DIR
+RUN make prepare && make build

test/Dockerfile.test-pg93

@@ -0,0 +1,18 @@
+FROM postgres:9.3
+ADD test/postgres-pgss.sh /docker-entrypoint-initdb.d/
+
+ENV GOPATH /go
+ENV CODE_DIR $GOPATH/src/github.com/pganalyze/collector
+ENV PATH $PATH:/usr/local/go/bin
+
+# Packages required for both building and packaging
+RUN apt-get update -qq && apt-get install -y -q build-essential git curl
+
+# Golang
+RUN curl -o go.tar.gz -sSL "https://storage.googleapis.com/golang/go1.6.2.linux-amd64.tar.gz"
+RUN tar -C /usr/local -xzf go.tar.gz
+
+# Build the collector
+COPY . $CODE_DIR
+WORKDIR $CODE_DIR
+RUN make prepare && make build

test/Dockerfile.test-pg94

@@ -0,0 +1,18 @@
+FROM postgres:9.4
+ADD test/postgres-pgss.sh /docker-entrypoint-initdb.d/
+
+ENV GOPATH /go
+ENV CODE_DIR $GOPATH/src/github.com/pganalyze/collector
+ENV PATH $PATH:/usr/local/go/bin
+
+# Packages required for both building and packaging
+RUN apt-get update -qq && apt-get install -y -q build-essential git curl
+
+# Golang
+RUN curl -o go.tar.gz -sSL "https://storage.googleapis.com/golang/go1.6.2.linux-amd64.tar.gz"
+RUN tar -C /usr/local -xzf go.tar.gz
+
+# Build the collector
+COPY . $CODE_DIR
+WORKDIR $CODE_DIR
+RUN make prepare && make build

test/Dockerfile.test-pg95

@@ -0,0 +1,18 @@
+FROM postgres:9.5
+ADD test/postgres-pgss.sh /docker-entrypoint-initdb.d/
+
+ENV GOPATH /go
+ENV CODE_DIR $GOPATH/src/github.com/pganalyze/collector
+ENV PATH $PATH:/usr/local/go/bin
+
+# Packages required for both building and packaging
+RUN apt-get update -qq && apt-get install -y -q build-essential git curl
+
+# Golang
+RUN curl -o go.tar.gz -sSL "https://storage.googleapis.com/golang/go1.6.2.linux-amd64.tar.gz"
+RUN tar -C /usr/local -xzf go.tar.gz
+
+# Build the collector
+COPY . $CODE_DIR
+WORKDIR $CODE_DIR
+RUN make prepare && make build

test/Dockerfile.test-pg96

@@ -0,0 +1,18 @@
+FROM postgres:9.6
+ADD test/postgres-pgss.sh /docker-entrypoint-initdb.d/
+
+ENV GOPATH /go
+ENV CODE_DIR $GOPATH/src/github.com/pganalyze/collector
+ENV PATH $PATH:/usr/local/go/bin
+
+# Packages required for both building and packaging
+RUN apt-get update -qq && apt-get install -y -q build-essential git curl
+
+# Golang
+RUN curl -o go.tar.gz -sSL "https://storage.googleapis.com/golang/go1.6.2.linux-amd64.tar.gz"
+RUN tar -C /usr/local -xzf go.tar.gz
+
+# Build the collector
+COPY . $CODE_DIR
+WORKDIR $CODE_DIR
+RUN make prepare && make build