sdn: fix initialization order to prevent crash on node startup

OsdnNode.Start()
   (node.pm == nil at this point)
   -> node.policy.Start()  (which is multitenant policy)
   -> mp.vnids.Start()
   -> go vmap.watchNetNamespaces()
   -> (net namespace event happens)
   -> watchNetNamespaces()
   -> vmap.policy.AddNetNamespace() (policy is multitenant)
   -> mp.updatePodNetwork()
   -> mp.node.podManager.UpdateLocalMulticastRules() (and podManager is still nil)

Create the PodManager earlier so it's not nil if we get early events.

Fixes: #13742

Author: dcbw

pkg/sdn/plugin/node.go

@@ -120,6 +120,7 @@ func NewNodePlugin(pluginName string, osClient *osclient.Client, kClient *kclien
 		osClient:           osClient,
 		ovs:                ovsif,
 		localIP:            selfIP,
+		podManager:         newPodManager(kClient, policy, mtu, ovsif),
 		hostName:           hostname,
 		podNetworkReady:    make(chan struct{}),
 		kubeletInitReady:   make(chan struct{}),
@@ -201,31 +202,11 @@ func (node *OsdnNode) Start() error {
 		return fmt.Errorf("Failed to get network information: %v", err)
 	}
 
-	nodeIPTables := newNodeIPTables(node.networkInfo.ClusterNetwork.String(), node.iptablesSyncPeriod)
-	if err = nodeIPTables.Setup(); err != nil {
-		return fmt.Errorf("Failed to set up iptables: %v", err)
-	}
-
 	node.localSubnetCIDR, err = node.getLocalSubnet()
 	if err != nil {
 		return err
 	}
 
-	networkChanged, err := node.SetupSDN()
-	if err != nil {
-		return err
-	}
-
-	err = node.SubnetStartNode()
-	if err != nil {
-		return err
-	}
-
-	if err = node.policy.Start(node); err != nil {
-		return err
-	}
-	go kwait.Forever(node.watchServices, 0)
-
 	// Wait for kubelet to init the plugin so we get a knetwork.Host
 	log.V(5).Infof("Waiting for kubelet network plugin initialization")
 	<-node.kubeletInitReady
@@ -238,12 +219,30 @@ func (node *OsdnNode) Start() error {
 			return true, nil
 		})
 
-	log.V(5).Infof("Creating and initializing openshift-sdn pod manager")
-	node.podManager, err = newPodManager(node.host, node.localSubnetCIDR, node.networkInfo, node.kClient, node.policy, node.mtu, node.ovs)
+	//**** After this point, all OsdnNode fields have been initialized
+
+	nodeIPTables := newNodeIPTables(node.networkInfo.ClusterNetwork.String(), node.iptablesSyncPeriod)
+	if err = nodeIPTables.Setup(); err != nil {
+		return fmt.Errorf("Failed to set up iptables: %v", err)
+	}
+
+	networkChanged, err := node.SetupSDN()
 	if err != nil {
 		return err
 	}
