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

@@ -220,6 +220,11 @@ func (node *OsdnNode) Start() error {
 		return err
 	}
 
+	// podManager must be created early because other goroutines
+	// may call into it before its started due to event watches
+	log.V(5).Infof("Creating openshift-sdn pod manager")
+	node.podManager = newPodManager(node.kClient, node.policy, node.mtu, node.oc)
+
 	err = node.SubnetStartNode()
 	if err != nil {
 		return err
@@ -242,12 +247,9 @@ 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.oc)
-	if err != nil {
-		return err
-	}
-	if err := node.podManager.Start(cniserver.CNIServerSocketPath); err != nil {
+	// Kubelet has initialized, now we have a valid node.host
+	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

@@ -36,45 +36,38 @@ 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
+	oc      *ovsController
+
+	// Things only accessed through the processCNIRequests() goroutine
+	// and thus can be set from Start()
 	ipamConfig      []byte
-	mtu             uint32
 	hostportHandler kubehostport.HostportHandler
 	host            knetwork.Host
-	oc              *ovsController
 }
 
 // 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, oc *ovsController) (*podManager, error) {
-	pm := newDefaultPodManager(host)
+func newPodManager(kClient *kclientset.Clientset, policy osdnPolicy, mtu uint32, oc *ovsController) *podManager {
+	pm := newDefaultPodManager()
 	pm.kClient = kClient
 	pm.policy = policy
 	pm.mtu = mtu
-	pm.hostportHandler = kubehostport.NewHostportHandler()
-	pm.podHandler = pm
 	pm.oc = oc
-
-	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,
 	}
 }
 
@@ -132,7 +125,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,