@@ -10,21 +10,118 @@ import (
1010
1111 kapierrors "k8s.io/apimachinery/pkg/api/errors"
1212 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
13+ ktypes "k8s.io/apimachinery/pkg/types"
14+ kerrors "k8s.io/apimachinery/pkg/util/errors"
1315 utilruntime "k8s.io/apimachinery/pkg/util/runtime"
1416 utilwait "k8s.io/apimachinery/pkg/util/wait"
1517 "k8s.io/apimachinery/pkg/watch"
1618
1719 networkapi "github.com/openshift/origin/pkg/network/apis/network"
1820 "github.com/openshift/origin/pkg/network/common"
21+ networkinformers "github.com/openshift/origin/pkg/network/generated/informers/internalversion"
1922)
2023
21- func (node * OsdnNode ) SubnetStartNode () {
22- node .watchSubnets ()
24+ type hostSubnetWatcher struct {
25+ oc * ovsController
26+ localIP string
27+ networkInfo * common.NetworkInfo
28+
29+ hostSubnetMap map [ktypes.UID ]* networkapi.HostSubnet
30+ }
31+
32+ func newHostSubnetWatcher (oc * ovsController , localIP string , networkInfo * common.NetworkInfo ) * hostSubnetWatcher {
33+ return & hostSubnetWatcher {
34+ oc : oc ,
35+ localIP : localIP ,
36+ networkInfo : networkInfo ,
37+
38+ hostSubnetMap : make (map [ktypes.UID ]* networkapi.HostSubnet ),
39+ }
40+ }
41+
42+ func (hsw * hostSubnetWatcher ) Start (networkInformers networkinformers.SharedInformerFactory ) {
43+ funcs := common .InformerFuncs (& networkapi.HostSubnet {}, hsw .handleAddOrUpdateHostSubnet , hsw .handleDeleteHostSubnet )
44+ networkInformers .Network ().InternalVersion ().HostSubnets ().Informer ().AddEventHandler (funcs )
45+ }
46+
47+ func (hsw * hostSubnetWatcher ) handleAddOrUpdateHostSubnet (obj , _ interface {}, eventType watch.EventType ) {
48+ hs := obj .(* networkapi.HostSubnet )
49+ glog .V (5 ).Infof ("Watch %s event for HostSubnet %q" , eventType , hs .Name )
50+
51+ if err := hsw .updateHostSubnet (hs ); err != nil {
52+ utilruntime .HandleError (err )
53+ }
54+ }
55+
56+ func (hsw * hostSubnetWatcher ) handleDeleteHostSubnet (obj interface {}) {
57+ hs := obj .(* networkapi.HostSubnet )
58+ glog .V (5 ).Infof ("Watch %s event for HostSubnet %q" , watch .Deleted , hs .Name )
59+
60+ if err := hsw .deleteHostSubnet (hs ); err != nil {
61+ utilruntime .HandleError (err )
62+ }
2363}
2464
25- func (node * OsdnNode ) watchSubnets () {
26- funcs := common .InformerFuncs (& networkapi.HostSubnet {}, node .handleAddOrUpdateHostSubnet , node .handleDeleteHostSubnet )
27- node .networkInformers .Network ().InternalVersion ().HostSubnets ().Informer ().AddEventHandler (funcs )
65+ func (hsw * hostSubnetWatcher ) updateHostSubnet (hs * networkapi.HostSubnet ) error {
66+ if hs .HostIP == hsw .localIP {
67+ return nil
68+ }
69+ oldSubnet , exists := hsw .hostSubnetMap [hs .UID ]
70+ if exists {
71+ if oldSubnet .HostIP == hs .HostIP {
72+ return nil
73+ } else {
74+ // Delete old subnet rules (ignore errors)
75+ hsw .oc .DeleteHostSubnetRules (oldSubnet )
76+ }
77+ }
78+ if err := hsw .networkInfo .ValidateNodeIP (hs .HostIP ); err != nil {
79+ return fmt .Errorf ("Ignoring invalid subnet for node %s: %v" , hs .HostIP , err )
80+ }
81+
82+ hsw .hostSubnetMap [hs .UID ] = hs
83+
84+ errList := []error {}
85+ if err := hsw .oc .AddHostSubnetRules (hs ); err != nil {
86+ errList = append (errList , fmt .Errorf ("error adding OVS flows for subnet %q: %v" , hs .Subnet , err ))
87+ }
88+ // Update multicast rules after all other changes have been processed
89+ if err := hsw .updateVXLANMulticastRules (); err != nil {
90+ errList = append (errList , fmt .Errorf ("error updating OVS VXLAN multicast flows: %v" , err ))
91+ }
92+
93+ return kerrors .NewAggregate (errList )
94+ }
95+
96+ func (hsw * hostSubnetWatcher ) deleteHostSubnet (hs * networkapi.HostSubnet ) error {
97+ if hs .HostIP == hsw .localIP {
98+ return nil
99+ }
100+ if _ , exists := hsw .hostSubnetMap [hs .UID ]; ! exists {
101+ return nil
102+ }
103+
104+ delete (hsw .hostSubnetMap , hs .UID )
105+
106+ errList := []error {}
107+ if err := hsw .oc .DeleteHostSubnetRules (hs ); err != nil {
108+ errList = append (errList , fmt .Errorf ("error deleting OVS flows for subnet %q: %v" , hs .Subnet , err ))
109+ }
110+ if err := hsw .updateVXLANMulticastRules (); err != nil {
111+ errList = append (errList , fmt .Errorf ("error updating OVS VXLAN multicast flows: %v" , err ))
112+ }
113+
114+ return kerrors .NewAggregate (errList )
115+ }
116+
117+ func (hsw * hostSubnetWatcher ) updateVXLANMulticastRules () error {
118+ remoteIPs := make ([]string , 0 , len (hsw .hostSubnetMap ))
119+ for _ , subnet := range hsw .hostSubnetMap {
120+ if subnet .HostIP != hsw .localIP {
121+ remoteIPs = append (remoteIPs , subnet .HostIP )
122+ }
123+ }
124+ return hsw .oc .UpdateVXLANMulticastFlows (remoteIPs )
28125}
29126
30127func (node * OsdnNode ) getLocalSubnet () (string , error ) {
@@ -68,59 +165,3 @@ func (node *OsdnNode) getLocalSubnet() (string, error) {
68165
69166 return subnet .Subnet , nil
70167}
71-
72- func (node * OsdnNode ) handleAddOrUpdateHostSubnet (obj , _ interface {}, eventType watch.EventType ) {
73- hs := obj .(* networkapi.HostSubnet )
74- glog .V (5 ).Infof ("Watch %s event for HostSubnet %q" , eventType , hs .Name )
75-
76- if hs .HostIP == node .localIP {
77- return
78- }
79- oldSubnet , exists := node .hostSubnetMap [string (hs .UID )]
80- if exists {
81- if oldSubnet .HostIP == hs .HostIP {
82- return
83- } else {
84- // Delete old subnet rules
85- node .DeleteHostSubnetRules (oldSubnet )
86- }
87- }
88- if err := node .networkInfo .ValidateNodeIP (hs .HostIP ); err != nil {
89- glog .Warningf ("Ignoring invalid subnet for node %s: %v" , hs .HostIP , err )
90- return
91- }
92- node .AddHostSubnetRules (hs )
93- node .hostSubnetMap [string (hs .UID )] = hs
94-
95- // Update multicast rules after all other changes have been processed
96- node .updateVXLANMulticastRules ()
97- }
98-
99- func (node * OsdnNode ) handleDeleteHostSubnet (obj interface {}) {
100- hs := obj .(* networkapi.HostSubnet )
101- glog .V (5 ).Infof ("Watch %s event for HostSubnet %q" , watch .Deleted , hs .Name )
102-
103- if hs .HostIP == node .localIP {
104- return
105- }
106- if _ , exists := node .hostSubnetMap [string (hs .UID )]; ! exists {
107- return
108- }
109-
110- delete (node .hostSubnetMap , string (hs .UID ))
111- node .DeleteHostSubnetRules (hs )
112-
113- node .updateVXLANMulticastRules ()
114- }
115-
116- func (node * OsdnNode ) updateVXLANMulticastRules () {
117- remoteIPs := make ([]string , 0 , len (node .hostSubnetMap ))
118- for _ , subnet := range node .hostSubnetMap {
119- if subnet .HostIP != node .localIP {
120- remoteIPs = append (remoteIPs , subnet .HostIP )
121- }
122- }
123- if err := node .oc .UpdateVXLANMulticastFlows (remoteIPs ); err != nil {
124- utilruntime .HandleError (fmt .Errorf ("Error updating OVS VXLAN multicast flows: %v" , err ))
125- }
126- }
0 commit comments