Skip to content

K
k3s
Commit 1f50bd22 authored by Saad Ali's avatar Saad Ali
Browse files
Options

Merge pull request #5854 from fgrzadkowski/delete_pod_cache

Delete pod_cache and rely on updating pod status by kublet.
parents 50452994 74da3b14
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 44 additions and 640 deletions
cmd/integration/integration.go
View file @ 1f50bd22
......@@ -195,7 +195,6 @@ func startComponents(firstManifestURL, secondManifestURL, apiVersion string) (st
ReadOnlyPort: portNumber,
PublicAddress: publicAddress,
CacheTimeout: 2 * time.Second,
SyncPodStatus: true,
EnableV1Beta3: true,
})
handler.delegate = m.Handler
......
cmd/kube-apiserver/app/server.go
View file @ 1f50bd22
......@@ -73,7 +73,6 @@ type APIServer struct {
RuntimeConfig util.ConfigurationMap
KubeletConfig client.KubeletConfig
ClusterName string
SyncPodStatus bool
EnableProfiling bool
}
......@@ -94,7 +93,6 @@ func NewAPIServer() *APIServer {
EnableLogsSupport: true,
MasterServiceNamespace: api.NamespaceDefault,
ClusterName: "kubernetes",
SyncPodStatus: true,
RuntimeConfig: make(util.ConfigurationMap),
KubeletConfig: client.KubeletConfig{
......@@ -149,7 +147,6 @@ func (s *APIServer) AddFlags(fs *pflag.FlagSet) {
fs.BoolVar(&s.AllowPrivileged, "allow_privileged", s.AllowPrivileged, "If true, allow privileged containers.")
fs.Var(&s.PortalNet, "portal_net", "A CIDR notation IP range from which to assign portal IPs. This must not overlap with any IP ranges assigned to nodes for pods.")
fs.StringVar(&s.MasterServiceNamespace, "master_service_namespace", s.MasterServiceNamespace, "The namespace from which the kubernetes master services should be injected into pods")
fs.BoolVar(&s.SyncPodStatus, "sync_pod_status", s.SyncPodStatus, "If true, periodically fetch pods statuses from kubelets.")
fs.Var(&s.RuntimeConfig, "runtime_config", "A set of key=value pairs that describe runtime configuration that may be passed to the apiserver.")
client.BindKubeletClientConfigFlags(fs, &s.KubeletConfig)
fs.StringVar(&s.ClusterName, "cluster_name", s.ClusterName, "The instance prefix for the cluster")
......@@ -250,7 +247,6 @@ func (s *APIServer) Run(_ []string) error {
EnableV1Beta3: v1beta3,
MasterServiceNamespace: s.MasterServiceNamespace,
ClusterName: s.ClusterName,
SyncPodStatus: s.SyncPodStatus,
}
m := master.New(config)
......
pkg/kubelet/kubelet.go
View file @ 1f50bd22
......@@ -1423,6 +1423,11 @@ func (kl *Kubelet) syncPod(pod *api.Pod, hasMirrorPod bool, runningPod kubeconta
if !hasMirrorPod && isStaticPod(pod) {
glog.V(4).Infof("Creating a mirror pod %q", podFullName)
// To make sure we will properly update static pod status we need to delete
// it from status manager. Otherwise it is possible that we will miss manual
// deletion of mirror pod in apiserver and will never reset its status to
// Running after recreating it.
kl.statusManager.DeletePodStatus(podFullName)
if err := kl.podManager.CreateMirrorPod(*pod, kl.hostname); err != nil {
glog.Errorf("Failed creating a mirror pod %q: %#v", podFullName, err)
}
......@@ -1800,6 +1805,29 @@ func (kl *Kubelet) GetHostname() string {
return kl.hostname
}
// Returns host IP or nil in case of error.
func (kl *Kubelet) GetHostIP() (net.IP, error) {
node, err := kl.GetNode()
if err != nil {
return nil, fmt.Errorf("Cannot get node: %v", err)
}
addresses := node.Status.Addresses
addressMap := make(map[api.NodeAddressType][]api.NodeAddress)
for i := range addresses {
addressMap[addresses[i].Type] = append(addressMap[addresses[i].Type], addresses[i])
}
if addresses, ok := addressMap[api.NodeLegacyHostIP]; ok {
return net.ParseIP(addresses[0].Address), nil
}
if addresses, ok := addressMap[api.NodeInternalIP]; ok {
return net.ParseIP(addresses[0].Address), nil
}
if addresses, ok := addressMap[api.NodeExternalIP]; ok {
return net.ParseIP(addresses[0].Address), nil
}
return nil, fmt.Errorf("Host IP unknown; known addresses: %v", addresses)
}
// GetPods returns all pods bound to the kubelet and their spec, and the mirror
// pods.
func (kl *Kubelet) GetPods() []api.Pod {
......@@ -2012,7 +2040,12 @@ func (kl *Kubelet) generatePodStatusByPod(pod *api.Pod) (api.PodStatus, error) {
podStatus.Info[c.Name] = containerStatus
}
podStatus.Conditions = append(podStatus.Conditions, getPodReadyCondition(spec, podStatus.Info)...)
podStatus.Host = kl.hostname
podStatus.Host = kl.GetHostname()
hostIP, err := kl.GetHostIP()
if err != nil {
return api.PodStatus{}, fmt.Errorf("Cannot get host IP: %v", err)
}
podStatus.HostIP = hostIP.String()
return *podStatus, nil
}
......
pkg/master/master.go
View file @ 1f50bd22
......@@ -115,9 +115,6 @@ type Config struct {
// The name of the cluster.
ClusterName string
// If true we will periodically probe pods statuses.
SyncPodStatus bool
}
// Master contains state for a Kubernetes cluster master/api server.
......@@ -372,20 +369,6 @@ func (m *Master) init(c *Config) {
m.nodeRegistry = registry
nodeStorage := minion.NewStorage(m.nodeRegistry, c.KubeletClient)
// 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(podCache.UpdateAllContainers, m.cacheTimeout)
go util.Forever(podCache.GarbageCollectPodStatus, time.Minute*30)
// Note the pod cache needs access to an un-decorated RESTStorage
podStorage = podStorage.WithPodStatus(podCache)
}
controllerStorage := controlleretcd.NewREST(c.EtcdHelper)
......
pkg/master/pod_cache.go deleted 100644 → 0
View file @ 50452994
/*
Copyright 2014 Google Inc. All rights reserved.
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 master
import (
"sync"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/pod"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/golang/glog"
)
// PodCache contains both a cache of container information, as well as the mechanism for keeping
// that cache up to date.
type PodCache struct {
containerInfo client.PodInfoGetter
pods pod.Registry
// For confirming existance of a node
nodes client.NodeInterface
// lock protects access to all fields below
lock sync.Mutex
// cached pod statuses.
podStatus map[objKey]api.PodStatus
// nodes that we know exist. Cleared at the beginning of each
// UpdateAllPods call.
currentNodes map[objKey]api.NodeStatus
}
type objKey struct {
namespace, name string
}
// NewPodCache returns a new PodCache which watches container information
// registered in the given PodRegistry.
// TODO(lavalamp): pods should be a client.PodInterface.
func NewPodCache(info client.PodInfoGetter, nodes client.NodeInterface, pods pod.Registry) *PodCache {
return &PodCache{
containerInfo: info,
pods: pods,
nodes: nodes,
currentNodes: map[objKey]api.NodeStatus{},
podStatus: map[objKey]api.PodStatus{},
}
}
// GetPodStatus gets the stored pod status.
func (p *PodCache) GetPodStatus(namespace, name string) (*api.PodStatus, error) {
status := p.getPodStatusInternal(namespace, name)
if status != nil {
return status, nil
}
return p.updateCacheAndReturn(namespace, name)
}
func (p *PodCache) updateCacheAndReturn(namespace, name string) (*api.PodStatus, error) {
pod, err := p.pods.GetPod(api.WithNamespace(api.NewContext(), namespace), name)
if err != nil {
return nil, err
}
if err := p.updatePodStatus(pod); err != nil {
return nil, err
}
status := p.getPodStatusInternal(namespace, name)
if status == nil {
glog.Warningf("nil status after successful update. that's odd... (%s %s)", namespace, name)
return nil, client.ErrPodInfoNotAvailable
}
return status, nil
}
func (p *PodCache) getPodStatusInternal(namespace, name string) *api.PodStatus {
p.lock.Lock()
defer p.lock.Unlock()
value, ok := p.podStatus[objKey{namespace, name}]
if !ok {
return nil
}
// Make a copy
return &value
}
func (p *PodCache) ClearPodStatus(namespace, name string) {
p.lock.Lock()
defer p.lock.Unlock()
delete(p.podStatus, objKey{namespace, name})
}
func (p *PodCache) getNodeStatusInCache(name string) (*api.NodeStatus, bool) {
p.lock.Lock()
defer p.lock.Unlock()
nodeStatus, cacheHit := p.currentNodes[objKey{"", name}]
return &nodeStatus, cacheHit
}
// lock must *not* be held
func (p *PodCache) getNodeStatus(name string) (*api.NodeStatus, error) {
nodeStatus, cacheHit := p.getNodeStatusInCache(name)
if cacheHit {
return nodeStatus, nil
}
// TODO: suppose there's N concurrent requests for node "foo"; in that case
// it might be useful to block all of them and only look up "foo" once.
// (This code will make up to N lookups.) One way of doing that would be to
// have a pool of M mutexes and require that before looking up "foo" you must
// lock mutex hash("foo") % M.
node, err := p.nodes.Get(name)
if err != nil {
glog.Errorf("Unexpected error verifying node existence: %+v", err)
return nil, err
}
p.lock.Lock()
defer p.lock.Unlock()
p.currentNodes[objKey{"", name}] = node.Status
return &node.Status, nil
}
func (p *PodCache) clearNodeStatus() {
p.lock.Lock()
defer p.lock.Unlock()
p.currentNodes = map[objKey]api.NodeStatus{}
}
func (p *PodCache) getHostAddress(addresses []api.NodeAddress) string {
addressMap := make(map[api.NodeAddressType][]api.NodeAddress)
for i := range addresses {
addressMap[addresses[i].Type] = append(addressMap[addresses[i].Type], addresses[i])
}
if addresses, ok := addressMap[api.NodeLegacyHostIP]; ok {
return addresses[0].Address
}
if addresses, ok := addressMap[api.NodeInternalIP]; ok {
return addresses[0].Address
}
if addresses, ok := addressMap[api.NodeExternalIP]; ok {
return addresses[0].Address
}
return ""
}
// TODO: once Host gets moved to spec, this can take a podSpec + metadata instead of an
// entire pod?
func (p *PodCache) updatePodStatus(pod *api.Pod) error {
newStatus, err := p.computePodStatus(pod)
p.lock.Lock()
defer p.lock.Unlock()
// Map accesses must be locked.
if err == nil {
p.podStatus[objKey{pod.Namespace, pod.Name}] = newStatus
}
return err
}
// computePodStatus always returns a new status, even if it also returns a non-nil error.
// TODO: once Host gets moved to spec, this can take a podSpec + metadata instead of an
// entire pod?
func (p *PodCache) computePodStatus(pod *api.Pod) (api.PodStatus, error) {
newStatus := pod.Status
if pod.Status.Host == "" {
// Not assigned.
newStatus.Phase = api.PodPending
newStatus.Conditions = append(newStatus.Conditions, pod.Status.Conditions...)
return newStatus, nil
}
nodeStatus, err := p.getNodeStatus(pod.Status.Host)
// Assigned to non-existing node.
if err != nil || len(nodeStatus.Conditions) == 0 {
glog.V(5).Infof("node doesn't exist: %v %v, setting pod %q status to unknown", err, nodeStatus, pod.Name)
newStatus.Phase = api.PodUnknown
newStatus.Conditions = append(newStatus.Conditions, pod.Status.Conditions...)
return newStatus, nil
}
// Assigned to an unhealthy node.
for _, condition := range nodeStatus.Conditions {
if (condition.Type == api.NodeReady || condition.Type == api.NodeReachable) && condition.Status == api.ConditionFalse {
glog.V(5).Infof("node status: %v, setting pod %q status to unknown", condition, pod.Name)
newStatus.Phase = api.PodUnknown
newStatus.Conditions = append(newStatus.Conditions, pod.Status.Conditions...)
return newStatus, nil
}
}
result, err := p.containerInfo.GetPodStatus(pod.Status.Host, pod.Namespace, pod.Name)
if err != nil {
glog.V(5).Infof("error getting pod %s status: %v, retry later", pod.Name, err)
} else {
newStatus.HostIP = p.getHostAddress(nodeStatus.Addresses)
newStatus.Info = result.Status.Info
newStatus.PodIP = result.Status.PodIP
newStatus.Phase = result.Status.Phase
newStatus.Conditions = result.Status.Conditions
}
return newStatus, err
}
func (p *PodCache) GarbageCollectPodStatus() {
pods, err := p.pods.ListPods(api.NewContext(), labels.Everything())
if err != nil {
glog.Errorf("Error getting pod list: %v", err)
return
}
keys := map[objKey]bool{}
for _, pod := range pods.Items {
keys[objKey{pod.Namespace, pod.Name}] = true
}
p.lock.Lock()
defer p.lock.Unlock()
for key := range p.podStatus {
if _, found := keys[key]; !found {
glog.Infof("Deleting orphaned cache entry: %v", key)
delete(p.podStatus, key)
}
}
}
// UpdateAllContainers updates information about all containers.
// Callers should let one call to UpdateAllContainers finish before
// calling again, or risk having new info getting clobbered by delayed
// old info.
func (p *PodCache) UpdateAllContainers() {
// TODO: this is silly, we should pro-actively update the pod status when
// the API server makes changes.
p.clearNodeStatus()
ctx := api.NewContext()
pods, err := p.pods.ListPods(ctx, labels.Everything())
if err != nil {
glog.Errorf("Error getting pod list: %v", err)
return
}
// TODO: this algorithm is 1 goroutine & RPC per pod. With a little work,
// it should be possible to make it 1 per *node*, which will be important
// at very large scales. (To be clear, the goroutines shouldn't matter--
// it's the RPCs that need to be minimized.)
var wg sync.WaitGroup
for i := range pods.Items {
pod := &pods.Items[i]
wg.Add(1)
go func() {
defer util.HandleCrash()
defer wg.Done()
err := p.updatePodStatus(pod)
if err != nil && err != client.ErrPodInfoNotAvailable {
glog.Errorf("Error getting info for pod %v/%v: %v", pod.Namespace, pod.Name, err)
}
}()
}
wg.Wait()
}
pkg/master/rest_to_nodes.go deleted 100644 → 0
View file @ 50452994
/*
Copyright 2014 Google Inc. All rights reserved.
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 master
import (
"errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/rest"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
// RESTStorageToNodes will take a RESTStorage object and return a client interface
// which will work for any use expecting a client.Nodes() interface. The advantage
// of doing this in server code is that it doesn't require an actual trip through
// the network.
//
// TODO: considering that the only difference between the various client types
// and RESTStorage type is the type of the arguments, maybe use "go generate" to
// write a specialized adaptor for every client type?
//
// TODO: this also means that pod and node API endpoints have to be colocated in the same
// process
func RESTStorageToNodes(storage rest.Storage) client.NodesInterface {
return &nodeAdaptor{storage}
}
type nodeAdaptor struct {
storage rest.Storage
}
func (n *nodeAdaptor) Nodes() client.NodeInterface {
return n
}
// Create creates a new node.
func (n *nodeAdaptor) Create(minion *api.Node) (*api.Node, error) {
return nil, errors.New("direct creation not implemented")
// TODO: apiserver should expose newOperation to make this easier.
// the actual code that should go here would start like this:
//
// ctx := api.NewDefaultContext()
// out, err := n.storage.Create(ctx, minion)
// if err != nil {
// return nil, err
// }
}
// List lists all the nodes in the cluster.
func (n *nodeAdaptor) List() (*api.NodeList, error) {
ctx := api.NewContext()
obj, err := n.storage.(rest.Lister).List(ctx, labels.Everything(), fields.Everything())
if err != nil {
return nil, err
}
return obj.(*api.NodeList), nil
}
// Get gets an existing node.
func (n *nodeAdaptor) Get(name string) (*api.Node, error) {
ctx := api.NewContext()
obj, err := n.storage.(rest.Getter).Get(ctx, name)
if err != nil {
return nil, err
}
return obj.(*api.Node), nil
}
// Delete deletes an existing node.
// TODO: implement
func (n *nodeAdaptor) Delete(name string) error {
return errors.New("direct deletion not implemented")
}
// Update updates an existing node.
func (n *nodeAdaptor) Update(minion *api.Node) (*api.Node, error) {
return nil, errors.New("direct update not implemented")
}
// Watch watches for nodes.
func (n *nodeAdaptor) Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
return nil, errors.New("direct watch not implemented")
}
pkg/registry/pod/etcd/etcd.go
View file @ 1f50bd22
......@@ -85,15 +85,6 @@ func (r *REST) ResourceLocation(ctx api.Context, name string) (*url.URL, http.Ro
return pod.ResourceLocation(r, ctx, name)
}
// WithPodStatus returns a rest object that decorates returned responses with extra
// status information.
func (r *REST) WithPodStatus(cache pod.PodStatusGetter) *REST {
store := *r
store.Decorator = pod.PodStatusDecorator(cache)
store.AfterDelete = rest.AllFuncs(store.AfterDelete, pod.PodStatusReset(cache))
return &store
}
// BindingREST implements the REST endpoint for binding pods to nodes when etcd is in use.
type BindingREST struct {
store *etcdgeneric.Etcd
......
pkg/registry/pod/rest.go
View file @ 1f50bd22
......@@ -25,7 +25,6 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/rest"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
......@@ -92,40 +91,6 @@ func (podStatusStrategy) ValidateUpdate(obj, old runtime.Object) fielderrors.Val
return validation.ValidatePodStatusUpdate(obj.(*api.Pod), old.(*api.Pod))
}
// PodStatusGetter is an interface used by Pods to fetch and retrieve status info.
type PodStatusGetter interface {
GetPodStatus(namespace, name string) (*api.PodStatus, error)
ClearPodStatus(namespace, name string)
}
// PodStatusDecorator returns a function that updates pod.Status based
// on the provided pod cache.
func PodStatusDecorator(cache PodStatusGetter) rest.ObjectFunc {
return func(obj runtime.Object) error {
pod := obj.(*api.Pod)
host := pod.Status.Host
if status, err := cache.GetPodStatus(pod.Namespace, pod.Name); err != nil {
pod.Status = api.PodStatus{
Phase: api.PodUnknown,
}
} else {
pod.Status = *status
}
pod.Status.Host = host
return nil
}
}
// PodStatusReset returns a function that clears the pod cache when the object
// is deleted.
func PodStatusReset(cache PodStatusGetter) rest.ObjectFunc {
return func(obj runtime.Object) error {
pod := obj.(*api.Pod)
cache.ClearPodStatus(pod.Namespace, pod.Name)
return nil
}
}
// MatchPod returns a generic matcher for a given label and field selector.
func MatchPod(label labels.Selector, field fields.Selector) generic.Matcher {
return generic.MatcherFunc(func(obj runtime.Object) (bool, error) {
......
pkg/registry/pod/rest_test.go deleted 100644 → 0
View file @ 50452994
/*
Copyright 2014 Google Inc. All rights reserved.
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 pod
import (
"fmt"
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
)
type fakeCache struct {
requestedNamespace string
requestedName string
clearedNamespace string
clearedName string
statusToReturn *api.PodStatus
errorToReturn error
}
func (f *fakeCache) GetPodStatus(namespace, name string) (*api.PodStatus, error) {
f.requestedNamespace = namespace
f.requestedName = name
return f.statusToReturn, f.errorToReturn
}
func (f *fakeCache) ClearPodStatus(namespace, name string) {
f.clearedNamespace = namespace
f.clearedName = name
}
func TestPodStatusDecorator(t *testing.T) {
cache := &fakeCache{statusToReturn: &api.PodStatus{Phase: api.PodRunning}}
pod := &api.Pod{}
if err := PodStatusDecorator(cache)(pod); err != nil {
t.Fatalf("unexpected error: %v", err)
}
if pod.Status.Phase != api.PodRunning {
t.Errorf("unexpected pod: %#v", pod)
}
pod = &api.Pod{
Status: api.PodStatus{
Host: "foo",
},
}
if err := PodStatusDecorator(cache)(pod); err != nil {
t.Fatalf("unexpected error: %v", err)
}
if pod.Status.Phase != api.PodRunning || pod.Status.Host != "foo" {
t.Errorf("unexpected pod: %#v", pod)
}
}
func TestPodStatusDecoratorError(t *testing.T) {
cache := &fakeCache{errorToReturn: fmt.Errorf("test error")}
pod := &api.Pod{}
if err := PodStatusDecorator(cache)(pod); err != nil {
t.Fatalf("unexpected error: %v", err)
}
if pod.Status.Phase != api.PodUnknown {
t.Errorf("unexpected pod: %#v", pod)
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment