Add support for pg_stat_statements to MetricBeat Postgresql module

metricbeat/module/postgresql/statement/_meta/data.json

@@ -0,0 +1,54 @@
+{
+    "@timestamp": "2017-10-12T08:05:34.853Z",
+    "beat": {
+        "hostname": "host.example.com",
+        "name": "host.example.com"
+    },
+    "metricset": {
+        "host": "postgresql:5432",
+        "module": "postgresql",
+        "name": "statements",
+        "rtt": 115
+    },
+    "postgresql": {
+        "statements": {
+            "user": {
+                "id": 10
+            },
+            "database": {
+                "oid": "12407"
+            },
+            "query": {
+                "id":  "3240664890",
+                "text": "SELECT pg_sleep(?);",
+                "calls": "2",
+                "rows": "2",
+                "time": {
+                    "total": 120066.497,
+                    "min": 60029.533,
+                    "max": 60036.964,
+                    "mean": 60033.2485,
+                    "stddev": 3.71549999999843
+                },
+                "memory": {
+                    "shared": {
+                        "hit": 0,
+                        "read": 0,
+                        "dirtied": 0,
+                        "written": 0,
+                    },
+                    "local": {
+                        "hit": 0,
+                        "read": 0,
+                        "dirtied": 0,
+                        "written": 0,
+                    },
+                    "temp": {
+                        "read": 0,
+                        "written": 0,
+                    }
+                }
+            }
+        }
+    }
+}

metricbeat/module/postgresql/statement/_meta/docs.asciidoc

@@ -0,0 +1,3 @@
+=== postgresql statement MetricSet
+
+This is the statement metricset of the module postgresql.

metricbeat/module/postgresql/statement/_meta/fields.yml

@@ -0,0 +1,91 @@
+- name: statement
+  type: group
+  description: >
+    One document per query per user per database, showing information related
+    invocation of that query, such as cpu usage and total time. Collected by
+    querying pg_stat_statements.
+  release: beta
+  fields:
+    - name: user.id
+      type: long
+      description: >
+        OID of the user logged into the backend that ran the query.
+    - name: database.oid
+      type: long
+      description: >
+        OID of the database the query was run on.
+    - name: query.id
+      type: long
+      description: >
+        ID of the statement.
+    - name: query.text
+      description: >
+        Query text
+    - name: query.calls
+      type: long
+      description: >
+        Number of times the query has been run.
+    - name: query.rows
+      type: long
+      description: >
+        Total number of rows returned by query.
+    - name: time.total
+      type: long
+      description: >
+        Total number of milliseconds spent running query.
+    - name: time.min
+      type: long
+      description: >
+        Minimum number of milliseconds spent running query.
+    - name: time.max
+      type: long
+      description: >
+        Maximum number of milliseconds spent running query.
+    - name: time.mean
+      type: long
+      description: >
+        Mean number of milliseconds spent running query.
+    - name: time.stddev
+      type: long
+      description: >
+        Population standard deviation of time spent running query, in milliseconds.
+    - name: memory.shared.hit
+      type: long
+      description: >
+        Total number of shared block cache hits by the query.
+    - name: memory.shared.read
+      type: long
+      description: >
+        Total number of shared block cache read by the query.
+    - name: memory.shared.dirtied
+      type: long
+      description: >
+        Total number of shared block cache dirtied by the query.
+    - name: memory.shared.written
+      type: long
+      description: >
+        Total number of shared block cache written by the query.
+    - name: memory.local.hit
+      type: long
+      description: >
+        Total number of local block cache hits by the query.
+    - name: memory.local.read
+      type: long
+      description: >
+        Total number of local block cache read by the query.
+    - name: memory.local.dirtied
+      type: long
+      description: >
+        Total number of local block cache dirtied by the query.
+    - name: memory.local.written
+      type: long
+      description: >
+        Total number of local block cache written by the query.
+    - name: memory.temp.read
+      type: long
+      description: >
+        Total number of temp block cache read by the query.
+    - name: memory.temp.written
+      type: long
+      description: >
+        Total number of temp block cache written by the query.

metricbeat/module/postgresql/statement/data.go

@@ -0,0 +1,47 @@
+package statement
+
+import (
+  s "github.com/elastic/beats/libbeat/common/schema"
+  c "github.com/elastic/beats/libbeat/common/schema/mapstrstr"
+)
+
+// Based on: https://www.postgresql.org/docs/9.2/static/monitoring-stats.html#PG-STAT-ACTIVITY-VIEW
+var schema = s.Schema{
+  "user": s.Object{
+    "id": c.Int("userid"),
+  },
+  "database": s.Object{
+    "oid": c.Int("dbid"),
+  },
+  "query": s.Object{
+    "id": c.Str("queryid"),
+    "text": c.Str("query"),
+    "calls": c.Int("datid"),
+    "rows": c.Int("rows"),
+    "time": s.Object{
+      "total": s.Object{"ms": c.Float("total_time")},
+      "min": s.Object{"ms": c.Float("min_time")},
+      "max": s.Object{"ms": c.Float("max_time")},
+      "mean": s.Object{"ms": c.Float("mean_time")},
+      "stddev": s.Object{"ms": c.Float("stddev_time")},
+    },
+    "memory": s.Object{
+      "shared": s.Object{
+        "hit": c.Int("shared_blks_hit"),
+        "read": c.Int("shared_blks_read"),
+        "dirtied": c.Int("shared_blks_dirtied"),
+        "written": c.Int("shared_blks_written"),
+      },
+      "local": s.Object{
+        "hit": c.Int("local_blks_hit"),
+        "read": c.Int("local_blks_read"),
+        "dirtied": c.Int("local_blks_dirtied"),
+        "written": c.Int("local_blks_written"),
+      },
+      "temp": s.Object{
+        "read": c.Int("temp_blks_read"),
+        "written": c.Int("temp_blks_written"),
+      },
+    },
+  },
+}

