forked from openshift/origin
-
Notifications
You must be signed in to change notification settings - Fork 0
/
lifecycle.go
314 lines (283 loc) · 11.7 KB
/
lifecycle.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
package support
import (
"fmt"
"time"
"github.com/golang/glog"
kapi "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
kerrors "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
kclient "github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
kutil "github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/wait"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
deployapi "github.com/openshift/origin/pkg/deploy/api"
deployutil "github.com/openshift/origin/pkg/deploy/util"
namer "github.com/openshift/origin/pkg/util/namer"
)
// HookExecutor executes a deployment lifecycle hook.
type HookExecutor struct {
// PodClient provides access to pods.
PodClient HookExecutorPodClient
}
// Execute executes hook in the context of deployment. The label is used to
// distinguish the kind of hook (e.g. pre, post).
func (e *HookExecutor) Execute(hook *deployapi.LifecycleHook, deployment *kapi.ReplicationController, label string) error {
var err error
switch {
case hook.ExecNewPod != nil:
err = e.executeExecNewPod(hook, deployment, label)
}
if err == nil {
return nil
}
// Retry failures are treated the same as Abort.
switch hook.FailurePolicy {
case deployapi.LifecycleHookFailurePolicyAbort, deployapi.LifecycleHookFailurePolicyRetry:
return fmt.Errorf("Hook failed, aborting: %s", err)
case deployapi.LifecycleHookFailurePolicyIgnore:
glog.Infof("Hook failed, ignoring: %s", err)
return nil
default:
return err
}
}
// executeExecNewPod executes a ExecNewPod hook by creating a new pod based on
// the hook parameters and deployment. The pod is then synchronously watched
// until the pod completes, and if the pod failed, an error is returned.
//
// The hook pod inherits the following from the container the hook refers to:
//
// * Environment (hook keys take precedence)
// * Working directory
// * Resources
func (e *HookExecutor) executeExecNewPod(hook *deployapi.LifecycleHook, deployment *kapi.ReplicationController, label string) error {
// Build a pod spec from the hook config and deployment
podSpec, err := makeHookPod(hook, deployment, label)
if err != nil {
return err
}
// Try to create the pod.
pod, err := e.PodClient.CreatePod(deployment.Namespace, podSpec)
if err != nil {
if !kerrors.IsAlreadyExists(err) {
return fmt.Errorf("couldn't create lifecycle pod for %s: %v", deployutil.LabelForDeployment(deployment), err)
}
} else {
glog.V(0).Infof("Created lifecycle pod %s for deployment %s", pod.Name, deployutil.LabelForDeployment(deployment))
}
stopChannel := make(chan struct{})
defer close(stopChannel)
nextPod := e.PodClient.PodWatch(pod.Namespace, pod.Name, pod.ResourceVersion, stopChannel)
glog.V(0).Infof("Waiting for hook pod %s/%s to complete", pod.Namespace, pod.Name)
for {
pod := nextPod()
switch pod.Status.Phase {
case kapi.PodSucceeded:
return nil
case kapi.PodFailed:
return fmt.Errorf(pod.Status.Message)
}
}
}
// makeHookPod makes a pod spec from a hook and deployment.
func makeHookPod(hook *deployapi.LifecycleHook, deployment *kapi.ReplicationController, label string) (*kapi.Pod, error) {
exec := hook.ExecNewPod
var baseContainer *kapi.Container
for _, container := range deployment.Spec.Template.Spec.Containers {
if container.Name == exec.ContainerName {
baseContainer = &container
break
}
}
if baseContainer == nil {
return nil, fmt.Errorf("no container named '%s' found in deployment template", exec.ContainerName)
}
// Build a merged environment; hook environment takes precedence over base
// container environment
envMap := map[string]string{}
mergedEnv := []kapi.EnvVar{}
for _, env := range baseContainer.Env {
envMap[env.Name] = env.Value
}
for _, env := range exec.Env {
envMap[env.Name] = env.Value
}
for k, v := range envMap {
mergedEnv = append(mergedEnv, kapi.EnvVar{Name: k, Value: v})
}
// Inherit resources from the base container
resources := kapi.ResourceRequirements{}
if err := kapi.Scheme.Convert(&baseContainer.Resources, &resources); err != nil {
return nil, fmt.Errorf("couldn't clone ResourceRequirements: %v", err)
}
// Assigning to a variable since its address is required
maxDeploymentDurationSeconds := deployapi.MaxDeploymentDurationSeconds
// Let the kubelet manage retries if requested
restartPolicy := kapi.RestartPolicyNever
if hook.FailurePolicy == deployapi.LifecycleHookFailurePolicyRetry {
restartPolicy = kapi.RestartPolicyOnFailure
}
pod := &kapi.Pod{
ObjectMeta: kapi.ObjectMeta{
Name: namer.GetPodName(deployment.Name, label),
Annotations: map[string]string{
deployapi.DeploymentAnnotation: deployment.Name,
},
Labels: map[string]string{
deployapi.DeployerPodForDeploymentLabel: deployment.Name,
},
},
Spec: kapi.PodSpec{
Containers: []kapi.Container{
{
Name: "lifecycle",
Image: baseContainer.Image,
Command: exec.Command,
WorkingDir: baseContainer.WorkingDir,
Env: mergedEnv,
Resources: resources,
},
},
ActiveDeadlineSeconds: &maxDeploymentDurationSeconds,
// Setting the node selector on the hook pod so that it is created
// on the same set of nodes as the deployment pods.
NodeSelector: deployment.Spec.Template.Spec.NodeSelector,
RestartPolicy: restartPolicy,
},
}
return pod, nil
}
// HookExecutorPodClient abstracts access to pods.
type HookExecutorPodClient interface {
CreatePod(namespace string, pod *kapi.Pod) (*kapi.Pod, error)
PodWatch(namespace, name, resourceVersion string, stopChannel chan struct{}) func() *kapi.Pod
}
// HookExecutorPodClientImpl is a pluggable HookExecutorPodClient.
type HookExecutorPodClientImpl struct {
CreatePodFunc func(namespace string, pod *kapi.Pod) (*kapi.Pod, error)
PodWatchFunc func(namespace, name, resourceVersion string, stopChannel chan struct{}) func() *kapi.Pod
}
func (i *HookExecutorPodClientImpl) CreatePod(namespace string, pod *kapi.Pod) (*kapi.Pod, error) {
return i.CreatePodFunc(namespace, pod)
}
func (i *HookExecutorPodClientImpl) PodWatch(namespace, name, resourceVersion string, stopChannel chan struct{}) func() *kapi.Pod {
return i.PodWatchFunc(namespace, name, resourceVersion, stopChannel)
}
// NewPodWatch creates a pod watching function which is backed by a
// FIFO/reflector pair. This avoids managing watches directly.
// A stop channel to close the watch's reflector is also returned.
// It is the caller's responsibility to defer closing the stop channel to prevent leaking resources.
func NewPodWatch(client kclient.Interface, namespace, name, resourceVersion string, stopChannel chan struct{}) func() *kapi.Pod {
fieldSelector, _ := fields.ParseSelector("metadata.name=" + name)
podLW := &deployutil.ListWatcherImpl{
ListFunc: func() (runtime.Object, error) {
return client.Pods(namespace).List(labels.Everything(), fieldSelector)
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return client.Pods(namespace).Watch(labels.Everything(), fieldSelector, resourceVersion)
},
}
queue := cache.NewFIFO(cache.MetaNamespaceKeyFunc)
cache.NewReflector(podLW, &kapi.Pod{}, queue, 1*time.Minute).RunUntil(stopChannel)
return func() *kapi.Pod {
obj := queue.Pop()
return obj.(*kapi.Pod)
}
}
// NewAcceptNewlyObservedReadyPods makes a new AcceptNewlyObservedReadyPods
// from a real client.
func NewAcceptNewlyObservedReadyPods(kclient kclient.Interface, timeout time.Duration, interval time.Duration) *AcceptNewlyObservedReadyPods {
return &AcceptNewlyObservedReadyPods{
timeout: timeout,
interval: interval,
acceptedPods: kutil.NewStringSet(),
getDeploymentPodStore: func(deployment *kapi.ReplicationController) (cache.Store, chan struct{}) {
selector := labels.Set(deployment.Spec.Selector).AsSelector()
store := cache.NewStore(cache.MetaNamespaceKeyFunc)
lw := &deployutil.ListWatcherImpl{
ListFunc: func() (runtime.Object, error) {
return kclient.Pods(deployment.Namespace).List(selector, fields.Everything())
},
WatchFunc: func(resourceVersion string) (watch.Interface, error) {
return kclient.Pods(deployment.Namespace).Watch(selector, fields.Everything(), resourceVersion)
},
}
stop := make(chan struct{})
cache.NewReflector(lw, &kapi.Pod{}, store, 10*time.Second).RunUntil(stop)
return store, stop
},
}
}
// AcceptNewlyObservedReadyPods is a kubectl.UpdateAcceptor which will accept
// a deployment if all the containers in all of the pods for the deployment
// are observed to be ready at least once.
//
// AcceptNewlyObservedReadyPods keeps track of the pods it has accepted for a
// deployment so that the acceptor can be reused across multiple batches of
// updates to a single controller. For example, if during the first acceptance
// call the deployment has 3 pods, the acceptor will validate those 3 pods. If
// the same acceptor instance is used again for the same deployment which now
// has 6 pods, only the latest 3 pods will be considered for acceptance. The
// status of the original 3 pods becomes irrelevant.
//
// Note that this struct is stateful and intended for use with a single
// deployment and should be discarded and recreated between deployments.
type AcceptNewlyObservedReadyPods struct {
// getDeploymentPodStore should return a Store containing all the pods for
// the named deployment, and a channel to stop whatever process is feeding
// the store.
getDeploymentPodStore func(deployment *kapi.ReplicationController) (cache.Store, chan struct{})
// timeout is how long to wait for pod readiness.
timeout time.Duration
// interval is how often to check for pod readiness
interval time.Duration
// acceptedPods keeps track of pods which have been previously accepted for
// a deployment.
acceptedPods kutil.StringSet
}
// Accept implements UpdateAcceptor.
func (c *AcceptNewlyObservedReadyPods) Accept(deployment *kapi.ReplicationController) error {
// Make a pod store to poll and ensure it gets cleaned up.
podStore, stopStore := c.getDeploymentPodStore(deployment)
defer close(stopStore)
// Start checking for pod updates.
glog.V(0).Infof("Waiting %.f seconds for pods owned by deployment %q to become ready (checking every %.f seconds; %d pods previously accepted)", c.timeout.Seconds(), deployutil.LabelForDeployment(deployment), c.interval.Seconds(), c.acceptedPods.Len())
err := wait.Poll(c.interval, c.timeout, func() (done bool, err error) {
// Check for pod readiness.
unready := kutil.NewStringSet()
for _, obj := range podStore.List() {
pod := obj.(*kapi.Pod)
// Skip previously accepted pods; we only want to verify newly observed
// and unaccepted pods.
if c.acceptedPods.Has(pod.Name) {
continue
}
if kapi.IsPodReady(pod) {
// If the pod is ready, track it as accepted.
c.acceptedPods.Insert(pod.Name)
} else {
// Otherwise, track it as unready.
unready.Insert(pod.Name)
}
}
// Check to see if we're done.
if unready.Len() == 0 {
glog.V(0).Infof("All pods ready for %s", deployutil.LabelForDeployment(deployment))
return true, nil
}
// Otherwise, try again later.
glog.V(4).Infof("Still waiting for %d pods to become ready for deployment %s", unready.Len(), deployutil.LabelForDeployment(deployment))
return false, nil
})
// Handle acceptance failure.
if err != nil {
if err == wait.ErrWaitTimeout {
return fmt.Errorf("pods for deployment %q took longer than %.f seconds to become ready", deployutil.LabelForDeployment(deployment), c.timeout.Seconds())
}
return fmt.Errorf("pod readiness check failed for deployment %q: %v", deployutil.LabelForDeployment(deployment), err)
}
return nil
}