[Heartbeat] Fix status for TCP checks (elastic#11118)

The heartbeat checks for TCP send/receive have been broken for a long time, since at least 6.3 from my testing. An error was returned, but the status was still set to 'up'.

This patch fixes that and adds tests.

(cherry picked from commit e5c3485)

CHANGELOG.next.asciidoc

@@ -65,6 +65,10 @@ https://github.com/elastic/beats/compare/v7.0.0-beta1...master[Check the HEAD di
 
 *Heartbeat*
 
+- Made monitors.d configuration part of the default config. {pull}9004[9004]
+- Fixed rare issue where TLS connections to endpoints with x509 certificates missing either notBefore or notAfter would cause the check to fail with a stacktrace.  {pull}9566[9566]
+- Fix checks for TCP send/receive data {pull}11118[11118]
+
 *Journalbeat*
 
 *Metricbeat*

heartbeat/monitors/active/tcp/task.go

@@ -70,7 +70,8 @@ func pingHost(
 		},
 	})
 	if err != nil {
-		event.PutValue("error", reason.FailValidate(err))
+		return reason.MakeValidateError(err)
 	}
+
 	return nil
 }

heartbeat/monitors/active/tcp/tcp_test.go

@@ -25,6 +25,7 @@ import (
 	"net/http/httptest"
 	"net/url"
 	"os"
+	"strconv"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -38,11 +39,16 @@ import (
 )
 
 func testTCPCheck(t *testing.T, host string, port uint16) *beat.Event {
-	config, err := common.NewConfigFrom(common.MapStr{
+	config := common.MapStr{
 		"hosts":   host,
 		"ports":   port,
 		"timeout": "1s",
-	})
+	}
+	return testTCPConfigCheck(t, config, host, port)
+}
+
+func testTCPConfigCheck(t *testing.T, configMap common.MapStr, host string, port uint16) *beat.Event {
+	config, err := common.NewConfigFrom(configMap)
 	require.NoError(t, err)
 
 	jobs, endpoints, err := create("tcp", config)
@@ -194,3 +200,110 @@ func TestUnreachableEndpointJob(t *testing.T) {
 		event.Fields,
 	)
 }
+
+func TestCheckUp(t *testing.T) {
+	host, port, ip, closeEcho, err := startEchoServer(t)
+	require.NoError(t, err)
+	defer closeEcho()
+
+	configMap := common.MapStr{
+		"hosts":         host,
+		"ports":         port,
+		"timeout":       "1s",
+		"check.receive": "echo123",
+		"check.send":    "echo123",
+	}
+
+	event := testTCPConfigCheck(t, configMap, host, port)
+
+	mapval.Test(
+		t,
+		mapval.Strict(mapval.Compose(
+			tcpMonitorChecks(host, ip, port, "up"),
+			hbtest.RespondingTCPChecks(),
+			hbtest.SimpleURLChecks(t, "tcp", host, port),
+			hbtest.SummaryChecks(1, 0),
+			mapval.MustCompile(mapval.Map{
+				"resolve": mapval.Map{
+					"ip":     ip,
+					"rtt.us": mapval.IsDuration,
+				},
+				"tcp": mapval.Map{
+					"rtt.validate.us": mapval.IsDuration,
+				},
+			}),
+		)),
+		event.Fields,
+	)
+}
+
+func TestCheckDown(t *testing.T) {
+	host, port, ip, closeEcho, err := startEchoServer(t)
+	require.NoError(t, err)
+	defer closeEcho()
+
+	configMap := common.MapStr{
+		"hosts":         host,
+		"ports":         port,
+		"timeout":       "1s",
+		"check.receive": "BOOM", // should fail
+		"check.send":    "echo123",
+	}
+
+	event := testTCPConfigCheck(t, configMap, host, port)
+
+	mapval.Test(
+		t,
+		mapval.Strict(mapval.Compose(
+			tcpMonitorChecks(host, ip, port, "down"),
+			hbtest.RespondingTCPChecks(),
+			hbtest.SimpleURLChecks(t, "tcp", host, port),
+			hbtest.SummaryChecks(0, 1),
+			mapval.MustCompile(mapval.Map{
+				"resolve": mapval.Map{
+					"ip":     ip,
+					"rtt.us": mapval.IsDuration,
+				},
+				"tcp": mapval.Map{
+					"rtt.validate.us": mapval.IsDuration,
+				},
+				"error": mapval.Map{
+					"type":    "validate",
+					"message": "received string mismatch",
+				},
+			}),
+		)),
+		event.Fields,
+	)
+}
+
+// startEchoServer starts a simple TCP echo server for testing. Only handles a single connection once.
+// Note you MUST connect to this server exactly once to avoid leaking a goroutine. This is only useful
+// for the specific tests used here.
+func startEchoServer(t *testing.T) (host string, port uint16, ip string, close func() error, err error) {
+	// Simple echo server
+	listener, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		return "", 0, "", nil, err
+	}
+	go func() {
+		conn, err := listener.Accept()
+		require.NoError(t, err)
+		buf := make([]byte, 1024)
+		rlen, err := conn.Read(buf)
+		require.NoError(t, err)
+		wlen, err := conn.Write(buf[:rlen])
+		require.NoError(t, err)
+		// Normally we'd retry partial writes, but for tests this is OK
+		require.Equal(t, wlen, rlen)
+	}()
+
+	ip, portStr, err := net.SplitHostPort(listener.Addr().String())
+	portUint64, err := strconv.ParseUint(portStr, 10, 16)
+	if err != nil {
+		listener.Close()
+		return "", 0, "", nil, err
+	}
+
+	return "localhost", uint16(portUint64), ip, listener.Close, nil
+}

heartbeat/reason/reason.go

@@ -68,4 +68,7 @@ func Fail(r Reason) common.MapStr {
 
 func FailIO(err error) common.MapStr { return Fail(IOError{err}) }
 
-func FailValidate(err error) common.MapStr { return Fail(ValidateError{err}) }
+// MakeValidateError creates an instance of ValidateError from the given error.
+func MakeValidateError(err error) ValidateError {
+	return ValidateError{err}
+}