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