forked from kubernetes/kubernetes
-
Notifications
You must be signed in to change notification settings - Fork 0
/
federation-daemonset.go
192 lines (169 loc) · 8.28 KB
/
federation-daemonset.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
/*
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 e2e
import (
"fmt"
"time"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5"
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
"k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
"k8s.io/kubernetes/pkg/util/wait"
"k8s.io/kubernetes/test/e2e/framework"
)
const (
FederatedDaemonSetName = "federated-daemonset"
FederatedDaemonSetTimeout = 60 * time.Second
FederatedDaemonSetMaxRetries = 3
)
// Create/delete daemonset api objects
var _ = framework.KubeDescribe("Federation daemonsets [Feature:Federation12]", func() {
var clusters map[string]*cluster // All clusters, keyed by cluster name
f := framework.NewDefaultFederatedFramework("federated-daemonset")
Describe("DaemonSet objects", func() {
BeforeEach(func() {
framework.SkipUnlessFederated(f.ClientSet)
clusters = map[string]*cluster{}
registerClusters(clusters, UserAgentName, "", f)
})
AfterEach(func() {
framework.SkipUnlessFederated(f.ClientSet)
unregisterClusters(clusters, f)
})
It("should be created and deleted successfully", func() {
framework.SkipUnlessFederated(f.ClientSet)
nsName := f.FederationNamespace.Name
daemonset := createDaemonSetOrFail(f.FederationClientset_1_5, nsName)
defer func() { // Cleanup
By(fmt.Sprintf("Deleting daemonset %q in namespace %q", daemonset.Name, nsName))
err := f.FederationClientset_1_5.Extensions().DaemonSets(nsName).Delete(daemonset.Name, &v1.DeleteOptions{})
framework.ExpectNoError(err, "Error deleting daemonset %q in namespace %q", daemonset.Name, nsName)
}()
// wait for daemonset shards being created
waitForDaemonSetShardsOrFail(nsName, daemonset, clusters)
daemonset = updateDaemonSetOrFail(f.FederationClientset_1_5, nsName)
waitForDaemonSetShardsUpdatedOrFail(nsName, daemonset, clusters)
})
})
})
func createDaemonSetOrFail(clientset *fedclientset.Clientset, namespace string) *v1beta1.DaemonSet {
if clientset == nil || len(namespace) == 0 {
Fail(fmt.Sprintf("Internal error: invalid parameters passed to createDaemonSetOrFail: clientset: %v, namespace: %v", clientset, namespace))
}
daemonset := &v1beta1.DaemonSet{
ObjectMeta: v1.ObjectMeta{
Name: FederatedDaemonSetName,
Namespace: namespace,
},
Spec: v1beta1.DaemonSetSpec{
Template: v1.PodTemplateSpec{
ObjectMeta: v1.ObjectMeta{
Labels: map[string]string{"aaa": "bbb"},
},
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "container1",
Image: "gcr.io/google_containers/serve_hostname:v1.4",
Ports: []v1.ContainerPort{{ContainerPort: 9376}},
},
},
},
},
},
}
By(fmt.Sprintf("Creating daemonset %q in namespace %q", daemonset.Name, namespace))
_, err := clientset.Extensions().DaemonSets(namespace).Create(daemonset)
framework.ExpectNoError(err, "Failed to create daemonset %s", daemonset.Name)
By(fmt.Sprintf("Successfully created federated daemonset %q in namespace %q", FederatedDaemonSetName, namespace))
return daemonset
}
func updateDaemonSetOrFail(clientset *fedclientset.Clientset, namespace string) *v1beta1.DaemonSet {
if clientset == nil || len(namespace) == 0 {
Fail(fmt.Sprintf("Internal error: invalid parameters passed to updateDaemonSetOrFail: clientset: %v, namespace: %v", clientset, namespace))
}
var newDaemonSet *v1beta1.DaemonSet
for retryCount := 0; retryCount < FederatedDaemonSetMaxRetries; retryCount++ {
daemonset, err := clientset.Extensions().DaemonSets(namespace).Get(FederatedDaemonSetName)
if err != nil {
framework.Failf("failed to get daemonset %q: %v", FederatedDaemonSetName, err)
}
// Update one of the data in the daemonset.
daemonset.Annotations = map[string]string{"ccc": "ddd"}
newDaemonSet, err = clientset.Extensions().DaemonSets(namespace).Update(daemonset)
if err == nil {
return newDaemonSet
}
if !errors.IsConflict(err) && !errors.IsServerTimeout(err) {
framework.Failf("failed to update daemonset %q: %v", FederatedDaemonSetName, err)
}
}
framework.Failf("too many retries updating daemonset %q", FederatedDaemonSetName)
return newDaemonSet
}
func waitForDaemonSetShardsOrFail(namespace string, daemonset *v1beta1.DaemonSet, clusters map[string]*cluster) {
framework.Logf("Waiting for daemonset %q in %d clusters", daemonset.Name, len(clusters))
for _, c := range clusters {
waitForDaemonSetOrFail(c.Clientset, namespace, daemonset, true, FederatedDaemonSetTimeout)
}
}
func waitForDaemonSetOrFail(clientset *kubeclientset.Clientset, namespace string, daemonset *v1beta1.DaemonSet, present bool, timeout time.Duration) {
By(fmt.Sprintf("Fetching a federated daemonset shard of daemonset %q in namespace %q from cluster", daemonset.Name, namespace))
var clusterDaemonSet *v1beta1.DaemonSet
err := wait.PollImmediate(framework.Poll, timeout, func() (bool, error) {
clusterDaemonSet, err := clientset.Extensions().DaemonSets(namespace).Get(daemonset.Name)
if (!present) && errors.IsNotFound(err) { // We want it gone, and it's gone.
By(fmt.Sprintf("Success: shard of federated daemonset %q in namespace %q in cluster is absent", daemonset.Name, namespace))
return true, nil // Success
}
if present && err == nil { // We want it present, and the Get succeeded, so we're all good.
By(fmt.Sprintf("Success: shard of federated daemonset %q in namespace %q in cluster is present", daemonset.Name, namespace))
return true, nil // Success
}
By(fmt.Sprintf("DaemonSet %q in namespace %q in cluster. Found: %v, waiting for Found: %v, trying again in %s (err=%v)", daemonset.Name, namespace, clusterDaemonSet != nil && err == nil, present, framework.Poll, err))
return false, nil
})
framework.ExpectNoError(err, "Failed to verify daemonset %q in namespace %q in cluster: Present=%v", daemonset.Name, namespace, present)
if present && clusterDaemonSet != nil {
Expect(util.ObjectMetaAndSpecEquivalent(clusterDaemonSet, daemonset))
}
}
func waitForDaemonSetShardsUpdatedOrFail(namespace string, daemonset *v1beta1.DaemonSet, clusters map[string]*cluster) {
framework.Logf("Waiting for daemonset %q in %d clusters", daemonset.Name, len(clusters))
for _, c := range clusters {
waitForDaemonSetUpdateOrFail(c.Clientset, namespace, daemonset, FederatedDaemonSetTimeout)
}
}
func waitForDaemonSetUpdateOrFail(clientset *kubeclientset.Clientset, namespace string, daemonset *v1beta1.DaemonSet, timeout time.Duration) {
By(fmt.Sprintf("Fetching a federated daemonset shard of daemonset %q in namespace %q from cluster", daemonset.Name, namespace))
err := wait.PollImmediate(framework.Poll, timeout, func() (bool, error) {
clusterDaemonSet, err := clientset.Extensions().DaemonSets(namespace).Get(daemonset.Name)
if err == nil { // We want it present, and the Get succeeded, so we're all good.
if util.ObjectMetaAndSpecEquivalent(clusterDaemonSet, daemonset) {
By(fmt.Sprintf("Success: shard of federated daemonset %q in namespace %q in cluster is updated", daemonset.Name, namespace))
return true, nil
} else {
By(fmt.Sprintf("Expected equal daemonsets. expected: %+v\nactual: %+v", *daemonset, *clusterDaemonSet))
}
By(fmt.Sprintf("DaemonSet %q in namespace %q in cluster, waiting for daemonset being updated, trying again in %s (err=%v)", daemonset.Name, namespace, framework.Poll, err))
return false, nil
}
By(fmt.Sprintf("DaemonSet %q in namespace %q in cluster, waiting for being updated, trying again in %s (err=%v)", daemonset.Name, namespace, framework.Poll, err))
return false, nil
})
framework.ExpectNoError(err, "Failed to verify daemonset %q in namespace %q in cluster", daemonset.Name, namespace)
}