lib/dispatchcloud/azure.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package dispatchcloud
   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 AzureProviderConfig 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 AzureProvider struct {
 189         azconfig          AzureProviderConfig
 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 NewAzureProvider(azcfg AzureProviderConfig, dispatcherID string) (prv InstanceProvider, err error) {
 200         ap := AzureProvider{}
 201         err = ap.setup(azcfg, dispatcherID)
 202         if err != nil {
 203                 return nil, err
 204         }
 205         return &ap, nil
 206 }
 207
 208 func (az *AzureProvider) setup(azcfg AzureProviderConfig, dispatcherID string) (err error) {
 209         az.azconfig = azcfg
 210         vmClient := compute.NewVirtualMachinesClient(az.azconfig.SubscriptionID)
 211         netClient := network.NewInterfacesClient(az.azconfig.SubscriptionID)
 212         storageAcctClient := storageacct.NewAccountsClient(az.azconfig.SubscriptionID)
 213
 214         az.azureEnv, err = azure.EnvironmentFromName(az.azconfig.CloudEnv)
 215         if err != nil {
 216                 return err
 217         }
 218
 219         authorizer, err := auth.ClientCredentialsConfig{
 220                 ClientID:     az.azconfig.ClientID,
 221                 ClientSecret: az.azconfig.ClientSecret,
 222                 TenantID:     az.azconfig.TenantID,
 223                 Resource:     az.azureEnv.ResourceManagerEndpoint,
 224                 AADEndpoint:  az.azureEnv.ActiveDirectoryEndpoint,
 225         }.Authorizer()
 226         if err != nil {
 227                 return err
 228         }
 229
 230         vmClient.Authorizer = authorizer
 231         netClient.Authorizer = authorizer
 232         storageAcctClient.Authorizer = authorizer
 233
 234         az.vmClient = &VirtualMachinesClientImpl{vmClient}
 235         az.netClient = &InterfacesClientImpl{netClient}
 236         az.storageAcctClient = storageAcctClient
 237
 238         az.dispatcherID = dispatcherID
 239         az.namePrefix = fmt.Sprintf("compute-%s-", az.dispatcherID)
 240
 241         return nil
 242 }
 243
 244 func (az *AzureProvider) Create(ctx context.Context,
 245         instanceType arvados.InstanceType,
 246         imageId ImageID,
 247         newTags InstanceTags,
 248         publicKey ssh.PublicKey) (Instance, error) {
 249
 250         if len(newTags["node-token"]) == 0 {
 251                 return nil, fmt.Errorf("Must provide tag 'node-token'")
 252         }
 253
 254         name, err := randutil.String(15, "abcdefghijklmnopqrstuvwxyz0123456789")
 255         if err != nil {
 256                 return nil, err
 257         }
 258
 259         name = az.namePrefix + name
 260         log.Printf("name is %v", name)
 261
 262         timestamp := time.Now().Format(time.RFC3339Nano)
 263
 264         tags := make(map[string]*string)
 265         tags["created-at"] = &timestamp
 266         for k, v := range newTags {
 267                 newstr := v
 268                 tags["dispatch-"+k] = &newstr
 269         }
 270
 271         tags["dispatch-instance-type"] = &instanceType.Name
 272
 273         nicParameters := network.Interface{
 274                 Location: &az.azconfig.Location,
 275                 Tags:     tags,
 276                 InterfacePropertiesFormat: &network.InterfacePropertiesFormat{
 277                         IPConfigurations: &[]network.InterfaceIPConfiguration{
 278                                 network.InterfaceIPConfiguration{
 279                                         Name: to.StringPtr("ip1"),
 280                                         InterfaceIPConfigurationPropertiesFormat: &network.InterfaceIPConfigurationPropertiesFormat{
 281                                                 Subnet: &network.Subnet{
 282                                                         ID: to.StringPtr(fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers"+
 283                                                                 "/Microsoft.Network/virtualnetworks/%s/subnets/%s",
 284                                                                 az.azconfig.SubscriptionID,
 285                                                                 az.azconfig.ResourceGroup,
 286                                                                 az.azconfig.Network,
 287                                                                 az.azconfig.Subnet)),
 288                                                 },
 289                                                 PrivateIPAllocationMethod: network.Dynamic,
 290                                         },
 291                                 },
 292                         },
 293                 },
 294         }
 295         nic, err := az.netClient.CreateOrUpdate(ctx, az.azconfig.ResourceGroup, name+"-nic", nicParameters)
 296         if err != nil {
 297                 return nil, WrapAzureError(err)
 298         }
 299
 300         log.Printf("Created NIC %v", *nic.ID)
 301
 302         instance_vhd := fmt.Sprintf("https://%s.blob.%s/%s/%s-os.vhd",
 303                 az.azconfig.StorageAccount,
 304                 az.azureEnv.StorageEndpointSuffix,
 305                 az.azconfig.BlobContainer,
 306                 name)
 307
 308         log.Printf("URI instance vhd %v", instance_vhd)
 309
 310         customData := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf(`#!/bin/sh
 311 echo '%s-%s' > /home/crunch/node-token`, name, newTags["node-token"])))
 312
 313         vmParameters := compute.VirtualMachine{
 314                 Location: &az.azconfig.Location,
 315                 Tags:     tags,
 316                 VirtualMachineProperties: &compute.VirtualMachineProperties{
 317                         HardwareProfile: &compute.HardwareProfile{
 318                                 VMSize: compute.VirtualMachineSizeTypes(instanceType.ProviderType),
 319                         },
 320                         StorageProfile: &compute.StorageProfile{
 321                                 OsDisk: &compute.OSDisk{
 322                                         OsType:       compute.Linux,
 323                                         Name:         to.StringPtr(name + "-os"),
 324                                         CreateOption: compute.FromImage,
 325                                         Image: &compute.VirtualHardDisk{
 326                                                 URI: to.StringPtr(string(imageId)),
 327                                         },
 328                                         Vhd: &compute.VirtualHardDisk{
 329                                                 URI: &instance_vhd,
 330                                         },
 331                                 },
 332                         },
 333                         NetworkProfile: &compute.NetworkProfile{
 334                                 NetworkInterfaces: &[]compute.NetworkInterfaceReference{
 335                                         compute.NetworkInterfaceReference{
 336                                                 ID: nic.ID,
 337                                                 NetworkInterfaceReferenceProperties: &compute.NetworkInterfaceReferenceProperties{
 338                                                         Primary: to.BoolPtr(true),
 339                                                 },
 340                                         },
 341                                 },
 342                         },
 343                         OsProfile: &compute.OSProfile{
 344                                 ComputerName:  &name,
 345                                 AdminUsername: to.StringPtr("crunch"),
 346                                 LinuxConfiguration: &compute.LinuxConfiguration{
 347                                         DisablePasswordAuthentication: to.BoolPtr(true),
 348                                         SSH: &compute.SSHConfiguration{
 349                                                 PublicKeys: &[]compute.SSHPublicKey{
 350                                                         compute.SSHPublicKey{
 351                                                                 Path:    to.StringPtr("/home/crunch/.ssh/authorized_keys"),
 352                                                                 KeyData: to.StringPtr(string(ssh.MarshalAuthorizedKey(publicKey))),
 353                                                         },
 354                                                 },
 355                                         },
 356                                 },
 357                                 CustomData: &customData,
 358                         },
 359                 },
 360         }
 361
 362         vm, err := az.vmClient.CreateOrUpdate(ctx, az.azconfig.ResourceGroup, name, vmParameters)
 363         if err != nil {
 364                 return nil, WrapAzureError(err)
 365         }
 366
 367         return &AzureInstance{
 368                 provider: az,
 369                 nic:      nic,
 370                 vm:       vm,
 371         }, nil
 372 }
 373
 374 func (az *AzureProvider) Instances(ctx context.Context) ([]Instance, error) {
 375         interfaces, err := az.ManageNics(ctx)
 376         if err != nil {
 377                 return nil, err
 378         }
 379
 380         result, err := az.vmClient.ListComplete(ctx, az.azconfig.ResourceGroup)
 381         if err != nil {
 382                 return nil, WrapAzureError(err)
 383         }
 384
 385         instances := make([]Instance, 0)
 386
 387         for ; result.NotDone(); err = result.Next() {
 388                 if err != nil {
 389                         return nil, WrapAzureError(err)
 390                 }
 391                 if strings.HasPrefix(*result.Value().Name, az.namePrefix) {
 392                         instances = append(instances, &AzureInstance{
 393                                 provider: az,
 394                                 vm:       result.Value(),
 395                                 nic:      interfaces[*(*result.Value().NetworkProfile.NetworkInterfaces)[0].ID]})
 396                 }
 397         }
 398         return instances, nil
 399 }
 400
 401 func (az *AzureProvider) ManageNics(ctx context.Context) (map[string]network.Interface, error) {
 402         result, err := az.netClient.ListComplete(ctx, az.azconfig.ResourceGroup)
 403         if err != nil {
 404                 return nil, WrapAzureError(err)
 405         }
 406
 407         interfaces := make(map[string]network.Interface)
 408
 409         timestamp := time.Now()
 410         wg := sync.WaitGroup{}
 411         deletechannel := make(chan string, 20)
 412         defer func() {
 413                 wg.Wait()
 414                 close(deletechannel)
 415         }()
 416         for i := 0; i < 4; i += 1 {
 417                 go func() {
 418                         for {
 419                                 nicname, ok := <-deletechannel
 420                                 if !ok {
 421                                         return
 422                                 }
 423                                 _, delerr := az.netClient.Delete(context.Background(), az.azconfig.ResourceGroup, nicname)
 424                                 if delerr != nil {
 425                                         log.Printf("Error deleting %v: %v", nicname, delerr)
 426                                 } else {
 427                                         log.Printf("Deleted %v", nicname)
 428                                 }
 429                                 wg.Done()
 430                         }
 431                 }()
 432         }
 433
 434         for ; result.NotDone(); err = result.Next() {
 435                 if err != nil {
 436                         log.Printf("Error listing nics: %v", err)
 437                         return interfaces, nil
 438                 }
 439                 if strings.HasPrefix(*result.Value().Name, az.namePrefix) {
 440                         if result.Value().VirtualMachine != nil {
 441                                 interfaces[*result.Value().ID] = result.Value()
 442                         } else {
 443                                 if result.Value().Tags["created-at"] != nil {
 444                                         created_at, err := time.Parse(time.RFC3339Nano, *result.Value().Tags["created-at"])
 445                                         if err == nil {
 446                                                 //log.Printf("found dangling NIC %v created %v seconds ago", *result.Value().Name, timestamp.Sub(created_at).Seconds())
 447                                                 if timestamp.Sub(created_at).Seconds() > az.azconfig.DeleteDanglingResourcesAfter {
 448                                                         log.Printf("Will delete %v because it is older than %v s", *result.Value().Name, az.azconfig.DeleteDanglingResourcesAfter)
 449                                                         wg.Add(1)
 450                                                         deletechannel <- *result.Value().Name
 451                                                 }
 452                                         }
 453                                 }
 454                         }
 455                 }
 456         }
 457         return interfaces, nil
 458 }
 459
 460 func (az *AzureProvider) ManageBlobs(ctx context.Context) {
 461         result, err := az.storageAcctClient.ListKeys(ctx, az.azconfig.ResourceGroup, az.azconfig.StorageAccount)
 462         if err != nil {
 463                 log.Printf("Couldn't get account keys %v", err)
 464                 return
 465         }
 466
 467         key1 := *(*result.Keys)[0].Value
 468         client, err := storage.NewBasicClientOnSovereignCloud(az.azconfig.StorageAccount, key1, az.azureEnv)
 469         if err != nil {
 470                 log.Printf("Couldn't make client %v", err)
 471                 return
 472         }
 473
 474         blobsvc := client.GetBlobService()
 475         blobcont := blobsvc.GetContainerReference(az.azconfig.BlobContainer)
 476
 477         timestamp := time.Now()
 478         wg := sync.WaitGroup{}
 479         deletechannel := make(chan storage.Blob, 20)
 480         defer func() {
 481                 wg.Wait()
 482                 close(deletechannel)
 483         }()
 484         for i := 0; i < 4; i += 1 {
 485                 go func() {
 486                         for {
 487                                 blob, ok := <-deletechannel
 488                                 if !ok {
 489                                         return
 490                                 }
 491                                 err := blob.Delete(nil)
 492                                 if err != nil {
 493                                         log.Printf("error deleting %v: %v", blob.Name, err)
 494                                 } else {
 495                                         log.Printf("Deleted blob %v", blob.Name)
 496                                 }
 497                                 wg.Done()
 498                         }
 499                 }()
 500         }
 501
 502         page := storage.ListBlobsParameters{Prefix: az.namePrefix}
 503
 504         for {
 505                 response, err := blobcont.ListBlobs(page)
 506                 if err != nil {
 507                         log.Printf("Error listing blobs %v", err)
 508                         return
 509                 }
 510                 for _, b := range response.Blobs {
 511                         age := timestamp.Sub(time.Time(b.Properties.LastModified))
 512                         if b.Properties.BlobType == storage.BlobTypePage &&
 513                                 b.Properties.LeaseState == "available" &&
 514                                 b.Properties.LeaseStatus == "unlocked" &&
 515                                 age.Seconds() > az.azconfig.DeleteDanglingResourcesAfter {
 516
 517                                 log.Printf("Blob %v is unlocked and not modified for %v seconds, will delete", b.Name, age.Seconds())
 518                                 wg.Add(1)
 519                                 deletechannel <- b
 520                         }
 521                 }
 522                 if response.NextMarker != "" {
 523                         page.Marker = response.NextMarker
 524                 } else {
 525                         break
 526                 }
 527         }
 528 }
 529
 530 func (az *AzureProvider) Stop() {
 531 }
 532
 533 type AzureInstance struct {
 534         provider *AzureProvider
 535         nic      network.Interface
 536         vm       compute.VirtualMachine
 537 }
 538
 539 func (ai *AzureInstance) ID() InstanceID {
 540         return InstanceID(*ai.vm.ID)
 541 }
 542
 543 func (ai *AzureInstance) String() string {
 544         return *ai.vm.Name
 545 }
 546
 547 func (ai *AzureInstance) SetTags(ctx context.Context, newTags InstanceTags) error {
 548         tags := make(map[string]*string)
 549
 550         for k, v := range ai.vm.Tags {
 551                 if !strings.HasPrefix(k, "dispatch-") {
 552                         tags[k] = v
 553                 }
 554         }
 555         for k, v := range newTags {
 556                 newstr := v
 557                 tags["dispatch-"+k] = &newstr
 558         }
 559
 560         vmParameters := compute.VirtualMachine{
 561                 Location: &ai.provider.azconfig.Location,
 562                 Tags:     tags,
 563         }
 564         vm, err := ai.provider.vmClient.CreateOrUpdate(ctx, ai.provider.azconfig.ResourceGroup, *ai.vm.Name, vmParameters)
 565         if err != nil {
 566                 return WrapAzureError(err)
 567         }
 568         ai.vm = vm
 569
 570         return nil
 571 }
 572
 573 func (ai *AzureInstance) Tags(ctx context.Context) (InstanceTags, error) {
 574         tags := make(map[string]string)
 575
 576         for k, v := range ai.vm.Tags {
 577                 if strings.HasPrefix(k, "dispatch-") {
 578                         tags[k[9:]] = *v
 579                 }
 580         }
 581
 582         return tags, nil
 583 }
 584
 585 func (ai *AzureInstance) Destroy(ctx context.Context) error {
 586         _, err := ai.provider.vmClient.Delete(ctx, ai.provider.azconfig.ResourceGroup, *ai.vm.Name)
 587         return WrapAzureError(err)
 588 }
 589
 590 func (ai *AzureInstance) Address() string {
 591         return *(*ai.nic.IPConfigurations)[0].PrivateIPAddress
 592 }
 593
 594 func (ai *AzureInstance) VerifyPublicKey(ctx context.Context, receivedKey ssh.PublicKey, client *ssh.Client) error {
 595         remoteFingerprint := ssh.FingerprintSHA256(receivedKey)
 596
 597         tags, _ := ai.Tags(ctx)
 598
 599         tg := tags["ssh-pubkey-fingerprint"]
 600         if tg != "" {
 601                 if remoteFingerprint == tg {
 602                         return nil
 603                 } else {
 604                         return fmt.Errorf("Key fingerprint did not match, expected %q got %q", tg, remoteFingerprint)
 605                 }
 606         }
 607
 608         nodetokenTag := tags["node-token"]
 609         if nodetokenTag == "" {
 610                 return fmt.Errorf("Missing node token tag")
 611         }
 612
 613         sess, err := client.NewSession()
 614         if err != nil {
 615                 return err
 616         }
 617
 618         nodetokenbytes, err := sess.Output("cat /home/crunch/node-token")
 619         if err != nil {
 620                 return err
 621         }
 622
 623         nodetoken := strings.TrimSpace(string(nodetokenbytes))
 624
 625         expectedToken := fmt.Sprintf("%s-%s", *ai.vm.Name, nodetokenTag)
 626         log.Printf("%q %q", nodetoken, expectedToken)
 627
 628         if strings.TrimSpace(nodetoken) != expectedToken {
 629                 return fmt.Errorf("Node token did not match, expected %q got %q", expectedToken, nodetoken)
 630         }
 631
 632         sess, err = client.NewSession()
 633         if err != nil {
 634                 return err
 635         }
 636
 637         keyfingerprintbytes, err := sess.Output("ssh-keygen -E sha256 -l -f /etc/ssh/ssh_host_rsa_key.pub")
 638         if err != nil {
 639                 return err
 640         }
 641
 642         sp := strings.Split(string(keyfingerprintbytes), " ")
 643
 644         log.Printf("%q %q", remoteFingerprint, sp[1])
 645
 646         if remoteFingerprint != sp[1] {
 647                 return fmt.Errorf("Key fingerprint did not match, expected %q got %q", sp[1], remoteFingerprint)
 648         }
 649
 650         tags["ssh-pubkey-fingerprint"] = sp[1]
 651         delete(tags, "node-token")
 652         ai.SetTags(ctx, tags)
 653         return nil
 654 }