lib/dispatchcloud/dispatcher.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         "crypto/md5"
  10         "encoding/json"
  11         "fmt"
  12         "net/http"
  13         "strings"
  14         "sync"
  15         "time"
  16
  17         "git.curoverse.com/arvados.git/lib/cloud"
  18         "git.curoverse.com/arvados.git/lib/dispatchcloud/container"
  19         "git.curoverse.com/arvados.git/lib/dispatchcloud/scheduler"
  20         "git.curoverse.com/arvados.git/lib/dispatchcloud/ssh_executor"
  21         "git.curoverse.com/arvados.git/lib/dispatchcloud/worker"
  22         "git.curoverse.com/arvados.git/sdk/go/arvados"
  23         "git.curoverse.com/arvados.git/sdk/go/auth"
  24         "git.curoverse.com/arvados.git/sdk/go/ctxlog"
  25         "git.curoverse.com/arvados.git/sdk/go/httpserver"
  26         "github.com/julienschmidt/httprouter"
  27         "github.com/prometheus/client_golang/prometheus"
  28         "github.com/prometheus/client_golang/prometheus/promhttp"
  29         "github.com/sirupsen/logrus"
  30         "golang.org/x/crypto/ssh"
  31 )
  32
  33 const (
  34         defaultPollInterval     = time.Second
  35         defaultStaleLockTimeout = time.Minute
  36 )
  37
  38 type pool interface {
  39         scheduler.WorkerPool
  40         Instances() []worker.InstanceView
  41         SetIdleBehavior(cloud.InstanceID, worker.IdleBehavior) error
  42         KillInstance(id cloud.InstanceID, reason string) error
  43         Stop()
  44 }
  45
  46 type dispatcher struct {
  47         Cluster       *arvados.Cluster
  48         Context       context.Context
  49         ArvClient     *arvados.Client
  50         AuthToken     string
  51         InstanceSetID cloud.InstanceSetID
  52
  53         logger      logrus.FieldLogger
  54         reg         *prometheus.Registry
  55         instanceSet cloud.InstanceSet
  56         pool        pool
  57         queue       scheduler.ContainerQueue
  58         httpHandler http.Handler
  59         sshKey      ssh.Signer
  60
  61         setupOnce sync.Once
  62         stop      chan struct{}
  63         stopped   chan struct{}
  64 }
  65
  66 // Start starts the dispatcher. Start can be called multiple times
  67 // with no ill effect.
  68 func (disp *dispatcher) Start() {
  69         disp.setupOnce.Do(disp.setup)
  70 }
  71
  72 // ServeHTTP implements service.Handler.
  73 func (disp *dispatcher) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  74         disp.Start()
  75         disp.httpHandler.ServeHTTP(w, r)
  76 }
  77
  78 // CheckHealth implements service.Handler.
  79 func (disp *dispatcher) CheckHealth() error {
  80         disp.Start()
  81         return nil
  82 }
  83
  84 // Stop dispatching containers and release resources. Typically used
  85 // in tests.
  86 func (disp *dispatcher) Close() {
  87         disp.Start()
  88         select {
  89         case disp.stop <- struct{}{}:
  90         default:
  91         }
  92         <-disp.stopped
  93 }
  94
  95 // Make a worker.Executor for the given instance.
  96 func (disp *dispatcher) newExecutor(inst cloud.Instance) worker.Executor {
  97         exr := ssh_executor.New(inst)
  98         exr.SetTargetPort(disp.Cluster.Containers.CloudVMs.SSHPort)
  99         exr.SetSigners(disp.sshKey)
 100         return exr
 101 }
 102
 103 func (disp *dispatcher) typeChooser(ctr *arvados.Container) (arvados.InstanceType, error) {
 104         return ChooseInstanceType(disp.Cluster, ctr)
 105 }
 106
 107 func (disp *dispatcher) setup() {
 108         disp.initialize()
 109         go disp.run()
 110 }
 111
 112 func (disp *dispatcher) initialize() {
 113         disp.logger = ctxlog.FromContext(disp.Context)
 114
 115         disp.ArvClient.AuthToken = disp.AuthToken
 116
 117         if disp.InstanceSetID == "" {
 118                 if strings.HasPrefix(disp.AuthToken, "v2/") {
 119                         disp.InstanceSetID = cloud.InstanceSetID(strings.Split(disp.AuthToken, "/")[1])
 120                 } else {
 121                         // Use some other string unique to this token
 122                         // that doesn't reveal the token itself.
 123                         disp.InstanceSetID = cloud.InstanceSetID(fmt.Sprintf("%x", md5.Sum([]byte(disp.AuthToken))))
 124                 }
 125         }
 126         disp.stop = make(chan struct{}, 1)
 127         disp.stopped = make(chan struct{})
 128
 129         if key, err := ssh.ParsePrivateKey([]byte(disp.Cluster.Containers.DispatchPrivateKey)); err != nil {
 130                 disp.logger.Fatalf("error parsing configured Containers.DispatchPrivateKey: %s", err)
 131         } else {
 132                 disp.sshKey = key
 133         }
 134
 135         disp.reg = prometheus.NewRegistry()
 136         instanceSet, err := newInstanceSet(disp.Cluster, disp.InstanceSetID, disp.logger, disp.reg)
 137         if err != nil {
 138                 disp.logger.Fatalf("error initializing driver: %s", err)
 139         }
 140         disp.instanceSet = instanceSet
 141         disp.pool = worker.NewPool(disp.logger, disp.ArvClient, disp.reg, disp.InstanceSetID, disp.instanceSet, disp.newExecutor, disp.sshKey.PublicKey(), disp.Cluster)
 142         disp.queue = container.NewQueue(disp.logger, disp.reg, disp.typeChooser, disp.ArvClient)
 143
 144         if disp.Cluster.ManagementToken == "" {
 145                 disp.httpHandler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 146                         http.Error(w, "Management API authentication is not configured", http.StatusForbidden)
 147                 })
 148         } else {
 149                 mux := httprouter.New()
 150                 mux.HandlerFunc("GET", "/arvados/v1/dispatch/containers", disp.apiContainers)
 151                 mux.HandlerFunc("POST", "/arvados/v1/dispatch/containers/kill", disp.apiInstanceKill)
 152                 mux.HandlerFunc("GET", "/arvados/v1/dispatch/instances", disp.apiInstances)
 153                 mux.HandlerFunc("POST", "/arvados/v1/dispatch/instances/hold", disp.apiInstanceHold)
 154                 mux.HandlerFunc("POST", "/arvados/v1/dispatch/instances/drain", disp.apiInstanceDrain)
 155                 mux.HandlerFunc("POST", "/arvados/v1/dispatch/instances/run", disp.apiInstanceRun)
 156                 mux.HandlerFunc("POST", "/arvados/v1/dispatch/instances/kill", disp.apiInstanceKill)
 157                 metricsH := promhttp.HandlerFor(disp.reg, promhttp.HandlerOpts{
 158                         ErrorLog: disp.logger,
 159                 })
 160                 mux.Handler("GET", "/metrics", metricsH)
 161                 mux.Handler("GET", "/metrics.json", metricsH)
 162                 disp.httpHandler = auth.RequireLiteralToken(disp.Cluster.ManagementToken, mux)
 163         }
 164 }
 165
 166 func (disp *dispatcher) run() {
 167         defer close(disp.stopped)
 168         defer disp.instanceSet.Stop()
 169         defer disp.pool.Stop()
 170
 171         staleLockTimeout := time.Duration(disp.Cluster.Containers.StaleLockTimeout)
 172         if staleLockTimeout == 0 {
 173                 staleLockTimeout = defaultStaleLockTimeout
 174         }
 175         pollInterval := time.Duration(disp.Cluster.Containers.CloudVMs.PollInterval)
 176         if pollInterval <= 0 {
 177                 pollInterval = defaultPollInterval
 178         }
 179         sched := scheduler.New(disp.Context, disp.queue, disp.pool, staleLockTimeout, pollInterval)
 180         sched.Start()
 181         defer sched.Stop()
 182
 183         <-disp.stop
 184 }
 185
 186 // Management API: all active and queued containers.
 187 func (disp *dispatcher) apiContainers(w http.ResponseWriter, r *http.Request) {
 188         var resp struct {
 189                 Items []container.QueueEnt `json:"items"`
 190         }
 191         qEntries, _ := disp.queue.Entries()
 192         for _, ent := range qEntries {
 193                 resp.Items = append(resp.Items, ent)
 194         }
 195         json.NewEncoder(w).Encode(resp)
 196 }
 197
 198 // Management API: all active instances (cloud VMs).
 199 func (disp *dispatcher) apiInstances(w http.ResponseWriter, r *http.Request) {
 200         var resp struct {
 201                 Items []worker.InstanceView `json:"items"`
 202         }
 203         resp.Items = disp.pool.Instances()
 204         json.NewEncoder(w).Encode(resp)
 205 }
 206
 207 // Management API: set idle behavior to "hold" for specified instance.
 208 func (disp *dispatcher) apiInstanceHold(w http.ResponseWriter, r *http.Request) {
 209         disp.apiInstanceIdleBehavior(w, r, worker.IdleBehaviorHold)
 210 }
 211
 212 // Management API: set idle behavior to "drain" for specified instance.
 213 func (disp *dispatcher) apiInstanceDrain(w http.ResponseWriter, r *http.Request) {
 214         disp.apiInstanceIdleBehavior(w, r, worker.IdleBehaviorDrain)
 215 }
 216
 217 // Management API: set idle behavior to "run" for specified instance.
 218 func (disp *dispatcher) apiInstanceRun(w http.ResponseWriter, r *http.Request) {
 219         disp.apiInstanceIdleBehavior(w, r, worker.IdleBehaviorRun)
 220 }
 221
 222 // Management API: shutdown/destroy specified instance now.
 223 func (disp *dispatcher) apiInstanceKill(w http.ResponseWriter, r *http.Request) {
 224         id := cloud.InstanceID(r.FormValue("instance_id"))
 225         if id == "" {
 226                 httpserver.Error(w, "instance_id parameter not provided", http.StatusBadRequest)
 227                 return
 228         }
 229         err := disp.pool.KillInstance(id, "via management API: "+r.FormValue("reason"))
 230         if err != nil {
 231                 httpserver.Error(w, err.Error(), http.StatusNotFound)
 232                 return
 233         }
 234 }
 235
 236 // Management API: send SIGTERM to specified container's crunch-run
 237 // process now.
 238 func (disp *dispatcher) apiContainerKill(w http.ResponseWriter, r *http.Request) {
 239         uuid := r.FormValue("container_uuid")
 240         if uuid == "" {
 241                 httpserver.Error(w, "container_uuid parameter not provided", http.StatusBadRequest)
 242                 return
 243         }
 244         if !disp.pool.KillContainer(uuid, "via management API: "+r.FormValue("reason")) {
 245                 httpserver.Error(w, "container not found", http.StatusNotFound)
 246                 return
 247         }
 248 }
 249
 250 func (disp *dispatcher) apiInstanceIdleBehavior(w http.ResponseWriter, r *http.Request, want worker.IdleBehavior) {
 251         id := cloud.InstanceID(r.FormValue("instance_id"))
 252         if id == "" {
 253                 httpserver.Error(w, "instance_id parameter not provided", http.StatusBadRequest)
 254                 return
 255         }
 256         err := disp.pool.SetIdleBehavior(id, want)
 257         if err != nil {
 258                 httpserver.Error(w, err.Error(), http.StatusNotFound)
 259                 return
 260         }
 261 }