forked from kubernetes/kubernetes
-
Notifications
You must be signed in to change notification settings - Fork 0
/
azure_blobDiskController.go
808 lines (677 loc) · 26.1 KB
/
azure_blobDiskController.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
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
/*
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 azure
import (
"bytes"
"encoding/binary"
"fmt"
"math"
"net/url"
"os"
"regexp"
"sync"
"strconv"
"strings"
"sync/atomic"
"time"
storage "github.com/Azure/azure-sdk-for-go/arm/storage"
azstorage "github.com/Azure/azure-sdk-for-go/storage"
"github.com/Azure/go-autorest/autorest/to"
"github.com/golang/glog"
"github.com/rubiojr/go-vhd/vhd"
kwait "k8s.io/apimachinery/pkg/util/wait"
"k8s.io/kubernetes/pkg/volume"
)
const (
vhdContainerName = "vhds"
useHTTPSForBlobBasedDisk = true
blobServiceName = "blob"
)
type storageAccountState struct {
name string
saType storage.SkuName
key string
diskCount int32
isValidating int32
defaultContainerCreated bool
}
//BlobDiskController : blob disk controller struct
type BlobDiskController struct {
common *controllerCommon
accounts map[string]*storageAccountState
}
var defaultContainerName = ""
var storageAccountNamePrefix = ""
var storageAccountNameMatch = ""
var initFlag int64
var accountsLock = &sync.Mutex{}
func newBlobDiskController(common *controllerCommon) (*BlobDiskController, error) {
c := BlobDiskController{common: common}
err := c.init()
if err != nil {
return nil, err
}
return &c, nil
}
// CreateVolume creates a VHD blob in a given storage account, will create the given storage account if it does not exist in current resource group
func (c *BlobDiskController) CreateVolume(name, storageAccount string, storageAccountType storage.SkuName, location string, requestGB int) (string, string, int, error) {
key, err := c.common.cloud.getStorageAccesskey(storageAccount)
if err != nil {
glog.V(2).Infof("azureDisk - no key found for storage account %s in resource group %s, begin to create a new storage account", storageAccount, c.common.resourceGroup)
cp := storage.AccountCreateParameters{
Sku: &storage.Sku{Name: storageAccountType},
Tags: &map[string]*string{"created-by": to.StringPtr("azure-dd")},
Location: &location}
cancel := make(chan struct{})
_, errchan := c.common.cloud.StorageAccountClient.Create(c.common.resourceGroup, storageAccount, cp, cancel)
err = <-errchan
if err != nil {
return "", "", 0, fmt.Errorf(fmt.Sprintf("Create Storage Account %s, error: %s", storageAccount, err))
}
key, err = c.common.cloud.getStorageAccesskey(storageAccount)
if err != nil {
return "", "", 0, fmt.Errorf("no key found for storage account %s even after creating a new storage account", storageAccount)
}
glog.Errorf("no key found for storage account %s in resource group %s", storageAccount, c.common.resourceGroup)
return "", "", 0, err
}
client, err := azstorage.NewBasicClientOnSovereignCloud(storageAccount, key, c.common.cloud.Environment)
if err != nil {
return "", "", 0, err
}
blobClient := client.GetBlobService()
container := blobClient.GetContainerReference(vhdContainerName)
_, err = container.CreateIfNotExists(&azstorage.CreateContainerOptions{Access: azstorage.ContainerAccessTypePrivate})
if err != nil {
return "", "", 0, err
}
diskName, diskURI, err := c.createVHDBlobDisk(blobClient, storageAccount, name, vhdContainerName, int64(requestGB))
if err != nil {
return "", "", 0, err
}
glog.V(4).Infof("azureDisk - created vhd blob uri: %s", diskURI)
return diskName, diskURI, requestGB, err
}
// DeleteVolume deletes a VHD blob
func (c *BlobDiskController) DeleteVolume(diskURI string) error {
glog.V(4).Infof("azureDisk - begin to delete volume %s", diskURI)
accountName, blob, err := c.common.cloud.getBlobNameAndAccountFromURI(diskURI)
if err != nil {
return fmt.Errorf("failed to parse vhd URI %v", err)
}
key, err := c.common.cloud.getStorageAccesskey(accountName)
if err != nil {
return fmt.Errorf("no key for storage account %s, err %v", accountName, err)
}
err = c.common.cloud.deleteVhdBlob(accountName, key, blob)
if err != nil {
glog.Warningf("azureDisk - failed to delete blob %s err: %v", diskURI, err)
detail := err.Error()
if strings.Contains(detail, errLeaseIDMissing) {
// disk is still being used
// see https://msdn.microsoft.com/en-us/library/microsoft.windowsazure.storage.blob.protocol.bloberrorcodestrings.leaseidmissing.aspx
return volume.NewDeletedVolumeInUseError(fmt.Sprintf("disk %q is still in use while being deleted", diskURI))
}
return fmt.Errorf("failed to delete vhd %v, account %s, blob %s, err: %v", diskURI, accountName, blob, err)
}
glog.V(4).Infof("azureDisk - blob %s deleted", diskURI)
return nil
}
// get diskURI https://foo.blob.core.windows.net/vhds/bar.vhd and return foo (account) and bar.vhd (blob name)
func (c *BlobDiskController) getBlobNameAndAccountFromURI(diskURI string) (string, string, error) {
scheme := "http"
if useHTTPSForBlobBasedDisk {
scheme = "https"
}
host := fmt.Sprintf("%s://(.*).%s.%s", scheme, blobServiceName, c.common.storageEndpointSuffix)
reStr := fmt.Sprintf("%s/%s/(.*)", host, vhdContainerName)
re := regexp.MustCompile(reStr)
res := re.FindSubmatch([]byte(diskURI))
if len(res) < 3 {
return "", "", fmt.Errorf("invalid vhd URI for regex %s: %s", reStr, diskURI)
}
return string(res[1]), string(res[2]), nil
}
func (c *BlobDiskController) createVHDBlobDisk(blobClient azstorage.BlobStorageClient, accountName, vhdName, containerName string, sizeGB int64) (string, string, error) {
container := blobClient.GetContainerReference(containerName)
_, err := container.CreateIfNotExists(&azstorage.CreateContainerOptions{Access: azstorage.ContainerAccessTypePrivate})
if err != nil {
return "", "", err
}
size := 1024 * 1024 * 1024 * sizeGB
vhdSize := size + vhd.VHD_HEADER_SIZE /* header size */
// Blob name in URL must end with '.vhd' extension.
vhdName = vhdName + ".vhd"
tags := make(map[string]string)
tags["createdby"] = "k8sAzureDataDisk"
glog.V(4).Infof("azureDisk - creating page blob %name in container %s account %s", vhdName, containerName, accountName)
blob := container.GetBlobReference(vhdName)
blob.Properties.ContentLength = vhdSize
blob.Metadata = tags
err = blob.PutPageBlob(nil)
if err != nil {
return "", "", fmt.Errorf("failed to put page blob %s in container %s: %v", vhdName, containerName, err)
}
// add VHD signature to the blob
h, err := createVHDHeader(uint64(size))
if err != nil {
blob.DeleteIfExists(nil)
return "", "", fmt.Errorf("failed to create vhd header, err: %v", err)
}
blobRange := azstorage.BlobRange{
Start: uint64(size),
End: uint64(vhdSize - 1),
}
if err = blob.WriteRange(blobRange, bytes.NewBuffer(h[:vhd.VHD_HEADER_SIZE]), nil); err != nil {
glog.Infof("azureDisk - failed to put header page for data disk %s in container %s account %s, error was %s\n",
vhdName, containerName, accountName, err.Error())
return "", "", err
}
scheme := "http"
if useHTTPSForBlobBasedDisk {
scheme = "https"
}
host := fmt.Sprintf("%s://%s.%s.%s", scheme, accountName, blobServiceName, c.common.storageEndpointSuffix)
uri := fmt.Sprintf("%s/%s/%s", host, containerName, vhdName)
return vhdName, uri, nil
}
// delete a vhd blob
func (c *BlobDiskController) deleteVhdBlob(accountName, accountKey, blobName string) error {
client, err := azstorage.NewBasicClientOnSovereignCloud(accountName, accountKey, c.common.cloud.Environment)
if err != nil {
return err
}
blobSvc := client.GetBlobService()
container := blobSvc.GetContainerReference(vhdContainerName)
blob := container.GetBlobReference(blobName)
return blob.Delete(nil)
}
//CreateBlobDisk : create a blob disk in a node
func (c *BlobDiskController) CreateBlobDisk(dataDiskName string, storageAccountType storage.SkuName, sizeGB int, forceStandAlone bool) (string, error) {
glog.V(4).Infof("azureDisk - creating blob data disk named:%s on StorageAccountType:%s StandAlone:%v", dataDiskName, storageAccountType, forceStandAlone)
var storageAccountName = ""
var err error
if forceStandAlone {
// we have to wait until the storage account is is created
storageAccountName = "p" + MakeCRC32(c.common.subscriptionID+c.common.resourceGroup+dataDiskName)
err = c.createStorageAccount(storageAccountName, storageAccountType, c.common.location, false)
if err != nil {
return "", err
}
} else {
storageAccountName, err = c.findSANameForDisk(storageAccountType)
if err != nil {
return "", err
}
}
blobClient, err := c.getBlobSvcClient(storageAccountName)
if err != nil {
return "", err
}
_, diskURI, err := c.createVHDBlobDisk(blobClient, storageAccountName, dataDiskName, defaultContainerName, int64(sizeGB))
if err != nil {
return "", err
}
if !forceStandAlone {
atomic.AddInt32(&c.accounts[storageAccountName].diskCount, 1)
}
return diskURI, nil
}
//DeleteBlobDisk : delete a blob disk from a node
func (c *BlobDiskController) DeleteBlobDisk(diskURI string, wasForced bool) error {
storageAccountName, vhdName, err := diskNameandSANameFromURI(diskURI)
if err != nil {
return err
}
_, ok := c.accounts[storageAccountName]
if !ok {
// the storage account is specified by user
glog.V(4).Infof("azureDisk - deleting volume %s", diskURI)
return c.DeleteVolume(diskURI)
}
// if forced (as in one disk = one storage account)
// delete the account completely
if wasForced {
return c.deleteStorageAccount(storageAccountName)
}
blobSvc, err := c.getBlobSvcClient(storageAccountName)
if err != nil {
return err
}
glog.V(4).Infof("azureDisk - About to delete vhd file %s on storage account %s container %s", vhdName, storageAccountName, defaultContainerName)
container := blobSvc.GetContainerReference(defaultContainerName)
blob := container.GetBlobReference(vhdName)
_, err = blob.DeleteIfExists(nil)
if c.accounts[storageAccountName].diskCount == -1 {
if diskCount, err := c.getDiskCount(storageAccountName); err != nil {
c.accounts[storageAccountName].diskCount = int32(diskCount)
} else {
glog.Warningf("azureDisk - failed to get disk count for %s however the delete disk operation was ok", storageAccountName)
return nil // we have failed to aquire a new count. not an error condition
}
}
atomic.AddInt32(&c.accounts[storageAccountName].diskCount, -1)
return err
}
// Init tries best effort to ensure that 2 accounts standard/premium were created
// to be used by shared blob disks. This to increase the speed pvc provisioning (in most of cases)
func (c *BlobDiskController) init() error {
if !c.shouldInit() {
return nil
}
c.setUniqueStrings()
// get accounts
accounts, err := c.getAllStorageAccounts()
if err != nil {
return err
}
c.accounts = accounts
if len(c.accounts) == 0 {
counter := 1
for counter <= storageAccountsCountInit {
accountType := storage.PremiumLRS
if n := math.Mod(float64(counter), 2); n == 0 {
accountType = storage.StandardLRS
}
// We don't really care if these calls failed
// at this stage, we are trying to ensure 2 accounts (Standard/Premium)
// are there ready for PVC creation
// if we failed here, the accounts will be created in the process
// of creating PVC
// nor do we care if they were partially created, as the entire
// account creation process is idempotent
go func(thisNext int) {
newAccountName := getAccountNameForNum(thisNext)
glog.Infof("azureDisk - BlobDiskController init process will create new storageAccount:%s type:%s", newAccountName, accountType)
err := c.createStorageAccount(newAccountName, accountType, c.common.location, true)
// TODO return created and error from
if err != nil {
glog.Infof("azureDisk - BlobDiskController init: create account %s with error:%s", newAccountName, err.Error())
} else {
glog.Infof("azureDisk - BlobDiskController init: created account %s", newAccountName)
}
}(counter)
counter = counter + 1
}
}
return nil
}
//Sets unique strings to be used as accountnames && || blob containers names
func (c *BlobDiskController) setUniqueStrings() {
uniqueString := c.common.resourceGroup + c.common.location + c.common.subscriptionID
hash := MakeCRC32(uniqueString)
//used to generate a unqie container name used by this cluster PVC
defaultContainerName = hash
storageAccountNamePrefix = fmt.Sprintf(storageAccountNameTemplate, hash)
// Used to filter relevant accounts (accounts used by shared PVC)
storageAccountNameMatch = storageAccountNamePrefix
// Used as a template to create new names for relevant accounts
storageAccountNamePrefix = storageAccountNamePrefix + "%s"
}
func (c *BlobDiskController) getStorageAccountKey(SAName string) (string, error) {
if account, exists := c.accounts[SAName]; exists && account.key != "" {
return c.accounts[SAName].key, nil
}
listKeysResult, err := c.common.cloud.StorageAccountClient.ListKeys(c.common.resourceGroup, SAName)
if err != nil {
return "", err
}
if listKeysResult.Keys == nil {
return "", fmt.Errorf("azureDisk - empty listKeysResult in storage account:%s keys", SAName)
}
for _, v := range *listKeysResult.Keys {
if v.Value != nil && *v.Value == "key1" {
if _, ok := c.accounts[SAName]; !ok {
glog.Warningf("azureDisk - account %s was not cached while getting keys", SAName)
return *v.Value, nil
}
}
c.accounts[SAName].key = *v.Value
return c.accounts[SAName].key, nil
}
return "", fmt.Errorf("couldn't find key named key1 in storage account:%s keys", SAName)
}
func (c *BlobDiskController) getBlobSvcClient(SAName string) (azstorage.BlobStorageClient, error) {
key := ""
var client azstorage.Client
var blobSvc azstorage.BlobStorageClient
var err error
if key, err = c.getStorageAccountKey(SAName); err != nil {
return blobSvc, err
}
if client, err = azstorage.NewBasicClientOnSovereignCloud(SAName, key, c.common.cloud.Environment); err != nil {
return blobSvc, err
}
blobSvc = client.GetBlobService()
return blobSvc, nil
}
func (c *BlobDiskController) ensureDefaultContainer(storageAccountName string) error {
var err error
var blobSvc azstorage.BlobStorageClient
// short circut the check via local cache
// we are forgiving the fact that account may not be in cache yet
if v, ok := c.accounts[storageAccountName]; ok && v.defaultContainerCreated {
return nil
}
// not cached, check existance and readiness
bExist, provisionState, _ := c.getStorageAccountState(storageAccountName)
// account does not exist
if !bExist {
return fmt.Errorf("azureDisk - account %s does not exist while trying to create/ensure default container", storageAccountName)
}
// account exists but not ready yet
if provisionState != storage.Succeeded {
// we don't want many attempts to validate the account readiness
// here hence we are locking
counter := 1
for swapped := atomic.CompareAndSwapInt32(&c.accounts[storageAccountName].isValidating, 0, 1); swapped != true; {
time.Sleep(3 * time.Second)
counter = counter + 1
// check if we passed the max sleep
if counter >= 20 {
return fmt.Errorf("azureDisk - timeout waiting to aquire lock to validate account:%s readiness", storageAccountName)
}
}
// swapped
defer func() {
c.accounts[storageAccountName].isValidating = 0
}()
// short circut the check again.
if v, ok := c.accounts[storageAccountName]; ok && v.defaultContainerCreated {
return nil
}
err = kwait.ExponentialBackoff(defaultBackOff, func() (bool, error) {
_, provisionState, err := c.getStorageAccountState(storageAccountName)
if err != nil {
glog.V(4).Infof("azureDisk - GetStorageAccount:%s err %s", storageAccountName, err.Error())
return false, err
}
if provisionState == storage.Succeeded {
return true, nil
}
glog.V(4).Infof("azureDisk - GetStorageAccount:%s not ready yet", storageAccountName)
// leave it for next loop/sync loop
return false, fmt.Errorf("azureDisk - Account %s has not been flagged Succeeded by ARM", storageAccountName)
})
// we have failed to ensure that account is ready for us to create
// the default vhd container
if err != nil {
return err
}
}
if blobSvc, err = c.getBlobSvcClient(storageAccountName); err != nil {
return err
}
container := blobSvc.GetContainerReference(defaultContainerName)
bCreated, err := container.CreateIfNotExists(&azstorage.CreateContainerOptions{Access: azstorage.ContainerAccessTypePrivate})
if err != nil {
return err
}
if bCreated {
glog.V(2).Infof("azureDisk - storage account:%s had no default container(%s) and it was created \n", storageAccountName, defaultContainerName)
}
// flag so we no longer have to check on ARM
c.accounts[storageAccountName].defaultContainerCreated = true
return nil
}
// Gets Disk counts per storage account
func (c *BlobDiskController) getDiskCount(SAName string) (int, error) {
// if we have it in cache
if c.accounts[SAName].diskCount != -1 {
return int(c.accounts[SAName].diskCount), nil
}
var err error
var blobSvc azstorage.BlobStorageClient
if err = c.ensureDefaultContainer(SAName); err != nil {
return 0, err
}
if blobSvc, err = c.getBlobSvcClient(SAName); err != nil {
return 0, err
}
params := azstorage.ListBlobsParameters{}
container := blobSvc.GetContainerReference(defaultContainerName)
response, err := container.ListBlobs(params)
if err != nil {
return 0, err
}
glog.V(4).Infof("azure-Disk - refreshed data count for account %s and found %v", SAName, len(response.Blobs))
c.accounts[SAName].diskCount = int32(len(response.Blobs))
return int(c.accounts[SAName].diskCount), nil
}
// shouldInit ensures that we only init the plugin once
// and we only do that in the controller
func (c *BlobDiskController) shouldInit() bool {
if os.Args[0] == "kube-controller-manager" || (os.Args[0] == "/hyperkube" && os.Args[1] == "controller-manager") {
swapped := atomic.CompareAndSwapInt64(&initFlag, 0, 1)
if swapped {
return true
}
}
return false
}
func (c *BlobDiskController) getAllStorageAccounts() (map[string]*storageAccountState, error) {
accountListResult, err := c.common.cloud.StorageAccountClient.List()
if err != nil {
return nil, err
}
if accountListResult.Value == nil {
return nil, fmt.Errorf("azureDisk - empty accountListResult")
}
accounts := make(map[string]*storageAccountState)
for _, v := range *accountListResult.Value {
if strings.Index(*v.Name, storageAccountNameMatch) != 0 {
continue
}
if v.Name == nil || v.Sku == nil {
glog.Infof("azureDisk - accountListResult Name or Sku is nil")
continue
}
glog.Infof("azureDisk - identified account %s as part of shared PVC accounts", *v.Name)
sastate := &storageAccountState{
name: *v.Name,
saType: (*v.Sku).Name,
diskCount: -1,
}
accounts[*v.Name] = sastate
}
return accounts, nil
}
func (c *BlobDiskController) createStorageAccount(storageAccountName string, storageAccountType storage.SkuName, location string, checkMaxAccounts bool) error {
bExist, _, _ := c.getStorageAccountState(storageAccountName)
if bExist {
newAccountState := &storageAccountState{
diskCount: -1,
saType: storageAccountType,
name: storageAccountName,
}
c.addAccountState(storageAccountName, newAccountState)
}
// Account Does not exist
if !bExist {
if len(c.accounts) == maxStorageAccounts && checkMaxAccounts {
return fmt.Errorf("azureDisk - can not create new storage account, current storage accounts count:%v Max is:%v", len(c.accounts), maxStorageAccounts)
}
glog.V(2).Infof("azureDisk - Creating storage account %s type %s \n", storageAccountName, string(storageAccountType))
cp := storage.AccountCreateParameters{
Sku: &storage.Sku{Name: storageAccountType},
Tags: &map[string]*string{"created-by": to.StringPtr("azure-dd")},
Location: &location}
cancel := make(chan struct{})
_, errChan := c.common.cloud.StorageAccountClient.Create(c.common.resourceGroup, storageAccountName, cp, cancel)
err := <-errChan
if err != nil {
return fmt.Errorf(fmt.Sprintf("Create Storage Account: %s, error: %s", storageAccountName, err))
}
newAccountState := &storageAccountState{
diskCount: -1,
saType: storageAccountType,
name: storageAccountName,
}
c.addAccountState(storageAccountName, newAccountState)
}
if !bExist {
// SA Accounts takes time to be provisioned
// so if this account was just created allow it sometime
// before polling
glog.V(2).Infof("azureDisk - storage account %s was just created, allowing time before polling status")
time.Sleep(25 * time.Second) // as observed 25 is the average time for SA to be provisioned
}
// finally, make sure that we default container is created
// before handing it back over
return c.ensureDefaultContainer(storageAccountName)
}
// finds a new suitable storageAccount for this disk
func (c *BlobDiskController) findSANameForDisk(storageAccountType storage.SkuName) (string, error) {
maxDiskCount := maxDisksPerStorageAccounts
SAName := ""
totalDiskCounts := 0
countAccounts := 0 // account of this type.
for _, v := range c.accounts {
// filter out any stand-alone disks/accounts
if strings.Index(v.name, storageAccountNameMatch) != 0 {
continue
}
// note: we compute avge stratified by type.
// this to enable user to grow per SA type to avoid low
//avg utilization on one account type skewing all data.
if v.saType == storageAccountType {
// compute average
dCount, err := c.getDiskCount(v.name)
if err != nil {
return "", err
}
totalDiskCounts = totalDiskCounts + dCount
countAccounts = countAccounts + 1
// empty account
if dCount == 0 {
glog.V(2).Infof("azureDisk - account %s identified for a new disk is because it has 0 allocated disks", v.name)
return v.name, nil // shortcircut, avg is good and no need to adjust
}
// if this account is less allocated
if dCount < maxDiskCount {
maxDiskCount = dCount
SAName = v.name
}
}
}
// if we failed to find storageaccount
if SAName == "" {
glog.V(2).Infof("azureDisk - failed to identify a suitable account for new disk and will attempt to create new account")
SAName = getAccountNameForNum(c.getNextAccountNum())
err := c.createStorageAccount(SAName, storageAccountType, c.common.location, true)
if err != nil {
return "", err
}
return SAName, nil
}
disksAfter := totalDiskCounts + 1 // with the new one!
avgUtilization := float64(disksAfter) / float64(countAccounts*maxDisksPerStorageAccounts)
aboveAvg := (avgUtilization > storageAccountUtilizationBeforeGrowing)
// avg are not create and we should craete more accounts if we can
if aboveAvg && countAccounts < maxStorageAccounts {
glog.V(2).Infof("azureDisk - shared storageAccounts utilzation(%v) > grow-at-avg-utilization (%v). New storage account will be created", avgUtilization, storageAccountUtilizationBeforeGrowing)
SAName = getAccountNameForNum(c.getNextAccountNum())
err := c.createStorageAccount(SAName, storageAccountType, c.common.location, true)
if err != nil {
return "", err
}
return SAName, nil
}
// avergates are not ok and we are at capacity(max storage accounts allowed)
if aboveAvg && countAccounts == maxStorageAccounts {
glog.Infof("azureDisk - shared storageAccounts utilzation(%v) > grow-at-avg-utilization (%v). But k8s maxed on SAs for PVC(%v). k8s will now exceed grow-at-avg-utilization without adding accounts",
avgUtilization, storageAccountUtilizationBeforeGrowing, maxStorageAccounts)
}
// we found a storage accounts && [ avg are ok || we reached max sa count ]
return SAName, nil
}
func (c *BlobDiskController) getNextAccountNum() int {
max := 0
for k := range c.accounts {
// filter out accounts that are for standalone
if strings.Index(k, storageAccountNameMatch) != 0 {
continue
}
num := getAccountNumFromName(k)
if num > max {
max = num
}
}
return max + 1
}
func (c *BlobDiskController) deleteStorageAccount(storageAccountName string) error {
resp, err := c.common.cloud.StorageAccountClient.Delete(c.common.resourceGroup, storageAccountName)
if err != nil {
return fmt.Errorf("azureDisk - Delete of storage account '%s' failed with status %s...%v", storageAccountName, resp.Status, err)
}
c.removeAccountState(storageAccountName)
glog.Infof("azureDisk - Storage Account %s was deleted", storageAccountName)
return nil
}
//Gets storage account exist, provisionStatus, Error if any
func (c *BlobDiskController) getStorageAccountState(storageAccountName string) (bool, storage.ProvisioningState, error) {
account, err := c.common.cloud.StorageAccountClient.GetProperties(c.common.resourceGroup, storageAccountName)
if err != nil {
return false, "", err
}
return true, account.AccountProperties.ProvisioningState, nil
}
func (c *BlobDiskController) addAccountState(key string, state *storageAccountState) {
accountsLock.Lock()
defer accountsLock.Unlock()
if _, ok := c.accounts[key]; !ok {
c.accounts[key] = state
}
}
func (c *BlobDiskController) removeAccountState(key string) {
accountsLock.Lock()
defer accountsLock.Unlock()
delete(c.accounts, key)
}
// pads account num with zeros as needed
func getAccountNameForNum(num int) string {
sNum := strconv.Itoa(num)
missingZeros := 3 - len(sNum)
strZero := ""
for missingZeros > 0 {
strZero = strZero + "0"
missingZeros = missingZeros - 1
}
sNum = strZero + sNum
return fmt.Sprintf(storageAccountNamePrefix, sNum)
}
func getAccountNumFromName(accountName string) int {
nameLen := len(accountName)
num, _ := strconv.Atoi(accountName[nameLen-3:])
return num
}
func createVHDHeader(size uint64) ([]byte, error) {
h := vhd.CreateFixedHeader(size, &vhd.VHDOptions{})
b := new(bytes.Buffer)
err := binary.Write(b, binary.BigEndian, h)
if err != nil {
return nil, err
}
return b.Bytes(), nil
}
func diskNameandSANameFromURI(diskURI string) (string, string, error) {
uri, err := url.Parse(diskURI)
if err != nil {
return "", "", err
}
hostName := uri.Host
storageAccountName := strings.Split(hostName, ".")[0]
segments := strings.Split(uri.Path, "/")
diskNameVhd := segments[len(segments)-1]
return storageAccountName, diskNameVhd, nil
}