Parse also the port from log sources

CHANGELOG.asciidoc

@@ -48,6 +48,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha1...master[Check the HEAD d
 - Fix installation of haproxy dashboard. {issue}9307[9307] {pull}9313[9313]
 - Don't generate incomplete configurations when logs collection is disabled by hints. {pull}9305[9305]
 - Stop runners disabled by hints after previously being started. {pull}9305[9305]
+- Use a separate field for source port on network inputs. {pull}9460[9460]
 
 *Heartbeat*
 

dev-tools/ecs-migration.yml

@@ -29,7 +29,7 @@
   alias: true
 
 - from: source
-  to: ["log.file.path", "log.source.ip"]
+  to: ["log.file.path", "log.source.ip", "log.source.port"]
   alias: false
 
 - from: beat.name

filebeat/_meta/fields.common.yml

@@ -17,6 +17,12 @@
       description: >
         Source IP from which the log event was read / sent from.
 
+    - name: log.source.port
+      type: long
+      required: false
+      description: >
+        Source port from which the log event was read / sent from.
+
     - name: log.offset
       type: long
       required: false

filebeat/docs/fields.asciidoc

@@ -4709,6 +4709,18 @@ required: False
 Source IP from which the log event was read / sent from.
 
 
+--
+
+*`log.source.port`*::
++
+--
+type: long
+
+required: False
+
+Source port from which the log event was read / sent from.
+
+
 --
 
 *`log.offset`*::

filebeat/input/syslog/input.go

@@ -18,6 +18,7 @@
 package syslog
 
 import (
+	"net"
 	"strings"
 	"sync"
 	"time"
@@ -201,14 +202,18 @@ func (p *Input) Wait() {
 }
 
 func createEvent(ev *event, metadata inputsource.NetworkMetadata, timezone *time.Location, log *logp.Logger) *beat.Event {
+	remoteHost, remotePort, _ := net.SplitHostPort(metadata.RemoteAddr.String())
 	f := common.MapStr{
 		"message": strings.TrimRight(ev.Message(), "\n"),
 		"log": common.MapStr{
 			"source": common.MapStr{
-				"ip": metadata.RemoteAddr.String(),
+				"ip": remoteHost,
 			},
 		},
 	}
+	if remotePort != "" {
+		f.Put("log.source.port", remotePort)
+	}
 
 	syslog := common.MapStr{}
 	event := common.MapStr{}

filebeat/input/tcp/input.go

@@ -19,6 +19,7 @@ package tcp
 
 import (
 	"fmt"
+	"net"
 	"sync"
 	"time"
 
@@ -126,17 +127,21 @@ func (p *Input) Wait() {
 }
 
 func createEvent(raw []byte, metadata inputsource.NetworkMetadata) *util.Data {
+	remoteHost, remotePort, _ := net.SplitHostPort(metadata.RemoteAddr.String())
 	data := util.NewData()
 	data.Event = beat.Event{
 		Timestamp: time.Now(),
 		Fields: common.MapStr{
 			"message": string(raw),
 			"log": common.MapStr{
 				"source": common.MapStr{
-					"ip": metadata.RemoteAddr.String(),
+					"ip": remoteHost,
 				},
 			},
 		},
 	}
+	if remotePort != "" {
+		data.Event.Fields.Put("log.source.port", remotePort)
+	}
 	return data
 }

filebeat/input/udp/input.go

@@ -18,6 +18,7 @@
 package udp
 
 import (
+	"net"
 	"sync"
 	"time"
 
@@ -66,6 +67,7 @@ func NewInput(
 
 	forwarder := harvester.NewForwarder(out)
 	callback := func(data []byte, metadata inputsource.NetworkMetadata) {
+		remoteHost, remotePort, _ := net.SplitHostPort(metadata.RemoteAddr.String())
 		e := util.NewData()
 		e.Event = beat.Event{
 			Timestamp: time.Now(),
@@ -76,11 +78,14 @@ func NewInput(
 				"message": string(data),
 				"log": common.MapStr{
 					"source": common.MapStr{
-						"ip": metadata.RemoteAddr.String(),
+						"ip": remoteHost,
 					},
 				},
 			},
 		}
+		if remotePort != "" {
+			e.Event.Fields.Put("log.source.port", remotePort)
+		}
 		forwarder.Send(e)
 	}