fallback to previously collected addresses if a sync loop fails

pkg/kubelet/cloudresource/cloud_request_manager.go

@@ -70,17 +70,10 @@ func NewSyncManager(cloud cloudprovider.Interface, nodeName types.NodeName, sync
 	}
 }
 
-func (m *cloudResourceSyncManager) updateAddresses(addrs []v1.NodeAddress, err error) {
-	m.nodeAddressesMonitor.L.Lock()
-	defer m.nodeAddressesMonitor.L.Unlock()
-	defer m.nodeAddressesMonitor.Broadcast()
-
-	m.nodeAddresses = addrs
-	m.nodeAddressesErr = err
-}
-
-// NodeAddresses does not wait for cloud provider to return a node addresses.
-// It always returns node addresses or an error.
+// NodeAddresses waits for the first sync loop to run. If no successful syncs
+// have run, it will return the most recent error. If node addresses have been
+// synced successfully, it will return the list of node addresses from the most
+// recent successful sync.
 func (m *cloudResourceSyncManager) NodeAddresses() ([]v1.NodeAddress, error) {
 	m.nodeAddressesMonitor.L.Lock()
 	defer m.nodeAddressesMonitor.L.Unlock()
@@ -94,33 +87,49 @@ func (m *cloudResourceSyncManager) NodeAddresses() ([]v1.NodeAddress, error) {
 	}
 }
 
-func (m *cloudResourceSyncManager) collectNodeAddresses(ctx context.Context, nodeName types.NodeName) {
-	klog.V(5).Infof("Requesting node addresses from cloud provider for node %q", nodeName)
-
+// getNodeAddresses calls the cloud provider to get a current list of node addresses.
+func (m *cloudResourceSyncManager) getNodeAddresses() ([]v1.NodeAddress, error) {
+	// TODO(roberthbailey): Can we do this without having credentials to talk to
+	// the cloud provider?
+	// TODO(justinsb): We can if CurrentNodeName() was actually CurrentNode() and
+	// returned an interface.
+	// TODO: If IP addresses couldn't be fetched from the cloud provider, should
+	// kubelet fallback on the other methods for getting the IP below?
 	instances, ok := m.cloud.Instances()
 	if !ok {
-		m.updateAddresses(nil, fmt.Errorf("failed to get instances from cloud provider"))
-		return
+		return nil, fmt.Errorf("failed to get instances from cloud provider")
 	}
+	return instances.NodeAddresses(context.TODO(), m.nodeName)
+}
+
+func (m *cloudResourceSyncManager) syncNodeAddresses() {
+	klog.V(5).Infof("Requesting node addresses from cloud provider for node %q", m.nodeName)
 
-	// TODO(roberthbailey): Can we do this without having credentials to talk
-	// to the cloud provider?
-	// TODO(justinsb): We can if CurrentNodeName() was actually CurrentNode() and returned an interface
-	// TODO: If IP addresses couldn't be fetched from the cloud provider, should kubelet fallback on the other methods for getting the IP below?
+	addrs, err := m.getNodeAddresses()
+
+	m.nodeAddressesMonitor.L.Lock()
+	defer m.nodeAddressesMonitor.L.Unlock()
+	defer m.nodeAddressesMonitor.Broadcast()
 
-	nodeAddresses, err := instances.NodeAddresses(ctx, nodeName)
 	if err != nil {
-		m.updateAddresses(nil, fmt.Errorf("failed to get node address from cloud provider: %v", err))
-		klog.V(2).Infof("Node addresses from cloud provider for node %q not collected", nodeName)
-	} else {
-		m.updateAddresses(nodeAddresses, nil)
-		klog.V(5).Infof("Node addresses from cloud provider for node %q collected", nodeName)
+		klog.V(2).Infof("Node addresses from cloud provider for node %q not collected: %v", m.nodeName, err)
+
+		if len(m.nodeAddresses) > 0 {
+			// in the event that a sync loop fails when a previous sync had
+			// succeeded, continue to use the old addresses.
+			return
+		}
+
+		m.nodeAddressesErr = fmt.Errorf("failed to get node address from cloud provider: %v", err)
+		return
 	}
+
+	klog.V(5).Infof("Node addresses from cloud provider for node %q collected", m.nodeName)
+	m.nodeAddressesErr = nil
+	m.nodeAddresses = addrs
 }
 
 // Run starts the cloud resource sync manager's sync loop.
 func (m *cloudResourceSyncManager) Run(stopCh <-chan struct{}) {
-	wait.Until(func() {
-		m.collectNodeAddresses(context.TODO(), m.nodeName)
-	}, m.syncPeriod, stopCh)
+	wait.Until(m.syncNodeAddresses, m.syncPeriod, stopCh)
 }

pkg/kubelet/cloudresource/cloud_request_manager_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package cloudresource
 
 import (
+	"errors"
 	"reflect"
 	"testing"
 	"time"
@@ -79,3 +80,54 @@ func TestNodeAddressesDelay(t *testing.T) {
 	}
 	t.Errorf("Timeout waiting for %q address to appear", cloud.Addresses[0].Address)
 }
+
+func TestNodeAddressesUsesLastSuccess(t *testing.T) {
+	cloud := &fake.FakeCloud{}
+	manager := NewSyncManager(cloud, "defaultNode", 0).(*cloudResourceSyncManager)
+
+	// These tests are stateful and order dependant.
+	tests := []struct {
+		name      string
+		addrs     []v1.NodeAddress
+		err       error
+		wantAddrs []v1.NodeAddress
+		wantErr   bool
+	}{
+		{
+			name:    "first sync loop encounters an error",
+			err:     errors.New("bad"),
+			wantErr: true,
+		},
+		{
+			name:      "subsequent sync loop succeeds",
+			addrs:     createNodeInternalIPAddress("10.0.1.12"),
+			wantAddrs: createNodeInternalIPAddress("10.0.1.12"),
+		},
+		{
+			name:      "subsequent sync loop encounters an error, last addresses returned",
+			err:       errors.New("bad"),
+			wantAddrs: createNodeInternalIPAddress("10.0.1.12"),
+		},
+		{
+			name:      "subsequent sync loop succeeds changing addresses",
+			addrs:     createNodeInternalIPAddress("10.0.1.13"),
+			wantAddrs: createNodeInternalIPAddress("10.0.1.13"),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cloud.Addresses = test.addrs
+			cloud.Err = test.err
+
+			manager.syncNodeAddresses()
+			nodeAddresses, err := manager.NodeAddresses()
+			if (err != nil) != test.wantErr {
+				t.Errorf("unexpected err: %v", err)
+			}
+			if got, want := nodeAddresses, test.wantAddrs; !reflect.DeepEqual(got, want) {
+				t.Errorf("Unexpected diff of node addresses: %v", diff.ObjectReflectDiff(got, want))
+			}
+		})
+	}
+}