[metricbeat] Add linux sockstat data to socket_summary metricset

metricbeat/docs/fields.asciidoc

@@ -25959,13 +25959,35 @@ All TCP connections
 
 
 
+*`system.socket.summary.tcp.memory`*::
++
+--
+type: integer
+
+format: bytes
+
+Memory used by TCP sockets in bytes, based on number of allocated pages and system page size. Corresponds to limits set in /proc/sys/net/ipv4/tcp_mem. Only available on Linux. 
+
+
+--
+
 [float]
 == all fields
 
 All TCP connections
 
 
 
+*`system.socket.summary.tcp.all.orphan`*::
++
+--
+type: integer
+
+A count of all orphaned tcp sockets. Only available on Linux.
+
+
+--
+
 *`system.socket.summary.tcp.all.count`*::
 +
 --
@@ -26023,6 +26045,18 @@ All UDP connections
 
 
 
+*`system.socket.summary.udp.memory`*::
++
+--
+type: integer
+
+format: bytes
+
+Memory used by UDP sockets in bytes, based on number of allocated pages and system page size. Corresponds to limits set in /proc/sys/net/ipv4/udp_mem. Only available on Linux. 
+
+
+--
+
 [float]
 == all fields
 

metricbeat/module/system/socket_summary/_meta/fields.yml

@@ -23,11 +23,20 @@
       description: >
         All TCP connections
       fields:
+        - name: memory
+          type: integer
+          format: bytes
+          description: >
+            Memory used by TCP sockets in bytes, based on number of allocated pages and system page size. Corresponds to limits set in /proc/sys/net/ipv4/tcp_mem. Only available on Linux. 
         - name: all
           type: group
           description: >
             All TCP connections
           fields:
+            - name: orphan
+              type: integer
+              description: >
+                A count of all orphaned tcp sockets. Only available on Linux.
             - name: count
               type: integer
               description: >
@@ -53,6 +62,11 @@
       description: >
         All UDP connections
       fields:
+        - name: memory
+          type: integer
+          format: bytes
+          description: >
+            Memory used by UDP sockets in bytes, based on number of allocated pages and system page size. Corresponds to limits set in /proc/sys/net/ipv4/udp_mem. Only available on Linux. 
         - name: all
           type: group
           description: >

metricbeat/module/system/socket_summary/_meta/testdata/sockstat

@@ -0,0 +1,6 @@
+sockets: used 249
+TCP: inuse 2 orphan 0 tw 0 alloc 12 mem 1
+UDP: inuse 2 mem 11
+UDPLITE: inuse 0
+RAW: inuse 0
+FRAG: inuse 0 memory 0

metricbeat/module/system/socket_summary/socket_summary.go

