Migrate system socket fields metricset to ECS

CHANGELOG.next.asciidoc

@@ -23,6 +23,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
   leaf field `user.group` is now the `group` field set. {pull}10275[10275]
 
 *Auditbeat*
+
 - Rename `process.exe` to `process.executable` in auditd module to align with ECS. {pull}9949[9949]
 - Rename `process.cwd` to `process.working_directory` in auditd module to align with ECS. {pull}10195[10195]
 - Change data type of `process.pid` and `process.ppid` to number in JSON output
@@ -74,6 +75,9 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 - Refactor Prometheus metric mappings {pull}9948[9948]
 - Removed Prometheus stats metricset in favor of just using Prometheus collector {pull}9948[9948]
+- Migrate system process metricset fields to ECS. {pull}10339[10339]
+- Migrate system socket metricset fields to ECS. {pull}10339[10339]
+- Renamed direction values in sockets to ECS recommendations, from incoming/outcoming to inbound/outbound. {pull}10339[10339]
 - Adjust Redis.info metricset fields to ECS. {pull}10319[10319]
 
 *Packetbeat*

dev-tools/ecs-migration.yml

@@ -1223,6 +1223,43 @@
   alias: true
   beat: metricbeat
 
+### System
+
+- from: system.socket.direction
+  to: network.direction
+  alias: true
+  beat: metricbeat
+
+- from: system.socket.family
+  to: network.type
+  alias: true
+  beat: metricbeat
+
+- from: system.socket.process.command
+  to: process.name
+  alias: true
+  beat: metricbeat
+
+- from: system.socket.process.exe
+  to: process.executable
+  alias: true
+  beat: metricbeat
+
+- from: system.socket.process.pid
+  to: process.pid
+  alias: true
+  beat: metricbeat
+
+- from: system.socket.user.id
+  to: user.id
+  alias: true
+  beat: metricbeat
+
+- from: system.socket.user.name
+  to: user.name
+  alias: true
+  beat: metricbeat
+
 ### Zookeeper
 
 - from: zookeeper.mntr.version

metricbeat/docs/fields.asciidoc

@@ -24399,24 +24399,18 @@ TCP sockets that are active.
 *`system.socket.direction`*::
 +
 --
-type: keyword
-
-example: incoming
-
-How the socket was initiated. Possible values are incoming, outgoing, or listening.
+type: alias
 
+alias to: network.direction
 
 --
 
 *`system.socket.family`*::
 +
 --
-type: keyword
-
-example: ipv4
-
-Address family.
+type: alias
 
+alias to: network.type
 
 --
 
@@ -24505,20 +24499,18 @@ Error describing the cause of the reverse lookup failure.
 *`system.socket.process.pid`*::
 +
 --
-type: long
-
-ID of the process that opened the socket.
+type: alias
 
+alias to: process.pid
 
 --
 
 *`system.socket.process.command`*::
 +
 --
-type: keyword
-
-Name of the command (limited to 20 chars by the OS).
+type: alias
 
+alias to: process.name
 
 --
 
@@ -24527,37 +24519,35 @@ Name of the command (limited to 20 chars by the OS).
 --
 type: keyword
 
+Full command line
 
 
 --
 
 *`system.socket.process.exe`*::
 +
 --
-type: keyword
-
-Absolute path to the executable.
+type: alias
 
+alias to: process.executable
 
 --
 
 *`system.socket.user.id`*::
 +
 --
-type: long
-
-UID of the user running the process.
+type: alias
 
+alias to: user.id
 
 --
 
 *`system.socket.user.name`*::
 +
 --
-type: keyword
-
-Name of the user running the process.
+type: alias
 
+alias to: user.full_name
 
 --
 

metricbeat/helper/socket/listeners.go

@@ -26,20 +26,27 @@ type Direction uint8
 
 const (
 	_ Direction = iota
-	// Incoming indicates a connection was established from the outside to
+	// Inbound indicates a connection was established from the outside to
 	// listening socket on this host.
-	Incoming
-	// Outgoing indicates a connection was established from this socket to an
+	Inbound
+	// Outbound indicates a connection was established from this socket to an
 	// external listening socket.
-	Outgoing
+	Outbound
 	// Listening indicates a socket that is listening.
 	Listening
 )
 
