Skip to content

K
k3s
Commit 673d061c authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #40838 from kow3ns/ss-fixes

Automatic merge from submit-queue (batch tested with PRs 40796, 40878, 36033, 40838, 41210) StatefulSet hardening **What this PR does / why we need it**: This PR contains the following changes to StatefulSet. Only one change effects the semantics of how the controller operates (This is described in #38418), and this change only brings the controller into conformance with its documented behavior. 1. pcb and pcb controller are removed and their functionality is encapsulated in StatefulPodControlInterface. This class modules the design contoller.PodControlInterface and provides an abstraction to clientset.Interface which is useful for testing purposes. 2. IdentityMappers has been removed to clarify what properties of a Pod are mutated by the controller. All mutations are performed in the UpdateStatefulPod method of the StatefulPodControlInterface. 3. The statefulSetIterator and petQueue classes are removed. These classes sorted Pods by CreationTimestamp. This is brittle and not resilient to clock skew. The current control loop, which implements the same logic, is in stateful_set_control.go. The Pods are now sorted and considered by their ordinal indices, as is outlined in the documentation. 4. StatefulSetController now checks to see if the Pods matching a StatefulSet's Selector also match the Name of the StatefulSet. This will make the controller resilient to overlapping, and will be enhanced by the addition of ControllerRefs. 5. The total lines of production code have been reduced, and the total number of unit tests has been increased. All new code has 100% unit coverage giving the module 83% coverage. Tests for StatefulSetController have been added, but it is not practical to achieve greater coverage in unit testing for this code (the e2e tests for StatefulSet cover these areas). 6. Issue #38418 is fixed in that StaefulSet will ensure that all Pods that are predecessors of another Pod are Running and Ready prior to launching a new Pod. This removes the potential for deadlock when a Pod needs to be rescheduled while its predecessor is hung in Pending or Initializing. 7. All reference to pet have been removed from the code and comments. **Which issue this PR fixes** fixes #38418,#36859 **Special notes for your reviewer**: **Release note**: ```release-note Fixes issue #38418 which, under circumstance, could cause StatefulSet to deadlock. Mediates issue #36859. StatefulSet only acts on Pods whose identity matches the StatefulSet, providing a partial mediation for overlapping controllers. ```
parents 45d122dd 4d99b4d8
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 749 additions and 1283 deletions
pkg/apis/apps/v1beta1/types.go
View file @ 673d061c
......@@ -21,6 +21,11 @@ import (
"k8s.io/kubernetes/pkg/api/v1"
)
const (
// StatefulSetInitAnnotation if present, and set to false, indicates that a Pod's readiness should be ignored.
StatefulSetInitAnnotation = "pod.alpha.kubernetes.io/initialized"
)
// +genclient=true
// StatefulSet represents a set of pods with consistent identities.
......
pkg/controller/statefulset/BUILD
View file @ 673d061c
......@@ -11,12 +11,10 @@ load(
go_library(
name = "go_default_library",
srcs = [
"fakes.go",
"identity_mappers.go",
"iterator.go",
"stateful_pod_control.go",
"stateful_set.go",
"stateful_set_control.go",
"stateful_set_utils.go",
"statefulpod.go",
],
tags = ["automanaged"],
deps = [
......@@ -25,19 +23,14 @@ go_library(
"//pkg/api/v1/pod:go_default_library",
"//pkg/apis/apps/v1beta1:go_default_library",
"//pkg/client/clientset_generated/clientset:go_default_library",
"//pkg/client/clientset_generated/clientset/typed/apps/v1beta1:go_default_library",
"//pkg/client/legacylisters:go_default_library",
"//pkg/controller:go_default_library",
"//vendor:github.com/golang/glog",
"//vendor:gopkg.in/inf.v0",
"//vendor:k8s.io/apimachinery/pkg/api/errors",
"//vendor:k8s.io/apimachinery/pkg/api/resource",
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
"//vendor:k8s.io/apimachinery/pkg/runtime",
"//vendor:k8s.io/apimachinery/pkg/types",
"//vendor:k8s.io/apimachinery/pkg/util/errors",
"//vendor:k8s.io/apimachinery/pkg/util/runtime",
"//vendor:k8s.io/apimachinery/pkg/util/sets",
"//vendor:k8s.io/apimachinery/pkg/util/wait",
"//vendor:k8s.io/apimachinery/pkg/watch",
"//vendor:k8s.io/client-go/kubernetes/typed/core/v1",
......@@ -51,33 +44,29 @@ go_library(
go_test(
name = "go_default_test",
srcs = [
"identity_mappers_test.go",
"iterator_test.go",
"stateful_pod_control_test.go",
"stateful_set_control_test.go",
"stateful_set_test.go",
"statefulpod_test.go",
"stateful_set_utils_test.go",
],
library = ":go_default_library",
tags = ["automanaged"],
deps = [
"//pkg/api:go_default_library",
"//pkg/api/testapi:go_default_library",
"//pkg/api/v1:go_default_library",
"//pkg/api/v1/pod:go_default_library",
"//pkg/apis/apps/v1beta1:go_default_library",
"//pkg/client/clientset_generated/clientset:go_default_library",
"//pkg/client/clientset_generated/clientset/fake:go_default_library",
"//pkg/client/clientset_generated/clientset/typed/apps/v1beta1:go_default_library",
"//pkg/client/clientset_generated/clientset/typed/apps/v1beta1/fake:go_default_library",
"//pkg/client/legacylisters:go_default_library",
"//pkg/controller:go_default_library",
"//vendor:k8s.io/apimachinery/pkg/api/errors",
"//vendor:k8s.io/apimachinery/pkg/api/resource",
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
"//vendor:k8s.io/apimachinery/pkg/runtime",
"//vendor:k8s.io/apimachinery/pkg/util/errors",
"//vendor:k8s.io/apimachinery/pkg/util/sets",
"//vendor:k8s.io/client-go/rest",
"//vendor:k8s.io/apimachinery/pkg/types",
"//vendor:k8s.io/client-go/testing",
"//vendor:k8s.io/client-go/tools/cache",
"//vendor:k8s.io/client-go/util/testing",
"//vendor:k8s.io/client-go/tools/record",
"//vendor:k8s.io/client-go/util/workqueue",
],
)
......
pkg/controller/statefulset/fakes.go deleted 100644 → 0
View file @ 45d122dd
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"fmt"
"time"
inf "gopkg.in/inf.v0"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/client-go/tools/record"
"k8s.io/kubernetes/pkg/api/v1"
apipod "k8s.io/kubernetes/pkg/api/v1/pod"
apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
)
func dec(i int64, exponent int) *inf.Dec {
return inf.NewDec(i, inf.Scale(-exponent))
}
func newPVC(name string) v1.PersistentVolumeClaim {
return v1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
Name: name,
},
Spec: v1.PersistentVolumeClaimSpec{
Resources: v1.ResourceRequirements{
Requests: v1.ResourceList{
v1.ResourceStorage: *resource.NewQuantity(1, resource.BinarySI),
},
},
},
}
}
func newStatefulSetWithVolumes(replicas int, name string, petMounts []v1.VolumeMount, podMounts []v1.VolumeMount) *apps.StatefulSet {
mounts := append(petMounts, podMounts...)
claims := []v1.PersistentVolumeClaim{}
for _, m := range petMounts {
claims = append(claims, newPVC(m.Name))
}
vols := []v1.Volume{}
for _, m := range podMounts {
vols = append(vols, v1.Volume{
Name: m.Name,
VolumeSource: v1.VolumeSource{
HostPath: &v1.HostPathVolumeSource{
Path: fmt.Sprintf("/tmp/%v", m.Name),
},
},
})
}
return &apps.StatefulSet{
TypeMeta: metav1.TypeMeta{
Kind: "StatefulSet",
APIVersion: "apps/v1beta1",
},
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: metav1.NamespaceDefault,
UID: types.UID("test"),
},
Spec: apps.StatefulSetSpec{
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{"foo": "bar"},
},
Replicas: func() *int32 { i := int32(replicas); return &i }(),
Template: v1.PodTemplateSpec{
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "nginx",
Image: "nginx",
VolumeMounts: mounts,
},
},
Volumes: vols,
},
},
VolumeClaimTemplates: claims,
ServiceName: "governingsvc",
},
}
}
func runningPod(ns, name string) *v1.Pod {
p := &v1.Pod{Status: v1.PodStatus{Phase: v1.PodRunning}}
p.Namespace = ns
p.Name = name
return p
}
func newPodList(ps *apps.StatefulSet, num int) []*v1.Pod {
// knownPods are pods in the system
knownPods := []*v1.Pod{}
for i := 0; i < num; i++ {
k, _ := newPCB(fmt.Sprintf("%v", i), ps)
knownPods = append(knownPods, k.pod)
}
return knownPods
}
func newStatefulSet(replicas int) *apps.StatefulSet {
petMounts := []v1.VolumeMount{
{Name: "datadir", MountPath: "/tmp/zookeeper"},
}
podMounts := []v1.VolumeMount{
{Name: "home", MountPath: "/home"},
}
return newStatefulSetWithVolumes(replicas, "foo", petMounts, podMounts)
}
func checkPodForMount(pod *v1.Pod, mountName string) error {
for _, c := range pod.Spec.Containers {
for _, v := range c.VolumeMounts {
if v.Name == mountName {
return nil
}
}
}
return fmt.Errorf("Found volume but no associated mount %v in pod %v", mountName, pod.Name)
}
func newFakePetClient() *fakePetClient {
return &fakePetClient{
pets: []*pcb{},
claims: []v1.PersistentVolumeClaim{},
recorder: &record.FakeRecorder{},
petHealthChecker: &defaultPetHealthChecker{},
}
}
type fakePetClient struct {
pets []*pcb
claims []v1.PersistentVolumeClaim
petsCreated int
petsDeleted int
claimsCreated int
claimsDeleted int
recorder record.EventRecorder
petHealthChecker
}
// Delete fakes pet client deletion.
func (f *fakePetClient) Delete(p *pcb) error {
pets := []*pcb{}
found := false
for i, pet := range f.pets {
if p.pod.Name == pet.pod.Name {
found = true
f.recorder.Eventf(pet.parent, v1.EventTypeNormal, "SuccessfulDelete", "pod: %v", pet.pod.Name)
continue
}
pets = append(pets, f.pets[i])
}
if !found {
// TODO: Return proper not found error
return fmt.Errorf("Delete failed: pod %v doesn't exist", p.pod.Name)
}
f.pets = pets
f.petsDeleted++
return nil
}
// Get fakes getting pets.
func (f *fakePetClient) Get(p *pcb) (*pcb, bool, error) {
for i, pet := range f.pets {
if p.pod.Name == pet.pod.Name {
return f.pets[i], true, nil
}
}
return nil, false, nil
}
// Create fakes pet creation.
func (f *fakePetClient) Create(p *pcb) error {
for _, pet := range f.pets {
if p.pod.Name == pet.pod.Name {
return fmt.Errorf("Create failed: pod %v already exists", p.pod.Name)
}
}
f.recorder.Eventf(p.parent, v1.EventTypeNormal, "SuccessfulCreate", "pod: %v", p.pod.Name)
f.pets = append(f.pets, p)
f.petsCreated++
return nil
}
// Update fakes pet updates.
func (f *fakePetClient) Update(expected, wanted *pcb) error {
found := false
pets := []*pcb{}
for i, pet := range f.pets {
if wanted.pod.Name == pet.pod.Name {
f.pets[i].pod.Annotations[apipod.PodHostnameAnnotation] = wanted.pod.Annotations[apipod.PodHostnameAnnotation]
f.pets[i].pod.Annotations[apipod.PodSubdomainAnnotation] = wanted.pod.Annotations[apipod.PodSubdomainAnnotation]
f.pets[i].pod.Spec = wanted.pod.Spec
found = true
}
pets = append(pets, f.pets[i])
}
f.pets = pets
if !found {
return fmt.Errorf("Cannot update pod %v not found", wanted.pod.Name)
}
// TODO: Delete pvcs/volumes that are in wanted but not in expected.
return nil
}
func (f *fakePetClient) getPodList() []*v1.Pod {
p := []*v1.Pod{}
for i, pet := range f.pets {
if pet.pod == nil {
continue
}
p = append(p, f.pets[i].pod)
}
return p
}
func (f *fakePetClient) deletePetAtIndex(index int) {
p := []*pcb{}
for i := range f.pets {
if i != index {
p = append(p, f.pets[i])
}
}
f.pets = p
}
func (f *fakePetClient) setHealthy(index int) error {
if len(f.pets) <= index {
return fmt.Errorf("Index out of range, len %v index %v", len(f.pets), index)
}
f.pets[index].pod.Status.Phase = v1.PodRunning
f.pets[index].pod.Annotations[StatefulSetInitAnnotation] = "true"
f.pets[index].pod.Status.Conditions = []v1.PodCondition{
{Type: v1.PodReady, Status: v1.ConditionTrue},
}
return nil
}
// isHealthy is a convenience wrapper around the default health checker.
// The first invocation returns not-healthy, but marks the pet healthy so
// subsequent invocations see it as healthy.
func (f *fakePetClient) isHealthy(pod *v1.Pod) bool {
if f.petHealthChecker.isHealthy(pod) {
return true
}
return false
}
func (f *fakePetClient) setDeletionTimestamp(index int) error {
if len(f.pets) <= index {
return fmt.Errorf("Index out of range, len %v index %v", len(f.pets), index)
}
f.pets[index].pod.DeletionTimestamp = &metav1.Time{Time: time.Now()}
return nil
}
// SyncPVCs fakes pvc syncing.
func (f *fakePetClient) SyncPVCs(pet *pcb) error {
v := pet.pvcs
updateClaims := map[string]v1.PersistentVolumeClaim{}
for i, update := range v {
updateClaims[update.Name] = v[i]
}
claimList := []v1.PersistentVolumeClaim{}
for i, existing := range f.claims {
if update, ok := updateClaims[existing.Name]; ok {
claimList = append(claimList, update)
delete(updateClaims, existing.Name)
} else {
claimList = append(claimList, f.claims[i])
}
}
for _, remaining := range updateClaims {
claimList = append(claimList, remaining)
f.claimsCreated++
f.recorder.Eventf(pet.parent, v1.EventTypeNormal, "SuccessfulCreate", "pvc: %v", remaining.Name)
}
f.claims = claimList
return nil
}
// DeletePVCs fakes pvc deletion.
func (f *fakePetClient) DeletePVCs(pet *pcb) error {
claimsToDelete := pet.pvcs
deleteClaimNames := sets.NewString()
for _, c := range claimsToDelete {
deleteClaimNames.Insert(c.Name)
}
pvcs := []v1.PersistentVolumeClaim{}
for i, existing := range f.claims {
if deleteClaimNames.Has(existing.Name) {
deleteClaimNames.Delete(existing.Name)
f.claimsDeleted++
f.recorder.Eventf(pet.parent, v1.EventTypeNormal, "SuccessfulDelete", "pvc: %v", existing.Name)
continue
}
pvcs = append(pvcs, f.claims[i])
}
f.claims = pvcs
if deleteClaimNames.Len() != 0 {
return fmt.Errorf("Claims %+v don't exist. Failed deletion.", deleteClaimNames)
}
return nil
}
pkg/controller/statefulset/identity_mappers.go deleted 100644 → 0
View file @ 45d122dd
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"crypto/md5"
"fmt"
"sort"
"strings"
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/kubernetes/pkg/api/v1"
podapi "k8s.io/kubernetes/pkg/api/v1/pod"
apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
)
// identityMapper is an interface for assigning identities to a pet.
// All existing identity mappers just append "-(index)" to the statefulset name to
// generate a unique identity. This is used in claims/DNS/hostname/petname
// etc. There's a more elegant way to achieve this mapping, but we're
// taking the simplest route till we have data on whether users will need
// more customization.
// Note that running a single identity mapper is not guaranteed to give
// your pet a unique identity. You must run them all. Order doesn't matter.
type identityMapper interface {
// SetIdentity takes an id and assigns the given pet an identity based
// on the stateful set spec. The is must be unique amongst members of the
// stateful set.
SetIdentity(id string, pet *v1.Pod)
// Identity returns the identity of the pet.
Identity(pod *v1.Pod) string
}
func newIdentityMappers(ps *apps.StatefulSet) []identityMapper {
return []identityMapper{
&NameIdentityMapper{ps},
&NetworkIdentityMapper{ps},
&VolumeIdentityMapper{ps},
}
}
// NetworkIdentityMapper assigns network identity to pets.
type NetworkIdentityMapper struct {
ps *apps.StatefulSet
}
// SetIdentity sets network identity on the pet.
func (n *NetworkIdentityMapper) SetIdentity(id string, pet *v1.Pod) {
pet.Annotations[podapi.PodHostnameAnnotation] = fmt.Sprintf("%v-%v", n.ps.Name, id)
pet.Annotations[podapi.PodSubdomainAnnotation] = n.ps.Spec.ServiceName
return
}
// Identity returns the network identity of the pet.
func (n *NetworkIdentityMapper) Identity(pet *v1.Pod) string {
return n.String(pet)
}
// String is a string function for the network identity of the pet.
func (n *NetworkIdentityMapper) String(pet *v1.Pod) string {
hostname := pet.Annotations[podapi.PodHostnameAnnotation]
subdomain := pet.Annotations[podapi.PodSubdomainAnnotation]
return strings.Join([]string{hostname, subdomain, n.ps.Namespace}, ".")
}
// VolumeIdentityMapper assigns storage identity to pets.
type VolumeIdentityMapper struct {
ps *apps.StatefulSet
}
// SetIdentity sets storage identity on the pet.
func (v *VolumeIdentityMapper) SetIdentity(id string, pet *v1.Pod) {
petVolumes := []v1.Volume{}
petClaims := v.GetClaims(id)
// These volumes will all go down with the pod. If a name matches one of
// the claims in the stateful set, it gets clobbered.
podVolumes := map[string]v1.Volume{}
for _, podVol := range pet.Spec.Volumes {
podVolumes[podVol.Name] = podVol
}
// Insert claims for the idempotent statefulset volumes
for name, claim := range petClaims {
// Volumes on a pet for which there are no associated claims on the
// statefulset are pod local, and die with the pod.
podVol, ok := podVolumes[name]
if ok {
// TODO: Validate and reject this.
glog.V(4).Infof("Overwriting existing volume source %v", podVol.Name)
}
newVol := v1.Volume{
Name: name,
VolumeSource: v1.VolumeSource{
PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
ClaimName: claim.Name,
// TODO: Use source definition to set this value when we have one.
ReadOnly: false,
},
},
}
petVolumes = append(petVolumes, newVol)
}
// Transfer any ephemeral pod volumes
for name, vol := range podVolumes {
if _, ok := petClaims[name]; !ok {
petVolumes = append(petVolumes, vol)
}
}
pet.Spec.Volumes = petVolumes
return
}
// Identity returns the storage identity of the pet.
func (v *VolumeIdentityMapper) Identity(pet *v1.Pod) string {
// TODO: Make this a hash?
return v.String(pet)
}
// String is a string function for the network identity of the pet.
func (v *VolumeIdentityMapper) String(pet *v1.Pod) string {
ids := []string{}
petVols := sets.NewString()
for _, petVol := range v.ps.Spec.VolumeClaimTemplates {
petVols.Insert(petVol.Name)
}
for _, podVol := range pet.Spec.Volumes {
// Volumes on a pet for which there are no associated claims on the
// statefulset are pod local, and die with the pod.
if !petVols.Has(podVol.Name) {
continue
}
if podVol.VolumeSource.PersistentVolumeClaim == nil {
// TODO: Is this a part of the identity?
ids = append(ids, fmt.Sprintf("%v:None", podVol.Name))
continue
}
ids = append(ids, fmt.Sprintf("%v:%v", podVol.Name, podVol.VolumeSource.PersistentVolumeClaim.ClaimName))
}
sort.Strings(ids)
return strings.Join(ids, "")
}
// GetClaims returns the volume claims associated with the given id.
// The claims belong to the statefulset. The id should be unique within a statefulset.
func (v *VolumeIdentityMapper) GetClaims(id string) map[string]v1.PersistentVolumeClaim {
petClaims := map[string]v1.PersistentVolumeClaim{}
for _, pvc := range v.ps.Spec.VolumeClaimTemplates {
claim := pvc
// TODO: Name length checking in validation.
claim.Name = fmt.Sprintf("%v-%v-%v", claim.Name, v.ps.Name, id)
claim.Namespace = v.ps.Namespace
claim.Labels = v.ps.Spec.Selector.MatchLabels
// TODO: We're assuming that the claim template has a volume QoS key, eg:
// volume.alpha.kubernetes.io/storage-class: anything
petClaims[pvc.Name] = claim
}
return petClaims
}
// GetClaimsForPet returns the pvcs for the given pet.
func (v *VolumeIdentityMapper) GetClaimsForPet(pet *v1.Pod) []v1.PersistentVolumeClaim {
// Strip out the "-(index)" from the pet name and use it to generate
// claim names.
id := strings.Split(pet.Name, "-")
petID := id[len(id)-1]
pvcs := []v1.PersistentVolumeClaim{}
for _, pvc := range v.GetClaims(petID) {
pvcs = append(pvcs, pvc)
}
return pvcs
}
// NameIdentityMapper assigns names to pets.
// It also puts the pet in the same namespace as the parent.
type NameIdentityMapper struct {
ps *apps.StatefulSet
}
// SetIdentity sets the pet namespace and name.
func (n *NameIdentityMapper) SetIdentity(id string, pet *v1.Pod) {
pet.Name = fmt.Sprintf("%v-%v", n.ps.Name, id)
pet.Namespace = n.ps.Namespace
return
}
// Identity returns the name identity of the pet.
func (n *NameIdentityMapper) Identity(pet *v1.Pod) string {
return n.String(pet)
}
// String is a string function for the name identity of the pet.
func (n *NameIdentityMapper) String(pet *v1.Pod) string {
return fmt.Sprintf("%v/%v", pet.Namespace, pet.Name)
}
// identityHash computes a hash of the pet by running all the above identity
// mappers.
func identityHash(ps *apps.StatefulSet, pet *v1.Pod) string {
id := ""
for _, idMapper := range newIdentityMappers(ps) {
id += idMapper.Identity(pet)
}
return fmt.Sprintf("%x", md5.Sum([]byte(id)))
}
// copyPetID gives the realPet the same identity as the expectedPet.
// Note that this is *not* a literal copy, but a copy of the fields that
// contribute to the pet's identity. The returned boolean 'needsUpdate' will
// be false if the realPet already has the same identity as the expectedPet.
func copyPetID(realPet, expectedPet *pcb) (pod v1.Pod, needsUpdate bool, err error) {
if realPet.pod == nil || expectedPet.pod == nil {
return pod, false, fmt.Errorf("Need a valid to and from pet for copy")
}
if realPet.parent.UID != expectedPet.parent.UID {
return pod, false, fmt.Errorf("Cannot copy pets with different parents")
}
ps := realPet.parent
if identityHash(ps, realPet.pod) == identityHash(ps, expectedPet.pod) {
return *realPet.pod, false, nil
}
copyPod := *realPet.pod
// This is the easiest way to give an identity to a pod. It won't work
// when we stop using names for id.
for _, idMapper := range newIdentityMappers(ps) {
idMapper.SetIdentity(expectedPet.id, &copyPod)
}
return copyPod, true, nil
}
pkg/controller/statefulset/identity_mappers_test.go deleted 100644 → 0
View file @ 45d122dd
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"fmt"
"reflect"
"strings"
"testing"
"k8s.io/kubernetes/pkg/api/v1"
apipod "k8s.io/kubernetes/pkg/api/v1/pod"
)
func TestPetIDName(t *testing.T) {
replicas := 3
ps := newStatefulSet(replicas)
for i := 0; i < replicas; i++ {
petName := fmt.Sprintf("%v-%d", ps.Name, i)
pcb, err := newPCB(fmt.Sprintf("%d", i), ps)
if err != nil {
t.Fatalf("Failed to generate pet %v", err)
}
pod := pcb.pod
if pod.Name != petName || pod.Namespace != ps.Namespace {
t.Errorf("Wrong name identity, expected %v", pcb.pod.Name)
}
}
}
func TestPetIDDNS(t *testing.T) {
replicas := 3
ps := newStatefulSet(replicas)
for i := 0; i < replicas; i++ {
petName := fmt.Sprintf("%v-%d", ps.Name, i)
petSubdomain := ps.Spec.ServiceName
pcb, err := newPCB(fmt.Sprintf("%d", i), ps)
pod := pcb.pod
if err != nil {
t.Fatalf("Failed to generate pet %v", err)
}
if hostname, ok := pod.Annotations[apipod.PodHostnameAnnotation]; !ok || hostname != petName {
t.Errorf("Wrong hostname: %v", hostname)
}
if subdomain, ok := pod.Annotations[apipod.PodSubdomainAnnotation]; !ok || subdomain != petSubdomain {
t.Errorf("Wrong subdomain: %v", subdomain)
}
}
}
func TestPetIDVolume(t *testing.T) {
replicas := 3
ps := newStatefulSet(replicas)
for i := 0; i < replicas; i++ {
pcb, err := newPCB(fmt.Sprintf("%d", i), ps)
if err != nil {
t.Fatalf("Failed to generate pet %v", err)
}
pod := pcb.pod
petName := fmt.Sprintf("%v-%d", ps.Name, i)
claimName := fmt.Sprintf("datadir-%v", petName)
for _, v := range pod.Spec.Volumes {
switch v.Name {
case "datadir":
c := v.VolumeSource.PersistentVolumeClaim
if c == nil || c.ClaimName != claimName {
t.Fatalf("Unexpected claim %v", c)
}
if err := checkPodForMount(pod, "datadir"); err != nil {
t.Errorf("Expected pod mount: %v", err)
}
case "home":
h := v.VolumeSource.HostPath
if h == nil || h.Path != "/tmp/home" {
t.Errorf("Unexpected modification to hostpath, expected /tmp/home got %+v", h)
}
default:
t.Errorf("Unexpected volume %v", v.Name)
}
}
}
// TODO: Check volume mounts.
}
func TestPetIDVolumeClaims(t *testing.T) {
replicas := 3
ps := newStatefulSet(replicas)
for i := 0; i < replicas; i++ {
pcb, err := newPCB(fmt.Sprintf("%v", i), ps)
if err != nil {
t.Fatalf("Failed to generate pet %v", err)
}
pvcs := pcb.pvcs
petName := fmt.Sprintf("%v-%d", ps.Name, i)
claimName := fmt.Sprintf("datadir-%v", petName)
if len(pvcs) != 1 || pvcs[0].Name != claimName {
t.Errorf("Wrong pvc expected %v got %v", claimName, pvcs[0].Name)
}
}
}
func TestPetIDCrossAssignment(t *testing.T) {
replicas := 3
ps := newStatefulSet(replicas)
nameMapper := &NameIdentityMapper{ps}
volumeMapper := &VolumeIdentityMapper{ps}
networkMapper := &NetworkIdentityMapper{ps}
// Check that the name is consistent across identity.
for i := 0; i < replicas; i++ {
pet, _ := newPCB(fmt.Sprintf("%v", i), ps)
p := pet.pod
name := strings.Split(nameMapper.Identity(p), "/")[1]
network := networkMapper.Identity(p)
volume := volumeMapper.Identity(p)
petVolume := strings.Split(volume, ":")[1]
if petVolume != fmt.Sprintf("datadir-%v", name) {
t.Errorf("Unexpected pet volume name %v, expected %v", petVolume, name)
}
if network != fmt.Sprintf("%v.%v.%v", name, ps.Spec.ServiceName, ps.Namespace) {
t.Errorf("Unexpected pet network ID %v, expected %v", network, name)
}
t.Logf("[%v] volume: %+v, network: %+v, name: %+v", i, volume, network, name)
}
}
func TestPetIDReset(t *testing.T) {
replicas := 2
ps := newStatefulSet(replicas)
firstPCB, err := newPCB("1", ps)
secondPCB, err := newPCB("2", ps)
if identityHash(ps, firstPCB.pod) == identityHash(ps, secondPCB.pod) {
t.Fatalf("Failed to generate uniquey identities:\n%+v\n%+v", firstPCB.pod.Spec, secondPCB.pod.Spec)
}
userAdded := v1.Volume{
Name: "test",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{Medium: v1.StorageMediumMemory},
},
}
firstPCB.pod.Spec.Volumes = append(firstPCB.pod.Spec.Volumes, userAdded)
pod, needsUpdate, err := copyPetID(firstPCB, secondPCB)
if err != nil {
t.Errorf("%v", err)
}
if !needsUpdate {
t.Errorf("expected update since identity of %v was reset", secondPCB.pod.Name)
}
if identityHash(ps, &pod) != identityHash(ps, secondPCB.pod) {
t.Errorf("Failed to copy identity for pod %v -> %v", firstPCB.pod.Name, secondPCB.pod.Name)
}
foundVol := false
for _, v := range pod.Spec.Volumes {
if reflect.DeepEqual(v, userAdded) {
foundVol = true
break
}
}
if !foundVol {
t.Errorf("User added volume was corrupted by reset action.")
}
}
pkg/controller/statefulset/iterator.go deleted 100644 → 0
View file @ 45d122dd
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"fmt"
"sort"
"github.com/golang/glog"
"k8s.io/kubernetes/pkg/api/v1"
apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
"k8s.io/kubernetes/pkg/controller"
)
// newPCB generates a new PCB using the id string as a unique qualifier
func newPCB(id string, ps *apps.StatefulSet) (*pcb, error) {
petPod, err := controller.GetPodFromTemplate(&ps.Spec.Template, ps, nil)
if err != nil {
return nil, err
}
for _, im := range newIdentityMappers(ps) {
im.SetIdentity(id, petPod)
}
petPVCs := []v1.PersistentVolumeClaim{}
vMapper := &VolumeIdentityMapper{ps}
for _, c := range vMapper.GetClaims(id) {
petPVCs = append(petPVCs, c)
}
// TODO: Replace id field with IdentityHash, since id is more than just an index.
return &pcb{pod: petPod, pvcs: petPVCs, id: id, parent: ps}, nil
}
// petQueue is a custom datastructure that's resembles a queue of pets.
type petQueue struct {
pets []*pcb
idMapper identityMapper
}
// enqueue enqueues the given pet, evicting any pets with the same id
func (pt *petQueue) enqueue(p *pcb) {
if p == nil {
pt.pets = append(pt.pets, nil)
return
}
// Pop an existing pet from the know list, append the new pet to the end.
petList := []*pcb{}
petID := pt.idMapper.Identity(p.pod)
for i := range pt.pets {
if petID != pt.idMapper.Identity(pt.pets[i].pod) {
petList = append(petList, pt.pets[i])
}
}
pt.pets = petList
p.event = syncPet
pt.pets = append(pt.pets, p)
}
// dequeue returns the last element of the queue
func (pt *petQueue) dequeue() *pcb {
if pt.empty() {
glog.Warningf("Dequeue invoked on an empty queue")
return nil
}
l := len(pt.pets) - 1
pet := pt.pets[l]
pt.pets = pt.pets[:l]
return pet
}
// empty returns true if the pet queue is empty.
func (pt *petQueue) empty() bool {
return len(pt.pets) == 0
}
// NewPetQueue returns a queue for tracking pets
func NewPetQueue(ps *apps.StatefulSet, podList []*v1.Pod) *petQueue {
pt := petQueue{pets: []*pcb{}, idMapper: &NameIdentityMapper{ps}}
// Seed the queue with existing pets. Assume all pets are scheduled for
// deletion, enqueuing a pet will "undelete" it. We always want to delete
// from the higher ids, so sort by creation timestamp.
sort.Sort(PodsByCreationTimestamp(podList))
vMapper := VolumeIdentityMapper{ps}
for i := range podList {
pod := podList[i]
pt.pets = append(pt.pets, &pcb{pod: pod, pvcs: vMapper.GetClaimsForPet(pod), parent: ps, event: deletePet, id: fmt.Sprintf("%v", i)})
}
return &pt
}
// statefulsetIterator implements a simple iterator over pets in the given statefulset.
type statefulSetIterator struct {
// ps is the statefulset for this iterator.
ps *apps.StatefulSet
// queue contains the elements to iterate over.
queue *petQueue
// errs is a list because we always want the iterator to drain.
errs []error
// petCount is the number of pets iterated over.
petCount int32
}
// Next returns true for as long as there are elements in the underlying queue.
func (pi *statefulSetIterator) Next() bool {
var pet *pcb
var err error
if pi.petCount < *(pi.ps.Spec.Replicas) {
pet, err = newPCB(fmt.Sprintf("%d", pi.petCount), pi.ps)
if err != nil {
pi.errs = append(pi.errs, err)
// Don't stop iterating over the set on errors. Caller handles nil.
pet = nil
}
pi.queue.enqueue(pet)
pi.petCount++
}
// Keep the iterator running till we've deleted pets in the queue.
return !pi.queue.empty()
}
// Value dequeues an element from the queue.
func (pi *statefulSetIterator) Value() *pcb {
return pi.queue.dequeue()
}
// NewStatefulSetIterator returns a new iterator. All pods in the given podList
// are used to seed the queue of the iterator.
func NewStatefulSetIterator(ps *apps.StatefulSet, podList []*v1.Pod) *statefulSetIterator {
pi := &statefulSetIterator{
ps: ps,
queue: NewPetQueue(ps, podList),
errs: []error{},
petCount: 0,
}
return pi
}
// PodsByCreationTimestamp sorts a list of Pods by creation timestamp, using their names as a tie breaker.
type PodsByCreationTimestamp []*v1.Pod
func (o PodsByCreationTimestamp) Len() int { return len(o) }
func (o PodsByCreationTimestamp) Swap(i, j int) { o[i], o[j] = o[j], o[i] }
func (o PodsByCreationTimestamp) Less(i, j int) bool {
if o[i].CreationTimestamp.Equal(o[j].CreationTimestamp) {
return o[i].Name < o[j].Name
}
return o[i].CreationTimestamp.Before(o[j].CreationTimestamp)
}
pkg/controller/statefulset/iterator_test.go deleted 100644 → 0
View file @ 45d122dd
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"fmt"
"testing"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/kubernetes/pkg/api/v1"
)
func TestPetQueueCreates(t *testing.T) {
replicas := 3
ps := newStatefulSet(replicas)
q := NewPetQueue(ps, []*v1.Pod{})
for i := 0; i < replicas; i++ {
pet, _ := newPCB(fmt.Sprintf("%v", i), ps)
q.enqueue(pet)
p := q.dequeue()
if p.event != syncPet {
t.Errorf("Failed to retrieve sync event from queue")
}
}
if q.dequeue() != nil {
t.Errorf("Expected no pods")
}
}
func TestPetQueueScaleDown(t *testing.T) {
replicas := 1
ps := newStatefulSet(replicas)
// knownPods are the pods in the system
knownPods := newPodList(ps, 3)
q := NewPetQueue(ps, knownPods)
// The iterator will insert a single replica, the enqueue
// mimics that behavior.
pet, _ := newPCB(fmt.Sprintf("%v", 0), ps)
q.enqueue(pet)
deletes := sets.NewString(fmt.Sprintf("%v-1", ps.Name), fmt.Sprintf("%v-2", ps.Name))
syncs := sets.NewString(fmt.Sprintf("%v-0", ps.Name))
// Confirm that 2 known pods are deleted
for i := 0; i < 3; i++ {
p := q.dequeue()
switch p.event {
case syncPet:
if !syncs.Has(p.pod.Name) {
t.Errorf("Unexpected sync %v expecting %+v", p.pod.Name, syncs)
}
case deletePet:
if !deletes.Has(p.pod.Name) {
t.Errorf("Unexpected deletes %v expecting %+v", p.pod.Name, deletes)
}
}
}
if q.dequeue() != nil {
t.Errorf("Expected no pods")
}
}
func TestPetQueueScaleUp(t *testing.T) {
replicas := 5
ps := newStatefulSet(replicas)
// knownPods are pods in the system
knownPods := newPodList(ps, 2)
q := NewPetQueue(ps, knownPods)
for i := 0; i < 5; i++ {
pet, _ := newPCB(fmt.Sprintf("%v", i), ps)
q.enqueue(pet)
}
for i := 4; i >= 0; i-- {
pet := q.dequeue()
expectedName := fmt.Sprintf("%v-%d", ps.Name, i)
if pet.event != syncPet || pet.pod.Name != expectedName {
t.Errorf("Unexpected pod %+v, expected %v", pet.pod.Name, expectedName)
}
}
}
func TestStatefulSetIteratorRelist(t *testing.T) {
replicas := 5
ps := newStatefulSet(replicas)
// knownPods are pods in the system
knownPods := newPodList(ps, 5)
for i := range knownPods {
knownPods[i].Spec.NodeName = fmt.Sprintf("foo-node-%v", i)
knownPods[i].Status.Phase = v1.PodRunning
}
pi := NewStatefulSetIterator(ps, knownPods)
// A simple resync should not change identity of pods in the system
i := 0
for pi.Next() {
p := pi.Value()
if identityHash(ps, p.pod) != identityHash(ps, knownPods[i]) {
t.Errorf("Got unexpected identity hash from iterator.")
}
if p.event != syncPet {
t.Errorf("Got unexpected sync event for %v: %v", p.pod.Name, p.event)
}
i++
}
if i != 5 {
t.Errorf("Unexpected iterations %v, this probably means too many/few pods", i)
}
// Scale to 0 should delete all pods in system
*(ps.Spec.Replicas) = 0
pi = NewStatefulSetIterator(ps, knownPods)
i = 0
for pi.Next() {
p := pi.Value()
if p.event != deletePet {
t.Errorf("Got unexpected sync event for %v: %v", p.pod.Name, p.event)
}
i++
}
if i != 5 {
t.Errorf("Unexpected iterations %v, this probably means too many/few pods", i)
}
// Relist with 0 replicas should no-op
pi = NewStatefulSetIterator(ps, []*v1.Pod{})
if pi.Next() != false {
t.Errorf("Unexpected iteration without any replicas or pods in system")
}
}
pkg/controller/statefulset/stateful_pod_control.go 0 → 100644
View file @ 673d061c
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"fmt"
"strings"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
errorutils "k8s.io/apimachinery/pkg/util/errors"
"k8s.io/client-go/tools/record"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/v1"
apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
)
// StatefulPodControlInterface defines the interface that StatefulSetController uses to create, update, and delete Pods,
// and to update the Status of a StatefulSet. It follows the design paradigms used for PodControl, but its
// implementation provides for PVC creation, ordered Pod creation, ordered Pod termination, and Pod identity enforcement.
// Like controller.PodControlInterface, it is implemented as an interface to provide for testing fakes.
type StatefulPodControlInterface interface {
// CreateStatefulPod create a Pod in a StatefulSet. Any PVCs necessary for the Pod are created prior to creating
// the Pod. If the returned error is nil the Pod and its PVCs have been created.
CreateStatefulPod(set *apps.StatefulSet, pod *v1.Pod) error
// UpdateStatefulPod Updates a Pod in a StatefulSet. If the Pod already has the correct identity and stable
// storage this method is a no-op. If the Pod must be mutated to conform to the Set, it is mutated and updated.
// pod is an in-out parameter, and any updates made to the pod are reflected as mutations to this parameter. If
// the create is successful, the returned error is nil.
UpdateStatefulPod(set *apps.StatefulSet, pod *v1.Pod) error
// DeleteStatefulPod deletes a Pod in a StatefulSet. The pods PVCs are not deleted. If the delete is successful,
// the returned error is nil.
DeleteStatefulPod(set *apps.StatefulSet, pod *v1.Pod) error
// UpdateStatefulSetStatus Updates the Status.Replicas of a StatefulSet. set is an in-out parameter, and any
// updates made to the set are made visible as mutations to the parameter. If the method is successful, the
// returned error is nil, and set has its Status.Replicas field set to replicas.
UpdateStatefulSetReplicas(set *apps.StatefulSet, replicas int32) error
}
func NewRealStatefulPodControl(client clientset.Interface, recorder record.EventRecorder) StatefulPodControlInterface {
return &realStatefulPodControl{client, recorder}
}
// realStatefulPodControl implements StatefulPodControlInterface using a clientset.Interface to communicate with the
// API server. The struct is package private as the internal details are irrelevant to importing packages.
type realStatefulPodControl struct {
client clientset.Interface
recorder record.EventRecorder
}
func (spc *realStatefulPodControl) CreateStatefulPod(set *apps.StatefulSet, pod *v1.Pod) error {
// Create the Pod's PVCs prior to creating the Pod
if err := spc.createPersistentVolumeClaims(set, pod); err != nil {
spc.recordPodEvent("create", set, pod, err)
return err
}
// If we created the PVCs attempt to create the Pod
_, err := spc.client.Core().Pods(set.Namespace).Create(pod)
// sink already exists errors
if apierrors.IsAlreadyExists(err) {
return err
}
spc.recordPodEvent("create", set, pod, err)
return err
}
func (spc *realStatefulPodControl) UpdateStatefulPod(set *apps.StatefulSet, pod *v1.Pod) error {
// we make a copy of the Pod on the stack and mutate the copy
// we copy back to pod to notify the caller of successful mutation
obj, err := api.Scheme.Copy(pod)
if err != nil {
return fmt.Errorf("unable to copy pod: %v", err)
}
podCopy := obj.(*v1.Pod)
for attempt := 0; attempt < maxUpdateRetries; attempt++ {
// assume the Pod is consistent
consistent := true
// if the Pod does not conform to it's identity, update the identity and dirty the Pod
if !identityMatches(set, podCopy) {
updateIdentity(set, podCopy)
consistent = false
}
// if the Pod does not conform to the StatefulSet's storage requirements, update the Pod's PVC's,
// dirty the Pod, and create any missing PVCs
if !storageMatches(set, podCopy) {
updateStorage(set, podCopy)
consistent = false
if err := spc.createPersistentVolumeClaims(set, podCopy); err != nil {
spc.recordPodEvent("update", set, pod, err)
return err
}
}
// if the Pod is not dirty do nothing
if consistent {
*pod = *podCopy
return nil
}
// commit the update, retrying on conflicts
_, err = spc.client.Core().Pods(set.Namespace).Update(podCopy)
if !apierrors.IsConflict(err) {
if err == nil {
*pod = *podCopy
}
spc.recordPodEvent("update", set, pod, err)
return err
}
conflicting, err := spc.client.Core().Pods(set.Namespace).Get(podCopy.Name, metav1.GetOptions{})
if err != nil {
spc.recordPodEvent("update", set, podCopy, err)
return err
}
*podCopy = *conflicting
}
spc.recordPodEvent("update", set, pod, updateConflictError)
return updateConflictError
}
func (spc *realStatefulPodControl) DeleteStatefulPod(set *apps.StatefulSet, pod *v1.Pod) error {
err := spc.client.Core().Pods(set.Namespace).Delete(pod.Name, nil)
spc.recordPodEvent("delete", set, pod, err)
return err
}
func (spc *realStatefulPodControl) UpdateStatefulSetReplicas(set *apps.StatefulSet, replicas int32) error {
if set.Status.Replicas == replicas {
return nil
}
obj, err := api.Scheme.Copy(set)
if err != nil {
return fmt.Errorf("unable to copy set: %v", err)
}
setCopy := obj.(*apps.StatefulSet)
setCopy.Status.Replicas = replicas
for attempt := 0; attempt < maxUpdateRetries; attempt++ {
_, err := spc.client.Apps().StatefulSets(setCopy.Namespace).UpdateStatus(setCopy)
if !apierrors.IsConflict(err) {
if err == nil {
*set = *setCopy
}
return err
}
conflicting, err := spc.client.Apps().StatefulSets(setCopy.Namespace).Get(setCopy.Name, metav1.GetOptions{})
if err != nil {
return err
}
conflicting.Status.Replicas = setCopy.Status.Replicas
*setCopy = *conflicting
}
return updateConflictError
}
// recordPodEvent records an event for verb applied to a Pod in a StatefulSet. If err is nil the generated event will
// have a reason of v1.EventTypeNormal. If err is not nil the generated event will have a reason of v1.EventTypeWarning.
func (spc *realStatefulPodControl) recordPodEvent(verb string, set *apps.StatefulSet, pod *v1.Pod, err error) {
if err == nil {
reason := fmt.Sprintf("Successful%s", strings.Title(verb))
message := fmt.Sprintf("%s Pod %s in StatefulSet %s successful",
strings.ToLower(verb), pod.Name, set.Name)
spc.recorder.Event(set, v1.EventTypeNormal, reason, message)
} else {
reason := fmt.Sprintf("Failed%s", strings.Title(verb))
message := fmt.Sprintf("%s Pod %s in StatefulSet %s failed error: %s",
strings.ToLower(verb), pod.Name, set.Name, err)
spc.recorder.Event(set, v1.EventTypeWarning, reason, message)
}
}
// recordClaimEvent records an event for verb applied to the PersistentVolumeClaim of a Pod in a StatefulSet. If err is
// nil the generated event will have a reason of v1.EventTypeNormal. If err is not nil the generated event will have a
// reason of v1.EventTypeWarning.
func (spc *realStatefulPodControl) recordClaimEvent(verb string, set *apps.StatefulSet, pod *v1.Pod, claim *v1.PersistentVolumeClaim, err error) {
if err == nil {
reason := fmt.Sprintf("Successful%s", strings.Title(verb))
message := fmt.Sprintf("%s Claim %s Pod %s in StatefulSet %s success",
strings.ToLower(verb), claim.Name, pod.Name, set.Name)
spc.recorder.Event(set, v1.EventTypeNormal, reason, message)
} else {
reason := fmt.Sprintf("Failed%s", strings.Title(verb))
message := fmt.Sprintf("%s Claim %s for Pod %s in StatefulSet %s failed error: %s",
strings.ToLower(verb), claim.Name, pod.Name, set.Name, err)
spc.recorder.Event(set, v1.EventTypeWarning, reason, message)
}
}
// createPersistentVolumeClaims creates all of the required PersistentVolumeClaims for pod, which mush be a member of
// set. If all of the claims for Pod are successfully created, the returned error is nil. If creation fails, this method
// may be called again until no error is returned, indicating the PersistentVolumeClaims for pod are consistent with
// set's Spec.
func (spc *realStatefulPodControl) createPersistentVolumeClaims(set *apps.StatefulSet, pod *v1.Pod) error {
var errs []error
for _, claim := range getPersistentVolumeClaims(set, pod) {
_, err := spc.client.Core().PersistentVolumeClaims(claim.Namespace).Get(claim.Name, metav1.GetOptions{})
if err != nil {
if apierrors.IsNotFound(err) {
_, err := spc.client.Core().PersistentVolumeClaims(claim.Namespace).Create(&claim)
if err != nil {
errs = append(errs, fmt.Errorf("Failed to create PVC %s: %s", claim.Name, err))
}
spc.recordClaimEvent("create", set, pod, &claim, err)
} else {
errs = append(errs, fmt.Errorf("Failed to retrieve PVC %s: %s", claim.Name, err))
spc.recordClaimEvent("create", set, pod, &claim, err)
}
}
// TODO: Check resource requirements and accessmodes, update if necessary
}
return errorutils.NewAggregate(errs)
}
var _ StatefulPodControlInterface = &realStatefulPodControl{}
pkg/controller/statefulset/stateful_set_control.go 0 → 100644
View file @ 673d061c
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"sort"
"k8s.io/kubernetes/pkg/api/v1"
apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
"github.com/golang/glog"
)
// StatefulSetControl implements the control logic for updating StatefulSets and their children Pods. It is implemented
// as an interface to allow for extensions that provide different semantics. Currently, there is only one implementation.
type StatefulSetControlInterface interface {
// UpdateStatefulSet implements the control logic for Pod creation, update, and deletion, and
// persistent volume creation, update, and deletion.
// If an implementation returns a non-nil error, the invocation will be retried using a rate-limited strategy.
// Implementors should sink any errors that they do not wish to trigger a retry, and they may feel free to
// exit exceptionally at any point provided they wish the update to be re-run at a later point in time.
UpdateStatefulSet(set *apps.StatefulSet, pods []*v1.Pod) error
}
// NewDefaultStatefulSetControl returns a new instance of the default implementation StatefulSetControlInterface that
// implements the documented semantics for StatefulSets. podControl is the PodControlInterface used to create, update,
// and delete Pods and to create PersistentVolumeClaims. You should use an instance returned from
// NewRealStatefulPodControl() for any scenario other than testing.
func NewDefaultStatefulSetControl(podControl StatefulPodControlInterface) StatefulSetControlInterface {
return &defaultStatefulSetControl{podControl}
}
type defaultStatefulSetControl struct {
podControl StatefulPodControlInterface
}
func (ssc *defaultStatefulSetControl) UpdateStatefulSet(set *apps.StatefulSet, pods []*v1.Pod) error {
replicaCount := int(*set.Spec.Replicas)
// slice that will contain all Pods such that 0 <= getOrdinal(pod) < set.Spec.Replicas
replicas := make([]*v1.Pod, replicaCount)
// slice that will contain all Pods such that set.Spec.Replicas <= getOrdinal(pod)
condemned := make([]*v1.Pod, 0, len(pods))
ready := 0
unhealthy := 0
// First we partition pods into two lists valid replicas and condemned Pods
for i := range pods {
//count the number of running and ready replicas
if isRunningAndReady(pods[i]) {
ready++
}
if ord := getOrdinal(pods[i]); 0 <= ord && ord < replicaCount {
// if the ordinal of the pod is within the range of the current number of replicas,
// insert it at the indirection of its ordinal
replicas[ord] = pods[i]
} else if ord >= replicaCount {
// if the ordinal is greater than the number of replicas add it to the condemned list
condemned = append(condemned, pods[i])
}
}
// for any empty indices in the sequence [0,set.Spec.Replicas) create a new Pod
for ord := 0; ord < replicaCount; ord++ {
if replicas[ord] == nil {
replicas[ord] = newStatefulSetPod(set, ord)
}
}
// count the number of unhealthy pods
for i := range replicas {
if !isHealthy(replicas[i]) {
unhealthy++
}
}
for i := range condemned {
if !isHealthy(condemned[i]) {
unhealthy++
}
}
// sort the condemned Pods by their ordinals
sort.Sort(ascendingOrdinal(condemned))
// if the current number of replicas has changed update the statefulSets replicas
if err := ssc.podControl.UpdateStatefulSetReplicas(set, int32(ready)); err != nil {
return err
}
// Examine each replica with respect to its ordinal
for i := range replicas {
// delete and recreate failed pods
if isFailed(replicas[i]) {
glog.V(2).Infof("StatefulSet %s is recreating failed Pod %s", set.Name, replicas[i].Name)
if err := ssc.podControl.DeleteStatefulPod(set, replicas[i]); err != nil {
return err
}
replicas[i] = newStatefulSetPod(set, i)
}
// If we find a Pod that has not been created we create the Pod immediately and return
if !isCreated(replicas[i]) {
return ssc.podControl.CreateStatefulPod(set, replicas[i])
}
// If we have a Pod that has been created but is not running and ready we can not make progress.
// We must ensure that all for each Pod, when we create it, all of its predecessors, with respect to its
// ordinal, are Running and Ready.
if !isRunningAndReady(replicas[i]) {
glog.V(2).Infof("StatefulSet %s is waiting for Pod %s to be Running and Ready",
set.Name, replicas[i].Name)
return nil
}
// Enforce the StatefulSet invariants,
if err := ssc.podControl.UpdateStatefulPod(set, replicas[i]); err != nil {
return err
}
}
// At this point, all of the current Replicas are Running and Ready, we can consider termination.
// We will wait for all predecessors to be Running and Ready prior to attempting a deletion.
// We will terminate Pods in a monotonically decreasing order over [len(pods),set.Spec.Replicas).
// Note that we do not resurrect Pods in this interval.
if unhealthy > 0 {
glog.V(2).Infof("StatefulSet %s is waiting on %d Pods", set.Name, unhealthy)
return nil
}
if target := len(condemned) - 1; target >= 0 {
glog.V(2).Infof("StatefulSet %s terminating Pod %s", set.Name, condemned[target])
return ssc.podControl.DeleteStatefulPod(set, condemned[target])
}
return nil
}
var _ StatefulSetControlInterface = &defaultStatefulSetControl{}
pkg/controller/statefulset/statefulpod_test.go deleted 100644 → 0
View file @ 45d122dd
/*
Copyright 2016 The Kubernetes Authors.
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 statefulset
import (
"fmt"
"net/http/httptest"
"testing"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
restclient "k8s.io/client-go/rest"
core "k8s.io/client-go/testing"
utiltesting "k8s.io/client-go/util/testing"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/testapi"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
)
func newPetClient(client *clientset.Clientset) *apiServerPetClient {
return &apiServerPetClient{
c: client,
}
}
func makeTwoDifferntPCB() (pcb1, pcb2 *pcb) {
userAdded := v1.Volume{
Name: "test",
VolumeSource: v1.VolumeSource{
EmptyDir: &v1.EmptyDirVolumeSource{Medium: v1.StorageMediumMemory},
},
}
ps := newStatefulSet(2)
pcb1, _ = newPCB("1", ps)
pcb2, _ = newPCB("2", ps)
pcb2.pod.Spec.Volumes = append(pcb2.pod.Spec.Volumes, userAdded)
return pcb1, pcb2
}
func TestUpdatePetWithoutRetry(t *testing.T) {
pcb1, pcb2 := makeTwoDifferntPCB()
// invalid pet with empty pod
invalidPcb := *pcb1
invalidPcb.pod = nil
testCases := []struct {
realPet *pcb
expectedPet *pcb
expectErr bool
requests int
}{
// case 0: error occurs, no need to update
{
realPet: pcb1,
expectedPet: &invalidPcb,
expectErr: true,
requests: 0,
},
// case 1: identical pet, no need to update
{
realPet: pcb1,
expectedPet: pcb1,
expectErr: false,
requests: 0,
},
// case 2: need to call update once
{
realPet: pcb1,
expectedPet: pcb2,
expectErr: false,
requests: 1,
},
}
for k, tc := range testCases {
body := runtime.EncodeOrDie(testapi.Default.Codec(), &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "empty_pod"}})
fakeHandler := utiltesting.FakeHandler{
StatusCode: 200,
ResponseBody: string(body),
}
testServer := httptest.NewServer(&fakeHandler)
client := clientset.NewForConfigOrDie(&restclient.Config{Host: testServer.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
petClient := newPetClient(client)
err := petClient.Update(tc.realPet, tc.expectedPet)
if tc.expectErr != (err != nil) {
t.Errorf("case %d: expect error(%v), got err: %v", k, tc.expectErr, err)
}
fakeHandler.ValidateRequestCount(t, tc.requests)
testServer.Close()
}
}
func TestUpdatePetWithFailure(t *testing.T) {
fakeHandler := utiltesting.FakeHandler{
StatusCode: 500,
ResponseBody: "{}",
}
testServer := httptest.NewServer(&fakeHandler)
defer testServer.Close()
client := clientset.NewForConfigOrDie(&restclient.Config{Host: testServer.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
petClient := newPetClient(client)
pcb1, pcb2 := makeTwoDifferntPCB()
if err := petClient.Update(pcb1, pcb2); err == nil {
t.Errorf("expect error, got nil")
}
// 1 Update and 1 GET, both of which fail
fakeHandler.ValidateRequestCount(t, 2)
}
func TestUpdatePetRetrySucceed(t *testing.T) {
pcb1, pcb2 := makeTwoDifferntPCB()
fakeClient := &fake.Clientset{}
fakeClient.AddReactor("get", "pods", func(action core.Action) (bool, runtime.Object, error) {
return true, pcb2.pod, nil
})
fakeClient.AddReactor("*", "*", func(action core.Action) (bool, runtime.Object, error) {
return true, nil, fmt.Errorf("Fake error")
})
petClient := apiServerPetClient{
c: fakeClient,
}
if err := petClient.Update(pcb1, pcb2); err != nil {
t.Errorf("unexpected error: %v", err)
}
actions := fakeClient.Actions()
if len(actions) != 2 {
t.Errorf("Expect 2 actions, got %d actions", len(actions))
}
for i := 0; i < len(actions); i++ {
a := actions[i]
if a.GetResource().Resource != "pods" {
t.Errorf("Unexpected action %+v", a)
continue
}
switch action := a.(type) {
case core.GetAction:
if i%2 == 0 {
t.Errorf("Unexpected Get action")
}
// Make sure the get is for the right pod
if action.GetName() != pcb2.pod.Name {
t.Errorf("Expected get pod %v, got %q instead", pcb2.pod.Name, action.GetName())
}
case core.UpdateAction:
if i%2 == 1 {
t.Errorf("Unexpected Update action")
}
default:
t.Errorf("Unexpected action %+v", a)
break
}
}
}
test/e2e/BUILD
View file @ 673d061c
......@@ -134,7 +134,6 @@ go_library(
"//pkg/controller/job:go_default_library",
"//pkg/controller/replicaset:go_default_library",
"//pkg/controller/replication:go_default_library",
"//pkg/controller/statefulset:go_default_library",
"//pkg/kubectl:go_default_library",
"//pkg/kubectl/cmd/util:go_default_library",
"//pkg/kubelet/api/v1alpha1/stats:go_default_library",
......@@ -177,7 +176,6 @@ go_library(
"//vendor:golang.org/x/net/websocket",
"//vendor:google.golang.org/api/compute/v1",
"//vendor:google.golang.org/api/googleapi",
"//vendor:gopkg.in/inf.v0",
"//vendor:k8s.io/apimachinery/pkg/api/errors",
"//vendor:k8s.io/apimachinery/pkg/api/resource",
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
......
test/e2e/statefulset.go
View file @ 673d061c
......@@ -24,8 +24,6 @@ import (
"strings"
"time"
inf "gopkg.in/inf.v0"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
apierrs "k8s.io/apimachinery/pkg/api/errors"
......@@ -44,7 +42,6 @@ import (
"k8s.io/kubernetes/pkg/api/v1"
apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
"k8s.io/kubernetes/pkg/controller/statefulset"
"k8s.io/kubernetes/test/e2e/framework"
)
......@@ -77,7 +74,7 @@ var _ = framework.KubeDescribe("StatefulSet", func() {
ns = f.Namespace.Name
})
framework.KubeDescribe("Basic StatefulSet functionality", func() {
framework.KubeDescribe("Basic StatefulSet functionality [StatefulSetBasic]", func() {
ssName := "ss"
labels := map[string]string{
"foo": "bar",
......@@ -144,7 +141,7 @@ var _ = framework.KubeDescribe("StatefulSet", func() {
framework.ExpectNoError(sst.execInStatefulPods(ss, cmd))
})
It("should handle healthy stateful pod restarts during scale", func() {
It("should not deadlock when a pod's predecessor fails", func() {
By("Creating statefulset " + ssName + " in namespace " + ns)
*(ss.Spec.Replicas) = 2
setInitializedAnnotation(ss, "false")
......@@ -169,8 +166,8 @@ var _ = framework.KubeDescribe("StatefulSet", func() {
By("Deleting healthy stateful pod at index 0.")
sst.deleteStatefulPodAtIndex(0, ss)
By("Confirming stateful pod at index 0 is not recreated.")
sst.confirmStatefulPodCount(1, ss, 10*time.Second)
By("Confirming stateful pod at index 0 is recreated.")
sst.waitForRunningAndReady(2, ss)
By("Deleting unhealthy stateful pod at index 1.")
sst.deleteStatefulPodAtIndex(1, ss)
......@@ -937,10 +934,10 @@ func (s *statefulSetTester) setHealthy(ss *apps.StatefulSet) {
framework.Failf("Found multiple non-healthy stateful pods: %v and %v", pod.Name, markedHealthyPod)
}
p, err := framework.UpdatePodWithRetries(s.c, pod.Namespace, pod.Name, func(update *v1.Pod) {
update.Annotations[statefulset.StatefulSetInitAnnotation] = "true"
update.Annotations[apps.StatefulSetInitAnnotation] = "true"
})
framework.ExpectNoError(err)
framework.Logf("Set annotation %v to %v on pod %v", statefulset.StatefulSetInitAnnotation, p.Annotations[statefulset.StatefulSetInitAnnotation], pod.Name)
framework.Logf("Set annotation %v to %v on pod %v", apps.StatefulSetInitAnnotation, p.Annotations[apps.StatefulSetInitAnnotation], pod.Name)
markedHealthyPod = pod.Name
}
}
......@@ -1016,7 +1013,7 @@ func deleteAllStatefulSets(c clientset.Interface, ns string) {
return true, nil
})
if pvcPollErr != nil {
errList = append(errList, fmt.Sprintf("Timeout waiting for pvc deletion."))
errList = append(errList, "Timeout waiting for pvc deletion.")
}
pollErr := wait.PollImmediate(statefulsetPoll, statefulsetTimeout, func() (bool, error) {
......@@ -1038,7 +1035,7 @@ func deleteAllStatefulSets(c clientset.Interface, ns string) {
return false, nil
})
if pollErr != nil {
errList = append(errList, fmt.Sprintf("Timeout waiting for pv provisioner to delete pvs, this might mean the test leaked pvs."))
errList = append(errList, "Timeout waiting for pv provisioner to delete pvs, this might mean the test leaked pvs.")
}
if len(errList) != 0 {
framework.ExpectNoError(fmt.Errorf("%v", strings.Join(errList, "\n")))
......@@ -1063,7 +1060,7 @@ func pollReadWithTimeout(statefulPod statefulPodTester, statefulPodNumber int, k
}
func isInitialized(pod v1.Pod) bool {
initialized, ok := pod.Annotations[statefulset.StatefulSetInitAnnotation]
initialized, ok := pod.Annotations[apps.StatefulSetInitAnnotation]
if !ok {
return false
}
......@@ -1074,10 +1071,6 @@ func isInitialized(pod v1.Pod) bool {
return inited
}
func dec(i int64, exponent int) *inf.Dec {
return inf.NewDec(i, inf.Scale(-exponent))
}
func newPVC(name string) v1.PersistentVolumeClaim {
return v1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
......
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