github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-03 09:22:44 +00:00
Code Issues Projects Releases Wiki Activity

kubelet: move common types to kubelet/types

Browse Source 
This would faciliate tasks such as moving code in pkg/kubelet to sub packages.
This commit is contained in:
Yu-Ju Hong 2015-10-06 17:44:08 -07:00
parent 27ff98c8f3
commit 098ab05997
28 changed files with 246 additions and 235 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/integration/integration.go
View File

@ -48,10 +48,10 @@ import (
"k8s.io/kubernetes/pkg/controller/node" "k8s.io/kubernetes/pkg/controller/node"
replicationControllerPkg "k8s.io/kubernetes/pkg/controller/replication" replicationControllerPkg "k8s.io/kubernetes/pkg/controller/replication"
"k8s.io/kubernetes/pkg/fields" "k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/kubelet"
"k8s.io/kubernetes/pkg/kubelet/cadvisor" "k8s.io/kubernetes/pkg/kubelet/cadvisor"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/labels" "k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/master" "k8s.io/kubernetes/pkg/master"
"k8s.io/kubernetes/pkg/tools/etcdtest" "k8s.io/kubernetes/pkg/tools/etcdtest"
@ -415,7 +415,7 @@ containers:
// Wait for the mirror pod to be created. // Wait for the mirror pod to be created.
podName := fmt.Sprintf("%s-localhost", desc) podName := fmt.Sprintf("%s-localhost", desc)
namespace := kubelet.NamespaceDefault namespace := kubeletTypes.NamespaceDefault
if err := wait.Poll(time.Second, longTestTimeout, if err := wait.Poll(time.Second, longTestTimeout,
podRunning(c, namespace, podName)); err != nil { podRunning(c, namespace, podName)); err != nil {
if pods, err := c.Pods(namespace).List(labels.Everything(), fields.Everything()); err == nil { if pods, err := c.Pods(namespace).List(labels.Everything(), fields.Everything()); err == nil {

23
cmd/kubelet/app/server.go
View File

@ -49,6 +49,7 @@ import (
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
"k8s.io/kubernetes/pkg/kubelet/network" "k8s.io/kubernetes/pkg/kubelet/network"
"k8s.io/kubernetes/pkg/kubelet/qos" "k8s.io/kubernetes/pkg/kubelet/qos"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/master/ports" "k8s.io/kubernetes/pkg/master/ports"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
"k8s.io/kubernetes/pkg/util/io" "k8s.io/kubernetes/pkg/util/io"
@ -148,8 +149,8 @@ type KubeletBootstrap interface {
StartGarbageCollection() StartGarbageCollection()
ListenAndServe(net.IP, uint, *kubelet.TLSOptions, bool) ListenAndServe(net.IP, uint, *kubelet.TLSOptions, bool)
ListenAndServeReadOnly(net.IP, uint) ListenAndServeReadOnly(net.IP, uint)
Run(<-chan kubelet.PodUpdate) Run(<-chan kubeletTypes.PodUpdate)
RunOnce(<-chan kubelet.PodUpdate) ([]kubelet.RunPodResult, error) RunOnce(<-chan kubeletTypes.PodUpdate) ([]kubelet.RunPodResult, error)
} }
// create and initialize a Kubelet instance // create and initialize a Kubelet instance
@ -173,9 +174,9 @@ func NewKubeletServer() *KubeletServer {
FileCheckFrequency: 20 * time.Second, FileCheckFrequency: 20 * time.Second,
HealthzBindAddress: net.ParseIP("127.0.0.1"), HealthzBindAddress: net.ParseIP("127.0.0.1"),
HealthzPort: 10248, HealthzPort: 10248,
HostNetworkSources: kubelet.AllSource, HostNetworkSources: kubeletTypes.AllSource,
HostPIDSources: kubelet.AllSource, HostPIDSources: kubeletTypes.AllSource,
HostIPCSources: kubelet.AllSource, HostIPCSources: kubeletTypes.AllSource,
HTTPCheckFrequency: 20 * time.Second, HTTPCheckFrequency: 20 * time.Second,
ImageGCHighThresholdPercent: 90, ImageGCHighThresholdPercent: 90,
ImageGCLowThresholdPercent: 80, ImageGCLowThresholdPercent: 80,
@ -283,17 +284,17 @@ func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
// KubeletConfig returns a KubeletConfig suitable for being run, or an error if the server setup // KubeletConfig returns a KubeletConfig suitable for being run, or an error if the server setup
// is not valid. It will not start any background processes. // is not valid. It will not start any background processes.
func (s *KubeletServer) KubeletConfig() (*KubeletConfig, error) { func (s *KubeletServer) KubeletConfig() (*KubeletConfig, error) {
hostNetworkSources, err := kubelet.GetValidatedSources(strings.Split(s.HostNetworkSources, ",")) hostNetworkSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostNetworkSources, ","))
if err != nil { if err != nil {
return nil, err return nil, err
} }
hostPIDSources, err := kubelet.GetValidatedSources(strings.Split(s.HostPIDSources, ",")) hostPIDSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostPIDSources, ","))
if err != nil { if err != nil {
return nil, err return nil, err
} }
hostIPCSources, err := kubelet.GetValidatedSources(strings.Split(s.HostIPCSources, ",")) hostIPCSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostIPCSources, ","))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -763,17 +764,17 @@ func makePodSourceConfig(kc *KubeletConfig) *config.PodConfig {
// define file config source // define file config source
if kc.ConfigFile != "" { if kc.ConfigFile != "" {
glog.Infof("Adding manifest file: %v", kc.ConfigFile) glog.Infof("Adding manifest file: %v", kc.ConfigFile)
config.NewSourceFile(kc.ConfigFile, kc.NodeName, kc.FileCheckFrequency, cfg.Channel(kubelet.FileSource)) config.NewSourceFile(kc.ConfigFile, kc.NodeName, kc.FileCheckFrequency, cfg.Channel(kubeletTypes.FileSource))
} }
// define url config source // define url config source
if kc.ManifestURL != "" { if kc.ManifestURL != "" {
glog.Infof("Adding manifest url %q with HTTP header %v", kc.ManifestURL, kc.ManifestURLHeader) glog.Infof("Adding manifest url %q with HTTP header %v", kc.ManifestURL, kc.ManifestURLHeader)
config.NewSourceURL(kc.ManifestURL, kc.ManifestURLHeader, kc.NodeName, kc.HTTPCheckFrequency, cfg.Channel(kubelet.HTTPSource)) config.NewSourceURL(kc.ManifestURL, kc.ManifestURLHeader, kc.NodeName, kc.HTTPCheckFrequency, cfg.Channel(kubeletTypes.HTTPSource))
} }
if kc.KubeClient != nil { if kc.KubeClient != nil {
glog.Infof("Watching apiserver") glog.Infof("Watching apiserver")
config.NewSourceApiserver(kc.KubeClient, kc.NodeName, cfg.Channel(kubelet.ApiserverSource)) config.NewSourceApiserver(kc.KubeClient, kc.NodeName, cfg.Channel(kubeletTypes.ApiserverSource))
} }
return cfg return cfg
} }

17
contrib/mesos/pkg/executor/executor.go
View File

@ -43,6 +43,7 @@ import (
"k8s.io/kubernetes/pkg/kubelet" "k8s.io/kubernetes/pkg/kubelet"
"k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
) )
@ -264,9 +265,9 @@ func (k *KubernetesExecutor) onInitialRegistration() {
defer close(k.initialRegComplete) defer close(k.initialRegComplete)
// emit an empty update to allow the mesos "source" to be marked as seen // emit an empty update to allow the mesos "source" to be marked as seen
k.updateChan <- kubelet.PodUpdate{ k.updateChan <- kubeletTypes.PodUpdate{
Pods: []*api.Pod{}, Pods: []*api.Pod{},
Op: kubelet.SET, Op: kubeletTypes.SET,
Source: k.sourcename, Source: k.sourcename,
} }
} }
@ -392,8 +393,8 @@ func (k *KubernetesExecutor) handleChangedApiserverPod(pod *api.Pod) {
oldPod.DeletionTimestamp = pod.DeletionTimestamp oldPod.DeletionTimestamp = pod.DeletionTimestamp
oldPod.DeletionGracePeriodSeconds = pod.DeletionGracePeriodSeconds oldPod.DeletionGracePeriodSeconds = pod.DeletionGracePeriodSeconds
update := kubelet.PodUpdate{ update := kubeletTypes.PodUpdate{
Op: kubelet.UPDATE, Op: kubeletTypes.UPDATE,
Pods: []*api.Pod{oldPod}, Pods: []*api.Pod{oldPod},
} }
k.updateChan <- update k.updateChan <- update
@ -565,8 +566,8 @@ func (k *KubernetesExecutor) launchTask(driver bindings.ExecutorDriver, taskId s
k.pods[podFullName] = pod k.pods[podFullName] = pod
// send the new pod to the kubelet which will spin it up // send the new pod to the kubelet which will spin it up
update := kubelet.PodUpdate{ update := kubeletTypes.PodUpdate{
Op: kubelet.ADD, Op: kubeletTypes.ADD,
Pods: []*api.Pod{pod}, Pods: []*api.Pod{pod},
} }
k.updateChan <- update k.updateChan <- update
@ -770,8 +771,8 @@ func (k *KubernetesExecutor) removePodTask(driver bindings.ExecutorDriver, tid,
delete(k.pods, pid) delete(k.pods, pid)
// tell the kubelet to remove the pod // tell the kubelet to remove the pod
update := kubelet.PodUpdate{ update := kubeletTypes.PodUpdate{
Op: kubelet.REMOVE, Op: kubeletTypes.REMOVE,
Pods: []*api.Pod{pod}, Pods: []*api.Pod{pod},
} }
k.updateChan <- update k.updateChan <- update

11
contrib/mesos/pkg/executor/executor_test.go
View File

@ -43,6 +43,7 @@ import (
"k8s.io/kubernetes/pkg/kubelet" "k8s.io/kubernetes/pkg/kubelet"
kconfig "k8s.io/kubernetes/pkg/kubelet/config" kconfig "k8s.io/kubernetes/pkg/kubelet/config"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/watch" "k8s.io/kubernetes/pkg/watch"
@ -66,15 +67,15 @@ func TestExecutorRegister(t *testing.T) {
executor.Init(mockDriver) executor.Init(mockDriver)
executor.Registered(mockDriver, nil, nil, nil) executor.Registered(mockDriver, nil, nil, nil)
initialPodUpdate := kubelet.PodUpdate{ initialPodUpdate := kubeletTypes.PodUpdate{
Pods: []*api.Pod{}, Pods: []*api.Pod{},
Op: kubelet.SET, Op: kubeletTypes.SET,
Source: executor.sourcename, Source: executor.sourcename,
} }
receivedInitialPodUpdate := false receivedInitialPodUpdate := false
select { select {
case m := <-updates: case m := <-updates:
update, ok := m.(kubelet.PodUpdate) update, ok := m.(kubeletTypes.PodUpdate)
if ok { if ok {
if reflect.DeepEqual(initialPodUpdate, update) { if reflect.DeepEqual(initialPodUpdate, update) {
receivedInitialPodUpdate = true receivedInitialPodUpdate = true
@ -212,7 +213,7 @@ func TestExecutorLaunchAndKillTask(t *testing.T) {
gotPodUpdate := false gotPodUpdate := false
select { select {
case m := <-updates: case m := <-updates:
update, ok := m.(kubelet.PodUpdate) update, ok := m.(kubeletTypes.PodUpdate)
if ok && len(update.Pods) == 1 { if ok && len(update.Pods) == 1 {
gotPodUpdate = true gotPodUpdate = true
} }
@ -360,7 +361,7 @@ func TestExecutorStaticPods(t *testing.T) {
if !ok { if !ok {
return return
} }
podUpdate, ok := update.(kubelet.PodUpdate) podUpdate, ok := update.(kubeletTypes.PodUpdate)
if !ok { if !ok {
continue continue
} }

15
contrib/mesos/pkg/executor/service/service.go
View File

@ -46,6 +46,7 @@ import (
kconfig "k8s.io/kubernetes/pkg/kubelet/config" kconfig "k8s.io/kubernetes/pkg/kubelet/config"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
utilio "k8s.io/kubernetes/pkg/util/io" utilio "k8s.io/kubernetes/pkg/util/io"
"k8s.io/kubernetes/pkg/util/mount" "k8s.io/kubernetes/pkg/util/mount"
@ -55,7 +56,7 @@ import (
const ( const (
// if we don't use this source then the kubelet will do funny, mirror things. // if we don't use this source then the kubelet will do funny, mirror things.
// @see ConfigSourceAnnotationKey // @see ConfigSourceAnnotationKey
MESOS_CFG_SOURCE = kubelet.ApiserverSource MESOS_CFG_SOURCE = kubeletTypes.ApiserverSource
) )
type KubeletExecutorServer struct { type KubeletExecutorServer struct {
@ -136,17 +137,17 @@ func (s *KubeletExecutorServer) Run(hks hyperkube.Interface, _ []string) error {
//cloud := cloudprovider.InitCloudProvider(s.CloudProvider, s.CloudConfigFile) //cloud := cloudprovider.InitCloudProvider(s.CloudProvider, s.CloudConfigFile)
//log.Infof("Successfully initialized cloud provider: %q from the config file: %q\n", s.CloudProvider, s.CloudConfigFile) //log.Infof("Successfully initialized cloud provider: %q from the config file: %q\n", s.CloudProvider, s.CloudConfigFile)
hostNetworkSources, err := kubelet.GetValidatedSources(strings.Split(s.HostNetworkSources, ",")) hostNetworkSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostNetworkSources, ","))
if err != nil { if err != nil {
return err return err
} }
hostPIDSources, err := kubelet.GetValidatedSources(strings.Split(s.HostPIDSources, ",")) hostPIDSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostPIDSources, ","))
if err != nil { if err != nil {
return err return err
} }
hostIPCSources, err := kubelet.GetValidatedSources(strings.Split(s.HostIPCSources, ",")) hostIPCSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostIPCSources, ","))
if err != nil { if err != nil {
return err return err
} }
@ -383,7 +384,7 @@ func (ks *KubeletExecutorServer) createAndInitKubelet(
go exec.InitializeStaticPodsSource(func() { go exec.InitializeStaticPodsSource(func() {
// Create file source only when we are called back. Otherwise, it is never marked unseen. // Create file source only when we are called back. Otherwise, it is never marked unseen.
fileSourceUpdates := pc.Channel(kubelet.FileSource) fileSourceUpdates := pc.Channel(kubeletTypes.FileSource)
kconfig.NewSourceFile(staticPodsConfigPath, kc.Hostname, kc.FileCheckFrequency, fileSourceUpdates) kconfig.NewSourceFile(staticPodsConfigPath, kc.Hostname, kc.FileCheckFrequency, fileSourceUpdates)
}) })
@ -449,7 +450,7 @@ func (kl *kubeletExecutor) ListenAndServe(address net.IP, port uint, tlsOptions
// runs the main kubelet loop, closing the kubeletFinished chan when the loop exits. // runs the main kubelet loop, closing the kubeletFinished chan when the loop exits.
// never returns. // never returns.
func (kl *kubeletExecutor) Run(updates <-chan kubelet.PodUpdate) { func (kl *kubeletExecutor) Run(updates <-chan kubeletTypes.PodUpdate) {
defer func() { defer func() {
close(kl.kubeletFinished) close(kl.kubeletFinished)
util.HandleCrash() util.HandleCrash()
@ -460,7 +461,7 @@ func (kl *kubeletExecutor) Run(updates <-chan kubelet.PodUpdate) {
// push updates through a closable pipe. when the executor indicates shutdown // push updates through a closable pipe. when the executor indicates shutdown
// via Done() we want to stop the Kubelet from processing updates. // via Done() we want to stop the Kubelet from processing updates.
pipe := make(chan kubelet.PodUpdate) pipe := make(chan kubeletTypes.PodUpdate)
go func() { go func() {
// closing pipe will cause our patched kubelet's syncLoop() to exit // closing pipe will cause our patched kubelet's syncLoop() to exit
defer close(pipe) defer close(pipe)

4
contrib/mesos/pkg/scheduler/scheduler.go
View File

@ -46,8 +46,8 @@ import (
"k8s.io/kubernetes/pkg/api/errors" "k8s.io/kubernetes/pkg/api/errors"
client "k8s.io/kubernetes/pkg/client/unversioned" client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/fields" "k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/kubelet"
"k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/container"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/labels" "k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/tools" "k8s.io/kubernetes/pkg/tools"
"k8s.io/kubernetes/pkg/util/sets" "k8s.io/kubernetes/pkg/util/sets"
@ -897,7 +897,7 @@ func (ks *KubernetesScheduler) recoverTasks() error {
ks.slaveHostNames.Register(slaveId, t.Offer.Host()) ks.slaveHostNames.Register(slaveId, t.Offer.Host())
} }
for _, pod := range podList.Items { for _, pod := range podList.Items {
if _, isMirrorPod := pod.Annotations[kubelet.ConfigMirrorAnnotationKey]; isMirrorPod { if _, isMirrorPod := pod.Annotations[kubeletTypes.ConfigMirrorAnnotationKey]; isMirrorPod {
// mirrored pods are never reconciled because the scheduler isn't responsible for // mirrored pods are never reconciled because the scheduler isn't responsible for
// scheduling them; they're started by the executor/kubelet upon instantiation and // scheduling them; they're started by the executor/kubelet upon instantiation and
// reflected in the apiserver afterward. the scheduler has no knowledge of them. // reflected in the apiserver afterward. the scheduler has no knowledge of them.

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

@ -22,7 +22,7 @@ import (
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
client "k8s.io/kubernetes/pkg/client/unversioned" client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/fields" "k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
) )
// NewSourceApiserver creates a config source that watches and pulls from the apiserver. // NewSourceApiserver creates a config source that watches and pulls from the apiserver.
@ -38,7 +38,7 @@ func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{}
for _, o := range objs { for _, o := range objs {
pods = append(pods, o.(*api.Pod)) pods = append(pods, o.(*api.Pod))
} }
updates <- kubelet.PodUpdate{Pods: pods, Op: kubelet.SET, Source: kubelet.ApiserverSource} updates <- kubeletTypes.PodUpdate{Pods: pods, Op: kubeletTypes.SET, Source: kubeletTypes.ApiserverSource}
} }
cache.NewReflector(lw, &api.Pod{}, cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc), 0).Run() cache.NewReflector(lw, &api.Pod{}, cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc), 0).Run()
} }

34
pkg/kubelet/config/apiserver_test.go
View File

@ -21,7 +21,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/watch" "k8s.io/kubernetes/pkg/watch"
) )
@ -67,8 +67,8 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update := got.(kubelet.PodUpdate) update := got.(kubeletTypes.PodUpdate)
expected := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod1v1) expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod1v1)
if !api.Semantic.DeepEqual(expected, update) { if !api.Semantic.DeepEqual(expected, update) {
t.Errorf("Expected %#v; Got %#v", expected, update) t.Errorf("Expected %#v; Got %#v", expected, update)
} }
@ -79,10 +79,10 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update = got.(kubelet.PodUpdate) update = got.(kubeletTypes.PodUpdate)
// Could be sorted either of these two ways: // Could be sorted either of these two ways:
expectedA := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod1v1, pod2) expectedA := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod1v1, pod2)
expectedB := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod2, pod1v1) expectedB := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod2, pod1v1)
if !api.Semantic.DeepEqual(expectedA, update) && !api.Semantic.DeepEqual(expectedB, update) { if !api.Semantic.DeepEqual(expectedA, update) && !api.Semantic.DeepEqual(expectedB, update) {
t.Errorf("Expected %#v or %#v, Got %#v", expectedA, expectedB, update) t.Errorf("Expected %#v or %#v, Got %#v", expectedA, expectedB, update)
@ -94,9 +94,9 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update = got.(kubelet.PodUpdate) update = got.(kubeletTypes.PodUpdate)
expectedA = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod1v2, pod2) expectedA = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod1v2, pod2)
expectedB = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod2, pod1v2) expectedB = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod2, pod1v2)
if !api.Semantic.DeepEqual(expectedA, update) && !api.Semantic.DeepEqual(expectedB, update) { if !api.Semantic.DeepEqual(expectedA, update) && !api.Semantic.DeepEqual(expectedB, update) {
t.Errorf("Expected %#v or %#v, Got %#v", expectedA, expectedB, update) t.Errorf("Expected %#v or %#v, Got %#v", expectedA, expectedB, update)
@ -108,8 +108,8 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update = got.(kubelet.PodUpdate) update = got.(kubeletTypes.PodUpdate)
expected = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod2) expected = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod2)
if !api.Semantic.DeepEqual(expected, update) { if !api.Semantic.DeepEqual(expected, update) {
t.Errorf("Expected %#v, Got %#v", expected, update) t.Errorf("Expected %#v, Got %#v", expected, update)
} }
@ -120,8 +120,8 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update = got.(kubelet.PodUpdate) update = got.(kubeletTypes.PodUpdate)
expected = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource) expected = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource)
if !api.Semantic.DeepEqual(expected, update) { if !api.Semantic.DeepEqual(expected, update) {
t.Errorf("Expected %#v, Got %#v", expected, update) t.Errorf("Expected %#v, Got %#v", expected, update)
} }
@ -150,7 +150,7 @@ func TestNewSourceApiserver_TwoNamespacesSameName(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update := got.(kubelet.PodUpdate) update := got.(kubeletTypes.PodUpdate)
// Make sure that we get both pods. Catches bug #2294. // Make sure that we get both pods. Catches bug #2294.
if !(len(update.Pods) == 2) { if !(len(update.Pods) == 2) {
t.Errorf("Expected %d, Got %d", 2, len(update.Pods)) t.Errorf("Expected %d, Got %d", 2, len(update.Pods))
@ -162,7 +162,7 @@ func TestNewSourceApiserver_TwoNamespacesSameName(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update = got.(kubelet.PodUpdate) update = got.(kubeletTypes.PodUpdate)
if !(len(update.Pods) == 1) { if !(len(update.Pods) == 1) {
t.Errorf("Expected %d, Got %d", 1, len(update.Pods)) t.Errorf("Expected %d, Got %d", 1, len(update.Pods))
} }
@ -184,8 +184,8 @@ func TestNewSourceApiserverInitialEmptySendsEmptyPodUpdate(t *testing.T) {
if !ok { if !ok {
t.Errorf("Unable to read from channel when expected") t.Errorf("Unable to read from channel when expected")
} }
update := got.(kubelet.PodUpdate) update := got.(kubeletTypes.PodUpdate)
expected := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource) expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource)
if !api.Semantic.DeepEqual(expected, update) { if !api.Semantic.DeepEqual(expected, update) {
t.Errorf("Expected %#v; Got %#v", expected, update) t.Errorf("Expected %#v; Got %#v", expected, update)
} }

4
pkg/kubelet/config/common.go
View File

@ -25,7 +25,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/latest" "k8s.io/kubernetes/pkg/api/latest"
"k8s.io/kubernetes/pkg/api/validation" "k8s.io/kubernetes/pkg/api/validation"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
utilyaml "k8s.io/kubernetes/pkg/util/yaml" utilyaml "k8s.io/kubernetes/pkg/util/yaml"
@ -56,7 +56,7 @@ func applyDefaults(pod *api.Pod, source string, isFile bool, nodeName string) er
glog.V(5).Infof("Generated Name %q for UID %q from URL %s", pod.Name, pod.UID, source) glog.V(5).Infof("Generated Name %q for UID %q from URL %s", pod.Name, pod.UID, source)
if pod.Namespace == "" { if pod.Namespace == "" {
pod.Namespace = kubelet.NamespaceDefault pod.Namespace = kubeletTypes.NamespaceDefault
} }
glog.V(5).Infof("Using namespace %q for pod %q from %s", pod.Namespace, pod.Name, source) glog.V(5).Infof("Using namespace %q for pod %q from %s", pod.Namespace, pod.Name, source)

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

@ -25,7 +25,6 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/validation" "k8s.io/kubernetes/pkg/api/validation"
"k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/client/record"
"k8s.io/kubernetes/pkg/kubelet"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
kubeletUtil "k8s.io/kubernetes/pkg/kubelet/util" kubeletUtil "k8s.io/kubernetes/pkg/kubelet/util"
@ -60,7 +59,7 @@ type PodConfig struct {
mux *config.Mux mux *config.Mux
// the channel of denormalized changes passed to listeners // the channel of denormalized changes passed to listeners
updates chan kubelet.PodUpdate updates chan kubeletTypes.PodUpdate
// contains the list of all configured sources // contains the list of all configured sources
sourcesLock sync.Mutex sourcesLock sync.Mutex
@ -70,7 +69,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, recorder record.EventRecorder) *PodConfig { func NewPodConfig(mode PodConfigNotificationMode, recorder record.EventRecorder) *PodConfig {
updates := make(chan kubelet.PodUpdate, 50) updates := make(chan kubeletTypes.PodUpdate, 50)
storage := newPodStorage(updates, mode, recorder) storage := newPodStorage(updates, mode, recorder)
podConfig := &PodConfig{ podConfig := &PodConfig{
pods: storage, pods: storage,
@ -101,7 +100,7 @@ func (c *PodConfig) SeenAllSources() bool {
} }
// Updates returns a channel of updates to the configuration, properly denormalized. // Updates returns a channel of updates to the configuration, properly denormalized.
func (c *PodConfig) Updates() <-chan kubelet.PodUpdate { func (c *PodConfig) Updates() <-chan kubeletTypes.PodUpdate {
return c.updates return c.updates
} }
@ -123,7 +122,7 @@ type podStorage struct {
// ensures that updates are delivered in strict order // ensures that updates are delivered in strict order
// on the updates channel // on the updates channel
updateLock sync.Mutex updateLock sync.Mutex
updates chan<- kubelet.PodUpdate updates chan<- kubeletTypes.PodUpdate
// contains the set of all sources that have sent at least one SET // contains the set of all sources that have sent at least one SET
sourcesSeenLock sync.Mutex sourcesSeenLock sync.Mutex
@ -136,7 +135,7 @@ type podStorage struct {
// TODO: PodConfigNotificationMode could be handled by a listener to the updates channel // TODO: PodConfigNotificationMode could be handled by a listener to the updates channel
// in the future, especially with multiple listeners. // in the future, especially with multiple listeners.
// TODO: allow initialization of the current state of the store with snapshotted version. // TODO: allow initialization of the current state of the store with snapshotted version.
func newPodStorage(updates chan<- kubelet.PodUpdate, mode PodConfigNotificationMode, recorder record.EventRecorder) *podStorage { func newPodStorage(updates chan<- kubeletTypes.PodUpdate, mode PodConfigNotificationMode, recorder record.EventRecorder) *podStorage {
return &podStorage{ return &podStorage{
pods: make(map[string]map[string]*api.Pod), pods: make(map[string]map[string]*api.Pod),
mode: mode, mode: mode,
@ -173,12 +172,12 @@ func (s *podStorage) Merge(source string, change interface{}) error {
s.updates <- *updates s.updates <- *updates
} }
if len(deletes.Pods) > 0 || len(adds.Pods) > 0 { if len(deletes.Pods) > 0 || len(adds.Pods) > 0 {
s.updates <- kubelet.PodUpdate{Pods: s.MergedState().([]*api.Pod), Op: kubelet.SET, Source: source} s.updates <- kubeletTypes.PodUpdate{Pods: s.MergedState().([]*api.Pod), Op: kubeletTypes.SET, Source: source}
} }
case PodConfigNotificationSnapshot: case PodConfigNotificationSnapshot:
if len(updates.Pods) > 0 || len(deletes.Pods) > 0 || len(adds.Pods) > 0 { if len(updates.Pods) > 0 || len(deletes.Pods) > 0 || len(adds.Pods) > 0 {
s.updates <- kubelet.PodUpdate{Pods: s.MergedState().([]*api.Pod), Op: kubelet.SET, Source: source} s.updates <- kubeletTypes.PodUpdate{Pods: s.MergedState().([]*api.Pod), Op: kubeletTypes.SET, Source: source}
} }
case PodConfigNotificationUnknown: case PodConfigNotificationUnknown:
@ -190,23 +189,23 @@ func (s *podStorage) Merge(source string, change interface{}) error {
return nil return nil
} }
func (s *podStorage) merge(source string, change interface{}) (adds, updates, deletes *kubelet.PodUpdate) { func (s *podStorage) merge(source string, change interface{}) (adds, updates, deletes *kubeletTypes.PodUpdate) {
s.podLock.Lock() s.podLock.Lock()
defer s.podLock.Unlock() defer s.podLock.Unlock()
adds = &kubelet.PodUpdate{Op: kubelet.ADD, Source: source} adds = &kubeletTypes.PodUpdate{Op: kubeletTypes.ADD, Source: source}
updates = &kubelet.PodUpdate{Op: kubelet.UPDATE, Source: source} updates = &kubeletTypes.PodUpdate{Op: kubeletTypes.UPDATE, Source: source}
deletes = &kubelet.PodUpdate{Op: kubelet.REMOVE, Source: source} deletes = &kubeletTypes.PodUpdate{Op: kubeletTypes.REMOVE, Source: source}
pods := s.pods[source] pods := s.pods[source]
if pods == nil { if pods == nil {
pods = make(map[string]*api.Pod) pods = make(map[string]*api.Pod)
} }
update := change.(kubelet.PodUpdate) update := change.(kubeletTypes.PodUpdate)
switch update.Op { switch update.Op {
case kubelet.ADD, kubelet.UPDATE: case kubeletTypes.ADD, kubeletTypes.UPDATE:
if update.Op == kubelet.ADD { if update.Op == kubeletTypes.ADD {
glog.V(4).Infof("Adding new pods from source %s : %v", source, update.Pods) glog.V(4).Infof("Adding new pods from source %s : %v", source, update.Pods)
} else { } else {
glog.V(4).Infof("Updating pods from source %s : %v", source, update.Pods) glog.V(4).Infof("Updating pods from source %s : %v", source, update.Pods)
@ -219,7 +218,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
if ref.Annotations == nil { if ref.Annotations == nil {
ref.Annotations = make(map[string]string) ref.Annotations = make(map[string]string)
} }
ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source ref.Annotations[kubeletTypes.ConfigSourceAnnotationKey] = source
if existing, found := pods[name]; found { if existing, found := pods[name]; found {
if checkAndUpdatePod(existing, ref) { if checkAndUpdatePod(existing, ref) {
// this is an update // this is an update
@ -235,7 +234,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
adds.Pods = append(adds.Pods, ref) adds.Pods = append(adds.Pods, ref)
} }
case kubelet.REMOVE: case kubeletTypes.REMOVE:
glog.V(4).Infof("Removing a pod %v", update) glog.V(4).Infof("Removing a pod %v", update)
for _, value := range update.Pods { for _, value := range update.Pods {
name := kubecontainer.GetPodFullName(value) name := kubecontainer.GetPodFullName(value)
@ -248,7 +247,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
// this is a no-op // this is a no-op
} }
case kubelet.SET: case kubeletTypes.SET:
glog.V(4).Infof("Setting pods for source %s", source) glog.V(4).Infof("Setting pods for source %s", source)
s.markSourceSet(source) s.markSourceSet(source)
// Clear the old map entries by just creating a new map // Clear the old map entries by just creating a new map
@ -262,7 +261,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
if ref.Annotations == nil { if ref.Annotations == nil {
ref.Annotations = make(map[string]string) ref.Annotations = make(map[string]string)
} }
ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source ref.Annotations[kubeletTypes.ConfigSourceAnnotationKey] = source
if existing, found := oldPods[name]; found { if existing, found := oldPods[name]; found {
pods[name] = existing pods[name] = existing
if checkAndUpdatePod(existing, ref) { if checkAndUpdatePod(existing, ref) {
@ -336,9 +335,9 @@ func filterInvalidPods(pods []*api.Pod, source string, recorder record.EventReco
// Annotations that the kubelet adds to the pod. // Annotations that the kubelet adds to the pod.
var localAnnotations = []string{ var localAnnotations = []string{
kubelet.ConfigSourceAnnotationKey, kubeletTypes.ConfigSourceAnnotationKey,
kubelet.ConfigMirrorAnnotationKey, kubeletTypes.ConfigMirrorAnnotationKey,
kubelet.ConfigFirstSeenAnnotationKey, kubeletTypes.ConfigFirstSeenAnnotationKey,
} }
func isLocalAnnotationKey(key string) bool { func isLocalAnnotationKey(key string) bool {
@ -380,7 +379,7 @@ func isAnnotationMapEqual(existingMap, candidateMap map[string]string) bool {
// recordFirstSeenTime records the first seen time of this pod. // recordFirstSeenTime records the first seen time of this pod.
func recordFirstSeenTime(pod *api.Pod) { func recordFirstSeenTime(pod *api.Pod) {
glog.V(4).Infof("Receiving a new pod %q", kubeletUtil.FormatPodName(pod)) glog.V(4).Infof("Receiving a new pod %q", kubeletUtil.FormatPodName(pod))
pod.Annotations[kubelet.ConfigFirstSeenAnnotationKey] = kubeletTypes.NewTimestamp().GetString() pod.Annotations[kubeletTypes.ConfigFirstSeenAnnotationKey] = kubeletTypes.NewTimestamp().GetString()
} }
// updateAnnotations returns an Annotation map containing the api annotation map plus // updateAnnotations returns an Annotation map containing the api annotation map plus
@ -421,7 +420,7 @@ func checkAndUpdatePod(existing, ref *api.Pod) bool {
// Overwrite the first-seen time with the existing one. This is our own // Overwrite the first-seen time with the existing one. This is our own
// internal annotation, there is no need to update. // internal annotation, there is no need to update.
ref.Annotations[kubelet.ConfigFirstSeenAnnotationKey] = existing.Annotations[kubelet.ConfigFirstSeenAnnotationKey] ref.Annotations[kubeletTypes.ConfigFirstSeenAnnotationKey] = existing.Annotations[kubeletTypes.ConfigFirstSeenAnnotationKey]
existing.Spec = ref.Spec existing.Spec = ref.Spec
existing.Labels = ref.Labels existing.Labels = ref.Labels
@ -435,7 +434,7 @@ func checkAndUpdatePod(existing, ref *api.Pod) bool {
func (s *podStorage) Sync() { func (s *podStorage) Sync() {
s.updateLock.Lock() s.updateLock.Lock()
defer s.updateLock.Unlock() defer s.updateLock.Unlock()
s.updates <- kubelet.PodUpdate{Pods: s.MergedState().([]*api.Pod), Op: kubelet.SET, Source: kubelet.AllSource} s.updates <- kubeletTypes.PodUpdate{Pods: s.MergedState().([]*api.Pod), Op: kubeletTypes.SET, Source: kubeletTypes.AllSource}
} }
// Object implements config.Accessor // Object implements config.Accessor

116
pkg/kubelet/config/config_test.go
View File

@ -23,7 +23,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/client/record"
"k8s.io/kubernetes/pkg/conversion" "k8s.io/kubernetes/pkg/conversion"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/securitycontext" "k8s.io/kubernetes/pkg/securitycontext"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
) )
@ -74,11 +74,11 @@ func CreateValidPod(name, namespace string) *api.Pod {
} }
} }
func CreatePodUpdate(op kubelet.PodOperation, source string, pods ...*api.Pod) kubelet.PodUpdate { func CreatePodUpdate(op kubeletTypes.PodOperation, source string, pods ...*api.Pod) kubeletTypes.PodUpdate {
return kubelet.PodUpdate{Pods: pods, Op: op, Source: source} return kubeletTypes.PodUpdate{Pods: pods, Op: op, Source: source}
} }
func createPodConfigTester(mode PodConfigNotificationMode) (chan<- interface{}, <-chan kubelet.PodUpdate, *PodConfig) { func createPodConfigTester(mode PodConfigNotificationMode) (chan<- interface{}, <-chan kubeletTypes.PodUpdate, *PodConfig) {
eventBroadcaster := record.NewBroadcaster() eventBroadcaster := record.NewBroadcaster()
config := NewPodConfig(mode, eventBroadcaster.NewRecorder(api.EventSource{Component: "kubelet"})) config := NewPodConfig(mode, eventBroadcaster.NewRecorder(api.EventSource{Component: "kubelet"}))
channel := config.Channel(TestSource) channel := config.Channel(TestSource)
@ -86,7 +86,7 @@ func createPodConfigTester(mode PodConfigNotificationMode) (chan<- interface{},
return channel, ch, config return channel, ch, config
} }
func expectPodUpdate(t *testing.T, ch <-chan kubelet.PodUpdate, expected ...kubelet.PodUpdate) { func expectPodUpdate(t *testing.T, ch <-chan kubeletTypes.PodUpdate, expected ...kubeletTypes.PodUpdate) {
for i := range expected { for i := range expected {
update := <-ch update := <-ch
sort.Sort(sortedPods(update.Pods)) sort.Sort(sortedPods(update.Pods))
@ -113,7 +113,7 @@ func expectPodUpdate(t *testing.T, ch <-chan kubelet.PodUpdate, expected ...kube
expectNoPodUpdate(t, ch) expectNoPodUpdate(t, ch)
} }
func expectNoPodUpdate(t *testing.T, ch <-chan kubelet.PodUpdate) { func expectNoPodUpdate(t *testing.T, ch <-chan kubeletTypes.PodUpdate) {
select { select {
case update := <-ch: case update := <-ch:
t.Errorf("Expected no update in channel, Got %#v", update) t.Errorf("Expected no update in channel, Got %#v", update)
@ -125,63 +125,63 @@ func TestNewPodAdded(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update // see an update
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")))
config.Sync() config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, kubeletTypes.AllSource, CreateValidPod("foo", "new")))
} }
func TestNewPodAddedInvalidNamespace(t *testing.T) { func TestNewPodAddedInvalidNamespace(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update // see an update
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", ""))
channel <- podUpdate channel <- podUpdate
config.Sync() config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, kubeletTypes.AllSource))
} }
func TestNewPodAddedDefaultNamespace(t *testing.T) { func TestNewPodAddedDefaultNamespace(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update // see an update
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "default")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "default"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "default"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "default")))
config.Sync() config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource, CreateValidPod("foo", "default"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, kubeletTypes.AllSource, CreateValidPod("foo", "default")))
} }
func TestNewPodAddedDifferentNamespaces(t *testing.T) { func TestNewPodAddedDifferentNamespaces(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update // see an update
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "default")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "default"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "default"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "default")))
// see an update in another namespace // see an update in another namespace
podUpdate = CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) podUpdate = CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")))
config.Sync() config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource, CreateValidPod("foo", "default"), CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, kubeletTypes.AllSource, CreateValidPod("foo", "default"), CreateValidPod("foo", "new")))
} }
func TestInvalidPodFiltered(t *testing.T) { func TestInvalidPodFiltered(t *testing.T) {
channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental) channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental)
// see an update // see an update
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")))
// add an invalid update // add an invalid update
podUpdate = CreatePodUpdate(kubelet.UPDATE, TestSource, &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}) podUpdate = CreatePodUpdate(kubeletTypes.UPDATE, TestSource, &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
channel <- podUpdate channel <- podUpdate
expectNoPodUpdate(t, ch) expectNoPodUpdate(t, ch)
} }
@ -190,81 +190,81 @@ func TestNewPodAddedSnapshotAndUpdates(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshotAndUpdates) channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshotAndUpdates)
// see an set // see an set
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo", "new")))
config.Sync() config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, kubeletTypes.AllSource, CreateValidPod("foo", "new")))
// container updates are separated as UPDATE // container updates are separated as UPDATE
pod := *podUpdate.Pods[0] pod := *podUpdate.Pods[0]
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}} pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}}
channel <- CreatePodUpdate(kubelet.ADD, TestSource, &pod) channel <- CreatePodUpdate(kubeletTypes.ADD, TestSource, &pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, &pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, &pod))
} }
func TestNewPodAddedSnapshot(t *testing.T) { func TestNewPodAddedSnapshot(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshot) channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshot)
// see an set // see an set
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo", "new")))
config.Sync() config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, kubeletTypes.AllSource, CreateValidPod("foo", "new")))
// container updates are separated as UPDATE // container updates are separated as UPDATE
pod := *podUpdate.Pods[0] pod := *podUpdate.Pods[0]
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}} pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}}
channel <- CreatePodUpdate(kubelet.ADD, TestSource, &pod) channel <- CreatePodUpdate(kubeletTypes.ADD, TestSource, &pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, &pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, TestSource, &pod))
} }
func TestNewPodAddedUpdatedRemoved(t *testing.T) { func TestNewPodAddedUpdatedRemoved(t *testing.T) {
channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental) channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental)
// should register an add // should register an add
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")))
// should ignore ADDs that are identical // should ignore ADDs that are identical
expectNoPodUpdate(t, ch) expectNoPodUpdate(t, ch)
// an kubelet.ADD should be converted to kubelet.UPDATE // an kubeletTypes.ADD should be converted to kubeletTypes.UPDATE
pod := CreateValidPod("foo", "new") pod := CreateValidPod("foo", "new")
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}} pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}}
podUpdate = CreatePodUpdate(kubelet.ADD, TestSource, pod) podUpdate = CreatePodUpdate(kubeletTypes.ADD, TestSource, pod)
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod))
podUpdate = CreatePodUpdate(kubelet.REMOVE, TestSource, &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "new"}}) podUpdate = CreatePodUpdate(kubeletTypes.REMOVE, TestSource, &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "new"}})
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.REMOVE, TestSource, pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.REMOVE, TestSource, pod))
} }
func TestNewPodAddedUpdatedSet(t *testing.T) { func TestNewPodAddedUpdatedSet(t *testing.T) {
channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental) channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental)
// should register an add // should register an add
podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"), CreateValidPod("foo2", "new"), CreateValidPod("foo3", "new")) podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"), CreateValidPod("foo2", "new"), CreateValidPod("foo3", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"), CreateValidPod("foo2", "new"), CreateValidPod("foo3", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"), CreateValidPod("foo2", "new"), CreateValidPod("foo3", "new")))
// should ignore ADDs that are identical // should ignore ADDs that are identical
expectNoPodUpdate(t, ch) expectNoPodUpdate(t, ch)
// should be converted to an kubelet.ADD, kubelet.REMOVE, and kubelet.UPDATE // should be converted to an kubeletTypes.ADD, kubeletTypes.REMOVE, and kubeletTypes.UPDATE
pod := CreateValidPod("foo2", "new") pod := CreateValidPod("foo2", "new")
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}} pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}}
podUpdate = CreatePodUpdate(kubelet.SET, TestSource, pod, CreateValidPod("foo3", "new"), CreateValidPod("foo4", "new")) podUpdate = CreatePodUpdate(kubeletTypes.SET, TestSource, pod, CreateValidPod("foo3", "new"), CreateValidPod("foo4", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, expectPodUpdate(t, ch,
CreatePodUpdate(kubelet.REMOVE, TestSource, CreateValidPod("foo", "new")), CreatePodUpdate(kubeletTypes.REMOVE, TestSource, CreateValidPod("foo", "new")),
CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo4", "new")), CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo4", "new")),
CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod))
} }
func TestPodUpdateAnnotations(t *testing.T) { func TestPodUpdateAnnotations(t *testing.T) {
@ -279,24 +279,24 @@ func TestPodUpdateAnnotations(t *testing.T) {
t.Fatalf("%v", err) t.Fatalf("%v", err)
} }
podUpdate := CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo1", "new"), clone.(*api.Pod), CreateValidPod("foo3", "new")) podUpdate := CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo1", "new"), clone.(*api.Pod), CreateValidPod("foo3", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new"))) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new")))
pod.Annotations["kubenetes.io/blah"] = "superblah" pod.Annotations["kubenetes.io/blah"] = "superblah"
podUpdate = CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new")) podUpdate = CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod))
pod.Annotations["kubernetes.io/otherblah"] = "doh" pod.Annotations["kubernetes.io/otherblah"] = "doh"
podUpdate = CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new")) podUpdate = CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod))
delete(pod.Annotations, "kubernetes.io/blah") delete(pod.Annotations, "kubernetes.io/blah")
podUpdate = CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new")) podUpdate = CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo1", "new"), pod, CreateValidPod("foo3", "new"))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod))
} }
func TestPodUpdateLables(t *testing.T) { func TestPodUpdateLables(t *testing.T) {
@ -311,13 +311,13 @@ func TestPodUpdateLables(t *testing.T) {
t.Fatalf("%v", err) t.Fatalf("%v", err)
} }
podUpdate := CreatePodUpdate(kubelet.SET, TestSource, clone.(*api.Pod)) podUpdate := CreatePodUpdate(kubeletTypes.SET, TestSource, clone.(*api.Pod))
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, pod))
pod.Labels["key"] = "newValue" pod.Labels["key"] = "newValue"
podUpdate = CreatePodUpdate(kubelet.SET, TestSource, pod) podUpdate = CreatePodUpdate(kubeletTypes.SET, TestSource, pod)
channel <- podUpdate channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod))
} }

