start and stop informers
@@ -74,6 +74,8 @@ const (
type StatusManager struct {
cm *v1.ConfigMap
f *Framework
+ informerCh chan struct{}
+ informersRunning bool
}
func NewStatusManager(f *Framework) *StatusManager {
@@ -94,31 +96,43 @@ func (sm *StatusManager) init() error {
return fmt.Errorf("error creating ConfigMap: %v", err)
}
+ go func() {
+ for _ = range time.NewTicker(flushInterval).C {
+ sm.flush()
+ }
+ }()
+
+ sm.startInformer()
+ return nil
+}
+
+func (sm *StatusManager) startInformer() {
newIndexer := func() cache.Indexers {
return cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}
}
- stopCh := make(chan struct{})
+
+ sm.informerCh = make(chan struct{})
cmInformer := informerv1.NewConfigMapInformer(sm.f.Clientset, "default", cmPollInterval, newIndexer())
cmInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
UpdateFunc: func(old, cur interface{}) {
curCm := cur.(*v1.ConfigMap)
if len(curCm.Data[exitKey]) > 0 {
glog.V(2).Infof("ConfigMap was updated with exit switch at %s", curCm.Data[exitKey])
- close(stopCh)
+ close(sm.informerCh)
sm.f.shutdown(0)
}
},
})
- go cmInformer.Run(stopCh)
-
- go func() {
- for _ = range time.NewTicker(flushInterval).C {
- sm.flush()
+ glog.V(4).Info("Started informers")
+ sm.informersRunning = true
+ go cmInformer.Run(sm.informerCh)
}
- }()
- return nil
+func (sm *StatusManager) stopInformer() {
+ glog.V(4).Info("Stopped informers")
+ sm.informersRunning = false
+ close(sm.informerCh)
}
func (sm *StatusManager) shutdown() {
@@ -154,13 +168,13 @@ func (sm *StatusManager) flush() {
sm.f.lock.Lock()
defer sm.f.lock.Unlock()
- // If master is in the process upgrading, we exit early
- if sm.masterUpgrading() {
- return
- }
-
// Loop until we successfully update the config map
for {
+ // Restart ConfigMap informer if it was previously shut down
+ if sm.masterUpgraded() && !sm.informersRunning {
+ sm.startInformer()
+ }
+
updatedCm, err := sm.f.Clientset.Core().ConfigMaps("default").Get(configMapName, metav1.GetOptions{})
if err != nil {
glog.Warningf("Error getting ConfigMap: %v", err)
@@ -171,13 +185,25 @@ func (sm *StatusManager) flush() {
}
// K8s considers its version of the ConfigMap to be latest, so we must get
- // the configmap from k8s first, then merge in our data.
+ // the configmap from k8s first.
+ // We give precedence to the master-upgraded and master-upgrading flags
+ // set by the external test framework, but otherwise we prioritize
+ // Ingress statuses set by StatusManager.
for key, value := range sm.cm.Data {
+ if key != masterUpgradedKey && key != masterUpgradingKey {
updatedCm.Data[key] = value
}
+ }
sm.cm = updatedCm
sm.cm.Name = configMapName
+ // If master is in the process upgrading, we exit early and turn off the
+ // ConfigMap informer.
+ if sm.masterUpgrading() && sm.informersRunning {
+ sm.stopInformer()
+ return
+ }
+
_, err = sm.f.Clientset.Core().ConfigMaps("default").Update(sm.cm)
if err != nil {
glog.Warningf("Error updating ConfigMap: %v", err)
