1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
7 // Dispatcher service for Crunch that runs containers locally.
20 "git.arvados.org/arvados.git/lib/config"
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/dispatch"
24 "github.com/sirupsen/logrus"
32 logrus.Fatalf("%q", err)
37 runningCmds map[string]*exec.Cmd
38 runningCmdsMutex sync.Mutex
39 waitGroup sync.WaitGroup
40 crunchRunCommand *string
44 logger := logrus.StandardLogger()
45 if os.Getenv("DEBUG") != "" {
46 logger.SetLevel(logrus.DebugLevel)
48 logger.Formatter = &logrus.JSONFormatter{
49 TimestampFormat: "2006-01-02T15:04:05.000000000Z07:00",
52 flags := flag.NewFlagSet("crunch-dispatch-local", flag.ExitOnError)
54 pollInterval := flags.Int(
57 "Interval in seconds to poll for queued containers")
59 crunchRunCommand = flags.String(
61 "/usr/bin/crunch-run",
62 "Crunch command to run container")
64 getVersion := flags.Bool(
67 "Print version information and exit.")
69 // Parse args; omit the first arg which is the command name
70 err := flags.Parse(os.Args[1:])
71 if err == flag.ErrHelp {
73 } else if err != nil {
75 } else if flags.NArg() != 0 {
76 return fmt.Errorf("unrecognized command line arguments: %v", flags.Args())
79 // Print version information if requested
81 fmt.Printf("crunch-dispatch-local %s\n", version)
85 loader := config.NewLoader(nil, logger)
86 cfg, err := loader.Load()
87 cluster, err := cfg.GetCluster("")
89 return fmt.Errorf("config error: %s", err)
92 logger.Printf("crunch-dispatch-local %s started", version)
94 runningCmds = make(map[string]*exec.Cmd)
96 var client arvados.Client
97 client.APIHost = cluster.Services.Controller.ExternalURL.Host
98 client.AuthToken = cluster.SystemRootToken
99 client.Insecure = cluster.TLS.Insecure
101 if client.APIHost != "" || client.AuthToken != "" {
102 // Copy real configs into env vars so [a]
103 // MakeArvadosClient() uses them, and [b] they get
104 // propagated to crunch-run via SLURM.
105 os.Setenv("ARVADOS_API_HOST", client.APIHost)
106 os.Setenv("ARVADOS_API_TOKEN", client.AuthToken)
107 os.Setenv("ARVADOS_API_HOST_INSECURE", "")
109 os.Setenv("ARVADOS_API_HOST_INSECURE", "1")
111 os.Setenv("ARVADOS_EXTERNAL_CLIENT", "")
113 logger.Warnf("Client credentials missing from config, so falling back on environment variables (deprecated).")
116 arv, err := arvadosclient.MakeArvadosClient()
118 logger.Errorf("error making Arvados client: %v", err)
123 ctx, cancel := context.WithCancel(context.Background())
125 dispatcher := dispatch.Dispatcher{
128 RunContainer: (&LocalRun{startFunc, make(chan bool, 8), ctx, cluster}).run,
129 PollPeriod: time.Duration(*pollInterval) * time.Second,
132 err = dispatcher.Run(ctx)
137 c := make(chan os.Signal, 1)
138 signal.Notify(c, os.Interrupt, syscall.SIGTERM, syscall.SIGQUIT)
140 logger.Printf("Received %s, shutting down", sig)
145 runningCmdsMutex.Lock()
146 // Finished dispatching; interrupt any crunch jobs that are still running
147 for _, cmd := range runningCmds {
148 cmd.Process.Signal(os.Interrupt)
150 runningCmdsMutex.Unlock()
152 // Wait for all running crunch jobs to complete / terminate
158 func startFunc(container arvados.Container, cmd *exec.Cmd) error {
162 type LocalRun struct {
163 startCmd func(container arvados.Container, cmd *exec.Cmd) error
164 concurrencyLimit chan bool
166 cluster *arvados.Cluster
171 // If the container is Locked, start a new crunch-run process and wait until
172 // crunch-run completes. If the priority is set to zero, set an interrupt
173 // signal to the crunch-run process.
175 // If the container is in any other state, or is not Complete/Cancelled after
176 // crunch-run terminates, mark the container as Cancelled.
177 func (lr *LocalRun) run(dispatcher *dispatch.Dispatcher,
178 container arvados.Container,
179 status <-chan arvados.Container) error {
181 uuid := container.UUID
183 if container.State == dispatch.Locked {
186 case lr.concurrencyLimit <- true:
188 case <-lr.ctx.Done():
192 defer func() { <-lr.concurrencyLimit }()
196 // Check for state updates after possibly
197 // waiting to be ready-to-run
206 defer waitGroup.Done()
208 cmd := exec.Command(*crunchRunCommand, "--runtime-engine="+lr.cluster.Containers.RuntimeEngine, uuid)
210 cmd.Stderr = os.Stderr
211 cmd.Stdout = os.Stderr
213 dispatcher.Logger.Printf("starting container %v", uuid)
215 // Add this crunch job to the list of runningCmds only if we
216 // succeed in starting crunch-run.
218 runningCmdsMutex.Lock()
219 if err := lr.startCmd(container, cmd); err != nil {
220 runningCmdsMutex.Unlock()
221 dispatcher.Logger.Warnf("error starting %q for %s: %s", *crunchRunCommand, uuid, err)
222 dispatcher.UpdateState(uuid, dispatch.Cancelled)
224 runningCmds[uuid] = cmd
225 runningCmdsMutex.Unlock()
227 // Need to wait for crunch-run to exit
228 done := make(chan struct{})
231 if _, err := cmd.Process.Wait(); err != nil {
232 dispatcher.Logger.Warnf("error while waiting for crunch job to finish for %v: %q", uuid, err)
234 dispatcher.Logger.Debugf("sending done")
244 // Interrupt the child process if priority changes to 0
245 if (c.State == dispatch.Locked || c.State == dispatch.Running) && c.Priority == 0 {
246 dispatcher.Logger.Printf("sending SIGINT to pid %d to cancel container %v", cmd.Process.Pid, uuid)
247 cmd.Process.Signal(os.Interrupt)
253 dispatcher.Logger.Printf("finished container run for %v", uuid)
255 // Remove the crunch job from runningCmds
256 runningCmdsMutex.Lock()
257 delete(runningCmds, uuid)
258 runningCmdsMutex.Unlock()
264 // If the container is not finalized, then change it to "Cancelled".
265 err := dispatcher.Arv.Get("containers", uuid, nil, &container)
267 dispatcher.Logger.Warnf("error getting final container state: %v", err)
269 if container.State == dispatch.Locked || container.State == dispatch.Running {
270 dispatcher.Logger.Warnf("after %q process termination, container state for %v is %q; updating it to %q",
271 *crunchRunCommand, uuid, container.State, dispatch.Cancelled)
272 dispatcher.UpdateState(uuid, dispatch.Cancelled)
275 // drain any subsequent status changes
279 dispatcher.Logger.Printf("finalized container %v", uuid)