diff --git a/cmd/integration/integration.go b/cmd/integration/integration.go index 00073904ac5..c711af689c7 100644 --- a/cmd/integration/integration.go +++ b/cmd/integration/integration.go @@ -48,10 +48,10 @@ import ( "k8s.io/kubernetes/pkg/controller/node" replicationControllerPkg "k8s.io/kubernetes/pkg/controller/replication" "k8s.io/kubernetes/pkg/fields" - "k8s.io/kubernetes/pkg/kubelet" "k8s.io/kubernetes/pkg/kubelet/cadvisor" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/dockertools" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/labels" "k8s.io/kubernetes/pkg/master" "k8s.io/kubernetes/pkg/tools/etcdtest" @@ -415,7 +415,7 @@ containers: // Wait for the mirror pod to be created. podName := fmt.Sprintf("%s-localhost", desc) - namespace := kubelet.NamespaceDefault + namespace := kubeletTypes.NamespaceDefault if err := wait.Poll(time.Second, longTestTimeout, podRunning(c, namespace, podName)); err != nil { if pods, err := c.Pods(namespace).List(labels.Everything(), fields.Everything()); err == nil { diff --git a/cmd/kubelet/app/server.go b/cmd/kubelet/app/server.go index b02d63e8d99..cd4d0cd4fc3 100644 --- a/cmd/kubelet/app/server.go +++ b/cmd/kubelet/app/server.go @@ -49,6 +49,7 @@ import ( "k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/network" "k8s.io/kubernetes/pkg/kubelet/qos" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/master/ports" "k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util/io" @@ -148,8 +149,8 @@ type KubeletBootstrap interface { StartGarbageCollection() ListenAndServe(net.IP, uint, *kubelet.TLSOptions, bool) ListenAndServeReadOnly(net.IP, uint) - Run(<-chan kubelet.PodUpdate) - RunOnce(<-chan kubelet.PodUpdate) ([]kubelet.RunPodResult, error) + Run(<-chan kubeletTypes.PodUpdate) + RunOnce(<-chan kubeletTypes.PodUpdate) ([]kubelet.RunPodResult, error) } // create and initialize a Kubelet instance @@ -173,9 +174,9 @@ func NewKubeletServer() *KubeletServer { FileCheckFrequency: 20 * time.Second, HealthzBindAddress: net.ParseIP("127.0.0.1"), HealthzPort: 10248, - HostNetworkSources: kubelet.AllSource, - HostPIDSources: kubelet.AllSource, - HostIPCSources: kubelet.AllSource, + HostNetworkSources: kubeletTypes.AllSource, + HostPIDSources: kubeletTypes.AllSource, + HostIPCSources: kubeletTypes.AllSource, HTTPCheckFrequency: 20 * time.Second, ImageGCHighThresholdPercent: 90, 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 // is not valid. It will not start any background processes. 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 { return nil, err } - hostPIDSources, err := kubelet.GetValidatedSources(strings.Split(s.HostPIDSources, ",")) + hostPIDSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostPIDSources, ",")) if err != nil { return nil, err } - hostIPCSources, err := kubelet.GetValidatedSources(strings.Split(s.HostIPCSources, ",")) + hostIPCSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostIPCSources, ",")) if err != nil { return nil, err } @@ -763,17 +764,17 @@ func makePodSourceConfig(kc *KubeletConfig) *config.PodConfig { // define file config source if 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 if kc.ManifestURL != "" { 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 { 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 } diff --git a/contrib/mesos/pkg/executor/executor.go b/contrib/mesos/pkg/executor/executor.go index 2889ed83d1f..f7ea7c64f66 100644 --- a/contrib/mesos/pkg/executor/executor.go +++ b/contrib/mesos/pkg/executor/executor.go @@ -43,6 +43,7 @@ import ( "k8s.io/kubernetes/pkg/kubelet" "k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/dockertools" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/util" ) @@ -264,9 +265,9 @@ func (k *KubernetesExecutor) onInitialRegistration() { defer close(k.initialRegComplete) // 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{}, - Op: kubelet.SET, + Op: kubeletTypes.SET, Source: k.sourcename, } } @@ -392,8 +393,8 @@ func (k *KubernetesExecutor) handleChangedApiserverPod(pod *api.Pod) { oldPod.DeletionTimestamp = pod.DeletionTimestamp oldPod.DeletionGracePeriodSeconds = pod.DeletionGracePeriodSeconds - update := kubelet.PodUpdate{ - Op: kubelet.UPDATE, + update := kubeletTypes.PodUpdate{ + Op: kubeletTypes.UPDATE, Pods: []*api.Pod{oldPod}, } k.updateChan <- update @@ -565,8 +566,8 @@ func (k *KubernetesExecutor) launchTask(driver bindings.ExecutorDriver, taskId s k.pods[podFullName] = pod // send the new pod to the kubelet which will spin it up - update := kubelet.PodUpdate{ - Op: kubelet.ADD, + update := kubeletTypes.PodUpdate{ + Op: kubeletTypes.ADD, Pods: []*api.Pod{pod}, } k.updateChan <- update @@ -770,8 +771,8 @@ func (k *KubernetesExecutor) removePodTask(driver bindings.ExecutorDriver, tid, delete(k.pods, pid) // tell the kubelet to remove the pod - update := kubelet.PodUpdate{ - Op: kubelet.REMOVE, + update := kubeletTypes.PodUpdate{ + Op: kubeletTypes.REMOVE, Pods: []*api.Pod{pod}, } k.updateChan <- update diff --git a/contrib/mesos/pkg/executor/executor_test.go b/contrib/mesos/pkg/executor/executor_test.go index afe5b3a0fa3..15540e53c21 100644 --- a/contrib/mesos/pkg/executor/executor_test.go +++ b/contrib/mesos/pkg/executor/executor_test.go @@ -43,6 +43,7 @@ import ( "k8s.io/kubernetes/pkg/kubelet" kconfig "k8s.io/kubernetes/pkg/kubelet/config" "k8s.io/kubernetes/pkg/kubelet/dockertools" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/watch" @@ -66,15 +67,15 @@ func TestExecutorRegister(t *testing.T) { executor.Init(mockDriver) executor.Registered(mockDriver, nil, nil, nil) - initialPodUpdate := kubelet.PodUpdate{ + initialPodUpdate := kubeletTypes.PodUpdate{ Pods: []*api.Pod{}, - Op: kubelet.SET, + Op: kubeletTypes.SET, Source: executor.sourcename, } receivedInitialPodUpdate := false select { case m := <-updates: - update, ok := m.(kubelet.PodUpdate) + update, ok := m.(kubeletTypes.PodUpdate) if ok { if reflect.DeepEqual(initialPodUpdate, update) { receivedInitialPodUpdate = true @@ -212,7 +213,7 @@ func TestExecutorLaunchAndKillTask(t *testing.T) { gotPodUpdate := false select { case m := <-updates: - update, ok := m.(kubelet.PodUpdate) + update, ok := m.(kubeletTypes.PodUpdate) if ok && len(update.Pods) == 1 { gotPodUpdate = true } @@ -360,7 +361,7 @@ func TestExecutorStaticPods(t *testing.T) { if !ok { return } - podUpdate, ok := update.(kubelet.PodUpdate) + podUpdate, ok := update.(kubeletTypes.PodUpdate) if !ok { continue } diff --git a/contrib/mesos/pkg/executor/service/service.go b/contrib/mesos/pkg/executor/service/service.go index 07110616526..c473e0a1b14 100644 --- a/contrib/mesos/pkg/executor/service/service.go +++ b/contrib/mesos/pkg/executor/service/service.go @@ -46,6 +46,7 @@ import ( kconfig "k8s.io/kubernetes/pkg/kubelet/config" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/dockertools" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/util" utilio "k8s.io/kubernetes/pkg/util/io" "k8s.io/kubernetes/pkg/util/mount" @@ -55,7 +56,7 @@ import ( const ( // if we don't use this source then the kubelet will do funny, mirror things. // @see ConfigSourceAnnotationKey - MESOS_CFG_SOURCE = kubelet.ApiserverSource + MESOS_CFG_SOURCE = kubeletTypes.ApiserverSource ) type KubeletExecutorServer struct { @@ -136,17 +137,17 @@ func (s *KubeletExecutorServer) Run(hks hyperkube.Interface, _ []string) error { //cloud := cloudprovider.InitCloudProvider(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 { return err } - hostPIDSources, err := kubelet.GetValidatedSources(strings.Split(s.HostPIDSources, ",")) + hostPIDSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostPIDSources, ",")) if err != nil { return err } - hostIPCSources, err := kubelet.GetValidatedSources(strings.Split(s.HostIPCSources, ",")) + hostIPCSources, err := kubeletTypes.GetValidatedSources(strings.Split(s.HostIPCSources, ",")) if err != nil { return err } @@ -383,7 +384,7 @@ func (ks *KubeletExecutorServer) createAndInitKubelet( go exec.InitializeStaticPodsSource(func() { // 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) }) @@ -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. // never returns. -func (kl *kubeletExecutor) Run(updates <-chan kubelet.PodUpdate) { +func (kl *kubeletExecutor) Run(updates <-chan kubeletTypes.PodUpdate) { defer func() { close(kl.kubeletFinished) 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 // via Done() we want to stop the Kubelet from processing updates. - pipe := make(chan kubelet.PodUpdate) + pipe := make(chan kubeletTypes.PodUpdate) go func() { // closing pipe will cause our patched kubelet's syncLoop() to exit defer close(pipe) diff --git a/contrib/mesos/pkg/scheduler/scheduler.go b/contrib/mesos/pkg/scheduler/scheduler.go index ad5d48b13c2..9be6420a7d3 100644 --- a/contrib/mesos/pkg/scheduler/scheduler.go +++ b/contrib/mesos/pkg/scheduler/scheduler.go @@ -46,8 +46,8 @@ import ( "k8s.io/kubernetes/pkg/api/errors" client "k8s.io/kubernetes/pkg/client/unversioned" "k8s.io/kubernetes/pkg/fields" - "k8s.io/kubernetes/pkg/kubelet" "k8s.io/kubernetes/pkg/kubelet/container" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/labels" "k8s.io/kubernetes/pkg/tools" "k8s.io/kubernetes/pkg/util/sets" @@ -897,7 +897,7 @@ func (ks *KubernetesScheduler) recoverTasks() error { ks.slaveHostNames.Register(slaveId, t.Offer.Host()) } 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 // scheduling them; they're started by the executor/kubelet upon instantiation and // reflected in the apiserver afterward. the scheduler has no knowledge of them. diff --git a/pkg/kubelet/config/apiserver.go b/pkg/kubelet/config/apiserver.go index c441b3f0d55..7dd56f432e3 100644 --- a/pkg/kubelet/config/apiserver.go +++ b/pkg/kubelet/config/apiserver.go @@ -22,7 +22,7 @@ import ( "k8s.io/kubernetes/pkg/client/cache" client "k8s.io/kubernetes/pkg/client/unversioned" "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. @@ -38,7 +38,7 @@ func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{} for _, o := range objs { 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() } diff --git a/pkg/kubelet/config/apiserver_test.go b/pkg/kubelet/config/apiserver_test.go index 11b3f982876..6b9986b3157 100644 --- a/pkg/kubelet/config/apiserver_test.go +++ b/pkg/kubelet/config/apiserver_test.go @@ -21,7 +21,7 @@ import ( "k8s.io/kubernetes/pkg/api" "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/watch" ) @@ -67,8 +67,8 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) { if !ok { t.Errorf("Unable to read from channel when expected") } - update := got.(kubelet.PodUpdate) - expected := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod1v1) + update := got.(kubeletTypes.PodUpdate) + expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod1v1) if !api.Semantic.DeepEqual(expected, update) { t.Errorf("Expected %#v; Got %#v", expected, update) } @@ -79,10 +79,10 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) { if !ok { 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: - expectedA := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod1v1, pod2) - expectedB := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod2, pod1v1) + expectedA := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod1v1, pod2) + expectedB := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod2, pod1v1) if !api.Semantic.DeepEqual(expectedA, update) && !api.Semantic.DeepEqual(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 { t.Errorf("Unable to read from channel when expected") } - update = got.(kubelet.PodUpdate) - expectedA = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod1v2, pod2) - expectedB = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod2, pod1v2) + update = got.(kubeletTypes.PodUpdate) + expectedA = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod1v2, pod2) + expectedB = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod2, pod1v2) if !api.Semantic.DeepEqual(expectedA, update) && !api.Semantic.DeepEqual(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 { t.Errorf("Unable to read from channel when expected") } - update = got.(kubelet.PodUpdate) - expected = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, pod2) + update = got.(kubeletTypes.PodUpdate) + expected = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource, pod2) if !api.Semantic.DeepEqual(expected, update) { t.Errorf("Expected %#v, Got %#v", expected, update) } @@ -120,8 +120,8 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) { if !ok { t.Errorf("Unable to read from channel when expected") } - update = got.(kubelet.PodUpdate) - expected = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource) + update = got.(kubeletTypes.PodUpdate) + expected = CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource) if !api.Semantic.DeepEqual(expected, update) { t.Errorf("Expected %#v, Got %#v", expected, update) } @@ -150,7 +150,7 @@ func TestNewSourceApiserver_TwoNamespacesSameName(t *testing.T) { if !ok { 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. if !(len(update.Pods) == 2) { t.Errorf("Expected %d, Got %d", 2, len(update.Pods)) @@ -162,7 +162,7 @@ func TestNewSourceApiserver_TwoNamespacesSameName(t *testing.T) { if !ok { t.Errorf("Unable to read from channel when expected") } - update = got.(kubelet.PodUpdate) + update = got.(kubeletTypes.PodUpdate) if !(len(update.Pods) == 1) { t.Errorf("Expected %d, Got %d", 1, len(update.Pods)) } @@ -184,8 +184,8 @@ func TestNewSourceApiserverInitialEmptySendsEmptyPodUpdate(t *testing.T) { if !ok { t.Errorf("Unable to read from channel when expected") } - update := got.(kubelet.PodUpdate) - expected := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource) + update := got.(kubeletTypes.PodUpdate) + expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.ApiserverSource) if !api.Semantic.DeepEqual(expected, update) { t.Errorf("Expected %#v; Got %#v", expected, update) } diff --git a/pkg/kubelet/config/common.go b/pkg/kubelet/config/common.go index 355d5d4c4ca..5d9abfdf901 100644 --- a/pkg/kubelet/config/common.go +++ b/pkg/kubelet/config/common.go @@ -25,7 +25,7 @@ import ( "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/latest" "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/util" 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) 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) diff --git a/pkg/kubelet/config/config.go b/pkg/kubelet/config/config.go index 99a08b69d3c..b9dcc9be15a 100644 --- a/pkg/kubelet/config/config.go +++ b/pkg/kubelet/config/config.go @@ -25,7 +25,6 @@ import ( "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/validation" "k8s.io/kubernetes/pkg/client/record" - "k8s.io/kubernetes/pkg/kubelet" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" kubeletUtil "k8s.io/kubernetes/pkg/kubelet/util" @@ -60,7 +59,7 @@ type PodConfig struct { mux *config.Mux // the channel of denormalized changes passed to listeners - updates chan kubelet.PodUpdate + updates chan kubeletTypes.PodUpdate // contains the list of all configured sources sourcesLock sync.Mutex @@ -70,7 +69,7 @@ type PodConfig struct { // NewPodConfig creates an object that can merge many configuration sources into a stream // of normalized updates to a pod configuration. 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) podConfig := &PodConfig{ pods: storage, @@ -101,7 +100,7 @@ func (c *PodConfig) SeenAllSources() bool { } // 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 } @@ -123,7 +122,7 @@ type podStorage struct { // ensures that updates are delivered in strict order // on the updates channel updateLock sync.Mutex - updates chan<- kubelet.PodUpdate + updates chan<- kubeletTypes.PodUpdate // contains the set of all sources that have sent at least one SET sourcesSeenLock sync.Mutex @@ -136,7 +135,7 @@ type podStorage struct { // TODO: PodConfigNotificationMode could be handled by a listener to the updates channel // in the future, especially with multiple listeners. // 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{ pods: make(map[string]map[string]*api.Pod), mode: mode, @@ -173,12 +172,12 @@ func (s *podStorage) Merge(source string, change interface{}) error { s.updates <- *updates } 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: 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: @@ -190,23 +189,23 @@ func (s *podStorage) Merge(source string, change interface{}) error { 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() defer s.podLock.Unlock() - adds = &kubelet.PodUpdate{Op: kubelet.ADD, Source: source} - updates = &kubelet.PodUpdate{Op: kubelet.UPDATE, Source: source} - deletes = &kubelet.PodUpdate{Op: kubelet.REMOVE, Source: source} + adds = &kubeletTypes.PodUpdate{Op: kubeletTypes.ADD, Source: source} + updates = &kubeletTypes.PodUpdate{Op: kubeletTypes.UPDATE, Source: source} + deletes = &kubeletTypes.PodUpdate{Op: kubeletTypes.REMOVE, Source: source} pods := s.pods[source] if pods == nil { pods = make(map[string]*api.Pod) } - update := change.(kubelet.PodUpdate) + update := change.(kubeletTypes.PodUpdate) switch update.Op { - case kubelet.ADD, kubelet.UPDATE: - if update.Op == kubelet.ADD { + case kubeletTypes.ADD, kubeletTypes.UPDATE: + if update.Op == kubeletTypes.ADD { glog.V(4).Infof("Adding new pods from source %s : %v", source, update.Pods) } else { 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 { ref.Annotations = make(map[string]string) } - ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source + ref.Annotations[kubeletTypes.ConfigSourceAnnotationKey] = source if existing, found := pods[name]; found { if checkAndUpdatePod(existing, ref) { // 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) } - case kubelet.REMOVE: + case kubeletTypes.REMOVE: glog.V(4).Infof("Removing a pod %v", update) for _, value := range update.Pods { name := kubecontainer.GetPodFullName(value) @@ -248,7 +247,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de // this is a no-op } - case kubelet.SET: + case kubeletTypes.SET: glog.V(4).Infof("Setting pods for source %s", source) s.markSourceSet(source) // 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 { ref.Annotations = make(map[string]string) } - ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source + ref.Annotations[kubeletTypes.ConfigSourceAnnotationKey] = source if existing, found := oldPods[name]; found { pods[name] = existing 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. var localAnnotations = []string{ - kubelet.ConfigSourceAnnotationKey, - kubelet.ConfigMirrorAnnotationKey, - kubelet.ConfigFirstSeenAnnotationKey, + kubeletTypes.ConfigSourceAnnotationKey, + kubeletTypes.ConfigMirrorAnnotationKey, + kubeletTypes.ConfigFirstSeenAnnotationKey, } 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. func recordFirstSeenTime(pod *api.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 @@ -421,7 +420,7 @@ func checkAndUpdatePod(existing, ref *api.Pod) bool { // Overwrite the first-seen time with the existing one. This is our own // 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.Labels = ref.Labels @@ -435,7 +434,7 @@ func checkAndUpdatePod(existing, ref *api.Pod) bool { func (s *podStorage) Sync() { s.updateLock.Lock() 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 diff --git a/pkg/kubelet/config/config_test.go b/pkg/kubelet/config/config_test.go index b2db9114d78..193f4504ab6 100644 --- a/pkg/kubelet/config/config_test.go +++ b/pkg/kubelet/config/config_test.go @@ -23,7 +23,7 @@ import ( "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/client/record" "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/types" ) @@ -74,11 +74,11 @@ func CreateValidPod(name, namespace string) *api.Pod { } } -func CreatePodUpdate(op kubelet.PodOperation, source string, pods ...*api.Pod) kubelet.PodUpdate { - return kubelet.PodUpdate{Pods: pods, Op: op, Source: source} +func CreatePodUpdate(op kubeletTypes.PodOperation, source string, pods ...*api.Pod) kubeletTypes.PodUpdate { + 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() config := NewPodConfig(mode, eventBroadcaster.NewRecorder(api.EventSource{Component: "kubelet"})) channel := config.Channel(TestSource) @@ -86,7 +86,7 @@ func createPodConfigTester(mode PodConfigNotificationMode) (chan<- interface{}, 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 { update := <-ch sort.Sort(sortedPods(update.Pods)) @@ -113,7 +113,7 @@ func expectPodUpdate(t *testing.T, ch <-chan kubelet.PodUpdate, expected ...kube expectNoPodUpdate(t, ch) } -func expectNoPodUpdate(t *testing.T, ch <-chan kubelet.PodUpdate) { +func expectNoPodUpdate(t *testing.T, ch <-chan kubeletTypes.PodUpdate) { select { case update := <-ch: t.Errorf("Expected no update in channel, Got %#v", update) @@ -125,63 +125,63 @@ func TestNewPodAdded(t *testing.T) { channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) // see an update - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")) channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"))) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))) 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) { channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) // see an update - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "")) channel <- podUpdate config.Sync() - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource)) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, kubeletTypes.AllSource)) } func TestNewPodAddedDefaultNamespace(t *testing.T) { channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) // see an update - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "default")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "default")) channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "default"))) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "default"))) 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) { channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental) // see an update - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "default")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "default")) 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 - podUpdate = CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) + podUpdate = CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")) channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new"))) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new"))) 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) { channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental) // see an update - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")) 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 - 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 expectNoPodUpdate(t, ch) } @@ -190,81 +190,81 @@ func TestNewPodAddedSnapshotAndUpdates(t *testing.T) { channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshotAndUpdates) // see an set - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")) channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo", "new"))) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo", "new"))) 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 pod := *podUpdate.Pods[0] pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}} - channel <- CreatePodUpdate(kubelet.ADD, TestSource, &pod) - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, &pod)) + channel <- CreatePodUpdate(kubeletTypes.ADD, TestSource, &pod) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, &pod)) } func TestNewPodAddedSnapshot(t *testing.T) { channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshot) // see an set - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")) channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, CreateValidPod("foo", "new"))) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, TestSource, CreateValidPod("foo", "new"))) 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 pod := *podUpdate.Pods[0] pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent}} - channel <- CreatePodUpdate(kubelet.ADD, TestSource, &pod) - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, &pod)) + channel <- CreatePodUpdate(kubeletTypes.ADD, TestSource, &pod) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.SET, TestSource, &pod)) } func TestNewPodAddedUpdatedRemoved(t *testing.T) { channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental) // should register an add - podUpdate := CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo", "new")) + podUpdate := CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo", "new")) 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 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.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 - 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 - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.REMOVE, TestSource, pod)) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.REMOVE, TestSource, pod)) } func TestNewPodAddedUpdatedSet(t *testing.T) { channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental) // 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 - 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 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.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 expectPodUpdate(t, ch, - CreatePodUpdate(kubelet.REMOVE, TestSource, CreateValidPod("foo", "new")), - CreatePodUpdate(kubelet.ADD, TestSource, CreateValidPod("foo4", "new")), - CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) + CreatePodUpdate(kubeletTypes.REMOVE, TestSource, CreateValidPod("foo", "new")), + CreatePodUpdate(kubeletTypes.ADD, TestSource, CreateValidPod("foo4", "new")), + CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod)) } func TestPodUpdateAnnotations(t *testing.T) { @@ -279,24 +279,24 @@ func TestPodUpdateAnnotations(t *testing.T) { 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 - 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" - 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 - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod)) 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 - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod)) 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 - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod)) } func TestPodUpdateLables(t *testing.T) { @@ -311,13 +311,13 @@ func TestPodUpdateLables(t *testing.T) { t.Fatalf("%v", err) } - podUpdate := CreatePodUpdate(kubelet.SET, TestSource, clone.(*api.Pod)) + podUpdate := CreatePodUpdate(kubeletTypes.SET, TestSource, clone.(*api.Pod)) channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, TestSource, pod)) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.ADD, TestSource, pod)) pod.Labels["key"] = "newValue" - podUpdate = CreatePodUpdate(kubelet.SET, TestSource, pod) + podUpdate = CreatePodUpdate(kubeletTypes.SET, TestSource, pod) channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, TestSource, pod)) + expectPodUpdate(t, ch, CreatePodUpdate(kubeletTypes.UPDATE, TestSource, pod)) } diff --git a/pkg/kubelet/config/file.go b/pkg/kubelet/config/file.go index ba86a397ee4..6ea4671923c 100644 --- a/pkg/kubelet/config/file.go +++ b/pkg/kubelet/config/file.go @@ -26,7 +26,7 @@ import ( "time" "k8s.io/kubernetes/pkg/api" - "k8s.io/kubernetes/pkg/kubelet" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/util" "github.com/golang/glog" @@ -66,7 +66,7 @@ func (s *sourceFile) extractFromPath() error { return err } // 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") } @@ -76,14 +76,14 @@ func (s *sourceFile) extractFromPath() error { if err != nil { 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(): pod, err := s.extractFromFile(path) if err != nil { 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: return fmt.Errorf("path is not a directory or file") diff --git a/pkg/kubelet/config/file_test.go b/pkg/kubelet/config/file_test.go index e4bb154088a..ef1c8ed9a1b 100644 --- a/pkg/kubelet/config/file_test.go +++ b/pkg/kubelet/config/file_test.go @@ -26,7 +26,7 @@ import ( "k8s.io/kubernetes/pkg/api/testapi" "k8s.io/kubernetes/pkg/api/unversioned" "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/securitycontext" "k8s.io/kubernetes/pkg/util" @@ -46,8 +46,8 @@ func TestUpdateOnNonExistentFile(t *testing.T) { NewSourceFile("random_non_existent_path", "localhost", time.Millisecond, ch) select { case got := <-ch: - update := got.(kubelet.PodUpdate) - expected := CreatePodUpdate(kubelet.SET, kubelet.FileSource) + update := got.(kubeletTypes.PodUpdate) + expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.FileSource) if !api.Semantic.DeepDerivative(expected, update) { t.Fatalf("Expected %#v, Got %#v", expected, update) } @@ -75,7 +75,7 @@ func TestReadPodsFromFile(t *testing.T) { var testCases = []struct { desc string pod runtime.Object - expected kubelet.PodUpdate + expected kubeletTypes.PodUpdate }{ { desc: "Simple pod", @@ -94,7 +94,7 @@ func TestReadPodsFromFile(t *testing.T) { SecurityContext: &api.PodSecurityContext{}, }, }, - expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{ + expected: CreatePodUpdate(kubeletTypes.SET, kubeletTypes.FileSource, &api.Pod{ ObjectMeta: api.ObjectMeta{ Name: "test-" + hostname, UID: "12345", @@ -137,7 +137,7 @@ func TestReadPodsFromFile(t *testing.T) { NewSourceFile(file.Name(), hostname, time.Millisecond, ch) select { case got := <-ch: - update := got.(kubelet.PodUpdate) + update := got.(kubeletTypes.PodUpdate) for _, pod := range update.Pods { if errs := validation.ValidatePod(pod); len(errs) > 0 { 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) } - update := (<-ch).(kubelet.PodUpdate) - expected := CreatePodUpdate(kubelet.SET, kubelet.FileSource) + update := (<-ch).(kubeletTypes.PodUpdate) + expected := CreatePodUpdate(kubeletTypes.SET, kubeletTypes.FileSource) if !api.Semantic.DeepEqual(expected, update) { t.Errorf("Expected %#v, Got %#v", expected, update) } diff --git a/pkg/kubelet/config/http.go b/pkg/kubelet/config/http.go index 8d9274b65cf..a3797f26a3a 100644 --- a/pkg/kubelet/config/http.go +++ b/pkg/kubelet/config/http.go @@ -25,7 +25,7 @@ import ( "time" "k8s.io/kubernetes/pkg/api" - "k8s.io/kubernetes/pkg/kubelet" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/util" "github.com/golang/glog" @@ -95,7 +95,7 @@ func (s *sourceURL) extractFromURL() error { } if len(data) == 0 { // 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) } // 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. 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 } @@ -126,7 +126,7 @@ func (s *sourceURL) extractFromURL() error { for i := range podList.Items { 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 } diff --git a/pkg/kubelet/config/http_test.go b/pkg/kubelet/config/http_test.go index 18b52242b68..145d7347d59 100644 --- a/pkg/kubelet/config/http_test.go +++ b/pkg/kubelet/config/http_test.go @@ -27,7 +27,7 @@ import ( "k8s.io/kubernetes/pkg/api/testapi" "k8s.io/kubernetes/pkg/api/unversioned" "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/util" "k8s.io/kubernetes/pkg/util/errors" @@ -128,7 +128,7 @@ func TestExtractPodsFromHTTP(t *testing.T) { var testCases = []struct { desc string pods runtime.Object - expected kubelet.PodUpdate + expected kubeletTypes.PodUpdate }{ { desc: "Single pod", @@ -148,8 +148,8 @@ func TestExtractPodsFromHTTP(t *testing.T) { SecurityContext: &api.PodSecurityContext{}, }, }, - expected: CreatePodUpdate(kubelet.SET, - kubelet.HTTPSource, + expected: CreatePodUpdate(kubeletTypes.SET, + kubeletTypes.HTTPSource, &api.Pod{ ObjectMeta: api.ObjectMeta{ UID: "111", @@ -206,15 +206,15 @@ func TestExtractPodsFromHTTP(t *testing.T) { }, }, }, - expected: CreatePodUpdate(kubelet.SET, - kubelet.HTTPSource, + expected: CreatePodUpdate(kubeletTypes.SET, + kubeletTypes.HTTPSource, &api.Pod{ ObjectMeta: api.ObjectMeta{ UID: "111", Name: "foo" + "-" + hostname, Namespace: "default", - SelfLink: getSelfLink("foo-"+hostname, kubelet.NamespaceDefault), + SelfLink: getSelfLink("foo-"+hostname, kubeletTypes.NamespaceDefault), }, Spec: api.PodSpec{ NodeName: hostname, @@ -237,7 +237,7 @@ func TestExtractPodsFromHTTP(t *testing.T) { Name: "bar" + "-" + hostname, Namespace: "default", - SelfLink: getSelfLink("bar-"+hostname, kubelet.NamespaceDefault), + SelfLink: getSelfLink("bar-"+hostname, kubeletTypes.NamespaceDefault), }, Spec: api.PodSpec{ NodeName: hostname, @@ -279,7 +279,7 @@ func TestExtractPodsFromHTTP(t *testing.T) { t.Errorf("%s: Unexpected error: %v", testCase.desc, err) continue } - update := (<-ch).(kubelet.PodUpdate) + update := (<-ch).(kubeletTypes.PodUpdate) if !api.Semantic.DeepEqual(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 { t.Fatalf("Unexpected error extracting from URL: %v", err) } - update := (<-ch).(kubelet.PodUpdate) + update := (<-ch).(kubeletTypes.PodUpdate) headerVal := fakeHandler.RequestReceived.Header["Metadata-Flavor"] if len(headerVal) != 1 || headerVal[0] != "Google" { diff --git a/pkg/kubelet/fake_pod_workers.go b/pkg/kubelet/fake_pod_workers.go index 14fb671f215..39691f01776 100644 --- a/pkg/kubelet/fake_pod_workers.go +++ b/pkg/kubelet/fake_pod_workers.go @@ -19,6 +19,7 @@ package kubelet import ( "k8s.io/kubernetes/pkg/api" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/types" ) @@ -30,12 +31,12 @@ type fakePodWorkers struct { 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() if err != nil { 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) } } diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index 8c0ed742a00..ff1b3cbca46 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -764,7 +764,7 @@ func (kl *Kubelet) StartGarbageCollection() { } // 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 { 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 } -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) uid := pod.UID start := time.Now() 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) } else { 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. var podStatus api.PodStatus - if updateType == SyncPodCreate { + if updateType == kubeletTypes.SyncPodCreate { // This is the first time we are syncing the pod. Record the latency // since kubelet first saw the pod if firstSeenTime is set. 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 // no changes are seen to the configuration, will synchronize the last known desired // 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.") var housekeepingTimestamp time.Time 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()) select { case u, open := <-updates: @@ -1940,16 +1940,16 @@ func (kl *Kubelet) syncLoopIteration(updates <-chan PodUpdate, handler SyncHandl return false } switch u.Op { - case ADD: + case kubeletTypes.ADD: glog.V(2).Infof("SyncLoop (ADD): %q", kubeletUtil.FormatPodNames(u.Pods)) handler.HandlePodAdditions(u.Pods) - case UPDATE: + case kubeletTypes.UPDATE: glog.V(2).Infof("SyncLoop (UPDATE): %q", kubeletUtil.FormatPodNames(u.Pods)) handler.HandlePodUpdates(u.Pods) - case REMOVE: + case kubeletTypes.REMOVE: glog.V(2).Infof("SyncLoop (REMOVE): %q", kubeletUtil.FormatPodNames(u.Pods)) handler.HandlePodDeletions(u.Pods) - case SET: + case kubeletTypes.SET: // TODO: Do we want to support this? glog.Errorf("Kubelet does not support snapshot update") } @@ -1962,7 +1962,7 @@ func (kl *Kubelet) syncLoopIteration(updates <-chan PodUpdate, handler SyncHandl 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) { 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)) }) // Note the number of containers for new pods. - if syncType == SyncPodCreate { + if syncType == kubeletTypes.SyncPodCreate { 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 // pod exists. 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 } mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod) - kl.dispatchWork(pod, SyncPodCreate, mirrorPod, start) + kl.dispatchWork(pod, kubeletTypes.SyncPodCreate, mirrorPod, start) 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. 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() for _, pod := range pods { mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod) - kl.dispatchWork(pod, SyncPodSync, mirrorPod, start) + kl.dispatchWork(pod, kubeletTypes.SyncPodSync, mirrorPod, start) } } diff --git a/pkg/kubelet/kubelet_test.go b/pkg/kubelet/kubelet_test.go index aa233717531..0b35968d489 100644 --- a/pkg/kubelet/kubelet_test.go +++ b/pkg/kubelet/kubelet_test.go @@ -47,6 +47,7 @@ import ( "k8s.io/kubernetes/pkg/kubelet/network" "k8s.io/kubernetes/pkg/kubelet/prober" "k8s.io/kubernetes/pkg/kubelet/status" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/types" "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) { testKubelet := newTestKubelet(t) @@ -323,12 +324,12 @@ func TestSyncLoopTimeUpdate(t *testing.T) { 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() if loopTime2.IsZero() { 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() if !loopTime3.After(loopTime1) { 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 kubelet.resyncInterval = time.Second * 30 - ch := make(chan PodUpdate) + ch := make(chan kubeletTypes.PodUpdate) close(ch) // 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) { - for _, updateType := range []SyncPodType{SyncPodCreate, SyncPodUpdate} { + for _, updateType := range []kubeletTypes.SyncPodType{kubeletTypes.SyncPodCreate, kubeletTypes.SyncPodUpdate} { testKubelet := newTestKubelet(t) kl := testKubelet.kubelet manager := testKubelet.fakeMirrorClient @@ -2667,7 +2668,7 @@ func TestCreateMirrorPod(t *testing.T) { Name: "bar", Namespace: "foo", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "file", + kubeletTypes.ConfigSourceAnnotationKey: "file", }, }, } @@ -2700,7 +2701,7 @@ func TestDeleteOutdatedMirrorPod(t *testing.T) { Name: "foo", Namespace: "ns", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "file", + kubeletTypes.ConfigSourceAnnotationKey: "file", }, }, Spec: api.PodSpec{ @@ -2716,8 +2717,8 @@ func TestDeleteOutdatedMirrorPod(t *testing.T) { Name: "foo", Namespace: "ns", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "api", - ConfigMirrorAnnotationKey: "mirror", + kubeletTypes.ConfigSourceAnnotationKey: "api", + kubeletTypes.ConfigMirrorAnnotationKey: "mirror", }, }, Spec: api.PodSpec{ @@ -2729,7 +2730,7 @@ func TestDeleteOutdatedMirrorPod(t *testing.T) { pods := []*api.Pod{pod, mirrorPod} kl.podManager.SetPods(pods) - err := kl.syncPod(pod, mirrorPod, container.Pod{}, SyncPodUpdate) + err := kl.syncPod(pod, mirrorPod, container.Pod{}, kubeletTypes.SyncPodUpdate) if err != nil { t.Errorf("unexpected error: %v", err) } @@ -2754,8 +2755,8 @@ func TestDeleteOrphanedMirrorPods(t *testing.T) { Name: "pod1", Namespace: "ns", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "api", - ConfigMirrorAnnotationKey: "mirror", + kubeletTypes.ConfigSourceAnnotationKey: "api", + kubeletTypes.ConfigMirrorAnnotationKey: "mirror", }, }, }, @@ -2765,8 +2766,8 @@ func TestDeleteOrphanedMirrorPods(t *testing.T) { Name: "pod2", Namespace: "ns", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "api", - ConfigMirrorAnnotationKey: "mirror", + kubeletTypes.ConfigSourceAnnotationKey: "api", + kubeletTypes.ConfigMirrorAnnotationKey: "mirror", }, }, }, @@ -2797,7 +2798,7 @@ func TestGetContainerInfoForMirrorPods(t *testing.T) { Name: "qux", Namespace: "ns", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "file", + kubeletTypes.ConfigSourceAnnotationKey: "file", }, }, Spec: api.PodSpec{ @@ -2812,8 +2813,8 @@ func TestGetContainerInfoForMirrorPods(t *testing.T) { Name: "qux", Namespace: "ns", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "api", - ConfigMirrorAnnotationKey: "mirror", + kubeletTypes.ConfigSourceAnnotationKey: "api", + kubeletTypes.ConfigMirrorAnnotationKey: "mirror", }, }, Spec: api.PodSpec{ @@ -2879,7 +2880,7 @@ func TestDoNotCacheStatusForStaticPods(t *testing.T) { Name: "staticFoo", Namespace: "new", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "file", + kubeletTypes.ConfigSourceAnnotationKey: "file", }, }, Spec: api.PodSpec{ @@ -2904,7 +2905,7 @@ func TestHostNetworkAllowed(t *testing.T) { capabilities.SetForTests(capabilities.Capabilities{ PrivilegedSources: capabilities.PrivilegedSources{ - HostNetworkSources: []string{ApiserverSource, FileSource}, + HostNetworkSources: []string{kubeletTypes.ApiserverSource, kubeletTypes.FileSource}, }, }) pod := &api.Pod{ @@ -2913,7 +2914,7 @@ func TestHostNetworkAllowed(t *testing.T) { Name: "foo", Namespace: "new", Annotations: map[string]string{ - ConfigSourceAnnotationKey: FileSource, + kubeletTypes.ConfigSourceAnnotationKey: kubeletTypes.FileSource, }, }, Spec: api.PodSpec{ @@ -2926,7 +2927,7 @@ func TestHostNetworkAllowed(t *testing.T) { }, } 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 { t.Errorf("expected pod infra creation to succeed: %v", err) } @@ -2947,7 +2948,7 @@ func TestHostNetworkDisallowed(t *testing.T) { Name: "foo", Namespace: "new", Annotations: map[string]string{ - ConfigSourceAnnotationKey: FileSource, + kubeletTypes.ConfigSourceAnnotationKey: kubeletTypes.FileSource, }, }, 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 { t.Errorf("expected pod infra creation to fail") } @@ -2986,7 +2987,7 @@ func TestPrivilegeContainerAllowed(t *testing.T) { }, } 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 { 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 { t.Errorf("expected pod infra creation to fail") } diff --git a/pkg/kubelet/mirror_client.go b/pkg/kubelet/mirror_client.go index f692e69edb3..a47a3f2847e 100644 --- a/pkg/kubelet/mirror_client.go +++ b/pkg/kubelet/mirror_client.go @@ -23,6 +23,7 @@ import ( "k8s.io/kubernetes/pkg/api" client "k8s.io/kubernetes/pkg/client/unversioned" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" ) // 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 { copyPod.Annotations[k] = v } - copyPod.Annotations[ConfigMirrorAnnotationKey] = MirrorType + copyPod.Annotations[kubeletTypes.ConfigMirrorAnnotationKey] = kubeletTypes.MirrorType _, err := mc.apiserverClient.Pods(copyPod.Namespace).Create(©Pod) return err @@ -80,7 +81,7 @@ func (mc *basicMirrorClient) DeleteMirrorPod(podFullName string) error { // Helper functions. func getPodSource(pod *api.Pod) (string, error) { if pod.Annotations != nil { - if source, ok := pod.Annotations[ConfigSourceAnnotationKey]; ok { + if source, ok := pod.Annotations[kubeletTypes.ConfigSourceAnnotationKey]; ok { return source, nil } } @@ -89,13 +90,13 @@ func getPodSource(pod *api.Pod) (string, error) { func isStaticPod(pod *api.Pod) bool { source, err := getPodSource(pod) - return err == nil && source != ApiserverSource + return err == nil && source != kubeletTypes.ApiserverSource } func isMirrorPod(pod *api.Pod) bool { - if value, ok := pod.Annotations[ConfigMirrorAnnotationKey]; !ok { + if value, ok := pod.Annotations[kubeletTypes.ConfigMirrorAnnotationKey]; !ok { return false } else { - return value == MirrorType + return value == kubeletTypes.MirrorType } } diff --git a/pkg/kubelet/pod_manager_test.go b/pkg/kubelet/pod_manager_test.go index b658330d7ed..f995e554ff2 100644 --- a/pkg/kubelet/pod_manager_test.go +++ b/pkg/kubelet/pod_manager_test.go @@ -21,6 +21,7 @@ import ( "testing" "k8s.io/kubernetes/pkg/api" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" ) // Stub out mirror client for testing purpose. @@ -40,8 +41,8 @@ func TestGetSetPods(t *testing.T) { Name: "bar", Namespace: "default", Annotations: map[string]string{ - ConfigSourceAnnotationKey: "api", - ConfigMirrorAnnotationKey: "mirror", + kubeletTypes.ConfigSourceAnnotationKey: "api", + kubeletTypes.ConfigMirrorAnnotationKey: "mirror", }, }, } @@ -50,7 +51,7 @@ func TestGetSetPods(t *testing.T) { UID: "123456789", Name: "bar", 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", Name: "taco", Namespace: "default", - Annotations: map[string]string{ConfigSourceAnnotationKey: "api"}, + Annotations: map[string]string{kubeletTypes.ConfigSourceAnnotationKey: "api"}, }, }, staticPod, diff --git a/pkg/kubelet/pod_workers.go b/pkg/kubelet/pod_workers.go index 01a28a25e10..c97561fc129 100644 --- a/pkg/kubelet/pod_workers.go +++ b/pkg/kubelet/pod_workers.go @@ -24,18 +24,19 @@ import ( "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/client/record" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/util" ) // PodWorkers is an abstract interface for testability. 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) 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 { // Protects all per worker fields. @@ -74,7 +75,7 @@ type workUpdate struct { updateCompleteFn func() // A string describing the type of this update, eg: create - updateType SyncPodType + updateType kubeletTypes.SyncPodType } 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. -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 var podUpdates chan workUpdate var exists bool diff --git a/pkg/kubelet/pod_workers_test.go b/pkg/kubelet/pod_workers_test.go index 8ae565ec418..a0cf22031b5 100644 --- a/pkg/kubelet/pod_workers_test.go +++ b/pkg/kubelet/pod_workers_test.go @@ -30,6 +30,7 @@ import ( kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/network" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/types" ) @@ -56,7 +57,7 @@ func createPodWorkers() (*podWorkers, map[types.UID][]string) { fakeRuntimeCache := createFakeRuntimeCache(fakeRecorder) podWorkers := newPodWorkers( 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() { lock.Lock() defer lock.Unlock() @@ -93,7 +94,7 @@ func TestUpdatePod(t *testing.T) { numPods := 20 for i := 0; i < numPods; i++ { 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) @@ -126,7 +127,7 @@ func TestForgetNonExistingPodWorkers(t *testing.T) { numPods := 20 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) @@ -162,12 +163,12 @@ type simpleFakeKubelet struct { 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 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.wg.Done() return nil @@ -353,8 +354,8 @@ func TestFakePodWorkers(t *testing.T) { kubeletForRealWorkers.wg.Add(1) fakeDocker.ContainerList = tt.containerList - realPodWorkers.UpdatePod(tt.pod, tt.mirrorPod, SyncPodUpdate, func() {}) - fakePodWorkers.UpdatePod(tt.pod, tt.mirrorPod, SyncPodUpdate, func() {}) + realPodWorkers.UpdatePod(tt.pod, tt.mirrorPod, kubeletTypes.SyncPodUpdate, func() {}) + fakePodWorkers.UpdatePod(tt.pod, tt.mirrorPod, kubeletTypes.SyncPodUpdate, func() {}) kubeletForRealWorkers.wg.Wait() diff --git a/pkg/kubelet/runonce.go b/pkg/kubelet/runonce.go index e2c992803f8..0b8d5b2255e 100644 --- a/pkg/kubelet/runonce.go +++ b/pkg/kubelet/runonce.go @@ -24,6 +24,7 @@ import ( "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" ) const ( @@ -39,7 +40,7 @@ type RunPodResult struct { } // 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 { case u := <-updates: 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) // We don't create mirror pods in this mode; pass a dummy boolean value // 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) } if retry >= RunOnceMaxRetries { diff --git a/pkg/kubelet/server_test.go b/pkg/kubelet/server_test.go index 664dd18e811..01f4e78e5de 100644 --- a/pkg/kubelet/server_test.go +++ b/pkg/kubelet/server_test.go @@ -37,6 +37,7 @@ import ( apierrs "k8s.io/kubernetes/pkg/api/errors" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/dockertools" + kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/util/httpstream" "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. func getPodName(name, namespace string) string { if namespace == "" { - namespace = NamespaceDefault + namespace = kubeletTypes.NamespaceDefault } return name + "_" + namespace } diff --git a/pkg/kubelet/types.go b/pkg/kubelet/types/pod_update.go similarity index 99% rename from pkg/kubelet/types.go rename to pkg/kubelet/types/pod_update.go index 7d0359cf528..9ef8c9c1617 100644 --- a/pkg/kubelet/types.go +++ b/pkg/kubelet/types/pod_update.go @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package kubelet +package types import ( "fmt" diff --git a/pkg/kubelet/types_test.go b/pkg/kubelet/types/pod_update_test.go similarity index 98% rename from pkg/kubelet/types_test.go rename to pkg/kubelet/types/pod_update_test.go index 057747ab2cb..a753bb58727 100644 --- a/pkg/kubelet/types_test.go +++ b/pkg/kubelet/types/pod_update_test.go @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package kubelet +package types import ( "testing" diff --git a/plugin/pkg/admission/serviceaccount/admission.go b/plugin/pkg/admission/serviceaccount/admission.go index d9178ea013a..26ee52e69ed 100644 --- a/plugin/pkg/admission/serviceaccount/admission.go +++ b/plugin/pkg/admission/serviceaccount/admission.go @@ -29,7 +29,7 @@ import ( client "k8s.io/kubernetes/pkg/client/unversioned" "k8s.io/kubernetes/pkg/controller/serviceaccount" "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/runtime" "k8s.io/kubernetes/pkg/util/sets" diff --git a/plugin/pkg/admission/serviceaccount/admission_test.go b/plugin/pkg/admission/serviceaccount/admission_test.go index 6812265fb8b..65549007e23 100644 --- a/plugin/pkg/admission/serviceaccount/admission_test.go +++ b/plugin/pkg/admission/serviceaccount/admission_test.go @@ -23,7 +23,7 @@ import ( "k8s.io/kubernetes/pkg/admission" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/client/unversioned/testclient" - "k8s.io/kubernetes/pkg/kubelet" + kubelet "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/types" )