mirror of https://github.com/jetkvm/kvm.git
Restructure to ensure we're releasing locks
We were holding locks too long and doing more structure copying than optimal.
This commit is contained in:
parent
af8bfc702c
commit
b6c1e49803
11
config.go
11
config.go
|
|
@ -7,6 +7,7 @@ import (
|
||||||
"strconv"
|
"strconv"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
|
"github.com/guregu/null/v6"
|
||||||
"github.com/jetkvm/kvm/internal/confparser"
|
"github.com/jetkvm/kvm/internal/confparser"
|
||||||
"github.com/jetkvm/kvm/internal/logging"
|
"github.com/jetkvm/kvm/internal/logging"
|
||||||
"github.com/jetkvm/kvm/internal/network/types"
|
"github.com/jetkvm/kvm/internal/network/types"
|
||||||
|
|
@ -254,6 +255,16 @@ func LoadConfig() {
|
||||||
loadedConfig.KeyboardLayout = "en-US"
|
loadedConfig.KeyboardLayout = "en-US"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// fixup old lldp modes enabled and all are now rx_and_tx
|
||||||
|
if loadedConfig.NetworkConfig.LLDPMode.String == "enabled" || loadedConfig.NetworkConfig.LLDPMode.String == "all" {
|
||||||
|
loadedConfig.NetworkConfig.LLDPMode = null.StringFrom("rx_and_tx")
|
||||||
|
}
|
||||||
|
|
||||||
|
// fixup old lldp mode basic is now rx_only
|
||||||
|
if loadedConfig.NetworkConfig.LLDPMode.String == "basic" {
|
||||||
|
loadedConfig.NetworkConfig.LLDPMode = null.StringFrom("rx_only")
|
||||||
|
}
|
||||||
|
|
||||||
config = &loadedConfig
|
config = &loadedConfig
|
||||||
|
|
||||||
logging.GetRootLogger().UpdateLogLevel(config.DefaultLogLevel)
|
logging.GetRootLogger().UpdateLogLevel(config.DefaultLogLevel)
|
||||||
|
|
|
||||||
|
|
@ -39,7 +39,7 @@ type testNetworkConfig struct {
|
||||||
IPv6Mode null.String `json:"ipv6_mode" one_of:"slaac,dhcpv6,slaac_and_dhcpv6,static,link_local,disabled" default:"slaac"`
|
IPv6Mode null.String `json:"ipv6_mode" one_of:"slaac,dhcpv6,slaac_and_dhcpv6,static,link_local,disabled" default:"slaac"`
|
||||||
IPv6Static *testIPv6StaticConfig `json:"ipv6_static,omitempty" required_if:"IPv6Mode=static"`
|
IPv6Static *testIPv6StaticConfig `json:"ipv6_static,omitempty" required_if:"IPv6Mode=static"`
|
||||||
|
|
||||||
LLDPMode null.String `json:"lldp_mode,omitempty" one_of:"disabled,basic,all" default:"basic"`
|
LLDPMode null.String `json:"lldp_mode,omitempty" one_of:"disabled,enabled,rx_only,tx_only,rx_and_tx,basic,all" default:"rx_and_tx"`
|
||||||
LLDPTxTLVs []string `json:"lldp_tx_tlvs,omitempty" one_of:"chassis,port,system,vlan" default:"chassis,port,system,vlan"`
|
LLDPTxTLVs []string `json:"lldp_tx_tlvs,omitempty" one_of:"chassis,port,system,vlan" default:"chassis,port,system,vlan"`
|
||||||
MDNSMode null.String `json:"mdns_mode,omitempty" one_of:"disabled,auto,ipv4_only,ipv6_only" default:"auto"`
|
MDNSMode null.String `json:"mdns_mode,omitempty" one_of:"disabled,auto,ipv4_only,ipv6_only" default:"auto"`
|
||||||
TimeSyncMode null.String `json:"time_sync_mode,omitempty" one_of:"ntp_only,ntp_and_http,http_only,custom" default:"ntp_and_http"`
|
TimeSyncMode null.String `json:"time_sync_mode,omitempty" one_of:"ntp_only,ntp_and_http,http_only,custom" default:"ntp_and_http"`
|
||||||
|
|
|
||||||
|
|
@ -14,16 +14,18 @@ import (
|
||||||
|
|
||||||
var defaultLogger = logging.GetSubsystemLogger("lldp")
|
var defaultLogger = logging.GetSubsystemLogger("lldp")
|
||||||
|
|
||||||
|
type RunningState struct {
|
||||||
|
Enabled bool
|
||||||
|
Running bool
|
||||||
|
Logger *zerolog.Logger
|
||||||
|
TPacket *afpacket.TPacket
|
||||||
|
Ctx context.Context
|
||||||
|
Cancel context.CancelFunc
|
||||||
|
}
|
||||||
|
|
||||||
type LLDP struct {
|
type LLDP struct {
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
|
|
||||||
l *zerolog.Logger
|
l *zerolog.Logger
|
||||||
tPacketRx *afpacket.TPacket
|
|
||||||
tPacketTx *afpacket.TPacket
|
|
||||||
pktSourceRx *gopacket.PacketSource
|
|
||||||
|
|
||||||
enableRx bool
|
|
||||||
enableTx bool
|
|
||||||
|
|
||||||
interfaceName string
|
interfaceName string
|
||||||
advertiseOptions *AdvertiseOptions
|
advertiseOptions *AdvertiseOptions
|
||||||
|
|
@ -32,15 +34,11 @@ type LLDP struct {
|
||||||
neighbors map[neighborCacheKey]Neighbor
|
neighbors map[neighborCacheKey]Neighbor
|
||||||
neighborsMu sync.RWMutex
|
neighborsMu sync.RWMutex
|
||||||
|
|
||||||
// State tracking
|
Tx *RunningState
|
||||||
txRunning bool
|
Rx *RunningState
|
||||||
txCtx context.Context
|
|
||||||
txCancel context.CancelFunc
|
|
||||||
|
|
||||||
rxRunning bool
|
pktSourceRx *gopacket.PacketSource
|
||||||
rxWaitGroup *sync.WaitGroup
|
rxWaitGroup *sync.WaitGroup
|
||||||
rxCtx context.Context
|
|
||||||
rxCancel context.CancelFunc
|
|
||||||
}
|
}
|
||||||
|
|
||||||
type AdvertiseOptions struct {
|
type AdvertiseOptions struct {
|
||||||
|
|
@ -74,8 +72,12 @@ func NewLLDP(opts *Options) *LLDP {
|
||||||
return &LLDP{
|
return &LLDP{
|
||||||
interfaceName: opts.InterfaceName,
|
interfaceName: opts.InterfaceName,
|
||||||
advertiseOptions: opts.AdvertiseOptions,
|
advertiseOptions: opts.AdvertiseOptions,
|
||||||
enableRx: opts.EnableRx,
|
Rx: &RunningState{
|
||||||
enableTx: opts.EnableTx,
|
Enabled: opts.EnableRx,
|
||||||
|
},
|
||||||
|
Tx: &RunningState{
|
||||||
|
Enabled: opts.EnableTx,
|
||||||
|
},
|
||||||
rxWaitGroup: &sync.WaitGroup{},
|
rxWaitGroup: &sync.WaitGroup{},
|
||||||
l: opts.Logger,
|
l: opts.Logger,
|
||||||
neighbors: make(map[neighborCacheKey]Neighbor),
|
neighbors: make(map[neighborCacheKey]Neighbor),
|
||||||
|
|
@ -84,14 +86,18 @@ func NewLLDP(opts *Options) *LLDP {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *LLDP) Start() error {
|
func (l *LLDP) Start() error {
|
||||||
if l.enableRx {
|
l.mu.RLock()
|
||||||
|
rxEnabled, txEnabled := l.Rx.Enabled, l.Tx.Enabled
|
||||||
|
l.mu.RUnlock()
|
||||||
|
|
||||||
|
if rxEnabled {
|
||||||
if err := l.startRx(); err != nil {
|
if err := l.startRx(); err != nil {
|
||||||
return fmt.Errorf("failed to start RX: %w", err)
|
return fmt.Errorf("failed to start RX: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start TX if enabled
|
// Start TX if enabled
|
||||||
if l.enableTx {
|
if txEnabled {
|
||||||
if err := l.startTx(); err != nil {
|
if err := l.startTx(); err != nil {
|
||||||
return fmt.Errorf("failed to start TX: %w", err)
|
return fmt.Errorf("failed to start TX: %w", err)
|
||||||
}
|
}
|
||||||
|
|
@ -103,25 +109,22 @@ func (l *LLDP) Start() error {
|
||||||
// StartRx starts the LLDP receiver if not already running
|
// StartRx starts the LLDP receiver if not already running
|
||||||
func (l *LLDP) startRx() error {
|
func (l *LLDP) startRx() error {
|
||||||
l.mu.RLock()
|
l.mu.RLock()
|
||||||
running := l.rxRunning
|
running := l.Rx.Running
|
||||||
enabled := l.enableRx
|
enabled := l.Rx.Enabled
|
||||||
l.mu.RUnlock()
|
l.mu.RUnlock()
|
||||||
|
|
||||||
if running || !enabled {
|
if running || !enabled {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := l.setUpCapture(); err != nil {
|
|
||||||
return fmt.Errorf("failed to set up capture: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return l.startCapture()
|
return l.startCapture()
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetAdvertiseOptions updates the advertise options and resends LLDP packets if TX is running
|
// SetAdvertiseOptions updates the advertise options and resends LLDP packets if TX is running
|
||||||
func (l *LLDP) SetAdvertiseOptions(opts *AdvertiseOptions) error {
|
func (l *LLDP) SetAdvertiseOptions(opts *AdvertiseOptions) error {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
txRunning := l.txRunning
|
logger := l.l
|
||||||
|
txRunning := l.Tx.Running
|
||||||
l.advertiseOptions = opts
|
l.advertiseOptions = opts
|
||||||
l.mu.Unlock()
|
l.mu.Unlock()
|
||||||
|
|
||||||
|
|
@ -130,7 +133,7 @@ func (l *LLDP) SetAdvertiseOptions(opts *AdvertiseOptions) error {
|
||||||
if err := l.sendTxPackets(); err != nil {
|
if err := l.sendTxPackets(); err != nil {
|
||||||
return fmt.Errorf("failed to resend LLDP packet with new options: %w", err)
|
return fmt.Errorf("failed to resend LLDP packet with new options: %w", err)
|
||||||
}
|
}
|
||||||
l.l.Info().Msg("advertise options changed, resent LLDP packet")
|
logger.Info().Msg("advertise options changed, resent LLDP packet")
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|
@ -138,8 +141,8 @@ func (l *LLDP) SetAdvertiseOptions(opts *AdvertiseOptions) error {
|
||||||
|
|
||||||
func (l *LLDP) SetRxAndTx(rx, tx bool) error {
|
func (l *LLDP) SetRxAndTx(rx, tx bool) error {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
l.enableRx = rx
|
l.Rx.Enabled = rx
|
||||||
l.enableTx = tx
|
l.Tx.Enabled = tx
|
||||||
l.mu.Unlock()
|
l.mu.Unlock()
|
||||||
|
|
||||||
// if rx is enabled, start the RX
|
// if rx is enabled, start the RX
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@ package lldp
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net"
|
"net"
|
||||||
|
|
@ -10,6 +11,7 @@ import (
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/google/gopacket"
|
"github.com/google/gopacket"
|
||||||
|
"github.com/google/gopacket/afpacket"
|
||||||
"github.com/google/gopacket/layers"
|
"github.com/google/gopacket/layers"
|
||||||
"github.com/rs/zerolog"
|
"github.com/rs/zerolog"
|
||||||
)
|
)
|
||||||
|
|
@ -30,21 +32,7 @@ var multicastAddrs = []string{
|
||||||
"01:00:0C:CC:CC:CC",
|
"01:00:0C:CC:CC:CC",
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *LLDP) setUpCapture() error {
|
func (l *LLDP) setUpPacketSourceUnderLock(tPacket *afpacket.TPacket, logger *zerolog.Logger) (*gopacket.PacketSource, error) {
|
||||||
l.mu.Lock()
|
|
||||||
defer l.mu.Unlock()
|
|
||||||
|
|
||||||
if l.tPacketRx != nil {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
logger := l.l.With().Str("interface", l.interfaceName).Logger()
|
|
||||||
tPacketRx, err := afPacketNewTPacket(l.interfaceName)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
logger.Info().Msg("created TPacketRx")
|
|
||||||
|
|
||||||
// set up multicast addresses
|
// set up multicast addresses
|
||||||
// otherwise the kernel might discard the packets
|
// otherwise the kernel might discard the packets
|
||||||
// another workaround would be to enable promiscuous mode but that's too tricky
|
// another workaround would be to enable promiscuous mode but that's too tricky
|
||||||
|
|
@ -72,23 +60,30 @@ func (l *LLDP) setUpCapture() error {
|
||||||
Msg("added multicast address")
|
Msg("added multicast address")
|
||||||
}
|
}
|
||||||
|
|
||||||
if err = tPacketRx.SetBPF(bpfFilter); err != nil {
|
if err := tPacket.SetBPF(bpfFilter); err != nil {
|
||||||
logger.Error().
|
logger.Error().
|
||||||
Err(err).
|
Err(err).
|
||||||
Msg("unable to set BPF filter")
|
Msg("unable to set BPF filter")
|
||||||
tPacketRx.Close()
|
return nil, err
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Info().Msg("BPF filter set")
|
logger.Info().Msg("BPF filter set")
|
||||||
|
return gopacket.NewPacketSource(tPacket, layers.LayerTypeEthernet), nil
|
||||||
l.pktSourceRx = gopacket.NewPacketSource(tPacketRx, layers.LayerTypeEthernet)
|
|
||||||
l.tPacketRx = tPacketRx
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *LLDP) doCapture(logger *zerolog.Logger) {
|
func (l *LLDP) doCapture(logger *zerolog.Logger) {
|
||||||
if l.pktSourceRx == nil || l.rxCtx == nil {
|
l.mu.Lock()
|
||||||
|
l.Rx.Running = true
|
||||||
|
ctx := l.Rx.Ctx
|
||||||
|
l.mu.Unlock()
|
||||||
|
|
||||||
|
defer func() {
|
||||||
|
l.mu.Lock()
|
||||||
|
l.Rx.Running = false
|
||||||
|
l.mu.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
|
if l.pktSourceRx == nil || ctx == nil {
|
||||||
logger.Error().Msg("packet source or RX context not initialized")
|
logger.Error().Msg("packet source or RX context not initialized")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
@ -101,7 +96,7 @@ func (l *LLDP) doCapture(logger *zerolog.Logger) {
|
||||||
for {
|
for {
|
||||||
// check if the context is done before blocking call
|
// check if the context is done before blocking call
|
||||||
select {
|
select {
|
||||||
case <-l.rxCtx.Done():
|
case <-ctx.Done():
|
||||||
logger.Info().Msg("RX context cancelled")
|
logger.Info().Msg("RX context cancelled")
|
||||||
return
|
return
|
||||||
default:
|
default:
|
||||||
|
|
@ -109,7 +104,6 @@ func (l *LLDP) doCapture(logger *zerolog.Logger) {
|
||||||
|
|
||||||
logger.Trace().Msg("waiting for next packet")
|
logger.Trace().Msg("waiting for next packet")
|
||||||
packet, err := l.pktSourceRx.NextPacket()
|
packet, err := l.pktSourceRx.NextPacket()
|
||||||
logger.Trace().Interface("packet", packet).Err(err).Msg("got next packet")
|
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Error().
|
logger.Error().
|
||||||
|
|
@ -117,16 +111,22 @@ func (l *LLDP) doCapture(logger *zerolog.Logger) {
|
||||||
Msg("error getting next packet")
|
Msg("error getting next packet")
|
||||||
|
|
||||||
// Immediately break for known unrecoverable errors
|
// Immediately break for known unrecoverable errors
|
||||||
if err == io.EOF || err == io.ErrUnexpectedEOF ||
|
if errors.Is(err, io.EOF) ||
|
||||||
err == io.ErrNoProgress || err == io.ErrClosedPipe || err == io.ErrShortBuffer ||
|
errors.Is(err, io.ErrUnexpectedEOF) ||
|
||||||
err == syscall.EBADF ||
|
errors.Is(err, io.ErrNoProgress) ||
|
||||||
|
errors.Is(err, io.ErrClosedPipe) ||
|
||||||
|
errors.Is(err, io.ErrShortBuffer) ||
|
||||||
|
errors.Is(err, syscall.EBADF) ||
|
||||||
strings.Contains(err.Error(), "use of closed file") {
|
strings.Contains(err.Error(), "use of closed file") {
|
||||||
|
logger.Error().Msg("unrecoverable error, stopping capture")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
logger.Trace().Interface("packet", packet).Msg("got next packet")
|
||||||
|
|
||||||
if err := l.handlePacket(packet, logger); err != nil {
|
if err := l.handlePacket(packet, logger); err != nil {
|
||||||
logger.Error().
|
logger.Error().
|
||||||
Err(err).
|
Err(err).
|
||||||
|
|
@ -136,30 +136,52 @@ func (l *LLDP) doCapture(logger *zerolog.Logger) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *LLDP) startCapture() error {
|
func (l *LLDP) prepareCapture() (bool, error) {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
defer l.mu.Unlock()
|
defer l.mu.Unlock()
|
||||||
|
|
||||||
if l.rxRunning {
|
if l.Rx.Running {
|
||||||
return nil // Already running
|
l.l.Debug().Msg("LLDP receiver already running")
|
||||||
|
return false, nil // Already running
|
||||||
}
|
}
|
||||||
|
|
||||||
if l.tPacketRx == nil {
|
logger := l.l.With().Str("interface", l.interfaceName).Str("direction", "rx").Logger()
|
||||||
return fmt.Errorf("AFPacket not initialized")
|
|
||||||
}
|
|
||||||
|
|
||||||
if l.pktSourceRx == nil {
|
|
||||||
return fmt.Errorf("packet source not initialized")
|
|
||||||
}
|
|
||||||
|
|
||||||
logger := l.l.With().Str("interface", l.interfaceName).Logger()
|
|
||||||
logger.Info().Msg("starting capture LLDP ethernet frames")
|
logger.Info().Msg("starting capture LLDP ethernet frames")
|
||||||
|
|
||||||
// Create a new context for this instance
|
tPacket, err := afPacketNewTPacket(l.interfaceName)
|
||||||
l.rxCtx, l.rxCancel = context.WithCancel(context.Background())
|
if err != nil {
|
||||||
l.rxRunning = true
|
logger.Error().Err(err).Msg("could not create TPacket instance for receiving LLDP packets")
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
|
||||||
go l.doCapture(&logger)
|
l.pktSourceRx, err = l.setUpPacketSourceUnderLock(tPacket, &logger)
|
||||||
|
if err != nil {
|
||||||
|
logger.Error().Err(err).Msg("could not create packet source for receiving LLDP packets")
|
||||||
|
tPacket.Close()
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
|
||||||
|
runningState := &RunningState{Enabled: true, Logger: &logger}
|
||||||
|
runningState.Ctx, runningState.Cancel = context.WithCancel(context.Background())
|
||||||
|
runningState.TPacket = tPacket
|
||||||
|
l.Rx = runningState
|
||||||
|
|
||||||
|
logger.Debug().Msg("created packet source for receiving LLDP packets")
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *LLDP) startCapture() error {
|
||||||
|
// set up the logger and context (inside a lock)
|
||||||
|
startRunning, err := l.prepareCapture()
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if startRunning {
|
||||||
|
logger := l.Rx.Logger
|
||||||
|
go l.doCapture(logger)
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -182,7 +204,7 @@ func (l *LLDP) handlePacket(packet gopacket.Packet, logger *zerolog.Logger) erro
|
||||||
|
|
||||||
lldpRaw := packet.Layer(layers.LayerTypeLinkLayerDiscovery)
|
lldpRaw := packet.Layer(layers.LayerTypeLinkLayerDiscovery)
|
||||||
if lldpRaw != nil {
|
if lldpRaw != nil {
|
||||||
l.l.Trace().Hex("packet", packet.Data()).Msg("received LLDP frame")
|
logger.Trace().Hex("packet", packet.Data()).Msg("received LLDP frame")
|
||||||
|
|
||||||
lldpInfo := packet.Layer(layers.LayerTypeLinkLayerDiscoveryInfo)
|
lldpInfo := packet.Layer(layers.LayerTypeLinkLayerDiscoveryInfo)
|
||||||
if lldpInfo == nil {
|
if lldpInfo == nil {
|
||||||
|
|
@ -198,7 +220,7 @@ func (l *LLDP) handlePacket(packet gopacket.Packet, logger *zerolog.Logger) erro
|
||||||
|
|
||||||
cdpRaw := packet.Layer(layers.LayerTypeCiscoDiscovery)
|
cdpRaw := packet.Layer(layers.LayerTypeCiscoDiscovery)
|
||||||
if cdpRaw != nil {
|
if cdpRaw != nil {
|
||||||
l.l.Trace().Hex("packet", packet.Data()).Msg("received CDP frame")
|
logger.Trace().Hex("packet", packet.Data()).Msg("received CDP frame")
|
||||||
|
|
||||||
cdpInfo := packet.Layer(layers.LayerTypeCiscoDiscoveryInfo)
|
cdpInfo := packet.Layer(layers.LayerTypeCiscoDiscoveryInfo)
|
||||||
if cdpInfo == nil {
|
if cdpInfo == nil {
|
||||||
|
|
@ -335,32 +357,33 @@ func (l *LLDP) stopCapture() error {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
defer l.mu.Unlock()
|
defer l.mu.Unlock()
|
||||||
|
|
||||||
if !l.rxRunning {
|
if !l.Rx.Running {
|
||||||
return nil // Already stopped
|
return nil // Already stopped
|
||||||
}
|
}
|
||||||
|
|
||||||
logger := l.l.With().Str("interface", l.interfaceName).Logger()
|
logger := l.Rx.Logger
|
||||||
|
rxCancel := l.Rx.Cancel
|
||||||
|
tPacket := l.Rx.TPacket
|
||||||
|
|
||||||
logger.Info().Msg("stopping LLDP receiver")
|
logger.Info().Msg("stopping LLDP receiver")
|
||||||
|
|
||||||
// Cancel context to signal stop
|
// Cancel context to signal stop
|
||||||
rxCancel := l.rxCancel
|
|
||||||
if rxCancel != nil {
|
if rxCancel != nil {
|
||||||
|
l.Rx.Cancel = nil // so we don't cancel again
|
||||||
rxCancel()
|
rxCancel()
|
||||||
l.rxCancel = nil
|
|
||||||
|
|
||||||
logger.Info().Msg("cancelled RX context, waiting for goroutine to finish")
|
logger.Info().Msg("cancelled RX context, waiting for goroutine to finish")
|
||||||
}
|
}
|
||||||
|
|
||||||
// stop the TPacketRx
|
// stop the TPacketRx
|
||||||
go func() {
|
go func() {
|
||||||
if l.tPacketRx == nil {
|
if tPacket == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// write an empty packet to the TPacketRx to interrupt the blocking read
|
// write an empty packet to the TPacketRx to interrupt the blocking read
|
||||||
// it's a shitty workaround until https://github.com/google/gopacket/pull/777 is merged,
|
// it's a shitty workaround until https://github.com/google/gopacket/pull/777 is merged,
|
||||||
// or we have a better solution, see https://github.com/google/gopacket/issues/1064
|
// or we have a better solution, see https://github.com/google/gopacket/issues/1064
|
||||||
_ = l.tPacketRx.WritePacketData([]byte{})
|
_ = tPacket.WritePacketData([]byte{})
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// wait for the goroutine to finish
|
// wait for the goroutine to finish
|
||||||
|
|
@ -368,19 +391,20 @@ func (l *LLDP) stopCapture() error {
|
||||||
l.rxWaitGroup.Wait()
|
l.rxWaitGroup.Wait()
|
||||||
logger.Info().Dur("duration", time.Since(start)).Msg("RX goroutine finished")
|
logger.Info().Dur("duration", time.Since(start)).Msg("RX goroutine finished")
|
||||||
|
|
||||||
l.rxRunning = false
|
l.Rx.Running = false
|
||||||
|
|
||||||
if l.tPacketRx != nil {
|
|
||||||
logger.Info().Msg("closing TPacketRx")
|
|
||||||
l.tPacketRx.Close()
|
|
||||||
l.tPacketRx = nil
|
|
||||||
}
|
|
||||||
|
|
||||||
|
// close the packet source first because it's constructed with the Rx.TPacket
|
||||||
if l.pktSourceRx != nil {
|
if l.pktSourceRx != nil {
|
||||||
logger.Info().Msg("closing packet source")
|
logger.Info().Msg("closing packet source")
|
||||||
l.pktSourceRx = nil
|
l.pktSourceRx = nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if tPacket != nil {
|
||||||
|
logger.Info().Msg("closing RX TPacket")
|
||||||
|
tPacket.Close()
|
||||||
|
l.Rx.TPacket = nil
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -9,7 +9,6 @@ import (
|
||||||
|
|
||||||
"github.com/google/gopacket"
|
"github.com/google/gopacket"
|
||||||
"github.com/google/gopacket/layers"
|
"github.com/google/gopacket/layers"
|
||||||
"github.com/rs/zerolog"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
|
@ -17,14 +16,10 @@ var (
|
||||||
lldpEtherType = layers.EthernetTypeLinkLayerDiscovery
|
lldpEtherType = layers.EthernetTypeLinkLayerDiscovery
|
||||||
)
|
)
|
||||||
|
|
||||||
func (l *LLDP) toPayloadValues() []layers.LinkLayerDiscoveryValue {
|
func (l *LLDP) toPayloadValues(opts *AdvertiseOptions) []layers.LinkLayerDiscoveryValue {
|
||||||
// See also: layers.LinkLayerDiscovery.SerializeTo()
|
// See also: layers.LinkLayerDiscovery.SerializeTo()
|
||||||
r := []layers.LinkLayerDiscoveryValue{}
|
r := []layers.LinkLayerDiscoveryValue{}
|
||||||
|
|
||||||
l.mu.RLock()
|
|
||||||
opts := l.advertiseOptions
|
|
||||||
l.mu.RUnlock()
|
|
||||||
|
|
||||||
if opts == nil {
|
if opts == nil {
|
||||||
return r
|
return r
|
||||||
}
|
}
|
||||||
|
|
@ -74,35 +69,31 @@ func (l *LLDP) toPayloadValues() []layers.LinkLayerDiscoveryValue {
|
||||||
func (l *LLDP) setUpTx() error {
|
func (l *LLDP) setUpTx() error {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
defer l.mu.Unlock()
|
defer l.mu.Unlock()
|
||||||
|
|
||||||
// Check if already set up (double-check pattern to prevent duplicate setup)
|
// Check if already set up (double-check pattern to prevent duplicate setup)
|
||||||
if l.tPacketTx != nil {
|
if l.Tx.TPacket != nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
logger := l.l.With().Str("interface", l.interfaceName).Logger()
|
logger := l.l.With().Str("interface", l.interfaceName).Str("direction", "tx").Logger()
|
||||||
tPacketTx, err := afPacketNewTPacket(l.interfaceName)
|
txState := &RunningState{Enabled: true, Logger: &logger}
|
||||||
|
txState.Ctx, txState.Cancel = context.WithCancel(context.Background())
|
||||||
|
|
||||||
|
var err error
|
||||||
|
txState.TPacket, err = afPacketNewTPacket(l.interfaceName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Info().Msg("created TPacket instance for sending LLDP packets")
|
logger.Info().Msg("created TPacket instance for sending LLDP packets")
|
||||||
|
l.Tx = txState
|
||||||
l.tPacketTx = tPacketTx
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *LLDP) sendTxPackets() error {
|
func (l *LLDP) generateTxPackets(interfaceName string, opts *AdvertiseOptions) (gopacket.SerializeBuffer, error) {
|
||||||
l.mu.RLock()
|
iface, err := net.InterfaceByName(interfaceName)
|
||||||
defer l.mu.RUnlock()
|
|
||||||
|
|
||||||
logger := l.l.With().Str("interface", l.interfaceName).Logger()
|
|
||||||
iface, err := net.InterfaceByName(l.interfaceName)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return nil, err
|
||||||
}
|
|
||||||
|
|
||||||
if l.tPacketTx == nil {
|
|
||||||
return fmt.Errorf("AFPacket not initialized")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// create payload
|
// create payload
|
||||||
|
|
@ -122,46 +113,76 @@ func (l *LLDP) sendTxPackets() error {
|
||||||
ID: []byte(iface.Name),
|
ID: []byte(iface.Name),
|
||||||
},
|
},
|
||||||
TTL: uint16(3600),
|
TTL: uint16(3600),
|
||||||
Values: l.toPayloadValues(),
|
Values: l.toPayloadValues(opts),
|
||||||
}
|
}
|
||||||
|
|
||||||
buf := gopacket.NewSerializeBuffer()
|
buf := gopacket.NewSerializeBuffer()
|
||||||
if err := gopacket.SerializeLayers(buf, gopacket.SerializeOptions{
|
err = gopacket.SerializeLayers(buf, gopacket.SerializeOptions{
|
||||||
FixLengths: true,
|
FixLengths: true,
|
||||||
ComputeChecksums: true,
|
ComputeChecksums: true,
|
||||||
}, ðFrame, &lldpFrame); err != nil {
|
}, ðFrame, &lldpFrame)
|
||||||
l.l.Error().Err(err).Msg("unable to serialize packet")
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return buf, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *LLDP) sendTxPackets() error {
|
||||||
|
l.mu.RLock()
|
||||||
|
interfaceName := l.interfaceName
|
||||||
|
opts := l.advertiseOptions
|
||||||
|
logger := l.Tx.Logger
|
||||||
|
tPacket := l.Tx.TPacket
|
||||||
|
l.mu.RUnlock()
|
||||||
|
|
||||||
|
if tPacket == nil {
|
||||||
|
logger.Error().Msg("AFPacket not initialized for Tx")
|
||||||
|
return fmt.Errorf("AFPacket not initialized for Tx")
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.Trace().Msg("generating LLDP Tx packets")
|
||||||
|
buf, err := l.generateTxPackets(interfaceName, opts)
|
||||||
|
if err != nil {
|
||||||
|
logger.Error().Err(err).Msg("unable to serialize packet")
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Trace().Hex("packet", buf.Bytes()).Msg("sending LLDP packet")
|
logger.Trace().Hex("packet", buf.Bytes()).Msg("sending LLDP packet")
|
||||||
|
start := time.Now()
|
||||||
|
|
||||||
// send packet
|
// send packet
|
||||||
if err := l.tPacketTx.WritePacketData(buf.Bytes()); err != nil {
|
if err := tPacket.WritePacketData(buf.Bytes()); err != nil {
|
||||||
l.l.Error().Err(err).Msg("unable to send packet")
|
logger.Error().Err(err).Msg("unable to send packet")
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
logger.Info().Dur("duration", time.Since(start)).Msg("sent LLDP packet")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
const txInterval = 30 * time.Second // Standard LLDP transmission interval
|
const txInterval = 30 * time.Second // Standard LLDP transmission interval
|
||||||
|
|
||||||
func (l *LLDP) doSendPeriodically(logger *zerolog.Logger, txCtx context.Context) {
|
func (l *LLDP) doSendPeriodically() {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
l.txRunning = true
|
l.Tx.Running = true
|
||||||
|
logger := l.Tx.Logger
|
||||||
|
txCtx := l.Tx.Ctx
|
||||||
l.mu.Unlock()
|
l.mu.Unlock()
|
||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
l.txRunning = false
|
l.Tx.Running = false
|
||||||
l.mu.Unlock()
|
l.mu.Unlock()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
logger.Info().Msg("starting LLDP transmitter")
|
||||||
ticker := time.NewTicker(txInterval)
|
ticker := time.NewTicker(txInterval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
// Send initial packet immediately
|
// Send initial packet immediately
|
||||||
|
logger.Trace().Msg("sending initial LLDP packet")
|
||||||
if err := l.sendTxPackets(); err != nil {
|
if err := l.sendTxPackets(); err != nil {
|
||||||
logger.Error().Err(err).Msg("error sending initial LLDP packet")
|
logger.Error().Err(err).Msg("error sending initial LLDP packet")
|
||||||
}
|
}
|
||||||
|
|
@ -169,6 +190,7 @@ func (l *LLDP) doSendPeriodically(logger *zerolog.Logger, txCtx context.Context)
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
|
logger.Trace().Msg("time to send periodic LLDP packet")
|
||||||
if err := l.sendTxPackets(); err != nil {
|
if err := l.sendTxPackets(); err != nil {
|
||||||
logger.Error().Err(err).Msg("error sending LLDP packet")
|
logger.Error().Err(err).Msg("error sending LLDP packet")
|
||||||
}
|
}
|
||||||
|
|
@ -181,50 +203,45 @@ func (l *LLDP) doSendPeriodically(logger *zerolog.Logger, txCtx context.Context)
|
||||||
|
|
||||||
func (l *LLDP) startTx() error {
|
func (l *LLDP) startTx() error {
|
||||||
l.mu.RLock()
|
l.mu.RLock()
|
||||||
running := l.txRunning
|
logger := l.l
|
||||||
enabled := l.enableTx
|
running := l.Tx.Running
|
||||||
cancel := l.txCancel
|
enabled := l.Tx.Enabled
|
||||||
|
previousCancel := l.Tx.Cancel
|
||||||
l.mu.RUnlock()
|
l.mu.RUnlock()
|
||||||
|
|
||||||
if running || !enabled {
|
if running || !enabled {
|
||||||
|
logger.Trace().Bool("running", running).Bool("enabled", enabled).Msg("alrady running Tx or not enabled")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
if cancel != nil {
|
if previousCancel != nil {
|
||||||
cancel()
|
logger.Trace().Msg("stopping previous Tx context before starting new one")
|
||||||
|
previousCancel()
|
||||||
}
|
}
|
||||||
|
|
||||||
l.mu.Lock()
|
|
||||||
l.txCtx, l.txCancel = context.WithCancel(context.Background())
|
|
||||||
l.mu.Unlock()
|
|
||||||
|
|
||||||
if err := l.setUpTx(); err != nil {
|
if err := l.setUpTx(); err != nil {
|
||||||
return fmt.Errorf("failed to set up TX: %w", err)
|
return fmt.Errorf("failed to set up TX: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
logger := l.l.With().Str("interface", l.interfaceName).Logger()
|
go l.doSendPeriodically()
|
||||||
logger.Info().Msg("starting LLDP transmitter")
|
|
||||||
|
|
||||||
go l.doSendPeriodically(&logger, l.txCtx)
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *LLDP) stopTx() error {
|
func (l *LLDP) stopTx() error {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
if !l.txRunning {
|
if !l.Tx.Running {
|
||||||
l.mu.Unlock()
|
l.mu.Unlock()
|
||||||
return nil // Already stopped
|
return nil // Already stopped
|
||||||
}
|
}
|
||||||
|
|
||||||
logger := l.l.With().Str("interface", l.interfaceName).Logger()
|
|
||||||
logger.Info().Msg("stopping LLDP transmitter")
|
|
||||||
|
|
||||||
// Cancel context to signal stop
|
// Cancel context to signal stop
|
||||||
txCancel := l.txCancel
|
txCancel := l.Tx.Cancel
|
||||||
l.txRunning = false
|
l.Tx.Running = false
|
||||||
l.mu.Unlock()
|
l.mu.Unlock()
|
||||||
|
|
||||||
|
l.Tx.Logger.Info().Msg("stopping LLDP transmitter")
|
||||||
|
|
||||||
// Cancel context (goroutine will handle cleanup)
|
// Cancel context (goroutine will handle cleanup)
|
||||||
if txCancel != nil {
|
if txCancel != nil {
|
||||||
txCancel()
|
txCancel()
|
||||||
|
|
|
||||||
|
|
@ -42,7 +42,7 @@ type NetworkConfig struct {
|
||||||
IPv6Mode null.String `json:"ipv6_mode,omitempty" one_of:"slaac,dhcpv6,slaac_and_dhcpv6,static,link_local,disabled" default:"slaac"`
|
IPv6Mode null.String `json:"ipv6_mode,omitempty" one_of:"slaac,dhcpv6,slaac_and_dhcpv6,static,link_local,disabled" default:"slaac"`
|
||||||
IPv6Static *IPv6StaticConfig `json:"ipv6_static,omitempty" required_if:"IPv6Mode=static"`
|
IPv6Static *IPv6StaticConfig `json:"ipv6_static,omitempty" required_if:"IPv6Mode=static"`
|
||||||
|
|
||||||
LLDPMode null.String `json:"lldp_mode,omitempty" one_of:"disabled,rx_only,tx_only,rx_and_tx,basic,all" default:"rx_and_tx"`
|
LLDPMode null.String `json:"lldp_mode,omitempty" one_of:"disabled,enabled,rx_only,tx_only,rx_and_tx,basic,all" default:"rx_and_tx"`
|
||||||
LLDPTxTLVs []string `json:"lldp_tx_tlvs,omitempty" one_of:"chassis,port,system,vlan" default:"chassis,port,system,vlan"`
|
LLDPTxTLVs []string `json:"lldp_tx_tlvs,omitempty" one_of:"chassis,port,system,vlan" default:"chassis,port,system,vlan"`
|
||||||
MDNSMode null.String `json:"mdns_mode,omitempty" one_of:"disabled,auto,ipv4_only,ipv6_only" default:"auto"`
|
MDNSMode null.String `json:"mdns_mode,omitempty" one_of:"disabled,auto,ipv4_only,ipv6_only" default:"auto"`
|
||||||
TimeSyncMode null.String `json:"time_sync_mode,omitempty" one_of:"ntp_only,ntp_and_http,http_only,custom" default:"ntp_and_http"`
|
TimeSyncMode null.String `json:"time_sync_mode,omitempty" one_of:"ntp_only,ntp_and_http,http_only,custom" default:"ntp_and_http"`
|
||||||
|
|
@ -54,7 +54,7 @@ type NetworkConfig struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *NetworkConfig) ShouldEnableLLDPTransmit() bool {
|
func (c *NetworkConfig) ShouldEnableLLDPTransmit() bool {
|
||||||
// backwards compatibility: `basic` mode will be `rx_only` due to privacy concerns
|
// backwards compatibility: `basic` mode will be `rx_only` due to privacy concerns, `enabled` and `all` will enable transmit
|
||||||
return c.LLDPMode.String != "rx_only" && c.LLDPMode.String != "disabled" && c.LLDPMode.String != "basic"
|
return c.LLDPMode.String != "rx_only" && c.LLDPMode.String != "disabled" && c.LLDPMode.String != "basic"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -211,6 +211,7 @@ func initNetwork() error {
|
||||||
},
|
},
|
||||||
Logger: networkLogger,
|
Logger: networkLogger,
|
||||||
})
|
})
|
||||||
|
|
||||||
if err := lldpService.Start(); err != nil {
|
if err := lldpService.Start(); err != nil {
|
||||||
networkLogger.Error().Err(err).Msg("failed to start LLDP service")
|
networkLogger.Error().Err(err).Msg("failed to start LLDP service")
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -414,7 +414,7 @@ func (nm *NetlinkManager) reconcileDefaultRoute(link *Link, expected map[string]
|
||||||
|
|
||||||
// add remaining expected default routes
|
// add remaining expected default routes
|
||||||
for _, gateway := range expected {
|
for _, gateway := range expected {
|
||||||
nm.logger.Warn().Str("gateway", gateway.String()).Msg("adding default route")
|
nm.logger.Info().Str("gateway", gateway.String()).Msg("adding default route")
|
||||||
|
|
||||||
route := &netlink.Route{
|
route := &netlink.Route{
|
||||||
Dst: &ipv4DefaultRoute,
|
Dst: &ipv4DefaultRoute,
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue