WIP

metricbeat/include/list.go

@@ -21,6 +21,11 @@ import (
 	_ "github.com/elastic/beats/metricbeat/module/mysql"
 	_ "github.com/elastic/beats/metricbeat/module/mysql/status"
 	_ "github.com/elastic/beats/metricbeat/module/nginx"
+	_ "github.com/elastic/beats/metricbeat/module/nginx/pluscache"
+	_ "github.com/elastic/beats/metricbeat/module/nginx/plustcpupstream"
+	_ "github.com/elastic/beats/metricbeat/module/nginx/plustcpzone"
+	_ "github.com/elastic/beats/metricbeat/module/nginx/plusupstream"
+	_ "github.com/elastic/beats/metricbeat/module/nginx/pluszone"
 	_ "github.com/elastic/beats/metricbeat/module/nginx/stubstatus"
 	_ "github.com/elastic/beats/metricbeat/module/postgresql"
 	_ "github.com/elastic/beats/metricbeat/module/postgresql/activity"

metricbeat/module/nginx/pluscache/_beat/docs.asciidoc

@@ -0,0 +1 @@
+=== Nginx PlusCache MetricSet

metricbeat/module/nginx/pluscache/_beat/fields.yml

@@ -0,0 +1,9 @@
+- name: pluscache
+  type: group
+  description: >
+    `pluscache` reads server cache status from Nginx ngx_http_status_module.
+  fields:
+    - name: hostname
+      type: keyword
+      description: >
+        Nginx hostname

metricbeat/module/nginx/pluscache/data.go

@@ -0,0 +1,42 @@
+package pluscache
+
+import (
+	"encoding/json"
+	"io"
+	"io/ioutil"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/metricbeat/module/nginx"
+)
+
+// Map body to []MapStr
+func eventMapping(m *MetricSet, body io.ReadCloser, hostname string, metricset string) ([]common.MapStr, error) {
+	// Nginx plus server caches:
+	b, err := ioutil.ReadAll(body)
+	if err != nil {
+		return nil, err
+	}
+
+	var caches map[string]interface{}
+	if err := json.Unmarshal([]byte(b), &caches); err != nil {
+		return nil, err
+	}
+	caches = nginx.Ftoi(caches)
+
+	events := []common.MapStr{}
+
+	for name, cache := range caches {
+		event := common.MapStr{
+			"hostname": hostname,
+			"name": name,
+		}
+
+		for k, v := range cache.(map[string]interface{}) {
+			event[k] = v
+		}
+
+		events = append(events, event)
+	}
+
+	return events, nil
+}

metricbeat/module/nginx/pluscache/pluscache.go

@@ -0,0 +1,117 @@
+// Package pluscache reads server cache status from Nginx, ngx_http_status_module is required.
+package pluscache
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/metricbeat/mb"
+)
+
+const (
+	// defaultScheme is the default scheme to use when it is not specified in
+	// the host config.
+	defaultScheme = "http"
+
+	// defaultPath is the default path to the ngx_http_status_module server cache endpoint on Nginx.
+	defaultPath = "/status/caches"
+)
+
+var (
+	debugf = logp.MakeDebug("nginx-status")
+)
+
+func init() {
+	if err := mb.Registry.AddMetricSet("nginx", "pluscache", New); err != nil {
+		panic(err)
+	}
+}
+
+// MetricSet for fetching Nginx plus status.
+type MetricSet struct {
+	mb.BaseMetricSet
+
+	client *http.Client // HTTP client that is reused across requests.
+	url    string       // Nginx pluscache endpoint URL.
+
+	requests int
+}
+
+// New creates new instance of MetricSet
+func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
+	// Additional configuration options
+	config := struct {
+		ServerStatusPath string `config:"server_status_path"`
+	}{
+		ServerStatusPath: defaultPath,
+	}
+
+	if err := base.Module().UnpackConfig(&config); err != nil {
+		return nil, err
+	}
+
+	u, err := getURL(config.ServerStatusPath, base.Host())
+	if err != nil {
+		return nil, err
+	}
+
+	debugf("nginx-pluscache URL=%s", u)
+	return &MetricSet{
+		BaseMetricSet: base,
+		url:           u.String(),
+		client:        &http.Client{Timeout: base.Module().Config().Timeout},
+		requests:      0,
+	}, nil
+}
+
+// Fetch makes an HTTP request to fetch status metrics from the pluscache endpoint.
+func (m *MetricSet) Fetch() ([]common.MapStr, error) {
+	req, err := http.NewRequest("GET", m.url, nil)
+	resp, err := m.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("error making http request: %v", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		return nil, fmt.Errorf("HTTP error %d: %s", resp.StatusCode, resp.Status)
+	}
+
+	return eventMapping(m, resp.Body, m.Host(), m.Name())
+}
+
+// getURL constructs a URL from the rawHost value and path if one was not set in the rawHost value.
+func getURL(statusPath, rawHost string) (*url.URL, error) {
+	u, err := url.Parse(rawHost)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing nginx host: %v", err)
+	}
+
+	if u.Scheme == "" {
+		// Add scheme and re-parse.
+		u, err = url.Parse(fmt.Sprintf("%s://%s", "http", rawHost))
+		if err != nil {
+			return nil, fmt.Errorf("error parsing nginx host: %v", err)
+		}
+	}
+
+	if u.Host == "" {
+		return nil, fmt.Errorf("error parsing nginx host: empty host")
+	}
+
+	if u.Path == "" {
+		// The path given in the host config takes precedence over the
+		// server_status_path config value.
+		path := statusPath
+		if !strings.HasPrefix(path, "/") {
+			path = "/" + path
+		}
+		u.Path = path
+	}
+
+	return u, nil
+}

