forked from openshift/origin
-
Notifications
You must be signed in to change notification settings - Fork 0
/
factory.go
177 lines (164 loc) · 6.32 KB
/
factory.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
package deployment
import (
"fmt"
"time"
kapi "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/client/cache"
"k8s.io/kubernetes/pkg/client/record"
kclient "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/util/flowcontrol"
utilruntime "k8s.io/kubernetes/pkg/util/runtime"
"k8s.io/kubernetes/pkg/util/sets"
"k8s.io/kubernetes/pkg/watch"
controller "github.com/openshift/origin/pkg/controller"
deployapi "github.com/openshift/origin/pkg/deploy/api"
deployutil "github.com/openshift/origin/pkg/deploy/util"
)
// DeploymentControllerFactory can create a DeploymentController that creates
// deployer pods in a configurable way.
type DeploymentControllerFactory struct {
// KubeClient is a Kubernetes client.
KubeClient kclient.Interface
// Codec is used for encoding/decoding.
Codec runtime.Codec
// ServiceAccount is the service account name to run deployer pods as
ServiceAccount string
// Environment is a set of environment which should be injected into all deployer pod containers.
Environment []kapi.EnvVar
// DeployerImage specifies which Docker image can support the default strategies.
DeployerImage string
}
// Create creates a DeploymentController.
func (factory *DeploymentControllerFactory) Create() controller.RunnableController {
deploymentLW := &cache.ListWatch{
// TODO: Investigate specifying annotation field selectors to fetch only 'deployments'
// Currently field selectors are not supported for replication controllers
ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {
return factory.KubeClient.ReplicationControllers(kapi.NamespaceAll).List(options)
},
WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {
return factory.KubeClient.ReplicationControllers(kapi.NamespaceAll).Watch(options)
},
}
deploymentQueue := cache.NewFIFO(cache.MetaNamespaceKeyFunc)
cache.NewReflector(deploymentLW, &kapi.ReplicationController{}, deploymentQueue, 2*time.Minute).Run()
eventBroadcaster := record.NewBroadcaster()
eventBroadcaster.StartRecordingToSink(factory.KubeClient.Events(""))
eventRecorder := eventBroadcaster.NewRecorder(kapi.EventSource{Component: "deployments-controller"})
c := &DeploymentController{
serviceAccount: factory.ServiceAccount,
deploymentClient: &deploymentClientImpl{
getDeploymentFunc: func(namespace, name string) (*kapi.ReplicationController, error) {
return factory.KubeClient.ReplicationControllers(namespace).Get(name)
},
updateDeploymentFunc: func(namespace string, deployment *kapi.ReplicationController) (*kapi.ReplicationController, error) {
return factory.KubeClient.ReplicationControllers(namespace).Update(deployment)
},
},
podClient: &podClientImpl{
getPodFunc: func(namespace, name string) (*kapi.Pod, error) {
return factory.KubeClient.Pods(namespace).Get(name)
},
createPodFunc: func(namespace string, pod *kapi.Pod) (*kapi.Pod, error) {
return factory.KubeClient.Pods(namespace).Create(pod)
},
deletePodFunc: func(namespace, name string) error {
return factory.KubeClient.Pods(namespace).Delete(name, nil)
},
updatePodFunc: func(namespace string, pod *kapi.Pod) (*kapi.Pod, error) {
return factory.KubeClient.Pods(namespace).Update(pod)
},
// Find deployer pods using the label they should all have which
// correlates them to the named deployment.
getDeployerPodsForFunc: func(namespace, name string) ([]kapi.Pod, error) {
labelSel, err := labels.Parse(fmt.Sprintf("%s=%s", deployapi.DeployerPodForDeploymentLabel, name))
if err != nil {
return []kapi.Pod{}, err
}
options := kapi.ListOptions{LabelSelector: labelSel}
pods, err := factory.KubeClient.Pods(namespace).List(options)
if err != nil {
return []kapi.Pod{}, err
}
return pods.Items, nil
},
},
makeContainer: func(strategy *deployapi.DeploymentStrategy) *kapi.Container {
return factory.makeContainer(strategy)
},
decodeConfig: func(deployment *kapi.ReplicationController) (*deployapi.DeploymentConfig, error) {
return deployutil.DecodeDeploymentConfig(deployment, factory.Codec)
},
recorder: eventRecorder,
}
return &controller.RetryController{
Queue: deploymentQueue,
RetryManager: controller.NewQueueRetryManager(
deploymentQueue,
cache.MetaNamespaceKeyFunc,
func(obj interface{}, err error, retries controller.Retry) bool {
if _, isFatal := err.(fatalError); isFatal {
utilruntime.HandleError(err)
return false
}
if retries.Count > 1 {
_, isActionableErr := err.(actionableError)
deployment, noReasonToPanic := obj.(*kapi.ReplicationController)
if isActionableErr && noReasonToPanic {
c.emitDeploymentEvent(deployment, kapi.EventTypeWarning, "FailedRetry", fmt.Sprintf("About to stop retrying %s: %v", deployment.Name, err))
}
return false
}
return true
},
flowcontrol.NewTokenBucketRateLimiter(1, 10),
),
Handle: func(obj interface{}) error {
deployment := obj.(*kapi.ReplicationController)
return c.Handle(deployment)
},
}
}
// makeContainer creates containers in the following way:
//
// 1. For the Recreate and Rolling strategies, strategy, use the factory's
// DeployerImage as the container image, and the factory's Environment
// as the container environment.
// 2. For all Custom strategies, or if the CustomParams field is set, use
// the strategy's image for the container image, and use the combination
// of the factory's Environment and the strategy's environment as the
// container environment.
//
func (factory *DeploymentControllerFactory) makeContainer(strategy *deployapi.DeploymentStrategy) *kapi.Container {
image := factory.DeployerImage
var environment []kapi.EnvVar
var command []string
set := sets.NewString()
// Use user-defined values from the strategy input.
if p := strategy.CustomParams; p != nil {
if len(p.Image) > 0 {
image = p.Image
}
if len(p.Command) > 0 {
command = p.Command
}
for _, env := range strategy.CustomParams.Environment {
set.Insert(env.Name)
environment = append(environment, env)
}
}
// Set default environment values
for _, env := range factory.Environment {
if set.Has(env.Name) {
continue
}
environment = append(environment, env)
}
return &kapi.Container{
Image: image,
Command: command,
Env: environment,
}
}