Add init container loading to the kubelet

parent 6cc6d293
......@@ -92,6 +92,8 @@ func GetPodReadyCondition(status PodStatus) *PodCondition {
return condition
}
// GetPodCondition extracts the provided condition from the given status and returns that.
// Returns nil and -1 if the condition is not present, and the the index of the located condition.
func GetPodCondition(status *PodStatus, conditionType PodConditionType) (int, *PodCondition) {
for i, c := range status.Conditions {
if c.Type == conditionType {
......@@ -131,17 +133,6 @@ func UpdatePodCondition(status *PodStatus, condition *PodCondition) bool {
}
}
// GetPodCondition extracts the provided condition from the given status and returns that.
// Returns nil if the condition is not present.
func GetPodCondition(status PodStatus, t PodConditionType) *PodCondition {
for i, c := range status.Conditions {
if c.Type == t {
return &status.Conditions[i]
}
}
return nil
}
// IsNodeReady returns true if a node is ready; false otherwise.
func IsNodeReady(node *Node) bool {
for _, c := range node.Status.Conditions {
......
......@@ -57,5 +57,15 @@ func fieldPath(pod *api.Pod, container *api.Container) (string, error) {
}
}
}
for i := range pod.Spec.InitContainers {
here := &pod.Spec.InitContainers[i]
if here.Name == container.Name {
if here.Name == "" {
return fmt.Sprintf("spec.initContainers[%d]", i), nil
} else {
return fmt.Sprintf("spec.initContainers{%s}", here.Name), nil
}
}
}
return "", fmt.Errorf("container %#v not found in pod %#v", container, pod)
}
......@@ -50,9 +50,10 @@ var (
)
var (
ErrRunContainer = errors.New("RunContainerError")
ErrKillContainer = errors.New("KillContainerError")
ErrVerifyNonRoot = errors.New("VerifyNonRootError")
ErrRunContainer = errors.New("RunContainerError")
ErrKillContainer = errors.New("KillContainerError")
ErrVerifyNonRoot = errors.New("VerifyNonRootError")
ErrRunInitContainer = errors.New("RunInitContainerError")
)
var (
......@@ -69,6 +70,7 @@ const (
KillContainer SyncAction = "KillContainer"
SetupNetwork SyncAction = "SetupNetwork"
TeardownNetwork SyncAction = "TeardownNetwork"
InitContainer SyncAction = "InitContainer"
)
// SyncResult is the result of sync action.
......
......@@ -1761,6 +1761,7 @@ func (kl *Kubelet) syncPod(o syncPodOptions) error {
!firstSeenTime.IsZero() {
metrics.PodStartLatency.Observe(metrics.SinceInMicroseconds(firstSeenTime))
}
// Update status in the status manager
kl.statusManager.SetPodStatus(pod, apiPodStatus)
......@@ -2347,6 +2348,10 @@ func hasHostPortConflicts(pods []*api.Pod) bool {
glog.Errorf("Pod %q: HostPort is already allocated, ignoring: %v", format.Pod(pod), errs)
return true
}
if errs := validation.AccumulateUniqueHostPorts(pod.Spec.InitContainers, &ports, field.NewPath("spec", "initContainers")); len(errs) > 0 {
glog.Errorf("Pod %q: HostPort is already allocated, ignoring: %v", format.Pod(pod), errs)
return true
}
}
return false
}
......@@ -3310,12 +3315,46 @@ func (kl *Kubelet) tryUpdateNodeStatus() error {
// This func is exported to simplify integration with 3rd party kubelet
// integrations like kubernetes-mesos.
func GetPhase(spec *api.PodSpec, info []api.ContainerStatus) api.PodPhase {
initialized := 0
pendingInitialization := 0
failedInitialization := 0
for _, container := range spec.InitContainers {
containerStatus, ok := api.GetContainerStatus(info, container.Name)
if !ok {
pendingInitialization++
continue
}
switch {
case containerStatus.State.Running != nil:
pendingInitialization++
case containerStatus.State.Terminated != nil:
if containerStatus.State.Terminated.ExitCode == 0 {
initialized++
} else {
failedInitialization++
}
case containerStatus.State.Waiting != nil:
if containerStatus.LastTerminationState.Terminated != nil {
if containerStatus.LastTerminationState.Terminated.ExitCode == 0 {
initialized++
} else {
failedInitialization++
}
} else {
pendingInitialization++
}
default:
pendingInitialization++
}
}
unknown := 0
running := 0
waiting := 0
stopped := 0
failed := 0
succeeded := 0
unknown := 0
for _, container := range spec.Containers {
containerStatus, ok := api.GetContainerStatus(info, container.Name)
if !ok {
......@@ -3344,7 +3383,13 @@ func GetPhase(spec *api.PodSpec, info []api.ContainerStatus) api.PodPhase {
}
}
if failedInitialization > 0 && spec.RestartPolicy == api.RestartPolicyNever {
return api.PodFailed
}
switch {
case pendingInitialization > 0:
fallthrough
case waiting > 0:
glog.V(5).Infof("pod waiting > 0, pending")
// One or more containers has not been started
......@@ -3409,8 +3454,10 @@ func (kl *Kubelet) generateAPIPodStatus(pod *api.Pod, podStatus *kubecontainer.P
// Assume info is ready to process
spec := &pod.Spec
s.Phase = GetPhase(spec, s.ContainerStatuses)
allStatus := append(append([]api.ContainerStatus{}, s.ContainerStatuses...), s.InitContainerStatuses...)
s.Phase = GetPhase(spec, allStatus)
kl.probeManager.UpdatePodStatus(pod.UID, s)
s.Conditions = append(s.Conditions, status.GeneratePodInitializedCondition(spec, s.InitContainerStatuses, s.Phase))
s.Conditions = append(s.Conditions, status.GeneratePodReadyCondition(spec, s.ContainerStatuses, s.Phase))
// s (the PodStatus we are creating) will not have a PodScheduled condition yet, because converStatusToAPIStatus()
// does not create one. If the existing PodStatus has a PodScheduled condition, then copy it into s and make sure
......@@ -3443,9 +3490,27 @@ func (kl *Kubelet) generateAPIPodStatus(pod *api.Pod, podStatus *kubecontainer.P
// alter the kubelet state at all.
func (kl *Kubelet) convertStatusToAPIStatus(pod *api.Pod, podStatus *kubecontainer.PodStatus) *api.PodStatus {
var apiPodStatus api.PodStatus
uid := pod.UID
apiPodStatus.PodIP = podStatus.IP
apiPodStatus.ContainerStatuses = kl.convertToAPIContainerStatuses(
pod, podStatus,
pod.Status.ContainerStatuses,
pod.Spec.Containers,
len(pod.Spec.InitContainers) > 0,
false,
)
apiPodStatus.InitContainerStatuses = kl.convertToAPIContainerStatuses(
pod, podStatus,
pod.Status.InitContainerStatuses,
pod.Spec.InitContainers,
len(pod.Spec.InitContainers) > 0,
true,
)
return &apiPodStatus
}
func (kl *Kubelet) convertToAPIContainerStatuses(pod *api.Pod, podStatus *kubecontainer.PodStatus, previousStatus []api.ContainerStatus, containers []api.Container, hasInitContainers, isInitContainer bool) []api.ContainerStatus {
convertContainerStatus := func(cs *kubecontainer.ContainerStatus) *api.ContainerStatus {
cid := cs.ID.String()
status := &api.ContainerStatus{
......@@ -3474,15 +3539,19 @@ func (kl *Kubelet) convertStatusToAPIStatus(pod *api.Pod, podStatus *kubecontain
}
// Fetch old containers statuses from old pod status.
oldStatuses := make(map[string]api.ContainerStatus, len(pod.Spec.Containers))
for _, status := range pod.Status.ContainerStatuses {
oldStatuses := make(map[string]api.ContainerStatus, len(containers))
for _, status := range previousStatus {
oldStatuses[status.Name] = status
}
// Set all container statuses to default waiting state
statuses := make(map[string]*api.ContainerStatus, len(pod.Spec.Containers))
statuses := make(map[string]*api.ContainerStatus, len(containers))
defaultWaitingState := api.ContainerState{Waiting: &api.ContainerStateWaiting{Reason: "ContainerCreating"}}
for _, container := range pod.Spec.Containers {
if hasInitContainers {
defaultWaitingState = api.ContainerState{Waiting: &api.ContainerStateWaiting{Reason: "PodInitializing"}}
}
for _, container := range containers {
status := &api.ContainerStatus{
Name: container.Name,
Image: container.Image,
......@@ -3498,7 +3567,6 @@ func (kl *Kubelet) convertStatusToAPIStatus(pod *api.Pod, podStatus *kubecontain
// Make the latest container status comes first.
sort.Sort(sort.Reverse(kubecontainer.SortContainerStatusesByCreationTime(podStatus.ContainerStatuses)))
// Set container statuses according to the statuses seen in pod status
containerSeen := map[string]int{}
for _, cStatus := range podStatus.ContainerStatuses {
......@@ -3520,13 +3588,13 @@ func (kl *Kubelet) convertStatusToAPIStatus(pod *api.Pod, podStatus *kubecontain
}
// Handle the containers failed to be started, which should be in Waiting state.
for _, container := range pod.Spec.Containers {
for _, container := range containers {
// If a container should be restarted in next syncpod, it is *Waiting*.
if !kubecontainer.ShouldContainerBeRestarted(&container, pod, podStatus) {
continue
}
status := statuses[container.Name]
reason, message, ok := kl.reasonCache.Get(uid, container.Name)
reason, message, ok := kl.reasonCache.Get(pod.UID, container.Name)
if !ok {
// In fact, we could also apply Waiting state here, but it is less informative,
// and the container will be restarted soon, so we prefer the original state here.
......@@ -3548,15 +3616,15 @@ func (kl *Kubelet) convertStatusToAPIStatus(pod *api.Pod, podStatus *kubecontain
statuses[container.Name] = status
}
apiPodStatus.ContainerStatuses = make([]api.ContainerStatus, 0)
var containerStatuses []api.ContainerStatus
for _, status := range statuses {
apiPodStatus.ContainerStatuses = append(apiPodStatus.ContainerStatuses, *status)
containerStatuses = append(containerStatuses, *status)
}
// Sort the container statuses since clients of this interface expect the list
// of containers in a pod has a deterministic order.
sort.Sort(kubetypes.SortedContainerStatuses(apiPodStatus.ContainerStatuses))
return &apiPodStatus
sort.Sort(kubetypes.SortedContainerStatuses(containerStatuses))
return containerStatuses
}
// Returns logs of current machine.
......
......@@ -207,6 +207,15 @@ func (m *manager) UpdatePodStatus(podUID types.UID, podStatus *api.PodStatus) {
}
podStatus.ContainerStatuses[i].Ready = ready
}
// init containers are ready if they have exited with success or if a readiness probe has
// succeeded.
for i, c := range podStatus.InitContainerStatuses {
var ready bool
if c.State.Terminated != nil && c.State.Terminated.ExitCode == 0 {
ready = true
}
podStatus.InitContainerStatuses[i].Ready = ready
}
}
func (m *manager) getWorker(podUID types.UID, containerName string, probeType probeType) (*worker, bool) {
......
......@@ -466,6 +466,13 @@ func (s *Server) getContainerLogs(request *restful.Request, response *restful.Re
}
}
if !containerExists {
for _, container := range pod.Spec.InitContainers {
if container.Name == containerName {
containerExists = true
}
}
}
if !containerExists {
response.WriteError(http.StatusNotFound, fmt.Errorf("container %q not found in pod %q\n", containerName, podID))
return
}
......
......@@ -77,3 +77,58 @@ func GeneratePodReadyCondition(spec *api.PodSpec, containerStatuses []api.Contai
Status: api.ConditionTrue,
}
}
// GeneratePodInitializedCondition returns initialized condition if all init containers in a pod are ready, else it
// returns an uninitialized condition.
func GeneratePodInitializedCondition(spec *api.PodSpec, containerStatuses []api.ContainerStatus, podPhase api.PodPhase) api.PodCondition {
// Find if all containers are ready or not.
if containerStatuses == nil && len(spec.InitContainers) > 0 {
return api.PodCondition{
Type: api.PodInitialized,
Status: api.ConditionFalse,
Reason: "UnknownContainerStatuses",
}
}
unknownContainers := []string{}
unreadyContainers := []string{}
for _, container := range spec.InitContainers {
if containerStatus, ok := api.GetContainerStatus(containerStatuses, container.Name); ok {
if !containerStatus.Ready {
unreadyContainers = append(unreadyContainers, container.Name)
}
} else {
unknownContainers = append(unknownContainers, container.Name)
}
}
// If all init containers are known and succeeded, just return PodCompleted.
if podPhase == api.PodSucceeded && len(unknownContainers) == 0 {
return api.PodCondition{
Type: api.PodInitialized,
Status: api.ConditionTrue,
Reason: "PodCompleted",
}
}
unreadyMessages := []string{}
if len(unknownContainers) > 0 {
unreadyMessages = append(unreadyMessages, fmt.Sprintf("containers with unknown status: %s", unknownContainers))
}
if len(unreadyContainers) > 0 {
unreadyMessages = append(unreadyMessages, fmt.Sprintf("containers with incomplete status: %s", unreadyContainers))
}
unreadyMessage := strings.Join(unreadyMessages, ", ")
if unreadyMessage != "" {
return api.PodCondition{
Type: api.PodInitialized,
Status: api.ConditionFalse,
Reason: "ContainersNotInitialized",
Message: unreadyMessage,
}
}
return api.PodCondition{
Type: api.PodInitialized,
Status: api.ConditionTrue,
}
}
......@@ -172,20 +172,14 @@ func (m *manager) SetContainerReadiness(podUID types.UID, containerID kubecontai
}
// Find the container to update.
containerIndex := -1
for i, c := range oldStatus.status.ContainerStatuses {
if c.ContainerID == containerID.String() {
containerIndex = i
break
}
}
if containerIndex == -1 {
containerStatus, _, ok := findContainerStatus(&oldStatus.status, containerID.String())
if !ok {
glog.Warningf("Container readiness changed for unknown container: %q - %q",
format.Pod(pod), containerID.String())
return
}
if oldStatus.status.ContainerStatuses[containerIndex].Ready == ready {
if containerStatus.Ready == ready {
glog.V(4).Infof("Container readiness unchanged (%v): %q - %q", ready,
format.Pod(pod), containerID.String())
return
......@@ -196,7 +190,8 @@ func (m *manager) SetContainerReadiness(podUID types.UID, containerID kubecontai
if err != nil {
return
}
status.ContainerStatuses[containerIndex].Ready = ready
containerStatus, _, _ = findContainerStatus(&status, containerID.String())
containerStatus.Ready = ready
// Update pod condition.
readyConditionIndex := -1
......@@ -217,6 +212,31 @@ func (m *manager) SetContainerReadiness(podUID types.UID, containerID kubecontai
m.updateStatusInternal(pod, status, false)
}
func findContainerStatus(status *api.PodStatus, containerID string) (containerStatus *api.ContainerStatus, init bool, ok bool) {
// Find the container to update.
containerIndex := -1
for i, c := range status.ContainerStatuses {
if c.ContainerID == containerID {
containerIndex = i
break
}
}
if containerIndex != -1 {
return &status.ContainerStatuses[containerIndex], false, true
}
for i, c := range status.InitContainerStatuses {
if c.ContainerID == containerID {
containerIndex = i
break
}
}
if containerIndex != -1 {
return &status.InitContainerStatuses[containerIndex], true, true
}
return nil, false, false
}
func (m *manager) TerminatePod(pod *api.Pod) {
m.podStatusesLock.Lock()
defer m.podStatusesLock.Unlock()
......@@ -233,6 +253,11 @@ func (m *manager) TerminatePod(pod *api.Pod) {
Terminated: &api.ContainerStateTerminated{},
}
}
for i := range status.InitContainerStatuses {
status.InitContainerStatuses[i].State = api.ContainerState{
Terminated: &api.ContainerStateTerminated{},
}
}
m.updateStatusInternal(pod, pod.Status, true)
}
......@@ -251,16 +276,27 @@ func (m *manager) updateStatusInternal(pod *api.Pod, status api.PodStatus, force
}
// Set ReadyCondition.LastTransitionTime.
if readyCondition := api.GetPodReadyCondition(status); readyCondition != nil {
if _, readyCondition := api.GetPodCondition(&status, api.PodReady); readyCondition != nil {
// Need to set LastTransitionTime.
lastTransitionTime := unversioned.Now()
oldReadyCondition := api.GetPodReadyCondition(oldStatus)
_, oldReadyCondition := api.GetPodCondition(&oldStatus, api.PodReady)
if oldReadyCondition != nil && readyCondition.Status == oldReadyCondition.Status {
lastTransitionTime = oldReadyCondition.LastTransitionTime
}
readyCondition.LastTransitionTime = lastTransitionTime
}
// Set InitializedCondition.LastTransitionTime.
if _, initCondition := api.GetPodCondition(&status, api.PodInitialized); initCondition != nil {
// Need to set LastTransitionTime.
lastTransitionTime := unversioned.Now()
_, oldInitCondition := api.GetPodCondition(&oldStatus, api.PodInitialized)
if oldInitCondition != nil && initCondition.Status == oldInitCondition.Status {
lastTransitionTime = oldInitCondition.LastTransitionTime
}
initCondition.LastTransitionTime = lastTransitionTime
}
// ensure that the start time does not change across updates.
if oldStatus.StartTime != nil && !oldStatus.StartTime.IsZero() {
status.StartTime = oldStatus.StartTime
......@@ -490,6 +526,8 @@ func normalizeStatus(status *api.PodStatus) *api.PodStatus {
normalizeTimeStamp(&condition.LastProbeTime)
normalizeTimeStamp(&condition.LastTransitionTime)
}
// update container statuses
for i := range status.ContainerStatuses {
cstatus := &status.ContainerStatuses[i]
normalizeContainerState(&cstatus.State)
......@@ -497,6 +535,15 @@ func normalizeStatus(status *api.PodStatus) *api.PodStatus {
}
// Sort the container statuses, so that the order won't affect the result of comparison
sort.Sort(kubetypes.SortedContainerStatuses(status.ContainerStatuses))
// update init container statuses
for i := range status.InitContainerStatuses {
cstatus := &status.InitContainerStatuses[i]
normalizeContainerState(&cstatus.State)
normalizeContainerState(&cstatus.LastTerminationState)
}
// Sort the container statuses, so that the order won't affect the result of comparison
sort.Sort(kubetypes.SortedContainerStatuses(status.InitContainerStatuses))
return status
}
......
......@@ -63,6 +63,11 @@ func canRunPod(pod *api.Pod) error {
return fmt.Errorf("pod with UID %q specified privileged container, but is disallowed", pod.UID)
}
}
for _, container := range pod.Spec.InitContainers {
if securitycontext.HasPrivilegedRequest(&container) {
return fmt.Errorf("pod with UID %q specified privileged container, but is disallowed", pod.UID)
}
}
}
return nil
}
......
......@@ -879,6 +879,109 @@ func deleteNS(c *client.Client, namespace string, timeout time.Duration) error {
return nil
}
func ContainerInitInvariant(older, newer runtime.Object) error {
oldPod := older.(*api.Pod)
newPod := newer.(*api.Pod)
if len(oldPod.Spec.InitContainers) == 0 {
return nil
}
if len(oldPod.Spec.InitContainers) != len(newPod.Spec.InitContainers) {
return fmt.Errorf("init container list changed")
}
if oldPod.UID != newPod.UID {
return fmt.Errorf("two different pods exist in the condition: %s vs %s", oldPod.UID, newPod.UID)
}
if err := initContainersInvariants(oldPod); err != nil {
return err
}
if err := initContainersInvariants(newPod); err != nil {
return err
}
oldInit, _, _ := podInitialized(oldPod)
newInit, _, _ := podInitialized(newPod)
if oldInit && !newInit {
// TODO: we may in the future enable resetting PodInitialized = false if the kubelet needs to restart it
// from scratch
return fmt.Errorf("pod cannot be initialized and then regress to not being initialized")
}
return nil
}
func podInitialized(pod *api.Pod) (ok bool, failed bool, err error) {
allInit := true
initFailed := false
for _, s := range pod.Status.InitContainerStatuses {
switch {
case initFailed && s.State.Waiting == nil:
return allInit, initFailed, fmt.Errorf("container %s is after a failed container but isn't waiting", s.Name)
case allInit && s.State.Waiting == nil:
return allInit, initFailed, fmt.Errorf("container %s is after an initializing container but isn't waiting", s.Name)
case s.State.Terminated == nil:
allInit = false
case s.State.Terminated.ExitCode != 0:
allInit = false
initFailed = true
case !s.Ready:
return allInit, initFailed, fmt.Errorf("container %s initialized but isn't marked as ready", s.Name)
}
}
return allInit, initFailed, nil
}
func initContainersInvariants(pod *api.Pod) error {
allInit, initFailed, err := podInitialized(pod)
if err != nil {
return err
}
if !allInit || initFailed {
for _, s := range pod.Status.ContainerStatuses {
if s.State.Waiting == nil || s.RestartCount != 0 {
return fmt.Errorf("container %s is not waiting but initialization not complete", s.Name)
}
if s.State.Waiting.Reason != "PodInitializing" {
return fmt.Errorf("container %s should have reason PodInitializing: %s", s.Name, s.State.Waiting.Reason)
}
}
}
_, c := api.GetPodCondition(&pod.Status, api.PodInitialized)
if c == nil {
return fmt.Errorf("pod does not have initialized condition")
}
if c.LastTransitionTime.IsZero() {
return fmt.Errorf("PodInitialized condition should always have a transition time")
}
switch {
case c.Status == api.ConditionUnknown:
return fmt.Errorf("PodInitialized condition should never be Unknown")
case c.Status == api.ConditionTrue && (initFailed || !allInit):
return fmt.Errorf("PodInitialized condition was True but all not all containers initialized")
case c.Status == api.ConditionFalse && (!initFailed && allInit):
return fmt.Errorf("PodInitialized condition was False but all containers initialized")
}
return nil
}
type InvariantFunc func(older, newer runtime.Object) error
func CheckInvariants(events []watch.Event, fns ...InvariantFunc) error {
errs := sets.NewString()
for i := range events {
j := i + 1
if j >= len(events) {
continue
}
for _, fn := range fns {
if err := fn(events[i].Object, events[j].Object); err != nil {
errs.Insert(err.Error())
}
}
}
if errs.Len() > 0 {
return fmt.Errorf("invariants violated:\n* %s", strings.Join(errs.List(), "\n* "))
}
return nil
}
// Waits default amount of time (PodStartTimeout) for the specified pod to become running.
// Returns an error if timeout occurs first, or pod goes in to failed state.
func WaitForPodRunningInNamespace(c *client.Client, podName string, namespace string) error {
......@@ -2218,7 +2321,11 @@ func DumpNodeDebugInfo(c *client.Client, nodeNames []string) {
continue
}
for _, p := range podList.Items {
Logf("%v started at %v (%d container statuses recorded)", p.Name, p.Status.StartTime, len(p.Status.ContainerStatuses))
Logf("%v started at %v (%d+%d container statuses recorded)", p.Name, p.Status.StartTime, len(p.Status.InitContainerStatuses), len(p.Status.ContainerStatuses))
for _, c := range p.Status.InitContainerStatuses {
Logf("\tInit container %v ready: %v, restart count %v",
c.Name, c.Ready, c.RestartCount)
}
for _, c := range p.Status.ContainerStatuses {
Logf("\tContainer %v ready: %v, restart count %v",
c.Name, c.Ready, c.RestartCount)
......
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