代码拉取完成,页面将自动刷新
/*
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"
"github.com/golang/glog"
)
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 {
glog.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) {
glog.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))
glog.V(2).Infof("Node addresses from cloud provider for node %q not collected", nodeName)
} else {
manager.setNodeAddressSafe(nodeAddresses, nil)
glog.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)
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。