Fix etcd client failover when an endpoint host crashes hard

README.md

@@ -107,7 +107,7 @@ enough to connect to ETCD, read all data, and reply to this first request. This
 Example PowerDNS configuration file:
 ```
 launch=remote
-remote-connection-string=pipe:command=/path/to/pdns-etcd3[,pdns-version=3|4|5][,<config>][,prefix=<string>][,timeout=<integer>][,log-<level>=<components>]
+remote-connection-string=pipe:command=/path/to/pdns-etcd3[,pdns-version=3|4|5][,<config>][,prefix=<string>][,timeout=<integer>][,dial-keep-alive-time=<duration>][,dial-keep-alive-timeout=<duration>][,auto-sync-interval=<duration>][,permit-without-stream=<bool>][,log-<level>=<components>]
 # since in pipe mode every instance connects to ETCD and loads the data for itself (uses memory), possibly do this:
 distributor-threads=1
 ```
@@ -199,6 +199,25 @@ are tagged by *#STANDALONE*):
   `timeout=<integer>` *config file* (in milliseconds, e.g. `1500` for 1.5 seconds)<br>
   An optional parameter which sets the dial timeout to ETCD. Must be a positive value (>= 1ms).<br>
   Defaults to 2 seconds.
+* `dial-keep-alive-time=<duration>` *#STANDALONE*<br>
+  Interval at which the client sends keep-alive pings to ETCD on the underlying gRPC (HTTP/2) connection.
+  These pings allow the client to detect a dead endpoint (e.g. a host that vanished without sending a TCP RST/FIN)
+  and rotate to another endpoint instead of waiting for the kernel TCP timeout (~13–15 minutes).
+  Set to `0` to disable keep-alive pings.<br>
+  Defaults to 10 seconds.
+* `dial-keep-alive-timeout=<duration>` *#STANDALONE*<br>
+  Time the client waits for an acknowledgement after sending a keep-alive ping. If no ack arrives within this timeout,
+  the connection is considered dead and the client reconnects (to another endpoint if available).<br>
+  Defaults to 5 seconds.
+* `auto-sync-interval=<duration>` *#STANDALONE*<br>
+  Interval at which the client refreshes its view of the ETCD cluster member list.
+  This makes new endpoints (e.g. cluster members added or rotated in after the client connected) reachable
+  without restarting the backend. Set to `0` to disable.<br>
+  Defaults to 1 minute.
+* `permit-without-stream=<bool>` *#STANDALONE*<br>
+  When true, the client sends keep-alive pings even when no RPC stream is active on the connection.
+  This is needed to detect a dead endpoint while the backend is idle (e.g. between watch events on a low-traffic deployment).<br>
+  Defaults to `true`.
 * `pdns-version=3|4|5`<br>
   The (major) PowerDNS version. Version 3 and 4 have incompatible protocols with the backend, so one must use the proper one.
   Version 5 is accepted, but works currently the same as 4 (no relevant API changes yet).<br>

src/const.go

