Merge pull request #29 from sapcc/timeouts

Timeouts and mutex

cmd/go-pmtud/main.go

@@ -4,4 +4,4 @@ import "github.com/sapcc/go-pmtud/internal/cmd"
 
 func main() {
 	cmd.Execute()
-}
+}

internal/arp/resolve.go

@@ -1,32 +1,48 @@
 package arp
 
 import (
+	"net"
+	"sync"
+	"time"
+
 	"github.com/go-logr/logr"
 	mdarp "github.com/mdlayher/arp"
 	"github.com/sapcc/go-pmtud/internal/config"
-	"net"
-	"time"
 )
 
+var mutex sync.Mutex
+
 type Resolver struct {
 	Log logr.Logger
 	Cfg *config.Config
 }
 
-func (r *Resolver) Resolve (ip string) (string, error) {
+func (r *Resolver) Resolve(ip string) (string, error) {
+	// avoid ARP DDoS towards single node
+	time.Sleep(time.Duration(r.Cfg.RandDelay) * time.Millisecond)
+
 	log := r.Log.WithName("arp-resolver").WithValues("ip", ip)
 	ifi, err := net.InterfaceByName(r.Cfg.ReplicationInterface)
 	if err != nil {
 		log.Error(err, "error getting interface")
 		return "", err
 	}
+
+	// Lock so only one ARP resolver runs at a time
+	mutex.Lock()
 	c, err := mdarp.Dial(ifi)
 	if err != nil {
 		log.Error(err, "error dialing")
 		return "", err
 	}
-	defer c.Close()
-	err = c.SetDeadline(time.Now().Add(1*time.Second))
+	defer func() {
+		err = c.Close()
+		if err != nil {
+			log.Error(err, "error closing arp client")
+		}
+		mutex.Unlock()
+	}()
+	err = c.SetDeadline(time.Now().Add(time.Duration(r.Cfg.ArpRequestTimeoutSeconds) * time.Second))
 	if err != nil {
 		log.Error(err, "error setting deadline")
 		return "", err
@@ -37,5 +53,6 @@ func (r *Resolver) Resolve (ip string) (string, error) {
 		log.Error(err, "error resolving mac for ip")
 		return "", err
 	}
+
 	return mac.String(), nil
 }

internal/cmd/command.go

@@ -3,7 +3,9 @@ package cmd
 import (
 	goflag "flag"
 	"fmt"
+	"math/rand"
 	"os"
+	"time"
 
 	conf "github.com/sapcc/go-pmtud/internal/config"
 	metr "github.com/sapcc/go-pmtud/internal/metrics"
@@ -49,10 +51,15 @@ func init() {
 	rootCmd.PersistentFlags().IntVar(&cfg.HealthPort, "health_port", 30041, "Port for healthz")
 	rootCmd.PersistentFlags().Uint16Var(&cfg.NfGroup, "nflog_group", 33, "NFLOG group")
 	rootCmd.PersistentFlags().IntVar(&cfg.TimeToLive, "ttl", 1, "TTL for resent packets")
+	rootCmd.PersistentFlags().IntVar(&cfg.ArpCacheTimeoutMinutes, "node-timeout-minutes", 5, "Timeout in minutes for node arp entry")
+	rootCmd.PersistentFlags().IntVar(&cfg.ArpRequestTimeoutSeconds, "arp-timeout-seconds", 1, "Timeout in seconds for node arp request")
 	rootCmd.PersistentFlags().StringVar(&cfg.KubeContext, "kube_context", "", "kube-context to use")
 	rootCmd.PersistentFlags().AddGoFlagSet(goflag.CommandLine)
 	_ = viper.BindPFlags(rootCmd.PersistentFlags())
 
+	rand.Seed(time.Now().UnixNano())
+	cfg.RandDelay = rand.Intn(1000) + 1000
+
 	metrics.Registry.MustRegister(metr.SentError, metr.Error, metr.ArpResolveError, metr.SentPacketsPeer, metr.SentPackets, metr.RecvPackets, metr.CallbackDuration)
 	cfg.PeerList = make(map[string]conf.PeerEntry)
 }

internal/config/config.go

@@ -7,22 +7,26 @@ import (
 
 type PeerEntry struct {
 	LastUpdated time.Time
-	Mac string
+	Mac         string
 }
 
 type Config struct {
 	//Peers []string
 	InterfaceNames []string
-	NodeName string
-	MetricsPort int
-	HealthPort int
-	TimeToLive int
-	NfGroup uint16
-	KubeContext string
+	NodeName       string
+	MetricsPort    int
+	HealthPort     int
+	TimeToLive     int
+	NfGroup        uint16
+	KubeContext    string
 
-	ReplicationInterface string
-	DefaultInterface string
-	InterfaceMtu int
-	PeerMutex sync.Mutex
-	PeerList map[string]PeerEntry
-}
+	ReplicationInterface     string
+	DefaultInterface         string
+	InterfaceMtu             int
+	PeerMutex                sync.Mutex
+	PeerList                 map[string]PeerEntry
+	ArpCacheTimeoutMinutes   int
+	ArpRequestTimeoutSeconds int
+
+	RandDelay int
+}

internal/nflog/pmtud.go

@@ -37,7 +37,7 @@ func (nfc *Controller) Start(startCtx context.Context) error {
 
 	nodeIface := cfg.ReplicationInterface
 	//ensure counters are reported
-	metrics.RecvPackets.WithLabelValues(cfg.NodeName).Add(0)
+	metrics.RecvPackets.WithLabelValues(cfg.NodeName, "").Add(0)
 	metrics.Error.WithLabelValues(cfg.NodeName).Add(0)
 
 	//TODO: make this a better logger

internal/node/controller.go

@@ -15,12 +15,12 @@ import (
 )
 
 type Reconciler struct {
-	Log logr.Logger
+	Log    logr.Logger
 	Client client.Client
-	Cfg *config.Config
+	Cfg    *config.Config
 }
 
-func (r *Reconciler) Reconcile (ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
+func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
 	log := r.Log.WithValues("node", request.Name)
 
 	// We do not consider our own node
@@ -31,7 +31,7 @@ func (r *Reconciler) Reconcile (ctx context.Context, request reconcile.Request)
 	// We do not want to update every mac on every update
 	e, ok := r.Cfg.PeerList[request.Name]
 	if ok {
-		if time.Now().Before(e.LastUpdated.Add(1 * time.Minute)) {
+		if time.Now().Before(e.LastUpdated.Add(time.Duration(r.Cfg.ArpCacheTimeoutMinutes) * time.Minute)) {
 			return reconcile.Result{}, nil
 		}
 	}
@@ -60,7 +60,7 @@ func (r *Reconciler) Reconcile (ctx context.Context, request reconcile.Request)
 	log.Info("found mac " + mac)
 	entry := config.PeerEntry{
 		LastUpdated: time.Now(),
-		Mac: mac,
+		Mac:         mac,
 	}
 	r.Cfg.PeerMutex.Lock()
 	r.Cfg.PeerList[request.Name] = entry

internal/util/interface.go

@@ -71,7 +71,7 @@ func GetInterfaceIp(name string, log logr.Logger) (string, error) {
 	}
 	for _, addr := range addrs {
 		var ip net.IP
-		switch v:= addr.(type) {
+		switch v := addr.(type) {
 		case *net.IPNet:
 			ip = v.IP
 		case *net.IPAddr:
@@ -91,4 +91,4 @@ func GetInterfaceIp(name string, log logr.Logger) (string, error) {
 	err = fmt.Errorf("%s is not connected to the network", name)
 	log.Error(err, "error finding interface ip")
 	return "", err
-}
+}