Skip to content

K
k3s
Commit 892942af authored by Clayton Coleman's avatar Clayton Coleman Committed by Eric Paris
Browse files
Options

Read BoundPods from etcd instead of ContainerManifestList

There are three values that uniquely identify a pod on a host - the configuration source (etcd, file, http), the pod name, and the pod namespace. This change ensures that configuration properly makes those names unique by changing podFullName to contain both name (currently ID in v1beta1, Name in v1beta3) and namespace. The Kubelet does not properly handle information requests for pods not in the default namespace at this time.
parent 332a03b0
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 664 additions and 336 deletions
pkg/api/conversion.go
View file @ 892942af
......@@ -17,9 +17,49 @@ limitations under the License.
package api
import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/conversion"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
)
// Codec is the identity codec for this package - it can only convert itself
// to itself.
var Codec = runtime.CodecFor(Scheme, "")
func init() {
Scheme.AddConversionFuncs(
// Convert ContainerManifest to BoundPod
func(in *ContainerManifest, out *BoundPod, s conversion.Scope) error {
out.Spec.Containers = in.Containers
out.Spec.Volumes = in.Volumes
out.Spec.RestartPolicy = in.RestartPolicy
out.ID = in.ID
out.UID = in.UUID
return nil
},
func(in *BoundPod, out *ContainerManifest, s conversion.Scope) error {
out.Containers = in.Spec.Containers
out.Volumes = in.Spec.Volumes
out.RestartPolicy = in.Spec.RestartPolicy
out.Version = "v1beta2"
out.ID = in.ID
out.UUID = in.UID
return nil
},
func(in *ContainerManifestList, out *BoundPods, s conversion.Scope) error {
if err := s.Convert(&in.Items, &out.Items, 0); err != nil {
return err
}
for i := range out.Items {
item := &out.Items[i]
item.ResourceVersion = in.ResourceVersion
}
return nil
},
func(in *BoundPods, out *ContainerManifestList, s conversion.Scope) error {
if err := s.Convert(&in.Items, &out.Items, 0); err != nil {
return err
}
out.ResourceVersion = in.ResourceVersion
return nil
})
}
pkg/api/register.go
View file @ 892942af
......@@ -40,7 +40,9 @@ func init() {
&Binding{},
&Event{},
&EventList{},
&ContainerManifest{},
&ContainerManifestList{},
&BoundPod{},
&BoundPods{},
)
}
......@@ -61,5 +63,7 @@ func (*ServerOp) IsAnAPIObject() {}
func (*ServerOpList) IsAnAPIObject() {}
func (*Event) IsAnAPIObject() {}
func (*EventList) IsAnAPIObject() {}
func (*ContainerManifest) IsAnAPIObject() {}
func (*ContainerManifestList) IsAnAPIObject() {}
func (*BoundPod) IsAnAPIObject() {}
func (*BoundPods) IsAnAPIObject() {}
pkg/api/serialization_test.go
View file @ 892942af
......@@ -170,6 +170,10 @@ func TestTypes(t *testing.T) {
t.Errorf("Couldn't make a %v? %v", kind, err)
continue
}
if _, err := runtime.FindTypeMeta(item); err != nil {
t.Logf("%s is not a TypeMeta and cannot be round tripped: %v", kind, err)
continue
}
runTest(t, v1beta1.Codec, item)
runTest(t, v1beta2.Codec, item)
runTest(t, api.Codec, item)
......
pkg/api/v1beta1/register.go
View file @ 892942af
......@@ -42,7 +42,9 @@ func init() {
&ServerOpList{},
&Event{},
&EventList{},
&ContainerManifest{},
&ContainerManifestList{},
&BoundPod{},
&BoundPods{},
)
}
......@@ -63,5 +65,7 @@ func (*ServerOp) IsAnAPIObject() {}
func (*ServerOpList) IsAnAPIObject() {}
func (*Event) IsAnAPIObject() {}
func (*EventList) IsAnAPIObject() {}
func (*ContainerManifest) IsAnAPIObject() {}
func (*ContainerManifestList) IsAnAPIObject() {}
func (*BoundPod) IsAnAPIObject() {}
func (*BoundPods) IsAnAPIObject() {}
pkg/api/v1beta2/register.go
View file @ 892942af
......@@ -42,7 +42,9 @@ func init() {
&ServerOpList{},
&Event{},
&EventList{},
&ContainerManifest{},
&ContainerManifestList{},
&BoundPod{},
&BoundPods{},
)
}
......@@ -63,5 +65,7 @@ func (*ServerOp) IsAnAPIObject() {}
func (*ServerOpList) IsAnAPIObject() {}
func (*Event) IsAnAPIObject() {}
func (*EventList) IsAnAPIObject() {}
func (*ContainerManifest) IsAnAPIObject() {}
func (*ContainerManifestList) IsAnAPIObject() {}
func (*BoundPod) IsAnAPIObject() {}
func (*BoundPods) IsAnAPIObject() {}
pkg/api/validation/validation.go
View file @ 892942af
......@@ -437,3 +437,25 @@ func ValidateReadOnlyPersistentDisks(volumes []api.Volume) errs.ErrorList {
}
return allErrs
}
// ValidateBoundPod tests if required fields on a bound pod are set.
func ValidateBoundPod(pod *api.BoundPod) (errors []error) {
if !util.IsDNSSubdomain(pod.ID) {
errors = append(errors, errs.NewFieldInvalid("id", pod.ID))
}
if !util.IsDNSSubdomain(pod.Namespace) {
errors = append(errors, errs.NewFieldInvalid("namespace", pod.Namespace))
}
containerManifest := &api.ContainerManifest{
Version: "v1beta2",
ID: pod.ID,
UUID: pod.UID,
Containers: pod.Spec.Containers,
Volumes: pod.Spec.Volumes,
RestartPolicy: pod.Spec.RestartPolicy,
}
if errs := ValidateManifest(containerManifest); len(errs) != 0 {
errors = append(errors, errs...)
}
return errors
}
pkg/api/validation/validation_test.go
View file @ 892942af
......@@ -823,3 +823,21 @@ func TestValidateReplicationController(t *testing.T) {
}
}
}
func TestValidateBoundPodNoName(t *testing.T) {
errorCases := map[string]api.BoundPod{
// manifest is tested in api/validation_test.go, ensure it is invoked
"empty version": {TypeMeta: api.TypeMeta{ID: "test"}, Spec: api.PodSpec{Containers: []api.Container{{Name: ""}}}},
// Name
"zero-length name": {TypeMeta: api.TypeMeta{ID: ""}},
"name > 255 characters": {TypeMeta: api.TypeMeta{ID: strings.Repeat("a", 256)}},
"name not a DNS subdomain": {TypeMeta: api.TypeMeta{ID: "a.b.c."}},
"name with underscore": {TypeMeta: api.TypeMeta{ID: "a_b_c"}},
}
for k, v := range errorCases {
if errs := ValidateBoundPod(&v); len(errs) == 0 {
t.Errorf("expected failure for %s", k)
}
}
}
pkg/kubelet/config/config.go
View file @ 892942af
......@@ -21,7 +21,9 @@ import (
"reflect"
"sync"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
apierrs "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/config"
......@@ -89,7 +91,7 @@ func (c *PodConfig) Sync() {
type podStorage struct {
podLock sync.RWMutex
// map of source name to pod name to pod reference
pods map[string]map[string]*kubelet.Pod
pods map[string]map[string]*api.BoundPod
mode PodConfigNotificationMode
// ensures that updates are delivered in strict order
......@@ -103,7 +105,7 @@ type podStorage struct {
// TODO: allow initialization of the current state of the store with snapshotted version.
func newPodStorage(updates chan<- kubelet.PodUpdate, mode PodConfigNotificationMode) *podStorage {
return &podStorage{
pods: make(map[string]map[string]*kubelet.Pod),
pods: make(map[string]map[string]*api.BoundPod),
mode: mode,
updates: updates,
}
......@@ -136,12 +138,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().([]kubelet.Pod), kubelet.SET}
s.updates <- kubelet.PodUpdate{s.MergedState().([]api.BoundPod), kubelet.SET}
}
case PodConfigNotificationSnapshot:
if len(updates.Pods) > 0 || len(deletes.Pods) > 0 || len(adds.Pods) > 0 {
s.updates <- kubelet.PodUpdate{s.MergedState().([]kubelet.Pod), kubelet.SET}
s.updates <- kubelet.PodUpdate{s.MergedState().([]api.BoundPod), kubelet.SET}
}
default:
......@@ -161,7 +163,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
pods := s.pods[source]
if pods == nil {
pods = make(map[string]*kubelet.Pod)
pods = make(map[string]*api.BoundPod)
}
update := change.(kubelet.PodUpdate)
......@@ -175,11 +177,11 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
filtered := filterInvalidPods(update.Pods, source)
for _, ref := range filtered {
name := ref.Name
name := podUniqueName(ref)
if existing, found := pods[name]; found {
if !reflect.DeepEqual(existing.Manifest, ref.Manifest) {
if !reflect.DeepEqual(existing.Spec, ref.Spec) {
// this is an update
existing.Manifest = ref.Manifest
existing.Spec = ref.Spec
updates.Pods = append(updates.Pods, *existing)
continue
}
......@@ -187,7 +189,10 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
continue
}
// this is an add
ref.Namespace = source
if ref.Annotations == nil {
ref.Annotations = make(map[string]string)
}
ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source
pods[name] = ref
adds.Pods = append(adds.Pods, *ref)
}
......@@ -195,7 +200,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
case kubelet.REMOVE:
glog.V(4).Infof("Removing a pod %v", update)
for _, value := range update.Pods {
name := value.Name
name := podUniqueName(&value)
if existing, found := pods[name]; found {
// this is a delete
delete(pods, name)
......@@ -209,23 +214,26 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
glog.V(4).Infof("Setting pods for source %s : %v", source, update)
// Clear the old map entries by just creating a new map
oldPods := pods
pods = make(map[string]*kubelet.Pod)
pods = make(map[string]*api.BoundPod)
filtered := filterInvalidPods(update.Pods, source)
for _, ref := range filtered {
name := ref.Name
name := podUniqueName(ref)
if existing, found := oldPods[name]; found {
pods[name] = existing
if !reflect.DeepEqual(existing.Manifest, ref.Manifest) {
if !reflect.DeepEqual(existing.Spec, ref.Spec) {
// this is an update
existing.Manifest = ref.Manifest
existing.Spec = ref.Spec
updates.Pods = append(updates.Pods, *existing)
continue
}
// this is a no-op
continue
}
ref.Namespace = source
if ref.Annotations == nil {
ref.Annotations = make(map[string]string)
}
ref.Annotations[kubelet.ConfigSourceAnnotationKey] = source
pods[name] = ref
adds.Pods = append(adds.Pods, *ref)
}
......@@ -246,20 +254,21 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
return adds, updates, deletes
}
func filterInvalidPods(pods []kubelet.Pod, source string) (filtered []*kubelet.Pod) {
func filterInvalidPods(pods []api.BoundPod, source string) (filtered []*api.BoundPod) {
names := util.StringSet{}
for i := range pods {
var errors []error
if names.Has(pods[i].Name) {
errors = append(errors, apierrs.NewFieldDuplicate("name", pods[i].Name))
name := podUniqueName(&pods[i])
if names.Has(name) {
errors = append(errors, apierrs.NewFieldDuplicate("name", pods[i].ID))
} else {
names.Insert(pods[i].Name)
names.Insert(name)
}
if errs := kubelet.ValidatePod(&pods[i]); len(errs) != 0 {
if errs := validation.ValidateBoundPod(&pods[i]); len(errs) != 0 {
errors = append(errors, errs...)
}
if len(errors) > 0 {
glog.Warningf("Pod %d (%s) from %s failed validation, ignoring: %v", i+1, pods[i].Name, source, errors)
glog.Warningf("Pod %d (%s) from %s failed validation, ignoring: %v", i+1, pods[i].ID, source, errors)
continue
}
filtered = append(filtered, &pods[i])
......@@ -271,20 +280,32 @@ func filterInvalidPods(pods []kubelet.Pod, source string) (filtered []*kubelet.P
func (s *podStorage) Sync() {
s.updateLock.Lock()
defer s.updateLock.Unlock()
s.updates <- kubelet.PodUpdate{s.MergedState().([]kubelet.Pod), kubelet.SET}
s.updates <- kubelet.PodUpdate{s.MergedState().([]api.BoundPod), kubelet.SET}
}
// Object implements config.Accessor
func (s *podStorage) MergedState() interface{} {
s.podLock.RLock()
defer s.podLock.RUnlock()
pods := make([]kubelet.Pod, 0)
for source, sourcePods := range s.pods {
var pods []api.BoundPod
for _, sourcePods := range s.pods {
for _, podRef := range sourcePods {
pod := *podRef
pod.Namespace = source
pods = append(pods, pod)
pod, err := api.Scheme.Copy(podRef)
if err != nil {
glog.Errorf("unable to copy pod: %v", err)
}
pods = append(pods, *pod.(*api.BoundPod))
}
}
return pods
}
// podUniqueName returns a value for a given pod that is unique across a source,
// which is the combination of namespace and ID.
func podUniqueName(pod *api.BoundPod) string {
namespace := pod.Namespace
if len(namespace) == 0 {
namespace = api.NamespaceDefault
}
return fmt.Sprintf("%s.%s", pod.ID, namespace)
}
pkg/kubelet/config/config_test.go
View file @ 892942af
......@@ -18,6 +18,7 @@ package config
import (
"reflect"
"sort"
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
......@@ -32,7 +33,7 @@ func expectEmptyChannel(t *testing.T, ch <-chan interface{}) {
}
}
type sortedPods []kubelet.Pod
type sortedPods []api.BoundPod
func (s sortedPods) Len() int {
return len(s)
......@@ -41,25 +42,27 @@ func (s sortedPods) Swap(i, j int) {
s[i], s[j] = s[j], s[i]
}
func (s sortedPods) Less(i, j int) bool {
if s[i].Namespace < s[j].Namespace {
return true
}
return s[i].Name < s[j].Name
return s[i].ID < s[j].ID
}
func CreateValidPod(name, namespace string) kubelet.Pod {
return kubelet.Pod{
Name: name,
Namespace: namespace,
Manifest: api.ContainerManifest{
Version: "v1beta1",
func CreateValidPod(name, namespace, source string) api.BoundPod {
return api.BoundPod{
TypeMeta: api.TypeMeta{
ID: name,
Namespace: namespace,
Annotations: map[string]string{kubelet.ConfigSourceAnnotationKey: source},
},
Spec: api.PodSpec{
RestartPolicy: api.RestartPolicy{Always: &api.RestartPolicyAlways{}},
},
}
}
func CreatePodUpdate(op kubelet.PodOperation, pods ...kubelet.Pod) kubelet.PodUpdate {
newPods := make([]kubelet.Pod, len(pods))
func CreatePodUpdate(op kubelet.PodOperation, pods ...api.BoundPod) kubelet.PodUpdate {
if len(pods) == 0 {
return kubelet.PodUpdate{Op: op}
}
newPods := make([]api.BoundPod, len(pods))
for i := range pods {
newPods[i] = pods[i]
}
......@@ -76,6 +79,7 @@ func createPodConfigTester(mode PodConfigNotificationMode) (chan<- interface{},
func expectPodUpdate(t *testing.T, ch <-chan kubelet.PodUpdate, expected ...kubelet.PodUpdate) {
for i := range expected {
update := <-ch
sort.Sort(sortedPods(update.Pods))
if !reflect.DeepEqual(expected[i], update) {
t.Fatalf("Expected %#v, Got %#v", expected[i], update)
}
......@@ -95,24 +99,63 @@ func TestNewPodAdded(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", ""))
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", "test")))
config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "new", "test")))
}
func TestNewPodAddedInvalidNamespace(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "", ""))
channel <- podUpdate
config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET))
}
func TestNewPodAddedDefaultNamespace(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "default", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "default", "test")))
config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "default", "test")))
}
func TestNewPodAddedDifferentNamespaces(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationIncremental)
// see an update
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "default", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "default", "test")))
// see an update in another namespace
podUpdate = CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", "test")))
config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "default", "test"), CreateValidPod("foo", "new", "test")))
}
func TestInvalidPodFiltered(t *testing.T) {
channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental)
// see an update
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", ""))
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", "test")))
// add an invalid update
podUpdate = CreatePodUpdate(kubelet.UPDATE, kubelet.Pod{Name: "foo"})
podUpdate = CreatePodUpdate(kubelet.UPDATE, api.BoundPod{TypeMeta: api.TypeMeta{ID: "foo"}})
channel <- podUpdate
expectNoPodUpdate(t, ch)
}
......@@ -121,16 +164,16 @@ func TestNewPodAddedSnapshotAndUpdates(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshotAndUpdates)
// see an set
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", ""))
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "new", "test")))
config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "new", "test")))
// container updates are separated as UPDATE
pod := podUpdate.Pods[0]
pod.Manifest.Containers = []api.Container{{Name: "bar", Image: "test"}}
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test"}}
channel <- CreatePodUpdate(kubelet.ADD, pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, pod))
}
......@@ -139,16 +182,16 @@ func TestNewPodAddedSnapshot(t *testing.T) {
channel, ch, config := createPodConfigTester(PodConfigNotificationSnapshot)
// see an set
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", ""))
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "new", "test")))
config.Sync()
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, CreateValidPod("foo", "new", "test")))
// container updates are separated as UPDATE
pod := podUpdate.Pods[0]
pod.Manifest.Containers = []api.Container{{Name: "bar", Image: "test"}}
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test"}}
channel <- CreatePodUpdate(kubelet.ADD, pod)
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.SET, pod))
}
......@@ -157,21 +200,21 @@ func TestNewPodAddedUpdatedRemoved(t *testing.T) {
channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental)
// should register an add
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", ""))
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", "test")))
// should ignore ADDs that are identical
expectNoPodUpdate(t, ch)
// an kubelet.ADD should be converted to kubelet.UPDATE
pod := CreateValidPod("foo", "test")
pod.Manifest.Containers = []api.Container{{Name: "bar", Image: "test"}}
pod := CreateValidPod("foo", "new", "test")
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test"}}
podUpdate = CreatePodUpdate(kubelet.ADD, pod)
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.UPDATE, pod))
podUpdate = CreatePodUpdate(kubelet.REMOVE, kubelet.Pod{Name: "foo"})
podUpdate = CreatePodUpdate(kubelet.REMOVE, api.BoundPod{TypeMeta: api.TypeMeta{ID: "foo", Namespace: "new"}})
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.REMOVE, pod))
}
......@@ -180,20 +223,20 @@ func TestNewPodAddedUpdatedSet(t *testing.T) {
channel, ch, _ := createPodConfigTester(PodConfigNotificationIncremental)
// should register an add
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", ""), CreateValidPod("foo2", ""), CreateValidPod("foo3", ""))
podUpdate := CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", ""), CreateValidPod("foo2", "new", ""), CreateValidPod("foo3", "new", ""))
channel <- podUpdate
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "test"), CreateValidPod("foo2", "test"), CreateValidPod("foo3", "test")))
expectPodUpdate(t, ch, CreatePodUpdate(kubelet.ADD, CreateValidPod("foo", "new", "test"), CreateValidPod("foo2", "new", "test"), CreateValidPod("foo3", "new", "test")))
// should ignore ADDs that are identical
expectNoPodUpdate(t, ch)
// should be converted to an kubelet.ADD, kubelet.REMOVE, and kubelet.UPDATE
pod := CreateValidPod("foo2", "test")
pod.Manifest.Containers = []api.Container{{Name: "bar", Image: "test"}}
podUpdate = CreatePodUpdate(kubelet.SET, pod, CreateValidPod("foo3", ""), CreateValidPod("foo4", "test"))
pod := CreateValidPod("foo2", "new", "test")
pod.Spec.Containers = []api.Container{{Name: "bar", Image: "test"}}
podUpdate = CreatePodUpdate(kubelet.SET, pod, CreateValidPod("foo3", "new", ""), CreateValidPod("foo4", "new", "test"))
channel <- podUpdate
expectPodUpdate(t, ch,
CreatePodUpdate(kubelet.REMOVE, CreateValidPod("foo", "test")),
CreatePodUpdate(kubelet.ADD, CreateValidPod("foo4", "test")),
CreatePodUpdate(kubelet.REMOVE, CreateValidPod("foo", "new", "test")),
CreatePodUpdate(kubelet.ADD, CreateValidPod("foo4", "new", "test")),
CreatePodUpdate(kubelet.UPDATE, pod))
}
pkg/kubelet/config/etcd.go
View file @ 892942af
......@@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"path"
"strconv"
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
......@@ -86,21 +87,26 @@ func (s *SourceEtcd) run() {
// eventToPods takes a watch.Event object, and turns it into a structured list of pods.
// It returns a list of containers, or an error if one occurs.
func eventToPods(ev watch.Event) ([]kubelet.Pod, error) {
pods := []kubelet.Pod{}
manifests, ok := ev.Object.(*api.ContainerManifestList)
func eventToPods(ev watch.Event) ([]api.BoundPod, error) {
pods := []api.BoundPod{}
boundPods, ok := ev.Object.(*api.BoundPods)
if !ok {
return pods, errors.New("unable to parse response as ContainerManifestList")
return pods, errors.New("unable to parse response as BoundPods")
}
for i, manifest := range manifests.Items {
name := manifest.ID
if name == "" {
name = fmt.Sprintf("%d", i+1)
for i, pod := range boundPods.Items {
if len(pod.ID) == 0 {
pod.ID = fmt.Sprintf("%d", i+1)
}
pods = append(pods, kubelet.Pod{
Name: name,
Manifest: manifest})
// TODO: generate random UID if not present
if pod.UID == "" && !pod.CreationTimestamp.IsZero() {
pod.UID = strconv.FormatInt(pod.CreationTimestamp.Unix(), 10)
}
// Backwards compatibility with old api servers
if len(pod.Namespace) == 0 {
pod.Namespace = api.NamespaceDefault
}
pods = append(pods, pod)
}
return pods, nil
......
pkg/kubelet/config/etcd_test.go
View file @ 892942af
......@@ -21,53 +21,52 @@ import (
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
func TestEventToPods(t *testing.T) {
tests := []struct {
input watch.Event
pods []kubelet.Pod
pods []api.BoundPod
fail bool
}{
{
input: watch.Event{Object: nil},
pods: []kubelet.Pod{},
pods: []api.BoundPod{},
fail: true,
},
{
input: watch.Event{Object: &api.ContainerManifestList{}},
pods: []kubelet.Pod{},
input: watch.Event{Object: &api.BoundPods{}},
pods: []api.BoundPod{},
fail: false,
},
{
input: watch.Event{
Object: &api.ContainerManifestList{
Items: []api.ContainerManifest{
{ID: "foo"},
{ID: "bar"},
Object: &api.BoundPods{
Items: []api.BoundPod{
{TypeMeta: api.TypeMeta{ID: "foo"}},
{TypeMeta: api.TypeMeta{ID: "bar"}},
},
},
},
pods: []kubelet.Pod{
{Name: "foo", Manifest: api.ContainerManifest{ID: "foo"}},
{Name: "bar", Manifest: api.ContainerManifest{ID: "bar"}},
pods: []api.BoundPod{
{TypeMeta: api.TypeMeta{ID: "foo", Namespace: "default"}, Spec: api.PodSpec{}},
{TypeMeta: api.TypeMeta{ID: "bar", Namespace: "default"}, Spec: api.PodSpec{}},
},
fail: false,
},
{
input: watch.Event{
Object: &api.ContainerManifestList{
Items: []api.ContainerManifest{
{ID: ""},
{ID: ""},
Object: &api.BoundPods{
Items: []api.BoundPod{
{TypeMeta: api.TypeMeta{ID: "1"}},
{TypeMeta: api.TypeMeta{ID: "2", Namespace: "foo"}},
},
},
},
pods: []kubelet.Pod{
{Name: "1", Manifest: api.ContainerManifest{ID: ""}},
{Name: "2", Manifest: api.ContainerManifest{ID: ""}},
pods: []api.BoundPod{
{TypeMeta: api.TypeMeta{ID: "1", Namespace: "default"}, Spec: api.PodSpec{}},
{TypeMeta: api.TypeMeta{ID: "2", Namespace: "foo"}, Spec: api.PodSpec{}},
},
fail: false,
},
......
pkg/kubelet/config/file.go
View file @ 892942af
......@@ -29,8 +29,10 @@ import (
"strings"
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/golang/glog"
"gopkg.in/v1/yaml"
)
......@@ -79,7 +81,7 @@ func (s *SourceFile) extractFromPath() error {
if err != nil {
return err
}
s.updates <- kubelet.PodUpdate{[]kubelet.Pod{pod}, kubelet.SET}
s.updates <- kubelet.PodUpdate{[]api.BoundPod{pod}, kubelet.SET}
default:
return fmt.Errorf("path is not a directory or file")
......@@ -88,28 +90,29 @@ func (s *SourceFile) extractFromPath() error {
return nil
}
func extractFromDir(name string) ([]kubelet.Pod, error) {
pods := []kubelet.Pod{}
func extractFromDir(name string) ([]api.BoundPod, error) {
files, err := filepath.Glob(filepath.Join(name, "[^.]*"))
if err != nil {
return pods, err
return nil, err
}
if len(files) == 0 {
return nil, nil
}
sort.Strings(files)
pods := []api.BoundPod{}
for _, file := range files {
pod, err := extractFromFile(file)
if err != nil {
return []kubelet.Pod{}, err
return nil, err
}
pods = append(pods, pod)
}
return pods, nil
}
func extractFromFile(name string) (kubelet.Pod, error) {
var pod kubelet.Pod
func extractFromFile(name string) (api.BoundPod, error) {
var pod api.BoundPod
file, err := os.Open(name)
if err != nil {
......@@ -123,15 +126,23 @@ func extractFromFile(name string) (kubelet.Pod, error) {
return pod, err
}
if err := yaml.Unmarshal(data, &pod.Manifest); err != nil {
return pod, fmt.Errorf("could not unmarshal manifest: %v", err)
manifest := &api.ContainerManifest{}
// TODO: use api.Scheme.DecodeInto
if err := yaml.Unmarshal(data, manifest); err != nil {
return pod, err
}
podName := pod.Manifest.ID
if podName == "" {
podName = simpleSubdomainSafeHash(name)
if err := api.Scheme.Convert(manifest, &pod); err != nil {
return pod, err
}
pod.ID = simpleSubdomainSafeHash(name)
if len(pod.UID) == 0 {
pod.UID = simpleSubdomainSafeHash(name)
}
if len(pod.Namespace) == 0 {
pod.Namespace = api.NamespaceDefault
}
pod.Name = podName
return pod, nil
}
......
pkg/kubelet/config/file_test.go
View file @ 892942af
......@@ -26,10 +26,57 @@ import (
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"gopkg.in/v1/yaml"
)
func ExampleManifestAndPod(id string) (api.ContainerManifest, api.BoundPod) {
manifest := api.ContainerManifest{
ID: id,
UUID: "uid",
Containers: []api.Container{
{
Name: "c" + id,
Image: "foo",
},
},
Volumes: []api.Volume{
{
Name: "host-dir",
Source: &api.VolumeSource{
HostDir: &api.HostDir{"/dir/path"},
},
},
},
}
expectedPod := api.BoundPod{
TypeMeta: api.TypeMeta{
ID: id,
UID: "uid",
Namespace: "default",
},
Spec: api.PodSpec{
Containers: []api.Container{
{
Name: "c" + id,
Image: "foo",
},
},
Volumes: []api.Volume{
{
Name: "host-dir",
Source: &api.VolumeSource{
HostDir: &api.HostDir{"/dir/path"},
},
},
},
},
}
return manifest, expectedPod
}
func TestExtractFromNonExistentFile(t *testing.T) {
ch := make(chan interface{}, 1)
c := SourceFile{"/some/fake/file", ch}
......@@ -70,16 +117,18 @@ func TestReadFromFile(t *testing.T) {
select {
case got := <-ch:
update := got.(kubelet.PodUpdate)
expected := CreatePodUpdate(kubelet.SET, kubelet.Pod{
Name: "test",
Manifest: api.ContainerManifest{
ID: "test",
Version: "v1beta1",
expected := CreatePodUpdate(kubelet.SET, api.BoundPod{
TypeMeta: api.TypeMeta{
ID: simpleSubdomainSafeHash(file.Name()),
UID: simpleSubdomainSafeHash(file.Name()),
Namespace: "default",
},
Spec: api.PodSpec{
Containers: []api.Container{{Image: "test/image"}},
},
})
if !reflect.DeepEqual(expected, update) {
t.Errorf("Expected %#v, Got %#v", expected, update)
t.Fatalf("Expected %#v, Got %#v", expected, update)
}
case <-time.After(2 * time.Millisecond):
......@@ -95,29 +144,31 @@ func TestExtractFromBadDataFile(t *testing.T) {
c := SourceFile{file.Name(), ch}
err := c.extractFromPath()
if err == nil {
t.Errorf("Expected error")
t.Fatalf("Expected error")
}
expectEmptyChannel(t, ch)
}
func TestExtractFromValidDataFile(t *testing.T) {
manifest := api.ContainerManifest{ID: ""}
manifest, expectedPod := ExampleManifestAndPod("id")
text, err := json.Marshal(manifest)
if err != nil {
t.Errorf("Unexpected error: %v", err)
t.Fatalf("Unexpected error: %v", err)
}
file := writeTestFile(t, os.TempDir(), "test_pod_config", string(text))
defer os.Remove(file.Name())
expectedPod.ID = simpleSubdomainSafeHash(file.Name())
ch := make(chan interface{}, 1)
c := SourceFile{file.Name(), ch}
err = c.extractFromPath()
if err != nil {
t.Errorf("Unexpected error: %v", err)
t.Fatalf("Unexpected error: %v", err)
}
update := (<-ch).(kubelet.PodUpdate)
expected := CreatePodUpdate(kubelet.SET, kubelet.Pod{Name: simpleSubdomainSafeHash(file.Name()), Manifest: manifest})
expected := CreatePodUpdate(kubelet.SET, expectedPod)
if !reflect.DeepEqual(expected, update) {
t.Errorf("Expected %#v, Got %#v", expected, update)
}
......@@ -126,7 +177,7 @@ func TestExtractFromValidDataFile(t *testing.T) {
func TestExtractFromEmptyDir(t *testing.T) {
dirName, err := ioutil.TempDir("", "foo")
if err != nil {
t.Errorf("Unexpected error: %v", err)
t.Fatalf("Unexpected error: %v", err)
}
defer os.RemoveAll(dirName)
......@@ -134,7 +185,7 @@ func TestExtractFromEmptyDir(t *testing.T) {
c := SourceFile{dirName, ch}
err = c.extractFromPath()
if err != nil {
t.Errorf("Unexpected error: %v", err)
t.Fatalf("Unexpected error: %v", err)
}
update := (<-ch).(kubelet.PodUpdate)
......@@ -145,54 +196,55 @@ func TestExtractFromEmptyDir(t *testing.T) {
}
func TestExtractFromDir(t *testing.T) {
manifests := []api.ContainerManifest{
{Version: "v1beta1", Containers: []api.Container{{Name: "1", Image: "foo"}}},
{Version: "v1beta1", Containers: []api.Container{{Name: "2", Image: "bar"}}},
}
manifest, expectedPod := ExampleManifestAndPod("1")
manifest2, expectedPod2 := ExampleManifestAndPod("2")
manifests := []api.ContainerManifest{manifest, manifest2}
pods := []api.BoundPod{expectedPod, expectedPod2}
files := make([]*os.File, len(manifests))
dirName, err := ioutil.TempDir("", "foo")
if err != nil {
t.Errorf("Unexpected error: %v", err)
t.Fatalf("Unexpected error: %v", err)
}
for i, manifest := range manifests {
data, err := json.Marshal(manifest)
if err != nil {
t.Errorf("Unexpected error: %v", err)
continue
}
file, err := ioutil.TempFile(dirName, manifest.ID)
if err != nil {
t.Errorf("Unexpected error: %v", err)
continue
}
name := file.Name()
if err := file.Close(); err != nil {
t.Errorf("Unexpected error: %v", err)
continue
}
ioutil.WriteFile(name, data, 0755)
files[i] = file
pods[i].ID = simpleSubdomainSafeHash(name)
}
ch := make(chan interface{}, 1)
c := SourceFile{dirName, ch}
err = c.extractFromPath()
if err != nil {
t.Errorf("Unexpected error: %v", err)
t.Fatalf("Unexpected error: %v", err)
}
update := (<-ch).(kubelet.PodUpdate)
expected := CreatePodUpdate(
kubelet.SET,
kubelet.Pod{Name: simpleSubdomainSafeHash(files[0].Name()), Manifest: manifests[0]},
kubelet.Pod{Name: simpleSubdomainSafeHash(files[1].Name()), Manifest: manifests[1]},
)
expected := CreatePodUpdate(kubelet.SET, pods...)
sort.Sort(sortedPods(update.Pods))
sort.Sort(sortedPods(expected.Pods))
if !reflect.DeepEqual(expected, update) {
t.Errorf("Expected %#v, Got %#v", expected, update)
t.Fatalf("Expected %#v, Got %#v", expected, update)
}
for i := range update.Pods {
if errs := kubelet.ValidatePod(&update.Pods[i]); len(errs) != 0 {
if errs := validation.ValidateBoundPod(&update.Pods[i]); len(errs) != 0 {
t.Errorf("Expected no validation errors on %#v, Got %#v", update.Pods[i], errs)
}
}
......
pkg/kubelet/config/http.go
View file @ 892942af
......@@ -28,6 +28,7 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/golang/glog"
"gopkg.in/v1/yaml"
)
......@@ -79,6 +80,7 @@ func (s *SourceURL) extractFromURL() error {
// First try as if it's a single manifest
var manifest api.ContainerManifest
// TODO: should be api.Scheme.Decode
singleErr := yaml.Unmarshal(data, &manifest)
if singleErr == nil {
if errs := validation.ValidateManifest(&manifest); len(errs) > 0 {
......@@ -86,16 +88,23 @@ func (s *SourceURL) extractFromURL() error {
}
}
if singleErr == nil {
pod := kubelet.Pod{Name: manifest.ID, Manifest: manifest}
if pod.Name == "" {
pod.Name = "1"
pod := api.BoundPod{}
if err := api.Scheme.Convert(&manifest, &pod); err != nil {
return err
}
if len(pod.ID) == 0 {
pod.ID = "1"
}
if len(pod.Namespace) == 0 {
pod.Namespace = api.NamespaceDefault
}
s.updates <- kubelet.PodUpdate{[]kubelet.Pod{pod}, kubelet.SET}
s.updates <- kubelet.PodUpdate{[]api.BoundPod{pod}, kubelet.SET}
return nil
}
// That didn't work, so try an array of manifests.
var manifests []api.ContainerManifest
// TODO: should be api.Scheme.Decode
multiErr := yaml.Unmarshal(data, &manifests)
// We're not sure if the person reading the logs is going to care about the single or
// multiple manifest unmarshalling attempt, so we need to put both in the logs, as is
......@@ -113,18 +122,24 @@ func (s *SourceURL) extractFromURL() error {
// array of manifests (and no error) when unmarshaled as such. In that case,
// if the single manifest at least had a Version, we return the single-manifest
// error (if any).
if len(manifests) == 0 && manifest.Version != "" {
if len(manifests) == 0 && len(manifest.Version) != 0 {
return singleErr
}
pods := []kubelet.Pod{}
for i, manifest := range manifests {
pod := kubelet.Pod{Name: manifest.ID, Manifest: manifest}
if pod.Name == "" {
pod.Name = fmt.Sprintf("%d", i+1)
list := api.ContainerManifestList{Items: manifests}
boundPods := &api.BoundPods{}
if err := api.Scheme.Convert(&list, boundPods); err != nil {
return err
}
for i := range boundPods.Items {
pod := &boundPods.Items[i]
if len(pod.ID) == 0 {
pod.ID = fmt.Sprintf("%d", i+1)
}
if len(pod.Namespace) == 0 {
pod.Namespace = api.NamespaceDefault
}
pods = append(pods, pod)
}
s.updates <- kubelet.PodUpdate{pods, kubelet.SET}
s.updates <- kubelet.PodUpdate{boundPods.Items, kubelet.SET}
return nil
}
......
pkg/kubelet/config/http_test.go
View file @ 892942af
......@@ -24,6 +24,7 @@ import (
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
)
......@@ -122,11 +123,12 @@ func TestExtractFromHTTP(t *testing.T) {
desc: "Single manifest",
manifests: api.ContainerManifest{Version: "v1beta1", ID: "foo"},
expected: CreatePodUpdate(kubelet.SET,
kubelet.Pod{
Name: "foo",
Manifest: api.ContainerManifest{
Version: "v1beta1",
ID: "foo",
api.BoundPod{
TypeMeta: api.TypeMeta{
ID: "foo",
Namespace: "default",
},
Spec: api.PodSpec{
RestartPolicy: api.RestartPolicy{Always: &api.RestartPolicyAlways{}},
},
}),
......@@ -138,13 +140,19 @@ func TestExtractFromHTTP(t *testing.T) {
{Version: "v1beta1", ID: "bar", Containers: []api.Container{{Name: "1", Image: "foo"}}},
},
expected: CreatePodUpdate(kubelet.SET,
kubelet.Pod{
Name: "1",
Manifest: api.ContainerManifest{Version: "v1beta1", ID: "", Containers: []api.Container{{Name: "1", Image: "foo"}}},
api.BoundPod{
TypeMeta: api.TypeMeta{
ID: "1",
Namespace: "default",
},
Spec: api.PodSpec{Containers: []api.Container{{Name: "1", Image: "foo"}}},
},
kubelet.Pod{
Name: "bar",
Manifest: api.ContainerManifest{Version: "v1beta1", ID: "bar", Containers: []api.Container{{Name: "1", Image: "foo"}}},
api.BoundPod{
TypeMeta: api.TypeMeta{
ID: "bar",
Namespace: "default",
},
Spec: api.PodSpec{Containers: []api.Container{{Name: "1", Image: "foo"}}},
}),
},
{
......@@ -167,13 +175,14 @@ func TestExtractFromHTTP(t *testing.T) {
c := SourceURL{testServer.URL, ch, nil}
if err := c.extractFromURL(); err != nil {
t.Errorf("%s: Unexpected error: %v", testCase.desc, err)
continue
}
update := (<-ch).(kubelet.PodUpdate)
if !reflect.DeepEqual(testCase.expected, update) {
t.Errorf("%s: Expected: %#v, Got: %#v", testCase.desc, testCase.expected, update)
}
for i := range update.Pods {
if errs := kubelet.ValidatePod(&update.Pods[i]); len(errs) != 0 {
if errs := validation.ValidateBoundPod(&update.Pods[i]); len(errs) != 0 {
t.Errorf("%s: Expected no validation errors on %#v, Got %#v", testCase.desc, update.Pods[i], errs)
}
}
......
pkg/kubelet/dockertools/docker.go
View file @ 892942af
......@@ -212,7 +212,7 @@ func GetKubeletDockerContainers(client DockerInterface, allContainers bool) (Doc
// TODO(dchen1107): Remove the old separator "--" by end of Oct
if !strings.HasPrefix(container.Names[0], "/"+containerNamePrefix+"_") &&
!strings.HasPrefix(container.Names[0], "/"+containerNamePrefix+"--") {
glog.Infof("Docker Container:%s is not managed by kubelet.", container.Names[0])
glog.Infof("Docker Container: %s is not managed by kubelet.", container.Names[0])
continue
}
result[DockerID(container.ID)] = container
......@@ -330,7 +330,7 @@ func inspectContainer(client DockerInterface, dockerID, containerName string) (*
}
// GetDockerPodInfo returns docker info for all containers in the pod/manifest.
func GetDockerPodInfo(client DockerInterface, manifest api.ContainerManifest, podFullName, uuid string) (api.PodInfo, error) {
func GetDockerPodInfo(client DockerInterface, manifest api.PodSpec, podFullName, uuid string) (api.PodInfo, error) {
info := api.PodInfo{}
containers, err := client.ListContainers(docker.ListContainersOptions{All: true})
......
pkg/kubelet/runonce.go
View file @ 892942af
......@@ -20,6 +20,7 @@ import (
"fmt"
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
"github.com/golang/glog"
)
......@@ -32,7 +33,7 @@ const (
)
type RunPodResult struct {
Pod *Pod
Pod *api.BoundPod
Err error
}
......@@ -50,7 +51,7 @@ 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 []Pod) (results []RunPodResult, err error) {
func (kl *Kubelet) runOnce(pods []api.BoundPod) (results []RunPodResult, err error) {
if kl.dockerPuller == nil {
kl.dockerPuller = dockertools.NewDockerPuller(kl.dockerClient, kl.pullQPS, kl.pullBurst)
}
......@@ -72,10 +73,10 @@ func (kl *Kubelet) runOnce(pods []Pod) (results []RunPodResult, err error) {
results = append(results, res)
if res.Err != nil {
// TODO(proppy): report which containers failed the pod.
glog.Infof("failed to start pod %q: %v", res.Pod.Name, res.Err)
failedPods = append(failedPods, res.Pod.Name)
glog.Infof("failed to start pod %q: %v", res.Pod.ID, res.Err)
failedPods = append(failedPods, res.Pod.ID)
} else {
glog.Infof("started pod %q", res.Pod.Name)
glog.Infof("started pod %q", res.Pod.ID)
}
}
if len(failedPods) > 0 {
......@@ -86,7 +87,7 @@ func (kl *Kubelet) runOnce(pods []Pod) (results []RunPodResult, err error) {
}
// runPod runs a single pod and wait until all containers are running.
func (kl *Kubelet) runPod(pod Pod) error {
func (kl *Kubelet) runPod(pod api.BoundPod) error {
delay := RunOnceRetryDelay
retry := 0
for {
......@@ -95,18 +96,18 @@ func (kl *Kubelet) runPod(pod Pod) error {
return fmt.Errorf("failed to get kubelet docker containers: %v", err)
}
if running := kl.isPodRunning(pod, dockerContainers); running {
glog.Infof("pod %q containers running", pod.Name)
glog.Infof("pod %q containers running", pod.ID)
return nil
}
glog.Infof("pod %q containers not running: syncing", pod.Name)
glog.Infof("pod %q containers not running: syncing", pod.ID)
if err = kl.syncPod(&pod, dockerContainers); err != nil {
return fmt.Errorf("error syncing pod: %v", err)
}
if retry >= RunOnceMaxRetries {
return fmt.Errorf("timeout error: pod %q containers not running after %d retries", pod.Name, RunOnceMaxRetries)
return fmt.Errorf("timeout error: pod %q containers not running after %d retries", pod.ID, RunOnceMaxRetries)
}
// TODO(proppy): health checking would be better than waiting + checking the state at the next iteration.
glog.Infof("pod %q containers synced, waiting for %v", pod.Name, delay)
glog.Infof("pod %q containers synced, waiting for %v", pod.ID, delay)
<-time.After(delay)
retry++
delay *= RunOnceRetryDelayBackoff
......@@ -114,9 +115,9 @@ func (kl *Kubelet) runPod(pod Pod) error {
}
// isPodRunning returns true if all containers of a manifest are running.
func (kl *Kubelet) isPodRunning(pod Pod, dockerContainers dockertools.DockerContainers) bool {
for _, container := range pod.Manifest.Containers {
if dockerContainer, found, _ := dockerContainers.FindPodContainer(GetPodFullName(&pod), pod.Manifest.UUID, container.Name); !found || dockerContainer.Status != "running" {
func (kl *Kubelet) isPodRunning(pod api.BoundPod, dockerContainers dockertools.DockerContainers) bool {
for _, container := range pod.Spec.Containers {
if dockerContainer, found, _ := dockerContainers.FindPodContainer(GetPodFullName(&pod), pod.UID, container.Name); !found || dockerContainer.Status != "running" {
glog.Infof("container %q not found (%v) or not running: %#v", container.Name, found, dockerContainer)
return false
}
......
pkg/kubelet/runonce_test.go
View file @ 892942af
......@@ -69,12 +69,12 @@ func TestRunOnce(t *testing.T) {
kb := &Kubelet{}
podContainers := []docker.APIContainers{
{
Names: []string{"/k8s_bar." + strconv.FormatUint(dockertools.HashContainer(&api.Container{Name: "bar"}), 16) + "_foo.test"},
Names: []string{"/k8s_bar." + strconv.FormatUint(dockertools.HashContainer(&api.Container{Name: "bar"}), 16) + "_foo.new.test"},
ID: "1234",
Status: "running",
},
{
Names: []string{"/k8s_net_foo.test_"},
Names: []string{"/k8s_net_foo.new.test_"},
ID: "9876",
Status: "running",
},
......@@ -106,12 +106,14 @@ func TestRunOnce(t *testing.T) {
t: t,
}
kb.dockerPuller = &dockertools.FakeDockerPuller{}
results, err := kb.runOnce([]Pod{
results, err := kb.runOnce([]api.BoundPod{
{
Name: "foo",
Namespace: "test",
Manifest: api.ContainerManifest{
ID: "foo",
TypeMeta: api.TypeMeta{
ID: "foo",
Namespace: "new",
Annotations: map[string]string{ConfigSourceAnnotationKey: "test"},
},
Spec: api.PodSpec{
Containers: []api.Container{
{Name: "bar"},
},
......@@ -124,7 +126,7 @@ func TestRunOnce(t *testing.T) {
if results[0].Err != nil {
t.Errorf("unexpected run pod error: %v", results[0].Err)
}
if results[0].Pod.Name != "foo" {
t.Errorf("unexpected pod: %q", results[0].Pod.Name)
if results[0].Pod.ID != "foo" {
t.Errorf("unexpected pod: %q", results[0].Pod.ID)
}
}
pkg/kubelet/server.go
View file @ 892942af
......@@ -119,15 +119,26 @@ func (s *Server) handleContainer(w http.ResponseWriter, req *http.Request) {
return
}
// This is to provide backward compatibility. It only supports a single manifest
var pod Pod
err = yaml.Unmarshal(data, &pod.Manifest)
var pod api.BoundPod
var containerManifest api.ContainerManifest
err = yaml.Unmarshal(data, &containerManifest)
if err != nil {
s.error(w, err)
return
}
pod.ID = containerManifest.ID
pod.UID = containerManifest.UUID
pod.Spec.Containers = containerManifest.Containers
pod.Spec.Volumes = containerManifest.Volumes
pod.Spec.RestartPolicy = containerManifest.RestartPolicy
//TODO: sha1 of manifest?
pod.Name = "1"
s.updates <- PodUpdate{[]Pod{pod}, SET}
if pod.ID == "" {
pod.ID = "1"
}
if pod.UID == "" {
pod.UID = "1"
}
s.updates <- PodUpdate{[]api.BoundPod{pod}, SET}
}
......@@ -139,16 +150,16 @@ func (s *Server) handleContainers(w http.ResponseWriter, req *http.Request) {
s.error(w, err)
return
}
var manifests []api.ContainerManifest
err = yaml.Unmarshal(data, &manifests)
var specs []api.PodSpec
err = yaml.Unmarshal(data, &specs)
if err != nil {
s.error(w, err)
return
}
pods := make([]Pod, len(manifests))
for i := range manifests {
pods[i].Name = fmt.Sprintf("%d", i+1)
pods[i].Manifest = manifests[i]
pods := make([]api.BoundPod, len(specs))
for i := range specs {
pods[i].ID = fmt.Sprintf("%d", i+1)
pods[i].Spec = specs[i]
}
s.updates <- PodUpdate{pods, SET}
......@@ -186,7 +197,14 @@ func (s *Server) handleContainerLogs(w http.ResponseWriter, req *http.Request) {
follow, _ := strconv.ParseBool(uriValues.Get("follow"))
tail := uriValues.Get("tail")
podFullName := GetPodFullName(&Pod{Name: podID, Namespace: "etcd"})
podFullName := GetPodFullName(&api.BoundPod{
TypeMeta: api.TypeMeta{
ID: podID,
// TODO: I am broken
Namespace: api.NamespaceDefault,
Annotations: map[string]string{ConfigSourceAnnotationKey: "etcd"},
},
})
fw := FlushWriter{writer: w}
if flusher, ok := w.(http.Flusher); ok {
......@@ -216,10 +234,17 @@ func (s *Server) handlePodInfo(w http.ResponseWriter, req *http.Request) {
return
}
// TODO: backwards compatibility with existing API, needs API change
podFullName := GetPodFullName(&Pod{Name: podID, Namespace: "etcd"})
podFullName := GetPodFullName(&api.BoundPod{
TypeMeta: api.TypeMeta{
ID: podID,
// TODO: I am broken
Namespace: api.NamespaceDefault,
Annotations: map[string]string{ConfigSourceAnnotationKey: "etcd"},
},
})
info, err := s.host.GetPodInfo(podFullName, podUUID)
if err == dockertools.ErrNoContainersInPod {
http.Error(w, "Pod does not exist", http.StatusNotFound)
http.Error(w, "api.BoundPod does not exist", http.StatusNotFound)
return
}
if err != nil {
......@@ -283,7 +308,14 @@ func (s *Server) handleRun(w http.ResponseWriter, req *http.Request) {
http.Error(w, "Unexpected path for command running", http.StatusBadRequest)
return
}
podFullName := GetPodFullName(&Pod{Name: podID, Namespace: "etcd"})
podFullName := GetPodFullName(&api.BoundPod{
TypeMeta: api.TypeMeta{
ID: podID,
// TODO: I am broken
Namespace: api.NamespaceDefault,
Annotations: map[string]string{ConfigSourceAnnotationKey: "etcd"},
},
})
command := strings.Split(u.Query().Get("cmd"), " ")
data, err := s.host.RunInContainer(podFullName, uuid, container, command)
if err != nil {
......@@ -327,10 +359,24 @@ func (s *Server) serveStats(w http.ResponseWriter, req *http.Request) {
errors.New("pod level status currently unimplemented")
case 3:
// Backward compatibility without uuid information
podFullName := GetPodFullName(&Pod{Name: components[1], Namespace: "etcd"})
podFullName := GetPodFullName(&api.BoundPod{
TypeMeta: api.TypeMeta{
ID: components[1],
// TODO: I am broken
Namespace: api.NamespaceDefault,
Annotations: map[string]string{ConfigSourceAnnotationKey: "etcd"},
},
})
stats, err = s.host.GetContainerInfo(podFullName, "", components[2], &query)
case 4:
podFullName := GetPodFullName(&Pod{Name: components[1], Namespace: "etcd"})
podFullName := GetPodFullName(&api.BoundPod{
TypeMeta: api.TypeMeta{
ID: components[1],
// TODO: I am broken
Namespace: "",
Annotations: map[string]string{ConfigSourceAnnotationKey: "etcd"},
},
})
stats, err = s.host.GetContainerInfo(podFullName, components[2], components[2], &query)
default:
http.Error(w, "unknown resource.", http.StatusNotFound)
......
pkg/kubelet/server_test.go
View file @ 892942af
......@@ -101,7 +101,23 @@ func readResp(resp *http.Response) (string, error) {
func TestContainer(t *testing.T) {
fw := newServerTest()
expected := []api.ContainerManifest{
{ID: "test_manifest"},
{
ID: "test_manifest",
UUID: "value",
Containers: []api.Container{
{
Name: "container",
},
},
Volumes: []api.Volume{
{
Name: "test",
},
},
RestartPolicy: api.RestartPolicy{
Never: &api.RestartPolicyNever{},
},
},
}
body := bytes.NewBuffer([]byte(util.EncodeJSON(expected[0]))) // Only send a single ContainerManifest
resp, err := http.Post(fw.testHTTPServer.URL+"/container", "application/json", body)
......@@ -114,7 +130,29 @@ func TestContainer(t *testing.T) {
if len(received) != 1 {
t.Errorf("Expected 1 manifest, but got %v", len(received))
}
expectedPods := []Pod{{Name: "1", Manifest: expected[0]}}
expectedPods := []api.BoundPod{
{
TypeMeta: api.TypeMeta{
ID: "test_manifest",
UID: "value",
},
Spec: api.PodSpec{
Containers: []api.Container{
{
Name: "container",
},
},
Volumes: []api.Volume{
{
Name: "test",
},
},
RestartPolicy: api.RestartPolicy{
Never: &api.RestartPolicyNever{},
},
},
},
}
if !reflect.DeepEqual(expectedPods, received[0]) {
t.Errorf("Expected %#v, but got %#v", expectedPods, received[0])
}
......@@ -123,8 +161,38 @@ func TestContainer(t *testing.T) {
func TestContainers(t *testing.T) {
fw := newServerTest()
expected := []api.ContainerManifest{
{ID: "test_manifest_1"},
{ID: "test_manifest_2"},
{
ID: "test_manifest_1",
Containers: []api.Container{
{
Name: "container",
},
},
Volumes: []api.Volume{
{
Name: "test",
},
},
RestartPolicy: api.RestartPolicy{
Never: &api.RestartPolicyNever{},
},
},
{
ID: "test_manifest_2",
Containers: []api.Container{
{
Name: "container2",
},
},
Volumes: []api.Volume{
{
Name: "test2",
},
},
RestartPolicy: api.RestartPolicy{
Never: &api.RestartPolicyNever{},
},
},
}
body := bytes.NewBuffer([]byte(util.EncodeJSON(expected)))
resp, err := http.Post(fw.testHTTPServer.URL+"/containers", "application/json", body)
......@@ -137,7 +205,48 @@ func TestContainers(t *testing.T) {
if len(received) != 1 {
t.Errorf("Expected 1 update, but got %v", len(received))
}
expectedPods := []Pod{{Name: "1", Manifest: expected[0]}, {Name: "2", Manifest: expected[1]}}
expectedPods := []api.BoundPod{
{
TypeMeta: api.TypeMeta{
ID: "1",
},
Spec: api.PodSpec{
Containers: []api.Container{
{
Name: "container",
},
},
Volumes: []api.Volume{
{
Name: "test",
},
},
RestartPolicy: api.RestartPolicy{
Never: &api.RestartPolicyNever{},
},
},
},
{
TypeMeta: api.TypeMeta{
ID: "2",
},
Spec: api.PodSpec{
Containers: []api.Container{
{
Name: "container2",
},
},
Volumes: []api.Volume{
{
Name: "test2",
},
},
RestartPolicy: api.RestartPolicy{
Never: &api.RestartPolicyNever{},
},
},
},
}
if !reflect.DeepEqual(expectedPods, received[0]) {
t.Errorf("Expected %#v, but got %#v", expectedPods, received[0])
}
......@@ -149,7 +258,7 @@ func TestPodInfo(t *testing.T) {
"goodpod": api.ContainerStatus{},
}
fw.fakeKubelet.infoFunc = func(name string) (api.PodInfo, error) {
if name == "goodpod.etcd" {
if name == "goodpod.default.etcd" {
return expected, nil
}
return nil, fmt.Errorf("bad pod %s", name)
......@@ -175,7 +284,7 @@ func TestContainerInfo(t *testing.T) {
fw := newServerTest()
expectedInfo := &info.ContainerInfo{}
podID := "somepod"
expectedPodID := "somepod" + ".etcd"
expectedPodID := "somepod" + ".default.etcd"
expectedContainerName := "goodcontainer"
fw.fakeKubelet.containerInfoFunc = func(podID, containerName string, req *info.ContainerInfoRequest) (*info.ContainerInfo, error) {
if podID != expectedPodID || containerName != expectedContainerName {
......@@ -278,7 +387,7 @@ func TestServeRunInContainer(t *testing.T) {
fw := newServerTest()
output := "foo bar"
podName := "foo"
expectedPodName := podName + ".etcd"
expectedPodName := podName + ".default.etcd"
expectedContainerName := "baz"
expectedCommand := "ls -a"
fw.fakeKubelet.runFunc = func(podFullName, uuid, containerName string, cmd []string) ([]byte, error) {
......@@ -317,7 +426,7 @@ func TestServeRunInContainerWithUUID(t *testing.T) {
fw := newServerTest()
output := "foo bar"
podName := "foo"
expectedPodName := podName + ".etcd"
expectedPodName := podName + ".default.etcd"
expectedUuid := "7e00838d_-_3523_-_11e4_-_8421_-_42010af0a720"
expectedContainerName := "baz"
expectedCommand := "ls -a"
......@@ -360,7 +469,7 @@ func TestContainerLogs(t *testing.T) {
fw := newServerTest()
output := "foo bar"
podName := "foo"
expectedPodName := podName + ".etcd"
expectedPodName := podName + ".default.etcd"
expectedContainerName := "baz"
expectedTail := ""
expectedFollow := false
......@@ -399,7 +508,7 @@ func TestContainerLogsWithTail(t *testing.T) {
fw := newServerTest()
output := "foo bar"
podName := "foo"
expectedPodName := podName + ".etcd"
expectedPodName := podName + ".default.etcd"
expectedContainerName := "baz"
expectedTail := "5"
expectedFollow := false
......@@ -438,7 +547,7 @@ func TestContainerLogsWithFollow(t *testing.T) {
fw := newServerTest()
output := "foo bar"
podName := "foo"
expectedPodName := podName + ".etcd"
expectedPodName := podName + ".default.etcd"
expectedContainerName := "baz"
expectedTail := ""
expectedFollow := true
......
pkg/kubelet/types.go
View file @ 892942af
......@@ -22,13 +22,7 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
)
// Pod represents the structure of a pod on the Kubelet, distinct from the apiserver
// representation of a Pod.
type Pod struct {
Namespace string
Name string
Manifest api.ContainerManifest
}
const ConfigSourceAnnotationKey = "kubernetes/config.source"
// PodOperation defines what changes will be made on a pod configuration.
type PodOperation int
......@@ -48,13 +42,13 @@ const (
// sending an array of size one and Op == ADD|REMOVE (with REMOVE, only the ID is required).
// For setting the state of the system to a given state for this source configuration, set
// Pods as desired and Op to SET, which will reset the system state to that specified in this
// operation for this source channel. To remove all pods, set Pods to empty array and Op to SET.
// operation for this source channel. To remove all pods, set Pods to empty object and Op to SET.
type PodUpdate struct {
Pods []Pod
Pods []api.BoundPod
Op PodOperation
}
// GetPodFullName returns a name that full identifies a pod across all config sources.
func GetPodFullName(pod *Pod) string {
return fmt.Sprintf("%s.%s", pod.Name, pod.Namespace)
// GetPodFullName returns a name that uniquely identifies a pod across all config sources.
func GetPodFullName(pod *api.BoundPod) string {
return fmt.Sprintf("%s.%s.%s", pod.ID, pod.Namespace, pod.Annotations[ConfigSourceAnnotationKey])
}
pkg/kubelet/validation.go deleted 100644 → 0
View file @ 332a03b0
/*
Copyright 2014 Google Inc. All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package kubelet
import (
apierrs "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
)
func ValidatePod(pod *Pod) (errors []error) {
if !util.IsDNSSubdomain(pod.Name) {
errors = append(errors, apierrs.NewFieldInvalid("name", pod.Name))
}
if errs := validation.ValidateManifest(&pod.Manifest); len(errs) != 0 {
errors = append(errors, errs...)
}
return errors
}
pkg/kubelet/validation_test.go deleted 100644 → 0
View file @ 332a03b0
/*
Copyright 2014 Google Inc. All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package kubelet_test
import (
"strings"
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
. "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
)
func TestValidatePodNoName(t *testing.T) {
errorCases := map[string]Pod{
// manifest is tested in api/validation_test.go, ensure it is invoked
"empty version": {Name: "test", Manifest: api.ContainerManifest{Version: "", ID: "abc"}},
// Name
"zero-length name": {Name: "", Manifest: api.ContainerManifest{Version: "v1beta1"}},
"name > 255 characters": {Name: strings.Repeat("a", 256), Manifest: api.ContainerManifest{Version: "v1beta1"}},
"name not a DNS subdomain": {Name: "a.b.c.", Manifest: api.ContainerManifest{Version: "v1beta1"}},
"name with underscore": {Name: "a_b_c", Manifest: api.ContainerManifest{Version: "v1beta1"}},
}
for k, v := range errorCases {
if errs := ValidatePod(&v); len(errs) == 0 {
t.Errorf("expected failure for %s", k)
}
}
}
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