X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/f739d736bbb60a8463f04f5d56c18d09157d820e..10440ac12d6771ab80469adf551d2cac8d3461e6:/services/crunch-dispatch-local/crunch-dispatch-local.go diff --git a/services/crunch-dispatch-local/crunch-dispatch-local.go b/services/crunch-dispatch-local/crunch-dispatch-local.go index fc10393626..e455981891 100644 --- a/services/crunch-dispatch-local/crunch-dispatch-local.go +++ b/services/crunch-dispatch-local/crunch-dispatch-local.go @@ -17,34 +17,29 @@ import ( "syscall" "time" - "git.curoverse.com/arvados.git/sdk/go/arvados" - "git.curoverse.com/arvados.git/sdk/go/arvadosclient" - "git.curoverse.com/arvados.git/sdk/go/dispatch" - "github.com/Sirupsen/logrus" + "git.arvados.org/arvados.git/lib/cmd" + "git.arvados.org/arvados.git/lib/config" + "git.arvados.org/arvados.git/sdk/go/arvados" + "git.arvados.org/arvados.git/sdk/go/arvadosclient" + "git.arvados.org/arvados.git/sdk/go/dispatch" + "github.com/sirupsen/logrus" ) var version = "dev" -func main() { - err := doMain() - if err != nil { - logrus.Fatalf("%q", err) - } -} - var ( runningCmds map[string]*exec.Cmd runningCmdsMutex sync.Mutex waitGroup sync.WaitGroup - crunchRunCommand *string + crunchRunCommand string ) -func doMain() error { - logger := logrus.StandardLogger() +func main() { + baseLogger := logrus.StandardLogger() if os.Getenv("DEBUG") != "" { - logger.SetLevel(logrus.DebugLevel) + baseLogger.SetLevel(logrus.DebugLevel) } - logger.Formatter = &logrus.JSONFormatter{ + baseLogger.Formatter = &logrus.JSONFormatter{ TimestampFormat: "2006-01-02T15:04:05.000000000Z07:00", } @@ -55,7 +50,7 @@ func doMain() error { 10, "Interval in seconds to poll for queued containers") - crunchRunCommand = flags.String( + flags.StringVar(&crunchRunCommand, "crunch-run-command", "/usr/bin/crunch-run", "Crunch command to run container") @@ -65,37 +60,72 @@ func doMain() error { false, "Print version information and exit.") - // Parse args; omit the first arg which is the command name - flags.Parse(os.Args[1:]) + if ok, code := cmd.ParseFlags(flags, os.Args[0], os.Args[1:], "", os.Stderr); !ok { + os.Exit(code) + } // Print version information if requested if *getVersion { fmt.Printf("crunch-dispatch-local %s\n", version) - return nil + return } + loader := config.NewLoader(nil, baseLogger) + cfg, err := loader.Load() + if err != nil { + fmt.Fprintf(os.Stderr, "error loading config: %s\n", err) + os.Exit(1) + } + cluster, err := cfg.GetCluster("") + if err != nil { + fmt.Fprintf(os.Stderr, "config error: %s\n", err) + os.Exit(1) + } + + logger := baseLogger.WithField("ClusterID", cluster.ClusterID) logger.Printf("crunch-dispatch-local %s started", version) runningCmds = make(map[string]*exec.Cmd) + var client arvados.Client + client.APIHost = cluster.Services.Controller.ExternalURL.Host + client.AuthToken = cluster.SystemRootToken + client.Insecure = cluster.TLS.Insecure + + if client.APIHost != "" || client.AuthToken != "" { + // Copy real configs into env vars so [a] + // MakeArvadosClient() uses them, and [b] they get + // propagated to crunch-run via SLURM. + os.Setenv("ARVADOS_API_HOST", client.APIHost) + os.Setenv("ARVADOS_API_TOKEN", client.AuthToken) + os.Setenv("ARVADOS_API_HOST_INSECURE", "") + if client.Insecure { + os.Setenv("ARVADOS_API_HOST_INSECURE", "1") + } + } else { + logger.Warnf("Client credentials missing from config, so falling back on environment variables (deprecated).") + } + arv, err := arvadosclient.MakeArvadosClient() if err != nil { logger.Errorf("error making Arvados client: %v", err) - return err + os.Exit(1) } arv.Retries = 25 + ctx, cancel := context.WithCancel(context.Background()) + dispatcher := dispatch.Dispatcher{ Logger: logger, Arv: arv, - RunContainer: run, + RunContainer: (&LocalRun{startFunc, make(chan bool, 8), ctx, cluster}).run, PollPeriod: time.Duration(*pollInterval) * time.Second, } - ctx, cancel := context.WithCancel(context.Background()) err = dispatcher.Run(ctx) if err != nil { - return err + logger.Error(err) + return } c := make(chan os.Signal, 1) @@ -115,15 +145,18 @@ func doMain() error { // Wait for all running crunch jobs to complete / terminate waitGroup.Wait() - - return nil } func startFunc(container arvados.Container, cmd *exec.Cmd) error { return cmd.Start() } -var startCmd = startFunc +type LocalRun struct { + startCmd func(container arvados.Container, cmd *exec.Cmd) error + concurrencyLimit chan bool + ctx context.Context + cluster *arvados.Cluster +} // Run a container. // @@ -133,16 +166,38 @@ var startCmd = startFunc // // If the container is in any other state, or is not Complete/Cancelled after // crunch-run terminates, mark the container as Cancelled. -func run(dispatcher *dispatch.Dispatcher, +func (lr *LocalRun) run(dispatcher *dispatch.Dispatcher, container arvados.Container, - status <-chan arvados.Container) { + status <-chan arvados.Container) error { uuid := container.UUID if container.State == dispatch.Locked { + + select { + case lr.concurrencyLimit <- true: + break + case <-lr.ctx.Done(): + return lr.ctx.Err() + } + + defer func() { <-lr.concurrencyLimit }() + + select { + case c := <-status: + // Check for state updates after possibly + // waiting to be ready-to-run + if c.Priority == 0 { + goto Finish + } + default: + break + } + waitGroup.Add(1) + defer waitGroup.Done() - cmd := exec.Command(*crunchRunCommand, uuid) + cmd := exec.Command(crunchRunCommand, "--runtime-engine="+lr.cluster.Containers.RuntimeEngine, uuid) cmd.Stdin = nil cmd.Stderr = os.Stderr cmd.Stdout = os.Stderr @@ -153,9 +208,9 @@ func run(dispatcher *dispatch.Dispatcher, // succeed in starting crunch-run. runningCmdsMutex.Lock() - if err := startCmd(container, cmd); err != nil { + if err := lr.startCmd(container, cmd); err != nil { runningCmdsMutex.Unlock() - dispatcher.Logger.Warnf("error starting %q for %s: %s", *crunchRunCommand, uuid, err) + dispatcher.Logger.Warnf("error starting %q for %s: %s", crunchRunCommand, uuid, err) dispatcher.UpdateState(uuid, dispatch.Cancelled) } else { runningCmds[uuid] = cmd @@ -194,9 +249,10 @@ func run(dispatcher *dispatch.Dispatcher, delete(runningCmds, uuid) runningCmdsMutex.Unlock() } - waitGroup.Done() } +Finish: + // If the container is not finalized, then change it to "Cancelled". err := dispatcher.Arv.Get("containers", uuid, nil, &container) if err != nil { @@ -204,7 +260,7 @@ func run(dispatcher *dispatch.Dispatcher, } if container.State == dispatch.Locked || container.State == dispatch.Running { dispatcher.Logger.Warnf("after %q process termination, container state for %v is %q; updating it to %q", - *crunchRunCommand, uuid, container.State, dispatch.Cancelled) + crunchRunCommand, uuid, container.State, dispatch.Cancelled) dispatcher.UpdateState(uuid, dispatch.Cancelled) } @@ -213,4 +269,5 @@ func run(dispatcher *dispatch.Dispatcher, } dispatcher.Logger.Printf("finalized container %v", uuid) + return nil }