forked from kubernetes-retired/contrib
-
Notifications
You must be signed in to change notification settings - Fork 0
/
gce_manager.go
232 lines (204 loc) · 6.68 KB
/
gce_manager.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
/*
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 gce
import (
"fmt"
"io"
"strings"
"sync"
"time"
"gopkg.in/gcfg.v1"
"github.com/golang/glog"
"golang.org/x/oauth2"
"golang.org/x/oauth2/google"
gce "google.golang.org/api/compute/v1"
provider_gce "k8s.io/kubernetes/pkg/cloudprovider/providers/gce"
"k8s.io/kubernetes/pkg/util/wait"
)
const (
operationWaitTimeout = 5 * time.Second
operationPollInterval = 100 * time.Millisecond
)
type migInformation struct {
config *Mig
basename string
}
// GceManager is handles gce communication and data caching.
type GceManager struct {
migs []*migInformation
migCache map[GceRef]*Mig
service *gce.Service
cacheMutex sync.Mutex
}
// CreateGceManager constructs gceManager object.
func CreateGceManager(configReader io.Reader) (*GceManager, error) {
// Create Google Compute Engine token.
tokenSource := google.ComputeTokenSource("")
if configReader != nil {
var cfg provider_gce.Config
if err := gcfg.ReadInto(&cfg, configReader); err != nil {
glog.Errorf("Couldn't read config: %v", err)
return nil, err
}
if cfg.Global.TokenURL == "" {
glog.Warning("Empty tokenUrl in cloud config")
} else {
glog.Infof("Using TokenSource from config %#v", tokenSource)
tokenSource = provider_gce.NewAltTokenSource(cfg.Global.TokenURL, cfg.Global.TokenBody)
}
} else {
glog.Infof("Using default TokenSource %#v", tokenSource)
}
// Create Google Compute Engine service.
client := oauth2.NewClient(oauth2.NoContext, tokenSource)
gceService, err := gce.New(client)
if err != nil {
return nil, err
}
manager := &GceManager{
migs: make([]*migInformation, 0),
service: gceService,
migCache: make(map[GceRef]*Mig),
}
go wait.Forever(func() {
manager.cacheMutex.Lock()
defer manager.cacheMutex.Unlock()
if err := manager.regenerateCache(); err != nil {
glog.Errorf("Error while regenerating Mig cache: %v", err)
}
}, time.Hour)
return manager, nil
}
// RegisterMig registers mig in Gce Manager.
func (m *GceManager) RegisterMig(mig *Mig) {
m.cacheMutex.Lock()
defer m.cacheMutex.Unlock()
m.migs = append(m.migs, &migInformation{
config: mig,
})
}
// GetMigSize gets MIG size.
func (m *GceManager) GetMigSize(mig *Mig) (int64, error) {
igm, err := m.service.InstanceGroupManagers.Get(mig.Project, mig.Zone, mig.Name).Do()
if err != nil {
return -1, err
}
return igm.TargetSize, nil
}
// SetMigSize sets MIG size.
func (m *GceManager) SetMigSize(mig *Mig, size int64) error {
op, err := m.service.InstanceGroupManagers.Resize(mig.Project, mig.Zone, mig.Name, size).Do()
if err != nil {
return err
}
if err := m.waitForOp(op, mig.Project, mig.Zone); err != nil {
return err
}
return nil
}
func (m *GceManager) waitForOp(operation *gce.Operation, project string, zone string) error {
for start := time.Now(); time.Since(start) < operationWaitTimeout; time.Sleep(operationPollInterval) {
glog.V(4).Infof("Waiting for operation %s %s %s", project, zone, operation.Name)
if op, err := m.service.ZoneOperations.Get(project, zone, operation.Name).Do(); err == nil {
glog.V(4).Infof("Operation %s %s %s status: %s", project, zone, operation.Name, op.Status)
if op.Status == "DONE" {
return nil
}
} else {
glog.Warningf("Error while getting operation %s on %s: %v", operation.Name, operation.TargetLink, err)
}
}
return fmt.Errorf("Timeout while waiting for operation %s on %s to complete.", operation.Name, operation.TargetLink)
}
// DeleteInstances deletes the given instances. All instances must be controlled by the same MIG.
func (m *GceManager) DeleteInstances(instances []*GceRef) error {
if len(instances) == 0 {
return nil
}
commonMig, err := m.GetMigForInstance(instances[0])
if err != nil {
return err
}
for _, instance := range instances {
mig, err := m.GetMigForInstance(instance)
if err != nil {
return err
}
if mig != commonMig {
return fmt.Errorf("Connot delete instances which don't belong to the same MIG.")
}
}
req := gce.InstanceGroupManagersDeleteInstancesRequest{
Instances: []string{},
}
for _, instance := range instances {
req.Instances = append(req.Instances, GenerateInstanceUrl(instance.Project, instance.Zone, instance.Name))
}
op, err := m.service.InstanceGroupManagers.DeleteInstances(commonMig.Project, commonMig.Zone, commonMig.Name, &req).Do()
if err != nil {
return err
}
if err := m.waitForOp(op, commonMig.Project, commonMig.Zone); err != nil {
return err
}
return nil
}
// GetMigForInstance returns MigConfig of the given Instance
func (m *GceManager) GetMigForInstance(instance *GceRef) (*Mig, error) {
m.cacheMutex.Lock()
defer m.cacheMutex.Unlock()
if mig, found := m.migCache[*instance]; found {
return mig, nil
}
for _, mig := range m.migs {
if mig.config.Project == instance.Project &&
mig.config.Zone == instance.Zone &&
strings.HasPrefix(instance.Name, mig.basename) {
if err := m.regenerateCache(); err != nil {
return nil, fmt.Errorf("Error while looking for MIG for instance %+v, error: %v", *instance, err)
}
if mig, found := m.migCache[*instance]; found {
return mig, nil
}
return nil, fmt.Errorf("Instance %+v does not belong to any configured MIG", *instance)
}
}
// Instance doesn't belong to any configured mig.
return nil, nil
}
func (m *GceManager) regenerateCache() error {
newMigCache := make(map[GceRef]*Mig)
for _, migInfo := range m.migs {
mig := migInfo.config
glog.V(4).Infof("Regenerating MIG information for %s %s %s", mig.Project, mig.Zone, mig.Name)
instanceGroupManager, err := m.service.InstanceGroupManagers.Get(mig.Project, mig.Zone, mig.Name).Do()
if err != nil {
return err
}
migInfo.basename = instanceGroupManager.BaseInstanceName
instances, err := m.service.InstanceGroupManagers.ListManagedInstances(mig.Project, mig.Zone, mig.Name).Do()
if err != nil {
glog.V(4).Infof("Failed MIG info request for %s %s %s: %v", mig.Project, mig.Zone, mig.Name, err)
return err
}
for _, instance := range instances.ManagedInstances {
project, zone, name, err := ParseInstanceUrl(instance.Instance)
if err != nil {
return err
}
newMigCache[GceRef{Project: project, Zone: zone, Name: name}] = mig
}
}
m.migCache = newMigCache
return nil
}