Passing socks5 test

Socks5 tests added

heartbeat/monitors/active/dialchain/socks5.go

@@ -47,7 +47,6 @@ func SOCKS5Layer(config *transport.ProxyConfig) Layer {
 		return afterDial(dialer, func(conn net.Conn) (net.Conn, error) {
 			// TODO: extract connection parameter from connection object?
 			// TODO: add proxy url to event?
-
 			timer.stop()
 			event.Fields.Put("socks5.rtt.connect", look.RTT(timer.duration()))
 			return conn, nil

heartbeat/monitors/active/tcp/tcp.go

@@ -154,7 +154,7 @@ func (jf *jobFactory) makeDirectEndpointJob(endpointURL *url.URL) (jobs.Job, err
 	return job, nil
 }
 
-// makeDirectEndpointJob makes jobs that use a Socks5 proxy to perform DNS lookups
+// makeSocksLookupEndpointJob makes jobs that use a Socks5 proxy to perform DNS lookups
 func (jf *jobFactory) makeSocksLookupEndpointJob(endpointURL *url.URL) (jobs.Job, error) {
 	return wrappers.WithURLField(endpointURL,
 		jobs.MakeSimpleJob(func(event *beat.Event) error {
@@ -169,12 +169,15 @@ func (jf *jobFactory) makeSocksLookupEndpointJob(endpointURL *url.URL) (jobs.Job
 // also which hostname should be passed to the TLS implementation for validation of the server cert.
 func (jf *jobFactory) dial(event *beat.Event, dialAddr string, canonicalURL *url.URL) error {
 	dc := &dialchain.DialerChain{
-		Net: dialchain.MakeConstAddrDialer(dialAddr, dialchain.TCPDialer(jf.config.Timeout)),
+		Net: dialchain.CreateNetDialer(jf.config.Timeout),
 	}
+
 	if jf.config.Socks5.URL != "" {
 		dc.AddLayer(dialchain.SOCKS5Layer(&jf.config.Socks5))
 	}
 
+	dc.AddLayer(dialchain.ConstAddrLayer(dialAddr))
+
 	isTLS := true
 	if canonicalURL.Scheme == "tcp" || canonicalURL.Scheme == "plain" {
 		isTLS = false

heartbeat/monitors/active/tcp/tcp_test.go

@@ -26,9 +26,11 @@ import (
 	"net/url"
 	"os"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/armon/go-socks5"
 	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/beats/v7/heartbeat/hbtest"
@@ -316,6 +318,69 @@ func TestNXDomainJob(t *testing.T) {
 	)
 }
 
+func TestSocks5Job(t *testing.T) {
+	scenarios := []struct {
+		name          string
+		localResolver bool
+	}{
+		{
+			name:          "using local resolver",
+			localResolver: true,
+		},
+		{
+			name:          "not using local resolver",
+			localResolver: false,
+		},
+	}
+
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			host, port, ip, closeEcho, err := startEchoServer(t)
+			require.NoError(t, err)
+			defer closeEcho()
+
+			_, proxyPort, proxyIp, closeProxy, err := startSocks5Server(t)
+			require.NoError(t, err)
+			defer closeProxy()
+
+			proxyURL := &url.URL{Scheme: "socks5", Host: net.JoinHostPort(proxyIp, fmt.Sprint(proxyPort))}
+			configMap := common.MapStr{
+				"hosts":                    host,
+				"ports":                    port,
+				"timeout":                  "1s",
+				"proxy_url":                proxyURL.String(),
+				"proxy_use_local_resolver": scenario.localResolver,
+				"check.receive":            "echo123",
+				"check.send":               "echo123",
+			}
+			event := testTCPConfigCheck(t, configMap, host, port)
+
+			testslike.Test(
+				t,
+				lookslike.Strict(lookslike.Compose(
+					hbtest.BaseChecks(ip, "up", "tcp"),
+					hbtest.RespondingTCPChecks(),
+					hbtest.SimpleURLChecks(t, "tcp", host, port),
+					hbtest.SummaryChecks(1, 0),
+					lookslike.MustCompile(map[string]interface{}{
+						"resolve": map[string]interface{}{
+							"ip":     ip,
+							"rtt.us": isdef.IsDuration,
+						},
+						"tcp": map[string]interface{}{
+							"rtt.validate.us": isdef.IsDuration,
+						},
+						"socks5": map[string]interface{}{
+							"rtt.connect.us": isdef.IsDuration,
+						},
+					}),
+				)),
+				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.
@@ -346,3 +411,41 @@ func startEchoServer(t *testing.T) (host string, port uint16, ip string, close f
 
 	return "localhost", uint16(portUint64), ip, listener.Close, nil
 }
+
+func startSocks5Server(t *testing.T) (host string, port uint16, ip string, close func() error, err error) {
+	host = "localhost"
+	config := &socks5.Config{}
+	server, err := socks5.New(config)
+	if err != nil {
+		return "", 0, "", nil, err
+	}
+
+	listener, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		return "", 0, "", nil, err
+	}
+	ip, portStr, err := net.SplitHostPort(listener.Addr().String())
+	portUint64, err := strconv.ParseUint(portStr, 10, 16)
+	if err != nil {
+		listener.Close()
+		return "", 0, "", nil, err
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		if err := server.Serve(listener); err != nil {
+			debugf("Error in SOCKS5 Test Server %v", err)
+		}
+		wg.Done()
+	}()
+
+	return host, uint16(portUint64), ip, func() error {
+		err := listener.Close()
+		if err != nil {
+			return err
+		}
+		wg.Wait()
+		return nil
+	}, nil
+}