Skip to content

K
k3s
Commit 81497f3e authored by Kris Rousey's avatar Kris Rousey
Browse files
Options

Changing the scheduler package to use *api.Pod instead of api.Pod to

avoid unnecessary shallow copies. The change rippled through a lot of code.
parent c3caf397
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 332 additions and 333 deletions
pkg/client/cache/listers.go
View file @ 81497f3e
......@@ -42,11 +42,11 @@ type StoreToPodLister struct {
// TODO Get rid of the selector because that is confusing because the user might not realize that there has already been
// some selection at the caching stage. Also, consistency will facilitate code generation. However, the pkg/client
// is inconsistent too.
func (s *StoreToPodLister) List(selector labels.Selector) (pods []api.Pod, err error) {
func (s *StoreToPodLister) List(selector labels.Selector) (pods []*api.Pod, err error) {
for _, m := range s.Store.List() {
pod := m.(*api.Pod)
if selector.Matches(labels.Set(pod.Labels)) {
pods = append(pods, *pod)
pods = append(pods, pod)
}
}
return pods, nil
......@@ -106,7 +106,7 @@ func (s *StoreToServiceLister) List() (services api.ServiceList, err error) {
// TODO: Move this back to scheduler as a helper function that takes a Store,
// rather than a method of StoreToServiceLister.
func (s *StoreToServiceLister) GetPodServices(pod api.Pod) (services []api.Service, err error) {
func (s *StoreToServiceLister) GetPodServices(pod *api.Pod) (services []api.Service, err error) {
var selector labels.Selector
var service api.Service
......
pkg/kubelet/config/apiserver.go
View file @ 81497f3e
......@@ -34,9 +34,9 @@ func NewSourceApiserver(c *client.Client, hostname string, updates chan<- interf
// newSourceApiserverFromLW holds creates a config source that watches and pulls from the apiserver.
func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{}) {
send := func(objs []interface{}) {
var pods []api.Pod
var pods []*api.Pod
for _, o := range objs {
pods = append(pods, *o.(*api.Pod))
pods = append(pods, o.(*api.Pod))
}
updates <- kubelet.PodUpdate{pods, kubelet.SET, kubelet.ApiserverSource}
}
......
pkg/kubelet/config/apiserver_test.go
View file @ 81497f3e
......@@ -42,20 +42,20 @@ func (lw fakePodLW) Watch(resourceVersion string) (watch.Interface, error) {
var _ cache.ListerWatcher = fakePodLW{}
func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
pod1v1 := api.Pod{
pod1v1 := &api.Pod{
ObjectMeta: api.ObjectMeta{Name: "p"},
Spec: api.PodSpec{Containers: []api.Container{{Image: "image/one"}}}}
pod1v2 := api.Pod{
pod1v2 := &api.Pod{
ObjectMeta: api.ObjectMeta{Name: "p"},
Spec: api.PodSpec{Containers: []api.Container{{Image: "image/two"}}}}
pod2 := api.Pod{
pod2 := &api.Pod{
ObjectMeta: api.ObjectMeta{Name: "q"},
Spec: api.PodSpec{Containers: []api.Container{{Image: "image/blah"}}}}
// Setup fake api client.
fakeWatch := watch.NewFake()
lw := fakePodLW{
listResp: &api.PodList{Items: []api.Pod{pod1v1}},
listResp: &api.PodList{Items: []api.Pod{*pod1v1}},
watchResp: fakeWatch,
}
......@@ -74,7 +74,7 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
}
// Add another pod
fakeWatch.Add(&pod2)
fakeWatch.Add(pod2)
got, ok = <-ch
if !ok {
t.Errorf("Unable to read from channel when expected")
......@@ -89,7 +89,7 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
}
// Modify pod1
fakeWatch.Modify(&pod1v2)
fakeWatch.Modify(pod1v2)
got, ok = <-ch
if !ok {
t.Errorf("Unable to read from channel when expected")
......@@ -103,7 +103,7 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
}
// Delete pod1
fakeWatch.Delete(&pod1v2)
fakeWatch.Delete(pod1v2)
got, ok = <-ch
if !ok {
t.Errorf("Unable to read from channel when expected")
......@@ -115,7 +115,7 @@ func TestNewSourceApiserver_UpdatesAndMultiplePods(t *testing.T) {
}
// Delete pod2
fakeWatch.Delete(&pod2)
fakeWatch.Delete(pod2)
got, ok = <-ch
if !ok {
t.Errorf("Unable to read from channel when expected")
......
pkg/kubelet/config/common.go
View file @ 81497f3e
......@@ -85,7 +85,7 @@ func applyDefaults(pod *api.Pod, source string, isFile bool, hostname string) er
type defaultFunc func(pod *api.Pod) error
func tryDecodeSinglePod(data []byte, defaultFn defaultFunc) (parsed bool, pod api.Pod, err error) {
func tryDecodeSinglePod(data []byte, defaultFn defaultFunc) (parsed bool, pod *api.Pod, err error) {
obj, err := api.Scheme.Decode(data)
if err != nil {
return false, pod, err
......@@ -104,7 +104,7 @@ func tryDecodeSinglePod(data []byte, defaultFn defaultFunc) (parsed bool, pod ap
err = fmt.Errorf("invalid pod: %v", errs)
return true, pod, err
}
return true, *newPod, nil
return true, newPod, nil
}
func tryDecodePodList(data []byte, defaultFn defaultFunc) (parsed bool, pods api.PodList, err error) {
......@@ -132,7 +132,7 @@ func tryDecodePodList(data []byte, defaultFn defaultFunc) (parsed bool, pods api
return true, *newPods, err
}
func tryDecodeSingleManifest(data []byte, defaultFn defaultFunc) (parsed bool, manifest v1beta1.ContainerManifest, pod api.Pod, err error) {
func tryDecodeSingleManifest(data []byte, defaultFn defaultFunc) (parsed bool, manifest v1beta1.ContainerManifest, pod *api.Pod, err error) {
// TODO: should be api.Scheme.Decode
// This is awful. DecodeInto() expects to find an APIObject, which
// Manifest is not. We keep reading manifest for now for compat, but
......@@ -144,6 +144,7 @@ func tryDecodeSingleManifest(data []byte, defaultFn defaultFunc) (parsed bool, m
// avoids writing a v1beta1.ContainerManifest -> api.Pod
// conversion which would be identical to the api.ContainerManifest ->
// api.Pod conversion.
pod = new(api.Pod)
if err = yaml.Unmarshal(data, &manifest); err != nil {
return false, manifest, pod, err
}
......@@ -155,10 +156,10 @@ func tryDecodeSingleManifest(data []byte, defaultFn defaultFunc) (parsed bool, m
err = fmt.Errorf("invalid manifest: %v", errs)
return false, manifest, pod, err
}
if err = api.Scheme.Convert(&newManifest, &pod); err != nil {
if err = api.Scheme.Convert(&newManifest, pod); err != nil {
return true, manifest, pod, err
}
if err := defaultFn(&pod); err != nil {
if err := defaultFn(pod); err != nil {
return true, manifest, pod, err
}
// Success.
......
pkg/kubelet/config/config.go
View file @ 81497f3e
......@@ -168,12 +168,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{s.MergedState().([]api.Pod), kubelet.SET, source}
s.updates <- kubelet.PodUpdate{s.MergedState().([]*api.Pod), kubelet.SET, source}
}
case PodConfigNotificationSnapshot:
if len(updates.Pods) > 0 || len(deletes.Pods) > 0 || len(adds.Pods) > 0 {
s.updates <- kubelet.PodUpdate{s.MergedState().([]api.Pod), kubelet.SET, source}
s.updates <- kubelet.PodUpdate{s.MergedState().([]*api.Pod), kubelet.SET, source}
}
default:
......@@ -212,7 +212,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
if !reflect.DeepEqual(existing.Spec, ref.Spec) {
// this is an update
existing.Spec = ref.Spec
updates.Pods = append(updates.Pods, *existing)
updates.Pods = append(updates.Pods, existing)
continue
}
// this is a no-op
......@@ -224,17 +224,17 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
}
ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source
pods[name] = ref
adds.Pods = append(adds.Pods, *ref)
adds.Pods = append(adds.Pods, ref)
}
case kubelet.REMOVE:
glog.V(4).Infof("Removing a pod %v", update)
for _, value := range update.Pods {
name := kubecontainer.GetPodFullName(&value)
name := kubecontainer.GetPodFullName(value)
if existing, found := pods[name]; found {
// this is a delete
delete(pods, name)
deletes.Pods = append(deletes.Pods, *existing)
deletes.Pods = append(deletes.Pods, existing)
continue
}
// this is a no-op
......@@ -255,7 +255,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
if !reflect.DeepEqual(existing.Spec, ref.Spec) {
// this is an update
existing.Spec = ref.Spec
updates.Pods = append(updates.Pods, *existing)
updates.Pods = append(updates.Pods, existing)
continue
}
// this is a no-op
......@@ -266,13 +266,13 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
}
ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source
pods[name] = ref
adds.Pods = append(adds.Pods, *ref)
adds.Pods = append(adds.Pods, ref)
}
for name, existing := range oldPods {
if _, found := pods[name]; !found {
// this is a delete
deletes.Pods = append(deletes.Pods, *existing)
deletes.Pods = append(deletes.Pods, existing)
}
}
......@@ -297,10 +297,9 @@ func (s *podStorage) seenSources(sources ...string) bool {
return s.sourcesSeen.HasAll(sources...)
}
func filterInvalidPods(pods []api.Pod, source string, recorder record.EventRecorder) (filtered []*api.Pod) {
func filterInvalidPods(pods []*api.Pod, source string, recorder record.EventRecorder) (filtered []*api.Pod) {
names := util.StringSet{}
for i := range pods {
pod := &pods[i]
for i, pod := range pods {
var errlist []error
if errs := validation.ValidatePod(pod); len(errs) != 0 {
errlist = append(errlist, errs...)
......@@ -330,21 +329,21 @@ func filterInvalidPods(pods []api.Pod, source string, recorder record.EventRecor
func (s *podStorage) Sync() {
s.updateLock.Lock()
defer s.updateLock.Unlock()
s.updates <- kubelet.PodUpdate{s.MergedState().([]api.Pod), kubelet.SET, kubelet.AllSource}
s.updates <- kubelet.PodUpdate{s.MergedState().([]*api.Pod), kubelet.SET, kubelet.AllSource}
}
// Object implements config.Accessor
func (s *podStorage) MergedState() interface{} {
s.podLock.RLock()
defer s.podLock.RUnlock()
pods := make([]api.Pod, 0)
pods := make([]*api.Pod, 0)
for _, sourcePods := range s.pods {
for _, podRef := range sourcePods {
pod, err := api.Scheme.Copy(podRef)
if err != nil {
glog.Errorf("unable to copy pod: %v", err)
}
pods = append(pods, *pod.(*api.Pod))
pods = append(pods, pod.(*api.Pod))
}
}
return pods
......
pkg/kubelet/config/config_test.go
View file @ 81497f3e
......@@ -39,7 +39,7 @@ func expectEmptyChannel(t *testing.T, ch <-chan interface{}) {
}
}
type sortedPods []api.Pod
type sortedPods []*api.Pod
func (s sortedPods) Len() int {
return len(s)
......@@ -51,8 +51,8 @@ func (s sortedPods) Less(i, j int) bool {
return s[i].Namespace < s[j].Namespace
}
func CreateValidPod(name, namespace, source string) api.Pod {
return api.Pod{
func CreateValidPod(name, namespace, source string) *api.Pod {
return &api.Pod{
ObjectMeta: api.ObjectMeta{
UID: types.UID(name), // for the purpose of testing, this is unique enough
Name: name,
......@@ -67,12 +67,8 @@ func CreateValidPod(name, namespace, source string) api.Pod {
}
}
func CreatePodUpdate(op kubelet.PodOperation, source string, pods ...api.Pod) kubelet.PodUpdate {
newPods := make([]api.Pod, len(pods))
for i := range pods {
newPods[i] = pods[i]
}
return kubelet.PodUpdate{newPods, op, source}
func CreatePodUpdate(op kubelet.PodOperation, source string, pods ...*api.Pod) kubelet.PodUpdate {
return kubelet.PodUpdate{Pods: pods, Op: op, Source: source}
}
func createPodConfigTester(mode PodConfigNotificationMode) (chan<- interface{}, <-chan kubelet.PodUpdate, *PodConfig) {
......@@ -162,7 +158,7 @@ func TestInvalidPodFiltered(t *testing.T) {
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, NoneSource, CreateValidPod("foo", "new", "test")))
// add an invalid update
podUpdate = CreatePodUpdate(kubelet.UPDATE, NoneSource, api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
podUpdate = CreatePodUpdate(kubelet.UPDATE, NoneSource, &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
channel <- podUpdate
expectNoPodUpdate(t, ch)
}
......@@ -179,10 +175,10 @@ func TestNewPodAddedSnapshotAndUpdates(t *testing.T) {
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource, CreateValidPod("foo", "new", "test")))
// 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}}
channel <- CreatePodUpdate(kubelet.ADD, NoneSource, pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, NoneSource, pod))
channel <- CreatePodUpdate(kubelet.ADD, NoneSource, &pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, NoneSource, &pod))
}
func TestNewPodAddedSnapshot(t *testing.T) {
......@@ -197,10 +193,10 @@ func TestNewPodAddedSnapshot(t *testing.T) {
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, kubelet.AllSource, CreateValidPod("foo", "new", "test")))
// 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}}
channel <- CreatePodUpdate(kubelet.ADD, NoneSource, pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, pod))
channel <- CreatePodUpdate(kubelet.ADD, NoneSource, &pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, TestSource, &pod))
}
func TestNewPodAddedUpdatedRemoved(t *testing.T) {
......@@ -221,7 +217,7 @@ func TestNewPodAddedUpdatedRemoved(t *testing.T) {
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, NoneSource, pod))
podUpdate = CreatePodUpdate(kubelet.REMOVE, NoneSource, api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "new"}})
podUpdate = CreatePodUpdate(kubelet.REMOVE, NoneSource, &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "new"}})
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.REMOVE, NoneSource, pod))
}
......
pkg/kubelet/config/file.go
View file @ 81497f3e
......@@ -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{[]api.Pod{}, kubelet.SET, kubelet.FileSource}
s.updates <- kubelet.PodUpdate{[]*api.Pod{}, kubelet.SET, kubelet.FileSource}
return fmt.Errorf("path does not exist, ignoring")
}
......@@ -83,7 +83,7 @@ func (s *sourceFile) extractFromPath() error {
if err != nil {
return err
}
s.updates <- kubelet.PodUpdate{[]api.Pod{pod}, kubelet.SET, kubelet.FileSource}
s.updates <- kubelet.PodUpdate{[]*api.Pod{pod}, kubelet.SET, kubelet.FileSource}
default:
return fmt.Errorf("path is not a directory or file")
......@@ -95,13 +95,13 @@ func (s *sourceFile) extractFromPath() error {
// Get as many pod configs as we can from a directory. Return an error iff something
// prevented us from reading anything at all. Do not return an error if only some files
// were problematic.
func (s *sourceFile) extractFromDir(name string) ([]api.Pod, error) {
func (s *sourceFile) extractFromDir(name string) ([]*api.Pod, error) {
dirents, err := filepath.Glob(filepath.Join(name, "[^.]*"))
if err != nil {
return nil, fmt.Errorf("glob failed: %v", err)
}
pods := make([]api.Pod, 0)
pods := make([]*api.Pod, 0)
if len(dirents) == 0 {
return pods, nil
}
......@@ -131,7 +131,7 @@ func (s *sourceFile) extractFromDir(name string) ([]api.Pod, error) {
return pods, nil
}
func (s *sourceFile) extractFromFile(filename string) (pod api.Pod, err error) {
func (s *sourceFile) extractFromFile(filename string) (pod *api.Pod, err error) {
glog.V(3).Infof("Reading config file %q", filename)
file, err := os.Open(filename)
if err != nil {
......
pkg/kubelet/config/file_test.go
View file @ 81497f3e
......@@ -83,7 +83,7 @@ func TestReadFromFile(t *testing.T) {
"id": "test",
"containers": [{ "name": "image", "image": "test/image", "imagePullPolicy": "PullAlways"}]
}`,
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, api.Pod{
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{
ObjectMeta: api.ObjectMeta{
Name: "test-" + hostname,
UID: "12345",
......@@ -109,7 +109,7 @@ func TestReadFromFile(t *testing.T) {
"uuid": "12345",
"containers": [{ "name": "image", "image": "test/image", "imagePullPolicy": "PullAlways"}]
}`,
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, api.Pod{
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{
ObjectMeta: api.ObjectMeta{
Name: "12345-" + hostname,
UID: "12345",
......@@ -136,7 +136,7 @@ func TestReadFromFile(t *testing.T) {
"id": "test",
"containers": [{ "name": "image", "image": "test/image", "imagePullPolicy": "PullAlways"}]
}`,
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, api.Pod{
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{
ObjectMeta: api.ObjectMeta{
Name: "test-" + hostname,
UID: "12345",
......@@ -169,7 +169,7 @@ func TestReadFromFile(t *testing.T) {
}
}
}`,
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, api.Pod{
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{
ObjectMeta: api.ObjectMeta{
Name: "test-" + hostname,
UID: "12345",
......@@ -200,7 +200,7 @@ func TestReadFromFile(t *testing.T) {
}
}
}`,
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, api.Pod{
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{
ObjectMeta: api.ObjectMeta{
Name: "12345-" + hostname,
UID: "12345",
......@@ -232,7 +232,7 @@ func TestReadFromFile(t *testing.T) {
"containers": [{ "name": "image", "image": "test/image" }]
}
}`,
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, api.Pod{
expected: CreatePodUpdate(kubelet.SET, kubelet.FileSource, &api.Pod{
ObjectMeta: api.ObjectMeta{
Name: "test-" + hostname,
UID: "12345",
......@@ -264,7 +264,7 @@ func TestReadFromFile(t *testing.T) {
case got := <-ch:
update := got.(kubelet.PodUpdate)
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)
}
}
......@@ -335,7 +335,7 @@ func TestExtractFromEmptyDir(t *testing.T) {
}
}
func ExampleManifestAndPod(id string) (v1beta1.ContainerManifest, api.Pod) {
func ExampleManifestAndPod(id string) (v1beta1.ContainerManifest, *api.Pod) {
hostname := "an-example-host"
manifest := v1beta1.ContainerManifest{
......@@ -358,7 +358,7 @@ func ExampleManifestAndPod(id string) (v1beta1.ContainerManifest, api.Pod) {
},
},
}
expectedPod := api.Pod{
expectedPod := &api.Pod{
ObjectMeta: api.ObjectMeta{
Name: id + "-" + hostname,
UID: types.UID(id),
......@@ -391,7 +391,7 @@ func TestExtractFromDir(t *testing.T) {
manifest2, expectedPod2 := ExampleManifestAndPod("2")
manifests := []v1beta1.ContainerManifest{manifest, manifest2}
pods := []api.Pod{expectedPod, expectedPod2}
pods := []*api.Pod{expectedPod, expectedPod2}
files := make([]*os.File, len(manifests))
dirName, err := ioutil.TempDir("", "foo")
......@@ -433,9 +433,9 @@ func TestExtractFromDir(t *testing.T) {
if !api.Semantic.DeepDerivative(expected, update) {
t.Fatalf("Expected %#v, Got %#v", expected, update)
}
for i := range update.Pods {
if errs := validation.ValidatePod(&update.Pods[i]); len(errs) != 0 {
t.Errorf("Expected no validation errors on %#v, Got %q", update.Pods[i], errs)
for _, pod := range update.Pods {
if errs := validation.ValidatePod(pod); len(errs) != 0 {
t.Errorf("Expected no validation errors on %#v, Got %q", pod, errs)
}
}
}
pkg/kubelet/config/http.go
View file @ 81497f3e
......@@ -74,7 +74,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{[]api.Pod{}, kubelet.SET, kubelet.HTTPSource}
s.updates <- kubelet.PodUpdate{[]*api.Pod{}, kubelet.SET, kubelet.HTTPSource}
return fmt.Errorf("zero-length data received from %v", s.url)
}
// Short circuit if the manifest has not changed since the last time it was read.
......@@ -91,12 +91,12 @@ func (s *sourceURL) extractFromURL() error {
return singleErr
}
// It parsed!
s.updates <- kubelet.PodUpdate{[]api.Pod{pod}, kubelet.SET, kubelet.HTTPSource}
s.updates <- kubelet.PodUpdate{[]*api.Pod{pod}, kubelet.SET, kubelet.HTTPSource}
return nil
}
// That didn't work, so try an array of manifests.
parsed, manifests, pods, multiErr := tryDecodeManifestList(data, s.applyDefaults)
parsed, manifests, podList, multiErr := tryDecodeManifestList(data, s.applyDefaults)
if parsed {
if multiErr != nil {
// It parsed but could not be used.
......@@ -110,7 +110,11 @@ func (s *sourceURL) extractFromURL() error {
return singleErr
}
// It parsed!
s.updates <- kubelet.PodUpdate{pods.Items, kubelet.SET, kubelet.HTTPSource}
pods := make([]*api.Pod, 0)
for i := range podList.Items {
pods = append(pods, &podList.Items[i])
}
s.updates <- kubelet.PodUpdate{pods, kubelet.SET, kubelet.HTTPSource}
return nil
}
......@@ -124,18 +128,22 @@ func (s *sourceURL) extractFromURL() error {
// It parsed but could not be used.
return singlePodErr
}
s.updates <- kubelet.PodUpdate{[]api.Pod{pod}, kubelet.SET, kubelet.HTTPSource}
s.updates <- kubelet.PodUpdate{[]*api.Pod{pod}, kubelet.SET, kubelet.HTTPSource}
return nil
}
// That didn't work, so try a list of pods.
parsed, pods, multiPodErr := tryDecodePodList(data, s.applyDefaults)
parsed, podList, multiPodErr := tryDecodePodList(data, s.applyDefaults)
if parsed {
if multiPodErr != nil {
// It parsed but could not be used.
return multiPodErr
}
s.updates <- kubelet.PodUpdate{pods.Items, kubelet.SET, kubelet.HTTPSource}
pods := make([]*api.Pod, 0)
for i := range podList.Items {
pods = append(pods, &podList.Items[i])
}
s.updates <- kubelet.PodUpdate{pods, kubelet.SET, kubelet.HTTPSource}
return nil
}
......
pkg/kubelet/config/http_test.go
View file @ 81497f3e
......@@ -130,7 +130,7 @@ func TestExtractManifestFromHTTP(t *testing.T) {
Containers: []v1beta1.Container{{Name: "1", Image: "foo", ImagePullPolicy: v1beta1.PullAlways}}},
expected: CreatePodUpdate(kubelet.SET,
kubelet.HTTPSource,
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "111",
Name: "foo" + "-" + hostname,
......@@ -155,7 +155,7 @@ func TestExtractManifestFromHTTP(t *testing.T) {
Containers: []v1beta1.Container{{Name: "ctr", Image: "image", ImagePullPolicy: "IfNotPresent"}}},
expected: CreatePodUpdate(kubelet.SET,
kubelet.HTTPSource,
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "111",
Name: "111" + "-" + hostname,
......@@ -180,7 +180,7 @@ func TestExtractManifestFromHTTP(t *testing.T) {
Containers: []v1beta1.Container{{Name: "1", Image: "foo", ImagePullPolicy: v1beta1.PullAlways}}},
expected: CreatePodUpdate(kubelet.SET,
kubelet.HTTPSource,
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "111",
Name: "foo" + "-" + hostname,
......@@ -209,7 +209,7 @@ func TestExtractManifestFromHTTP(t *testing.T) {
},
expected: CreatePodUpdate(kubelet.SET,
kubelet.HTTPSource,
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "111",
Name: "foo" + "-" + hostname,
......@@ -227,7 +227,7 @@ func TestExtractManifestFromHTTP(t *testing.T) {
ImagePullPolicy: "Always"}},
},
},
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "222",
Name: "bar" + "-" + hostname,
......@@ -283,9 +283,9 @@ func TestExtractManifestFromHTTP(t *testing.T) {
if !api.Semantic.DeepEqual(testCase.expected, update) {
t.Errorf("%s: Expected: %#v, Got: %#v", testCase.desc, testCase.expected, update)
}
for i := range update.Pods {
if errs := validation.ValidatePod(&update.Pods[i]); len(errs) != 0 {
t.Errorf("%s: Expected no validation errors on %#v, Got %v", testCase.desc, update.Pods[i], errors.NewAggregate(errs))
for _, pod := range update.Pods {
if errs := validation.ValidatePod(pod); len(errs) != 0 {
t.Errorf("%s: Expected no validation errors on %#v, Got %v", testCase.desc, pod, errors.NewAggregate(errs))
}
}
}
......@@ -317,7 +317,7 @@ func TestExtractPodsFromHTTP(t *testing.T) {
},
expected: CreatePodUpdate(kubelet.SET,
kubelet.HTTPSource,
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "111",
Name: "foo" + "-" + hostname,
......@@ -355,7 +355,7 @@ func TestExtractPodsFromHTTP(t *testing.T) {
},
expected: CreatePodUpdate(kubelet.SET,
kubelet.HTTPSource,
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "111",
Name: "foo" + "-" + hostname,
......@@ -406,7 +406,7 @@ func TestExtractPodsFromHTTP(t *testing.T) {
},
expected: CreatePodUpdate(kubelet.SET,
kubelet.HTTPSource,
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "111",
Name: "foo" + "-" + hostname,
......@@ -424,7 +424,7 @@ func TestExtractPodsFromHTTP(t *testing.T) {
ImagePullPolicy: "Always"}},
},
},
api.Pod{
&api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "222",
Name: "bar" + "-" + hostname,
......@@ -472,9 +472,9 @@ func TestExtractPodsFromHTTP(t *testing.T) {
if !api.Semantic.DeepEqual(testCase.expected, update) {
t.Errorf("%s: Expected: %#v, Got: %#v", testCase.desc, testCase.expected, update)
}
for i := range update.Pods {
if errs := validation.ValidatePod(&update.Pods[i]); len(errs) != 0 {
t.Errorf("%s: Expected no validation errors on %#v, Got %v", testCase.desc, update.Pods[i], errors.NewAggregate(errs))
for _, pod := range update.Pods {
if errs := validation.ValidatePod(pod); len(errs) != 0 {
t.Errorf("%s: Expected no validation errors on %#v, Got %v", testCase.desc, pod, errors.NewAggregate(errs))
}
}
}
......
pkg/kubelet/kubelet.go
View file @ 81497f3e
......@@ -91,7 +91,7 @@ type SyncHandler interface {
// Syncs current state to match the specified pods. SyncPodType specified what
// type of sync is occuring per pod. StartTime specifies the time at which
// syncing began (for use in monitoring).
SyncPods(pods []api.Pod, podSyncTypes map[types.UID]metrics.SyncPodType, mirrorPods map[string]api.Pod,
SyncPods(pods []*api.Pod, podSyncTypes map[types.UID]metrics.SyncPodType, mirrorPods map[string]api.Pod,
startTime time.Time) error
}
......@@ -1319,7 +1319,7 @@ func (kl *Kubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecont
// Stores all volumes defined by the set of pods into a map.
// Keys for each entry are in the format (POD_ID)/(VOLUME_NAME)
func getDesiredVolumes(pods []api.Pod) map[string]api.Volume {
func getDesiredVolumes(pods []*api.Pod) map[string]api.Volume {
desiredVolumes := make(map[string]api.Volume)
for _, pod := range pods {
for _, volume := range pod.Spec.Volumes {
......@@ -1330,10 +1330,10 @@ func getDesiredVolumes(pods []api.Pod) map[string]api.Volume {
return desiredVolumes
}
func (kl *Kubelet) cleanupOrphanedPods(pods []api.Pod) error {
func (kl *Kubelet) cleanupOrphanedPods(pods []*api.Pod) error {
desired := util.NewStringSet()
for i := range pods {
desired.Insert(string(pods[i].UID))
for _, pod := range pods {
desired.Insert(string(pod.UID))
}
found, err := kl.listPodsFromDisk()
if err != nil {
......@@ -1353,7 +1353,7 @@ func (kl *Kubelet) cleanupOrphanedPods(pods []api.Pod) error {
// Compares the map of current volumes to the map of desired volumes.
// If an active volume does not have a respective desired volume, clean it up.
func (kl *Kubelet) cleanupOrphanedVolumes(pods []api.Pod, running []*docker.Container) error {
func (kl *Kubelet) cleanupOrphanedVolumes(pods []*api.Pod, running []*docker.Container) error {
desiredVolumes := getDesiredVolumes(pods)
currentVolumes := kl.getPodVolumesFromDisk()
runningSet := util.StringSet{}
......@@ -1388,7 +1388,7 @@ func (kl *Kubelet) cleanupOrphanedVolumes(pods []api.Pod, running []*docker.Cont
}
// SyncPods synchronizes the configured list of pods (desired state) with the host current state.
func (kl *Kubelet) SyncPods(allPods []api.Pod, podSyncTypes map[types.UID]metrics.SyncPodType,
func (kl *Kubelet) SyncPods(allPods []*api.Pod, podSyncTypes map[types.UID]metrics.SyncPodType,
mirrorPods map[string]api.Pod, start time.Time) error {
defer func() {
metrics.SyncPodsLatency.Observe(metrics.SinceInMicroseconds(start))
......@@ -1397,15 +1397,15 @@ func (kl *Kubelet) SyncPods(allPods []api.Pod, podSyncTypes map[types.UID]metric
// Remove obsolete entries in podStatus where the pod is no longer considered bound to this node.
podFullNames := make(map[string]bool)
for _, pod := range allPods {
podFullNames[kubecontainer.GetPodFullName(&pod)] = true
podFullNames[kubecontainer.GetPodFullName(pod)] = true
}
kl.statusManager.RemoveOrphanedStatuses(podFullNames)
// Filter out the rejected pod. They don't have running containers.
kl.handleNotFittingPods(allPods)
var pods []api.Pod
var pods []*api.Pod
for _, pod := range allPods {
status, ok := kl.statusManager.GetPodStatus(kubecontainer.GetPodFullName(&pod))
status, ok := kl.statusManager.GetPodStatus(kubecontainer.GetPodFullName(pod))
if ok && status.Phase == api.PodFailed {
continue
}
......@@ -1423,8 +1423,7 @@ func (kl *Kubelet) SyncPods(allPods []api.Pod, podSyncTypes map[types.UID]metric
}
// Check for any containers that need starting
for ix := range pods {
pod := &pods[ix]
for _, pod := range pods {
podFullName := kubecontainer.GetPodFullName(pod)
uid := pod.UID
desiredPods[uid] = empty{}
......@@ -1504,7 +1503,7 @@ func (kl *Kubelet) SyncPods(allPods []api.Pod, podSyncTypes map[types.UID]metric
return err
}
type podsByCreationTime []api.Pod
type podsByCreationTime []*api.Pod
func (s podsByCreationTime) Len() int {
return len(s)
......@@ -1519,31 +1518,30 @@ func (s podsByCreationTime) Less(i, j int) bool {
}
// checkHostPortConflicts detects pods with conflicted host ports.
func checkHostPortConflicts(pods []api.Pod) (fitting []api.Pod, notFitting []api.Pod) {
func checkHostPortConflicts(pods []*api.Pod) (fitting []*api.Pod, notFitting []*api.Pod) {
ports := map[int]bool{}
extract := func(p *api.ContainerPort) int { return p.HostPort }
// Respect the pod creation order when resolving conflicts.
sort.Sort(podsByCreationTime(pods))
for i := range pods {
pod := &pods[i]
for _, pod := range pods {
if errs := validation.AccumulateUniquePorts(pod.Spec.Containers, ports, extract); len(errs) != 0 {
glog.Errorf("Pod %q: HostPort is already allocated, ignoring: %v", kubecontainer.GetPodFullName(pod), errs)
notFitting = append(notFitting, *pod)
notFitting = append(notFitting, pod)
continue
}
fitting = append(fitting, *pod)
fitting = append(fitting, pod)
}
return
}
// checkCapacityExceeded detects pods that exceeds node's resources.
func (kl *Kubelet) checkCapacityExceeded(pods []api.Pod) (fitting []api.Pod, notFitting []api.Pod) {
func (kl *Kubelet) checkCapacityExceeded(pods []*api.Pod) (fitting []*api.Pod, notFitting []*api.Pod) {
info, err := kl.GetCachedMachineInfo()
if err != nil {
glog.Errorf("error getting machine info: %v", err)
return pods, []api.Pod{}
return pods, nil
}
// Respect the pod creation order when resolving conflicts.
......@@ -1554,14 +1552,14 @@ func (kl *Kubelet) checkCapacityExceeded(pods []api.Pod) (fitting []api.Pod, not
}
// checkNodeSelectorMatching detects pods that do not match node's labels.
func (kl *Kubelet) checkNodeSelectorMatching(pods []api.Pod) (fitting []api.Pod, notFitting []api.Pod) {
func (kl *Kubelet) checkNodeSelectorMatching(pods []*api.Pod) (fitting []*api.Pod, notFitting []*api.Pod) {
node, err := kl.GetNode()
if err != nil {
glog.Errorf("error getting node: %v", err)
return pods, []api.Pod{}
return pods, nil
}
for _, pod := range pods {
if !scheduler.PodMatchesNodeLabels(&pod, node) {
if !scheduler.PodMatchesNodeLabels(pod, node) {
notFitting = append(notFitting, pod)
continue
}
......@@ -1572,25 +1570,25 @@ func (kl *Kubelet) checkNodeSelectorMatching(pods []api.Pod) (fitting []api.Pod,
// handleNotfittingPods handles pods that do not fit on the node.
// Currently conflicts on Port.HostPort values, matching node's labels and exceeding node's capacity are handled.
func (kl *Kubelet) handleNotFittingPods(pods []api.Pod) {
func (kl *Kubelet) handleNotFittingPods(pods []*api.Pod) {
fitting, notFitting := checkHostPortConflicts(pods)
for _, pod := range notFitting {
kl.recorder.Eventf(&pod, "hostPortConflict", "Cannot start the pod due to host port conflict.")
kl.statusManager.SetPodStatus(&pod, api.PodStatus{
kl.recorder.Eventf(pod, "hostPortConflict", "Cannot start the pod due to host port conflict.")
kl.statusManager.SetPodStatus(pod, api.PodStatus{
Phase: api.PodFailed,
Message: "Pod cannot be started due to host port conflict"})
}
fitting, notFitting = kl.checkNodeSelectorMatching(fitting)
for _, pod := range notFitting {
kl.recorder.Eventf(&pod, "nodeSelectorMismatching", "Cannot start the pod due to node selector mismatch.")
kl.statusManager.SetPodStatus(&pod, api.PodStatus{
kl.recorder.Eventf(pod, "nodeSelectorMismatching", "Cannot start the pod due to node selector mismatch.")
kl.statusManager.SetPodStatus(pod, api.PodStatus{
Phase: api.PodFailed,
Message: "Pod cannot be started due to node selector mismatch"})
}
fitting, notFitting = kl.checkCapacityExceeded(fitting)
for _, pod := range notFitting {
kl.recorder.Eventf(&pod, "capacityExceeded", "Cannot start the pod due to exceeded capacity.")
kl.statusManager.SetPodStatus(&pod, api.PodStatus{
kl.recorder.Eventf(pod, "capacityExceeded", "Cannot start the pod due to exceeded capacity.")
kl.statusManager.SetPodStatus(pod, api.PodStatus{
Phase: api.PodFailed,
Message: "Pod cannot be started due to exceeded capacity"})
}
......@@ -1716,7 +1714,7 @@ func (kl *Kubelet) GetHostIP() (net.IP, error) {
// GetPods returns all pods bound to the kubelet and their spec, and the mirror
// pods.
func (kl *Kubelet) GetPods() []api.Pod {
func (kl *Kubelet) GetPods() []*api.Pod {
return kl.podManager.GetPods()
}
......
pkg/kubelet/pod_manager.go
View file @ 81497f3e
......@@ -43,11 +43,11 @@ import (
// also be removed.
type podManager interface {
GetPods() []api.Pod
GetPods() []*api.Pod
GetPodByFullName(podFullName string) (*api.Pod, bool)
GetPodByName(namespace, name string) (*api.Pod, bool)
GetPodsAndMirrorMap() ([]api.Pod, map[string]api.Pod)
SetPods(pods []api.Pod)
GetPodsAndMirrorMap() ([]*api.Pod, map[string]api.Pod)
SetPods(pods []*api.Pod)
UpdatePods(u PodUpdate, podSyncTypes map[types.UID]metrics.SyncPodType)
DeleteOrphanedMirrorPods()
TranslatePodUID(uid types.UID) types.UID
......@@ -78,7 +78,7 @@ type basicPodManager struct {
func newBasicPodManager(apiserverClient client.Interface) *basicPodManager {
pm := &basicPodManager{}
pm.mirrorClient = newBasicMirrorClient(apiserverClient)
pm.SetPods([]api.Pod{})
pm.SetPods(nil)
return pm
}
......@@ -127,27 +127,26 @@ func (self *basicPodManager) UpdatePods(u PodUpdate, podSyncTypes map[types.UID]
}
// Set the internal pods based on the new pods.
func (self *basicPodManager) SetPods(newPods []api.Pod) {
func (self *basicPodManager) SetPods(newPods []*api.Pod) {
self.lock.Lock()
defer self.lock.Unlock()
self.setPods(newPods)
}
func (self *basicPodManager) setPods(newPods []api.Pod) {
func (self *basicPodManager) setPods(newPods []*api.Pod) {
podByUID := make(map[types.UID]*api.Pod)
mirrorPodByUID := make(map[types.UID]*api.Pod)
podByFullName := make(map[string]*api.Pod)
mirrorPodByFullName := make(map[string]*api.Pod)
for i := range newPods {
pod := newPods[i]
podFullName := kubecontainer.GetPodFullName(&pod)
if isMirrorPod(&pod) {
mirrorPodByUID[pod.UID] = &pod
mirrorPodByFullName[podFullName] = &pod
for _, pod := range newPods {
podFullName := kubecontainer.GetPodFullName(pod)
if isMirrorPod(pod) {
mirrorPodByUID[pod.UID] = pod
mirrorPodByFullName[podFullName] = pod
} else {
podByUID[pod.UID] = &pod
podByFullName[podFullName] = &pod
podByUID[pod.UID] = pod
podByFullName[podFullName] = pod
}
}
......@@ -157,21 +156,19 @@ func (self *basicPodManager) setPods(newPods []api.Pod) {
self.mirrorPodByFullName = mirrorPodByFullName
}
func applyUpdates(changed []api.Pod, current []api.Pod) []api.Pod {
updated := []api.Pod{}
func applyUpdates(changed []*api.Pod, current []*api.Pod) []*api.Pod {
updated := []*api.Pod{}
m := map[types.UID]*api.Pod{}
for i := range changed {
pod := &changed[i]
for _, pod := range changed {
m[pod.UID] = pod
}
for i := range current {
pod := &current[i]
for _, pod := range current {
if m[pod.UID] != nil {
updated = append(updated, *m[pod.UID])
updated = append(updated, m[pod.UID])
glog.V(4).Infof("pod with UID: %q has a new spec %+v", pod.UID, *m[pod.UID])
} else {
updated = append(updated, *pod)
updated = append(updated, pod)
glog.V(4).Infof("pod with UID: %q stay with the same spec %+v", pod.UID, *pod)
}
}
......@@ -180,20 +177,20 @@ func applyUpdates(changed []api.Pod, current []api.Pod) []api.Pod {
}
// GetPods returns the regular pods bound to the kubelet and their spec.
func (self *basicPodManager) GetPods() []api.Pod {
func (self *basicPodManager) GetPods() []*api.Pod {
self.lock.RLock()
defer self.lock.RUnlock()
return podsMapToPods(self.podByUID)
}
// Returns all pods (including mirror pods).
func (self *basicPodManager) getAllPods() []api.Pod {
func (self *basicPodManager) getAllPods() []*api.Pod {
return append(podsMapToPods(self.podByUID), podsMapToPods(self.mirrorPodByUID)...)
}
// GetPodsAndMirrorMap returns the a copy of the regular pods and the mirror
// pods indexed by full name.
func (self *basicPodManager) GetPodsAndMirrorMap() ([]api.Pod, map[string]api.Pod) {
func (self *basicPodManager) GetPodsAndMirrorMap() ([]*api.Pod, map[string]api.Pod) {
self.lock.RLock()
defer self.lock.RUnlock()
mirrorPods := make(map[string]api.Pod)
......@@ -270,10 +267,10 @@ func (self *basicPodManager) IsMirrorPodOf(mirrorPod, pod *api.Pod) bool {
return api.Semantic.DeepEqual(&pod.Spec, &mirrorPod.Spec)
}
func podsMapToPods(UIDMap map[types.UID]*api.Pod) []api.Pod {
pods := make([]api.Pod, 0, len(UIDMap))
func podsMapToPods(UIDMap map[types.UID]*api.Pod) []*api.Pod {
pods := make([]*api.Pod, 0, len(UIDMap))
for _, pod := range UIDMap {
pods = append(pods, *pod)
pods = append(pods, pod)
}
return pods
}
pkg/kubelet/pod_manager_test.go
View file @ 81497f3e
......@@ -34,7 +34,7 @@ func newFakePodManager() (*basicPodManager, *fakeMirrorClient) {
// Tests that pods/maps are properly set after the pod update, and the basic
// methods work correctly.
func TestGetSetPods(t *testing.T) {
mirrorPod := api.Pod{
mirrorPod := &api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "987654321",
Name: "bar",
......@@ -45,7 +45,7 @@ func TestGetSetPods(t *testing.T) {
},
},
}
staticPod := api.Pod{
staticPod := &api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "123456789",
Name: "bar",
......@@ -54,7 +54,7 @@ func TestGetSetPods(t *testing.T) {
},
}
expectedPods := []api.Pod{
expectedPods := []*api.Pod{
{
ObjectMeta: api.ObjectMeta{
UID: "999999999",
......@@ -98,11 +98,11 @@ func TestGetSetPods(t *testing.T) {
// Test the basic Get methods.
actualPod, ok := podManager.GetPodByFullName("bar_default")
if !ok || !reflect.DeepEqual(actualPod, &staticPod) {
if !ok || !reflect.DeepEqual(actualPod, staticPod) {
t.Errorf("unable to get pod by full name; expected: %#v, got: %#v", staticPod, actualPod)
}
actualPod, ok = podManager.GetPodByName("default", "bar")
if !ok || !reflect.DeepEqual(actualPod, &staticPod) {
if !ok || !reflect.DeepEqual(actualPod, staticPod) {
t.Errorf("unable to get pod by name; expected: %#v, got: %#v", staticPod, actualPod)
}
......
pkg/kubelet/runonce.go
View file @ 81497f3e
......@@ -51,16 +51,15 @@ func (kl *Kubelet) RunOnce(updates <-chan PodUpdate) ([]RunPodResult, error) {
}
// runOnce runs a given set of pods and returns their status.
func (kl *Kubelet) runOnce(pods []api.Pod, retryDelay time.Duration) (results []RunPodResult, err error) {
func (kl *Kubelet) runOnce(pods []*api.Pod, retryDelay time.Duration) (results []RunPodResult, err error) {
kl.handleNotFittingPods(pods)
ch := make(chan RunPodResult)
for i := range pods {
pod := pods[i] // Make a copy
go func() {
for _, pod := range pods {
go func(pod *api.Pod) {
err := kl.runPod(pod, retryDelay)
ch <- RunPodResult{&pod, err}
}()
ch <- RunPodResult{pod, err}
}(pod)
}
glog.Infof("waiting for %d pods", len(pods))
......@@ -84,7 +83,7 @@ func (kl *Kubelet) runOnce(pods []api.Pod, retryDelay time.Duration) (results []
}
// runPod runs a single pod and wait until all containers are running.
func (kl *Kubelet) runPod(pod api.Pod, retryDelay time.Duration) error {
func (kl *Kubelet) runPod(pod *api.Pod, retryDelay time.Duration) error {
delay := retryDelay
retry := 0
for {
......@@ -104,7 +103,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); err != nil {
if err = kl.syncPod(pod, nil, p); err != nil {
return fmt.Errorf("error syncing pod: %v", err)
}
if retry >= RunOnceMaxRetries {
......@@ -119,7 +118,7 @@ func (kl *Kubelet) runPod(pod api.Pod, retryDelay time.Duration) error {
}
// isPodRunning returns true if all containers of a manifest are running.
func (kl *Kubelet) isPodRunning(pod api.Pod, runningPod container.Pod) (bool, error) {
func (kl *Kubelet) isPodRunning(pod *api.Pod, runningPod container.Pod) (bool, error) {
for _, container := range pod.Spec.Containers {
c := runningPod.FindContainerByName(container.Name)
if c == nil {
......
pkg/kubelet/runonce_test.go
View file @ 81497f3e
......@@ -148,7 +148,7 @@ func TestRunOnce(t *testing.T) {
kb.containerManager = dockertools.NewDockerManager(kb.dockerClient, kb.recorder, dockertools.PodInfraContainerImage, 0, 0)
kb.containerManager.Puller = &dockertools.FakeDockerPuller{}
pods := []api.Pod{
pods := []*api.Pod{
{
ObjectMeta: api.ObjectMeta{
UID: "12345678",
......
pkg/kubelet/server.go
View file @ 81497f3e
......@@ -102,7 +102,7 @@ type HostInterface interface {
GetRootInfo(req *cadvisorApi.ContainerInfoRequest) (*cadvisorApi.ContainerInfo, error)
GetDockerVersion() ([]uint, error)
GetCachedMachineInfo() (*cadvisorApi.MachineInfo, error)
GetPods() []api.Pod
GetPods() []*api.Pod
GetPodByName(namespace, name string) (*api.Pod, bool)
GetPodStatus(name string) (api.PodStatus, error)
RunInContainer(name string, uid types.UID, container string, cmd []string) ([]byte, error)
......@@ -279,8 +279,9 @@ func (s *Server) handleContainerLogs(w http.ResponseWriter, req *http.Request) {
// handlePods returns a list of pod bound to the Kubelet and their spec
func (s *Server) handlePods(w http.ResponseWriter, req *http.Request) {
pods := s.host.GetPods()
podList := &api.PodList{
Items: pods,
podList := new(api.PodList)
for _, pod := range pods {
podList.Items = append(podList.Items, *pod)
}
data, err := latest.Codec.Encode(podList)
if err != nil {
......
pkg/kubelet/server_test.go
View file @ 81497f3e
......@@ -44,7 +44,7 @@ type fakeKubelet struct {
containerInfoFunc func(podFullName string, uid types.UID, containerName string, req *cadvisorApi.ContainerInfoRequest) (*cadvisorApi.ContainerInfo, error)
rootInfoFunc func(query *cadvisorApi.ContainerInfoRequest) (*cadvisorApi.ContainerInfo, error)
machineInfoFunc func() (*cadvisorApi.MachineInfo, error)
podsFunc func() []api.Pod
podsFunc func() []*api.Pod
logFunc func(w http.ResponseWriter, req *http.Request)
runFunc func(podFullName string, uid types.UID, containerName string, cmd []string) ([]byte, error)
dockerVersionFunc func() ([]uint, error)
......@@ -79,7 +79,7 @@ func (fk *fakeKubelet) GetCachedMachineInfo() (*cadvisorApi.MachineInfo, error)
return fk.machineInfoFunc()
}
func (fk *fakeKubelet) GetPods() []api.Pod {
func (fk *fakeKubelet) GetPods() []*api.Pod {
return fk.podsFunc()
}
......
pkg/kubelet/types.go
View file @ 81497f3e
......@@ -64,7 +64,7 @@ const (
// functionally similar, this helps our unit tests properly check that the correct PodUpdates
// are generated.
type PodUpdate struct {
Pods []api.Pod
Pods []*api.Pod
Op PodOperation
Source string
}
......
pkg/scheduler/generic_scheduler.go
View file @ 81497f3e
......@@ -30,7 +30,7 @@ import (
type FailedPredicateMap map[string]util.StringSet
type FitError struct {
Pod api.Pod
Pod *api.Pod
FailedPredicates FailedPredicateMap
}
......@@ -51,7 +51,7 @@ type genericScheduler struct {
randomLock sync.Mutex
}
func (g *genericScheduler) Schedule(pod api.Pod, minionLister MinionLister) (string, error) {
func (g *genericScheduler) Schedule(pod *api.Pod, minionLister MinionLister) (string, error) {
minions, err := minionLister.List()
if err != nil {
return "", err
......@@ -97,7 +97,7 @@ func (g *genericScheduler) selectHost(priorityList HostPriorityList) (string, er
// Filters the minions to find the ones that fit based on the given predicate functions
// Each minion is passed through the predicate functions to determine if it is a fit
func findNodesThatFit(pod api.Pod, podLister PodLister, predicates map[string]FitPredicate, nodes api.NodeList) (api.NodeList, FailedPredicateMap, error) {
func findNodesThatFit(pod *api.Pod, podLister PodLister, predicates map[string]FitPredicate, nodes api.NodeList) (api.NodeList, FailedPredicateMap, error) {
filtered := []api.Node{}
machineToPods, err := MapPodsToMachines(podLister)
failedPredicateMap := FailedPredicateMap{}
......@@ -133,7 +133,7 @@ func findNodesThatFit(pod api.Pod, podLister PodLister, predicates map[string]Fi
// Each priority function can also have its own weight
// The minion scores returned by the priority function are multiplied by the weights to get weighted scores
// All scores are finally combined (added) to get the total weighted scores of all minions
func prioritizeNodes(pod api.Pod, podLister PodLister, priorityConfigs []PriorityConfig, minionLister MinionLister) (HostPriorityList, error) {
func prioritizeNodes(pod *api.Pod, podLister PodLister, priorityConfigs []PriorityConfig, minionLister MinionLister) (HostPriorityList, error) {
result := HostPriorityList{}
// If no priority configs are provided, then the EqualPriority function is applied
......@@ -177,7 +177,7 @@ func getBestHosts(list HostPriorityList) []string {
}
// EqualPriority is a prioritizer function that gives an equal weight of one to all nodes
func EqualPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
func EqualPriority(_ *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
nodes, err := minionLister.List()
if err != nil {
fmt.Errorf("failed to list nodes: %v", err)
......
pkg/scheduler/generic_scheduler_test.go
View file @ 81497f3e
......@@ -27,19 +27,19 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
)
func falsePredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func falsePredicate(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
return false, nil
}
func truePredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func truePredicate(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
return true, nil
}
func matchesPredicate(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func matchesPredicate(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
return pod.Name == node, nil
}
func numericPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
func numericPriority(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
nodes, err := minionLister.List()
result := []HostPriority{}
......@@ -60,7 +60,7 @@ func numericPriority(pod api.Pod, podLister PodLister, minionLister MinionLister
return result, nil
}
func reverseNumericPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
func reverseNumericPriority(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
var maxScore float64
minScore := math.MaxFloat64
reverseResult := []HostPriority{}
......@@ -165,7 +165,7 @@ func TestGenericScheduler(t *testing.T) {
predicates map[string]FitPredicate
prioritizers []PriorityConfig
nodes []string
pod api.Pod
pod *api.Pod
expectedHost string
expectsErr bool
}{
......@@ -189,7 +189,7 @@ func TestGenericScheduler(t *testing.T) {
predicates: map[string]FitPredicate{"matches": matchesPredicate},
prioritizers: []PriorityConfig{{Function: EqualPriority, Weight: 1}},
nodes: []string{"machine1", "machine2"},
pod: api.Pod{ObjectMeta: api.ObjectMeta{Name: "machine2"}},
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "machine2"}},
expectedHost: "machine2",
name: "test 3",
},
......@@ -204,7 +204,7 @@ func TestGenericScheduler(t *testing.T) {
predicates: map[string]FitPredicate{"matches": matchesPredicate},
prioritizers: []PriorityConfig{{Function: numericPriority, Weight: 1}},
nodes: []string{"3", "2", "1"},
pod: api.Pod{ObjectMeta: api.ObjectMeta{Name: "2"}},
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "2"}},
expectedHost: "2",
name: "test 5",
},
......@@ -212,7 +212,7 @@ func TestGenericScheduler(t *testing.T) {
predicates: map[string]FitPredicate{"true": truePredicate},
prioritizers: []PriorityConfig{{Function: numericPriority, Weight: 1}, {Function: reverseNumericPriority, Weight: 2}},
nodes: []string{"3", "2", "1"},
pod: api.Pod{ObjectMeta: api.ObjectMeta{Name: "2"}},
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Name: "2"}},
expectedHost: "1",
name: "test 6",
},
......@@ -227,7 +227,7 @@ func TestGenericScheduler(t *testing.T) {
for _, test := range tests {
random := rand.New(rand.NewSource(0))
scheduler := NewGenericScheduler(test.predicates, test.prioritizers, FakePodLister([]api.Pod{}), random)
scheduler := NewGenericScheduler(test.predicates, test.prioritizers, FakePodLister([]*api.Pod{}), random)
machine, err := scheduler.Schedule(test.pod, FakeMinionLister(makeNodeList(test.nodes)))
if test.expectsErr {
if err == nil {
......@@ -247,7 +247,7 @@ func TestGenericScheduler(t *testing.T) {
func TestFindFitAllError(t *testing.T) {
nodes := []string{"3", "2", "1"}
predicates := map[string]FitPredicate{"true": truePredicate, "false": falsePredicate}
_, predicateMap, err := findNodesThatFit(api.Pod{}, FakePodLister([]api.Pod{}), predicates, makeNodeList(nodes))
_, predicateMap, err := findNodesThatFit(&api.Pod{}, FakePodLister([]*api.Pod{}), predicates, makeNodeList(nodes))
if err != nil {
t.Errorf("unexpected error: %v", err)
......@@ -271,8 +271,8 @@ func TestFindFitAllError(t *testing.T) {
func TestFindFitSomeError(t *testing.T) {
nodes := []string{"3", "2", "1"}
predicates := map[string]FitPredicate{"true": truePredicate, "match": matchesPredicate}
pod := api.Pod{ObjectMeta: api.ObjectMeta{Name: "1"}}
_, predicateMap, err := findNodesThatFit(pod, FakePodLister([]api.Pod{}), predicates, makeNodeList(nodes))
pod := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "1"}}
_, predicateMap, err := findNodesThatFit(pod, FakePodLister([]*api.Pod{}), predicates, makeNodeList(nodes))
if err != nil {
t.Errorf("unexpected error: %v", err)
......
pkg/scheduler/listers.go
View file @ 81497f3e
......@@ -39,14 +39,14 @@ func (f FakeMinionLister) List() (api.NodeList, error) {
// PodLister interface represents anything that can list pods for a scheduler.
type PodLister interface {
// TODO: make this exactly the same as client's Pods(ns).List() method, by returning a api.PodList
List(labels.Selector) ([]api.Pod, error)
List(labels.Selector) ([]*api.Pod, error)
}
// FakePodLister implements PodLister on an []api.Pods for test purposes.
type FakePodLister []api.Pod
type FakePodLister []*api.Pod
// List returns []api.Pod matching a query.
func (f FakePodLister) List(s labels.Selector) (selected []api.Pod, err error) {
func (f FakePodLister) List(s labels.Selector) (selected []*api.Pod, err error) {
for _, pod := range f {
if s.Matches(labels.Set(pod.Labels)) {
selected = append(selected, pod)
......@@ -60,7 +60,7 @@ type ServiceLister interface {
// Lists all the services
List() (api.ServiceList, error)
// Gets the services for the given pod
GetPodServices(api.Pod) ([]api.Service, error)
GetPodServices(*api.Pod) ([]api.Service, error)
}
// FakeServiceLister implements ServiceLister on []api.Service for test purposes.
......@@ -72,7 +72,7 @@ func (f FakeServiceLister) List() (api.ServiceList, error) {
}
// GetPodServices gets the services that have the selector that match the labels on the given pod
func (f FakeServiceLister) GetPodServices(pod api.Pod) (services []api.Service, err error) {
func (f FakeServiceLister) GetPodServices(pod *api.Pod) (services []api.Service, err error) {
var selector labels.Selector
for _, service := range f {
......
pkg/scheduler/predicates.go
View file @ 81497f3e
......@@ -80,11 +80,11 @@ func isVolumeConflict(volume api.Volume, pod *api.Pod) bool {
// are exclusive so if there is already a volume mounted on that node, another pod can't schedule
// there. This is GCE specific for now.
// TODO: migrate this into some per-volume specific code?
func NoDiskConflict(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func NoDiskConflict(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
manifest := &(pod.Spec)
for ix := range manifest.Volumes {
for podIx := range existingPods {
if isVolumeConflict(manifest.Volumes[ix], &existingPods[podIx]) {
if isVolumeConflict(manifest.Volumes[ix], existingPods[podIx]) {
return false, nil
}
}
......@@ -111,31 +111,31 @@ func getResourceRequest(pod *api.Pod) resourceRequest {
return result
}
func CheckPodsExceedingCapacity(pods []api.Pod, capacity api.ResourceList) (fitting []api.Pod, notFitting []api.Pod) {
func CheckPodsExceedingCapacity(pods []*api.Pod, capacity api.ResourceList) (fitting []*api.Pod, notFitting []*api.Pod) {
totalMilliCPU := capacity.Cpu().MilliValue()
totalMemory := capacity.Memory().Value()
milliCPURequested := int64(0)
memoryRequested := int64(0)
for ix := range pods {
podRequest := getResourceRequest(&pods[ix])
for _, pod := range pods {
podRequest := getResourceRequest(pod)
fitsCPU := totalMilliCPU == 0 || (totalMilliCPU-milliCPURequested) >= podRequest.milliCPU
fitsMemory := totalMemory == 0 || (totalMemory-memoryRequested) >= podRequest.memory
if !fitsCPU || !fitsMemory {
// the pod doesn't fit
notFitting = append(notFitting, pods[ix])
notFitting = append(notFitting, pod)
continue
}
// the pod fits
milliCPURequested += podRequest.milliCPU
memoryRequested += podRequest.memory
fitting = append(fitting, pods[ix])
fitting = append(fitting, pod)
}
return
}
// PodFitsResources calculates fit based on requested, rather than used resources
func (r *ResourceFit) PodFitsResources(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
podRequest := getResourceRequest(&pod)
func (r *ResourceFit) PodFitsResources(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
podRequest := getResourceRequest(pod)
if podRequest.milliCPU == 0 && podRequest.memory == 0 {
// no resources requested always fits.
return true, nil
......@@ -144,7 +144,7 @@ func (r *ResourceFit) PodFitsResources(pod api.Pod, existingPods []api.Pod, node
if err != nil {
return false, err
}
pods := []api.Pod{}
pods := []*api.Pod{}
copy(pods, existingPods)
pods = append(existingPods, pod)
_, exceeding := CheckPodsExceedingCapacity(pods, info.Status.Capacity)
......@@ -180,15 +180,15 @@ type NodeSelector struct {
info NodeInfo
}
func (n *NodeSelector) PodSelectorMatches(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func (n *NodeSelector) PodSelectorMatches(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
minion, err := n.info.GetNodeInfo(node)
if err != nil {
return false, err
}
return PodMatchesNodeLabels(&pod, minion), nil
return PodMatchesNodeLabels(pod, minion), nil
}
func PodFitsHost(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func PodFitsHost(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
if len(pod.Spec.Host) == 0 {
return true, nil
}
......@@ -222,7 +222,7 @@ func NewNodeLabelPredicate(info NodeInfo, labels []string, presence bool) FitPre
// Alternately, eliminating minions that have a certain label, regardless of value, is also useful
// A minion may have a label with "retiring" as key and the date as the value
// and it may be desirable to avoid scheduling new pods on this minion
func (n *NodeLabelChecker) CheckNodeLabelPresence(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func (n *NodeLabelChecker) CheckNodeLabelPresence(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
var exists bool
minion, err := n.info.GetNodeInfo(node)
if err != nil {
......@@ -264,7 +264,7 @@ func NewServiceAffinityPredicate(podLister PodLister, serviceLister ServiceListe
// - L is listed in the ServiceAffinity object that is passed into the function
// - the pod does not have any NodeSelector for L
// - some other pod from the same service is already scheduled onto a minion that has value V for label L
func (s *ServiceAffinity) CheckServiceAffinity(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func (s *ServiceAffinity) CheckServiceAffinity(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
var affinitySelector labels.Selector
// check if the pod being scheduled has the affinity labels specified in its NodeSelector
......@@ -292,7 +292,7 @@ func (s *ServiceAffinity) CheckServiceAffinity(pod api.Pod, existingPods []api.P
return false, err
}
// consider only the pods that belong to the same namespace
nsServicePods := []api.Pod{}
nsServicePods := []*api.Pod{}
for _, nsPod := range servicePods {
if nsPod.Namespace == pod.Namespace {
nsServicePods = append(nsServicePods, nsPod)
......@@ -333,7 +333,7 @@ func (s *ServiceAffinity) CheckServiceAffinity(pod api.Pod, existingPods []api.P
return affinitySelector.Matches(labels.Set(minion.Labels)), nil
}
func PodFitsPorts(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func PodFitsPorts(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
existingPorts := getUsedPorts(existingPods...)
wantPorts := getUsedPorts(pod)
for wport := range wantPorts {
......@@ -347,7 +347,7 @@ func PodFitsPorts(pod api.Pod, existingPods []api.Pod, node string) (bool, error
return true, nil
}
func getUsedPorts(pods ...api.Pod) map[int]bool {
func getUsedPorts(pods ...*api.Pod) map[int]bool {
ports := make(map[int]bool)
for _, pod := range pods {
for _, container := range pod.Spec.Containers {
......@@ -361,12 +361,12 @@ func getUsedPorts(pods ...api.Pod) map[int]bool {
// MapPodsToMachines obtains a list of pods and pivots that list into a map where the keys are host names
// and the values are the list of pods running on that host.
func MapPodsToMachines(lister PodLister) (map[string][]api.Pod, error) {
machineToPods := map[string][]api.Pod{}
func MapPodsToMachines(lister PodLister) (map[string][]*api.Pod, error) {
machineToPods := map[string][]*api.Pod{}
// TODO: perform more targeted query...
pods, err := lister.List(labels.Everything())
if err != nil {
return map[string][]api.Pod{}, err
return map[string][]*api.Pod{}, err
}
for _, scheduledPod := range pods {
host := scheduledPod.Spec.Host
......
pkg/scheduler/priorities.go
View file @ 81497f3e
......@@ -39,7 +39,7 @@ func calculateScore(requested, capacity int64, node string) int {
// Calculate the occupancy on a node. 'node' has information about the resources on the node.
// 'pods' is a list of pods currently scheduled on the node.
func calculateOccupancy(pod api.Pod, node api.Node, pods []api.Pod) HostPriority {
func calculateOccupancy(pod *api.Pod, node api.Node, pods []*api.Pod) HostPriority {
totalMilliCPU := int64(0)
totalMemory := int64(0)
for _, existingPod := range pods {
......@@ -78,7 +78,7 @@ func calculateOccupancy(pod api.Pod, node api.Node, pods []api.Pod) HostPriority
// It calculates the percentage of memory and CPU requested by pods scheduled on the node, and prioritizes
// based on the minimum of the average of the fraction of requested to capacity.
// Details: (Sum(requested cpu) / Capacity + Sum(requested memory) / Capacity) * 50
func LeastRequestedPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
func LeastRequestedPriority(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
nodes, err := minionLister.List()
if err != nil {
return HostPriorityList{}, err
......@@ -108,7 +108,7 @@ func NewNodeLabelPriority(label string, presence bool) PriorityFunction {
// CalculateNodeLabelPriority checks whether a particular label exists on a minion or not, regardless of its value.
// If presence is true, prioritizes minions that have the specified label, regardless of value.
// If presence is false, prioritizes minions that do not have the specified label.
func (n *NodeLabelPrioritizer) CalculateNodeLabelPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
func (n *NodeLabelPrioritizer) CalculateNodeLabelPriority(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
var score int
minions, err := minionLister.List()
if err != nil {
......@@ -141,7 +141,7 @@ func (n *NodeLabelPrioritizer) CalculateNodeLabelPriority(pod api.Pod, podLister
// close the two metrics are to each other.
// Detail: score = 10 - abs(cpuFraction-memoryFraction)*10. The algorithm is partly inspired by:
// "Wei Huang et al. An Energy Efficient Virtual Machine Placement Algorithm with Balanced Resource Utilization"
func BalancedResourceAllocation(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
func BalancedResourceAllocation(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
nodes, err := minionLister.List()
if err != nil {
return HostPriorityList{}, err
......@@ -155,7 +155,7 @@ func BalancedResourceAllocation(pod api.Pod, podLister PodLister, minionLister M
return list, nil
}
func calculateBalancedResourceAllocation(pod api.Pod, node api.Node, pods []api.Pod) HostPriority {
func calculateBalancedResourceAllocation(pod *api.Pod, node api.Node, pods []*api.Pod) HostPriority {
totalMilliCPU := int64(0)
totalMemory := int64(0)
score := int(0)
......
pkg/scheduler/scheduler.go
View file @ 81497f3e
......@@ -23,5 +23,5 @@ import (
// Scheduler is an interface implemented by things that know how to schedule pods
// onto machines.
type Scheduler interface {
Schedule(api.Pod, MinionLister) (selectedMachine string, err error)
Schedule(*api.Pod, MinionLister) (selectedMachine string, err error)
}
pkg/scheduler/scheduler_test.go
View file @ 81497f3e
......@@ -31,7 +31,7 @@ type schedulerTester struct {
}
// Call if you know exactly where pod should get scheduled.
func (st *schedulerTester) expectSchedule(pod api.Pod, expected string) {
func (st *schedulerTester) expectSchedule(pod *api.Pod, expected string) {
actual, err := st.scheduler.Schedule(pod, st.minionLister)
if err != nil {
st.t.Errorf("Unexpected error %v\nTried to scheduler: %#v", err, pod)
......@@ -43,7 +43,7 @@ func (st *schedulerTester) expectSchedule(pod api.Pod, expected string) {
}
// Call if you can't predict where pod will be scheduled.
func (st *schedulerTester) expectSuccess(pod api.Pod) {
func (st *schedulerTester) expectSuccess(pod *api.Pod) {
_, err := st.scheduler.Schedule(pod, st.minionLister)
if err != nil {
st.t.Errorf("Unexpected error %v\nTried to scheduler: %#v", err, pod)
......@@ -52,19 +52,19 @@ func (st *schedulerTester) expectSuccess(pod api.Pod) {
}
// Call if pod should *not* schedule.
func (st *schedulerTester) expectFailure(pod api.Pod) {
func (st *schedulerTester) expectFailure(pod *api.Pod) {
_, err := st.scheduler.Schedule(pod, st.minionLister)
if err == nil {
st.t.Error("Unexpected non-error")
}
}
func newPod(host string, hostPorts ...int) api.Pod {
func newPod(host string, hostPorts ...int) *api.Pod {
networkPorts := []api.ContainerPort{}
for _, port := range hostPorts {
networkPorts = append(networkPorts, api.ContainerPort{HostPort: port})
}
return api.Pod{
return &api.Pod{
Spec: api.PodSpec{
Host: host,
Containers: []api.Container{
......
pkg/scheduler/spreading.go
View file @ 81497f3e
......@@ -34,9 +34,9 @@ func NewServiceSpreadPriority(serviceLister ServiceLister) PriorityFunction {
// CalculateSpreadPriority spreads pods by minimizing the number of pods belonging to the same service
// on the same machine.
func (s *ServiceSpread) CalculateSpreadPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
func (s *ServiceSpread) CalculateSpreadPriority(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
var maxCount int
var nsServicePods []api.Pod
var nsServicePods []*api.Pod
services, err := s.serviceLister.GetPodServices(pod)
if err == nil {
......@@ -101,8 +101,8 @@ func NewServiceAntiAffinityPriority(serviceLister ServiceLister, label string) P
// CalculateAntiAffinityPriority spreads pods by minimizing the number of pods belonging to the same service
// on machines with the same value for a particular label.
// The label to be considered is provided to the struct (ServiceAntiAffinity).
func (s *ServiceAntiAffinity) CalculateAntiAffinityPriority(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
var nsServicePods []api.Pod
func (s *ServiceAntiAffinity) CalculateAntiAffinityPriority(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error) {
var nsServicePods []*api.Pod
services, err := s.serviceLister.GetPodServices(pod)
if err == nil {
......
pkg/scheduler/spreading_test.go
View file @ 81497f3e
......@@ -40,36 +40,37 @@ func TestServiceSpreadPriority(t *testing.T) {
Host: "machine2",
}
tests := []struct {
pod api.Pod
pods []api.Pod
pod *api.Pod
pods []*api.Pod
nodes []string
services []api.Service
expectedList HostPriorityList
test string
}{
{
pod: new(api.Pod),
nodes: []string{"machine1", "machine2"},
expectedList: []HostPriority{{"machine1", 10}, {"machine2", 10}},
test: "nothing scheduled",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{{Spec: zone1Spec}},
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{{Spec: zone1Spec}},
nodes: []string{"machine1", "machine2"},
expectedList: []HostPriority{{"machine1", 10}, {"machine2", 10}},
test: "no services",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}}},
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}}},
nodes: []string{"machine1", "machine2"},
services: []api.Service{{Spec: api.ServiceSpec{Selector: map[string]string{"key": "value"}}}},
expectedList: []HostPriority{{"machine1", 10}, {"machine2", 10}},
test: "different services",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
},
......@@ -79,8 +80,8 @@ func TestServiceSpreadPriority(t *testing.T) {
test: "two pods, one service pod",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: "ns1"}},
......@@ -93,8 +94,8 @@ func TestServiceSpreadPriority(t *testing.T) {
test: "five pods, one service pod in no namespace",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: "ns1"}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
......@@ -106,8 +107,8 @@ func TestServiceSpreadPriority(t *testing.T) {
test: "four pods, one service pod in default namespace",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: "ns1"}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: "ns1"}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: "ns2"}},
......@@ -120,8 +121,8 @@ func TestServiceSpreadPriority(t *testing.T) {
test: "five pods, one service pod in specific namespace",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -132,8 +133,8 @@ func TestServiceSpreadPriority(t *testing.T) {
test: "three pods, two service pods on different machines",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -145,8 +146,8 @@ func TestServiceSpreadPriority(t *testing.T) {
test: "four pods, three service pods",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -203,14 +204,15 @@ func TestZoneSpreadPriority(t *testing.T) {
"machine21": zone2, "machine22": zone2,
}
tests := []struct {
pod api.Pod
pods []api.Pod
pod *api.Pod
pods []*api.Pod
nodes map[string]map[string]string
services []api.Service
expectedList HostPriorityList
test string
}{
{
pod: new(api.Pod),
nodes: labeledNodes,
expectedList: []HostPriority{{"machine11", 10}, {"machine12", 10},
{"machine21", 10}, {"machine22", 10},
......@@ -218,8 +220,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "nothing scheduled",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{{Spec: zone1Spec}},
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{{Spec: zone1Spec}},
nodes: labeledNodes,
expectedList: []HostPriority{{"machine11", 10}, {"machine12", 10},
{"machine21", 10}, {"machine22", 10},
......@@ -227,8 +229,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "no services",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}}},
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}}},
nodes: labeledNodes,
services: []api.Service{{Spec: api.ServiceSpec{Selector: map[string]string{"key": "value"}}}},
expectedList: []HostPriority{{"machine11", 10}, {"machine12", 10},
......@@ -237,8 +239,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "different services",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone0Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -251,8 +253,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "three pods, one service pod",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -265,8 +267,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "three pods, two service pods on different machines",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1, Namespace: api.NamespaceDefault}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -280,8 +282,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "three service label match pods in different namespaces",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -295,8 +297,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "four pods, three service pods",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels2}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......@@ -309,8 +311,8 @@ func TestZoneSpreadPriority(t *testing.T) {
test: "service with partial pod label matches",
},
{
pod: api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []api.Pod{
pod: &api.Pod{ObjectMeta: api.ObjectMeta{Labels: labels1}},
pods: []*api.Pod{
{Spec: zone0Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone1Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
{Spec: zone2Spec, ObjectMeta: api.ObjectMeta{Labels: labels1}},
......
pkg/scheduler/types.go
View file @ 81497f3e
......@@ -21,7 +21,7 @@ import (
)
// FitPredicate is a function that indicates if a pod fits into an existing node.
type FitPredicate func(pod api.Pod, existingPods []api.Pod, node string) (bool, error)
type FitPredicate func(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error)
// HostPriority represents the priority of scheduling to a particular host, lower priority is better.
type HostPriority struct {
......@@ -46,7 +46,7 @@ func (h HostPriorityList) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
}
type PriorityFunction func(pod api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error)
type PriorityFunction func(pod *api.Pod, podLister PodLister, minionLister MinionLister) (HostPriorityList, error)
type PriorityConfig struct {
Function PriorityFunction
......
plugin/pkg/scheduler/factory/factory_test.go
View file @ 81497f3e
......@@ -115,19 +115,19 @@ func TestCreateFromEmptyConfig(t *testing.T) {
factory.CreateFromConfig(policy)
}
func PredicateOne(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func PredicateOne(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
return true, nil
}
func PredicateTwo(pod api.Pod, existingPods []api.Pod, node string) (bool, error) {
func PredicateTwo(pod *api.Pod, existingPods []*api.Pod, node string) (bool, error) {
return true, nil
}
func PriorityOne(pod api.Pod, podLister algorithm.PodLister, minionLister algorithm.MinionLister) (algorithm.HostPriorityList, error) {
func PriorityOne(pod *api.Pod, podLister algorithm.PodLister, minionLister algorithm.MinionLister) (algorithm.HostPriorityList, error) {
return []algorithm.HostPriority{}, nil
}
func PriorityTwo(pod api.Pod, podLister algorithm.PodLister, minionLister algorithm.MinionLister) (algorithm.HostPriorityList, error) {
func PriorityTwo(pod *api.Pod, podLister algorithm.PodLister, minionLister algorithm.MinionLister) (algorithm.HostPriorityList, error) {
return []algorithm.HostPriority{}, nil
}
......
plugin/pkg/scheduler/modeler.go
View file @ 81497f3e
......@@ -123,7 +123,7 @@ func (s *SimpleModeler) ForgetPodByKey(key string) {
}
// Extract names for readable logging.
func podNames(pods []api.Pod) []string {
func podNames(pods []*api.Pod) []string {
out := make([]string, len(pods))
for i := range pods {
out[i] = fmt.Sprintf("'%v/%v (%v)'", pods[i].Namespace, pods[i].Name, pods[i].UID)
......@@ -131,7 +131,7 @@ func podNames(pods []api.Pod) []string {
return out
}
func (s *SimpleModeler) listPods(selector labels.Selector) (pods []api.Pod, err error) {
func (s *SimpleModeler) listPods(selector labels.Selector) (pods []*api.Pod, err error) {
assumed, err := s.assumedPods.List(selector)
if err != nil {
return nil, err
......@@ -140,20 +140,20 @@ func (s *SimpleModeler) listPods(selector labels.Selector) (pods []api.Pod, err
// Goal here is to stop making assumptions about a pod once it shows
// up in one of these other lists.
for _, pod := range assumed {
qExist, err := s.queuedPods.Exists(&pod)
qExist, err := s.queuedPods.Exists(pod)
if err != nil {
return nil, err
}
if qExist {
s.assumedPods.Store.Delete(&pod)
s.assumedPods.Store.Delete(pod)
continue
}
sExist, err := s.scheduledPods.Exists(&pod)
sExist, err := s.scheduledPods.Exists(pod)
if err != nil {
return nil, err
}
if sExist {
s.assumedPods.Store.Delete(&pod)
s.assumedPods.Store.Delete(pod)
continue
}
}
......@@ -190,6 +190,6 @@ type simpleModelerPods struct {
}
// List returns pods known and assumed to exist.
func (s simpleModelerPods) List(selector labels.Selector) (pods []api.Pod, err error) {
func (s simpleModelerPods) List(selector labels.Selector) (pods []*api.Pod, err error) {
return s.simpleModeler.listPods(selector)
}
plugin/pkg/scheduler/modeler_test.go
View file @ 81497f3e
......@@ -30,15 +30,15 @@ type nn struct {
type names []nn
func (ids names) list() []api.Pod {
out := make([]api.Pod, len(ids))
for i, id := range ids {
out[i] = api.Pod{
func (ids names) list() []*api.Pod {
out := make([]*api.Pod, 0, len(ids))
for _, id := range ids {
out = append(out, &api.Pod{
ObjectMeta: api.ObjectMeta{
Namespace: id.namespace,
Name: id.name,
},
}
})
}
return out
}
......@@ -54,9 +54,9 @@ func (ids names) has(pod *api.Pod) bool {
func TestModeler(t *testing.T) {
table := []struct {
queuedPods []api.Pod
scheduledPods []api.Pod
assumedPods []api.Pod
queuedPods []*api.Pod
scheduledPods []*api.Pod
assumedPods []*api.Pod
expectPods names
}{
{
......@@ -79,16 +79,16 @@ func TestModeler(t *testing.T) {
for _, item := range table {
q := &cache.StoreToPodLister{cache.NewStore(cache.MetaNamespaceKeyFunc)}
for i := range item.queuedPods {
q.Store.Add(&item.queuedPods[i])
for _, pod := range item.queuedPods {
q.Store.Add(pod)
}
s := &cache.StoreToPodLister{cache.NewStore(cache.MetaNamespaceKeyFunc)}
for i := range item.scheduledPods {
s.Store.Add(&item.scheduledPods[i])
for _, pod := range item.scheduledPods {
s.Store.Add(pod)
}
m := NewSimpleModeler(q, s)
for i := range item.assumedPods {
m.AssumePod(&item.assumedPods[i])
for _, pod := range item.assumedPods {
m.AssumePod(pod)
}
list, err := m.PodLister().List(labels.Everything())
......@@ -98,14 +98,14 @@ func TestModeler(t *testing.T) {
found := 0
for _, pod := range list {
if item.expectPods.has(&pod) {
if item.expectPods.has(pod) {
found++
} else {
t.Errorf("found unexpected pod %#v", pod)
}
}
if e, a := item.expectPods, found; len(e) != a {
t.Errorf("Expected pods:\n%+v\nFound pods:\n%v\n", e, list)
t.Errorf("Expected pods:\n%+v\nFound pods:\n%s\n", podNames(e.list()), podNames(list))
}
}
}
plugin/pkg/scheduler/scheduler.go
View file @ 81497f3e
......@@ -112,7 +112,7 @@ func (s *Scheduler) scheduleOne() {
defer func() {
metrics.E2eSchedulingLatency.Observe(metrics.SinceInMicroseconds(start))
}()
dest, err := s.config.Algorithm.Schedule(*pod, s.config.MinionLister)
dest, err := s.config.Algorithm.Schedule(pod, s.config.MinionLister)
metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInMicroseconds(start))
if err != nil {
glog.V(1).Infof("Failed to schedule: %v", pod)
......
plugin/pkg/scheduler/scheduler_test.go
View file @ 81497f3e
......@@ -59,7 +59,7 @@ type mockScheduler struct {
err error
}
func (es mockScheduler) Schedule(pod api.Pod, ml scheduler.MinionLister) (string, error) {
func (es mockScheduler) Schedule(pod *api.Pod, ml scheduler.MinionLister) (string, error) {
return es.machine, es.err
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment