Skip to content

feat(functions): wait for submitted deployments #202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tkkhq wants to merge 1 commit into from
Closed

feat(functions): wait for submitted deployments #202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion docs/functions.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,13 +61,16 @@ target instead of the active context.

Cloud deploys use latest-wins queueing. If the function is already deploying,
the new source replaces any older queued deploy and runs next. Delete supersedes
queued deploys; later deploys are rejected until deletion finishes.
queued deploys; later deploys are rejected until deletion finishes. Add `--wait`
to return only after each deployment submitted by the command becomes active or
one ends without becoming active.

## Examples

```bash
# Deploy everything, or a single function by name or path
volcano functions deploy --all
volcano functions deploy --all --wait
volcano functions deploy -f get-notes
volcano functions deploy -f volcano/functions/get-notes.js

Expand Down
84 changes: 74 additions & 10 deletions internal/cmd/functions/deploy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"os"
"path/filepath"
"strings"
"time"

"github.com/spf13/cobra"

Expand All @@ -19,19 +20,25 @@ import (
cliruntime "github.com/Kong/volcano-cli/internal/runtime"
)

const deployBatchSize = 100
const (
deployBatchSize = 100
deployPollInterval = 2 * time.Second
maxDeploymentReadFailures = 3
)

type deployOptions struct {
deps cliruntime.Deps
file string
all bool
batchAll bool
wait bool
out io.Writer
}

func newDeploy(deps cliruntime.Deps, batchAll bool) *cobra.Command {
var file string
var all bool
var wait bool
cmd := &cobra.Command{
Use: "deploy",
Short: "Deploy functions",
Expand All @@ -46,7 +53,8 @@ Usage:
The CLI scans volcano/functions, detects the runtime from source file extensions,
packages source with dependency manifests and shared libraries, and uploads the
archive to Volcano. Cloud deploy-all uploads are split into batches of up to
100 functions; local deploy-all uploads each function individually.`,
100 functions; local deploy-all uploads each function individually. Use --wait
to return only after every submitted deployment becomes active or one fails.`,
cliruntime.CommandPath(deps, "functions deploy --all"),
cliruntime.CommandPath(deps, "functions deploy -a"),
cliruntime.CommandPath(deps, "functions deploy -f get-notes"),
Expand All @@ -58,12 +66,14 @@ archive to Volcano. Cloud deploy-all uploads are split into batches of up to
file: strings.TrimSpace(file),
all: all,
batchAll: batchAll,
wait: wait,
out: cmd.OutOrStdout(),
})
},
}
cmd.Flags().StringVarP(&file, "file", "f", "", "Deploy a specific function by name or path")
cmd.Flags().BoolVarP(&all, "all", "a", false, "Deploy all functions")
cmd.Flags().BoolVar(&wait, "wait", false, "Wait for each submitted deployment to become active")
return cmd
}

Expand Down Expand Up @@ -133,9 +143,9 @@ func runDeploy(ctx context.Context, opts deployOptions) error {
}

if opts.all {
return runDeployAll(ctx, opts.out, service, baseDir, sources, opts.batchAll, manifest.Declarations)
return runDeployAll(ctx, opts.out, service, baseDir, sources, opts.batchAll, manifest.Declarations, opts.wait, opts.deps)
}
return runDeployOne(ctx, opts.out, service, baseDir, sources[0], manifest.Declarations)
return runDeployOne(ctx, opts.out, service, baseDir, sources[0], manifest.Declarations, opts.wait, opts.deps)
}

func excludeDurable(sources []clifunction.SourceInfo, durableNames map[string]bool, out io.Writer) []clifunction.SourceInfo {
Expand Down Expand Up @@ -172,7 +182,7 @@ func applyVariableDeclaration(pkg *clifunction.Package, declarations map[string]
pkg.Variables = declaration.Variables
}

func runDeployOne(ctx context.Context, out io.Writer, service clifunction.Service, baseDir string, source clifunction.SourceInfo, declarations map[string]projectconfig.FunctionVariableDeclaration) error {
func runDeployOne(ctx context.Context, out io.Writer, service clifunction.Service, baseDir string, source clifunction.SourceInfo, declarations map[string]projectconfig.FunctionVariableDeclaration, wait bool, deps cliruntime.Deps) error {
fmt.Fprintf(out, "\n[1/1] Deploying %s...\n", source.Name)
printSourceSummary(out, source)
pkg, err := clifunction.PackageSource(source, baseDir)
Expand All @@ -190,10 +200,10 @@ func runDeployOne(ctx context.Context, out io.Writer, service clifunction.Servic
fmt.Fprintf(out, " Deployed %s\n", deployed.Name)
fmt.Fprintln(out)
output.Success(out, "1/1 functions deployment started")
return nil
return waitForSubmittedDeployments(ctx, deps, out, service, []apiclient.Function{*deployed}, wait)
}

func runDeployAll(ctx context.Context, out io.Writer, service clifunction.Service, baseDir string, sources []clifunction.SourceInfo, batch bool, declarations map[string]projectconfig.FunctionVariableDeclaration) error {
func runDeployAll(ctx context.Context, out io.Writer, service clifunction.Service, baseDir string, sources []clifunction.SourceInfo, batch bool, declarations map[string]projectconfig.FunctionVariableDeclaration, wait bool, deps cliruntime.Deps) error {
packages := make([]clifunction.Package, 0, len(sources))
var totalSize int64
for i, source := range sources {
Expand All @@ -211,12 +221,13 @@ func runDeployAll(ctx context.Context, out io.Writer, service clifunction.Servic

fmt.Fprintf(out, "\nTotal upload size: %s\n", archive.FormatSize(totalSize))
if !batch {
return runDeployAllIndividually(ctx, out, service, packages)
return runDeployAllIndividually(ctx, out, service, packages, wait, deps)
}

totalStarted := 0
totalFailed := 0
batchCount := 0
var submitted []apiclient.Function
for start := 0; start < len(packages); start += deployBatchSize {
end := min(start+deployBatchSize, len(packages))
fmt.Fprintf(out, "\nUploading batch %d-%d of %d...\n", start+1, end, len(packages))
Expand All @@ -228,6 +239,7 @@ func runDeployAll(ctx context.Context, out io.Writer, service clifunction.Servic
for _, fn := range resp.Data {
fmt.Fprintf(out, " Deployed %s\n", fn.Name)
}
submitted = append(submitted, resp.Data...)
failures := batchFailures(resp)
for _, failure := range failures {
fmt.Fprintf(out, " ✗ Failed %s: %s\n", failure.Name, failure.Error)
Expand All @@ -245,26 +257,78 @@ func runDeployAll(ctx context.Context, out io.Writer, service clifunction.Servic
return errors.New(message)
}
output.Success(out, "%d/%d functions deployment started across %d batch(es)", totalStarted, len(sources), batchCount)
return nil
return waitForSubmittedDeployments(ctx, deps, out, service, submitted, wait)
}

func runDeployAllIndividually(ctx context.Context, out io.Writer, service clifunction.Service, packages []clifunction.Package) error {
func runDeployAllIndividually(ctx context.Context, out io.Writer, service clifunction.Service, packages []clifunction.Package, wait bool, deps cliruntime.Deps) error {
totalStarted := 0
var submitted []apiclient.Function
for i, pkg := range packages {
fmt.Fprintf(out, "\nUploading %s (%d/%d)...\n", pkg.Name, i+1, len(packages))
deployed, err := service.DeployPackage(ctx, pkg)
if err != nil {
return fmt.Errorf("failed to deploy function %s: %w", pkg.Name, err)
}
fmt.Fprintf(out, " Deployed %s\n", deployed.Name)
submitted = append(submitted, *deployed)
totalStarted++
}

fmt.Fprintln(out)
output.Success(out, "%d/%d functions deployment started", totalStarted, len(packages))
return waitForSubmittedDeployments(ctx, deps, out, service, submitted, wait)
}

func waitForSubmittedDeployments(ctx context.Context, deps cliruntime.Deps, out io.Writer, service clifunction.Service, functions []apiclient.Function, wait bool) error {
if !wait {
return nil
}
for _, fn := range functions {
deploymentID := fn.CurrentDeploymentId
if fn.PendingDeploymentId != nil {
deploymentID = fn.PendingDeploymentId
}
if deploymentID == nil {
return fmt.Errorf("function %q response did not identify its submitted deployment", fn.Name)
}
if err := waitForSubmittedDeployment(ctx, deps, out, service, fn, deploymentID.String()); err != nil {
return err
}
}
return nil
}

func waitForSubmittedDeployment(ctx context.Context, deps cliruntime.Deps, out io.Writer, service clifunction.Service, fn apiclient.Function, deploymentID string) error {
ticker := cliruntime.NewTicker(deps, deployPollInterval)
defer ticker.Stop()
readFailures := 0
for {
deployment, err := service.ResolveDeployment(ctx, fn.Id, deploymentID)
if err != nil {
readFailures++
if readFailures >= maxDeploymentReadFailures {
return fmt.Errorf("failed to read function %q deployment %s after %d attempts: %w", fn.Name, deploymentID, readFailures, err)
}
} else {
readFailures = 0
switch deployment.Status {
case apiclient.FunctionDeploymentStatusActive:
output.Success(out, "Function '%s' deployment %s is active", fn.Name, deploymentID)
return nil
case apiclient.FunctionDeploymentStatusProvisioning,
apiclient.FunctionDeploymentStatusQueued:
default:
return fmt.Errorf("function %q deployment %s finished with status %s", fn.Name, deploymentID, deployment.Status)
}
}
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C():
}
}
}

func printSourceSummary(out io.Writer, source clifunction.SourceInfo) {
detectedFrom := filepath.Ext(source.Path)
if source.IsDir {
Expand Down
144 changes: 144 additions & 0 deletions internal/cmd/functions/deploy_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"os"
"path/filepath"
"testing"
"time"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
Expand Down Expand Up @@ -241,6 +242,149 @@ func TestFunctionsDeployAllChunksLargeBatches(t *testing.T) {
assert.Contains(t, out, "105/105 functions deployment started across 2 batch(es)")
}

func TestFunctionsDeployWaitUsesSubmittedDeploymentID(t *testing.T) {
setFunctionCommandTestHome(t)
saveFunctionCommandTestConfig(t)
projectDir := t.TempDir()
t.Chdir(projectDir)
require.NoError(t, writeProjectFile(filepath.Join("volcano", "functions", "hello.js"), `exports.handler = async () => ({ statusCode: 200 });`))

const submittedDeploymentID = "33333333-3333-4333-8333-333333333333"
const externalDeploymentID = "44444444-4444-4444-8444-444444444444"
deploymentReads := 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case writeFunctionRuntimesCommandResponse(w, r):
return
case r.Method == http.MethodPost && r.URL.Path == "/projects/"+functionProjectID+"/functions/batch":
function := functionCommandPayload(functionID, "hello")
function["pending_deployment_id"] = submittedDeploymentID
writeFunctionCommandJSON(t, w, http.StatusAccepted, map[string]any{
"batch_id": "77777777-7777-4777-8777-777777777777",
"data": []any{function},
})
case r.Method == http.MethodGet && r.URL.Path == "/projects/"+functionProjectID+"/functions/"+functionID+"/deployments":
deploymentReads++
status := "queued"
if deploymentReads > 1 {
status = "active"
}
writeFunctionCommandJSON(t, w, http.StatusOK, map[string]any{
"data": []any{
functionDeploymentCommandPayload(externalDeploymentID, functionID, "failed"),
functionDeploymentCommandPayload(submittedDeploymentID, functionID, status),
},
"has_more": false,
"page": 1,
"total": 2,
})
default:
http.NotFound(w, r)
}
}))
defer server.Close()

ticker := &functionDeployTicker{ch: make(chan time.Time, 1)}
ticker.ch <- time.Now()
deps := cliruntime.Deps{
HTTPClient: server.Client(), APIBaseURL: server.URL,
NewTicker: func(time.Duration) cliruntime.Ticker { return ticker },
}
out, err := executeFunctionsCommand(t, New(deps), "deploy", "--all", "--wait")
require.NoError(t, err)
assert.Contains(t, out, "Function 'hello' deployment "+submittedDeploymentID+" is active")
}

type functionDeployTicker struct {
ch chan time.Time
}

func (t *functionDeployTicker) C() <-chan time.Time { return t.ch }
func (t *functionDeployTicker) Reset(time.Duration) {}
func (t *functionDeployTicker) Stop() {}

func TestFunctionsDeployWaitRejectsRolledBackDeployment(t *testing.T) {
setFunctionCommandTestHome(t)
saveFunctionCommandTestConfig(t)
projectDir := t.TempDir()
t.Chdir(projectDir)
require.NoError(t, writeProjectFile(filepath.Join("volcano", "functions", "hello.js"), `exports.handler = async () => ({ statusCode: 200 });`))

const submittedDeploymentID = "33333333-3333-4333-8333-333333333333"
const previousDeploymentID = "44444444-4444-4444-8444-444444444444"
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case writeFunctionRuntimesCommandResponse(w, r):
return
case r.Method == http.MethodPost && r.URL.Path == "/projects/"+functionProjectID+"/functions/batch":
function := functionCommandPayload(functionID, "hello")
function["current_deployment_id"] = submittedDeploymentID
writeFunctionCommandJSON(t, w, http.StatusAccepted, map[string]any{
"batch_id": "77777777-7777-4777-8777-777777777777",
"data": []any{function},
})
case r.Method == http.MethodGet && r.URL.Path == "/projects/"+functionProjectID+"/functions/"+functionID+"/deployments":
writeFunctionCommandJSON(t, w, http.StatusOK, map[string]any{
"data": []any{
functionDeploymentCommandPayload(previousDeploymentID, functionID, "active"),
functionDeploymentCommandPayload(submittedDeploymentID, functionID, "failed"),
},
"has_more": false,
"page": 1,
"total": 2,
})
default:
http.NotFound(w, r)
}
}))
defer server.Close()

_, err := executeFunctionsCommand(t, New(cliruntime.Deps{HTTPClient: server.Client(), APIBaseURL: server.URL}), "deploy", "--all", "--wait")
require.ErrorContains(t, err, "deployment "+submittedDeploymentID+" finished with status failed")
}

func TestFunctionsDeployWaitStopsAfterPersistentReadErrors(t *testing.T) {
setFunctionCommandTestHome(t)
saveFunctionCommandTestConfig(t)
projectDir := t.TempDir()
t.Chdir(projectDir)
require.NoError(t, writeProjectFile(filepath.Join("volcano", "functions", "hello.js"), `exports.handler = async () => ({ statusCode: 200 });`))

const submittedDeploymentID = "33333333-3333-4333-8333-333333333333"
deploymentReads := 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case writeFunctionRuntimesCommandResponse(w, r):
return
case r.Method == http.MethodPost && r.URL.Path == "/projects/"+functionProjectID+"/functions/batch":
function := functionCommandPayload(functionID, "hello")
function["current_deployment_id"] = submittedDeploymentID
writeFunctionCommandJSON(t, w, http.StatusAccepted, map[string]any{
"batch_id": "77777777-7777-4777-8777-777777777777",
"data": []any{function},
})
case r.Method == http.MethodGet && r.URL.Path == "/projects/"+functionProjectID+"/functions/"+functionID+"/deployments":
deploymentReads++
http.Error(w, "temporary failure", http.StatusServiceUnavailable)
default:
http.NotFound(w, r)
}
}))
defer server.Close()

ticker := &functionDeployTicker{ch: make(chan time.Time, maxDeploymentReadFailures-1)}
for range maxDeploymentReadFailures - 1 {
ticker.ch <- time.Now()
}
deps := cliruntime.Deps{
HTTPClient: server.Client(), APIBaseURL: server.URL,
NewTicker: func(time.Duration) cliruntime.Ticker { return ticker },
}
_, err := executeFunctionsCommand(t, New(deps), "deploy", "--all", "--wait")
require.ErrorContains(t, err, "failed to read function \"hello\" deployment "+submittedDeploymentID+" after 3 attempts")
assert.Equal(t, maxDeploymentReadFailures, deploymentReads)
}

func TestLocalFunctionsDeployAllUsesSingleFunctionUploads(t *testing.T) {
setFunctionCommandTestHome(t)
saveFunctionCommandTestConfig(t)
Expand Down
13 changes: 13 additions & 0 deletions internal/cmd/functions/helpers_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,19 @@ func functionCommandPayload(id, name string) map[string]any {
}
}

func functionDeploymentCommandPayload(id, functionID, status string) map[string]any {
return map[string]any{
"created_at": "2026-05-20T00:00:00Z",
"deploy_source": "request",
"function_id": functionID,
"id": id,
"operation": "update",
"project_id": functionProjectID,
"status": status,
"updated_at": "2026-05-20T00:00:00Z",
}
}

func functionRuntimeCommandPayload(name, language string, isDefault bool, fileExtensions []string, entrypoint, handler string, dependencyManifests []string) map[string]any {
return map[string]any{
"name": name,
Expand Down