Merge pull request #128390 from aroradaman/netlink-bump

bump: github.com/vishvananda/netlink

go.mod

@@ -60,7 +60,7 @@ require (
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.9.0
-	github.com/vishvananda/netlink v1.3.0
+	github.com/vishvananda/netlink v1.3.1-0.20240905180732-b1ce50cfa9be
 	github.com/vishvananda/netns v0.0.4
 	go.etcd.io/etcd/api/v3 v3.5.16
 	go.etcd.io/etcd/client/pkg/v3 v3.5.16

go.sum

@@ -497,8 +497,8 @@ github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG
 github.com/tmc/grpc-websocket-proxy v0.0.0-20220101234140-673ab2c3ae75 h1:6fotK7otjonDflCTK0BCfls4SPy3NcCVb5dqqmbRknE=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20220101234140-673ab2c3ae75/go.mod h1:KO6IkyS8Y3j8OdNO85qEYBsRPuteD+YciPomcXdrMnk=
 github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
-github.com/vishvananda/netlink v1.3.0 h1:X7l42GfcV4S6E4vHTsw48qbrV+9PVojNfIhZcwQdrZk=
-github.com/vishvananda/netlink v1.3.0/go.mod h1:i6NetklAujEcC6fK0JPjT8qSwWyO0HLn4UKG+hGqeJs=
+github.com/vishvananda/netlink v1.3.1-0.20240905180732-b1ce50cfa9be h1:xdCMvyhnKzaepIUgVpUmTJo/+H1AQ7HuFYn1hv7/Neo=
+github.com/vishvananda/netlink v1.3.1-0.20240905180732-b1ce50cfa9be/go.mod h1:i6NetklAujEcC6fK0JPjT8qSwWyO0HLn4UKG+hGqeJs=
 github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8=
 github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=

vendor/github.com/vishvananda/netlink/conntrack_linux.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"strings"
 	"time"
 
 	"github.com/vishvananda/netlink/nl"
@@ -70,7 +71,7 @@ func ConntrackUpdate(table ConntrackTableType, family InetFamily, flow *Conntrac
 // ConntrackDeleteFilter deletes entries on the specified table on the base of the filter
 // conntrack -D [table] parameters         Delete conntrack or expectation
 //
-// Deprecated: use [ConntrackDeleteFilter] instead.
+// Deprecated: use [ConntrackDeleteFilters] instead.
 func ConntrackDeleteFilter(table ConntrackTableType, family InetFamily, filter CustomConntrackFilter) (uint, error) {
 	return pkgHandle.ConntrackDeleteFilters(table, family, filter)
 }
@@ -158,6 +159,7 @@ func (h *Handle) ConntrackDeleteFilters(table ConntrackTableType, family InetFam
 	}
 
 	var matched uint
+	var errMsgs []string
 	for _, dataRaw := range res {
 		flow := parseRawData(dataRaw)
 		for _, filter := range filters {
@@ -165,14 +167,18 @@ func (h *Handle) ConntrackDeleteFilters(table ConntrackTableType, family InetFam
 				req2 := h.newConntrackRequest(table, family, nl.IPCTNL_MSG_CT_DELETE, unix.NLM_F_ACK)
 				// skip the first 4 byte that are the netfilter header, the newConntrackRequest is adding it already
 				req2.AddRawData(dataRaw[4:])
-				req2.Execute(unix.NETLINK_NETFILTER, 0)
-				matched++
-				// flow is already deleted, no need to match on other filters and continue to the next flow.
-				break
+				if _, err = req2.Execute(unix.NETLINK_NETFILTER, 0); err == nil {
+					matched++
+					// flow is already deleted, no need to match on other filters and continue to the next flow.
+					break
+				}
+				errMsgs = append(errMsgs, fmt.Sprintf("failed to delete conntrack flow '%s': %s", flow.String(), err.Error()))
 			}
 		}
 	}
-
+	if len(errMsgs) > 0 {
+		return matched, fmt.Errorf(strings.Join(errMsgs, "; "))
+	}
 	return matched, nil
 }
 

vendor/github.com/vishvananda/netlink/conntrack_unspecified.go

@@ -33,7 +33,7 @@ func ConntrackTableFlush(table ConntrackTableType) error {
 // ConntrackDeleteFilter deletes entries on the specified table on the base of the filter
 // conntrack -D [table] parameters         Delete conntrack or expectation
 //
-// Deprecated: use [ConntrackDeleteFilter] instead.
+// Deprecated: use [ConntrackDeleteFilters] instead.
 func ConntrackDeleteFilter(table ConntrackTableType, family InetFamily, filter *ConntrackFilter) (uint, error) {
 	return 0, ErrNotImplemented
 }

vendor/github.com/vishvananda/netlink/filter_linux.go

@@ -920,9 +920,11 @@ func parseActions(tables []syscall.NetlinkRouteAttr) ([]Action, error) {
 				actionnStatistic = (*ActionStatistic)(s)
 			}
 		}
-		action.Attrs().Statistics = actionnStatistic
-		action.Attrs().Timestamp = actionTimestamp
-		actions = append(actions, action)
+		if action != nil {
+			action.Attrs().Statistics = actionnStatistic
+			action.Attrs().Timestamp = actionTimestamp
+			actions = append(actions, action)
+		}
 	}
 	return actions, nil
 }

vendor/github.com/vishvananda/netlink/nl/nl_linux.go