8
pkg/kubelet/config/file.go
View File

@ -26,7 +26,7 @@ import (
"time" "time"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
"github.com/golang/glog" "github.com/golang/glog"
@ -66,7 +66,7 @@ func (s *sourceFile) extractFromPath() error {
return err return err
} }
// Emit an update with an empty PodList to allow FileSource to be marked as seen // Emit an update with an empty PodList to allow FileSource to be marked as seen
s.updates <- kubelet.PodUpdate{Pods: []*api.Pod{}, Op: kubelet.SET, Source: kubelet.FileSource} s.updates <- kubeletTypes.PodUpdate{Pods: []*api.Pod{}, Op: kubeletTypes.SET, Source: kubeletTypes.FileSource}
return fmt.Errorf("path does not exist, ignoring") return fmt.Errorf("path does not exist, ignoring")
} }
@ -76,14 +76,14 @@ func (s *sourceFile) extractFromPath() error {
if err != nil { if err != nil {
return err return err
} }
s.updates <- kubelet.PodUpdate{Pods: pods, Op: kubelet.SET, Source: kubelet.FileSource} s.updates <- kubeletTypes.PodUpdate{Pods: pods, Op: kubeletTypes.SET, Source: kubeletTypes.FileSource}
case statInfo.Mode().IsRegular(): case statInfo.Mode().IsRegular():
pod, err := s.extractFromFile(path) pod, err := s.extractFromFile(path)
if err != nil { if err != nil {
return err return err
} }
s.updates <- kubelet.PodUpdate{Pods: []*api.Pod{pod}, Op: kubelet.SET, Source: kubelet.FileSource} s.updates <- kubeletTypes.PodUpdate{Pods: []*api.Pod{pod}, Op: kubeletTypes.SET, Source: kubeletTypes.FileSource}
default: default:
return fmt.Errorf("path is not a directory or file") return fmt.Errorf("path is not a directory or file")

16
pkg/kubelet/config/file_test.go
View File

@ -26,7 +26,7 @@ import (
"k8s.io/kubernetes/pkg/api/testapi" "k8s.io/kubernetes/pkg/api/testapi"
"k8s.io/kubernetes/pkg/api/unversioned" "k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/api/validation" "k8s.io/kubernetes/pkg/api/validation"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/securitycontext" "k8s.io/kubernetes/pkg/securitycontext"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
@ -46,8 +46,8 @@ func TestUpdateOnNonExistentFile(t *testing.T) {
NewSourceFile("random_non_existent_path", "localhost", time.Millisecond, ch) NewSourceFile("random_non_existent_path", "localhost", time.Millisecond, ch)
select { select {
case got := <-ch: case got := <-ch:
update := got.(kubelet.PodUpdate) update := got.(kubeletTypes.PodUpdate)
expected := CreatePodUpdate(kubelet.SET, kubelet.FileSource) expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.FileSource)
if !api.Semantic.DeepDerivative(expected, update) { if !api.Semantic.DeepDerivative(expected, update) {
t.Fatalf("Expected %#v, Got %#v", expected, update) t.Fatalf("Expected %#v, Got %#v", expected, update)
} }
@ -75,7 +75,7 @@ func TestReadPodsFromFile(t *testing.T) {
var testCases = []struct { var testCases = []struct {
desc string desc string
pod runtime.Object pod runtime.Object
expected kubelet.PodUpdate expected kubeletTypes.PodUpdate
}{ }{
{ {
desc: "Simple pod", desc: "Simple pod",
@ -94,7 +94,7 @@ func TestReadPodsFromFile(t *testing.T) {
SecurityContext: &api.PodSecurityContext{}, SecurityContext: &api.PodSecurityContext{},
}, },
}, },
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{ expected: CreatePodUpdate(kubeletTypes.SET, kubeletTypes.FileSource, &api.Pod{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
Name: "test-" + hostname, Name: "test-" + hostname,
UID: "12345", UID: "12345",
@ -137,7 +137,7 @@ func TestReadPodsFromFile(t *testing.T) {
NewSourceFile(file.Name(), hostname, time.Millisecond, ch) NewSourceFile(file.Name(), hostname, time.Millisecond, ch)
select { select {
case got := <-ch: case got := <-ch:
update := got.(kubelet.PodUpdate) update := got.(kubeletTypes.PodUpdate)
for _, pod := range update.Pods { for _, pod := range update.Pods {
if errs := validation.ValidatePod(pod); len(errs) > 0 { if errs := validation.ValidatePod(pod); len(errs) > 0 {
t.Errorf("%s: Invalid pod %#v, %#v", testCase.desc, pod, errs) t.Errorf("%s: Invalid pod %#v, %#v", testCase.desc, pod, errs)
@ -180,8 +180,8 @@ func TestExtractFromEmptyDir(t *testing.T) {
t.Fatalf("Unexpected error: %v", err) t.Fatalf("Unexpected error: %v", err)
} }
update := (<-ch).(kubelet.PodUpdate) update := (<-ch).(kubeletTypes.PodUpdate)
expected := CreatePodUpdate(kubelet.SET, kubelet.FileSource) expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.FileSource)
if !api.Semantic.DeepEqual(expected, update) { if !api.Semantic.DeepEqual(expected, update) {
t.Errorf("Expected %#v, Got %#v", expected, update) t.Errorf("Expected %#v, Got %#v", expected, update)
} }

8
pkg/kubelet/config/http.go
View File

@ -25,7 +25,7 @@ import (
"time" "time"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
"github.com/golang/glog" "github.com/golang/glog"
@ -95,7 +95,7 @@ func (s *sourceURL) extractFromURL() error {
} }
if len(data) == 0 { if len(data) == 0 {
// Emit an update with an empty PodList to allow HTTPSource to be marked as seen // Emit an update with an empty PodList to allow HTTPSource to be marked as seen
s.updates <- kubelet.PodUpdate{Pods: []*api.Pod{}, Op: kubelet.SET, Source: kubelet.HTTPSource} s.updates <- kubeletTypes.PodUpdate{Pods: []*api.Pod{}, Op: kubeletTypes.SET, Source: kubeletTypes.HTTPSource}
return fmt.Errorf("zero-length data received from %v", s.url) return fmt.Errorf("zero-length data received from %v", s.url)
} }
// Short circuit if the data has not changed since the last time it was read. // Short circuit if the data has not changed since the last time it was read.
@ -111,7 +111,7 @@ func (s *sourceURL) extractFromURL() error {
// It parsed but could not be used. // It parsed but could not be used.
return singlePodErr return singlePodErr
} }
s.updates <- kubelet.PodUpdate{Pods: []*api.Pod{pod}, Op: kubelet.SET, Source: kubelet.HTTPSource} s.updates <- kubeletTypes.PodUpdate{Pods: []*api.Pod{pod}, Op: kubeletTypes.SET, Source: kubeletTypes.HTTPSource}
return nil return nil
} }
@ -126,7 +126,7 @@ func (s *sourceURL) extractFromURL() error {
for i := range podList.Items { for i := range podList.Items {
pods = append(pods, &podList.Items[i]) pods = append(pods, &podList.Items[i])
} }
s.updates <- kubelet.PodUpdate{Pods: pods, Op: kubelet.SET, Source: kubelet.HTTPSource} s.updates <- kubeletTypes.PodUpdate{Pods: pods, Op: kubeletTypes.SET, Source: kubeletTypes.HTTPSource}
return nil return nil
} }

20
pkg/kubelet/config/http_test.go
View File

@ -27,7 +27,7 @@ import (
"k8s.io/kubernetes/pkg/api/testapi" "k8s.io/kubernetes/pkg/api/testapi"
"k8s.io/kubernetes/pkg/api/unversioned" "k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/api/validation" "k8s.io/kubernetes/pkg/api/validation"
"k8s.io/kubernetes/pkg/kubelet" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
"k8s.io/kubernetes/pkg/util/errors" "k8s.io/kubernetes/pkg/util/errors"
@ -128,7 +128,7 @@ func TestExtractPodsFromHTTP(t *testing.T) {
var testCases = []struct { var testCases = []struct {
desc string desc string
pods runtime.Object pods runtime.Object
expected kubelet.PodUpdate expected kubeletTypes.PodUpdate
}{ }{
{ {
desc: "Single pod", desc: "Single pod",
@ -148,8 +148,8 @@ func TestExtractPodsFromHTTP(t *testing.T) {
SecurityContext: &api.PodSecurityContext{}, SecurityContext: &api.PodSecurityContext{},
}, },
}, },
expected: CreatePodUpdate(kubelet.SET, expected: CreatePodUpdate(kubeletTypes.SET,
kubelet.HTTPSource, kubeletTypes.HTTPSource,
&api.Pod{ &api.Pod{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
UID: "111", UID: "111",
@ -206,15 +206,15 @@ func TestExtractPodsFromHTTP(t *testing.T) {
}, },
}, },
}, },
expected: CreatePodUpdate(kubelet.SET, expected: CreatePodUpdate(kubeletTypes.SET,
kubelet.HTTPSource, kubeletTypes.HTTPSource,
&api.Pod{ &api.Pod{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
UID: "111", UID: "111",
Name: "foo" + "-" + hostname, Name: "foo" + "-" + hostname,
Namespace: "default", Namespace: "default",
SelfLink: getSelfLink("foo-"+hostname, kubelet.NamespaceDefault), SelfLink: getSelfLink("foo-"+hostname, kubeletTypes.NamespaceDefault),
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
NodeName: hostname, NodeName: hostname,
@ -237,7 +237,7 @@ func TestExtractPodsFromHTTP(t *testing.T) {
Name: "bar" + "-" + hostname, Name: "bar" + "-" + hostname,
Namespace: "default", Namespace: "default",
SelfLink: getSelfLink("bar-"+hostname, kubelet.NamespaceDefault), SelfLink: getSelfLink("bar-"+hostname, kubeletTypes.NamespaceDefault),
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
NodeName: hostname, NodeName: hostname,
@ -279,7 +279,7 @@ func TestExtractPodsFromHTTP(t *testing.T) {
t.Errorf("%s: Unexpected error: %v", testCase.desc, err) t.Errorf("%s: Unexpected error: %v", testCase.desc, err)
continue continue
} }
update := (<-ch).(kubelet.PodUpdate) update := (<-ch).(kubeletTypes.PodUpdate)
if !api.Semantic.DeepEqual(testCase.expected, update) { if !api.Semantic.DeepEqual(testCase.expected, update) {
t.Errorf("%s: Expected: %#v, Got: %#v", testCase.desc, testCase.expected, update) t.Errorf("%s: Expected: %#v, Got: %#v", testCase.desc, testCase.expected, update)
@ -325,7 +325,7 @@ func TestURLWithHeader(t *testing.T) {
if err := c.extractFromURL(); err != nil { if err := c.extractFromURL(); err != nil {
t.Fatalf("Unexpected error extracting from URL: %v", err) t.Fatalf("Unexpected error extracting from URL: %v", err)
} }
update := (<-ch).(kubelet.PodUpdate) update := (<-ch).(kubeletTypes.PodUpdate)
headerVal := fakeHandler.RequestReceived.Header["Metadata-Flavor"] headerVal := fakeHandler.RequestReceived.Header["Metadata-Flavor"]
if len(headerVal) != 1 || headerVal[0] != "Google" { if len(headerVal) != 1 || headerVal[0] != "Google" {

5
pkg/kubelet/fake_pod_workers.go
View File

@ -19,6 +19,7 @@ package kubelet
import ( import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
) )
@ -30,12 +31,12 @@ type fakePodWorkers struct {
t TestingInterface t TestingInterface
} }
func (f *fakePodWorkers) UpdatePod(pod *api.Pod, mirrorPod *api.Pod, updateType SyncPodType, updateComplete func()) { func (f *fakePodWorkers) UpdatePod(pod *api.Pod, mirrorPod *api.Pod, updateType kubeletTypes.SyncPodType, updateComplete func()) {
pods, err := f.runtimeCache.GetPods() pods, err := f.runtimeCache.GetPods()
if err != nil { if err != nil {
f.t.Errorf("Unexpected error: %v", err) f.t.Errorf("Unexpected error: %v", err)
} }
if err := f.syncPodFn(pod, mirrorPod, kubecontainer.Pods(pods).FindPodByID(pod.UID), SyncPodUpdate); err != nil { if err := f.syncPodFn(pod, mirrorPod, kubecontainer.Pods(pods).FindPodByID(pod.UID), kubeletTypes.SyncPodUpdate); err != nil {
f.t.Errorf("Unexpected error: %v", err) f.t.Errorf("Unexpected error: %v", err)
} }
} }

32
pkg/kubelet/kubelet.go
View File

@ -764,7 +764,7 @@ func (kl *Kubelet) StartGarbageCollection() {
} }
// Run starts the kubelet reacting to config updates // Run starts the kubelet reacting to config updates
func (kl *Kubelet) Run(updates <-chan PodUpdate) { func (kl *Kubelet) Run(updates <-chan kubeletTypes.PodUpdate) {
if kl.logServer == nil { if kl.logServer == nil {
kl.logServer = http.StripPrefix("/logs/", http.FileServer(http.Dir("/var/log/"))) kl.logServer = http.StripPrefix("/logs/", http.FileServer(http.Dir("/var/log/")))
} }
@ -1235,12 +1235,12 @@ func (kl *Kubelet) makePodDataDirs(pod *api.Pod) error {
return nil return nil
} }
func (kl *Kubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType SyncPodType) error { func (kl *Kubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType kubeletTypes.SyncPodType) error {
podFullName := kubecontainer.GetPodFullName(pod) podFullName := kubecontainer.GetPodFullName(pod)
uid := pod.UID uid := pod.UID
start := time.Now() start := time.Now()
var firstSeenTime time.Time var firstSeenTime time.Time
if firstSeenTimeStr, ok := pod.Annotations[ConfigFirstSeenAnnotationKey]; !ok { if firstSeenTimeStr, ok := pod.Annotations[kubeletTypes.ConfigFirstSeenAnnotationKey]; !ok {
glog.V(3).Infof("First seen time not recorded for pod %q", pod.UID) glog.V(3).Infof("First seen time not recorded for pod %q", pod.UID)
} else { } else {
firstSeenTime = kubeletTypes.ConvertToTimestamp(firstSeenTimeStr).Get() firstSeenTime = kubeletTypes.ConvertToTimestamp(firstSeenTimeStr).Get()
@ -1319,7 +1319,7 @@ func (kl *Kubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecont
// it's OK to pretend like the kubelet started them after it restarted. // it's OK to pretend like the kubelet started them after it restarted.
var podStatus api.PodStatus var podStatus api.PodStatus
if updateType == SyncPodCreate { if updateType == kubeletTypes.SyncPodCreate {
// This is the first time we are syncing the pod. Record the latency // This is the first time we are syncing the pod. Record the latency
// since kubelet first saw the pod if firstSeenTime is set. // since kubelet first saw the pod if firstSeenTime is set.
if !firstSeenTime.IsZero() { if !firstSeenTime.IsZero() {
@ -1889,7 +1889,7 @@ func (kl *Kubelet) canAdmitPod(pods []*api.Pod, pod *api.Pod) (bool, string, str
// any new change seen, will run a sync against desired state and running state. If // any new change seen, will run a sync against desired state and running state. If
// no changes are seen to the configuration, will synchronize the last known desired // no changes are seen to the configuration, will synchronize the last known desired
// 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 kubeletTypes.PodUpdate, handler SyncHandler) {
glog.Info("Starting kubelet main sync loop.") glog.Info("Starting kubelet main sync loop.")
var housekeepingTimestamp time.Time var housekeepingTimestamp time.Time
for { for {
@ -1931,7 +1931,7 @@ func (kl *Kubelet) syncLoop(updates <-chan PodUpdate, handler SyncHandler) {
} }
} }
func (kl *Kubelet) syncLoopIteration(updates <-chan PodUpdate, handler SyncHandler) bool { func (kl *Kubelet) syncLoopIteration(updates <-chan kubeletTypes.PodUpdate, handler SyncHandler) bool {
kl.syncLoopMonitor.Store(time.Now()) kl.syncLoopMonitor.Store(time.Now())
select { select {
case u, open := <-updates: case u, open := <-updates:
@ -1940,16 +1940,16 @@ func (kl *Kubelet) syncLoopIteration(updates <-chan PodUpdate, handler SyncHandl
return false return false
} }
switch u.Op { switch u.Op {
case ADD: case kubeletTypes.ADD:
glog.V(2).Infof("SyncLoop (ADD): %q", kubeletUtil.FormatPodNames(u.Pods)) glog.V(2).Infof("SyncLoop (ADD): %q", kubeletUtil.FormatPodNames(u.Pods))
handler.HandlePodAdditions(u.Pods) handler.HandlePodAdditions(u.Pods)
case UPDATE: case kubeletTypes.UPDATE:
glog.V(2).Infof("SyncLoop (UPDATE): %q", kubeletUtil.FormatPodNames(u.Pods)) glog.V(2).Infof("SyncLoop (UPDATE): %q", kubeletUtil.FormatPodNames(u.Pods))
handler.HandlePodUpdates(u.Pods) handler.HandlePodUpdates(u.Pods)
case REMOVE: case kubeletTypes.REMOVE:
glog.V(2).Infof("SyncLoop (REMOVE): %q", kubeletUtil.FormatPodNames(u.Pods)) glog.V(2).Infof("SyncLoop (REMOVE): %q", kubeletUtil.FormatPodNames(u.Pods))
handler.HandlePodDeletions(u.Pods) handler.HandlePodDeletions(u.Pods)
case SET: case kubeletTypes.SET:
// TODO: Do we want to support this? // TODO: Do we want to support this?
glog.Errorf("Kubelet does not support snapshot update") glog.Errorf("Kubelet does not support snapshot update")
} }
@ -1962,7 +1962,7 @@ func (kl *Kubelet) syncLoopIteration(updates <-chan PodUpdate, handler SyncHandl
return true return true
} }
func (kl *Kubelet) dispatchWork(pod *api.Pod, syncType SyncPodType, mirrorPod *api.Pod, start time.Time) { func (kl *Kubelet) dispatchWork(pod *api.Pod, syncType kubeletTypes.SyncPodType, mirrorPod *api.Pod, start time.Time) {
if kl.podIsTerminated(pod) { if kl.podIsTerminated(pod) {
return return
} }
@ -1971,7 +1971,7 @@ func (kl *Kubelet) dispatchWork(pod *api.Pod, syncType SyncPodType, mirrorPod *a
metrics.PodWorkerLatency.WithLabelValues(syncType.String()).Observe(metrics.SinceInMicroseconds(start)) metrics.PodWorkerLatency.WithLabelValues(syncType.String()).Observe(metrics.SinceInMicroseconds(start))
}) })
// Note the number of containers for new pods. // Note the number of containers for new pods.
if syncType == SyncPodCreate { if syncType == kubeletTypes.SyncPodCreate {
metrics.ContainersPerPodCount.Observe(float64(len(pod.Spec.Containers))) metrics.ContainersPerPodCount.Observe(float64(len(pod.Spec.Containers)))
} }
} }
@ -1982,7 +1982,7 @@ func (kl *Kubelet) handleMirrorPod(mirrorPod *api.Pod, start time.Time) {
// corresponding static pod. Send update to the pod worker if the static // corresponding static pod. Send update to the pod worker if the static
// pod exists. // pod exists.
if pod, ok := kl.podManager.GetPodByMirrorPod(mirrorPod); ok { if pod, ok := kl.podManager.GetPodByMirrorPod(mirrorPod); ok {
kl.dispatchWork(pod, SyncPodUpdate, mirrorPod, start) kl.dispatchWork(pod, kubeletTypes.SyncPodUpdate, mirrorPod, start)
} }
} }
@ -2007,7 +2007,7 @@ func (kl *Kubelet) HandlePodAdditions(pods []*api.Pod) {
continue continue
} }
mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod) mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
kl.dispatchWork(pod, SyncPodCreate, mirrorPod, start) kl.dispatchWork(pod, kubeletTypes.SyncPodCreate, mirrorPod, start)
kl.probeManager.AddPod(pod) kl.probeManager.AddPod(pod)
} }
} }
@ -2023,7 +2023,7 @@ func (kl *Kubelet) HandlePodUpdates(pods []*api.Pod) {
// TODO: Evaluate if we need to validate and reject updates. // TODO: Evaluate if we need to validate and reject updates.
mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod) mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
kl.dispatchWork(pod, SyncPodUpdate, mirrorPod, start) kl.dispatchWork(pod, kubeletTypes.SyncPodUpdate, mirrorPod, start)
} }
} }
@ -2048,7 +2048,7 @@ func (kl *Kubelet) HandlePodSyncs(pods []*api.Pod) {
start := time.Now() start := time.Now()
for _, pod := range pods { for _, pod := range pods {
mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod) mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
kl.dispatchWork(pod, SyncPodSync, mirrorPod, start) kl.dispatchWork(pod, kubeletTypes.SyncPodSync, mirrorPod, start)
} }
} }

51
pkg/kubelet/kubelet_test.go
View File

@ -47,6 +47,7 @@ import (
"k8s.io/kubernetes/pkg/kubelet/network" "k8s.io/kubernetes/pkg/kubelet/network"
"k8s.io/kubernetes/pkg/kubelet/prober" "k8s.io/kubernetes/pkg/kubelet/prober"
"k8s.io/kubernetes/pkg/kubelet/status" "k8s.io/kubernetes/pkg/kubelet/status"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
@ -311,7 +312,7 @@ func TestKubeletDirsCompat(t *testing.T) {
} }
} }
var emptyPodUIDs map[types.UID]SyncPodType var emptyPodUIDs map[types.UID]kubeletTypes.SyncPodType
func TestSyncLoopTimeUpdate(t *testing.T) { func TestSyncLoopTimeUpdate(t *testing.T) {
testKubelet := newTestKubelet(t) testKubelet := newTestKubelet(t)
@ -323,12 +324,12 @@ func TestSyncLoopTimeUpdate(t *testing.T) {
t.Errorf("Unexpected sync loop time: %s, expected 0", loopTime1) t.Errorf("Unexpected sync loop time: %s, expected 0", loopTime1)
} }
kubelet.syncLoopIteration(make(chan PodUpdate), kubelet) kubelet.syncLoopIteration(make(chan kubeletTypes.PodUpdate), kubelet)
loopTime2 := kubelet.LatestLoopEntryTime() loopTime2 := kubelet.LatestLoopEntryTime()
if loopTime2.IsZero() { if loopTime2.IsZero() {
t.Errorf("Unexpected sync loop time: 0, expected non-zero value.") t.Errorf("Unexpected sync loop time: 0, expected non-zero value.")
} }
kubelet.syncLoopIteration(make(chan PodUpdate), kubelet) kubelet.syncLoopIteration(make(chan kubeletTypes.PodUpdate), kubelet)
loopTime3 := kubelet.LatestLoopEntryTime() loopTime3 := kubelet.LatestLoopEntryTime()
if !loopTime3.After(loopTime1) { if !loopTime3.After(loopTime1) {
t.Errorf("Sync Loop Time was not updated correctly. Second update timestamp should be greater than first update timestamp") t.Errorf("Sync Loop Time was not updated correctly. Second update timestamp should be greater than first update timestamp")
@ -345,7 +346,7 @@ func TestSyncLoopAbort(t *testing.T) {
// the channel close // the channel close
kubelet.resyncInterval = time.Second * 30 kubelet.resyncInterval = time.Second * 30
ch := make(chan PodUpdate) ch := make(chan kubeletTypes.PodUpdate)
close(ch) close(ch)
// sanity check (also prevent this test from hanging in the next step) // sanity check (also prevent this test from hanging in the next step)
@ -2657,7 +2658,7 @@ func TestUpdateNodeStatusError(t *testing.T) {
} }
func TestCreateMirrorPod(t *testing.T) { func TestCreateMirrorPod(t *testing.T) {
for _, updateType := range []SyncPodType{SyncPodCreate, SyncPodUpdate} { for _, updateType := range []kubeletTypes.SyncPodType{kubeletTypes.SyncPodCreate, kubeletTypes.SyncPodUpdate} {
testKubelet := newTestKubelet(t) testKubelet := newTestKubelet(t)
kl := testKubelet.kubelet kl := testKubelet.kubelet
manager := testKubelet.fakeMirrorClient manager := testKubelet.fakeMirrorClient
@ -2667,7 +2668,7 @@ func TestCreateMirrorPod(t *testing.T) {
Name: "bar", Name: "bar",
Namespace: "foo", Namespace: "foo",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "file", kubeletTypes.ConfigSourceAnnotationKey: "file",
}, },
}, },
} }
@ -2700,7 +2701,7 @@ func TestDeleteOutdatedMirrorPod(t *testing.T) {
Name: "foo", Name: "foo",
Namespace: "ns", Namespace: "ns",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "file", kubeletTypes.ConfigSourceAnnotationKey: "file",
}, },
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
@ -2716,8 +2717,8 @@ func TestDeleteOutdatedMirrorPod(t *testing.T) {
Name: "foo", Name: "foo",
Namespace: "ns", Namespace: "ns",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "api", kubeletTypes.ConfigSourceAnnotationKey: "api",
ConfigMirrorAnnotationKey: "mirror", kubeletTypes.ConfigMirrorAnnotationKey: "mirror",
}, },
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
@ -2729,7 +2730,7 @@ func TestDeleteOutdatedMirrorPod(t *testing.T) {
pods := []*api.Pod{pod, mirrorPod} pods := []*api.Pod{pod, mirrorPod}
kl.podManager.SetPods(pods) kl.podManager.SetPods(pods)
err := kl.syncPod(pod, mirrorPod, container.Pod{}, SyncPodUpdate) err := kl.syncPod(pod, mirrorPod, container.Pod{}, kubeletTypes.SyncPodUpdate)
if err != nil { if err != nil {
t.Errorf("unexpected error: %v", err) t.Errorf("unexpected error: %v", err)
} }
@ -2754,8 +2755,8 @@ func TestDeleteOrphanedMirrorPods(t *testing.T) {
Name: "pod1", Name: "pod1",
Namespace: "ns", Namespace: "ns",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "api", kubeletTypes.ConfigSourceAnnotationKey: "api",
ConfigMirrorAnnotationKey: "mirror", kubeletTypes.ConfigMirrorAnnotationKey: "mirror",
}, },
}, },
}, },
@ -2765,8 +2766,8 @@ func TestDeleteOrphanedMirrorPods(t *testing.T) {
Name: "pod2", Name: "pod2",
Namespace: "ns", Namespace: "ns",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "api", kubeletTypes.ConfigSourceAnnotationKey: "api",
ConfigMirrorAnnotationKey: "mirror", kubeletTypes.ConfigMirrorAnnotationKey: "mirror",
}, },
}, },
}, },
@ -2797,7 +2798,7 @@ func TestGetContainerInfoForMirrorPods(t *testing.T) {
Name: "qux", Name: "qux",
Namespace: "ns", Namespace: "ns",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "file", kubeletTypes.ConfigSourceAnnotationKey: "file",
}, },
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
@ -2812,8 +2813,8 @@ func TestGetContainerInfoForMirrorPods(t *testing.T) {
Name: "qux", Name: "qux",
Namespace: "ns", Namespace: "ns",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "api", kubeletTypes.ConfigSourceAnnotationKey: "api",
ConfigMirrorAnnotationKey: "mirror", kubeletTypes.ConfigMirrorAnnotationKey: "mirror",
}, },
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
@ -2879,7 +2880,7 @@ func TestDoNotCacheStatusForStaticPods(t *testing.T) {
Name: "staticFoo", Name: "staticFoo",
Namespace: "new", Namespace: "new",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "file", kubeletTypes.ConfigSourceAnnotationKey: "file",
}, },
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
@ -2904,7 +2905,7 @@ func TestHostNetworkAllowed(t *testing.T) {
capabilities.SetForTests(capabilities.Capabilities{ capabilities.SetForTests(capabilities.Capabilities{
PrivilegedSources: capabilities.PrivilegedSources{ PrivilegedSources: capabilities.PrivilegedSources{
HostNetworkSources: []string{ApiserverSource, FileSource}, HostNetworkSources: []string{kubeletTypes.ApiserverSource, kubeletTypes.FileSource},
}, },
}) })
pod := &api.Pod{ pod := &api.Pod{
@ -2913,7 +2914,7 @@ func TestHostNetworkAllowed(t *testing.T) {
Name: "foo", Name: "foo",
Namespace: "new", Namespace: "new",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: FileSource, kubeletTypes.ConfigSourceAnnotationKey: kubeletTypes.FileSource,
}, },
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
@ -2926,7 +2927,7 @@ func TestHostNetworkAllowed(t *testing.T) {
}, },
} }
kubelet.podManager.SetPods([]*api.Pod{pod}) kubelet.podManager.SetPods([]*api.Pod{pod})
err := kubelet.syncPod(pod, nil, container.Pod{}, SyncPodUpdate) err := kubelet.syncPod(pod, nil, container.Pod{}, kubeletTypes.SyncPodUpdate)
if err != nil { if err != nil {
t.Errorf("expected pod infra creation to succeed: %v", err) t.Errorf("expected pod infra creation to succeed: %v", err)
} }
@ -2947,7 +2948,7 @@ func TestHostNetworkDisallowed(t *testing.T) {
Name: "foo", Name: "foo",
Namespace: "new", Namespace: "new",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: FileSource, kubeletTypes.ConfigSourceAnnotationKey: kubeletTypes.FileSource,
}, },
}, },
Spec: api.PodSpec{ Spec: api.PodSpec{
@ -2959,7 +2960,7 @@ func TestHostNetworkDisallowed(t *testing.T) {
}, },
}, },
} }
err := kubelet.syncPod(pod, nil, container.Pod{}, SyncPodUpdate) err := kubelet.syncPod(pod, nil, container.Pod{}, kubeletTypes.SyncPodUpdate)
if err == nil { if err == nil {
t.Errorf("expected pod infra creation to fail") t.Errorf("expected pod infra creation to fail")
} }
@ -2986,7 +2987,7 @@ func TestPrivilegeContainerAllowed(t *testing.T) {
}, },
} }
kubelet.podManager.SetPods([]*api.Pod{pod}) kubelet.podManager.SetPods([]*api.Pod{pod})
err := kubelet.syncPod(pod, nil, container.Pod{}, SyncPodUpdate) err := kubelet.syncPod(pod, nil, container.Pod{}, kubeletTypes.SyncPodUpdate)
if err != nil { if err != nil {
t.Errorf("expected pod infra creation to succeed: %v", err) t.Errorf("expected pod infra creation to succeed: %v", err)
} }
@ -3012,7 +3013,7 @@ func TestPrivilegeContainerDisallowed(t *testing.T) {
}, },
}, },
} }
err := kubelet.syncPod(pod, nil, container.Pod{}, SyncPodUpdate) err := kubelet.syncPod(pod, nil, container.Pod{}, kubeletTypes.SyncPodUpdate)
if err == nil { if err == nil {
t.Errorf("expected pod infra creation to fail") t.Errorf("expected pod infra creation to fail")
} }

