multisocket plugin (#6882)

* multisocket plugin improves performance in multiprocessor systems Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * - refactoring - update doc Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * remove port from reuseport plugin README Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * rename reuseport plugin to numsockets plugin Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * Add Recommendations to numsockets README Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * added numsockets test; made NUM_SOCKETS mandatory in doc Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * restart and whoami tests for numsockets plugin Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * default value for numsockets Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * caddy up Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * add numsockets to plugin.cfg Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * - rename numsockets plugin to multisocket - default as GOMAXPROCS - update README Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> * resolve conflicts Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com> --------- Signed-off-by: Viktor Rodionov <33463837+Shmillerov@users.noreply.github.com>
2024-11-13 20:40:25 +03:00 · 2024-11-13 20:40:25 +03:00 · 6c39f4bae7
commit 6c39f4bae7
parent 43fdf737d6
11 changed files with 438 additions and 56 deletions
--- a/core/dnsserver/config.go
+++ b/core/dnsserver/config.go
@ -24,6 +24,10 @@ type Config struct {
 	// The port to listen on.
 	Port string

+	// The number of servers that will listen on one port.
+	// By default, one server will be running.
+	NumSockets int
+
 	// Root points to a base directory we find user defined "things".
 	// First consumer is the file plugin to looks for zone files in this place.
 	Root string
--- a/core/dnsserver/register.go
+++ b/core/dnsserver/register.go
@ -134,69 +134,23 @@ func (h *dnsContext) InspectServerBlocks(sourceFile string, serverBlocks []caddy

 // MakeServers uses the newly-created siteConfigs to create and return a list of server instances.
 func (h *dnsContext) MakeServers() ([]caddy.Server, error) {
-	// Copy the Plugin, ListenHosts and Debug from first config in the block
-	// to all other config in the same block . Doing this results in zones
-	// sharing the same plugin instances and settings as other zones in
-	// the same block.
-	for _, c := range h.configs {
-		c.Plugin = c.firstConfigInBlock.Plugin
-		c.ListenHosts = c.firstConfigInBlock.ListenHosts
-		c.Debug = c.firstConfigInBlock.Debug
-		c.Stacktrace = c.firstConfigInBlock.Stacktrace
-
-		// Fork TLSConfig for each encrypted connection
-		c.TLSConfig = c.firstConfigInBlock.TLSConfig.Clone()
-		c.ReadTimeout = c.firstConfigInBlock.ReadTimeout
-		c.WriteTimeout = c.firstConfigInBlock.WriteTimeout
-		c.IdleTimeout = c.firstConfigInBlock.IdleTimeout
-		c.TsigSecret = c.firstConfigInBlock.TsigSecret
-	}
+	// Copy parameters from first config in the block to all other config in the same block
+	propagateConfigParams(h.configs)

 	// we must map (group) each config to a bind address
 	groups, err := groupConfigsByListenAddr(h.configs)
 	if err != nil {
 		return nil, err
 	}
+
 	// then we create a server for each group
 	var servers []caddy.Server
 	for addr, group := range groups {
-		// switch on addr
-		switch tr, _ := parse.Transport(addr); tr {
-		case transport.DNS:
-			s, err := NewServer(addr, group)
-			if err != nil {
-				return nil, err
-			}
-			servers = append(servers, s)
-
-		case transport.TLS:
-			s, err := NewServerTLS(addr, group)
-			if err != nil {
-				return nil, err
-			}
-			servers = append(servers, s)
-
-		case transport.QUIC:
-			s, err := NewServerQUIC(addr, group)
-			if err != nil {
-				return nil, err
-			}
-			servers = append(servers, s)
-
-		case transport.GRPC:
-			s, err := NewServergRPC(addr, group)
-			if err != nil {
-				return nil, err
-			}
-			servers = append(servers, s)
-
-		case transport.HTTPS:
-			s, err := NewServerHTTPS(addr, group)
-			if err != nil {
-				return nil, err
-			}
-			servers = append(servers, s)
+		serversForGroup, err := makeServersForGroup(addr, group)
+		if err != nil {
+			return nil, err
 		}
+		servers = append(servers, serversForGroup...)
 	}

 	// For each server config, check for View Filter plugins
@ -299,6 +253,27 @@ func (h *dnsContext) validateZonesAndListeningAddresses() error {
 	return nil
 }

+// propagateConfigParams copies the necessary parameters from first config in the block
+// to all other config in the same block. Doing this results in zones
+// sharing the same plugin instances and settings as other zones in
+// the same block.
+func propagateConfigParams(configs []*Config) {
+	for _, c := range configs {
+		c.Plugin = c.firstConfigInBlock.Plugin
+		c.ListenHosts = c.firstConfigInBlock.ListenHosts
+		c.Debug = c.firstConfigInBlock.Debug
+		c.Stacktrace = c.firstConfigInBlock.Stacktrace
+		c.NumSockets = c.firstConfigInBlock.NumSockets
+
+		// Fork TLSConfig for each encrypted connection
+		c.TLSConfig = c.firstConfigInBlock.TLSConfig.Clone()
+		c.ReadTimeout = c.firstConfigInBlock.ReadTimeout
+		c.WriteTimeout = c.firstConfigInBlock.WriteTimeout
+		c.IdleTimeout = c.firstConfigInBlock.IdleTimeout
+		c.TsigSecret = c.firstConfigInBlock.TsigSecret
+	}
+}
+
 // groupConfigsByListenAddr groups site configs by their listen
 // (bind) address, so sites that use the same listener can be served
 // on the same server instance. The return value maps the listen
@ -320,6 +295,63 @@ func groupConfigsByListenAddr(configs []*Config) (map[string][]*Config, error) {
 	return groups, nil
 }

+// makeServersForGroup creates servers for a specific transport and group.
+// It creates as many servers as specified in the NumSockets configuration.
+// If the NumSockets param is not specified, one server is created by default.
+func makeServersForGroup(addr string, group []*Config) ([]caddy.Server, error) {
+	// that is impossible, but better to check
+	if len(group) == 0 {
+		return nil, fmt.Errorf("no configs for group defined")
+	}
+	// create one server by default if no NumSockets specified
+	numSockets := 1
+	if group[0].NumSockets > 0 {
+		numSockets = group[0].NumSockets
+	}
+
+	var servers []caddy.Server
+	for range numSockets {
+		// switch on addr
+		switch tr, _ := parse.Transport(addr); tr {
+		case transport.DNS:
+			s, err := NewServer(addr, group)
+			if err != nil {
+				return nil, err
+			}
+			servers = append(servers, s)
+
+		case transport.TLS:
+			s, err := NewServerTLS(addr, group)
+			if err != nil {
+				return nil, err
+			}
+			servers = append(servers, s)
+
+		case transport.QUIC:
+			s, err := NewServerQUIC(addr, group)
+			if err != nil {
+				return nil, err
+			}
+			servers = append(servers, s)
+
+		case transport.GRPC:
+			s, err := NewServergRPC(addr, group)
+			if err != nil {
+				return nil, err
+			}
+			servers = append(servers, s)
+
+		case transport.HTTPS:
+			s, err := NewServerHTTPS(addr, group)
+			if err != nil {
+				return nil, err
+			}
+			servers = append(servers, s)
+		}
+	}
+	return servers, nil
+}
+
 // DefaultPort is the default port.
 const DefaultPort = transport.Port

--- a/core/dnsserver/zdirectives.go
+++ b/core/dnsserver/zdirectives.go
@ -16,6 +16,7 @@ var Directives = []string{
 	"cancel",
 	"tls",
 	"timeouts",
+	"multisocket",
 	"reload",
 	"nsid",
 	"bufsize",
--- a/core/plugin/zplugin.go
+++ b/core/plugin/zplugin.go
@ -39,6 +39,7 @@ import (
 	_ "github.com/coredns/coredns/plugin/metadata"
 	_ "github.com/coredns/coredns/plugin/metrics"
 	_ "github.com/coredns/coredns/plugin/minimal"
+	_ "github.com/coredns/coredns/plugin/multisocket"
 	_ "github.com/coredns/coredns/plugin/nsid"
 	_ "github.com/coredns/coredns/plugin/pprof"
 	_ "github.com/coredns/coredns/plugin/ready"
--- a/go.mod
+++ b/go.mod
@ -12,7 +12,7 @@ require (
 	github.com/aws/aws-sdk-go v1.55.5
 	github.com/aws/aws-sdk-go-v2/config v1.27.39
 	github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.33.3
-	github.com/coredns/caddy v1.1.1
+	github.com/coredns/caddy v1.1.2-0.20241029205200-8de985351a98
 	github.com/dnstap/golang-dnstap v0.4.0
 	github.com/expr-lang/expr v1.16.9
 	github.com/farsightsec/golang-framestream v0.3.0
--- a/go.sum
+++ b/go.sum
@ -92,8 +92,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
-github.com/coredns/caddy v1.1.1 h1:2eYKZT7i6yxIfGP3qLJoJ7HAsDJqYB+X68g4NYjSrE0=
-github.com/coredns/caddy v1.1.1/go.mod h1:A6ntJQlAWuQfFlsd9hvigKbo2WS0VUs2l1e2F+BawD4=
+github.com/coredns/caddy v1.1.2-0.20241029205200-8de985351a98 h1:c+Epklw9xk6BZ1OFBPWLA2PcL8QalKvl3if8CP9x8uw=
+github.com/coredns/caddy v1.1.2-0.20241029205200-8de985351a98/go.mod h1:A6ntJQlAWuQfFlsd9hvigKbo2WS0VUs2l1e2F+BawD4=
 github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI=
--- a/plugin.cfg
+++ b/plugin.cfg
@ -25,6 +25,7 @@ geoip:geoip
 cancel:cancel
 tls:tls
 timeouts:timeouts
+multisocket:multisocket
 reload:reload
 nsid:nsid
 bufsize:bufsize
--- a/plugin/multisocket/README.md
+++ b/plugin/multisocket/README.md
@ -0,0 +1,68 @@
+# multisocket
+
+## Name
+
+*multisocket* - allows to start multiple servers that will listen on one port.
+
+## Description
+
+With *multisocket*, you can define the number of servers that will listen on the same port. The SO_REUSEPORT socket 
+option allows to open multiple listening sockets at the same address and port. In this case, kernel distributes incoming 
+connections between sockets.
+
+Enabling this option allows to start multiple servers, which increases the throughput of CoreDNS in environments with a 
+large number of CPU cores.
+
+## Syntax
+
+~~~
+multisocket [NUM_SOCKETS]
+~~~
+
+* **NUM_SOCKETS** - the number of servers that will listen on one port. Default value is equal to GOMAXPROCS.
+
+## Examples
+
+Start 5 TCP/UDP servers on the same port.
+
+~~~ corefile
+. {
+	multisocket 5
+	forward . /etc/resolv.conf
+}
+~~~
+
+Do not define `NUM_SOCKETS`, in this case it will take a value equal to GOMAXPROCS.
+
+~~~ corefile
+. {
+	multisocket
+	forward . /etc/resolv.conf
+}
+~~~
+
+## Recommendations
+
+The tests of the `multisocket` plugin, which were conducted for `NUM_SOCKETS` from 1 to 10, did not reveal any side 
+effects or performance degradation.
+
+This means that the `multisocket` plugin can be used with a default value that is equal to GOMAXPROCS.
+
+However, to achieve the best results, it is recommended to consider the specific environment and plugins used in 
+CoreDNS. To determine the optimal configuration, it is advisable to conduct performance tests with different 
+`NUM_SOCKETS`, measuring Queries Per Second (QPS) and system load.
+
+If conducting such tests is difficult, follow these recommendations:
+1. Determine the maximum CPU consumption of CoreDNS server without `multisocket` plugin. Estimate how much CPU CoreDNS
+   actually consumes in specific environment under maximum load.
+2. Align `NUM_SOCKETS` with the estimated CPU usage and CPU limits or system's available resources.
+   Examples:
+   - If CoreDNS consumes 4 CPUs and 8 CPUs are available, set `NUM_SOCKETS` to 2.
+   - If CoreDNS consumes 8 CPUs and 64 CPUs are available, set `NUM_SOCKETS` to 8.
+
+## Limitations
+
+The SO_REUSEPORT socket option is not available for some operating systems. It is available since Linux Kernel 3.9 and 
+not available for Windows at all.
+
+Using this plugin with a system that does not support SO_REUSEPORT will cause an `address already in use` error.
--- a/plugin/multisocket/multisocket.go
+++ b/plugin/multisocket/multisocket.go
@ -0,0 +1,51 @@
+package multisocket
+
+import (
+	"fmt"
+	"runtime"
+	"strconv"
+
+	"github.com/coredns/caddy"
+	"github.com/coredns/coredns/core/dnsserver"
+	"github.com/coredns/coredns/plugin"
+)
+
+const pluginName = "multisocket"
+
+func init() { plugin.Register(pluginName, setup) }
+
+func setup(c *caddy.Controller) error {
+	err := parseNumSockets(c)
+	if err != nil {
+		return plugin.Error(pluginName, err)
+	}
+	return nil
+}
+
+func parseNumSockets(c *caddy.Controller) error {
+	config := dnsserver.GetConfig(c)
+	c.Next() // "multisocket"
+
+	args := c.RemainingArgs()
+
+	if len(args) > 1 || c.Next() {
+		return c.ArgErr()
+	}
+
+	if len(args) == 0 {
+		// Nothing specified; use default that is equal to GOMAXPROCS.
+		config.NumSockets = runtime.GOMAXPROCS(0)
+		return nil
+	}
+
+	numSockets, err := strconv.Atoi(args[0])
+	if err != nil {
+		return fmt.Errorf("invalid num sockets: %w", err)
+	}
+	if numSockets < 1 {
+		return fmt.Errorf("num sockets can not be zero or negative: %d", numSockets)
+	}
+	config.NumSockets = numSockets
+
+	return nil
+}
--- a/plugin/multisocket/multisocket_test.go
+++ b/plugin/multisocket/multisocket_test.go
@ -0,0 +1,55 @@
+package multisocket
+
+import (
+	"runtime"
+	"strings"
+	"testing"
+
+	"github.com/coredns/caddy"
+	"github.com/coredns/coredns/core/dnsserver"
+)
+
+func TestMultisocket(t *testing.T) {
+	tests := []struct {
+		input              string
+		shouldErr          bool
+		expectedNumSockets int
+		expectedErrContent string // substring from the expected error. Empty for positive cases.
+	}{
+		// positive
+		{`multisocket`, false, runtime.GOMAXPROCS(0), ""},
+		{`multisocket 2`, false, 2, ""},
+		{` multisocket 1`, false, 1, ""},
+		{`multisocket text`, true, 0, "invalid num sockets"},
+		{`multisocket 0`, true, 0, "num sockets can not be zero or negative"},
+		{`multisocket -1`, true, 0, "num sockets can not be zero or negative"},
+		{`multisocket 2 2`, true, 0, "Wrong argument count or unexpected line ending after '2'"},
+		{`multisocket 2 {
+			block
+		}`, true, 0, "Unexpected token '{', expecting argument"},
+	}
+
+	for i, test := range tests {
+		c := caddy.NewTestController("dns", test.input)
+		err := setup(c)
+		cfg := dnsserver.GetConfig(c)
+
+		if test.shouldErr && err == nil {
+			t.Errorf("Test %d: Expected error but found %s for input %s", i, err, test.input)
+		}
+
+		if err != nil {
+			if !test.shouldErr {
+				t.Errorf("Test %d: Expected no error but found one for input %s. Error was: %v", i, test.input, err)
+			}
+
+			if !strings.Contains(err.Error(), test.expectedErrContent) {
+				t.Errorf("Test %d: Expected error to contain: %v, found error: %v, input: %s", i, test.expectedErrContent, err, test.input)
+			}
+		}
+
+		if cfg.NumSockets != test.expectedNumSockets {
+			t.Errorf("Test %d: Expected num sockets to be %d, found %d", i, test.expectedNumSockets, cfg.NumSockets)
+		}
+	}
+}
--- a/test/multisocket_test.go
+++ b/test/multisocket_test.go
@ -0,0 +1,169 @@
+package test
+
+import (
+	"fmt"
+	"net"
+	"testing"
+
+	"github.com/miekg/dns"
+)
+
+// These tests need a fixed port, because :0 selects a random port for each socket, but we need all sockets to be on
+// the same port.
+
+func TestMultisocket(t *testing.T) {
+	tests := []struct {
+		name            string
+		corefile        string
+		expectedServers int
+		expectedErr     string
+		expectedPort    string
+	}{
+		{
+			name: "no multisocket",
+			corefile: `.:5054 {
+			}`,
+			expectedServers: 1,
+			expectedPort:    "5054",
+		},
+		{
+			name: "multisocket 1",
+			corefile: `.:5055 {
+				multisocket 1
+			}`,
+			expectedServers: 1,
+			expectedPort:    "5055",
+		},
+		{
+			name: "multisocket 2",
+			corefile: `.:5056 {
+				multisocket 2
+			}`,
+			expectedServers: 2,
+			expectedPort:    "5056",
+		},
+		{
+			name: "multisocket 100",
+			corefile: `.:5057 {
+				multisocket 100
+			}`,
+			expectedServers: 100,
+			expectedPort:    "5057",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s, err := CoreDNSServer(test.corefile)
+			defer s.Stop()
+			if err != nil {
+				t.Fatalf("Could not get CoreDNS serving instance: %s", err)
+			}
+			// check number of servers
+			if len(s.Servers()) != test.expectedServers {
+				t.Fatalf("Expected %d servers, got %d", test.expectedServers, len(s.Servers()))
+			}
+
+			// check that ports are the same
+			for _, listener := range s.Servers() {
+				if listener.Addr().String() != listener.LocalAddr().String() {
+					t.Fatalf("Expected tcp address %s to be on the same port as udp address %s",
+						listener.LocalAddr().String(), listener.Addr().String())
+				}
+				_, port, err := net.SplitHostPort(listener.Addr().String())
+				if err != nil {
+					t.Fatalf("Could not get port from listener addr: %s", err)
+				}
+				if port != test.expectedPort {
+					t.Fatalf("Expected port %s, got %s", test.expectedPort, port)
+				}
+			}
+		})
+	}
+}
+
+func TestMultisocket_Restart(t *testing.T) {
+	tests := []struct {
+		name             string
+		numSocketsBefore int
+		numSocketsAfter  int
+	}{
+		{
+			name:             "increase",
+			numSocketsBefore: 1,
+			numSocketsAfter:  2,
+		},
+		{
+			name:             "decrease",
+			numSocketsBefore: 2,
+			numSocketsAfter:  1,
+		},
+		{
+			name:             "no changes",
+			numSocketsBefore: 2,
+			numSocketsAfter:  2,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			corefile := `.:5058 {
+				multisocket %d
+			}`
+			srv, err := CoreDNSServer(fmt.Sprintf(corefile, test.numSocketsBefore))
+			if err != nil {
+				t.Fatalf("Could not get CoreDNS serving instance: %s", err)
+			}
+			if test.numSocketsBefore != len(srv.Servers()) {
+				t.Fatalf("Expected %d servers, got %d", test.numSocketsBefore, len(srv.Servers()))
+			}
+
+			newSrv, err := srv.Restart(NewInput(fmt.Sprintf(corefile, test.numSocketsAfter)))
+			if err != nil {
+				t.Fatalf("Could not get CoreDNS serving instance: %s", err)
+			}
+			if test.numSocketsAfter != len(newSrv.Servers()) {
+				t.Fatalf("Expected %d servers, got %d", test.numSocketsAfter, len(newSrv.Servers()))
+			}
+			newSrv.Stop()
+		})
+	}
+}
+
+// Just check that server with multisocket works
+func TestMultisocket_WhoAmI(t *testing.T) {
+	corefile := `.:5059 {
+		multisocket
+		whoami
+	}`
+	s, udp, tcp, err := CoreDNSServerAndPorts(corefile)
+	if err != nil {
+		t.Fatalf("Could not get CoreDNS serving instance: %s", err)
+	}
+	defer s.Stop()
+
+	m := new(dns.Msg)
+	m.SetQuestion("whoami.example.org.", dns.TypeA)
+
+	// check udp
+	cl := dns.Client{Net: "udp"}
+	udpResp, err := dns.Exchange(m, udp)
+	if err != nil {
+		t.Fatalf("Expected to receive reply, but didn't: %v", err)
+	}
+	// check tcp
+	cl.Net = "tcp"
+	tcpResp, _, err := cl.Exchange(m, tcp)
+	if err != nil {
+		t.Fatalf("Expected to receive reply, but didn't: %v", err)
+	}
+
+	for _, resp := range []*dns.Msg{udpResp, tcpResp} {
+		if resp.Rcode != dns.RcodeSuccess {
+			t.Fatalf("Expected RcodeSuccess, got %v", resp.Rcode)
+		}
+		if len(resp.Extra) != 2 {
+			t.Errorf("Expected 2 RRs in additional section, got %d", len(resp.Extra))
+		}
+	}
+}