@@ -4,6 +4,7 @@ package nl
 import (
 	"bytes"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"net"
 	"os"
@@ -11,6 +12,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"syscall"
+	"time"
 	"unsafe"
 
 	"github.com/vishvananda/netns"
@@ -656,9 +658,11 @@ func NewNetlinkRequest(proto, flags int) *NetlinkRequest {
 }
 
 type NetlinkSocket struct {
-	fd   int32
-	file *os.File
-	lsa  unix.SockaddrNetlink
+	fd             int32
+	file           *os.File
+	lsa            unix.SockaddrNetlink
+	sendTimeout    int64 // Access using atomic.Load/StoreInt64
+	receiveTimeout int64 // Access using atomic.Load/StoreInt64
 	sync.Mutex
 }
 
@@ -802,8 +806,44 @@ func (s *NetlinkSocket) GetFd() int {
 	return int(s.fd)
 }
 
+func (s *NetlinkSocket) GetTimeouts() (send, receive time.Duration) {
+	return time.Duration(atomic.LoadInt64(&s.sendTimeout)),
+		time.Duration(atomic.LoadInt64(&s.receiveTimeout))
+}
+
 func (s *NetlinkSocket) Send(request *NetlinkRequest) error {
-	return unix.Sendto(int(s.fd), request.Serialize(), 0, &s.lsa)
+	rawConn, err := s.file.SyscallConn()
+	if err != nil {
+		return err
+	}
+	var (
+		deadline time.Time
+		innerErr error
+	)
+	sendTimeout := atomic.LoadInt64(&s.sendTimeout)
+	if sendTimeout != 0 {
+		deadline = time.Now().Add(time.Duration(sendTimeout))
+	}
+	if err := s.file.SetWriteDeadline(deadline); err != nil {
+		return err
+	}
+	serializedReq := request.Serialize()
+	err = rawConn.Write(func(fd uintptr) (done bool) {
+		innerErr = unix.Sendto(int(s.fd), serializedReq, 0, &s.lsa)
+		return innerErr != unix.EWOULDBLOCK
+	})
+	if innerErr != nil {
+		return innerErr
+	}
+	if err != nil {
+		// The timeout was previously implemented using SO_SNDTIMEO on a blocking
+		// socket. So, continue to return EAGAIN when the timeout is reached.
+		if errors.Is(err, os.ErrDeadlineExceeded) {
+			return unix.EAGAIN
+		}
+		return err
+	}
+	return nil
 }
 
 func (s *NetlinkSocket) Receive() ([]syscall.NetlinkMessage, *unix.SockaddrNetlink, error) {
@@ -812,20 +852,33 @@ func (s *NetlinkSocket) Receive() ([]syscall.NetlinkMessage, *unix.SockaddrNetli
 		return nil, nil, err
 	}
 	var (
+		deadline time.Time
 		fromAddr *unix.SockaddrNetlink
 		rb       [RECEIVE_BUFFER_SIZE]byte
 		nr       int
 		from     unix.Sockaddr
 		innerErr error
 	)
+	receiveTimeout := atomic.LoadInt64(&s.receiveTimeout)
+	if receiveTimeout != 0 {
+		deadline = time.Now().Add(time.Duration(receiveTimeout))
+	}
+	if err := s.file.SetReadDeadline(deadline); err != nil {
+		return nil, nil, err
+	}
 	err = rawConn.Read(func(fd uintptr) (done bool) {
 		nr, from, innerErr = unix.Recvfrom(int(fd), rb[:], 0)
 		return innerErr != unix.EWOULDBLOCK
 	})
 	if innerErr != nil {
-		err = innerErr
+		return nil, nil, innerErr
 	}
 	if err != nil {
+		// The timeout was previously implemented using SO_RCVTIMEO on a blocking
+		// socket. So, continue to return EAGAIN when the timeout is reached.
+		if errors.Is(err, os.ErrDeadlineExceeded) {
+			return nil, nil, unix.EAGAIN
+		}
 		return nil, nil, err
 	}
 	fromAddr, ok := from.(*unix.SockaddrNetlink)
@@ -847,16 +900,14 @@ func (s *NetlinkSocket) Receive() ([]syscall.NetlinkMessage, *unix.SockaddrNetli
 
 // SetSendTimeout allows to set a send timeout on the socket
 func (s *NetlinkSocket) SetSendTimeout(timeout *unix.Timeval) error {
-	// Set a send timeout of SOCKET_SEND_TIMEOUT, this will allow the Send to periodically unblock and avoid that a routine
-	// remains stuck on a send on a closed fd
-	return unix.SetsockoptTimeval(int(s.fd), unix.SOL_SOCKET, unix.SO_SNDTIMEO, timeout)
+	atomic.StoreInt64(&s.sendTimeout, timeout.Nano())
+	return nil
 }
 
 // SetReceiveTimeout allows to set a receive timeout on the socket
 func (s *NetlinkSocket) SetReceiveTimeout(timeout *unix.Timeval) error {
-	// Set a read timeout of SOCKET_READ_TIMEOUT, this will allow the Read to periodically unblock and avoid that a routine
-	// remains stuck on a recvmsg on a closed fd
-	return unix.SetsockoptTimeval(int(s.fd), unix.SOL_SOCKET, unix.SO_RCVTIMEO, timeout)
+	atomic.StoreInt64(&s.receiveTimeout, timeout.Nano())
+	return nil
 }
 
 // SetReceiveBufferSize allows to set a receive buffer size on the socket

vendor/modules.txt

@@ -597,7 +597,7 @@ github.com/syndtr/gocapability/capability
 # github.com/tmc/grpc-websocket-proxy v0.0.0-20220101234140-673ab2c3ae75
 ## explicit; go 1.15
 github.com/tmc/grpc-websocket-proxy/wsproxy
-# github.com/vishvananda/netlink v1.3.0
+# github.com/vishvananda/netlink v1.3.1-0.20240905180732-b1ce50cfa9be
 ## explicit; go 1.12
 github.com/vishvananda/netlink
 github.com/vishvananda/netlink/nl