11
pkg/kubelet/mirror_client.go
View File

@ -23,6 +23,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
client "k8s.io/kubernetes/pkg/client/unversioned" client "k8s.io/kubernetes/pkg/client/unversioned"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
) )
// Mirror client is used to create/delete a mirror pod. // Mirror client is used to create/delete a mirror pod.
@ -54,7 +55,7 @@ func (mc *basicMirrorClient) CreateMirrorPod(pod *api.Pod) error {
for k, v := range pod.Annotations { for k, v := range pod.Annotations {
copyPod.Annotations[k] = v copyPod.Annotations[k] = v
} }
copyPod.Annotations[ConfigMirrorAnnotationKey] = MirrorType copyPod.Annotations[kubeletTypes.ConfigMirrorAnnotationKey] = kubeletTypes.MirrorType
_, err := mc.apiserverClient.Pods(copyPod.Namespace).Create(&copyPod) _, err := mc.apiserverClient.Pods(copyPod.Namespace).Create(&copyPod)
return err return err
@ -80,7 +81,7 @@ func (mc *basicMirrorClient) DeleteMirrorPod(podFullName string) error {
// Helper functions. // Helper functions.
func getPodSource(pod *api.Pod) (string, error) { func getPodSource(pod *api.Pod) (string, error) {
if pod.Annotations != nil { if pod.Annotations != nil {
if source, ok := pod.Annotations[ConfigSourceAnnotationKey]; ok { if source, ok := pod.Annotations[kubeletTypes.ConfigSourceAnnotationKey]; ok {
return source, nil return source, nil
} }
} }
@ -89,13 +90,13 @@ func getPodSource(pod *api.Pod) (string, error) {
func isStaticPod(pod *api.Pod) bool { func isStaticPod(pod *api.Pod) bool {
source, err := getPodSource(pod) source, err := getPodSource(pod)
return err == nil && source != ApiserverSource return err == nil && source != kubeletTypes.ApiserverSource
} }
func isMirrorPod(pod *api.Pod) bool { func isMirrorPod(pod *api.Pod) bool {
if value, ok := pod.Annotations[ConfigMirrorAnnotationKey]; !ok { if value, ok := pod.Annotations[kubeletTypes.ConfigMirrorAnnotationKey]; !ok {
return false return false
} else { } else {
return value == MirrorType return value == kubeletTypes.MirrorType
} }
} }

9
pkg/kubelet/pod_manager_test.go
View File

@ -21,6 +21,7 @@ import (
"testing" "testing"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
) )
// Stub out mirror client for testing purpose. // Stub out mirror client for testing purpose.
@ -40,8 +41,8 @@ func TestGetSetPods(t *testing.T) {
Name: "bar", Name: "bar",
Namespace: "default", Namespace: "default",
Annotations: map[string]string{ Annotations: map[string]string{
ConfigSourceAnnotationKey: "api", kubeletTypes.ConfigSourceAnnotationKey: "api",
ConfigMirrorAnnotationKey: "mirror", kubeletTypes.ConfigMirrorAnnotationKey: "mirror",
}, },
}, },
} }
@ -50,7 +51,7 @@ func TestGetSetPods(t *testing.T) {
UID: "123456789", UID: "123456789",
Name: "bar", Name: "bar",
Namespace: "default", Namespace: "default",
Annotations: map[string]string{ConfigSourceAnnotationKey: "file"}, Annotations: map[string]string{kubeletTypes.ConfigSourceAnnotationKey: "file"},
}, },
} }
@ -60,7 +61,7 @@ func TestGetSetPods(t *testing.T) {
UID: "999999999", UID: "999999999",
Name: "taco", Name: "taco",
Namespace: "default", Namespace: "default",
Annotations: map[string]string{ConfigSourceAnnotationKey: "api"}, Annotations: map[string]string{kubeletTypes.ConfigSourceAnnotationKey: "api"},
}, },
}, },
staticPod, staticPod,

