github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-28 05:57:25 +00:00
Code Issues Projects Releases Wiki Activity

Batch updates of multiple Pods.

Browse Source
This commit is contained in:
Wojciech Tyczynski 2015-02-18 16:08:32 +01:00
parent e81d9abcac
commit 25c8f07c67
2 changed files with 30 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/kubelet/config/config.go
View File

@ -60,7 +60,7 @@ type PodConfig struct {
// NewPodConfig creates an object that can merge many configuration sources into a stream // NewPodConfig creates an object that can merge many configuration sources into a stream
// of normalized updates to a pod configuration. // of normalized updates to a pod configuration.
func NewPodConfig(mode PodConfigNotificationMode) *PodConfig { func NewPodConfig(mode PodConfigNotificationMode) *PodConfig {
updates := make(chan kubelet.PodUpdate, 1) updates := make(chan kubelet.PodUpdate, 50)
storage := newPodStorage(updates, mode) storage := newPodStorage(updates, mode)
podConfig := &PodConfig{ podConfig := &PodConfig{
pods: storage, pods: storage,

42
pkg/kubelet/kubelet.go
View File

@ -1389,24 +1389,25 @@ func filterHostPortConflicts(pods []api.BoundPod) []api.BoundPod {
// state every sync_frequency seconds. Never returns. // state every sync_frequency seconds. Never returns.
func (kl *Kubelet) syncLoop(updates <-chan PodUpdate, handler SyncHandler) { func (kl *Kubelet) syncLoop(updates <-chan PodUpdate, handler SyncHandler) {
for { for {
unsyncedPod := false
select { select {
case u := <-updates: case u := <-updates:
switch u.Op { kl.updatePods(u)
case SET: unsyncedPod = true
glog.V(3).Infof("SET: Containers changed")
kl.pods = u.Pods
kl.pods = filterHostPortConflicts(kl.pods)
case UPDATE:
glog.V(3).Infof("Update: Containers changed")
kl.pods = updateBoundPods(u.Pods, kl.pods)
kl.pods = filterHostPortConflicts(kl.pods)
default:
panic("syncLoop does not support incremental changes")
}
case <-time.After(kl.resyncInterval): case <-time.After(kl.resyncInterval):
glog.V(4).Infof("Periodic sync") glog.V(4).Infof("Periodic sync")
} }
// If we already caught some update, try to wait for some short time
// to possibly batch it with other incoming updates.
for ; unsyncedPod; {
select {
case u := <-updates:
kl.updatePods(u)
case <-time.After(5 * time.Millisecond):
// Break the for loop.
unsyncedPod = false
}
}
err := handler.SyncPods(kl.pods) err := handler.SyncPods(kl.pods)
if err != nil { if err != nil {
@ -1415,6 +1416,21 @@ func (kl *Kubelet) syncLoop(updates <-chan PodUpdate, handler SyncHandler) {
} }
} }
func (kl *Kubelet) updatePods(u PodUpdate) {
switch u.Op {
case SET:
glog.V(3).Infof("SET: Containers changed")
kl.pods = u.Pods
kl.pods = filterHostPortConflicts(kl.pods)
case UPDATE:
glog.V(3).Infof("Update: Containers changed")
kl.pods = updateBoundPods(u.Pods, kl.pods)
kl.pods = filterHostPortConflicts(kl.pods)
default:
panic("syncLoop does not support incremental changes")
}
}
// Returns Docker version for this Kubelet. // Returns Docker version for this Kubelet.
func (kl *Kubelet) GetDockerVersion() ([]uint, error) { func (kl *Kubelet) GetDockerVersion() ([]uint, error) {
if kl.dockerClient == nil { if kl.dockerClient == nil {