+// Names for the direction of a connection
+const (
+	InboundName   = "inbound"
+	OutboundName  = "outbound"
+	ListeningName = "listening"
+)
+
 var directionNames = map[Direction]string{
-	Incoming:  "incoming",
-	Outgoing:  "outgoing",
-	Listening: "listening",
+	Inbound:   InboundName,
+	Outbound:  OutboundName,
+	Listening: ListeningName,
 }
 
 func (d Direction) String() string {
@@ -103,7 +110,7 @@ func (t *ListenerTable) Put(proto uint8, ip net.IP, port int) {
 
 // Direction returns whether the connection was incoming or outgoing based on
 // the protocol and local address. It compares the given local address to the
-// listeners in the table for the protocol and returns Incoming if there is a
+// listeners in the table for the protocol and returns Inbound if there is a
 // match. If remotePort is 0 then Listening is returned.
 func (t *ListenerTable) Direction(
 	proto uint8,
@@ -117,13 +124,13 @@ func (t *ListenerTable) Direction(
 	// Are there any listeners on the given protocol?
 	ports, exists := t.data[proto]
 	if !exists {
-		return Outgoing
+		return Outbound
 	}
 
 	// Is there any listener on the port?
 	interfaces, exists := ports[localPort]
 	if !exists {
-		return Outgoing
+		return Outbound
 	}
 
 	// Is there a listener that specific interface? OR
@@ -132,13 +139,13 @@ func (t *ListenerTable) Direction(
 	for _, ip := range interfaces.ips {
 		switch {
 		case ip.Equal(localIP):
-			return Incoming
+			return Inbound
 		case ip.Equal(net.IPv4zero) && isIPv4:
-			return Incoming
+			return Inbound
 		case ip.Equal(net.IPv6zero) && !isIPv4:
-			return Incoming
+			return Inbound
 		}
 	}
 
-	return Outgoing
+	return Outbound
 }

metricbeat/helper/socket/ptable.go

@@ -34,6 +34,7 @@ type Proc struct {
 	Command    string
 	Executable string
 	CmdLine    string
+	Args       []string
 }
 
 // ProcTable contains all of the active processes (if the current user is root).
@@ -101,6 +102,7 @@ func (t *ProcTable) Refresh() error {
 			if cmdline, err := p.CmdLine(); err != nil {
 				errs = append(errs, err)
 			} else {
+				proc.Args = cmdline
 				proc.CmdLine = strings.Join(cmdline, " ")
 			}
 		}

metricbeat/mb/testing/data_generator.go

@@ -77,29 +77,25 @@ func WriteEventsCond(f mb.EventsFetcher, t testing.TB, cond func(e common.MapStr
 		return fmt.Errorf("no events were generated")
 	}
 
-	var event *common.MapStr
-	if cond == nil {
-		event = &events[0]
-	} else {
-		for _, e := range events {
-			if cond(e) {
-				event = &e
-				break
-			}
-		}
-		if event == nil {
-			return fmt.Errorf("no events satisfied the condition")
-		}
+	event, err := SelectEvent(events, cond)
+	if err != nil {
+		return err
 	}
 
-	fullEvent := CreateFullEvent(f, *event)
+	fullEvent := CreateFullEvent(f, event)
 	WriteEventToDataJSON(t, fullEvent, "")
 	return nil
 }
 
 // WriteEventsReporterV2 fetches events and writes the first event to a ./_meta/data.json
 // file.
 func WriteEventsReporterV2(f mb.ReportingMetricSetV2, t testing.TB, path string) error {
+	return WriteEventsReporterV2Cond(f, t, path, nil)
+}
+
+// WriteEventsReporterV2Cond fetches events and writes the first event that matches
+// the condition to a file.
+func WriteEventsReporterV2Cond(f mb.ReportingMetricSetV2, t testing.TB, path string, cond func(common.MapStr) bool) error {
 	if !*dataFlag {
 		t.Skip("skip data generation tests")
 	}
@@ -113,7 +109,12 @@ func WriteEventsReporterV2(f mb.ReportingMetricSetV2, t testing.TB, path string)
 		return fmt.Errorf("no events were generated")
 	}
 
-	e := StandardizeEvent(f, events[0], mb.AddMetricSetInfo)
+	match, err := SelectEventV2(f, events, cond)
+	if err != nil {
+		return err
+	}
+
+	e := StandardizeEvent(f, match, mb.AddMetricSetInfo)
 
 	WriteEventToDataJSON(t, e, path)
 	return nil
@@ -169,7 +170,11 @@ func WriteEventToDataJSON(t testing.TB, fullEvent beat.Event, postfixPath string
 		t.Fatal(err)
 	}
 
-	p = path.Join(p, postfixPath, "_meta", "data.json")
+	if stat, err := os.Stat(postfixPath); err == nil && stat.IsDir() {
+		p = path.Join(p, postfixPath, "_meta", "data.json")
+	} else {
+		p = postfixPath
+	}
 
 	fields := fullEvent.Fields
 	fields["@timestamp"] = fullEvent.Timestamp
@@ -183,3 +188,30 @@ func WriteEventToDataJSON(t testing.TB, fullEvent beat.Event, postfixPath string
 		t.Fatal(err)
 	}
 }
+
+// SelectEvent selects the first event that matches an specific condition
+func SelectEvent(events []common.MapStr, cond func(e common.MapStr) bool) (common.MapStr, error) {
+	if cond == nil && len(events) > 0 {
+		return events[0], nil
+	}
+	for _, e := range events {
+		if cond(e) {
+			return e, nil
+		}
+	}
+	return nil, fmt.Errorf("no events satisfied the condition")
+}
+
+// SelectEventV2 selects the first event that matches an specific condition
+func SelectEventV2(f mb.ReportingMetricSetV2, events []mb.Event, cond func(e common.MapStr) bool) (mb.Event, error) {
+	if cond == nil && len(events) > 0 {
+		return events[0], nil
+	}
+	for _, e := range events {
+		fields := StandardizeEvent(f, e, mb.AddMetricSetInfo).Fields
+		if cond(fields) {
+			return e, nil
+		}
+	}
+	return mb.Event{}, fmt.Errorf("no events satisfied the condition")
+}