Merge pull request #22759 from dcbw/proxy-init-node-network-ready

sdn: wait for proxy initialization before declaring node network ready

Author: openshift-merge-robot

pkg/cmd/openshift-sdn/cmd.go

@@ -15,6 +15,7 @@ import (
 
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/tools/record"
 	kubeproxyconfig "k8s.io/kubernetes/pkg/proxy/apis/config"
 	"k8s.io/kubernetes/pkg/util/interrupt"
 
@@ -28,19 +29,23 @@ import (
 	"github.com/openshift/origin/pkg/version"
 )
 
+const openshiftCNIFile string = "80-openshift-network.conf"
+
 // OpenShiftSDN stores the variables needed to initialize the real networking
 // processess from the command line.
 type OpenShiftSDN struct {
 	ConfigFilePath            string
 	KubeConfigFilePath        string
 	URLOnlyKubeConfigFilePath string
+	cniConfFile               string
 
 	NodeConfig  *configapi.NodeConfig
 	ProxyConfig *kubeproxyconfig.KubeProxyConfiguration
 
-	informers *informers
-	OsdnNode  *sdnnode.OsdnNode
-	OsdnProxy *sdnproxy.OsdnProxy
+	informers   *informers
+	OsdnNode    *sdnnode.OsdnNode
+	sdnRecorder record.EventRecorder
+	OsdnProxy   *sdnproxy.OsdnProxy
 }
 
 var networkLong = `
@@ -97,6 +102,11 @@ func (sdn *OpenShiftSDN) Run(c *cobra.Command, errout io.Writer, stopCh chan str
 		klog.Fatal(err)
 	}
 
+	// Exit early if we can't create the CNI config file directory
+	if err := os.MkdirAll(filepath.Base(sdn.cniConfFile), 0755); err != nil {
+		klog.Fatal(err)
+	}
+
 	// Set up a watch on our config file; if it changes, we should exit -
 	// (we don't have the ability to dynamically reload config changes).
 	if err := watchForChanges(sdn.ConfigFilePath, stopCh); err != nil {
@@ -164,6 +174,12 @@ func (sdn *OpenShiftSDN) ValidateAndParse() error {
 		return err
 	}
 
+	cniConfDir := "/etc/cni/net.d"
+	if val, ok := sdn.NodeConfig.KubeletArguments["cni-conf-dir"]; ok && len(val) == 1 {
+		cniConfDir = val[0]
+	}
+	sdn.cniConfFile = filepath.Join(cniConfDir, openshiftCNIFile)
+
 	return nil
 }
 
@@ -200,9 +216,17 @@ func (sdn *OpenShiftSDN) Start(stopCh <-chan struct{}) error {
 	if err != nil {
 		return err
 	}
-	sdn.runProxy()
+	proxyInitChan := make(chan bool)
+	sdn.runProxy(proxyInitChan)
 	sdn.informers.start(stopCh)
 
+	klog.V(2).Infof("openshift-sdn network plugin waiting for proxy startup to comlete")
+	<-proxyInitChan
+	klog.V(2).Infof("openshift-sdn network plugin registering startup")
+	if err := sdn.writeConfigFile(); err != nil {
+		klog.Fatal(err)
+	}
+	klog.V(2).Infof("openshift-sdn network plugin ready")
 	return nil
 }
 

pkg/cmd/openshift-sdn/proxy.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"sync"
 	"time"
 
 	"k8s.io/api/core/v1"
@@ -127,8 +128,9 @@ func (sdn *OpenShiftSDN) initProxy() error {
 	return err
 }
 
-// runProxy starts the configured proxy process.
-func (sdn *OpenShiftSDN) runProxy() {
+// runProxy starts the configured proxy process and closes the provided channel
+// when the proxy has initialized
+func (sdn *OpenShiftSDN) runProxy(waitChan chan<- bool) {
 	protocol := utiliptables.ProtocolIpv4
 	bindAddr := net.ParseIP(sdn.ProxyConfig.BindAddress)
 	if bindAddr.To4() == nil {
@@ -270,10 +272,6 @@ func (sdn *OpenShiftSDN) runProxy() {
 		proxier = hybridProxier
 	}
 
-	iptInterface.AddReloadFunc(proxier.Sync)
-	serviceConfig.RegisterEventHandler(servicesHandler)
-	go serviceConfig.Run(utilwait.NeverStop)
-
 	endpointsConfig := pconfig.NewEndpointsConfig(
 		sdn.informers.KubeInformers.Core().V1().Endpoints(),
 		sdn.ProxyConfig.ConfigSyncPeriod.Duration,
@@ -283,7 +281,15 @@ func (sdn *OpenShiftSDN) runProxy() {
 		klog.Fatalf("error: node proxy plugin startup failed: %v", err)
 	}
 	endpointsHandler = sdn.OsdnProxy
-	endpointsConfig.RegisterEventHandler(endpointsHandler)
+
+	// Wrap the proxy to know when it finally initializes
+	waitingProxy := newWaitingProxyHandler(servicesHandler, endpointsHandler, waitChan)
+
+	iptInterface.AddReloadFunc(proxier.Sync)
+	serviceConfig.RegisterEventHandler(waitingProxy)
+	go serviceConfig.Run(utilwait.NeverStop)
+
+	endpointsConfig.RegisterEventHandler(waitingProxy)
 	go endpointsConfig.Run(utilwait.NeverStop)
 
 	// Start up healthz server
@@ -346,3 +352,73 @@ func getNodeIP(client kv1core.CoreV1Interface, hostname string) (net.IP, error)
 
 	return nodeIP, nil
 }
+
+type waitingProxyHandler struct {
+	sync.Mutex
+
+	// waitChan will be closed when both services and endpoints have
+	// been synced in the proxy
+	waitChan    chan<- bool
+	initialized bool
+
+	serviceChild    pconfig.ServiceHandler
+	serviceSynced   bool
+	endpointsChild  pconfig.EndpointsHandler
+	endpointsSynced bool
+}
+
+func newWaitingProxyHandler(serviceChild pconfig.ServiceHandler, endpointsChild pconfig.EndpointsHandler, waitChan chan<- bool) *waitingProxyHandler {
+	return &waitingProxyHandler{
+		serviceChild:   serviceChild,
+		endpointsChild: endpointsChild,
+		waitChan:       waitChan,
+	}
+}
+
+func (wph *waitingProxyHandler) checkInitialized() {
+	if !wph.initialized && wph.serviceSynced && wph.endpointsSynced {
+		klog.V(2).Info("openshift-sdn proxy services and endpoints initialized")
+		wph.initialized = true
+		close(wph.waitChan)
+	}
+}
+
+func (wph *waitingProxyHandler) OnServiceAdd(service *v1.Service) {
+	wph.serviceChild.OnServiceAdd(service)
+}
+
+func (wph *waitingProxyHandler) OnServiceUpdate(oldService, service *v1.Service) {
+	wph.serviceChild.OnServiceUpdate(oldService, service)
+}
+
+func (wph *waitingProxyHandler) OnServiceDelete(service *v1.Service) {
+	wph.serviceChild.OnServiceDelete(service)
+}
+
+func (wph *waitingProxyHandler) OnServiceSynced() {
+	wph.serviceChild.OnServiceSynced()
+	wph.Lock()
+	defer wph.Unlock()
+	wph.serviceSynced = true
+	wph.checkInitialized()
+}
+
+func (wph *waitingProxyHandler) OnEndpointsAdd(endpoints *v1.Endpoints) {
+	wph.endpointsChild.OnEndpointsAdd(endpoints)
+}
+
+func (wph *waitingProxyHandler) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) {
+	wph.endpointsChild.OnEndpointsUpdate(oldEndpoints, endpoints)
+}
+
+func (wph *waitingProxyHandler) OnEndpointsDelete(endpoints *v1.Endpoints) {
+	wph.endpointsChild.OnEndpointsDelete(endpoints)
+}
+
+func (wph *waitingProxyHandler) OnEndpointsSynced() {
+	wph.endpointsChild.OnEndpointsSynced()
+	wph.Lock()
+	defer wph.Unlock()
+	wph.endpointsSynced = true
+	wph.checkInitialized()
+}

pkg/cmd/openshift-sdn/sdn.go

@@ -1,6 +1,7 @@
 package openshift_sdn
 
 import (
+	"io/ioutil"
 	"strings"
 
 	kclientv1 "k8s.io/api/core/v1"
@@ -26,10 +27,6 @@ func (sdn *OpenShiftSDN) initSDN() error {
 	if val, ok := sdn.NodeConfig.KubeletArguments["cni-bin-dir"]; ok && len(val) == 1 {
 		cniBinDir = val[0]
 	}
-	cniConfDir := "/etc/cni/net.d"
-	if val, ok := sdn.NodeConfig.KubeletArguments["cni-conf-dir"]; ok && len(val) == 1 {
-		cniConfDir = val[0]
-	}
 
 	// dockershim + kube CNI driver delegates hostport handling to plugins,
 	// while CRI-O handles hostports itself. Thus we need to disable the
@@ -38,7 +35,7 @@ func (sdn *OpenShiftSDN) initSDN() error {
 
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartRecordingToSink(&kv1core.EventSinkImpl{Interface: sdn.informers.KubeClient.CoreV1().Events("")})
-	eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, kclientv1.EventSource{Component: "openshift-sdn", Host: sdn.NodeConfig.NodeName})
+	sdn.sdnRecorder = eventBroadcaster.NewRecorder(scheme.Scheme, kclientv1.EventSource{Component: "openshift-sdn", Host: sdn.NodeConfig.NodeName})
 
 	var err error
 	sdn.OsdnNode, err = sdnnode.New(&sdnnode.OsdnNodeConfig{
@@ -47,7 +44,6 @@ func (sdn *OpenShiftSDN) initSDN() error {
 		SelfIP:             sdn.NodeConfig.NodeIP,
 		RuntimeEndpoint:    runtimeEndpoint,
 		CNIBinDir:          cniBinDir,
-		CNIConfDir:         cniConfDir,
 		MTU:                sdn.NodeConfig.NetworkConfig.MTU,
 		NetworkClient:      sdn.informers.NetworkClient,
 		KClient:            sdn.informers.KubeClient,
@@ -57,7 +53,7 @@ func (sdn *OpenShiftSDN) initSDN() error {
 		MasqueradeBit:      sdn.ProxyConfig.IPTables.MasqueradeBit,
 		ProxyMode:          sdn.ProxyConfig.Mode,
 		EnableHostports:    enableHostports,
-		Recorder:           eventRecorder,
+		Recorder:           sdn.sdnRecorder,
 	})
 	return err
 }
@@ -66,3 +62,18 @@ func (sdn *OpenShiftSDN) initSDN() error {
 func (sdn *OpenShiftSDN) runSDN() error {
 	return sdn.OsdnNode.Start()
 }
+
+func (sdn *OpenShiftSDN) writeConfigFile() error {
+	// Make an event that openshift-sdn started
+	sdn.sdnRecorder.Eventf(&kclientv1.ObjectReference{Kind: "Node", Name: sdn.NodeConfig.NodeName}, kclientv1.EventTypeNormal, "Starting", "openshift-sdn done initializing node networking.")
+
+	// Write our CNI config file out to disk to signal to kubelet that
+	// our network plugin is ready
+	return ioutil.WriteFile(sdn.cniConfFile, []byte(`
+{
+  "cniVersion": "0.3.1",
+  "name": "openshift-sdn",
+  "type": "openshift-sdn"
+}
+`), 0644)
+}

pkg/network/node/node.go

@@ -4,10 +4,7 @@ package node
 
 import (
 	"fmt"
-	"io/ioutil"
 	"net"
-	"os"
-	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -47,10 +44,7 @@ import (
 	"github.com/openshift/origin/pkg/util/ovs"
 )
 
-const (
-	openshiftCNIFile = "80-openshift-network.conf"
-	hostLocalDataDir = "/var/lib/cni/networks"
-)
+const hostLocalDataDir = "/var/lib/cni/networks"
 
 type osdnPolicy interface {
 	Name() string
@@ -77,7 +71,6 @@ type OsdnNodeConfig struct {
 	MTU             uint32
 	EnableHostports bool
 	CNIBinDir       string
-	CNIConfDir      string
 
 	NetworkClient networkclient.Interface
 	KClient       kubernetes.Interface
@@ -105,7 +98,6 @@ type OsdnNode struct {
 	useConnTrack       bool
 	iptablesSyncPeriod time.Duration
 	mtu                uint32
-	cniDirPath         string
 
 	// Synchronizes operations on egressPolicies
 	egressPoliciesLock sync.Mutex
@@ -151,10 +143,6 @@ func New(c *OsdnNodeConfig) (*OsdnNode, error) {
 		return nil, fmt.Errorf("%q plugin is not compatible with proxy-mode %q", c.PluginName, c.ProxyMode)
 	}
 
-	// If our CNI config file exists, remove it so that kubelet doesn't think
-	// we're ready yet
-	os.Remove(filepath.Join(c.CNIConfDir, openshiftCNIFile))
-
 	if err := c.setNodeIP(); err != nil {
 		return nil, err
 	}
@@ -182,7 +170,6 @@ func New(c *OsdnNodeConfig) (*OsdnNode, error) {
 		kubeInformers:      c.KubeInformers,
 		networkInformers:   c.NetworkInformers,
 		egressIP:           newEgressIPWatcher(oc, c.SelfIP, c.MasqueradeBit),
-		cniDirPath:         c.CNIConfDir,
 
 		runtimeEndpoint: c.RuntimeEndpoint,
 		// 2 minutes is the current default value used in kubelet
@@ -337,34 +324,11 @@ func (node *OsdnNode) Start() error {
 		}
 	}
 
-	if err := os.MkdirAll(node.cniDirPath, 0755); err != nil {
-		return err
-	}
-
 	go kwait.Forever(node.policy.SyncVNIDRules, time.Hour)
 	go kwait.Forever(func() {
 		gatherPeriodicMetrics(node.oc.ovs)
 	}, time.Minute*2)
 
-	klog.V(2).Infof("openshift-sdn network plugin registering startup")
-
-	// Make an event that openshift-sdn started
-	node.recorder.Eventf(&corev1.ObjectReference{Kind: "Node", Name: node.hostName}, corev1.EventTypeNormal, "Starting", "Starting openshift-sdn.")
-
-	// Write our CNI config file out to disk to signal to kubelet that
-	// our network plugin is ready
-	err = ioutil.WriteFile(filepath.Join(node.cniDirPath, openshiftCNIFile), []byte(`
-{
-  "cniVersion": "0.3.1",
-  "name": "openshift-sdn",
-  "type": "openshift-sdn"
-}
-`), 0644)
-	if err != nil {
-		return err
-	}
-
-	klog.V(2).Infof("openshift-sdn network plugin ready")
 	return nil
 }