[Metricbeat] Add Zookeeper connections metricset

metricbeat/docs/fields.asciidoc

@@ -26258,6 +26258,53 @@ ZooKeeper metrics collected by the four-letter monitoring commands.
 
 
 
+[float]
+== connection fields
+
+connections
+
+
+
+*`zookeeper.connection.interest_ops`*::
++
+--
+type: long
+
+Interest ops
+
+
+--
+
+*`zookeeper.connection.queued`*::
++
+--
+type: long
+
+Queued connections
+
+
+--
+
+*`zookeeper.connection.received`*::
++
+--
+type: long
+
+Received connections
+
+
+--
+
+*`zookeeper.connection.sent`*::
++
+--
+type: long
+
+Connections sent
+
+
+--
+
 [float]
 == mntr fields
 

metricbeat/docs/modules/zookeeper.asciidoc

@@ -43,10 +43,14 @@ metricbeat.modules:
 
 The following metricsets are available:
 
+* <<metricbeat-metricset-zookeeper-connection,connection>>
+
 * <<metricbeat-metricset-zookeeper-mntr,mntr>>
 
 * <<metricbeat-metricset-zookeeper-server,server>>
 
+include::zookeeper/connection.asciidoc[]
+
 include::zookeeper/mntr.asciidoc[]
 
 include::zookeeper/server.asciidoc[]

metricbeat/docs/modules/zookeeper/connection.asciidoc

@@ -0,0 +1,23 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[metricbeat-metricset-zookeeper-connection]]
+=== ZooKeeper connection metricset
+
+beta[]
+
+include::../../../module/zookeeper/connection/_meta/docs.asciidoc[]
+
+
+==== Fields
+
+For a description of each field in the metricset, see the
+<<exported-fields-zookeeper,exported fields>> section.
+
+Here is an example document generated by this metricset:
+
+[source,json]
+----
+include::../../../module/zookeeper/connection/_meta/data.json[]
+----

metricbeat/docs/modules_list.asciidoc

@@ -166,7 +166,8 @@ This file is generated! See scripts/docs_collector.py
 .2+| .2+|  |<<metricbeat-metricset-windows-perfmon,perfmon>> beta[]  
 |<<metricbeat-metricset-windows-service,service>>   
 |<<metricbeat-module-zookeeper,ZooKeeper>>     |image:./images/icon-yes.png[Prebuilt dashboards are available]    |  
-.2+| .2+|  |<<metricbeat-metricset-zookeeper-mntr,mntr>>   
+.3+| .3+|  |<<metricbeat-metricset-zookeeper-connection,connection>> beta[]  
+|<<metricbeat-metricset-zookeeper-mntr,mntr>>   
 |<<metricbeat-metricset-zookeeper-server,server>>   
 |================================
 

metricbeat/module/zookeeper/connection/_meta/data.json

@@ -0,0 +1,27 @@
+{
+    "@timestamp": "2017-10-12T08:05:34.853Z",
+    "client": {
+        "ip": "172.17.0.1",
+        "port": 47728
+    },
+    "event": {
+        "dataset": "zookeeper.connection",
+        "duration": 115000,
+        "module": "zookeeper"
+    },
+    "metricset": {
+        "name": "connection"
+    },
+    "service": {
+        "address": "localhost:2181",
+        "type": "zookeeper"
+    },
+    "zookeeper": {
+        "connection": {
+            "interest_ops": 0,
+            "queued": 0,
+            "received": 1,
+            "sent": 0
+        }
+    }
+}

metricbeat/module/zookeeper/connection/_meta/docs.asciidoc

@@ -0,0 +1 @@
+This is the 'connection' metricset of the module zookeeper.

metricbeat/module/zookeeper/connection/_meta/fields.yml

@@ -0,0 +1,22 @@
+- name: connection
+  type: group
+  release: beta
+  description: >
+    connections
+  fields:
+    - name: interest_ops
+      type: long
+      description: >
+        Interest ops
+    - name: queued
+      type: long
+      description: >
+        Queued connections
+    - name: received
+      type: long
+      description: >
+        Received connections
+    - name: sent
+      type: long
+      description: >
+        Connections sent

metricbeat/module/zookeeper/connection/connection.go

