1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
18 "git.arvados.org/arvados.git/lib/cmd"
19 "git.arvados.org/arvados.git/lib/dispatchcloud"
20 "git.arvados.org/arvados.git/lib/service"
21 "git.arvados.org/arvados.git/sdk/go/arvados"
22 "git.arvados.org/arvados.git/sdk/go/arvadosclient"
23 "git.arvados.org/arvados.git/sdk/go/auth"
24 "git.arvados.org/arvados.git/sdk/go/ctxlog"
25 "git.arvados.org/arvados.git/sdk/go/dispatch"
26 "git.arvados.org/arvados.git/sdk/go/health"
27 "github.com/julienschmidt/httprouter"
28 "github.com/prometheus/client_golang/prometheus"
29 "github.com/prometheus/client_golang/prometheus/promhttp"
30 "github.com/sirupsen/logrus"
33 var DispatchCommand cmd.Handler = service.Command(arvados.ServiceNameDispatchLSF, newHandler)
35 func newHandler(ctx context.Context, cluster *arvados.Cluster, token string, reg *prometheus.Registry) service.Handler {
36 ac, err := arvados.NewClientFromConfig(cluster)
38 return service.ErrorHandler(ctx, cluster, fmt.Errorf("error initializing client from cluster config: %s", err))
51 type dispatcher struct {
52 Cluster *arvados.Cluster
53 Context context.Context
54 ArvClient *arvados.Client
56 Registry *prometheus.Registry
58 logger logrus.FieldLogger
61 arvDispatcher *dispatch.Dispatcher
62 httpHandler http.Handler
69 // Start starts the dispatcher. Start can be called multiple times
70 // with no ill effect.
71 func (disp *dispatcher) Start() {
72 disp.initOnce.Do(func() {
75 disp.checkLsfQueueForOrphans()
76 err := disp.arvDispatcher.Run(disp.Context)
78 disp.logger.Error(err)
85 // ServeHTTP implements service.Handler.
86 func (disp *dispatcher) ServeHTTP(w http.ResponseWriter, r *http.Request) {
88 disp.httpHandler.ServeHTTP(w, r)
91 // CheckHealth implements service.Handler.
92 func (disp *dispatcher) CheckHealth() error {
96 return errors.New("stopped")
102 // Done implements service.Handler.
103 func (disp *dispatcher) Done() <-chan struct{} {
107 // Stop dispatching containers and release resources. Used by tests.
108 func (disp *dispatcher) Close() {
111 case disp.stop <- struct{}{}:
117 func (disp *dispatcher) init() {
118 disp.logger = ctxlog.FromContext(disp.Context)
119 disp.lsfcli.logger = disp.logger
120 disp.lsfqueue = lsfqueue{
122 period: time.Duration(disp.Cluster.Containers.CloudVMs.PollInterval),
123 lsfcli: &disp.lsfcli,
125 disp.ArvClient.AuthToken = disp.AuthToken
126 disp.stop = make(chan struct{}, 1)
127 disp.stopped = make(chan struct{})
129 arv, err := arvadosclient.New(disp.ArvClient)
131 disp.logger.Fatalf("Error making Arvados client: %v", err)
134 arv.ApiToken = disp.AuthToken
135 disp.arvDispatcher = &dispatch.Dispatcher{
138 BatchSize: disp.Cluster.API.MaxItemsPerResponse,
139 RunContainer: disp.runContainer,
140 PollPeriod: time.Duration(disp.Cluster.Containers.CloudVMs.PollInterval),
141 MinRetryPeriod: time.Duration(disp.Cluster.Containers.MinRetryPeriod),
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)
149 mux := httprouter.New()
150 metricsH := promhttp.HandlerFor(disp.Registry, promhttp.HandlerOpts{
151 ErrorLog: disp.logger,
153 mux.Handler("GET", "/metrics", metricsH)
154 mux.Handler("GET", "/metrics.json", metricsH)
155 mux.Handler("GET", "/_health/:check", &health.Handler{
156 Token: disp.Cluster.ManagementToken,
158 Routes: health.Routes{"ping": disp.CheckHealth},
160 disp.httpHandler = auth.RequireLiteralToken(disp.Cluster.ManagementToken, mux)
164 func (disp *dispatcher) runContainer(_ *dispatch.Dispatcher, ctr arvados.Container, status <-chan arvados.Container) {
165 ctx, cancel := context.WithCancel(disp.Context)
168 if ctr.State != dispatch.Locked {
169 // already started by prior invocation
170 } else if _, ok := disp.lsfqueue.JobID(ctr.UUID); !ok {
171 disp.logger.Printf("Submitting container %s to LSF", ctr.UUID)
172 cmd := []string{disp.Cluster.Containers.CrunchRunCommand}
173 cmd = append(cmd, "--runtime-engine="+disp.Cluster.Containers.RuntimeEngine)
174 cmd = append(cmd, disp.Cluster.Containers.CrunchRunArgumentsList...)
175 if err := disp.submit(ctr, cmd); err != nil {
177 switch err := err.(type) {
178 case dispatchcloud.ConstraintsNotSatisfiableError:
179 var logBuf bytes.Buffer
180 fmt.Fprintf(&logBuf, "cannot run container %s: %s\n", ctr.UUID, err)
181 if len(err.AvailableTypes) == 0 {
182 fmt.Fprint(&logBuf, "No instance types are configured.\n")
184 fmt.Fprint(&logBuf, "Available instance types:\n")
185 for _, t := range err.AvailableTypes {
187 "Type %q: %d VCPUs, %d RAM, %d Scratch, %f Price\n",
188 t.Name, t.VCPUs, t.RAM, t.Scratch, t.Price,
192 text = logBuf.String()
193 disp.arvDispatcher.UpdateState(ctr.UUID, dispatch.Cancelled)
195 text = fmt.Sprintf("Error submitting container %s to LSF: %s", ctr.UUID, err)
197 disp.logger.Print(text)
199 lr := arvadosclient.Dict{"log": arvadosclient.Dict{
200 "object_uuid": ctr.UUID,
201 "event_type": "dispatch",
202 "properties": map[string]string{"text": text}}}
203 disp.arvDispatcher.Arv.Create("logs", lr, nil)
205 disp.arvDispatcher.Unlock(ctr.UUID)
210 disp.logger.Printf("Start monitoring container %v in state %q", ctr.UUID, ctr.State)
211 defer disp.logger.Printf("Done monitoring container %s", ctr.UUID)
213 // If the container disappears from the lsf queue, there is
214 // no point in waiting for further dispatch updates: just
215 // clean up and return.
216 go func(uuid string) {
217 for ctx.Err() == nil {
218 if _, ok := disp.lsfqueue.JobID(uuid); !ok {
219 disp.logger.Printf("container %s job disappeared from LSF queue", uuid)
226 for done := false; !done; {
229 // Disappeared from lsf queue
230 if err := disp.arvDispatcher.Arv.Get("containers", ctr.UUID, nil, &ctr); err != nil {
231 disp.logger.Printf("error getting final container state for %s: %s", ctr.UUID, err)
234 case dispatch.Running:
235 disp.arvDispatcher.UpdateState(ctr.UUID, dispatch.Cancelled)
236 case dispatch.Locked:
237 disp.arvDispatcher.Unlock(ctr.UUID)
240 case updated, ok := <-status:
245 if updated.State != ctr.State {
246 disp.logger.Infof("container %s changed state from %s to %s", ctr.UUID, ctr.State, updated.State)
249 if ctr.Priority == 0 {
250 disp.logger.Printf("container %s has state %s, priority %d: cancel lsf job", ctr.UUID, ctr.State, ctr.Priority)
253 disp.lsfqueue.SetPriority(ctr.UUID, int64(ctr.Priority))
257 disp.logger.Printf("container %s is done", ctr.UUID)
259 // Try "bkill" every few seconds until the LSF job disappears
261 ticker := time.NewTicker(5 * time.Second)
263 for jobid, ok := disp.lsfqueue.JobID(ctr.UUID); ok; _, ok = disp.lsfqueue.JobID(ctr.UUID) {
264 err := disp.lsfcli.Bkill(jobid)
266 disp.logger.Warnf("%s: bkill(%d): %s", ctr.UUID, jobid, err)
272 func (disp *dispatcher) submit(container arvados.Container, crunchRunCommand []string) error {
273 // Start with an empty slice here to ensure append() doesn't
274 // modify crunchRunCommand's underlying array
276 crArgs = append(crArgs, crunchRunCommand...)
277 crArgs = append(crArgs, container.UUID)
278 crScript := execScript(crArgs)
280 bsubArgs, err := disp.bsubArgs(container)
284 return disp.lsfcli.Bsub(crScript, bsubArgs, disp.ArvClient)
287 func (disp *dispatcher) bkill(ctr arvados.Container) {
288 if jobid, ok := disp.lsfqueue.JobID(ctr.UUID); !ok {
289 disp.logger.Debugf("bkill(%s): redundant, job not in queue", ctr.UUID)
290 } else if err := disp.lsfcli.Bkill(jobid); err != nil {
291 disp.logger.Warnf("%s: bkill(%d): %s", ctr.UUID, jobid, err)
295 func (disp *dispatcher) bsubArgs(container arvados.Container) ([]string, error) {
296 args := []string{"bsub"}
297 args = append(args, disp.Cluster.Containers.LSF.BsubArgumentsList...)
298 args = append(args, "-J", container.UUID)
299 args = append(args, disp.bsubConstraintArgs(container)...)
300 if u := disp.Cluster.Containers.LSF.BsubSudoUser; u != "" {
301 args = append([]string{"sudo", "-E", "-u", u}, args...)
306 func (disp *dispatcher) bsubConstraintArgs(container arvados.Container) []string {
307 // TODO: propagate container.SchedulingParameters.Partitions
308 tmp := int64(math.Ceil(float64(dispatchcloud.EstimateScratchSpace(&container)) / 1048576))
309 vcpus := container.RuntimeConstraints.VCPUs
310 mem := int64(math.Ceil(float64(container.RuntimeConstraints.RAM+
311 container.RuntimeConstraints.KeepCacheRAM+
312 int64(disp.Cluster.Containers.ReserveExtraRAM)) / 1048576))
314 "-R", fmt.Sprintf("rusage[mem=%dMB:tmp=%dMB] affinity[core(%d)]", mem, tmp, vcpus),
318 func (disp *dispatcher) checkLsfQueueForOrphans() {
319 disp.logger.Warn("FIXME: checkLsfQueueForOrphans")
322 func execScript(args []string) []byte {
323 s := "#!/bin/sh\nexec"
324 for _, w := range args {
326 s += strings.Replace(w, `'`, `'\''`, -1)
329 return []byte(s + "\n")