lib/cloud/azure.go

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