代码拉取完成,页面将自动刷新
/*
Copyright 2016 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 config
import (
"k8s.io/client-go/pkg/util/wait"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/client/cache"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
fed "k8s.io/kubernetes/pkg/dns/federation"
"k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/watch"
"time"
"github.com/golang/glog"
)
// Sync manages synchronization of the config map.
type Sync interface {
// Once does a blocking synchronization of the config map. If the
// ConfigMap fails to validate, this method will return nil, err.
Once() (*Config, error)
// Start a periodic synchronization of the configuration map. When a
// successful configuration map update is detected, the
// configuration will be sent to the channel.
//
// It is an error to call this more than once.
Periodic() <-chan *Config
}
// NewSync for ConfigMap from namespace `ns` and `name`.
func NewSync(client clientset.Interface, ns string, name string) Sync {
sync := &kubeSync{
ns: ns,
name: name,
client: client,
channel: make(chan *Config),
}
listWatch := &cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
options.FieldSelector = fields.Set{"metadata.name": name}.AsSelector()
return client.Core().ConfigMaps(ns).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
options.FieldSelector = fields.Set{"metadata.name": name}.AsSelector()
return client.Core().ConfigMaps(ns).Watch(options)
},
}
store, controller := cache.NewInformer(
listWatch,
&api.ConfigMap{},
time.Duration(0),
cache.ResourceEventHandlerFuncs{
AddFunc: sync.onAdd,
DeleteFunc: sync.onDelete,
UpdateFunc: sync.onUpdate,
})
sync.store = store
sync.controller = controller
return sync
}
// kubeSync implements Sync for the Kubernetes API.
type kubeSync struct {
ns string
name string
client clientset.Interface
store cache.Store
controller *cache.Controller
channel chan *Config
latestVersion string
}
var _ Sync = (*kubeSync)(nil)
func (sync *kubeSync) Once() (*Config, error) {
cm, err := sync.client.Core().ConfigMaps(sync.ns).Get(sync.name)
if err != nil {
glog.Errorf("Error getting ConfigMap %v:%v err: %v",
sync.ns, sync.name, err)
return nil, err
}
config, _, err := sync.processUpdate(cm)
return config, err
}
func (sync *kubeSync) Periodic() <-chan *Config {
go sync.controller.Run(wait.NeverStop)
return sync.channel
}
func (sync *kubeSync) toConfigMap(obj interface{}) *api.ConfigMap {
cm, ok := obj.(*api.ConfigMap)
if !ok {
glog.Fatalf("Expected ConfigMap, got %T", obj)
}
return cm
}
func (sync *kubeSync) onAdd(obj interface{}) {
cm := sync.toConfigMap(obj)
glog.V(2).Infof("ConfigMap %s:%s was created", sync.ns, sync.name)
config, updated, err := sync.processUpdate(cm)
if updated && err == nil {
sync.channel <- config
}
}
func (sync *kubeSync) onDelete(_ interface{}) {
glog.V(2).Infof("ConfigMap %s:%s was deleted, reverting to default configuration",
sync.ns, sync.name)
sync.latestVersion = ""
sync.channel <- NewDefaultConfig()
}
func (sync *kubeSync) onUpdate(_, obj interface{}) {
cm := sync.toConfigMap(obj)
glog.V(2).Infof("ConfigMap %s:%s was updated", sync.ns, sync.name)
config, changed, err := sync.processUpdate(cm)
if changed && err == nil {
sync.channel <- config
}
}
func (sync *kubeSync) processUpdate(cm *api.ConfigMap) (config *Config, changed bool, err error) {
glog.V(4).Infof("processUpdate ConfigMap %+v", *cm)
if cm.ObjectMeta.ResourceVersion != sync.latestVersion {
glog.V(3).Infof("Updating config to version %v (was %v)",
cm.ObjectMeta.ResourceVersion, sync.latestVersion)
changed = true
sync.latestVersion = cm.ObjectMeta.ResourceVersion
} else {
glog.V(4).Infof("Config was unchanged (version %v)", sync.latestVersion)
return
}
config = &Config{}
if err = sync.updateFederations(cm, config); err != nil {
glog.Errorf("Invalid configuration, ignoring update")
return
}
if err = config.Validate(); err != nil {
glog.Errorf("Invalid onfiguration: %v (value was %+v), ignoring update",
err, config)
config = nil
return
}
return
}
func (sync *kubeSync) updateFederations(cm *api.ConfigMap, config *Config) (err error) {
if flagValue, ok := cm.Data["federations"]; ok {
config.Federations = make(map[string]string)
if err = fed.ParseFederationsFlag(flagValue, config.Federations); err != nil {
glog.Errorf("Invalid federations value: %v (value was %q)",
err, cm.Data["federations"])
return
}
glog.V(2).Infof("Updated federations to %v", config.Federations)
} else {
glog.V(2).Infof("No federations present")
}
return
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。