"syscall"
"time"
+ "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"
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",
}
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")
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, logger)
+ 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 {
- return fmt.Errorf("config error: %s", err)
+ 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)
if client.Insecure {
os.Setenv("ARVADOS_API_HOST_INSECURE", "1")
}
- os.Setenv("ARVADOS_EXTERNAL_CLIENT", "")
} 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
err = dispatcher.Run(ctx)
if err != nil {
- return err
+ logger.Error(err)
+ return
}
c := make(chan os.Signal, 1)
// Wait for all running crunch jobs to complete / terminate
waitGroup.Wait()
-
- return nil
}
func startFunc(container arvados.Container, cmd *exec.Cmd) error {
waitGroup.Add(1)
defer waitGroup.Done()
- cmd := exec.Command(*crunchRunCommand, "--runtime-engine="+lr.cluster.Containers.RuntimeEngine, uuid)
+ cmd := exec.Command(crunchRunCommand, "--runtime-engine="+lr.cluster.Containers.RuntimeEngine, uuid)
cmd.Stdin = nil
cmd.Stderr = os.Stderr
cmd.Stdout = os.Stderr
runningCmdsMutex.Lock()
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
}
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)
}