From bf99565fbb7ff2454355881e82eeb71f1134f130 Mon Sep 17 00:00:00 2001 From: Mike Danese Date: Tue, 4 Dec 2018 13:26:55 -0800 Subject: [PATCH] fallback to previously collected addresses if a sync loop fails --- .../cloudresource/cloud_request_manager.go | 71 +++++++++++-------- .../cloud_request_manager_test.go | 52 ++++++++++++++ 2 files changed, 92 insertions(+), 31 deletions(-) diff --git a/pkg/kubelet/cloudresource/cloud_request_manager.go b/pkg/kubelet/cloudresource/cloud_request_manager.go index d6112b1307..879e131e37 100644 --- a/pkg/kubelet/cloudresource/cloud_request_manager.go +++ b/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 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) + + addrs, err := m.getNodeAddresses() + + m.nodeAddressesMonitor.L.Lock() + defer m.nodeAddressesMonitor.L.Unlock() + defer m.nodeAddressesMonitor.Broadcast() + + if err != nil { + 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 } - // 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? - - 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(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) } diff --git a/pkg/kubelet/cloudresource/cloud_request_manager_test.go b/pkg/kubelet/cloudresource/cloud_request_manager_test.go index 928fe58eef..e6d829e844 100644 --- a/pkg/kubelet/cloudresource/cloud_request_manager_test.go +++ b/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)) + } + }) + } +}