metricbeat/module/postgresql/statement/statement.go

@@ -0,0 +1,52 @@
+package statement
+
+import (
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	"github.com/elastic/beats/metricbeat/mb"
+)
+
+// init registers the MetricSet with the central registry as soon as the program
+// starts. The New function will be called later to instantiate an instance of
+// the MetricSet for each host defined in the module's configuration. After the
+// MetricSet has been created then Fetch will begin to be called periodically.
+func init() {
+	mb.Registry.MustAddMetricSet("postgresql", "statement", New)
+}
+
+// MetricSet holds any configuration or state information. It must implement
+// the mb.MetricSet interface. And this is best achieved by embedding
+// mb.BaseMetricSet because it implements all of the required mb.MetricSet
+// interface methods except for Fetch.
+type MetricSet struct {
+	mb.BaseMetricSet
+	counter int
+}
+
+// New creates a new instance of the MetricSet. New is responsible for unpacking
+// any MetricSet specific configuration options if there are any.
+func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
+	cfgwarn.Beta("The statement metricset is beta")
+
+	config := struct{}{}
+	if err := base.Module().UnpackConfig(&config); err != nil {
+		return nil, err
+	}
+
+	return &MetricSet{
+		BaseMetricSet: base,
+		counter:       1,
+	}, nil
+}
+
+// Fetch methods implements the data gathering and data conversion to the right
+// format. It publishes the event which is then forwarded to the output. In case
+// of an error set the Error field of mb.Event or simply call report.Error().
+func (m *MetricSet) Fetch(report mb.ReporterV2) {
+	report.Event(mb.Event{
+		MetricSetFields: common.MapStr{
+			"counter": m.counter,
+		},
+	})
+	m.counter++
+}

metricbeat/module/postgresql/statement/statement_integration_test.go

@@ -0,0 +1,94 @@
+// +build integration
+
+package statement
+
+import (
+  "testing"
+
+  "github.com/elastic/beats/libbeat/common"
+  "github.com/elastic/beats/libbeat/tests/compose"
+  mbtest "github.com/elastic/beats/metricbeat/mb/testing"
+  "github.com/elastic/beats/metricbeat/module/postgresql"
+
+  "github.com/stretchr/testify/assert"
+)
+
+func TestFetch(t *testing.T) {
+  compose.EnsureUp(t, "postgresql")
+
+  f := mbtest.NewEventsFetcher(t, getConfig())
+  events, err := f.Fetch()
+  if !assert.NoError(t, err) {
+    t.FailNow()
+  }
+
+  assert.True(t, len(events) > 0)
+  event := events[0]
+
+  t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
+
+  // Check event fields
+  assert.Contains(t, event, "user")
+  assert.Contains(t, event["user"].(common.MapStr), "id")
+
+  assert.Contains(t, event, "database")
+  db_oid := event["database"].(common.MapStr)["oid"].(int64)
+  assert.True(t, db_oid > 0)
+
+  assert.Contains(t, event, "query")
+  query := event["query"].(common.MapStr)
+  assert.Contains(t, query, "id")
+  assert.Contains(t, query, "text")
+  assert.Contains(t, query, "calls")
+  assert.Contains(t, query, "rows")
+
+  assert.Contains(t, query, "time")
+  time := query["time"].(common.MapStr)
+  assert.Contains(t, time, "total")
+  assert.Contains(t, time, "min")
+  assert.Contains(t, time, "max")
+  assert.Contains(t, time, "mean")
+  assert.Contains(t, time, "stddev")
+
+  assert.Contains(t, query["memory"], "shared")
+  memory := query["memory"].(common.MapStr)
+
+  assert.Contains(t, memory, "shared")
+  shared := memory["shared"].(common.MapStr)
+  assert.Contains(t, shared, "hit")
+  assert.Contains(t, shared, "read")
+  assert.Contains(t, shared, "dirtied")
+  assert.Contains(t, shared, "written")
+
+  assert.Contains(t, memory, "local")
+  local := memory["local"].(common.MapStr)
+  assert.Contains(t, local, "hit")
+  assert.Contains(t, local, "read")
+  assert.Contains(t, local, "dirtied")
+  assert.Contains(t, local, "written")
+
+  assert.Contains(t, memory, "temp")
+  temp := memory["temp"].(common.MapStr)
+  assert.Contains(t, temp, "read")
+  assert.Contains(t, temp, "written")
+}
+
+func TestData(t *testing.T) {
+  compose.EnsureUp(t, "postgresql")
+  f := mbtest.NewEventsFetcher(t, getConfig())
+
+  err := mbtest.WriteEvents(f, t)
+  if err != nil {
+    t.Fatal("write", err)
+  }
+}
+
+func getConfig() map[string]interface{} {
+  return map[string]interface{}{
+    "module":     "postgresql",
+    "metricsets": []string{"statement"},
+    "hosts":      []string{postgresql.GetEnvDSN()},
+    "username":   postgresql.GetEnvUsername(),
+    "password":   postgresql.GetEnvPassword(),
+  }
+}