@@ -0,0 +1,80 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package connection
+
+import (
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	"github.com/elastic/beats/metricbeat/mb"
+	"github.com/elastic/beats/metricbeat/mb/parse"
+	"github.com/elastic/beats/metricbeat/module/zookeeper"
+)
+
+// 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("zookeeper", "connection", New,
+		mb.WithHostParser(parse.PassThruHostParser),
+	)
+}
+
+// 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
+}
+
+// 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 zookeeper connection metricset is beta.")
+
+	config := struct{}{}
+	if err := base.Module().UnpackConfig(&config); err != nil {
+		return nil, err
+	}
+
+	return &MetricSet{
+		BaseMetricSet: base,
+	}, nil
+}
+
+// Fetch fetches metrics from ZooKeeper by making a tcp connection to the
+// command port and sending the "cons" command and parsing the output.
+func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
+	outputReader, err := zookeeper.RunCommand("cons", m.Host(), m.Module().Config().Timeout)
+	if err != nil {
+		return errors.Wrap(err, "'cons' command failed")
+	}
+
+	events, err := m.parseCons(outputReader)
+	if err != nil {
+		return errors.Wrap(err, "error parsing response from zookeeper")
+	}
+
+	for _, event := range events {
+		reporter.Event(event)
+	}
+
+	return nil
+}

metricbeat/module/zookeeper/connection/connection_integration_test.go

@@ -0,0 +1,43 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build integration
+
+package connection
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/metricbeat/module/zookeeper"
+
+	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
+)
+
+func TestData(t *testing.T) {
+	f := mbtest.NewReportingMetricSetV2Error(t, getConfig())
+	if err := mbtest.WriteEventsReporterV2Error(f, t, ""); err != nil {
+		t.Fatal("write", err)
+	}
+}
+
+func getConfig() map[string]interface{} {
+	return map[string]interface{}{
+		"module":     "zookeeper",
+		"metricsets": []string{"connection"},
+		"hosts":      []string{zookeeper.GetZookeeperEnvHost() + ":" + zookeeper.GetZookeeperEnvPort()},
+	}
+}

metricbeat/module/zookeeper/connection/connection_test.go

@@ -0,0 +1,82 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package connection
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/elastic/beats/libbeat/common"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var srvrTestInput = `/172.17.0.1:55218[0](queued=0,recved=1,sent=0)
+/172.17.0.2:55218[55](queued=11,recved=22,sent=333)
+/2001:0db8:85a3:0000:0000:8a2e:0370:7334:55218[0](queued=11,recved=22,sent=333)
+`
+
+func TestParser(t *testing.T) {
+	conns := MetricSet{}
+
+	mapStr, err := conns.parseCons(bytes.NewReader([]byte(srvrTestInput)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.True(t, len(mapStr) == 3)
+	firstLine := mapStr[0]
+	secondLine := mapStr[1]
+	thirdLine := mapStr[2]
+
+	firstLineClient, ok := firstLine.RootFields["client"]
+	assert.True(t, ok)
+
+	firstLineClientMap, ok := firstLineClient.(common.MapStr)
+	assert.True(t, ok)
+
+	secondLineClient, ok := secondLine.RootFields["client"]
+	assert.True(t, ok)
+
+	secondLineClientMap, ok := secondLineClient.(common.MapStr)
+	assert.True(t, ok)
+
+	thirdLineClient, ok := thirdLine.RootFields["client"]
+	assert.True(t, ok)
+
+	thirdLineClientMap, ok := thirdLineClient.(common.MapStr)
+	assert.True(t, ok)
+
+	assert.Equal(t, "172.17.0.1", firstLineClientMap["ip"])
+	assert.Equal(t, "172.17.0.2", secondLineClientMap["ip"])
+	assert.Equal(t, "2001:0db8:85a3:0000:0000:8a2e:0370:7334", thirdLineClientMap["ip"])
+
+	assert.Equal(t, int64(55218), firstLineClientMap["port"])
+	assert.Equal(t, int64(55218), secondLineClientMap["port"])
+
+	assert.Equal(t, int64(0), firstLine.MetricSetFields["interest_ops"])
+	assert.Equal(t, int64(55), secondLine.MetricSetFields["interest_ops"])
+
+	assert.Equal(t, int64(0), firstLine.MetricSetFields["queued"])
+	assert.Equal(t, int64(11), secondLine.MetricSetFields["queued"])
+
+	assert.Equal(t, int64(1), firstLine.MetricSetFields["received"])
+	assert.Equal(t, int64(22), secondLine.MetricSetFields["received"])
+
+	assert.Equal(t, int64(0), firstLine.MetricSetFields["sent"])
+	assert.Equal(t, int64(333), secondLine.MetricSetFields["sent"])
+}