@@ -20,6 +20,7 @@ package socket_summary
 import (
 	"syscall"
 
+	"github.com/pkg/errors"
 	"github.com/shirou/gopsutil/net"
 
 	"github.com/elastic/beats/libbeat/common"
@@ -43,6 +44,7 @@ func init() {
 // interface methods except for Fetch.
 type MetricSet struct {
 	mb.BaseMetricSet
+	sockstat string
 }
 
 // New creates a new instance of the MetricSet. New is responsible for unpacking
@@ -114,17 +116,24 @@ func calculateConnStats(conns []net.ConnectionStat) common.MapStr {
 // 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) {
-
+func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	// all network connections
 	conns, err := net.Connections("inet")
 
 	if err != nil {
-		report.Error(err)
-		return
+		return errors.Wrap(err, "error getting connections")
+	}
+
+	stats := calculateConnStats(conns)
+	newStats, err := applyEnhancements(stats, m)
+	if err != nil {
+		m.Logger().Debugf("error applying enhancements: %s", err)
+		newStats = stats
 	}
 
 	report.Event(mb.Event{
-		MetricSetFields: calculateConnStats(conns),
+		MetricSetFields: newStats,
 	})
+
+	return nil
 }

metricbeat/module/system/socket_summary/sockstat_linux.go

@@ -0,0 +1,134 @@
+// 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 linux
+
+package socket_summary
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/metricbeat/module/system"
+)
+
+// SockStat contains data from /proc/net/sockstat
+type SockStat struct {
+	//The count of all sockets in use on the system, in the most liberal definition. See `ss -s` and `ss -a` for more.
+	SocketsUsed int
+	//Total in use TCP sockets
+	TCPInUse int
+	//Total 'orphaned' TCP sockets
+	TCPOrphan int
+	//Sockets in TIME_WAIT
+	TCPTW int
+	//Total allocated sockets
+	TCPAlloc int
+	//Socket memory use, in pages
+	TCPMem int
+	//In use UDP sockets
+	UDPInUse int
+	//Socket memory use, in pages
+	UDPMem int
+	//UDP-Lite in use sockets
+	UDPLiteInUse int
+	//In Use raw sockets
+	RawInUse int
+	//FRAG sockets in use
+	FragInUse int
+	//Frag memory, in bytes
+	FragMemory int
+}
+
+// applyEnhancements gets a list of platform-specific enhancements and apply them to our mapStr object.
+func applyEnhancements(data common.MapStr, m *MetricSet) (common.MapStr, error) {
+	systemModule, ok := m.BaseMetricSet.Module().(*system.Module)
+	if !ok {
+		return nil, errors.New("unexpected module type")
+	}
+	dir := filepath.Join(systemModule.HostFS, "/proc/net/sockstat")
+	pageSize := os.Getpagesize()
+
+	stat, err := parseSockstat(dir)
+	if err != nil {
+		return nil, errors.Wrap(err, "error getting sockstat data")
+	}
+	data.Put("tcp.all.orphan", stat.TCPOrphan)
+	data.Put("tcp.memory", pageSize*stat.TCPMem)
+	data.Put("udp.memory", pageSize*stat.UDPMem)
+
+	return data, nil
+
+}
+
+// parseSockstat parses the ipv4 sockstat file
+//see net/ipv4/proc.c
+func parseSockstat(path string) (SockStat, error) {
+	fd, err := os.Open(path)
+	if err != nil {
+		return SockStat{}, err
+	}
+
+	var ss SockStat
+	scanfLines := []string{
+		"sockets: used %d",
+		"TCP: inuse %d orphan %d tw %d alloc %d mem %d",
+		"UDP: inuse %d mem %d",
+		"UDPLITE: inuse %d",
+		"RAW: inuse %d",
+		"FRAG: inuse %d memory %d",
+	}
+	scanfOut := [][]interface{}{
+		{&ss.SocketsUsed},
+		{&ss.TCPInUse, &ss.TCPOrphan, &ss.TCPTW, &ss.TCPAlloc, &ss.TCPMem},
+		{&ss.UDPInUse, &ss.UDPMem},
+		{&ss.UDPLiteInUse},
+		{&ss.RawInUse},
+		{&ss.FragInUse, &ss.FragMemory},
+	}
+
+	scanner := bufio.NewScanner(fd)
+
+	iter := 0
+	for scanner.Scan() {
+		//bail if we've iterated more times than expected
+		if iter >= len(scanfLines) {
+			return ss, nil
+		}
+		txt := scanner.Text()
+		count, err := fmt.Sscanf(txt, scanfLines[iter], scanfOut[iter]...)
+		if err != nil {
+			return ss, errors.Wrap(err, "error reading sockstat")
+		}
+		if count != len(scanfOut[iter]) {
+			return ss, fmt.Errorf("did not match fields in line %s", scanfLines[iter])
+		}
+
+		iter++
+	}
+
+	if err = scanner.Err(); err != nil {
+		return ss, errors.Wrap(err, "error in scan")
+	}
+
+	return ss, nil
+}

metricbeat/module/system/socket_summary/sockstat_linux_test.go

@@ -0,0 +1,37 @@
+// 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 linux
+
+package socket_summary
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSockstat(t *testing.T) {
+	out, err := parseSockstat("./_meta/testdata/sockstat")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Equal(t, out.TCPMem, 1)
+	assert.Equal(t, out.TCPOrphan, 0)
+	assert.Equal(t, out.UDPMem, 11)
+}

metricbeat/module/system/socket_summary/sockstat_other.go

@@ -0,0 +1,28 @@
+// 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 !linux
+
+package socket_summary
+
+import "github.com/elastic/beats/libbeat/common"
+
+//a stub function for non-linux systems
+//get a list of platform-specific enhancements and apply them to our mapStr object.
+func applyEnhancements(data common.MapStr, m *MetricSet) (common.MapStr, error) {
+	return data, nil
+}