forked from kubernetes/kubernetes
-
Notifications
You must be signed in to change notification settings - Fork 0
/
statusupdater.go
189 lines (163 loc) · 5.72 KB
/
statusupdater.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
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
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 node
import (
"fmt"
"time"
log "github.com/golang/glog"
"k8s.io/kubernetes/cmd/kubelet/app/options"
"k8s.io/kubernetes/contrib/mesos/pkg/runtime"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/client/cache"
client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/cloudprovider/providers/mesos"
"k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/util/sets"
)
const (
nodeStatusUpdateRetry = 5
slaveReadyReason = "SlaveReady"
slaveReadyMessage = "mesos reports ready status"
)
type StatusUpdater struct {
client *client.Client
relistPeriod time.Duration
heartBeatPeriod time.Duration
nowFunc func() time.Time
}
func NewStatusUpdater(client *client.Client, relistPeriod time.Duration, nowFunc func() time.Time) *StatusUpdater {
kubecfg := options.NewKubeletServer() // only create to get the config, this is without side-effects
return &StatusUpdater{
client: client,
relistPeriod: relistPeriod,
heartBeatPeriod: kubecfg.NodeStatusUpdateFrequency,
nowFunc: nowFunc,
}
}
func (u *StatusUpdater) Run(terminate <-chan struct{}) error {
nodeStore := cache.NewStore(cache.MetaNamespaceKeyFunc)
nodeLW := cache.NewListWatchFromClient(u.client, "nodes", api.NamespaceAll, fields.Everything())
cache.NewReflector(nodeLW, &api.Node{}, nodeStore, u.relistPeriod).Run()
monitor := func() {
// build up a set of listed slave nodes without a kubelet
slaves, err := mesos.CloudProvider.ListWithoutKubelet()
if err != nil {
log.Errorf("Error listing slaves without kubelet: %v", err)
return
}
slavesWithoutKubelet := sets.NewString(slaves...)
// update status for nodes which do not have a kubelet running and
// which are still existing as slave. This status update must be done
// before the node controller counts down the NodeMonitorGracePeriod
nodes := nodeStore.List()
for _, n := range nodes {
node := n.(*api.Node)
if !slavesWithoutKubelet.Has(node.Spec.ExternalID) {
// let the kubelet do its job updating the status, or the
// node controller will remove this node if the node does not even
// exist anymore
continue
}
err := u.updateStatus(node)
if err != nil {
log.Errorf("Error updating node status: %v", err)
}
}
}
go runtime.Until(monitor, u.heartBeatPeriod, terminate)
return nil
}
func (u *StatusUpdater) updateStatus(n *api.Node) error {
for i := 0; i < nodeStatusUpdateRetry; i++ {
if err := u.tryUpdateStatus(n); err != nil && !errors.IsConflict(err) {
log.Errorf("Error updating node status, will retry: %v", err)
} else {
return nil
}
}
return fmt.Errorf("Update node status exceeds retry count")
}
// nodeWithUpdatedStatus clones the given node and updates the NodeReady condition.
// The updated node is return and a boolean indicating whether the node was changed
// at all.
func (u *StatusUpdater) nodeWithUpdatedStatus(n *api.Node) (*api.Node, bool, error) {
readyCondition := getCondition(&n.Status, api.NodeReady)
currentTime := unversioned.NewTime(u.nowFunc())
// avoid flapping by waiting at least twice the kubetlet update frequency, i.e.
// give the kubelet the chance twice to update the heartbeat. This is necessary
// because we only poll the Mesos master state.json once in a while and we
// know that that the information from there can easily be outdated.
gracePeriod := u.heartBeatPeriod * 2
if readyCondition != nil && !currentTime.After(readyCondition.LastHeartbeatTime.Add(gracePeriod)) {
return n, false, nil
}
clone, err := api.Scheme.DeepCopy(n)
if err != nil {
return nil, false, err
}
n = clone.(*api.Node)
newNodeReadyCondition := api.NodeCondition{
Type: api.NodeReady,
Status: api.ConditionTrue,
Reason: slaveReadyReason,
Message: slaveReadyMessage,
LastHeartbeatTime: currentTime,
}
found := false
for i := range n.Status.Conditions {
c := &n.Status.Conditions[i]
if c.Type == api.NodeReady {
if c.Status == newNodeReadyCondition.Status {
newNodeReadyCondition.LastTransitionTime = c.LastTransitionTime
} else {
newNodeReadyCondition.LastTransitionTime = currentTime
}
n.Status.Conditions[i] = newNodeReadyCondition
found = true
break
}
}
if !found {
newNodeReadyCondition.LastTransitionTime = currentTime
n.Status.Conditions = append(n.Status.Conditions, newNodeReadyCondition)
}
return n, true, nil
}
// tryUpdateStatus updates the status of the given node and tries to persist that
// on the apiserver
func (u *StatusUpdater) tryUpdateStatus(n *api.Node) error {
n, updated, err := u.nodeWithUpdatedStatus(n)
if err != nil {
return err
}
if !updated {
return nil
}
_, err = u.client.Nodes().UpdateStatus(n)
return err
}
// getCondition returns a condition object for the specific condition
// type, nil if the condition is not set.
func getCondition(status *api.NodeStatus, conditionType api.NodeConditionType) *api.NodeCondition {
if status == nil {
return nil
}
for i := range status.Conditions {
if status.Conditions[i].Type == conditionType {
return &status.Conditions[i]
}
}
return nil
}