28222121081

Committed 26 Jun 2026 06:45AM UTC coverage: 27.011% (-0.01%) from 27.024%

Build # 28222121081

Build Type

Pull #6941

github

Committed by

zbb88888

Commit Message

refactor(speaker): improve BFD logging, type safety and peer state monitoring

- BFDDetectionMultiplier: uint32 → uint8 with pflag.Uint8 (RFC 5880 §6.8.1)
- Add ms→μs overflow validation (MaxUint32/1000 boundary)
- BFD stats: log only on errors (Warningf) or recovery (Infof)
- BFD peer state: diff caching, log only on state change
- Add watchPeerState() for BGP peer state change monitoring with BFD state
- Move newBFDPeerConfig() to bfd.go with explicit uint32 cast

Signed-off-by: zbb88888 <jmdxjsjgcxy@gmail.com>

Pull Request Pull Request #6941: fix: improve BFD logging, type safety and peer state check

Coverage Stats

15 of 74 new or added lines in 3 files covered. (20.27%)

5 existing lines in 2 files now uncovered.

16273 of 60245 relevant lines covered (27.01%)

0.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/pkg/speaker/controller.go

package speaker

import (
        "time"

        corev1 "k8s.io/api/core/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        utilruntime "k8s.io/apimachinery/pkg/util/runtime"
        "k8s.io/apimachinery/pkg/util/wait"
        kubeinformers "k8s.io/client-go/informers"
        "k8s.io/client-go/kubernetes/scheme"
        typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
        listerv1 "k8s.io/client-go/listers/core/v1"
        "k8s.io/client-go/tools/cache"
        "k8s.io/client-go/tools/record"
        "k8s.io/klog/v2"

        kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
        kubeovninformer "github.com/kubeovn/kube-ovn/pkg/client/informers/externalversions"
        kubeovnlister "github.com/kubeovn/kube-ovn/pkg/client/listers/kubeovn/v1"
        "github.com/kubeovn/kube-ovn/pkg/util"
)

const controllerAgentName = "ovn-speaker"

type Controller struct {
        config *Configuration

        podsLister listerv1.PodLister
        podsSynced cache.InformerSynced

        subnetsLister kubeovnlister.SubnetLister
        subnetSynced  cache.InformerSynced

        servicesLister listerv1.ServiceLister
        servicesSynced cache.InformerSynced

        eipLister kubeovnlister.IptablesEIPLister
        eipSynced cache.InformerSynced

        natgatewayLister kubeovnlister.VpcNatGatewayLister
        natgatewaySynced cache.InformerSynced

        informerFactory        kubeinformers.SharedInformerFactory
        podInformerFactory     kubeinformers.SharedInformerFactory
        kubeovnInformerFactory kubeovninformer.SharedInformerFactory
        recorder               record.EventRecorder

        // lastBFDPeerStates caches the most recent BFD session state per peer address.
        // Used by logBFDStatus to suppress repeated logs when state is unchanged.
        lastBFDPeerStates     map[string]string
        lastBFDStatsHasErrors bool
}