@@ -25,19 +25,27 @@ var (
 )
 
 const (
-	defaultEndpointIPv4 = "127.0.0.1:2379"
-	defaultEndpointIPv6 = "[::1]:2379"
-	defaultDialTimeout  = 2 * time.Second
-	minimumDialTimeout  = 10 * time.Millisecond
+	defaultEndpointIPv4         = "127.0.0.1:2379"
+	defaultEndpointIPv6         = "[::1]:2379"
+	defaultDialTimeout          = 2 * time.Second
+	minimumDialTimeout          = 10 * time.Millisecond
+	defaultDialKeepAliveTime    = 10 * time.Second
+	defaultDialKeepAliveTimeout = 5 * time.Second
+	defaultAutoSyncInterval     = 60 * time.Second
+	defaultPermitWithoutStream  = true
 )
 
 const (
-	pdnsVersionParam = "pdns-version"
-	prefixParam      = "prefix"
-	logParamPrefix   = "log-"
-	configFileParam  = "config-file"
-	endpointsParam   = "endpoints"
-	dialTimeoutParam = "timeout"
+	pdnsVersionParam          = "pdns-version"
+	prefixParam               = "prefix"
+	logParamPrefix            = "log-"
+	configFileParam           = "config-file"
+	endpointsParam            = "endpoints"
+	dialTimeoutParam          = "timeout"
+	dialKeepAliveTimeParam    = "dial-keep-alive-time"
+	dialKeepAliveTimeoutParam = "dial-keep-alive-timeout"
+	autoSyncIntervalParam     = "auto-sync-interval"
+	permitWithoutStreamParam  = "permit-without-stream"
 )
 
 const (

src/etcd.go

@@ -46,11 +46,19 @@ func (cli *etcdClient) Setup(args *programArgs) (logMessages []string, err error
 		return
 	}
 	cfg := clientv3.Config{
-		DialTimeout: *args.DialTimeout,
-		Endpoints:   strings.Split(*args.Endpoints, `|`),
+		DialTimeout:          *args.DialTimeout,
+		DialKeepAliveTime:    *args.DialKeepAliveTime,
+		DialKeepAliveTimeout: *args.DialKeepAliveTimeout,
+		PermitWithoutStream:  *args.PermitWithoutStream,
+		AutoSyncInterval:     *args.AutoSyncInterval,
+		Endpoints:            strings.Split(*args.Endpoints, `|`),
 	}
 	logMessages = append(logMessages,
 		fmt.Sprintf("%s: %s", dialTimeoutParam, *args.DialTimeout),
+		fmt.Sprintf("%s: %s", dialKeepAliveTimeParam, *args.DialKeepAliveTime),
+		fmt.Sprintf("%s: %s", dialKeepAliveTimeoutParam, *args.DialKeepAliveTimeout),
+		fmt.Sprintf("%s: %s", autoSyncIntervalParam, *args.AutoSyncInterval),
+		fmt.Sprintf("%s: %v", permitWithoutStreamParam, *args.PermitWithoutStream),
 		fmt.Sprintf("%s: %s", endpointsParam, *args.Endpoints),
 	)
 	cli.Client, err = clientv3.New(cfg)

src/integration_test.go

@@ -52,7 +52,7 @@
 	}
 }
 
 func putT(t *testing.T, prefix, key, value string) int64 {
 	t.Helper()
 	if resp, err := cli.Put(prefix+key, value, 10*time.Second); err != nil {
 		Fatalf(t, "failed to put %q: %s", prefix+key, err)
@@ -95,12 +95,20 @@
 	defer closeNoError(outR) // this should be done automatically by pdns-etcd3, but just in case
 	config := ""
 	timeout, _ := time.ParseDuration("5s")
+	keepAliveTime := defaultDialKeepAliveTime
+	keepAliveTimeout := defaultDialKeepAliveTimeout
+	autoSync := defaultAutoSyncInterval
+	permitWithoutStream := defaultPermitWithoutStream
 	prefix := ""
 	args = programArgs{
-		ConfigFile:  &config,
-		Endpoints:   &etcd.Endpoint,
-		DialTimeout: &timeout,
-		Prefix:      &prefix,
+		ConfigFile:           &config,
+		Endpoints:            &etcd.Endpoint,
+		DialTimeout:          &timeout,
+		DialKeepAliveTime:    &keepAliveTime,
+		DialKeepAliveTimeout: &keepAliveTimeout,
+		AutoSyncInterval:     &autoSync,
+		PermitWithoutStream:  &permitWithoutStream,
+		Prefix:               &prefix,
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

src/pdns-etcd3.go

@@ -33,14 +33,21 @@ import (
 )
 
 type programArgs struct {
-	ConfigFile  *string
-	Endpoints   *string
-	DialTimeout *time.Duration
-	Prefix      *string
+	ConfigFile           *string
+	Endpoints            *string
+	DialTimeout          *time.Duration
+	DialKeepAliveTime    *time.Duration
+	DialKeepAliveTimeout *time.Duration
+	AutoSyncInterval     *time.Duration
+	PermitWithoutStream  *bool
+	Prefix               *string
 }
 
 func (pa programArgs) String() string {
-	return fmt.Sprintf("ConfigFile=%s, Endpoints=%s, DialTimeout=%s, Prefix=%s", val2str(pa.ConfigFile), val2str(pa.Endpoints), val2str(pa.DialTimeout), val2str(pa.Prefix))
+	return fmt.Sprintf("ConfigFile=%s, Endpoints=%s, DialTimeout=%s, DialKeepAliveTime=%s, DialKeepAliveTimeout=%s, AutoSyncInterval=%s, PermitWithoutStream=%s, Prefix=%s",
+		val2str(pa.ConfigFile), val2str(pa.Endpoints), val2str(pa.DialTimeout),
+		val2str(pa.DialKeepAliveTime), val2str(pa.DialKeepAliveTimeout), val2str(pa.AutoSyncInterval),
+		val2str(pa.PermitWithoutStream), val2str(pa.Prefix))
 }
 
 type statusType struct {
@@ -124,6 +131,14 @@ func readParameters(params objectType[string], client *pdnsClient) error {
 		case !standalone && k == dialTimeoutParam:
 			mdt := minimumDialTimeout
 			err = setDurationParameterFunc(args.DialTimeout, &mdt)(v)
+		case !standalone && k == dialKeepAliveTimeParam:
+			err = setDurationParameterFunc(args.DialKeepAliveTime, nil)(v)
+		case !standalone && k == dialKeepAliveTimeoutParam:
+			err = setDurationParameterFunc(args.DialKeepAliveTimeout, nil)(v)
+		case !standalone && k == autoSyncIntervalParam:
+			err = setDurationParameterFunc(args.AutoSyncInterval, nil)(v)
+		case !standalone && k == permitWithoutStreamParam:
+			err = setBooleanParameterFunc(args.PermitWithoutStream)(v)
 		case !standalone && k == prefixParam:
 			*args.Prefix = v
 		case k == pdnsVersionParam:
@@ -300,13 +315,17 @@ func Main(programVersion VersionType, gitVersion string) {
 }
 
 var (
-	standaloneArg  = flag.String("standalone", "", `Use a standalone mode determined by the given URL (unix:///path/to/socket[?relative=<bool>] or http://<listen-address>:<listen-port>)`)
-	configFileArg  = flag.String(configFileParam, "", "Use the given configuration file for the ETCD connection (overrides -endpoints)")
-	endpointsArg   = flag.String(endpointsParam, defaultEndpointIPv6+"|"+defaultEndpointIPv4, "Use the endpoints configuration for ETCD connection")
-	dialTimeoutArg = flag.Duration(dialTimeoutParam, defaultDialTimeout, "ETCD dial timeout")
-	prefixArg      = flag.String(prefixParam, "", "Global key prefix")
-	pdnsVersionArg = flag.String(pdnsVersionParam, "", "default PDNS version")
-	loggingArgs    = func() map[logrus.Level]*string {
+	standaloneArg           = flag.String("standalone", "", `Use a standalone mode determined by the given URL (unix:///path/to/socket[?relative=<bool>] or http://<listen-address>:<listen-port>)`)
+	configFileArg           = flag.String(configFileParam, "", "Use the given configuration file for the ETCD connection (overrides -endpoints)")
+	endpointsArg            = flag.String(endpointsParam, defaultEndpointIPv6+"|"+defaultEndpointIPv4, "Use the endpoints configuration for ETCD connection")
+	dialTimeoutArg          = flag.Duration(dialTimeoutParam, defaultDialTimeout, "ETCD dial timeout")
+	dialKeepAliveTimeArg    = flag.Duration(dialKeepAliveTimeParam, defaultDialKeepAliveTime, "ETCD dial keep-alive ping interval (0 to disable)")
+	dialKeepAliveTimeoutArg = flag.Duration(dialKeepAliveTimeoutParam, defaultDialKeepAliveTimeout, "ETCD dial keep-alive ping timeout")
+	autoSyncIntervalArg     = flag.Duration(autoSyncIntervalParam, defaultAutoSyncInterval, "ETCD member list auto-sync interval (0 to disable)")
+	permitWithoutStreamArg  = flag.Bool(permitWithoutStreamParam, defaultPermitWithoutStream, "send ETCD client keep-alive pings even with no active RPC stream")
+	prefixArg               = flag.String(prefixParam, "", "Global key prefix")
+	pdnsVersionArg          = flag.String(pdnsVersionParam, "", "default PDNS version")
+	loggingArgs             = func() map[logrus.Level]*string {
 		args := map[logrus.Level]*string{}
 		for _, level := range logrus.AllLevels {
 			args[level] = flag.String(logParamPrefix+level.String(), "", fmt.Sprintf("Set logging level %s to the given components (separated by +)", level))
@@ -327,10 +346,14 @@ func main(programVersion VersionType, gitVersion string, cmdLineArgs []string, o
 	log.main().Printf("pdns-etcd3 %s, Copyright © 2016-2026 nix <https://keybase.io/nixn>", releaseVersion)
 	// handle arguments // TODO handle more arguments, f.e. 'show-defaults' standalone command
 	args = programArgs{
-		ConfigFile:  configFileArg,
-		Endpoints:   endpointsArg,
-		DialTimeout: dialTimeoutArg,
-		Prefix:      prefixArg,
+		ConfigFile:           configFileArg,
+		Endpoints:            endpointsArg,
+		DialTimeout:          dialTimeoutArg,
+		DialKeepAliveTime:    dialKeepAliveTimeArg,
+		DialKeepAliveTimeout: dialKeepAliveTimeoutArg,
+		AutoSyncInterval:     autoSyncIntervalArg,
+		PermitWithoutStream:  permitWithoutStreamArg,
+		Prefix:               prefixArg,
 	}
 	if err := flag.CommandLine.Parse(cmdLineArgs); err != nil { // same as flag.Parse(), but we can pass the arguments instead of being fixed to os.Args[1:] (needed for integration testing)
 		log.main().Panicf("failed to parse command line arguments: %s", err)