cloud_request_manager.go

/*
Copyright 2018 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cloudresource

import (
	"context"
	"fmt"
	"sync"
	"time"

	"k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/wait"
	cloudprovider "k8s.io/cloud-provider"

	"k8s.io/klog"
)

var nodeAddressesRetryPeriod = 5 * time.Second

// SyncManager is an interface for making requests to a cloud provider
type SyncManager interface {
	Run(stopCh <-chan struct{})
	NodeAddresses() ([]v1.NodeAddress, error)
}

var _ SyncManager = &cloudResourceSyncManager{}

type cloudResourceSyncManager struct {
	// Cloud provider interface.
	cloud cloudprovider.Interface
	// Sync period
	syncPeriod time.Duration

	nodeAddressesMux sync.Mutex
	nodeAddressesErr error
	nodeAddresses    []v1.NodeAddress

	nodeName types.NodeName
}

// NewSyncManager creates a manager responsible for collecting resources
// from a cloud provider through requests that are sensitive to timeouts and hanging
func NewSyncManager(cloud cloudprovider.Interface, nodeName types.NodeName, syncPeriod time.Duration) SyncManager {
	return &cloudResourceSyncManager{
		cloud:      cloud,
		syncPeriod: syncPeriod,
		nodeName:   nodeName,
	}
}

func (manager *cloudResourceSyncManager) getNodeAddressSafe() ([]v1.NodeAddress, error) {
	manager.nodeAddressesMux.Lock()
	defer manager.nodeAddressesMux.Unlock()

	return manager.nodeAddresses, manager.nodeAddressesErr
}

func (manager *cloudResourceSyncManager) setNodeAddressSafe(nodeAddresses []v1.NodeAddress, err error) {
	manager.nodeAddressesMux.Lock()
	defer manager.nodeAddressesMux.Unlock()

	manager.nodeAddresses = nodeAddresses
	manager.nodeAddressesErr = err
}

// NodeAddresses does not wait for cloud provider to return a node addresses.
// It always returns node addresses or an error.
func (manager *cloudResourceSyncManager) NodeAddresses() ([]v1.NodeAddress, error) {
	// wait until there is something
	for {
		nodeAddresses, err := manager.getNodeAddressSafe()
		if len(nodeAddresses) == 0 && err == nil {
			klog.V(5).Infof("Waiting for %v for cloud provider to provide node addresses", nodeAddressesRetryPeriod)
			time.Sleep(nodeAddressesRetryPeriod)
			continue
		}
		return nodeAddresses, err
	}
}

func (manager *cloudResourceSyncManager) collectNodeAddresses(ctx context.Context, nodeName types.NodeName) {
	klog.V(5).Infof("Requesting node addresses from cloud provider for node %q", nodeName)

	instances, ok := manager.cloud.Instances()
	if !ok {
		manager.setNodeAddressSafe(nil, fmt.Errorf("failed to get instances from cloud provider"))
		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 {
		manager.setNodeAddressSafe(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 {
		manager.setNodeAddressSafe(nodeAddresses, nil)
		klog.V(5).Infof("Node addresses from cloud provider for node %q collected", nodeName)
	}
}

func (manager *cloudResourceSyncManager) Run(stopCh <-chan struct{}) {
	wait.Until(func() {
		manager.collectNodeAddresses(context.TODO(), manager.nodeName)
	}, manager.syncPeriod, stopCh)
}