func NewController(config *Configuration) *Controller {
        utilruntime.Must(kubeovnv1.AddToScheme(scheme.Scheme))
        klog.V(4).Info("Creating event broadcaster")
        eventBroadcaster := record.NewBroadcaster()
        eventBroadcaster.StartLogging(klog.Infof)
        eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: config.KubeClient.CoreV1().Events(corev1.NamespaceAll)})
        recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})

        informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(config.KubeClient, 0,
                kubeinformers.WithTransform(util.TrimManagedFields),
                kubeinformers.WithTweakListOptions(func(listOption *metav1.ListOptions) {
                        listOption.AllowWatchBookmarks = true
                }))
        podInformerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(config.KubeClient, 0,
                kubeinformers.WithTransform(util.TrimManagedFields),
                kubeinformers.WithTweakListOptions(func(listOption *metav1.ListOptions) {
                        listOption.FieldSelector = "spec.hostNetwork=false"
                        listOption.AllowWatchBookmarks = true
                }))
        kubeovnInformerFactory := kubeovninformer.NewSharedInformerFactoryWithOptions(config.KubeOvnClient, 0,
                kubeovninformer.WithTransform(util.TrimManagedFields),
                kubeovninformer.WithTweakListOptions(func(listOption *metav1.ListOptions) {
                        listOption.AllowWatchBookmarks = true
                }))

        podInformer := podInformerFactory.Core().V1().Pods()
        subnetInformer := kubeovnInformerFactory.Kubeovn().V1().Subnets()
        serviceInformer := informerFactory.Core().V1().Services()
        eipInformer := kubeovnInformerFactory.Kubeovn().V1().IptablesEIPs()
        natgatewayInformer := kubeovnInformerFactory.Kubeovn().V1().VpcNatGateways()

        controller := &Controller{
                config: config,

                podsLister:       podInformer.Lister(),
                podsSynced:       podInformer.Informer().HasSynced,
                subnetsLister:    subnetInformer.Lister(),
                subnetSynced:     subnetInformer.Informer().HasSynced,
                servicesLister:   serviceInformer.Lister(),
                servicesSynced:   serviceInformer.Informer().HasSynced,
                eipLister:        eipInformer.Lister(),
                eipSynced:        eipInformer.Informer().HasSynced,
                natgatewayLister: natgatewayInformer.Lister(),
                natgatewaySynced: natgatewayInformer.Informer().HasSynced,

                informerFactory:        informerFactory,
                podInformerFactory:     podInformerFactory,
                kubeovnInformerFactory: kubeovnInformerFactory,
                recorder:               recorder,
        }

        return controller
}

func (c *Controller) Run(stopCh <-chan struct{}) {
        defer utilruntime.HandleCrash()
        c.informerFactory.Start(stopCh)
        c.podInformerFactory.Start(stopCh)
        c.kubeovnInformerFactory.Start(stopCh)

        if !cache.WaitForCacheSync(stopCh, c.podsSynced, c.subnetSynced, c.servicesSynced, c.eipSynced) {
                util.LogFatalAndExit(nil, "failed to wait for caches to sync")
                return
        }

        klog.Info("Started workers")
        go wait.Until(c.Reconcile, 5*time.Second, stopCh)

        <-stopCh
        klog.Info("Shutting down workers")
}

func (c *Controller) Reconcile() {
        if c.config.NatGwMode {
                err := c.syncEIPRoutes()
                if err != nil {
                        klog.Errorf("failed to reconcile EIPs: %s", err.Error())
                }
        } else {
                c.syncSubnetRoutes()
        }

        c.logBFDStatus()
}

