lib/cloud/azure/azure.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package azure
   6
   7 import (
   8         "context"
   9         "encoding/base64"
  10         "fmt"
  11         "net/http"
  12         "regexp"
  13         "strconv"
  14         "strings"
  15         "sync"
  16         "time"
  17
  18         "git.curoverse.com/arvados.git/lib/cloud"
  19         "git.curoverse.com/arvados.git/sdk/go/arvados"
  20         "github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2018-06-01/compute"
  21         "github.com/Azure/azure-sdk-for-go/services/network/mgmt/2018-06-01/network"
  22         storageacct "github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2018-02-01/storage"
  23         "github.com/Azure/azure-sdk-for-go/storage"
  24         "github.com/Azure/go-autorest/autorest"
  25         "github.com/Azure/go-autorest/autorest/azure"
  26         "github.com/Azure/go-autorest/autorest/azure/auth"
  27         "github.com/Azure/go-autorest/autorest/to"
  28         "github.com/jmcvetta/randutil"
  29         "github.com/mitchellh/mapstructure"
  30         "github.com/sirupsen/logrus"
  31         "golang.org/x/crypto/ssh"
  32 )
  33
  34 type AzureInstanceSetConfig struct {
  35         SubscriptionID               string  "SubscriptionID"
  36         ClientID                     string  "ClientID"
  37         ClientSecret                 string  "ClientSecret"
  38         TenantID                     string  "TenantID"
  39         CloudEnvironment             string  "CloudEnvironment"
  40         ResourceGroup                string  "ResourceGroup"
  41         Location                     string  "Location"
  42         Network                      string  "Network"
  43         Subnet                       string  "Subnet"
  44         StorageAccount               string  "StorageAccount"
  45         BlobContainer                string  "BlobContainer"
  46         DeleteDanglingResourcesAfter float64 "DeleteDanglingResourcesAfter"
  47 }
  48
  49 type VirtualMachinesClientWrapper interface {
  50         CreateOrUpdate(ctx context.Context,
  51                 resourceGroupName string,
  52                 VMName string,
  53                 parameters compute.VirtualMachine) (result compute.VirtualMachine, err error)
  54         Delete(ctx context.Context, resourceGroupName string, VMName string) (result *http.Response, err error)
  55         ListComplete(ctx context.Context, resourceGroupName string) (result compute.VirtualMachineListResultIterator, err error)
  56 }
  57
  58 type VirtualMachinesClientImpl struct {
  59         inner compute.VirtualMachinesClient
  60 }
  61
  62 func (cl *VirtualMachinesClientImpl) CreateOrUpdate(ctx context.Context,
  63         resourceGroupName string,
  64         VMName string,
  65         parameters compute.VirtualMachine) (result compute.VirtualMachine, err error) {
  66
  67         future, err := cl.inner.CreateOrUpdate(ctx, resourceGroupName, VMName, parameters)
  68         if err != nil {
  69                 return compute.VirtualMachine{}, WrapAzureError(err)
  70         }
  71         future.WaitForCompletionRef(ctx, cl.inner.Client)
  72         r, err := future.Result(cl.inner)
  73         return r, WrapAzureError(err)
  74 }
  75
  76 func (cl *VirtualMachinesClientImpl) Delete(ctx context.Context, resourceGroupName string, VMName string) (result *http.Response, err error) {
  77         future, err := cl.inner.Delete(ctx, resourceGroupName, VMName)
  78         if err != nil {
  79                 return nil, WrapAzureError(err)
  80         }
  81         err = future.WaitForCompletionRef(ctx, cl.inner.Client)
  82         return future.Response(), WrapAzureError(err)
  83 }
  84
  85 func (cl *VirtualMachinesClientImpl) ListComplete(ctx context.Context, resourceGroupName string) (result compute.VirtualMachineListResultIterator, err error) {
  86         r, err := cl.inner.ListComplete(ctx, resourceGroupName)
  87         return r, WrapAzureError(err)
  88 }
  89
  90 type InterfacesClientWrapper interface {
  91         CreateOrUpdate(ctx context.Context,
  92                 resourceGroupName string,
  93                 networkInterfaceName string,
  94                 parameters network.Interface) (result network.Interface, err error)
  95         Delete(ctx context.Context, resourceGroupName string, networkInterfaceName string) (result *http.Response, err error)
  96         ListComplete(ctx context.Context, resourceGroupName string) (result network.InterfaceListResultIterator, err error)
  97 }
  98
  99 type InterfacesClientImpl struct {
 100         inner network.InterfacesClient
 101 }
 102
 103 func (cl *InterfacesClientImpl) Delete(ctx context.Context, resourceGroupName string, VMName string) (result *http.Response, err error) {
 104         future, err := cl.inner.Delete(ctx, resourceGroupName, VMName)
 105         if err != nil {
 106                 return nil, WrapAzureError(err)
 107         }
 108         err = future.WaitForCompletionRef(ctx, cl.inner.Client)
 109         return future.Response(), WrapAzureError(err)
 110 }
 111
 112 func (cl *InterfacesClientImpl) CreateOrUpdate(ctx context.Context,
 113         resourceGroupName string,
 114         networkInterfaceName string,
 115         parameters network.Interface) (result network.Interface, err error) {
 116
 117         future, err := cl.inner.CreateOrUpdate(ctx, resourceGroupName, networkInterfaceName, parameters)
 118         if err != nil {
 119                 return network.Interface{}, WrapAzureError(err)
 120         }
 121         future.WaitForCompletionRef(ctx, cl.inner.Client)
 122         r, err := future.Result(cl.inner)
 123         return r, WrapAzureError(err)
 124 }
 125
 126 func (cl *InterfacesClientImpl) ListComplete(ctx context.Context, resourceGroupName string) (result network.InterfaceListResultIterator, err error) {
 127         r, err := cl.inner.ListComplete(ctx, resourceGroupName)
 128         return r, WrapAzureError(err)
 129 }
 130
 131 var quotaRe = regexp.MustCompile(`(?i:exceed|quota|limit)`)
 132
 133 type AzureRateLimitError struct {
 134         azure.RequestError
 135         earliestRetry time.Time
 136 }
 137
 138 func (ar *AzureRateLimitError) EarliestRetry() time.Time {
 139         return ar.earliestRetry
 140 }
 141
 142 type AzureQuotaError struct {
 143         azure.RequestError
 144 }
 145
 146 func (ar *AzureQuotaError) IsQuotaError() bool {
 147         return true
 148 }
 149
 150 func WrapAzureError(err error) error {
 151         de, ok := err.(autorest.DetailedError)
 152         if !ok {
 153                 return err
 154         }
 155         rq, ok := de.Original.(*azure.RequestError)
 156         if !ok {
 157                 return err
 158         }
 159         if rq.Response == nil {
 160                 return err
 161         }
 162         if rq.Response.StatusCode == 429 || len(rq.Response.Header["Retry-After"]) >= 1 {
 163                 // API throttling
 164                 ra := rq.Response.Header["Retry-After"][0]
 165                 earliestRetry, parseErr := http.ParseTime(ra)
 166                 if parseErr != nil {
 167                         // Could not parse as a timestamp, must be number of seconds
 168                         dur, parseErr := strconv.ParseInt(ra, 10, 64)
 169                         if parseErr == nil {
 170                                 earliestRetry = time.Now().Add(time.Duration(dur) * time.Second)
 171                         } else {
 172                                 // Couldn't make sense of retry-after,
 173                                 // so set retry to 20 seconds
 174                                 earliestRetry = time.Now().Add(20 * time.Second)
 175                         }
 176                 }
 177                 return &AzureRateLimitError{*rq, earliestRetry}
 178         }
 179         if rq.ServiceError == nil {
 180                 return err
 181         }
 182         if quotaRe.FindString(rq.ServiceError.Code) != "" || quotaRe.FindString(rq.ServiceError.Message) != "" {
 183                 return &AzureQuotaError{*rq}
 184         }
 185         return err
 186 }
 187
 188 type AzureInstanceSet struct {
 189         azconfig          AzureInstanceSetConfig
 190         vmClient          VirtualMachinesClientWrapper
 191         netClient         InterfacesClientWrapper
 192         storageAcctClient storageacct.AccountsClient
 193         azureEnv          azure.Environment
 194         interfaces        map[string]network.Interface
 195         dispatcherID      string
 196         namePrefix        string
 197         ctx               context.Context
 198         stopFunc          context.CancelFunc
 199         stopWg            sync.WaitGroup
 200         deleteNIC         chan string
 201         deleteBlob        chan storage.Blob
 202         logger            logrus.FieldLogger
 203 }
 204
 205 func NewAzureInstanceSet(config map[string]interface{}, dispatcherID cloud.InstanceSetID, logger logrus.FieldLogger) (prv cloud.InstanceSet, err error) {
 206         azcfg := AzureInstanceSetConfig{}
 207         if err = mapstructure.Decode(config, &azcfg); err != nil {
 208                 return nil, err
 209         }
 210         ap := AzureInstanceSet{logger: logger}
 211         err = ap.setup(azcfg, string(dispatcherID))
 212         if err != nil {
 213                 return nil, err
 214         }
 215         return &ap, nil
 216 }
 217
 218 func (az *AzureInstanceSet) setup(azcfg AzureInstanceSetConfig, dispatcherID string) (err error) {
 219         az.azconfig = azcfg
 220         vmClient := compute.NewVirtualMachinesClient(az.azconfig.SubscriptionID)
 221         netClient := network.NewInterfacesClient(az.azconfig.SubscriptionID)
 222         storageAcctClient := storageacct.NewAccountsClient(az.azconfig.SubscriptionID)
 223
 224         az.azureEnv, err = azure.EnvironmentFromName(az.azconfig.CloudEnvironment)
 225         if err != nil {
 226                 return err
 227         }
 228
 229         authorizer, err := auth.ClientCredentialsConfig{
 230                 ClientID:     az.azconfig.ClientID,
 231                 ClientSecret: az.azconfig.ClientSecret,
 232                 TenantID:     az.azconfig.TenantID,
 233                 Resource:     az.azureEnv.ResourceManagerEndpoint,
 234                 AADEndpoint:  az.azureEnv.ActiveDirectoryEndpoint,
 235         }.Authorizer()
 236         if err != nil {
 237                 return err
 238         }
 239
 240         vmClient.Authorizer = authorizer
 241         netClient.Authorizer = authorizer
 242         storageAcctClient.Authorizer = authorizer
 243
 244         az.vmClient = &VirtualMachinesClientImpl{vmClient}
 245         az.netClient = &InterfacesClientImpl{netClient}
 246         az.storageAcctClient = storageAcctClient
 247
 248         az.dispatcherID = dispatcherID
 249         az.namePrefix = fmt.Sprintf("compute-%s-", az.dispatcherID)
 250
 251         az.ctx, az.stopFunc = context.WithCancel(context.Background())
 252         go func() {
 253                 az.stopWg.Add(1)
 254                 defer az.stopWg.Done()
 255
 256                 tk := time.NewTicker(5 * time.Minute)
 257                 for {
 258                         select {
 259                         case <-az.ctx.Done():
 260                                 tk.Stop()
 261                                 return
 262                         case <-tk.C:
 263                                 az.ManageBlobs()
 264                         }
 265                 }
 266         }()
 267
 268         az.deleteNIC = make(chan string)
 269         az.deleteBlob = make(chan storage.Blob)
 270
 271         for i := 0; i < 4; i++ {
 272                 go func() {
 273                         for {
 274                                 nicname, ok := <-az.deleteNIC
 275                                 if !ok {
 276                                         return
 277                                 }
 278                                 _, delerr := az.netClient.Delete(context.Background(), az.azconfig.ResourceGroup, nicname)
 279                                 if delerr != nil {
 280                                         az.logger.WithError(delerr).Warnf("Error deleting %v", nicname)
 281                                 } else {
 282                                         az.logger.Printf("Deleted NIC %v", nicname)
 283                                 }
 284                         }
 285                 }()
 286                 go func() {
 287                         for {
 288                                 blob, ok := <-az.deleteBlob
 289                                 if !ok {
 290                                         return
 291                                 }
 292                                 err := blob.Delete(nil)
 293                                 if err != nil {
 294                                         az.logger.WithError(err).Warnf("Error deleting %v", blob.Name)
 295                                 } else {
 296                                         az.logger.Printf("Deleted blob %v", blob.Name)
 297                                 }
 298                         }
 299                 }()
 300         }
 301
 302         return nil
 303 }
 304
 305 func (az *AzureInstanceSet) Create(
 306         instanceType arvados.InstanceType,
 307         imageID cloud.ImageID,
 308         newTags cloud.InstanceTags,
 309         publicKey ssh.PublicKey) (cloud.Instance, error) {
 310
 311         az.stopWg.Add(1)
 312         defer az.stopWg.Done()
 313
 314         if len(newTags["node-token"]) == 0 {
 315                 return nil, fmt.Errorf("Must provide tag 'node-token'")
 316         }
 317
 318         name, err := randutil.String(15, "abcdefghijklmnopqrstuvwxyz0123456789")
 319         if err != nil {
 320                 return nil, err
 321         }
 322
 323         name = az.namePrefix + name
 324
 325         timestamp := time.Now().Format(time.RFC3339Nano)
 326
 327         tags := make(map[string]*string)
 328         tags["created-at"] = &timestamp
 329         for k, v := range newTags {
 330                 newstr := v
 331                 tags["dispatch-"+k] = &newstr
 332         }
 333
 334         tags["dispatch-instance-type"] = &instanceType.Name
 335
 336         nicParameters := network.Interface{
 337                 Location: &az.azconfig.Location,
 338                 Tags:     tags,
 339                 InterfacePropertiesFormat: &network.InterfacePropertiesFormat{
 340                         IPConfigurations: &[]network.InterfaceIPConfiguration{
 341                                 network.InterfaceIPConfiguration{
 342                                         Name: to.StringPtr("ip1"),
 343                                         InterfaceIPConfigurationPropertiesFormat: &network.InterfaceIPConfigurationPropertiesFormat{
 344                                                 Subnet: &network.Subnet{
 345                                                         ID: to.StringPtr(fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers"+
 346                                                                 "/Microsoft.Network/virtualnetworks/%s/subnets/%s",
 347                                                                 az.azconfig.SubscriptionID,
 348                                                                 az.azconfig.ResourceGroup,
 349                                                                 az.azconfig.Network,
 350                                                                 az.azconfig.Subnet)),
 351                                                 },
 352                                                 PrivateIPAllocationMethod: network.Dynamic,
 353                                         },
 354                                 },
 355                         },
 356                 },
 357         }
 358         nic, err := az.netClient.CreateOrUpdate(az.ctx, az.azconfig.ResourceGroup, name+"-nic", nicParameters)
 359         if err != nil {
 360                 return nil, WrapAzureError(err)
 361         }
 362
 363         instanceVhd := fmt.Sprintf("https://%s.blob.%s/%s/%s-os.vhd",
 364                 az.azconfig.StorageAccount,
 365                 az.azureEnv.StorageEndpointSuffix,
 366                 az.azconfig.BlobContainer,
 367                 name)
 368
 369         customData := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf(`#!/bin/sh
 370 echo '%s-%s' > /home/crunch/node-token`, name, newTags["node-token"])))
 371
 372         vmParameters := compute.VirtualMachine{
 373                 Location: &az.azconfig.Location,
 374                 Tags:     tags,
 375                 VirtualMachineProperties: &compute.VirtualMachineProperties{
 376                         HardwareProfile: &compute.HardwareProfile{
 377                                 VMSize: compute.VirtualMachineSizeTypes(instanceType.ProviderType),
 378                         },
 379                         StorageProfile: &compute.StorageProfile{
 380                                 OsDisk: &compute.OSDisk{
 381                                         OsType:       compute.Linux,
 382                                         Name:         to.StringPtr(name + "-os"),
 383                                         CreateOption: compute.FromImage,
 384                                         Image: &compute.VirtualHardDisk{
 385                                                 URI: to.StringPtr(string(imageID)),
 386                                         },
 387                                         Vhd: &compute.VirtualHardDisk{
 388                                                 URI: &instanceVhd,
 389                                         },
 390                                 },
 391                         },
 392                         NetworkProfile: &compute.NetworkProfile{
 393                                 NetworkInterfaces: &[]compute.NetworkInterfaceReference{
 394                                         compute.NetworkInterfaceReference{
 395                                                 ID: nic.ID,
 396                                                 NetworkInterfaceReferenceProperties: &compute.NetworkInterfaceReferenceProperties{
 397                                                         Primary: to.BoolPtr(true),
 398                                                 },
 399                                         },
 400                                 },
 401                         },
 402                         OsProfile: &compute.OSProfile{
 403                                 ComputerName:  &name,
 404                                 AdminUsername: to.StringPtr("crunch"),
 405                                 LinuxConfiguration: &compute.LinuxConfiguration{
 406                                         DisablePasswordAuthentication: to.BoolPtr(true),
 407                                         SSH: &compute.SSHConfiguration{
 408                                                 PublicKeys: &[]compute.SSHPublicKey{
 409                                                         compute.SSHPublicKey{
 410                                                                 Path:    to.StringPtr("/home/crunch/.ssh/authorized_keys"),
 411                                                                 KeyData: to.StringPtr(string(ssh.MarshalAuthorizedKey(publicKey))),
 412                                                         },
 413                                                 },
 414                                         },
 415                                 },
 416                                 CustomData: &customData,
 417                         },
 418                 },
 419         }
 420
 421         vm, err := az.vmClient.CreateOrUpdate(az.ctx, az.azconfig.ResourceGroup, name, vmParameters)
 422         if err != nil {
 423                 return nil, WrapAzureError(err)
 424         }
 425
 426         return &AzureInstance{
 427                 provider: az,
 428                 nic:      nic,
 429                 vm:       vm,
 430         }, nil
 431 }
 432
 433 func (az *AzureInstanceSet) Instances(cloud.InstanceTags) ([]cloud.Instance, error) {
 434         az.stopWg.Add(1)
 435         defer az.stopWg.Done()
 436
 437         interfaces, err := az.ManageNics()
 438         if err != nil {
 439                 return nil, err
 440         }
 441
 442         result, err := az.vmClient.ListComplete(az.ctx, az.azconfig.ResourceGroup)
 443         if err != nil {
 444                 return nil, WrapAzureError(err)
 445         }
 446
 447         instances := make([]cloud.Instance, 0)
 448
 449         for ; result.NotDone(); err = result.Next() {
 450                 if err != nil {
 451                         return nil, WrapAzureError(err)
 452                 }
 453                 if strings.HasPrefix(*result.Value().Name, az.namePrefix) {
 454                         instances = append(instances, &AzureInstance{
 455                                 provider: az,
 456                                 vm:       result.Value(),
 457                                 nic:      interfaces[*(*result.Value().NetworkProfile.NetworkInterfaces)[0].ID]})
 458                 }
 459         }
 460         return instances, nil
 461 }
 462
 463 // ManageNics returns a list of Azure network interface resources.
 464 // Also performs garbage collection of NICs which have "namePrefix", are
 465 // not associated with a virtual machine and have a "create-at" time
 466 // more than DeleteDanglingResourcesAfter (to prevent racing and
 467 // deleting newly created NICs) in the past are deleted.
 468 func (az *AzureInstanceSet) ManageNics() (map[string]network.Interface, error) {
 469         az.stopWg.Add(1)
 470         defer az.stopWg.Done()
 471
 472         result, err := az.netClient.ListComplete(az.ctx, az.azconfig.ResourceGroup)
 473         if err != nil {
 474                 return nil, WrapAzureError(err)
 475         }
 476
 477         interfaces := make(map[string]network.Interface)
 478
 479         timestamp := time.Now()
 480         for ; result.NotDone(); err = result.Next() {
 481                 if err != nil {
 482                         az.logger.WithError(err).Warnf("Error listing nics")
 483                         return interfaces, nil
 484                 }
 485                 if strings.HasPrefix(*result.Value().Name, az.namePrefix) {
 486                         if result.Value().VirtualMachine != nil {
 487                                 interfaces[*result.Value().ID] = result.Value()
 488                         } else {
 489                                 if result.Value().Tags["created-at"] != nil {
 490                                         createdAt, err := time.Parse(time.RFC3339Nano, *result.Value().Tags["created-at"])
 491                                         if err == nil {
 492                                                 if timestamp.Sub(createdAt).Seconds() > az.azconfig.DeleteDanglingResourcesAfter {
 493                                                         az.logger.Printf("Will delete %v because it is older than %v s", *result.Value().Name, az.azconfig.DeleteDanglingResourcesAfter)
 494                                                         az.deleteNIC <- *result.Value().Name
 495                                                 }
 496                                         }
 497                                 }
 498                         }
 499                 }
 500         }
 501         return interfaces, nil
 502 }
 503
 504 // ManageBlobs garbage collects blobs (VM disk images) in the
 505 // configured storage account container.  It will delete blobs which
 506 // have "namePrefix", are "available" (which means they are not
 507 // leased to a VM) and haven't been modified for
 508 // DeleteDanglingResourcesAfter seconds.
 509 func (az *AzureInstanceSet) ManageBlobs() {
 510         result, err := az.storageAcctClient.ListKeys(az.ctx, az.azconfig.ResourceGroup, az.azconfig.StorageAccount)
 511         if err != nil {
 512                 az.logger.WithError(err).Warn("Couldn't get account keys")
 513                 return
 514         }
 515
 516         key1 := *(*result.Keys)[0].Value
 517         client, err := storage.NewBasicClientOnSovereignCloud(az.azconfig.StorageAccount, key1, az.azureEnv)
 518         if err != nil {
 519                 az.logger.WithError(err).Warn("Couldn't make client")
 520                 return
 521         }
 522
 523         blobsvc := client.GetBlobService()
 524         blobcont := blobsvc.GetContainerReference(az.azconfig.BlobContainer)
 525
 526         page := storage.ListBlobsParameters{Prefix: az.namePrefix}
 527         timestamp := time.Now()
 528
 529         for {
 530                 response, err := blobcont.ListBlobs(page)
 531                 if err != nil {
 532                         az.logger.WithError(err).Warn("Error listing blobs")
 533                         return
 534                 }
 535                 for _, b := range response.Blobs {
 536                         age := timestamp.Sub(time.Time(b.Properties.LastModified))
 537                         if b.Properties.BlobType == storage.BlobTypePage &&
 538                                 b.Properties.LeaseState == "available" &&
 539                                 b.Properties.LeaseStatus == "unlocked" &&
 540                                 age.Seconds() > az.azconfig.DeleteDanglingResourcesAfter {
 541
 542                                 az.logger.Printf("Blob %v is unlocked and not modified for %v seconds, will delete", b.Name, age.Seconds())
 543                                 az.deleteBlob <- b
 544                         }
 545                 }
 546                 if response.NextMarker != "" {
 547                         page.Marker = response.NextMarker
 548                 } else {
 549                         break
 550                 }
 551         }
 552 }
 553
 554 func (az *AzureInstanceSet) Stop() {
 555         az.stopFunc()
 556         az.stopWg.Wait()
 557         close(az.deleteNIC)
 558         close(az.deleteBlob)
 559 }
 560
 561 type AzureInstance struct {
 562         provider *AzureInstanceSet
 563         nic      network.Interface
 564         vm       compute.VirtualMachine
 565 }
 566
 567 func (ai *AzureInstance) ID() cloud.InstanceID {
 568         return cloud.InstanceID(*ai.vm.ID)
 569 }
 570
 571 func (ai *AzureInstance) String() string {
 572         return *ai.vm.Name
 573 }
 574
 575 func (ai *AzureInstance) ProviderType() string {
 576         return string(ai.vm.VirtualMachineProperties.HardwareProfile.VMSize)
 577 }
 578
 579 func (ai *AzureInstance) SetTags(newTags cloud.InstanceTags) error {
 580         ai.provider.stopWg.Add(1)
 581         defer ai.provider.stopWg.Done()
 582
 583         tags := make(map[string]*string)
 584
 585         for k, v := range ai.vm.Tags {
 586                 if !strings.HasPrefix(k, "dispatch-") {
 587                         tags[k] = v
 588                 }
 589         }
 590         for k, v := range newTags {
 591                 newstr := v
 592                 tags["dispatch-"+k] = &newstr
 593         }
 594
 595         vmParameters := compute.VirtualMachine{
 596                 Location: &ai.provider.azconfig.Location,
 597                 Tags:     tags,
 598         }
 599         vm, err := ai.provider.vmClient.CreateOrUpdate(ai.provider.ctx, ai.provider.azconfig.ResourceGroup, *ai.vm.Name, vmParameters)
 600         if err != nil {
 601                 return WrapAzureError(err)
 602         }
 603         ai.vm = vm
 604
 605         return nil
 606 }
 607
 608 func (ai *AzureInstance) Tags() cloud.InstanceTags {
 609         tags := make(map[string]string)
 610
 611         for k, v := range ai.vm.Tags {
 612                 if strings.HasPrefix(k, "dispatch-") {
 613                         tags[k[9:]] = *v
 614                 }
 615         }
 616
 617         return tags
 618 }
 619
 620 func (ai *AzureInstance) Destroy() error {
 621         ai.provider.stopWg.Add(1)
 622         defer ai.provider.stopWg.Done()
 623
 624         _, err := ai.provider.vmClient.Delete(ai.provider.ctx, ai.provider.azconfig.ResourceGroup, *ai.vm.Name)
 625         return WrapAzureError(err)
 626 }
 627
 628 func (ai *AzureInstance) Address() string {
 629         return *(*ai.nic.IPConfigurations)[0].PrivateIPAddress
 630 }
 631
 632 func (ai *AzureInstance) VerifyHostKey(receivedKey ssh.PublicKey, client *ssh.Client) error {
 633         ai.provider.stopWg.Add(1)
 634         defer ai.provider.stopWg.Done()
 635
 636         remoteFingerprint := ssh.FingerprintSHA256(receivedKey)
 637
 638         tags := ai.Tags()
 639
 640         tg := tags["ssh-pubkey-fingerprint"]
 641         if tg != "" {
 642                 if remoteFingerprint == tg {
 643                         return nil
 644                 } else {
 645                         return fmt.Errorf("Key fingerprint did not match, expected %q got %q", tg, remoteFingerprint)
 646                 }
 647         }
 648
 649         nodetokenTag := tags["node-token"]
 650         if nodetokenTag == "" {
 651                 return fmt.Errorf("Missing node token tag")
 652         }
 653
 654         sess, err := client.NewSession()
 655         if err != nil {
 656                 return err
 657         }
 658
 659         nodetokenbytes, err := sess.Output("cat /home/crunch/node-token")
 660         if err != nil {
 661                 return err
 662         }
 663
 664         nodetoken := strings.TrimSpace(string(nodetokenbytes))
 665
 666         expectedToken := fmt.Sprintf("%s-%s", *ai.vm.Name, nodetokenTag)
 667
 668         if strings.TrimSpace(nodetoken) != expectedToken {
 669                 return fmt.Errorf("Node token did not match, expected %q got %q", expectedToken, nodetoken)
 670         }
 671
 672         sess, err = client.NewSession()
 673         if err != nil {
 674                 return err
 675         }
 676
 677         keyfingerprintbytes, err := sess.Output("ssh-keygen -E sha256 -l -f /etc/ssh/ssh_host_rsa_key.pub")
 678         if err != nil {
 679                 return err
 680         }
 681
 682         sp := strings.Split(string(keyfingerprintbytes), " ")
 683
 684         if remoteFingerprint != sp[1] {
 685                 return fmt.Errorf("Key fingerprint did not match, expected %q got %q", sp[1], remoteFingerprint)
 686         }
 687
 688         tags["ssh-pubkey-fingerprint"] = sp[1]
 689         delete(tags, "node-token")
 690         ai.SetTags(tags)
 691         return nil
 692 }