5785145364

Committed 04 Aug 2023 10:26AM UTC coverage: 52.633% (-0.07%) from 52.704%

Build # 5785145364

Build Type

push

github

Committed by

jcaamano

Commit Message

libovsdb: give monitor setup time to process than normal transactions

In certain scaling situations, it may take longer than types.OVSDBTimeout for the
monitoring of OVSDB tables to be set up. This change is a follow-up to commit
da9408d3b, accounting for cases where ovnkube-master
has a lot of startup data. The timeout for monitoring introduced in commit
6ae4edb8d did not account for that.

Reported-at: https://issues.redhat.com/browse/OCPBUGS-17147
Signed-off-by: Flavio Fernandes <flaviof@redhat.com>

Run Details

23915 of 45437 relevant lines covered (52.63%)

93.92 hits per line

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

64.23

/go-controller/pkg/ovn/controller/unidling/unidle.go

package unidling

import (
        "context"
        "fmt"
        "sync"

        libovsdbcache "github.com/ovn-org/libovsdb/cache"
        libovsdbclient "github.com/ovn-org/libovsdb/client"
        "github.com/ovn-org/libovsdb/model"
        "github.com/ovn-org/libovsdb/ovsdb"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/sbdb"
        ovntypes "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"

        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
        kapi "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/types"
        utilruntime "k8s.io/apimachinery/pkg/util/runtime"
        "k8s.io/client-go/tools/cache"
        "k8s.io/client-go/tools/record"
        "k8s.io/klog/v2"
)

// unidlingController checks periodically the OVN events db
// and generates a Kubernetes NeedPods events with the Service
// associated to the VIP
type unidlingController struct {
        eventQueue    chan sbdb.ControllerEvent
        eventRecorder record.EventRecorder
        // Map of load balancers to service namespace
        serviceVIPToName     map[ServiceVIPKey]types.NamespacedName
        serviceVIPToNameLock sync.Mutex
        sbClient             libovsdbclient.Client
}

// NewController creates a new unidling controller
func NewController(recorder record.EventRecorder, serviceInformer cache.SharedIndexInformer, sbClient libovsdbclient.Client) (*unidlingController, error) {
        uc := &unidlingController{
                eventQueue:       make(chan sbdb.ControllerEvent),
                eventRecorder:    recorder,
                serviceVIPToName: map[ServiceVIPKey]types.NamespacedName{},
                sbClient:         sbClient,
        }

        klog.Info("Registering OVN SB ControllerEvent handler")
        // add all empty lb backend events to a channel
        sbClient.Cache().AddEventHandler(
                &libovsdbcache.EventHandlerFuncs{
                        AddFunc: func(table string, m model.Model) {
                                if event, ok := m.(*sbdb.ControllerEvent); ok {
                                        if event.EventType == sbdb.ControllerEventEventTypeEmptyLbBackends {
                                                uc.eventQueue <- *event
                                        }
                                }
                        },
                },
        )

        // FIXME: libovsdb event handlers should be added before the Monitor is set
        // manually populate the current events. we may get an event twice, but this
        // shouldn't cause issues as we'll just log an error message
        klog.Info("Populating Initial ContollerEvent events")

        var controllerEvents []sbdb.ControllerEvent
        ctx, cancel := context.WithTimeout(context.Background(), ovntypes.OVSDBTimeout)
        defer cancel()
        err := sbClient.List(ctx, &controllerEvents)
        if err != nil {
                return nil, err
        }
        go func() {
                for _, event := range controllerEvents {
                        uc.eventQueue <- event
                }
        }()

        // we only process events on unidling, there is no reconcilation
        klog.Info("Setting up event handlers for services")
        _, err = serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: uc.onServiceAdd,
                UpdateFunc: func(old, new interface{}) {
                        uc.onServiceDelete(old)
                        uc.onServiceAdd(new)
                },
                DeleteFunc: uc.onServiceDelete,
        })
        if err != nil {
                return nil, err
        }
        return uc, nil
}

func (uc *unidlingController) onServiceAdd(obj interface{}) {
        svc := obj.(*kapi.Service)
        if util.ServiceTypeHasClusterIP(svc) && util.IsClusterIPSet(svc) {
                for _, ip := range util.GetClusterIPs(svc) {
                        for _, svcPort := range svc.Spec.Ports {
                                vip := util.JoinHostPortInt32(ip, svcPort.Port)
                                uc.AddServiceVIPToName(vip, svcPort.Protocol, svc.Namespace, svc.Name)
                        }
                }
        }
}