1	package speaker
2
3	import (
4	"time"
5
6	corev1 "k8s.io/api/core/v1"
7	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
8	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
9	"k8s.io/apimachinery/pkg/util/wait"
10	kubeinformers "k8s.io/client-go/informers"
11	"k8s.io/client-go/kubernetes/scheme"
12	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
13	listerv1 "k8s.io/client-go/listers/core/v1"
14	"k8s.io/client-go/tools/cache"
15	"k8s.io/client-go/tools/record"
16	"k8s.io/klog/v2"
17
18	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
19	kubeovninformer "github.com/kubeovn/kube-ovn/pkg/client/informers/externalversions"
20	kubeovnlister "github.com/kubeovn/kube-ovn/pkg/client/listers/kubeovn/v1"
21	"github.com/kubeovn/kube-ovn/pkg/util"
22	)
23
24	const controllerAgentName = "ovn-speaker"
25
26	type Controller struct {
27	config *Configuration
28
29	podsLister listerv1.PodLister
30	podsSynced cache.InformerSynced
31
32	subnetsLister kubeovnlister.SubnetLister
33	subnetSynced cache.InformerSynced
34
35	servicesLister listerv1.ServiceLister
36	servicesSynced cache.InformerSynced
37
38	eipLister kubeovnlister.IptablesEIPLister
39	eipSynced cache.InformerSynced
40
41	natgatewayLister kubeovnlister.VpcNatGatewayLister
42	natgatewaySynced cache.InformerSynced
43
44	informerFactory kubeinformers.SharedInformerFactory
45	podInformerFactory kubeinformers.SharedInformerFactory
46	kubeovnInformerFactory kubeovninformer.SharedInformerFactory
47	recorder record.EventRecorder
48
49	// lastBFDPeerStates caches the most recent BFD session state per peer address.
50	// Used by logBFDStatus to suppress repeated logs when state is unchanged.
51	lastBFDPeerStates map[string]string
52	lastBFDStatsHasErrors bool
53	}
54
55	func NewController(config Configuration) Controller {	×
56	utilruntime.Must(kubeovnv1.AddToScheme(scheme.Scheme))	×
57	klog.V(4).Info("Creating event broadcaster")	×
58	eventBroadcaster := record.NewBroadcaster()	×
59	eventBroadcaster.StartLogging(klog.Infof)	×
60	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: config.KubeClient.CoreV1().Events(corev1.NamespaceAll)})	×
61	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})	×
62		×
63	informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(config.KubeClient, 0,	×
64	kubeinformers.WithTransform(util.TrimManagedFields),	×
65	kubeinformers.WithTweakListOptions(func(listOption *metav1.ListOptions) {	×
66	listOption.AllowWatchBookmarks = true	×
67	}))	×
68	podInformerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(config.KubeClient, 0,	×
69	kubeinformers.WithTransform(util.TrimManagedFields),	×
70	kubeinformers.WithTweakListOptions(func(listOption *metav1.ListOptions) {	×
71	listOption.FieldSelector = "spec.hostNetwork=false"	×
72	listOption.AllowWatchBookmarks = true	×
73	}))	×
74	kubeovnInformerFactory := kubeovninformer.NewSharedInformerFactoryWithOptions(config.KubeOvnClient, 0,	×
75	kubeovninformer.WithTransform(util.TrimManagedFields),	×
76	kubeovninformer.WithTweakListOptions(func(listOption *metav1.ListOptions) {	×
77	listOption.AllowWatchBookmarks = true	×
78	}))	×
79
80	podInformer := podInformerFactory.Core().V1().Pods()	×
81	subnetInformer := kubeovnInformerFactory.Kubeovn().V1().Subnets()	×
82	serviceInformer := informerFactory.Core().V1().Services()	×
83	eipInformer := kubeovnInformerFactory.Kubeovn().V1().IptablesEIPs()	×
84	natgatewayInformer := kubeovnInformerFactory.Kubeovn().V1().VpcNatGateways()	×
85		×
86	controller := &Controller{	×
87	config: config,	×
88		×
89	podsLister: podInformer.Lister(),	×
90	podsSynced: podInformer.Informer().HasSynced,	×
91	subnetsLister: subnetInformer.Lister(),	×
92	subnetSynced: subnetInformer.Informer().HasSynced,	×
93	servicesLister: serviceInformer.Lister(),	×
94	servicesSynced: serviceInformer.Informer().HasSynced,	×
95	eipLister: eipInformer.Lister(),	×
96	eipSynced: eipInformer.Informer().HasSynced,	×
97	natgatewayLister: natgatewayInformer.Lister(),	×
98	natgatewaySynced: natgatewayInformer.Informer().HasSynced,	×
99		×
100	informerFactory: informerFactory,	×
101	podInformerFactory: podInformerFactory,	×
102	kubeovnInformerFactory: kubeovnInformerFactory,	×
103	recorder: recorder,	×
104	}	×
105		×
106	return controller	×
107	}
108
109	func (c *Controller) Run(stopCh <-chan struct{}) {	×
110	defer utilruntime.HandleCrash()	×
111	c.informerFactory.Start(stopCh)	×
112	c.podInformerFactory.Start(stopCh)	×
113	c.kubeovnInformerFactory.Start(stopCh)	×
114		×
115	if !cache.WaitForCacheSync(stopCh, c.podsSynced, c.subnetSynced, c.servicesSynced, c.eipSynced) {	×
116	util.LogFatalAndExit(nil, "failed to wait for caches to sync")	×
117	return	×
118	}	×
119
120	klog.Info("Started workers")	×
121	go wait.Until(c.Reconcile, 5*time.Second, stopCh)	×
122		×
123	<-stopCh	×
124	klog.Info("Shutting down workers")	×
125	}
126
127	func (c *Controller) Reconcile() {	×
128	if c.config.NatGwMode {	×
129	err := c.syncEIPRoutes()	×
130	if err != nil {	×
131	klog.Errorf("failed to reconcile EIPs: %s", err.Error())	×
132	}	×
133	} else {	×
134	c.syncSubnetRoutes()	×
135	}	×
136
NEW 137	c.logBFDStatus()	×
138	}

kubeovn / kube-ovn / 28222121081

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous