mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-26 05:03:09 +00:00
Revert "Sync all pods with store before starting RC Manager."
This reverts commit 1b93ee7b35
.
This commit is contained in:
parent
2d85c6e4b0
commit
d802778c20
@ -31,7 +31,6 @@ import (
|
|||||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||||
"k8s.io/kubernetes/pkg/client/record"
|
"k8s.io/kubernetes/pkg/client/record"
|
||||||
"k8s.io/kubernetes/pkg/controller/framework"
|
"k8s.io/kubernetes/pkg/controller/framework"
|
||||||
"k8s.io/kubernetes/pkg/fields"
|
|
||||||
"k8s.io/kubernetes/pkg/labels"
|
"k8s.io/kubernetes/pkg/labels"
|
||||||
"k8s.io/kubernetes/pkg/runtime"
|
"k8s.io/kubernetes/pkg/runtime"
|
||||||
)
|
)
|
||||||
@ -433,39 +432,6 @@ func FilterActivePods(pods []api.Pod) []*api.Pod {
|
|||||||
return result
|
return result
|
||||||
}
|
}
|
||||||
|
|
||||||
// SyncAllPodsWithStore lists all pods and inserts them into the given store.
|
|
||||||
// Though this function is written in a generic manner, it is only used by the
|
|
||||||
// controllers for a specific purpose, to synchronously populate the store
|
|
||||||
// with the first batch of pods that would otherwise be sent by the Informer.
|
|
||||||
// Doing this avoids more complicated forms of synchronization with the
|
|
||||||
// Informer, though it also means that the controller calling this function
|
|
||||||
// will receive "OnUpdate" events for all the pods in the store, instead of
|
|
||||||
// "OnAdd". This should be ok, since most controllers are level triggered
|
|
||||||
// and make decisions based on the contents of the store.
|
|
||||||
//
|
|
||||||
// TODO: Extend this logic to load arbitrary local state for the controllers
|
|
||||||
// instead of just pods.
|
|
||||||
func SyncAllPodsWithStore(kubeClient clientset.Interface, store cache.Store) {
|
|
||||||
var allPods *api.PodList
|
|
||||||
var err error
|
|
||||||
listOptions := api.ListOptions{LabelSelector: labels.Everything(), FieldSelector: fields.Everything()}
|
|
||||||
for {
|
|
||||||
if allPods, err = kubeClient.Core().Pods(api.NamespaceAll).List(listOptions); err != nil {
|
|
||||||
glog.Warningf("Retrying pod list: %v", err)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
break
|
|
||||||
}
|
|
||||||
pods := []interface{}{}
|
|
||||||
for i := range allPods.Items {
|
|
||||||
p := allPods.Items[i]
|
|
||||||
glog.V(4).Infof("Initializing store with pod %v/%v", p.Namespace, p.Name)
|
|
||||||
pods = append(pods, &p)
|
|
||||||
}
|
|
||||||
store.Replace(pods, allPods.ResourceVersion)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// ControllersByCreationTimestamp sorts a list of ReplicationControllers by creation timestamp, using their names as a tie breaker.
|
// ControllersByCreationTimestamp sorts a list of ReplicationControllers by creation timestamp, using their names as a tie breaker.
|
||||||
type ControllersByCreationTimestamp []*api.ReplicationController
|
type ControllersByCreationTimestamp []*api.ReplicationController
|
||||||
|
|
||||||
|
@ -187,7 +187,6 @@ func (rm *ReplicationManager) SetEventRecorder(recorder record.EventRecorder) {
|
|||||||
func (rm *ReplicationManager) Run(workers int, stopCh <-chan struct{}) {
|
func (rm *ReplicationManager) Run(workers int, stopCh <-chan struct{}) {
|
||||||
defer utilruntime.HandleCrash()
|
defer utilruntime.HandleCrash()
|
||||||
glog.Infof("Starting RC Manager")
|
glog.Infof("Starting RC Manager")
|
||||||
controller.SyncAllPodsWithStore(rm.kubeClient, rm.podStore.Store)
|
|
||||||
go rm.rcController.Run(stopCh)
|
go rm.rcController.Run(stopCh)
|
||||||
go rm.podController.Run(stopCh)
|
go rm.podController.Run(stopCh)
|
||||||
for i := 0; i < workers; i++ {
|
for i := 0; i < workers; i++ {
|
||||||
|
@ -892,27 +892,3 @@ func TestOverlappingRCs(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestRCManagerInit(t *testing.T) {
|
|
||||||
// Insert a stable rc into the replication manager's store and make sure
|
|
||||||
// it syncs pods with the apiserver before making any decisions.
|
|
||||||
rc := newReplicationController(2)
|
|
||||||
response := runtime.EncodeOrDie(testapi.Default.Codec(), newPodList(nil, 2, api.PodRunning, rc))
|
|
||||||
fakeHandler := utiltesting.FakeHandler{
|
|
||||||
StatusCode: 200,
|
|
||||||
ResponseBody: response,
|
|
||||||
}
|
|
||||||
testServer := httptest.NewServer(&fakeHandler)
|
|
||||||
// TODO: Uncomment when fix #19254
|
|
||||||
// defer testServer.Close()
|
|
||||||
|
|
||||||
c := clientset.NewForConfigOrDie(&client.Config{Host: testServer.URL, ContentConfig: client.ContentConfig{GroupVersion: testapi.Default.GroupVersion()}})
|
|
||||||
manager := NewReplicationManager(c, controller.NoResyncPeriodFunc, BurstReplicas)
|
|
||||||
manager.rcStore.Store.Add(rc)
|
|
||||||
manager.podStoreSynced = alwaysReady
|
|
||||||
controller.SyncAllPodsWithStore(manager.kubeClient, manager.podStore.Store)
|
|
||||||
fakePodControl := &controller.FakePodControl{}
|
|
||||||
manager.podControl = fakePodControl
|
|
||||||
manager.syncReplicationController(getKey(rc, t))
|
|
||||||
validateSyncReplication(t, fakePodControl, 0, 0)
|
|
||||||
}
|
|
||||||
|
Loading…
Reference in New Issue
Block a user