-	if err := node.podManager.Start(cniserver.CNIServerSocketPath); err != nil {
+
+	err = node.SubnetStartNode()
+	if err != nil {
+		return err
+	}
+
+	if err = node.policy.Start(node); err != nil {
+		return err
+	}
+	go kwait.Forever(node.watchServices, 0)
+
+	log.V(5).Infof("Starting openshift-sdn pod manager")
+	if err := node.podManager.Start(cniserver.CNIServerSocketPath, node.host, node.localSubnetCIDR, node.networkInfo.ClusterNetwork); err != nil {
 		return err
 	}
 

pkg/sdn/plugin/pod.go

@@ -38,45 +38,39 @@ type podManager struct {
 	cniServer  *cniserver.CNIServer
 	// Request queue for pod operations incoming from the CNIServer
 	requests chan (*cniserver.PodRequest)
-	// Tracks pod :: IP address for hostport handling
+	// Tracks pod :: IP address for hostport and multicast handling
 	runningPods     map[string]*runningPod
 	runningPodsLock sync.Mutex
 
 	// Live pod setup/teardown stuff not used in testing code
-	kClient         *kclientset.Clientset
-	policy          osdnPolicy
+	kClient *kclientset.Clientset
+	policy  osdnPolicy
+	mtu     uint32
+	ovs     *ovs.Interface
+
+	// Things only accessed through the processCNIRequests() goroutine
+	// and thus can be set from Start()
 	ipamConfig      []byte
-	mtu             uint32
 	hostportHandler kubehostport.HostportHandler
 	host            knetwork.Host
-	ovs             *ovs.Interface
 }
 
 // Creates a new live podManager; used by node code
-func newPodManager(host knetwork.Host, localSubnetCIDR string, netInfo *NetworkInfo, kClient *kclientset.Clientset, policy osdnPolicy, mtu uint32, ovs *ovs.Interface) (*podManager, error) {
-	pm := newDefaultPodManager(host)
+func newPodManager(kClient *kclientset.Clientset, policy osdnPolicy, mtu uint32, ovs *ovs.Interface) *podManager {
+	pm := newDefaultPodManager()
 	pm.kClient = kClient
 	pm.policy = policy
 	pm.mtu = mtu
-	pm.hostportHandler = kubehostport.NewHostportHandler()
 	pm.podHandler = pm
 	pm.ovs = ovs
-
-	var err error
-	pm.ipamConfig, err = getIPAMConfig(netInfo.ClusterNetwork, localSubnetCIDR)
-	if err != nil {
-		return nil, err
-	}
-
-	return pm, nil
+	return pm
 }
 
 // Creates a new basic podManager; used by testcases
-func newDefaultPodManager(host knetwork.Host) *podManager {
+func newDefaultPodManager() *podManager {
 	return &podManager{
 		runningPods: make(map[string]*runningPod),
 		requests:    make(chan *cniserver.PodRequest, 20),
-		host:        host,
 	}
 }
 
@@ -134,7 +128,15 @@ func getIPAMConfig(clusterNetwork *net.IPNet, localSubnet string) ([]byte, error
 }
 
 // Start the CNI server and start processing requests from it
-func (m *podManager) Start(socketPath string) error {
+func (m *podManager) Start(socketPath string, host knetwork.Host, localSubnetCIDR string, clusterNetwork *net.IPNet) error {
+	m.host = host
+	m.hostportHandler = kubehostport.NewHostportHandler()
+
+	var err error
+	if m.ipamConfig, err = getIPAMConfig(clusterNetwork, localSubnetCIDR); err != nil {
+		return err
+	}
+
 	go m.processCNIRequests()
 
 	m.cniServer = cniserver.NewCNIServer(socketPath)

pkg/sdn/plugin/pod_test.go

@@ -368,9 +368,10 @@ func TestPodManager(t *testing.T) {
 
 	for k, tc := range testcases {
 		podTester := newPodTester(t, k, socketPath)
-		podManager := newDefaultPodManager(newFakeHost())
+		podManager := newDefaultPodManager()
 		podManager.podHandler = podTester
-		podManager.Start(socketPath)
+		_, net, _ := net.ParseCIDR("1.2.0.0/16")
+		podManager.Start(socketPath, newFakeHost(), "1.2.3.0/24", net)
 
 		// Add pods to our expected pod list before kicking off the
 		// actual pod setup to ensure we don't concurrently access
@@ -463,9 +464,10 @@ func TestDirectPodUpdate(t *testing.T) {
 	socketPath := filepath.Join(tmpDir, "cni-server.sock")
 
 	podTester := newPodTester(t, "update", socketPath)
-	podManager := newDefaultPodManager(newFakeHost())
+	podManager := newDefaultPodManager()
 	podManager.podHandler = podTester
-	podManager.Start(socketPath)
+	_, net, _ := net.ParseCIDR("1.2.0.0/16")
+	podManager.Start(socketPath, newFakeHost(), "1.2.3.0/24", net)
 
 	op := &operation{
 		command:   cniserver.CNI_UPDATE,