github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #5205 from fgrzadkowski/sync_pod_status 

Periodically update pod status from kubelet.
pull/6/head
Victor Marmol 2015-03-16 17:04:31 -07:00
parent fa23519387 336525a27d
commit 009737e69e
11 changed files with 116 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/integration/integration.go
View File

@ -657,7 +657,7 @@ func runServiceTest(client *client.Client) {
glog.Fatalf("Failed to create service: %v, %v", svc1, err)
}
// create an identical service in the default namespace
// create an identical service in the non-default namespace
svc3 := &api.Service{
ObjectMeta: api.ObjectMeta{Name: "service1"},
Spec: api.ServiceSpec{

1
pkg/api/validation/validation.go
View File

@ -691,6 +691,7 @@ func ValidatePodStatusUpdate(newPod, oldPod *api.Pod) errs.ValidationErrorList {
allErrs = append(allErrs, errs.NewFieldInvalid("status.host", newPod.Status.Host, "pod host cannot be changed directly"))
}
// For status update we ignore changes to pod spec.
newPod.Spec = oldPod.Spec
return allErrs

5
pkg/client/fake_pods.go
View File

@ -63,3 +63,8 @@ func (c *FakePods) Bind(bind *api.Binding) error {
c.Fake.Actions = append(c.Fake.Actions, FakeAction{Action: "bind-pod", Value: bind.Name})
return nil
}
func (c *FakePods) UpdateStatus(name string, status *api.PodStatus) (*api.Pod, error) {
c.Fake.Actions = append(c.Fake.Actions, FakeAction{Action: "update-status-pod", Value: name})
return &api.Pod{}, nil
}

21
pkg/client/pods.go
View File

@ -39,6 +39,7 @@ type PodInterface interface {
Update(pod *api.Pod) (*api.Pod, error)
Watch(label, field labels.Selector, resourceVersion string) (watch.Interface, error)
Bind(binding *api.Binding) error
UpdateStatus(name string, status *api.PodStatus) (*api.Pod, error)
}
// pods implements PodsNamespacer interface
@ -62,7 +63,7 @@ func (c *pods) List(selector labels.Selector) (result *api.PodList, err error) {
return
}
// GetPod takes the name of the pod, and returns the corresponding Pod object, and an error if it occurs
// Get takes the name of the pod, and returns the corresponding Pod object, and an error if it occurs
func (c *pods) Get(name string) (result *api.Pod, err error) {
if len(name) == 0 {
return nil, errors.New("name is required parameter to Get")
@ -73,19 +74,19 @@ func (c *pods) Get(name string) (result *api.Pod, err error) {
return
}
// DeletePod takes the name of the pod, and returns an error if one occurs
// Delete takes the name of the pod, and returns an error if one occurs
func (c *pods) Delete(name string) error {
return c.r.Delete().Namespace(c.ns).Resource("pods").Name(name).Do().Error()
}
// CreatePod takes the representation of a pod. Returns the server's representation of the pod, and an error, if it occurs.
// Create takes the representation of a pod. Returns the server's representation of the pod, and an error, if it occurs.
func (c *pods) Create(pod *api.Pod) (result *api.Pod, err error) {
result = &api.Pod{}
err = c.r.Post().Namespace(c.ns).Resource("pods").Body(pod).Do().Into(result)
return
}
// UpdatePod takes the representation of a pod to update. Returns the server's representation of the pod, and an error, if it occurs.
// Update takes the representation of a pod to update. Returns the server's representation of the pod, and an error, if it occurs.
func (c *pods) Update(pod *api.Pod) (result *api.Pod, err error) {
result = &api.Pod{}
if len(pod.ResourceVersion) == 0 {
@ -112,3 +113,15 @@ func (c *pods) Watch(label, field labels.Selector, resourceVersion string) (watc
func (c *pods) Bind(binding *api.Binding) error {
return c.r.Post().Namespace(c.ns).Resource("pods").Name(binding.Name).SubResource("binding").Body(binding).Do().Error()
}
// UpdateStatus takes the name of the pod and the new status. Returns the server's representation of the pod, and an error, if it occurs.
func (c *pods) UpdateStatus(name string, newStatus *api.PodStatus) (result *api.Pod, err error) {
result = &api.Pod{}
pod, err := c.Get(name)
if err != nil {
return
}
pod.Status = *newStatus
err = c.r.Put().Namespace(c.ns).Resource("pods").Name(pod.Name).SubResource("status").Body(pod).Do().Into(result)
return
}

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

@ -31,7 +31,7 @@ func NewSourceApiserver(client *client.Client, hostname string, updates chan<- i
newSourceApiserverFromLW(lw, updates)
}
// newSourceApiserverFromLW holds creates a config source that watches an pulls from the apiserver.
// newSourceApiserverFromLW holds creates a config source that watches and pulls from the apiserver.
func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{}) {
send := func(objs []interface{}) {
var pods []api.Pod

1
pkg/kubelet/dockertools/fake_docker_client.go
View File

@ -126,6 +126,7 @@ func (f *FakeDockerClient) StartContainer(id string, hostConfig *docker.HostConf
Running: true,
Pid: 42,
},
NetworkSettings: &docker.NetworkSettings{IPAddress: "1.2.3.4"},
}
return f.Err
}

67
pkg/kubelet/kubelet.go
View File

@ -285,7 +285,7 @@ type Kubelet struct {
// the EventRecorder to use
recorder record.EventRecorder
// A pod status cache currently used to store rejected pods and their statuses.
// A pod status cache stores statuses for pods (both rejected and synced).
podStatusesLock sync.RWMutex
podStatuses map[string]api.PodStatus
}
@ -568,6 +568,7 @@ func (kl *Kubelet) Run(updates <-chan PodUpdate) {
glog.Warning("No api server defined - no node status update will be sent.")
}
go kl.syncNodeStatus()
go util.Forever(kl.syncStatus, kl.resyncInterval)
kl.syncLoop(updates, kl)
}
@ -1346,6 +1347,17 @@ func (kl *Kubelet) computePodContainerChanges(pod *api.Pod, containersInPod dock
func (kl *Kubelet) syncPod(pod *api.Pod, containersInPod dockertools.DockerContainers) error {
podFullName := GetPodFullName(pod)
uid := pod.UID
// Before returning, regenerate status and store it in the cache.
defer func() {
status, err := kl.generatePodStatus(podFullName, uid)
if err != nil {
glog.Errorf("Unable to generate status for pod with name %q and uid %q info with error(%v)", podFullName, uid, err)
} else {
kl.setPodStatusInCache(podFullName, status)
}
}()
containerChanges, err := kl.computePodContainerChanges(pod, containersInPod)
glog.V(3).Infof("Got container changes for pod %q: %+v", podFullName, containerChanges)
if err != nil {
@ -1715,6 +1727,40 @@ func (kl *Kubelet) syncLoop(updates <-chan PodUpdate, handler SyncHandler) {
}
}
// syncStatus syncs pods statuses with the apiserver.
func (kl *Kubelet) syncStatus() {
glog.V(3).Infof("Syncing pods status")
statuses := make(map[string]api.PodStatus)
func() {
kl.podLock.Lock()
defer kl.podLock.Unlock()
for _, pod := range kl.pods {
source := pod.Annotations[ConfigSourceAnnotationKey]
if source != ApiserverSource {
glog.V(3).Infof("Pod status for %q is not updated due to its source %s", pod.Name, source)
continue
}
status, err := kl.GetPodStatus(GetPodFullName(&pod), pod.UID)
if err != nil {
glog.Warningf("Error getting pod %q status: %v, retry later", pod.Name, err)
continue
}
statuses[GetPodFullName(&pod)] = status
}
}()
for podFullName, status := range statuses {
name, namespace := ParsePodFullName(podFullName)
pod, err := kl.kubeClient.Pods(namespace).UpdateStatus(name, &status)
if err != nil {
glog.Warningf("Error updating status for pod %s: %v (full pod: %s)", name, err, pod)
} else {
glog.V(3).Infof("Status for pod %q updated successfully: %s", name, pod)
}
}
}
// Updated the Kubelet's internal pods with those provided by the update.
// Records new and updated pods in newPods and updatedPods.
func (kl *Kubelet) updatePods(u PodUpdate, podSyncTypes map[types.UID]metrics.SyncPodType) {
@ -2000,19 +2046,23 @@ func (kl *Kubelet) GetPodByFullName(podFullName string) (*api.PodSpec, bool) {
// GetPodStatus returns information from Docker about the containers in a pod
func (kl *Kubelet) GetPodStatus(podFullName string, uid types.UID) (api.PodStatus, error) {
// Check to see if we have a cached version of the status.
cachedPodStatus, found := kl.getPodStatusFromCache(podFullName)
if found {
glog.V(3).Infof("Returning cached status for %s", podFullName)
return cachedPodStatus, nil
}
return kl.generatePodStatus(podFullName, uid)
}
func (kl *Kubelet) generatePodStatus(podFullName string, uid types.UID) (api.PodStatus, error) {
glog.V(3).Infof("Generating status for %s", podFullName)
var podStatus api.PodStatus
spec, found := kl.GetPodByFullName(podFullName)
if !found {
return podStatus, fmt.Errorf("Couldn't find spec for pod %s", podFullName)
}
// Check to see if the pod has been rejected.
mappedPodStatus, ok := kl.getPodStatusFromCache(podFullName)
if ok {
return mappedPodStatus, nil
}
info, err := dockertools.GetDockerPodInfo(kl.dockerClient, *spec, podFullName, uid)
if err != nil {
@ -2042,6 +2092,7 @@ func (kl *Kubelet) GetPodStatus(podFullName string, uid types.UID) (api.PodStatu
if found {
podStatus.PodIP = netContainerInfo.PodIP
}
podStatus.Host = kl.hostname
return podStatus, nil
}

22
pkg/kubelet/kubelet_test.go
View File

@ -447,7 +447,7 @@ func TestSyncPodsDoesNothing(t *testing.T) {
t.Errorf("unexpected error: %v", err)
}
waitGroup.Wait()
verifyCalls(t, fakeDocker, []string{"list", "list", "list", "inspect_container", "inspect_container"})
verifyCalls(t, fakeDocker, []string{"list", "list", "list", "inspect_container", "inspect_container", "list", "inspect_container", "inspect_container"})
}
func TestSyncPodsWithTerminationLog(t *testing.T) {
@ -481,7 +481,7 @@ func TestSyncPodsWithTerminationLog(t *testing.T) {
}
waitGroup.Wait()
verifyCalls(t, fakeDocker, []string{
"list", "list", "list", "create", "start", "inspect_container", "create", "start"})
"list", "list", "list", "create", "start", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
fakeDocker.Lock()
parts := strings.Split(fakeDocker.Container.HostConfig.Binds[0], ":")
@ -531,7 +531,7 @@ func TestSyncPodsCreatesNetAndContainer(t *testing.T) {
waitGroup.Wait()
verifyCalls(t, fakeDocker, []string{
"list", "list", "list", "create", "start", "inspect_container", "create", "start"})
"list", "list", "list", "create", "start", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
fakeDocker.Lock()
@ -584,7 +584,7 @@ func TestSyncPodsCreatesNetAndContainerPullsImage(t *testing.T) {
waitGroup.Wait()
verifyCalls(t, fakeDocker, []string{
"list", "list", "list", "create", "start", "inspect_container", "create", "start"})
"list", "list", "list", "create", "start", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
fakeDocker.Lock()
@ -634,7 +634,7 @@ func TestSyncPodsWithPodInfraCreatesContainer(t *testing.T) {
waitGroup.Wait()
verifyCalls(t, fakeDocker, []string{
"list", "list", "list", "inspect_container", "inspect_image", "list", "create", "start"})
"list", "list", "list", "inspect_container", "inspect_image", "list", "create", "start", "list", "inspect_container", "inspect_container"})
fakeDocker.Lock()
if len(fakeDocker.Created) != 1 ||
@ -691,7 +691,7 @@ func TestSyncPodsWithPodInfraCreatesContainerCallsHandler(t *testing.T) {
waitGroup.Wait()
verifyCalls(t, fakeDocker, []string{
"list", "list", "list", "inspect_container", "inspect_image", "list", "create", "start"})
"list", "list", "list", "inspect_container", "inspect_image", "list", "create", "start", "list", "inspect_container", "inspect_container"})
fakeDocker.Lock()
if len(fakeDocker.Created) != 1 ||
@ -760,7 +760,7 @@ func TestSyncPodsDeletesWithNoPodInfraContainer(t *testing.T) {
waitGroup.Wait()
verifyCalls(t, fakeDocker, []string{
"list", "list", "list", "list", "inspect_container", "inspect_container", "stop", "create", "start", "inspect_container", "create", "start"})
"list", "list", "list", "list", "inspect_container", "inspect_container", "list", "inspect_container", "inspect_container", "stop", "create", "start", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
// A map iteration is used to delete containers, so must not depend on
// order here.
@ -898,7 +898,7 @@ func TestSyncPodDeletesDuplicate(t *testing.T) {
t.Errorf("unexpected error: %v", err)
}
verifyCalls(t, fakeDocker, []string{"list", "stop"})
verifyCalls(t, fakeDocker, []string{"list", "stop", "list"})
// Expect one of the duplicates to be killed.
if len(fakeDocker.Stopped) != 1 || (fakeDocker.Stopped[0] != "1234" && fakeDocker.Stopped[0] != "4567") {
t.Errorf("Wrong containers were stopped: %v", fakeDocker.Stopped)
@ -940,7 +940,7 @@ func TestSyncPodBadHash(t *testing.T) {
}
//verifyCalls(t, fakeDocker, []string{"list", "stop", "list", "create", "start", "stop", "create", "start", "inspect_container"})
verifyCalls(t, fakeDocker, []string{"list", "stop", "stop", "create", "start", "inspect_container", "create", "start"})
verifyCalls(t, fakeDocker, []string{"list", "stop", "stop", "create", "start", "inspect_container", "create", "start", "list", "inspect_container", "inspect_container"})
// A map interation is used to delete containers, so must not depend on
// order here.
@ -993,7 +993,7 @@ func TestSyncPodUnhealthy(t *testing.T) {
t.Errorf("unexpected error: %v", err)
}
verifyCalls(t, fakeDocker, []string{"list", "stop", "create", "start"})
verifyCalls(t, fakeDocker, []string{"list", "stop", "create", "start", "list", "inspect_container"})
// A map interation is used to delete containers, so must not depend on
// order here.
@ -1683,7 +1683,7 @@ func TestSyncPodEventHandlerFails(t *testing.T) {
t.Errorf("unexpected error: %v", err)
}
verifyCalls(t, fakeDocker, []string{"list", "list", "create", "start", "stop"})
verifyCalls(t, fakeDocker, []string{"list", "list", "create", "start", "stop", "list"})
if len(fakeDocker.Stopped) != 1 {
t.Errorf("Wrong containers were stopped: %v", fakeDocker.Stopped)

2
pkg/kubelet/pod_workers.go
View File

@ -31,7 +31,7 @@ import (
type syncPodFnType func(*api.Pod, dockertools.DockerContainers) error
type podWorkers struct {
// Protects podUpdates field.
// Protects all per worker fields.
podLock sync.Mutex
// Tracks all running per-pod goroutines - per-pod goroutine will be

8
pkg/kubelet/types.go
View File

@ -18,6 +18,7 @@ package kubelet
import (
"fmt"
"strings"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
)
@ -68,12 +69,19 @@ type PodUpdate struct {
}
// GetPodFullName returns a name that uniquely identifies a pod across all config sources.
// NOTE: If changed ParsePodFullName must be also updated.
func GetPodFullName(pod *api.Pod) string {
// Use underscore as the delimiter because it is not allowed in pod name
// (DNS subdomain format), while allowed in the container name format.
return fmt.Sprintf("%s_%s", pod.Name, pod.Namespace)
}
// ParsePodFullName parses full name generated by GetPodFullName and returns parts of it.
func ParsePodFullName(podFullName string) (name, namespace string) {
nameParts := strings.Split(podFullName, "_")
return nameParts[0], nameParts[1]
}
// Build the pod full name from pod name and namespace.
func BuildPodFullName(name, namespace string) string {
return name + "_" + namespace

23
pkg/master/master.go
View File

@ -371,20 +371,19 @@ func (m *Master) init(c *Config) {
m.nodeRegistry = registry
nodeStorage := minion.NewREST(m.nodeRegistry)
// TODO: unify the storage -> registry and storage -> client patterns
nodeStorageClient := RESTStorageToNodes(nodeStorage)
podCache := NewPodCache(
c.KubeletClient,
nodeStorageClient.Nodes(),
podRegistry,
)
if c.SyncPodStatus {
go util.Forever(func() { podCache.UpdateAllContainers() }, m.cacheTimeout)
}
go util.Forever(func() { podCache.GarbageCollectPodStatus() }, time.Minute*30)
// TODO: unify the storage -> registry and storage -> client patterns
nodeStorageClient := RESTStorageToNodes(nodeStorage)
// TODO: refactor podCache to sit on top of podStorage via status calls
podStorage = podStorage.WithPodStatus(podCache)
podCache := NewPodCache(
c.KubeletClient,
nodeStorageClient.Nodes(),
podRegistry,
)
go util.Forever(func() { podCache.UpdateAllContainers() }, m.cacheTimeout)
go util.Forever(func() { podCache.GarbageCollectPodStatus() }, time.Minute*30)
podStorage = podStorage.WithPodStatus(podCache)
}
// TODO: Factor out the core API registration
m.storage = map[string]apiserver.RESTStorage{