@@ -79,8 +79,6 @@ type OVSDBMonitor struct {
7979
8080 // syncQueue used to notify ovsdb update
8181 syncQueue workqueue.RateLimitingInterface
82-
83- OvsdbReconnChan chan struct {}
8482}
8583
8684// NewOVSDBMonitor create a new instance of OVSDBMonitor
@@ -97,7 +95,6 @@ func NewOVSDBMonitor() (*OVSDBMonitor, error) {
9795 localEndpointHardwareAddrCacheLock : sync.RWMutex {},
9896 localEndpointHardwareAddrCache : make (map [string ]uint32 ),
9997 syncQueue : workqueue .NewRateLimitingQueue (workqueue .DefaultItemBasedRateLimiter ()),
100- OvsdbReconnChan : make (chan struct {}),
10198 }
10299
103100 return monitor , nil
@@ -130,32 +127,17 @@ func (monitor *OVSDBMonitor) Run(stopChan <-chan struct{}) {
130127 klog .Infof ("start ovsdb monitor" )
131128 defer klog .Infof ("shutting down ovsdb monitor" )
132129
133- err := monitor .startOvsdbMonitor (false )
130+ err := monitor .startOvsdbMonitor ()
134131 if err != nil {
135132 klog .Fatalf ("unable start ovsdb monitor: %s" , err )
136133 }
137- for {
138- select {
139- case <- monitor .OvsdbReconnChan :
140- klog .Infof ("reconnect ovsdb monitor" )
141- monitor .ovsClient .Disconnect ()
142- monitor .ovsClient , err = ovsdb .ConnectUnix (ovsdb .DEFAULT_SOCK )
143- if err != nil {
144- klog .Fatalf ("unable reconnect ovsdb socket: %s" , err )
145- }
146- err := monitor .startOvsdbMonitor (true )
147- if err != nil {
148- klog .Fatalf ("unable start ovsdb monitor: %s" , err )
149- }
150- case <- stopChan :
151- return
152- }
153- }
134+
135+ <- stopChan
154136}
155137
156- func (monitor * OVSDBMonitor ) startOvsdbMonitor (isReconnect bool ) error {
138+ func (monitor * OVSDBMonitor ) startOvsdbMonitor () error {
157139 klog .Infof ("start monitor ovsdb %s" , "Open_vSwitch" )
158- monitor .ovsClient .Register (ovsUpdateHandlerFunc (monitor .handleOvsUpdates ( false ) ))
140+ monitor .ovsClient .Register (ovsUpdateHandlerFunc (monitor .handleOvsUpdates ))
159141
160142 selectAll := ovsdb.MonitorSelect {
161143 Initial : true ,
@@ -170,11 +152,10 @@ func (monitor *OVSDBMonitor) startOvsdbMonitor(isReconnect bool) error {
170152 "Open_vSwitch" : {Select : selectAll , Columns : []string {"ovs_version" }},
171153 }
172154
173- initial , err := monitor .ovsClient .Monitor ("Open_vSwitch" , nil , requests )
155+ err := monitor .ovsClient .Monitor ("Open_vSwitch" , nil , requests )
174156 if err != nil {
175157 return fmt .Errorf ("monitor ovsdb %s: %s" , "Open_vSwitch" , err )
176158 }
177- monitor .handleOvsUpdates (isReconnect )(* initial )
178159
179160 return nil
180161}
@@ -586,50 +567,34 @@ func (monitor *OVSDBMonitor) processEndpointDel(rowupdate ovsdb.RowUpdate) {
586567 }
587568}
588569
589- func (monitor * OVSDBMonitor ) handleOvsUpdates (isReconn bool ) func (updates ovsdb.TableUpdates ) {
590- return func (updates ovsdb.TableUpdates ) {
591- monitor .cacheLock .Lock ()
592- defer monitor .cacheLock .Unlock ()
593-
594- if isReconn {
595- for table := range monitor .ovsdbCache {
596- for uuid , row := range monitor .ovsdbCache [table ] {
597- if _ , ok := updates .Updates [table ].Rows [uuid ]; ! ok {
598- updates .Updates [table ].Rows [uuid ] = ovsdb.RowUpdate {
599- Uuid : ovsdb.UUID {GoUuid : uuid },
600- Old : row ,
601- New : ovsdb.Row {},
602- }
603- }
604- }
605- }
606- }
570+ func (monitor * OVSDBMonitor ) handleOvsUpdates (updates ovsdb.TableUpdates ) {
571+ monitor .cacheLock .Lock ()
572+ defer monitor .cacheLock .Unlock ()
607573
608- for table , tableUpdate := range updates .Updates {
609- if _ , ok := monitor .ovsdbCache [table ]; ! ok {
610- monitor .ovsdbCache [table ] = make (map [string ]ovsdb.Row )
611- }
612- for uuid , row := range tableUpdate .Rows {
613- empty := ovsdb.Row {}
614- if ! reflect .DeepEqual (row .New , empty ) {
615- if table == "Interface" {
616- go monitor .processEndpointUpdate (row )
617- }
618- if table == "Port" {
619- go monitor .processPortUpdate (row )
620- }
621-
622- monitor.ovsdbCache [table ][uuid ] = row .New
623- } else {
624- if table == "Interface" {
625- go monitor .processEndpointDel (row )
626- }
574+ for table , tableUpdate := range updates .Updates {
575+ if _ , ok := monitor .ovsdbCache [table ]; ! ok {
576+ monitor .ovsdbCache [table ] = make (map [string ]ovsdb.Row )
577+ }
578+ for uuid , row := range tableUpdate .Rows {
579+ empty := ovsdb.Row {}
580+ if ! reflect .DeepEqual (row .New , empty ) {
581+ if table == "Interface" {
582+ go monitor .processEndpointUpdate (row )
583+ }
584+ if table == "Port" {
585+ go monitor .processPortUpdate (row )
586+ }
627587
628- delete (monitor .ovsdbCache [table ], uuid )
588+ monitor.ovsdbCache [table ][uuid ] = row .New
589+ } else {
590+ if table == "Interface" {
591+ go monitor .processEndpointDel (row )
629592 }
593+
594+ delete (monitor .ovsdbCache [table ], uuid )
630595 }
631596 }
632-
633- monitor .syncQueue .Add ("ovsdb-event" )
634597 }
598+
599+ monitor .syncQueue .Add ("ovsdb-event" )
635600}
0 commit comments