lib/cloud/loopback/loopback.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package loopback
   6
   7 import (
   8         "bytes"
   9         "crypto/rand"
  10         "crypto/rsa"
  11         "encoding/json"
  12         "errors"
  13         "io"
  14         "os"
  15         "os/exec"
  16         "os/user"
  17         "strings"
  18         "sync"
  19         "syscall"
  20
  21         "git.arvados.org/arvados.git/lib/cloud"
  22         "git.arvados.org/arvados.git/lib/dispatchcloud/test"
  23         "git.arvados.org/arvados.git/sdk/go/arvados"
  24         "github.com/sirupsen/logrus"
  25         "golang.org/x/crypto/ssh"
  26 )
  27
  28 // Driver is the loopback implementation of the cloud.Driver interface.
  29 var Driver = cloud.DriverFunc(newInstanceSet)
  30
  31 var (
  32         errUnimplemented = errors.New("function not implemented by loopback driver")
  33         errQuota         = quotaError("loopback driver is always at quota")
  34 )
  35
  36 type quotaError string
  37
  38 func (e quotaError) IsQuotaError() bool { return true }
  39 func (e quotaError) Error() string      { return string(e) }
  40
  41 type instanceSet struct {
  42         instanceSetID cloud.InstanceSetID
  43         logger        logrus.FieldLogger
  44         instances     []*instance
  45         mtx           sync.Mutex
  46 }
  47
  48 func newInstanceSet(config json.RawMessage, instanceSetID cloud.InstanceSetID, _ cloud.SharedResourceTags, logger logrus.FieldLogger) (cloud.InstanceSet, error) {
  49         is := &instanceSet{
  50                 instanceSetID: instanceSetID,
  51                 logger:        logger,
  52         }
  53         return is, nil
  54 }
  55
  56 func (is *instanceSet) Create(it arvados.InstanceType, _ cloud.ImageID, tags cloud.InstanceTags, _ cloud.InitCommand, pubkey ssh.PublicKey) (cloud.Instance, error) {
  57         is.mtx.Lock()
  58         defer is.mtx.Unlock()
  59         if len(is.instances) > 0 {
  60                 return nil, errQuota
  61         }
  62         // A crunch-run process running in a previous instance may
  63         // have marked the node as broken. In the loopback scenario a
  64         // destroy+create cycle doesn't fix whatever was broken -- but
  65         // nothing else will either, so the best we can do is remove
  66         // the "broken" flag and try again.
  67         if err := os.Remove("/var/lock/crunch-run-broken"); err == nil {
  68                 is.logger.Info("removed /var/lock/crunch-run-broken")
  69         } else if !errors.Is(err, os.ErrNotExist) {
  70                 return nil, err
  71         }
  72         u, err := user.Current()
  73         if err != nil {
  74                 return nil, err
  75         }
  76         hostRSAKey, err := rsa.GenerateKey(rand.Reader, 1024)
  77         if err != nil {
  78                 return nil, err
  79         }
  80         hostKey, err := ssh.NewSignerFromKey(hostRSAKey)
  81         if err != nil {
  82                 return nil, err
  83         }
  84         hostPubKey, err := ssh.NewPublicKey(hostRSAKey.Public())
  85         if err != nil {
  86                 return nil, err
  87         }
  88         inst := &instance{
  89                 is:           is,
  90                 instanceType: it,
  91                 adminUser:    u.Username,
  92                 tags:         tags,
  93                 hostPubKey:   hostPubKey,
  94                 sshService: test.SSHService{
  95                         HostKey:        hostKey,
  96                         AuthorizedUser: u.Username,
  97                         AuthorizedKeys: []ssh.PublicKey{pubkey},
  98                 },
  99         }
 100         inst.sshService.Exec = inst.sshExecFunc
 101         go inst.sshService.Start()
 102         is.instances = []*instance{inst}
 103         return inst, nil
 104 }
 105
 106 func (is *instanceSet) Instances(cloud.InstanceTags) ([]cloud.Instance, error) {
 107         is.mtx.Lock()
 108         defer is.mtx.Unlock()
 109         var ret []cloud.Instance
 110         for _, inst := range is.instances {
 111                 ret = append(ret, inst)
 112         }
 113         return ret, nil
 114 }
 115
 116 func (is *instanceSet) Stop() {
 117         is.mtx.Lock()
 118         defer is.mtx.Unlock()
 119         for _, inst := range is.instances {
 120                 inst.sshService.Close()
 121         }
 122 }
 123
 124 type instance struct {
 125         is           *instanceSet
 126         instanceType arvados.InstanceType
 127         adminUser    string
 128         tags         cloud.InstanceTags
 129         hostPubKey   ssh.PublicKey
 130         sshService   test.SSHService
 131 }
 132
 133 func (i *instance) ID() cloud.InstanceID                                    { return cloud.InstanceID(i.instanceType.ProviderType) }
 134 func (i *instance) String() string                                          { return i.instanceType.ProviderType }
 135 func (i *instance) ProviderType() string                                    { return i.instanceType.ProviderType }
 136 func (i *instance) Address() string                                         { return i.sshService.Address() }
 137 func (i *instance) PriceHistory(arvados.InstanceType) []cloud.InstancePrice { return nil }
 138 func (i *instance) RemoteUser() string                                      { return i.adminUser }
 139 func (i *instance) Tags() cloud.InstanceTags                                { return i.tags }
 140 func (i *instance) SetTags(tags cloud.InstanceTags) error {
 141         i.tags = tags
 142         return nil
 143 }
 144 func (i *instance) Destroy() error {
 145         i.is.mtx.Lock()
 146         defer i.is.mtx.Unlock()
 147         i.is.instances = i.is.instances[:0]
 148         return nil
 149 }
 150 func (i *instance) VerifyHostKey(pubkey ssh.PublicKey, _ *ssh.Client) error {
 151         if !bytes.Equal(pubkey.Marshal(), i.hostPubKey.Marshal()) {
 152                 return errors.New("host key mismatch")
 153         }
 154         return nil
 155 }
 156 func (i *instance) sshExecFunc(env map[string]string, command string, stdin io.Reader, stdout, stderr io.Writer) uint32 {
 157         cmd := exec.Command("sh", "-c", strings.TrimPrefix(command, "sudo "))
 158         cmd.Stdin = stdin
 159         cmd.Stdout = stdout
 160         cmd.Stderr = stderr
 161         for k, v := range env {
 162                 cmd.Env = append(cmd.Env, k+"="+v)
 163         }
 164         // Prevent child process from using our tty.
 165         cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
 166         err := cmd.Run()
 167         if err == nil {
 168                 return 0
 169         } else if err, ok := err.(*exec.ExitError); !ok {
 170                 return 1
 171         } else if code := err.ExitCode(); code < 0 {
 172                 return 1
 173         } else {
 174                 return uint32(code)
 175         }
 176 }