X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e31979e7ef752c9adef3d3fa6096a47d720d524e..a9be3117466506dffc39617be1c58406c5914e4b:/lib/lsf/dispatch.go?ds=sidebyside

diff --git a/lib/lsf/dispatch.go b/lib/lsf/dispatch.go
index 6e35b7de92..e2348337e6 100644
--- a/lib/lsf/dispatch.go
+++ b/lib/lsf/dispatch.go
@@ -6,6 +6,8 @@ package lsf
 
 import (
 	"context"
+	"crypto/hmac"
+	"crypto/sha256"
 	"errors"
 	"fmt"
 	"math"
@@ -119,7 +121,7 @@ func (disp *dispatcher) init() {
 	disp.lsfcli.logger = disp.logger
 	disp.lsfqueue = lsfqueue{
 		logger: disp.logger,
-		period: time.Duration(disp.Cluster.Containers.CloudVMs.PollInterval),
+		period: disp.Cluster.Containers.CloudVMs.PollInterval.Duration(),
 		lsfcli: &disp.lsfcli,
 	}
 	disp.ArvClient.AuthToken = disp.AuthToken
@@ -167,7 +169,7 @@ func (disp *dispatcher) runContainer(_ *dispatch.Dispatcher, ctr arvados.Contain
 
 	if ctr.State != dispatch.Locked {
 		// already started by prior invocation
-	} else if _, ok := disp.lsfqueue.JobID(ctr.UUID); !ok {
+	} else if _, ok := disp.lsfqueue.Lookup(ctr.UUID); !ok {
 		disp.logger.Printf("Submitting container %s to LSF", ctr.UUID)
 		cmd := []string{disp.Cluster.Containers.CrunchRunCommand}
 		cmd = append(cmd, "--runtime-engine="+disp.Cluster.Containers.RuntimeEngine)
@@ -181,16 +183,38 @@ func (disp *dispatcher) runContainer(_ *dispatch.Dispatcher, ctr arvados.Contain
 	disp.logger.Printf("Start monitoring container %v in state %q", ctr.UUID, ctr.State)
 	defer disp.logger.Printf("Done monitoring container %s", ctr.UUID)
 
-	// If the container disappears from the lsf queue, there is
-	// no point in waiting for further dispatch updates: just
-	// clean up and return.
 	go func(uuid string) {
+		cancelled := false
 		for ctx.Err() == nil {
-			if _, ok := disp.lsfqueue.JobID(uuid); !ok {
+			qent, ok := disp.lsfqueue.Lookup(uuid)
+			if !ok {
+				// If the container disappears from
+				// the lsf queue, there is no point in
+				// waiting for further dispatch
+				// updates: just clean up and return.
 				disp.logger.Printf("container %s job disappeared from LSF queue", uuid)
 				cancel()
 				return
 			}
+			if !cancelled && qent.Stat == "PEND" && strings.Contains(qent.PendReason, "There are no suitable hosts for the job") {
+				disp.logger.Printf("container %s: %s", uuid, qent.PendReason)
+				err := disp.arvDispatcher.Arv.Update("containers", uuid, arvadosclient.Dict{
+					"container": map[string]interface{}{
+						"runtime_status": map[string]string{
+							"error": qent.PendReason,
+						},
+					},
+				}, nil)
+				if err != nil {
+					disp.logger.Printf("error setting runtime_status on %s: %s", uuid, err)
+					continue // retry
+				}
+				err = disp.arvDispatcher.UpdateState(uuid, dispatch.Cancelled)
+				if err != nil {
+					continue // retry (UpdateState() already logged the error)
+				}
+				cancelled = true
+			}
 		}
 	}(ctr.UUID)
 
@@ -234,12 +258,12 @@ func (disp *dispatcher) runContainer(_ *dispatch.Dispatcher, ctr arvados.Contain
 
 	// Try "bkill" every few seconds until the LSF job disappears
 	// from the queue.
-	ticker := time.NewTicker(5 * time.Second)
+	ticker := time.NewTicker(disp.Cluster.Containers.CloudVMs.PollInterval.Duration() / 2)
 	defer ticker.Stop()
-	for jobid, ok := disp.lsfqueue.JobID(ctr.UUID); ok; _, ok = disp.lsfqueue.JobID(ctr.UUID) {
-		err := disp.lsfcli.Bkill(jobid)
+	for qent, ok := disp.lsfqueue.Lookup(ctr.UUID); ok; _, ok = disp.lsfqueue.Lookup(ctr.UUID) {
+		err := disp.lsfcli.Bkill(qent.ID)
 		if err != nil {
-			disp.logger.Warnf("%s: bkill(%d): %s", ctr.UUID, jobid, err)
+			disp.logger.Warnf("%s: bkill(%s): %s", ctr.UUID, qent.ID, err)
 		}
 		<-ticker.C
 	}
@@ -252,7 +276,12 @@ func (disp *dispatcher) submit(container arvados.Container, crunchRunCommand []s
 	var crArgs []string
 	crArgs = append(crArgs, crunchRunCommand...)
 	crArgs = append(crArgs, container.UUID)
-	crScript := execScript(crArgs)
+
+	h := hmac.New(sha256.New, []byte(disp.Cluster.SystemRootToken))
+	fmt.Fprint(h, container.UUID)
+	authsecret := fmt.Sprintf("%x", h.Sum(nil))
+
+	crScript := execScript(crArgs, map[string]string{"GatewayAuthSecret": authsecret})
 
 	bsubArgs, err := disp.bsubArgs(container)
 	if err != nil {
@@ -262,10 +291,10 @@ func (disp *dispatcher) submit(container arvados.Container, crunchRunCommand []s
 }
 
 func (disp *dispatcher) bkill(ctr arvados.Container) {
-	if jobid, ok := disp.lsfqueue.JobID(ctr.UUID); !ok {
+	if qent, ok := disp.lsfqueue.Lookup(ctr.UUID); !ok {
 		disp.logger.Debugf("bkill(%s): redundant, job not in queue", ctr.UUID)
-	} else if err := disp.lsfcli.Bkill(jobid); err != nil {
-		disp.logger.Warnf("%s: bkill(%d): %s", ctr.UUID, jobid, err)
+	} else if err := disp.lsfcli.Bkill(qent.ID); err != nil {
+		disp.logger.Warnf("%s: bkill(%s): %s", ctr.UUID, qent.ID, err)
 	}
 }
 
@@ -284,11 +313,16 @@ func (disp *dispatcher) bsubArgs(container arvados.Container) ([]string, error)
 		"%M": fmt.Sprintf("%d", mem),
 		"%T": fmt.Sprintf("%d", tmp),
 		"%U": container.UUID,
+		"%G": fmt.Sprintf("%d", container.RuntimeConstraints.CUDA.DeviceCount),
 	}
 
 	re := regexp.MustCompile(`%.`)
 	var substitutionErrors string
-	for _, a := range disp.Cluster.Containers.LSF.BsubArgumentsList {
+	argumentTemplate := disp.Cluster.Containers.LSF.BsubArgumentsList
+	if container.RuntimeConstraints.CUDA.DeviceCount > 0 {
+		argumentTemplate = append(argumentTemplate, disp.Cluster.Containers.LSF.BsubCUDAArguments...)
+	}
+	for _, a := range argumentTemplate {
 		args = append(args, re.ReplaceAllStringFunc(a, func(s string) string {
 			subst := repl[s]
 			if len(subst) == 0 {
@@ -326,8 +360,14 @@ func (disp *dispatcher) checkLsfQueueForOrphans() {
 	}
 }
 
-func execScript(args []string) []byte {
-	s := "#!/bin/sh\nexec"
+func execScript(args []string, env map[string]string) []byte {
+	s := "#!/bin/sh\n"
+	for k, v := range env {
+		s += k + `='`
+		s += strings.Replace(v, `'`, `'\''`, -1)
+		s += `' `
+	}
+	s += `exec`
 	for _, w := range args {
 		s += ` '`
 		s += strings.Replace(w, `'`, `'\''`, -1)