func (uc *unidlingController) onServiceDelete(obj interface{}) {
        svc, ok := obj.(*kapi.Service)
        if !ok {
                tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
                if !ok {
                        utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
                        return
                }
                svc, ok = tombstone.Obj.(*kapi.Service)
                if !ok {
                        utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a Service: %#v", obj))
                        return
                }
        }

        if util.ServiceTypeHasClusterIP(svc) && util.IsClusterIPSet(svc) {
                for _, ip := range util.GetClusterIPs(svc) {
                        for _, svcPort := range svc.Spec.Ports {
                                vip := util.JoinHostPortInt32(ip, svcPort.Port)
                                uc.DeleteServiceVIPToName(vip, svcPort.Protocol)
                        }
                }
        }
}

// ServiceVIPKey is used for looking up service namespace information for a
// particular load balancer
type ServiceVIPKey struct {
        // Load balancer VIP in the form "ip:port"
        vip string
        // Protocol used by the load balancer
        protocol kapi.Protocol
}

// AddServiceVIPToName associates a k8s service name with a load balancer VIP
func (uc *unidlingController) AddServiceVIPToName(vip string, protocol kapi.Protocol, namespace, name string) {
        uc.serviceVIPToNameLock.Lock()
        defer uc.serviceVIPToNameLock.Unlock()
        uc.serviceVIPToName[ServiceVIPKey{vip, protocol}] = types.NamespacedName{Namespace: namespace, Name: name}
}

// GetServiceVIPToName retrieves the associated k8s service name for a load balancer VIP
func (uc *unidlingController) GetServiceVIPToName(vip string, protocol kapi.Protocol) (types.NamespacedName, bool) {
        uc.serviceVIPToNameLock.Lock()
        defer uc.serviceVIPToNameLock.Unlock()
        namespace, ok := uc.serviceVIPToName[ServiceVIPKey{vip, protocol}]
        return namespace, ok
}

// DeleteServiceVIPToName retrieves the associated k8s service name for a load balancer VIP
func (uc *unidlingController) DeleteServiceVIPToName(vip string, protocol kapi.Protocol) {
        uc.serviceVIPToNameLock.Lock()
        defer uc.serviceVIPToNameLock.Unlock()
        delete(uc.serviceVIPToName, ServiceVIPKey{vip, protocol})
}

func (uc *unidlingController) Run(stopCh <-chan struct{}) {
        for {
                select {
                case event := <-uc.eventQueue:
                        if err := uc.handleLbEmptyBackendsEvent(event); err != nil {
                                klog.Error(err)
                        }
                case <-stopCh:
                        return
                }
        }
}

func (uc *unidlingController) handleLbEmptyBackendsEvent(event sbdb.ControllerEvent) error {
        op, err := uc.sbClient.Where(
                &event,
        ).Delete()
        if err != nil {
                return err
        }
        ctx, cancel := context.WithTimeout(context.Background(), ovntypes.OVSDBTimeout)
        defer cancel()
        result, err := uc.sbClient.Transact(ctx, op...)
        if err != nil {
                return err
        }
        _, err = ovsdb.CheckOperationResults(result, op)
        if err != nil {
                return err
        }

        vip, ok := event.EventInfo["vip"]
        if !ok {
                return err
        }
        proto := event.EventInfo["protocol"]
        var protocol kapi.Protocol
        if proto == "udp" {
                protocol = kapi.ProtocolUDP
        } else if proto == "sctp" {
                protocol = kapi.ProtocolSCTP
        } else {
                protocol = kapi.ProtocolTCP
        }

        serviceName, ok := uc.GetServiceVIPToName(vip, protocol)

        if !ok {
                return fmt.Errorf("can't find service for vip %s:%s", protocol, vip)
        }

        serviceRef := kapi.ObjectReference{
                Kind:      "Service",
                Namespace: serviceName.Namespace,
                Name:      serviceName.Name,
        }
        klog.V(5).Infof("Sending a NeedPods event for service %s in namespace %s.", serviceName.Name, serviceName.Namespace)
        uc.eventRecorder.Eventf(&serviceRef, kapi.EventTypeNormal, "NeedPods", "The service %s needs pods", serviceName.Name)

        return nil
}

