filebeat/input/{tcp,udp}: don't panic on nil addresses (#33837)

The callback may be called with a nil RemoteAddr in the metadata
parameter, so ensure that this is not nil before attempting to deference
it.

(cherry picked from commit 105fbda)

CHANGELOG.next.asciidoc

@@ -47,6 +47,12 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 *Heartbeat*
 
 - Fix broken zip URL monitors. NOTE: Zip URL Monitors will be removed in version 8.7 and replaced with project monitors. {pull}33723[33723]
+- Fix bug affecting let's encrypt and other users of cross-signed certs, where cert expiration was incorrectly calculated. {issue}33215[33215]
+- Fix broken disable feature for kibana configured monitors. {pull}33293[33293]
+- Fix states client support for output options. {pull}33405[33405]
+- Fix states client reloader under managed mode. {pull}33405[33405]
+- Fix bug where states.duration_ms was incorrect type. {pull}33563[33563]
+- Fix handling of long UDP messages in UDP input. {issue}33836[33836] {pull}33837[33837]
 
 *Auditbeat*
 

filebeat/input/tcp/input.go

@@ -71,7 +71,7 @@ func NewInput(
 
 	cb := func(data []byte, metadata inputsource.NetworkMetadata) {
 		event := createEvent(data, metadata)
-		forwarder.Send(event)
+		_ = forwarder.Send(event)
 	}
 
 	splitFunc, err := streaming.SplitFunc(config.Framing, []byte(config.LineDelimiter))
@@ -128,15 +128,18 @@ func (p *Input) Wait() {
 }
 
 func createEvent(raw []byte, metadata inputsource.NetworkMetadata) beat.Event {
-	return beat.Event{
+	evt := beat.Event{
 		Timestamp: time.Now(),
 		Fields: mapstr.M{
 			"message": string(raw),
-			"log": mapstr.M{
-				"source": mapstr.M{
-					"address": metadata.RemoteAddr.String(),
-				},
-			},
 		},
 	}
+	if metadata.RemoteAddr != nil {
+		evt.Fields["log"] = mapstr.M{
+			"source": mapstr.M{
+				"address": metadata.RemoteAddr.String(),
+			},
+		}
+	}
+	return evt
 }

filebeat/input/udp/input.go

@@ -65,20 +65,23 @@ func NewInput(
 
 	forwarder := harvester.NewForwarder(out)
 	callback := func(data []byte, metadata inputsource.NetworkMetadata) {
-		forwarder.Send(beat.Event{
+		evt := beat.Event{
 			Timestamp: time.Now(),
 			Meta: mapstr.M{
 				"truncated": metadata.Truncated,
 			},
 			Fields: mapstr.M{
 				"message": string(data),
-				"log": mapstr.M{
-					"source": mapstr.M{
-						"address": metadata.RemoteAddr.String(),
-					},
-				},
 			},
-		})
+		}
+		if metadata.RemoteAddr != nil {
+			evt.Fields["log"] = mapstr.M{
+				"source": mapstr.M{
+					"address": metadata.RemoteAddr.String(),
+				},
+			}
+		}
+		_ = forwarder.Send(evt)
 	}
 
 	udp := udp.New(&config.Config, callback)