metricbeat/module/nginx/pluscache/pluscache_integration_test.go

@@ -0,0 +1,42 @@
+// +build integration
+
+package pluscache
+
+import (
+	"testing"
+
+	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
+	"github.com/elastic/beats/metricbeat/module/nginx"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFetch(t *testing.T) {
+	f := mbtest.NewEventFetcher(t, getConfig())
+	event, err := f.Fetch()
+	if !assert.NoError(t, err) {
+		t.FailNow()
+	}
+
+	t.Logf("%s/%s event: %+v", f.Module().Name(), f.Name(), event)
+
+	// Check number of fields.
+	assert.Equal(t, 10, len(event))
+}
+
+func TestData(t *testing.T) {
+	f := mbtest.NewEventFetcher(t, getConfig())
+
+	err := mbtest.WriteEvent(f, t)
+	if err != nil {
+		t.Fatal("write", err)
+	}
+}
+
+func getConfig() map[string]interface{} {
+	return map[string]interface{}{
+		"module":     "nginx",
+		"metricsets": []string{"pluscache"},
+		"hosts":      []string{nginx.GetNginxEnvHost()},
+	}
+}

metricbeat/module/nginx/plustcpupstream/_beat/docs.asciidoc

@@ -0,0 +1 @@
+=== Nginx PlusTCPUpstream MetricSet

metricbeat/module/nginx/plustcpupstream/_beat/fields.yml

@@ -0,0 +1,9 @@
+- name: plustcpupstream
+  type: group
+  description: >
+    `plustcpupstream` reads server tcpupstream status from Nginx ngx_http_status_module.
+  fields:
+    - name: hostname
+      type: keyword
+      description: >
+        Nginx hostname

metricbeat/module/nginx/plustcpupstream/data.go

@@ -0,0 +1,42 @@
+package plustcpupstream
+
+import (
+	"encoding/json"
+	"io"
+	"io/ioutil"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/metricbeat/module/nginx"
+)
+
+// Map body to []MapStr
+func eventMapping(m *MetricSet, body io.ReadCloser, hostname string, metricset string) ([]common.MapStr, error) {
+	// Nginx plus server tcpupstreams:
+	b, err := ioutil.ReadAll(body)
+	if err != nil {
+		return nil, err
+	}
+
+	var tcpupstreams map[string]interface{}
+	if err := json.Unmarshal([]byte(b), &tcpupstreams); err != nil {
+		return nil, err
+	}
+	tcpupstreams = nginx.Ftoi(tcpupstreams)
+
+	events := []common.MapStr{}
+
+	for name, tcpupstream := range tcpupstreams {
+		event := common.MapStr{
+			"hostname": hostname,
+			"name": name,
+		}
+
+		for k, v := range tcpupstream.(map[string]interface{}) {
+			event[k] = v
+		}
+
+		events = append(events, event)
+	}
+
+	return events, nil
+}