1	package unidling
2
3	import (
4	"context"
5	"fmt"
6	"sync"
7
8	libovsdbcache "github.com/ovn-org/libovsdb/cache"
9	libovsdbclient "github.com/ovn-org/libovsdb/client"
10	"github.com/ovn-org/libovsdb/model"
11	"github.com/ovn-org/libovsdb/ovsdb"
12	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/sbdb"
13	ovntypes "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
14
15	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
16	kapi "k8s.io/api/core/v1"
17	"k8s.io/apimachinery/pkg/types"
18	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
19	"k8s.io/client-go/tools/cache"
20	"k8s.io/client-go/tools/record"
21	"k8s.io/klog/v2"
22	)
23
24	// unidlingController checks periodically the OVN events db
25	// and generates a Kubernetes NeedPods events with the Service
26	// associated to the VIP
27	type unidlingController struct {
28	eventQueue chan sbdb.ControllerEvent
29	eventRecorder record.EventRecorder
30	// Map of load balancers to service namespace
31	serviceVIPToName map[ServiceVIPKey]types.NamespacedName
32	serviceVIPToNameLock sync.Mutex
33	sbClient libovsdbclient.Client
34	}
35
36	// NewController creates a new unidling controller
37	func NewController(recorder record.EventRecorder, serviceInformer cache.SharedIndexInformer, sbClient libovsdbclient.Client) (*unidlingController, error) {	1✔
38	uc := &unidlingController{	1✔
39	eventQueue: make(chan sbdb.ControllerEvent),	1✔
40	eventRecorder: recorder,	1✔
41	serviceVIPToName: map[ServiceVIPKey]types.NamespacedName{},	1✔
42	sbClient: sbClient,	1✔
43	}	1✔
44		1✔
45	klog.Info("Registering OVN SB ControllerEvent handler")	1✔
46	// add all empty lb backend events to a channel	1✔
47	sbClient.Cache().AddEventHandler(	1✔
48	&libovsdbcache.EventHandlerFuncs{	1✔
49	AddFunc: func(table string, m model.Model) {	1✔
50	if event, ok := m.(*sbdb.ControllerEvent); ok {	×
51	if event.EventType == sbdb.ControllerEventEventTypeEmptyLbBackends {	×
52	uc.eventQueue <- *event	×
53	}	×
54	}
55	},
56	},
57	)
58
59	// FIXME: libovsdb event handlers should be added before the Monitor is set
60	// manually populate the current events. we may get an event twice, but this
61	// shouldn't cause issues as we'll just log an error message
62	klog.Info("Populating Initial ContollerEvent events")	1✔
63		1✔
64	var controllerEvents []sbdb.ControllerEvent	1✔
65	ctx, cancel := context.WithTimeout(context.Background(), ovntypes.OVSDBTimeout)	1✔
66	defer cancel()	1✔
67	err := sbClient.List(ctx, &controllerEvents)	1✔
68	if err != nil {	1✔
69	return nil, err	×
70	}	×
71	go func() {	2✔
72	for _, event := range controllerEvents {	2✔
73	uc.eventQueue <- event	1✔
74	}	1✔
75	}()
76
77	// we only process events on unidling, there is no reconcilation
78	klog.Info("Setting up event handlers for services")	1✔
79	_, err = serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{	1✔
80	AddFunc: uc.onServiceAdd,	1✔
81	UpdateFunc: func(old, new interface{}) {	1✔
82	uc.onServiceDelete(old)	×
83	uc.onServiceAdd(new)	×
84	},	×
85	DeleteFunc: uc.onServiceDelete,
86	})
87	if err != nil {	1✔
88	return nil, err	×
89	}	×
90	return uc, nil	1✔
91	}
92
93	func (uc *unidlingController) onServiceAdd(obj interface{}) {	1✔
94	svc := obj.(*kapi.Service)	1✔
95	if util.ServiceTypeHasClusterIP(svc) && util.IsClusterIPSet(svc) {	2✔
96	for _, ip := range util.GetClusterIPs(svc) {	2✔
97	for _, svcPort := range svc.Spec.Ports {	2✔
98	vip := util.JoinHostPortInt32(ip, svcPort.Port)	1✔
99	uc.AddServiceVIPToName(vip, svcPort.Protocol, svc.Namespace, svc.Name)	1✔
100	}	1✔
101	}
102	}
103	}
104
105	func (uc *unidlingController) onServiceDelete(obj interface{}) {	×
106	svc, ok := obj.(*kapi.Service)	×
107	if !ok {	×
108	tombstone, ok := obj.(cache.DeletedFinalStateUnknown)	×
109	if !ok {	×
110	utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))	×
111	return	×
112	}	×
113	svc, ok = tombstone.Obj.(*kapi.Service)	×
114	if !ok {	×
115	utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a Service: %#v", obj))	×
116	return	×
117	}	×
118	}
119
120	if util.ServiceTypeHasClusterIP(svc) && util.IsClusterIPSet(svc) {	×
121	for _, ip := range util.GetClusterIPs(svc) {	×
122	for _, svcPort := range svc.Spec.Ports {	×
123	vip := util.JoinHostPortInt32(ip, svcPort.Port)	×
124	uc.DeleteServiceVIPToName(vip, svcPort.Protocol)	×
125	}	×
126	}
127	}
128	}
129
130	// ServiceVIPKey is used for looking up service namespace information for a
131	// particular load balancer
132	type ServiceVIPKey struct {
133	// Load balancer VIP in the form "ip:port"
134	vip string
135	// Protocol used by the load balancer
136	protocol kapi.Protocol
137	}
138
139	// AddServiceVIPToName associates a k8s service name with a load balancer VIP
140	func (uc *unidlingController) AddServiceVIPToName(vip string, protocol kapi.Protocol, namespace, name string) {	1✔
141	uc.serviceVIPToNameLock.Lock()	1✔
142	defer uc.serviceVIPToNameLock.Unlock()	1✔
143	uc.serviceVIPToName[ServiceVIPKey{vip, protocol}] = types.NamespacedName{Namespace: namespace, Name: name}	1✔
144	}	1✔
145
146	// GetServiceVIPToName retrieves the associated k8s service name for a load balancer VIP
147	func (uc *unidlingController) GetServiceVIPToName(vip string, protocol kapi.Protocol) (types.NamespacedName, bool) {	1✔
148	uc.serviceVIPToNameLock.Lock()	1✔
149	defer uc.serviceVIPToNameLock.Unlock()	1✔
150	namespace, ok := uc.serviceVIPToName[ServiceVIPKey{vip, protocol}]	1✔
151	return namespace, ok	1✔
152	}	1✔
153
154	// DeleteServiceVIPToName retrieves the associated k8s service name for a load balancer VIP
155	func (uc *unidlingController) DeleteServiceVIPToName(vip string, protocol kapi.Protocol) {	×
156	uc.serviceVIPToNameLock.Lock()	×
157	defer uc.serviceVIPToNameLock.Unlock()	×
158	delete(uc.serviceVIPToName, ServiceVIPKey{vip, protocol})	×
159	}	×
160
161	func (uc *unidlingController) Run(stopCh <-chan struct{}) {	1✔
162	for {	3✔
163	select {	2✔
164	case event := <-uc.eventQueue:	1✔
165	if err := uc.handleLbEmptyBackendsEvent(event); err != nil {	1✔
166	klog.Error(err)	×
167	}	×
168	case <-stopCh:	1✔
169	return	1✔
170	}
171	}
172	}
173
174	func (uc *unidlingController) handleLbEmptyBackendsEvent(event sbdb.ControllerEvent) error {	1✔
175	op, err := uc.sbClient.Where(	1✔
176	&event,	1✔
177	).Delete()	1✔
178	if err != nil {	1✔
179	return err	×
180	}	×
181	ctx, cancel := context.WithTimeout(context.Background(), ovntypes.OVSDBTimeout)	1✔
182	defer cancel()	1✔
183	result, err := uc.sbClient.Transact(ctx, op...)	1✔
184	if err != nil {	1✔
185	return err	×
186	}	×
187	_, err = ovsdb.CheckOperationResults(result, op)	1✔
188	if err != nil {	1✔
189	return err	×
190	}	×
191
192	vip, ok := event.EventInfo["vip"]	1✔
193	if !ok {	1✔
194	return err	×
195	}	×
196	proto := event.EventInfo["protocol"]	1✔
197	var protocol kapi.Protocol	1✔
198	if proto == "udp" {	1✔
199	protocol = kapi.ProtocolUDP	×
200	} else if proto == "sctp" {	1✔
201	protocol = kapi.ProtocolSCTP	×
202	} else {	1✔
203	protocol = kapi.ProtocolTCP	1✔
204	}	1✔
205
206	serviceName, ok := uc.GetServiceVIPToName(vip, protocol)	1✔
207		1✔
208	if !ok {	1✔
209	return fmt.Errorf("can't find service for vip %s:%s", protocol, vip)	×
210	}	×
211
212	serviceRef := kapi.ObjectReference{	1✔
213	Kind: "Service",	1✔
214	Namespace: serviceName.Namespace,	1✔
215	Name: serviceName.Name,	1✔
216	}	1✔
217	klog.V(5).Infof("Sending a NeedPods event for service %s in namespace %s.", serviceName.Name, serviceName.Namespace)	1✔
218	uc.eventRecorder.Eventf(&serviceRef, kapi.EventTypeNormal, "NeedPods", "The service %s needs pods", serviceName.Name)	1✔
219		1✔
220	return nil	1✔
221	}

ovn-org / ovn-kubernetes / 5785145364

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