feat(socket_listener): Add vsock support to socket listener and writer (#14172)

docs/LICENSE_OF_DEPENDENCIES.md

@@ -242,6 +242,7 @@ following works:
 - github.com/mdlayher/genetlink [MIT License](https://github.com/mdlayher/genetlink/blob/master/LICENSE.md)
 - github.com/mdlayher/netlink [MIT License](https://github.com/mdlayher/netlink/blob/master/LICENSE.md)
 - github.com/mdlayher/socket [MIT License](https://github.com/mdlayher/socket/blob/master/LICENSE.md)
+- github.com/mdlayher/vsock [MIT License](https://github.com/mdlayher/vsock/blob/master/LICENSE.md)
 - github.com/microsoft/ApplicationInsights-Go [MIT License](https://github.com/microsoft/ApplicationInsights-Go/blob/master/LICENSE)
 - github.com/microsoft/go-mssqldb [BSD 3-Clause "New" or "Revised" License](https://github.com/microsoft/go-mssqldb/blob/master/LICENSE.txt)
 - github.com/miekg/dns [BSD 3-Clause Clear License](https://github.com/miekg/dns/blob/master/LICENSE)

go.mod

@@ -128,6 +128,7 @@ require (
 	github.com/lxc/lxd v0.0.0-20220920163450-e9b4b514106a
 	github.com/matttproud/golang_protobuf_extensions v1.0.4
 	github.com/mdlayher/apcupsd v0.0.0-20220319200143-473c7b5f3c6a
+	github.com/mdlayher/vsock v1.1.1
 	github.com/microsoft/ApplicationInsights-Go v0.4.4
 	github.com/microsoft/go-mssqldb v1.5.0
 	github.com/miekg/dns v1.1.56

go.sum

@@ -1095,6 +1095,8 @@ github.com/mdlayher/socket v0.0.0-20211102153432-57e3fa563ecb/go.mod h1:nFZ1EtZY
 github.com/mdlayher/socket v0.1.1/go.mod h1:mYV5YIZAfHh4dzDVzI8x8tWLWCliuX8Mon5Awbj+qDs=
 github.com/mdlayher/socket v0.2.3 h1:XZA2X2TjdOwNoNPVPclRCURoX/hokBY8nkTmRZFEheM=
 github.com/mdlayher/socket v0.2.3/go.mod h1:bz12/FozYNH/VbvC3q7TRIK/Y6dH1kCKsXaUeXi/FmY=
+github.com/mdlayher/vsock v1.1.1 h1:8lFuiXQnmICBrCIIA9PMgVSke6Fg6V4+r0v7r55k88I=
+github.com/mdlayher/vsock v1.1.1/go.mod h1:Y43jzcy7KM3QB+/FK15pfqGxDMCMzUXWegEfIbSM18U=
 github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE=
 github.com/mholt/archiver/v3 v3.5.0/go.mod h1:qqTTPUK/HZPFgFQ/TJ3BzvTpF/dPtFVJXdQbCmeMxwc=
 github.com/microsoft/ApplicationInsights-Go v0.4.4 h1:G4+H9WNs6ygSCe6sUyxRc2U81TI5Es90b2t/MwX5KqY=

plugins/inputs/socket_listener/README.md

@@ -42,6 +42,7 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   # service_address = "udp6://:8094"
   # service_address = "unix:///tmp/telegraf.sock"
   # service_address = "unixgram:///tmp/telegraf.sock"
+  # service_address = "vsock://cid:port"
 
   ## Change the file mode bits on unix sockets.  These permissions may not be
   ## respected by some platforms, to safely restrict write permissions it is best

plugins/inputs/socket_listener/sample.conf

@@ -11,6 +11,7 @@
   # service_address = "udp6://:8094"
   # service_address = "unix:///tmp/telegraf.sock"
   # service_address = "unixgram:///tmp/telegraf.sock"
+  # service_address = "vsock://cid:port"
 
   ## Change the file mode bits on unix sockets.  These permissions may not be
   ## respected by some platforms, to safely restrict write permissions it is best

plugins/inputs/socket_listener/socket_listener.go

@@ -223,6 +223,22 @@ func (sl *SocketListener) Start(acc telegraf.Accumulator) error {
 			return err
 		}
 		sl.listener = psl
+	case "vsock":
+		ssl := &streamListener{
+			ReadBufferSize:  int(sl.ReadBufferSize),
+			ReadTimeout:     sl.ReadTimeout,
+			KeepAlivePeriod: sl.KeepAlivePeriod,
+			MaxConnections:  sl.MaxConnections,
+			Encoding:        sl.ContentEncoding,
+			Splitter:        sl.splitter,
+			Parser:          sl.parser,
+			Log:             sl.Log,
+		}
+
+		if err := ssl.setupVsock(u); err != nil {
+			return err
+		}
+		sl.listener = ssl
 	default:
 		return fmt.Errorf("unknown protocol %q in %q", u.Scheme, sl.ServiceAddress)
 	}

plugins/inputs/socket_listener/stream_listener.go

@@ -6,10 +6,12 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"net/url"
 	"os"
 	"strconv"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -17,6 +19,7 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/internal"
+	"github.com/mdlayher/vsock"
 )
 
 type hasSetReadBuffer interface {
@@ -83,6 +86,30 @@ func (l *streamListener) setupUnix(u *url.URL, tlsCfg *tls.Config, socketMode st
 	return nil
 }
 
+func (l *streamListener) setupVsock(u *url.URL) error {
+	var err error
+
+	addrTuple := strings.SplitN(u.String(), ":", 2)
+
+	// Check address string for containing two tokens
+	if len(addrTuple) < 2 {
+		return fmt.Errorf("CID and/or port number missing")
+	}
+	// Parse CID and port number from address string both being 32-bit
+	// source: https://man7.org/linux/man-pages/man7/vsock.7.html
+	cid, _ := strconv.ParseUint(addrTuple[0], 10, 32)
+	if (cid >= uint64(math.Pow(2, 32))-1) && (cid <= 0) {
+		return fmt.Errorf("CID %d is out of range", cid)
+	}
+	port, _ := strconv.ParseUint(addrTuple[1], 10, 32)
+	if (port >= uint64(math.Pow(2, 32))-1) && (port <= 0) {
+		return fmt.Errorf("Port numner %d is out of range", port)
+	}
+
+	l.listener, err = vsock.Listen(uint32(port), nil)
+	return err
+}
+
 func (l *streamListener) setupConnection(conn net.Conn) error {
 	if c, ok := conn.(*tls.Conn); ok {
 		conn = c.NetConn()

plugins/outputs/socket_writer/README.md

@@ -31,6 +31,7 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   # address = "udp6://127.0.0.1:8094"
   # address = "unix:///tmp/telegraf.sock"
   # address = "unixgram:///tmp/telegraf.sock"
+  # address = "vsock://cid:port"
 
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"

plugins/outputs/socket_writer/sample.conf

@@ -11,6 +11,7 @@
   # address = "udp6://127.0.0.1:8094"
   # address = "unix:///tmp/telegraf.sock"
   # address = "unixgram:///tmp/telegraf.sock"
+  # address = "vsock://cid:port"
 
   ## Optional TLS Config
   # tls_ca = "/etc/telegraf/ca.pem"

plugins/outputs/socket_writer/socket_writer.go

@@ -6,7 +6,9 @@ import (
 	_ "embed"
 	"errors"
 	"fmt"
+	"math"
 	"net"
+	"strconv"
 	"strings"
 	"time"
 
@@ -16,6 +18,7 @@ import (
 	tlsint "github.com/influxdata/telegraf/plugins/common/tls"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/serializers"
+	"github.com/mdlayher/vsock"
 )
 
 //go:embed sample.conf
@@ -55,13 +58,38 @@ func (sw *SocketWriter) Connect() error {
 	}
 
 	var c net.Conn
-	if tlsCfg == nil {
-		c, err = net.Dial(spl[0], spl[1])
+
+	if spl[0] == "vsock" {
+		addrTuple := strings.SplitN(spl[1], ":", 2)
+
+		// Check address string for containing two
+		if len(addrTuple) < 2 {
+			return fmt.Errorf("CID and/or port number missing")
+		}
+
+		// Parse CID and port number from address string both being 32-bit
+		// source: https://man7.org/linux/man-pages/man7/vsock.7.html
+		cid, _ := strconv.ParseUint(addrTuple[0], 10, 32)
+		if (cid >= uint64(math.Pow(2, 32))-1) && (cid <= 0) {
+			return fmt.Errorf("CID %d is out of range", cid)
+		}
+		port, _ := strconv.ParseUint(addrTuple[1], 10, 32)
+		if (port >= uint64(math.Pow(2, 32))-1) && (port <= 0) {
+			return fmt.Errorf("Port numner %d is out of range", port)
+		}
+		c, err = vsock.Dial(uint32(cid), uint32(port), nil)
+		if err != nil {
+			return err
+		}
 	} else {
-		c, err = tls.Dial(spl[0], spl[1], tlsCfg)
-	}
-	if err != nil {
-		return err
+		if tlsCfg == nil {
+			c, err = net.Dial(spl[0], spl[1])
+		} else {
+			c, err = tls.Dial(spl[0], spl[1], tlsCfg)
+		}
+		if err != nil {
+			return err
+		}
 	}
 
 	if err := sw.setKeepAlive(c); err != nil {