forked from kubernetes/kubernetes
-
Notifications
You must be signed in to change notification settings - Fork 0
/
gce_instances.go
410 lines (360 loc) · 12.5 KB
/
gce_instances.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
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
/*
Copyright 2017 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 gce
import (
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"time"
"cloud.google.com/go/compute/metadata"
"github.com/golang/glog"
computealpha "google.golang.org/api/compute/v0.alpha"
compute "google.golang.org/api/compute/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/cloudprovider"
)
func newInstancesMetricContext(request, zone string) *metricContext {
return &metricContext{
start: time.Now(),
attributes: []string{"instances_" + request, unusedMetricLabel, zone},
}
}
// NodeAddresses is an implementation of Instances.NodeAddresses.
func (gce *GCECloud) NodeAddresses(_ types.NodeName) ([]v1.NodeAddress, error) {
internalIP, err := metadata.Get("instance/network-interfaces/0/ip")
if err != nil {
return nil, fmt.Errorf("couldn't get internal IP: %v", err)
}
externalIP, err := metadata.Get("instance/network-interfaces/0/access-configs/0/external-ip")
if err != nil {
return nil, fmt.Errorf("couldn't get external IP: %v", err)
}
return []v1.NodeAddress{
{Type: v1.NodeInternalIP, Address: internalIP},
{Type: v1.NodeExternalIP, Address: externalIP},
}, nil
}
// This method will not be called from the node that is requesting this ID.
// i.e. metadata service and other local methods cannot be used here
func (gce *GCECloud) NodeAddressesByProviderID(providerID string) ([]v1.NodeAddress, error) {
return []v1.NodeAddress{}, errors.New("unimplemented")
}
// InstanceTypeByProviderID returns the cloudprovider instance type of the node
// with the specified unique providerID This method will not be called from the
// node that is requesting this ID. i.e. metadata service and other local
// methods cannot be used here
func (gce *GCECloud) InstanceTypeByProviderID(providerID string) (string, error) {
return "", errors.New("unimplemented")
}
// ExternalID returns the cloud provider ID of the node with the specified NodeName (deprecated).
func (gce *GCECloud) ExternalID(nodeName types.NodeName) (string, error) {
instanceName := mapNodeNameToInstanceName(nodeName)
if gce.useMetadataServer {
// Use metadata, if possible, to fetch ID. See issue #12000
if gce.isCurrentInstance(instanceName) {
externalInstanceID, err := getCurrentExternalIDViaMetadata()
if err == nil {
return externalInstanceID, nil
}
}
}
// Fallback to GCE API call if metadata server fails to retrieve ID
inst, err := gce.getInstanceByName(instanceName)
if err != nil {
return "", err
}
return strconv.FormatUint(inst.ID, 10), nil
}
// InstanceID returns the cloud provider ID of the node with the specified NodeName.
func (gce *GCECloud) InstanceID(nodeName types.NodeName) (string, error) {
instanceName := mapNodeNameToInstanceName(nodeName)
if gce.useMetadataServer {
// Use metadata, if possible, to fetch ID. See issue #12000
if gce.isCurrentInstance(instanceName) {
projectID, zone, err := getProjectAndZone()
if err == nil {
return projectID + "/" + zone + "/" + canonicalizeInstanceName(instanceName), nil
}
}
}
instance, err := gce.getInstanceByName(instanceName)
if err != nil {
return "", err
}
return gce.projectID + "/" + instance.Zone + "/" + instance.Name, nil
}
// InstanceType returns the type of the specified node with the specified NodeName.
func (gce *GCECloud) InstanceType(nodeName types.NodeName) (string, error) {
instanceName := mapNodeNameToInstanceName(nodeName)
if gce.useMetadataServer {
// Use metadata, if possible, to fetch ID. See issue #12000
if gce.isCurrentInstance(instanceName) {
mType, err := getCurrentMachineTypeViaMetadata()
if err == nil {
return mType, nil
}
}
}
instance, err := gce.getInstanceByName(instanceName)
if err != nil {
return "", err
}
return instance.Type, nil
}
func (gce *GCECloud) AddSSHKeyToAllInstances(user string, keyData []byte) error {
return wait.Poll(2*time.Second, 30*time.Second, func() (bool, error) {
project, err := gce.service.Projects.Get(gce.projectID).Do()
if err != nil {
glog.Errorf("Could not get project: %v", err)
return false, nil
}
keyString := fmt.Sprintf("%s:%s %s@%s", user, strings.TrimSpace(string(keyData)), user, user)
found := false
for _, item := range project.CommonInstanceMetadata.Items {
if item.Key == "sshKeys" {
if strings.Contains(*item.Value, keyString) {
// We've already added the key
glog.Info("SSHKey already in project metadata")
return true, nil
}
value := *item.Value + "\n" + keyString
item.Value = &value
found = true
break
}
}
if !found {
// This is super unlikely, so log.
glog.Infof("Failed to find sshKeys metadata, creating a new item")
project.CommonInstanceMetadata.Items = append(project.CommonInstanceMetadata.Items,
&compute.MetadataItems{
Key: "sshKeys",
Value: &keyString,
})
}
mc := newInstancesMetricContext("add_ssh_key", "")
op, err := gce.service.Projects.SetCommonInstanceMetadata(
gce.projectID, project.CommonInstanceMetadata).Do()
if err != nil {
glog.Errorf("Could not Set Metadata: %v", err)
mc.Observe(err)
return false, nil
}
if err := gce.waitForGlobalOp(op, mc); err != nil {
glog.Errorf("Could not Set Metadata: %v", err)
return false, nil
}
glog.Infof("Successfully added sshKey to project metadata")
return true, nil
})
}
// GetAllZones returns all the zones in which nodes are running
func (gce *GCECloud) GetAllZones() (sets.String, error) {
// Fast-path for non-multizone
if len(gce.managedZones) == 1 {
return sets.NewString(gce.managedZones...), nil
}
// TODO: Caching, but this is currently only called when we are creating a volume,
// which is a relatively infrequent operation, and this is only # zones API calls
zones := sets.NewString()
// TODO: Parallelize, although O(zones) so not too bad (N <= 3 typically)
for _, zone := range gce.managedZones {
mc := newInstancesMetricContext("list", zone)
// We only retrieve one page in each zone - we only care about existence
listCall := gce.service.Instances.List(gce.projectID, zone)
// No filter: We assume that a zone is either used or unused
// We could only consider running nodes (like we do in List above),
// but probably if instances are starting we still want to consider them.
// I think we should wait until we have a reason to make the
// call one way or the other; we generally can't guarantee correct
// volume spreading if the set of zones is changing
// (and volume spreading is currently only a heuristic).
// Long term we want to replace GetAllZones (which primarily supports volume
// spreading) with a scheduler policy that is able to see the global state of
// volumes and the health of zones.
// Just a minimal set of fields - we only care about existence
listCall = listCall.Fields("items(name)")
res, err := listCall.Do()
if err != nil {
return nil, mc.Observe(err)
}
mc.Observe(nil)
if len(res.Items) != 0 {
zones.Insert(zone)
}
}
return zones, nil
}
// Implementation of Instances.CurrentNodeName
func (gce *GCECloud) CurrentNodeName(hostname string) (types.NodeName, error) {
return types.NodeName(hostname), nil
}
// AliasRanges returns a list of CIDR ranges that are assigned to the
// `node` for allocation to pods. Returns a list of the form
// "<ip>/<netmask>".
func (gce *GCECloud) AliasRanges(nodeName types.NodeName) (cidrs []string, err error) {
var instance *gceInstance
instance, err = gce.getInstanceByName(mapNodeNameToInstanceName(nodeName))
if err != nil {
return
}
var res *computealpha.Instance
res, err = gce.serviceAlpha.Instances.Get(
gce.projectID, instance.Zone, instance.Name).Do()
if err != nil {
return
}
for _, networkInterface := range res.NetworkInterfaces {
for _, aliasIpRange := range networkInterface.AliasIpRanges {
cidrs = append(cidrs, aliasIpRange.IpCidrRange)
}
}
return
}
// Gets the named instances, returning cloudprovider.InstanceNotFound if any instance is not found
func (gce *GCECloud) getInstancesByNames(names []string) ([]*gceInstance, error) {
instances := make(map[string]*gceInstance)
remaining := len(names)
nodeInstancePrefix := gce.nodeInstancePrefix
for _, name := range names {
name = canonicalizeInstanceName(name)
if !strings.HasPrefix(name, gce.nodeInstancePrefix) {
glog.Warningf("instance '%s' does not conform to prefix '%s', removing filter", name, gce.nodeInstancePrefix)
nodeInstancePrefix = ""
}
instances[name] = nil
}
for _, zone := range gce.managedZones {
if remaining == 0 {
break
}
pageToken := ""
page := 0
for ; page == 0 || (pageToken != "" && page < maxPages); page++ {
listCall := gce.service.Instances.List(gce.projectID, zone)
if nodeInstancePrefix != "" {
// Add the filter for hosts
listCall = listCall.Filter("name eq " + nodeInstancePrefix + ".*")
}
// TODO(zmerlynn): Internal bug 29524655
// listCall = listCall.Fields("items(name,id,disks,machineType)")
if pageToken != "" {
listCall.PageToken(pageToken)
}
res, err := listCall.Do()
if err != nil {
return nil, err
}
pageToken = res.NextPageToken
for _, i := range res.Items {
name := i.Name
if _, ok := instances[name]; !ok {
continue
}
instance := &gceInstance{
Zone: zone,
Name: name,
ID: i.Id,
Disks: i.Disks,
Type: lastComponent(i.MachineType),
}
instances[name] = instance
remaining--
}
}
if page >= maxPages {
glog.Errorf("getInstancesByNames exceeded maxPages=%d for Instances.List: truncating.", maxPages)
}
}
instanceArray := make([]*gceInstance, len(names))
for i, name := range names {
name = canonicalizeInstanceName(name)
instance := instances[name]
if instance == nil {
glog.Errorf("Failed to retrieve instance: %q", name)
return nil, cloudprovider.InstanceNotFound
}
instanceArray[i] = instances[name]
}
return instanceArray, nil
}
// Gets the named instance, returning cloudprovider.InstanceNotFound if the instance is not found
func (gce *GCECloud) getInstanceByName(name string) (*gceInstance, error) {
// Avoid changing behaviour when not managing multiple zones
for _, zone := range gce.managedZones {
name = canonicalizeInstanceName(name)
mc := newInstancesMetricContext("get", zone)
res, err := gce.service.Instances.Get(gce.projectID, zone, name).Do()
mc.Observe(err)
if err != nil {
glog.Errorf("getInstanceByName: failed to get instance %s; err: %v", name, err)
if isHTTPErrorCode(err, http.StatusNotFound) {
continue
}
return nil, err
}
return &gceInstance{
Zone: lastComponent(res.Zone),
Name: res.Name,
ID: res.Id,
Disks: res.Disks,
Type: lastComponent(res.MachineType),
}, nil
}
return nil, cloudprovider.InstanceNotFound
}
func getInstanceIDViaMetadata() (string, error) {
result, err := metadata.Get("instance/hostname")
if err != nil {
return "", err
}
parts := strings.Split(result, ".")
if len(parts) == 0 {
return "", fmt.Errorf("unexpected response: %s", result)
}
return parts[0], nil
}
func getCurrentExternalIDViaMetadata() (string, error) {
externalID, err := metadata.Get("instance/id")
if err != nil {
return "", fmt.Errorf("couldn't get external ID: %v", err)
}
return externalID, nil
}
func getCurrentMachineTypeViaMetadata() (string, error) {
mType, err := metadata.Get("instance/machine-type")
if err != nil {
return "", fmt.Errorf("couldn't get machine type: %v", err)
}
parts := strings.Split(mType, "/")
if len(parts) != 4 {
return "", fmt.Errorf("unexpected response for machine type: %s", mType)
}
return parts[3], nil
}
// isCurrentInstance uses metadata server to check if specified
// instanceID matches current machine's instanceID
func (gce *GCECloud) isCurrentInstance(instanceID string) bool {
currentInstanceID, err := getInstanceIDViaMetadata()
if err != nil {
// Log and swallow error
glog.Errorf("Failed to fetch instanceID via Metadata: %v", err)
return false
}
return currentInstanceID == canonicalizeInstanceName(instanceID)
}