From 0105f624283b748b237deaf851e88c5923083357 Mon Sep 17 00:00:00 2001 From: Lennart Espe <3391295+lnsp@users.noreply.github.com> Date: Sun, 26 Jul 2026 14:11:39 +0200 Subject: [PATCH] Add valar batch commands for RFC-0003 batch executions Batch services could not be created through the CLI at all: kind is only settable when a service is created, and neither SubmitArtifact nor SubmitBuild sent it, so batch was reachable only by hand-rolled HTTP. builds push gains --kind. It is validated before the artifact upload, because the upload is what creates the service, and it is repeated on the build request because the server reads an absent kind as "function" and answers 409 on a mismatch with an existing batch service. valar batch adds run, list, inspect, logs, cancel and config. Two behaviours worth noting. run --follow waits for the execution to finish and then prints its output, rather than streaming live: batch containers routinely exit in under a second, so a tail started when the status turns "running" usually attaches to an already-closed log and returns nothing. Waiting first makes the output deterministic, which was not true of the first version. And inspect translates blocked_gate into something actionable. The five gates look identical from outside -- the execution simply has not started -- but "no single runner has room" and "batch is at its configured share" call for different responses. config only sends a PATCH when a flag was actually passed, so reading the configuration cannot silently rewrite it. --- api/client.go | 127 +++++++++++++++++- cmd/batch.go | 333 ++++++++++++++++++++++++++++++++++++++++++++++ cmd/batch_test.go | 80 +++++++++++ cmd/builds.go | 26 +++- cmd/root.go | 2 + 5 files changed, 563 insertions(+), 5 deletions(-) create mode 100644 cmd/batch.go create mode 100644 cmd/batch_test.go diff --git a/api/client.go b/api/client.go index 697ee5a..abd7b3f 100644 --- a/api/client.go +++ b/api/client.go @@ -159,11 +159,15 @@ func (client *Client) StreamServiceLogs(project, service string, w io.Writer, fo } // SubmitArtifact submits a new build input artifact to the server. -func (client *Client) SubmitArtifact(project, service string, archive io.ReadCloser) (*Artifact, error) { +func (client *Client) SubmitArtifact(project, service, kind string, archive io.ReadCloser) (*Artifact, error) { var ( artifact Artifact path = fmt.Sprintf("/projects/%s/services/%s/artifacts", project, service) ) + // The body is the artifact itself, so kind travels as a query parameter. + if kind != "" { + path += "?kind=" + url.QueryEscape(kind) + } if err := client.request(http.MethodPost, path, &artifact, archive); err != nil { return nil, err } @@ -511,6 +515,121 @@ func (client *Client) VerifyDomain(project, domain string) (*Domain, error) { return &dom, nil } +// BatchExecution is one run-to-completion execution of a batch service. +type BatchExecution struct { + ID string `json:"id"` + Service string `json:"service"` + Build string `json:"build"` + Owner string `json:"owner"` + Status string `json:"status"` + Args []string `json:"args"` + Environment map[string]string `json:"environment"` + Error string `json:"error"` + ExitCode int32 `json:"exitCode"` + Attempts int32 `json:"attempts"` + // BlockedGate names the admission gate that last refused this execution. + // Empty on a pending execution means it is simply queued behind others + // rather than blocked on capacity. + BlockedGate string `json:"blockedGate"` + CreatedAt time.Time `json:"createdAt"` + FinishedAt *time.Time `json:"finishedAt"` +} + +// BatchExecutionRequest triggers an execution. An empty Build asks the server +// for the latest successful build. +type BatchExecutionRequest struct { + Build string `json:"build,omitempty"` + Args []string `json:"args,omitempty"` + Environment map[string]string `json:"environment,omitempty"` +} + +// BatchConfig is a batch service's kind-specific configuration. +type BatchConfig struct { + MaxParallelism int32 `json:"maxParallelism"` + TimeoutSeconds int32 `json:"timeoutSeconds"` + Weight int32 `json:"weight"` +} + +func batchPath(project, service string) string { + return fmt.Sprintf("/projects/%s/services/%s/batch/executions", project, service) +} + +// TriggerBatchExecution queues a new execution. +func (client *Client) TriggerBatchExecution(project, service string, req *BatchExecutionRequest) (*BatchExecution, error) { + payload, err := json.Marshal(req) + if err != nil { + return nil, err + } + var exec BatchExecution + if err := client.request(http.MethodPost, batchPath(project, service), &exec, bytes.NewReader(payload)); err != nil { + return nil, err + } + return &exec, nil +} + +// ListBatchExecutions lists a batch service's executions, newest first. +func (client *Client) ListBatchExecutions(project, service string) ([]BatchExecution, error) { + var execs []BatchExecution + if err := client.request(http.MethodGet, batchPath(project, service), &execs, nil); err != nil { + return nil, err + } + return execs, nil +} + +// InspectBatchExecution retrieves a single execution. +func (client *Client) InspectBatchExecution(project, service, execution string) (*BatchExecution, error) { + var exec BatchExecution + path := batchPath(project, service) + "/" + execution + if err := client.request(http.MethodGet, path, &exec, nil); err != nil { + return nil, err + } + return &exec, nil +} + +// CancelBatchExecution cancels a pending or running execution. +func (client *Client) CancelBatchExecution(project, service, execution string) error { + var resp struct{} + path := batchPath(project, service) + "/" + execution + "/cancel" + return client.request(http.MethodPost, path, &resp, nil) +} + +// StreamBatchExecutionLogs writes one execution's container output to w. +// +// Batch logs are addressed per execution rather than per service: a batch +// service has no deployment for the service-wide endpoint to derive a label +// from. +func (client *Client) StreamBatchExecutionLogs(project, service, execution string, w io.Writer, follow bool) error { + path := batchPath(project, service) + "/" + execution + "/logs" + if follow { + path += "?follow" + } + return client.streamRequest(http.MethodGet, path, w) +} + +// GetBatchConfig retrieves a batch service's configuration. +func (client *Client) GetBatchConfig(project, service string) (*BatchConfig, error) { + var cfg BatchConfig + path := fmt.Sprintf("/projects/%s/services/%s/config", project, service) + if err := client.request(http.MethodGet, path, &cfg, nil); err != nil { + return nil, err + } + return &cfg, nil +} + +// SetBatchConfig replaces a batch service's configuration. +func (client *Client) SetBatchConfig(project, service string, cfg *BatchConfig) (*BatchConfig, error) { + payload, err := json.Marshal(cfg) + if err != nil { + return nil, err + } + var out BatchConfig + path := fmt.Sprintf("/projects/%s/services/%s/config", project, service) + if err := client.request(http.MethodPatch, path, &out, bytes.NewReader(payload)); err != nil { + return nil, err + } + return &out, nil +} + func (client *Client) ListSchedules(project, service string) ([]Schedule, error) { var ( path = fmt.Sprintf("/projects/%s/services/%s/schedules", project, service) @@ -600,7 +719,11 @@ type DeployRequest struct { type BuildRequest struct { Artifact string `json:"artifact"` - Build struct { + // Kind is the service workload type. Empty means function. It must be + // repeated on every build for an existing batch service: the server treats + // an absent kind as function and answers 409 on a mismatch. + Kind string `json:"kind,omitempty"` + Build struct { Constructor string `json:"constructor"` Environment []KVPair `json:"environment"` } `json:"build"` diff --git a/cmd/batch.go b/cmd/batch.go new file mode 100644 index 0000000..d8bbfbb --- /dev/null +++ b/cmd/batch.go @@ -0,0 +1,333 @@ +package cmd + +import ( + "fmt" + "os" + "strconv" + "strings" + "text/tabwriter" + "time" + + "github.com/spf13/cobra" + "github.com/valar/cli/api" + "github.com/valar/cli/config" +) + +var ( + batchService string + + batchCmd = &cobra.Command{ + Use: "batch", + Short: "Manage batch executions of a service.", + Long: `Manage batch executions of a service. + +Batch services run containers to completion instead of serving traffic. Create +one by pushing a build with --kind batch, then trigger executions on demand.`, + } + + batchRunBuild string + batchRunEnv []string + batchRunFollow bool + batchRunWait bool + batchRunCmd = &cobra.Command{ + Use: "run [-- args...]", + Short: "Trigger a new batch execution.", + Long: `Trigger a new batch execution. + +Arguments after -- replace the image's default command. Without --build the +latest successful build is used.`, + Run: runAndHandle(func(cmd *cobra.Command, args []string) error { + cfg, client, err := batchContext() + if err != nil { + return err + } + env, err := parseBatchEnv(batchRunEnv) + if err != nil { + return err + } + exec, err := client.TriggerBatchExecution(cfg.Project(), cfg.Service(), &api.BatchExecutionRequest{ + Build: batchRunBuild, + Args: args, + Environment: env, + }) + if err != nil { + return err + } + fmt.Printf("Execution %s %s\n", exec.ID, colorize(exec.Status)) + if !batchRunFollow && !batchRunWait { + return nil + } + final, err := awaitBatchCompletion(client, cfg, exec.ID) + if err != nil { + return err + } + if batchRunFollow { + // Deliberately fetched after completion rather than streamed + // live. Batch containers routinely finish in under a second, so + // a tail started on "running" usually attaches to a log that is + // already closed and the stream ends before anything arrives. + // Waiting first makes the output deterministic. + if err := client.StreamBatchExecutionLogs(cfg.Project(), cfg.Service(), exec.ID, os.Stdout, false); err != nil { + fmt.Fprintf(os.Stderr, "could not read execution output: %v\n", err) + } + } + fmt.Printf("Execution %s %s (exit %d)\n", final.ID, colorize(final.Status), final.ExitCode) + if final.Status != "succeeded" { + if final.Error != "" { + fmt.Fprintln(os.Stderr, final.Error) + } + os.Exit(1) + } + return nil + }), + } + + batchListCmd = &cobra.Command{ + Use: "list", + Short: "List batch executions of a service.", + Run: runAndHandle(func(cmd *cobra.Command, args []string) error { + cfg, client, err := batchContext() + if err != nil { + return err + } + execs, err := client.ListBatchExecutions(cfg.Project(), cfg.Service()) + if err != nil { + return err + } + tw := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) + fmt.Fprintln(tw, "ID\tSTATUS\tEXIT\tBUILD\tCREATED\tNOTE") + for _, exec := range execs { + fmt.Fprintln(tw, strings.Join([]string{ + exec.ID, + colorize(exec.Status), + strconv.Itoa(int(exec.ExitCode)), + shortID(exec.Build), + exec.CreatedAt.Local().Format(time.RFC822), + batchNote(exec), + }, "\t")) + } + return tw.Flush() + }), + } + + batchInspectCmd = &cobra.Command{ + Use: "inspect execution", + Short: "Show the details of a batch execution.", + Args: cobra.ExactArgs(1), + Run: runAndHandle(func(cmd *cobra.Command, args []string) error { + cfg, client, err := batchContext() + if err != nil { + return err + } + exec, err := client.InspectBatchExecution(cfg.Project(), cfg.Service(), args[0]) + if err != nil { + return err + } + tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintf(tw, "ID:\t%s\n", exec.ID) + fmt.Fprintf(tw, "Status:\t%s\n", colorize(exec.Status)) + fmt.Fprintf(tw, "Exit code:\t%d\n", exec.ExitCode) + fmt.Fprintf(tw, "Build:\t%s\n", exec.Build) + fmt.Fprintf(tw, "Owner:\t%s\n", exec.Owner) + fmt.Fprintf(tw, "Attempts:\t%d\n", exec.Attempts) + fmt.Fprintf(tw, "Created:\t%s\n", exec.CreatedAt.Local().Format(time.RFC1123)) + if exec.FinishedAt != nil { + fmt.Fprintf(tw, "Finished:\t%s\n", exec.FinishedAt.Local().Format(time.RFC1123)) + } + if len(exec.Args) > 0 { + fmt.Fprintf(tw, "Args:\t%s\n", strings.Join(exec.Args, " ")) + } + for key, value := range exec.Environment { + fmt.Fprintf(tw, "Env:\t%s=%s\n", key, value) + } + if exec.BlockedGate != "" { + fmt.Fprintf(tw, "Blocked by:\t%s\n", batchGateReason(exec.BlockedGate)) + } + if exec.Error != "" { + fmt.Fprintf(tw, "Error:\t%s\n", exec.Error) + } + return tw.Flush() + }), + } + + batchLogsFollow bool + batchLogsCmd = &cobra.Command{ + Use: "logs execution", + Short: "Show the container output of a batch execution.", + Args: cobra.ExactArgs(1), + Run: runAndHandle(func(cmd *cobra.Command, args []string) error { + cfg, client, err := batchContext() + if err != nil { + return err + } + return client.StreamBatchExecutionLogs(cfg.Project(), cfg.Service(), args[0], os.Stdout, batchLogsFollow) + }), + } + + batchCancelCmd = &cobra.Command{ + Use: "cancel execution", + Short: "Cancel a pending or running batch execution.", + Args: cobra.ExactArgs(1), + Run: runAndHandle(func(cmd *cobra.Command, args []string) error { + cfg, client, err := batchContext() + if err != nil { + return err + } + if err := client.CancelBatchExecution(cfg.Project(), cfg.Service(), args[0]); err != nil { + return err + } + fmt.Printf("Execution %s cancelled\n", args[0]) + return nil + }), + } + + batchConfigParallelism int32 + batchConfigTimeout int32 + batchConfigWeight int32 + batchConfigCmd = &cobra.Command{ + Use: "config [--max-parallelism n] [--timeout seconds] [--weight n]", + Short: "Show or change the batch configuration of a service.", + Run: runAndHandle(func(cmd *cobra.Command, args []string) error { + cfg, client, err := batchContext() + if err != nil { + return err + } + current, err := client.GetBatchConfig(cfg.Project(), cfg.Service()) + if err != nil { + return err + } + // A PATCH with no flags would silently rewrite the config with the + // values it just read, so only write when something was asked for. + changed := false + for name, target := range map[string]*int32{ + "max-parallelism": ¤t.MaxParallelism, + "timeout": ¤t.TimeoutSeconds, + "weight": ¤t.Weight, + } { + if cmd.Flags().Changed(name) { + switch name { + case "max-parallelism": + *target = batchConfigParallelism + case "timeout": + *target = batchConfigTimeout + case "weight": + *target = batchConfigWeight + } + changed = true + } + } + if changed { + if current, err = client.SetBatchConfig(cfg.Project(), cfg.Service(), current); err != nil { + return err + } + } + tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintf(tw, "Max parallelism:\t%s\n", zeroAsDefault(current.MaxParallelism, "1")) + fmt.Fprintf(tw, "Timeout:\t%s\n", zeroAsDefault(current.TimeoutSeconds, "3600 (platform default)")) + fmt.Fprintf(tw, "Weight:\t%s\n", zeroAsDefault(current.Weight, "1 (platform default)")) + return tw.Flush() + }), + } +) + +// batchContext resolves the service config and an API client together, since +// every batch subcommand needs both. +func batchContext() (config.ServiceConfig, *api.Client, error) { + cfg, err := config.NewServiceConfigWithFallback(functionConfiguration, &batchService, globalConfiguration) + if err != nil { + return nil, nil, err + } + client, err := globalConfiguration.APIClient() + if err != nil { + return nil, nil, err + } + return cfg, client, nil +} + +func parseBatchEnv(pairs []string) (map[string]string, error) { + if len(pairs) == 0 { + return nil, nil + } + env := make(map[string]string, len(pairs)) + for _, pair := range pairs { + key, value, found := strings.Cut(pair, "=") + if !found || key == "" { + return nil, fmt.Errorf("invalid environment entry %q, want KEY=VALUE", pair) + } + env[key] = value + } + return env, nil +} + +// batchGateReason translates an admission gate into something actionable. The +// four situations look identical from the outside but have different remedies. +func batchGateReason(gate string) string { + switch gate { + case "follower_fit": + return "no single runner has room for this execution (capacity is fragmented)" + case "overcommit": + return "would exceed a runner's memory overcommit ceiling" + case "batch_budget": + return "batch work is at its configured share of the cluster" + case "dispatch_window": + return "too many batch jobs already queued on the scheduler" + case "stale_view": + return "the control plane has lost sight of the cluster" + default: + return gate + } +} + +func batchNote(exec api.BatchExecution) string { + if exec.BlockedGate != "" { + return batchGateReason(exec.BlockedGate) + } + return exec.Error +} + +func zeroAsDefault(value int32, fallback string) string { + if value == 0 { + return fallback + } + return strconv.Itoa(int(value)) +} + +func shortID(id string) string { + if len(id) > 8 { + return id[:8] + } + return id +} + +const batchPollInterval = 2 * time.Second + +func awaitBatchCompletion(client *api.Client, cfg config.ServiceConfig, id string) (*api.BatchExecution, error) { + for { + exec, err := client.InspectBatchExecution(cfg.Project(), cfg.Service(), id) + if err != nil { + return nil, err + } + switch exec.Status { + case "succeeded", "failed", "cancelled": + return exec, nil + } + time.Sleep(batchPollInterval) + } +} + +func initBatchCmd() { + batchCmd.PersistentFlags().StringVarP(&batchService, "service", "s", "", "The batch service to operate on") + + batchRunCmd.Flags().StringVarP(&batchRunBuild, "build", "b", "", "Build to run (default: latest successful)") + batchRunCmd.Flags().StringArrayVarP(&batchRunEnv, "env", "e", nil, "Environment variable as KEY=VALUE, repeatable") + batchRunCmd.Flags().BoolVarP(&batchRunFollow, "follow", "f", false, "Wait for the execution to finish, then print its output") + batchRunCmd.Flags().BoolVarP(&batchRunWait, "wait", "w", false, "Wait for the execution to finish and exit non-zero if it failed") + batchLogsCmd.Flags().BoolVarP(&batchLogsFollow, "follow", "f", false, "Follow the output") + + batchConfigCmd.Flags().Int32Var(&batchConfigParallelism, "max-parallelism", 0, "Maximum concurrent executions") + batchConfigCmd.Flags().Int32Var(&batchConfigTimeout, "timeout", 0, "Maximum seconds a single execution may run") + batchConfigCmd.Flags().Int32Var(&batchConfigWeight, "weight", 0, "Relative share of batch capacity for this project") + + batchCmd.AddCommand(batchRunCmd, batchListCmd, batchInspectCmd, batchLogsCmd, batchCancelCmd, batchConfigCmd) + rootCmd.AddCommand(batchCmd) +} diff --git a/cmd/batch_test.go b/cmd/batch_test.go new file mode 100644 index 0000000..fbab81a --- /dev/null +++ b/cmd/batch_test.go @@ -0,0 +1,80 @@ +package cmd + +import ( + "testing" +) + +func TestParseBatchEnv(t *testing.T) { + for _, tc := range []struct { + Name string + Input []string + Want map[string]string + WantErr bool + }{ + {"Empty", nil, nil, false}, + {"Single", []string{"A=1"}, map[string]string{"A": "1"}, false}, + {"Multiple", []string{"A=1", "B=2"}, map[string]string{"A": "1", "B": "2"}, false}, + // A value may legitimately contain '=' -- only the first splits. + {"ValueWithEquals", []string{"DSN=a=b"}, map[string]string{"DSN": "a=b"}, false}, + {"EmptyValue", []string{"A="}, map[string]string{"A": ""}, false}, + {"NoEquals", []string{"A"}, nil, true}, + {"NoKey", []string{"=1"}, nil, true}, + } { + t.Run(tc.Name, func(t *testing.T) { + got, err := parseBatchEnv(tc.Input) + if tc.WantErr { + if err == nil { + t.Fatalf("expected an error for %q", tc.Input) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != len(tc.Want) { + t.Fatalf("got %v, want %v", got, tc.Want) + } + for k, v := range tc.Want { + if got[k] != v { + t.Errorf("key %q: got %q, want %q", k, got[k], v) + } + } + }) + } +} + +func TestValidateServiceKind(t *testing.T) { + for _, kind := range []string{"", "function", "batch"} { + if err := validateServiceKind(kind); err != nil { + t.Errorf("kind %q should be accepted: %v", kind, err) + } + } + for _, kind := range []string{"job", "Batch", "BATCH"} { + if err := validateServiceKind(kind); err == nil { + t.Errorf("kind %q should be rejected", kind) + } + } +} + +// Every gate the server can report should map to an explanation. An unmapped +// gate would surface as a bare identifier, which tells a user nothing about +// what to do next. +func TestBatchGateReason_CoversEveryServerGate(t *testing.T) { + for _, gate := range []string{"follower_fit", "overcommit", "batch_budget", "dispatch_window", "stale_view"} { + if reason := batchGateReason(gate); reason == gate { + t.Errorf("gate %q has no explanation", gate) + } + } + if got := batchGateReason("something_new"); got != "something_new" { + t.Errorf("unknown gate should pass through, got %q", got) + } +} + +func TestZeroAsDefault(t *testing.T) { + if got := zeroAsDefault(0, "fallback"); got != "fallback" { + t.Errorf("got %q, want fallback", got) + } + if got := zeroAsDefault(3, "fallback"); got != "3" { + t.Errorf("got %q, want 3", got) + } +} diff --git a/cmd/builds.go b/cmd/builds.go index edfaee6..22ec172 100644 --- a/cmd/builds.go +++ b/cmd/builds.go @@ -19,7 +19,21 @@ import ( "golang.org/x/crypto/ssh/terminal" ) -var buildService string +var ( + buildService string + buildPushKind string +) + +// validateServiceKind rejects a bad kind before uploading an artifact, since +// the upload is what creates the service. +func validateServiceKind(kind string) error { + switch kind { + case "", "function", "batch": + return nil + default: + return fmt.Errorf("unknown service kind %q, want \"function\" or \"batch\"", kind) + } +} var buildCmd = &cobra.Command{ Use: "build [--service service]", @@ -384,13 +398,18 @@ var buildPushCmd = &cobra.Command{ return fmt.Errorf("package archive failed: %w", err) } defer targzFile.Close() - artifact, err := client.SubmitArtifact(serviceCfg.Project(), serviceCfg.Service(), targzFile) + if err := validateServiceKind(buildPushKind); err != nil { + return err + } + artifact, err := client.SubmitArtifact(serviceCfg.Project(), serviceCfg.Service(), buildPushKind, targzFile) if err != nil { return err } - // Submit build request + // Submit build request. The kind is repeated here because the server + // reads an absent kind as "function" and rejects a mismatch with 409. var buildReq api.BuildRequest buildReq.Artifact = artifact.Artifact + buildReq.Kind = buildPushKind buildReq.Build.Constructor = serviceCfg.Build().Constructor for _, kv := range serviceCfg.Build().Environment { buildReq.Build.Environment = append(buildReq.Build.Environment, api.KVPair(kv)) @@ -413,6 +432,7 @@ func initBuildsCmd() { buildLogsCmd.PersistentFlags().BoolVarP(&logsFollow, "follow", "f", false, "Follow the logs") buildLogsCmd.PersistentFlags().BoolVarP(&logsRaw, "raw", "r", false, "Dump the unformatted log content") buildPushCmd.Flags().BoolVar(&buildPushNoDeploy, "no-deploy", false, "Only build, skip deploy action") + buildPushCmd.Flags().StringVar(&buildPushKind, "kind", "", `Workload type for a new service: "function" (default) or "batch"`) buildCmd.AddCommand(buildListCmd, buildInspectCmd, buildLogsCmd, buildAbortCmd, buildStatusCmd, buildWatchCmd, buildPushCmd) rootCmd.AddCommand(buildCmd) } diff --git a/cmd/root.go b/cmd/root.go index 9cb667c..5b07254 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -49,6 +49,8 @@ func init() { initConfigCmd() // Configure cron.go initCronCmd() + // Configure batch.go + initBatchCmd() } func runAndHandle(f func(*cobra.Command, []string) error) func(*cobra.Command, []string) {