9
pkg/kubelet/pod_workers.go
View File

@ -24,18 +24,19 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/client/record"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
) )
// PodWorkers is an abstract interface for testability. // PodWorkers is an abstract interface for testability.
type PodWorkers interface { type PodWorkers interface {
UpdatePod(pod *api.Pod, mirrorPod *api.Pod, updateType SyncPodType, updateComplete func()) UpdatePod(pod *api.Pod, mirrorPod *api.Pod, updateType kubeletTypes.SyncPodType, updateComplete func())
ForgetNonExistingPodWorkers(desiredPods map[types.UID]empty) ForgetNonExistingPodWorkers(desiredPods map[types.UID]empty)
ForgetWorker(uid types.UID) ForgetWorker(uid types.UID)
} }
type syncPodFnType func(*api.Pod, *api.Pod, kubecontainer.Pod, SyncPodType) error type syncPodFnType func(*api.Pod, *api.Pod, kubecontainer.Pod, kubeletTypes.SyncPodType) error
type podWorkers struct { type podWorkers struct {
// Protects all per worker fields. // Protects all per worker fields.
@ -74,7 +75,7 @@ type workUpdate struct {
updateCompleteFn func() updateCompleteFn func()
// A string describing the type of this update, eg: create // A string describing the type of this update, eg: create
updateType SyncPodType updateType kubeletTypes.SyncPodType
} }
func newPodWorkers(runtimeCache kubecontainer.RuntimeCache, syncPodFn syncPodFnType, func newPodWorkers(runtimeCache kubecontainer.RuntimeCache, syncPodFn syncPodFnType,
@ -121,7 +122,7 @@ func (p *podWorkers) managePodLoop(podUpdates <-chan workUpdate) {
} }
// Apply the new setting to the specified pod. updateComplete is called when the update is completed. // Apply the new setting to the specified pod. updateComplete is called when the update is completed.
func (p *podWorkers) UpdatePod(pod *api.Pod, mirrorPod *api.Pod, updateType SyncPodType, updateComplete func()) { func (p *podWorkers) UpdatePod(pod *api.Pod, mirrorPod *api.Pod, updateType kubeletTypes.SyncPodType, updateComplete func()) {
uid := pod.UID uid := pod.UID
var podUpdates chan workUpdate var podUpdates chan workUpdate
var exists bool var exists bool

15
pkg/kubelet/pod_workers_test.go
View File

@ -30,6 +30,7 @@ import (
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
"k8s.io/kubernetes/pkg/kubelet/network" "k8s.io/kubernetes/pkg/kubelet/network"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
) )
@ -56,7 +57,7 @@ func createPodWorkers() (*podWorkers, map[types.UID][]string) {
fakeRuntimeCache := createFakeRuntimeCache(fakeRecorder) fakeRuntimeCache := createFakeRuntimeCache(fakeRecorder)
podWorkers := newPodWorkers( podWorkers := newPodWorkers(
fakeRuntimeCache, fakeRuntimeCache,
func(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType SyncPodType) error { func(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType kubeletTypes.SyncPodType) error {
func() { func() {
lock.Lock() lock.Lock()
defer lock.Unlock() defer lock.Unlock()
@ -93,7 +94,7 @@ func TestUpdatePod(t *testing.T) {
numPods := 20 numPods := 20
for i := 0; i < numPods; i++ { for i := 0; i < numPods; i++ {
for j := i; j < numPods; j++ { for j := i; j < numPods; j++ {
podWorkers.UpdatePod(newPod(string(j), string(i)), nil, SyncPodCreate, func() {}) podWorkers.UpdatePod(newPod(string(j), string(i)), nil, kubeletTypes.SyncPodCreate, func() {})
} }
} }
drainWorkers(podWorkers, numPods) drainWorkers(podWorkers, numPods)
@ -126,7 +127,7 @@ func TestForgetNonExistingPodWorkers(t *testing.T) {
numPods := 20 numPods := 20
for i := 0; i < numPods; i++ { for i := 0; i < numPods; i++ {
podWorkers.UpdatePod(newPod(string(i), "name"), nil, SyncPodUpdate, func() {}) podWorkers.UpdatePod(newPod(string(i), "name"), nil, kubeletTypes.SyncPodUpdate, func() {})
} }
drainWorkers(podWorkers, numPods) drainWorkers(podWorkers, numPods)
@ -162,12 +163,12 @@ type simpleFakeKubelet struct {
wg sync.WaitGroup wg sync.WaitGroup
} }
func (kl *simpleFakeKubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType SyncPodType) error { func (kl *simpleFakeKubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType kubeletTypes.SyncPodType) error {
kl.pod, kl.mirrorPod, kl.runningPod = pod, mirrorPod, runningPod kl.pod, kl.mirrorPod, kl.runningPod = pod, mirrorPod, runningPod
return nil return nil
} }
func (kl *simpleFakeKubelet) syncPodWithWaitGroup(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType SyncPodType) error { func (kl *simpleFakeKubelet) syncPodWithWaitGroup(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecontainer.Pod, updateType kubeletTypes.SyncPodType) error {
kl.pod, kl.mirrorPod, kl.runningPod = pod, mirrorPod, runningPod kl.pod, kl.mirrorPod, kl.runningPod = pod, mirrorPod, runningPod
kl.wg.Done() kl.wg.Done()
return nil return nil
@ -353,8 +354,8 @@ func TestFakePodWorkers(t *testing.T) {
kubeletForRealWorkers.wg.Add(1) kubeletForRealWorkers.wg.Add(1)
fakeDocker.ContainerList = tt.containerList fakeDocker.ContainerList = tt.containerList
realPodWorkers.UpdatePod(tt.pod, tt.mirrorPod, SyncPodUpdate, func() {}) realPodWorkers.UpdatePod(tt.pod, tt.mirrorPod, kubeletTypes.SyncPodUpdate, func() {})
fakePodWorkers.UpdatePod(tt.pod, tt.mirrorPod, SyncPodUpdate, func() {}) fakePodWorkers.UpdatePod(tt.pod, tt.mirrorPod, kubeletTypes.SyncPodUpdate, func() {})
kubeletForRealWorkers.wg.Wait() kubeletForRealWorkers.wg.Wait()

5
pkg/kubelet/runonce.go
View File

@ -24,6 +24,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/container"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
) )
const ( const (
@ -39,7 +40,7 @@ type RunPodResult struct {
} }
// RunOnce polls from one configuration update and run the associated pods. // RunOnce polls from one configuration update and run the associated pods.
func (kl *Kubelet) RunOnce(updates <-chan PodUpdate) ([]RunPodResult, error) { func (kl *Kubelet) RunOnce(updates <-chan kubeletTypes.PodUpdate) ([]RunPodResult, error) {
select { select {
case u := <-updates: case u := <-updates:
glog.Infof("processing manifest with %d pods", len(u.Pods)) glog.Infof("processing manifest with %d pods", len(u.Pods))
@ -109,7 +110,7 @@ func (kl *Kubelet) runPod(pod *api.Pod, retryDelay time.Duration) error {
glog.Infof("pod %q containers not running: syncing", pod.Name) glog.Infof("pod %q containers not running: syncing", pod.Name)
// We don't create mirror pods in this mode; pass a dummy boolean value // We don't create mirror pods in this mode; pass a dummy boolean value
// to sycnPod. // to sycnPod.
if err = kl.syncPod(pod, nil, p, SyncPodUpdate); err != nil { if err = kl.syncPod(pod, nil, p, kubeletTypes.SyncPodUpdate); err != nil {
return fmt.Errorf("error syncing pod: %v", err) return fmt.Errorf("error syncing pod: %v", err)
} }
if retry >= RunOnceMaxRetries { if retry >= RunOnceMaxRetries {

3
pkg/kubelet/server_test.go
View File

@ -37,6 +37,7 @@ import (
apierrs "k8s.io/kubernetes/pkg/api/errors" apierrs "k8s.io/kubernetes/pkg/api/errors"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util/httpstream" "k8s.io/kubernetes/pkg/util/httpstream"
"k8s.io/kubernetes/pkg/util/httpstream/spdy" "k8s.io/kubernetes/pkg/util/httpstream/spdy"
@ -172,7 +173,7 @@ func readResp(resp *http.Response) (string, error) {
// A helper function to return the correct pod name. // A helper function to return the correct pod name.
func getPodName(name, namespace string) string { func getPodName(name, namespace string) string {
if namespace == "" { if namespace == "" {
namespace = NamespaceDefault namespace = kubeletTypes.NamespaceDefault
} }
return name + "_" + namespace return name + "_" + namespace
} }

2
pkg/kubelet/types.go → pkg/kubelet/types/pod_update.go
View File

@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
package kubelet package types
import ( import (
"fmt" "fmt"

2
pkg/kubelet/types_test.go → pkg/kubelet/types/pod_update_test.go
View File

@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
package kubelet package types
import ( import (
"testing" "testing"

2
plugin/pkg/admission/serviceaccount/admission.go
View File

@ -29,7 +29,7 @@ import (
client "k8s.io/kubernetes/pkg/client/unversioned" client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/controller/serviceaccount" "k8s.io/kubernetes/pkg/controller/serviceaccount"
"k8s.io/kubernetes/pkg/fields" "k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/kubelet" kubelet "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/labels" "k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/util/sets" "k8s.io/kubernetes/pkg/util/sets"

2
plugin/pkg/admission/serviceaccount/admission_test.go
View File

@ -23,7 +23,7 @@ import (
"k8s.io/kubernetes/pkg/admission" "k8s.io/kubernetes/pkg/admission"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/client/unversioned/testclient" "k8s.io/kubernetes/pkg/client/unversioned/testclient"
"k8s.io/kubernetes/pkg/kubelet" kubelet "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
) )