Skip to content

Commit 3a060c4

Browse files
BoretsBorets
authored
Improve task dev server status output (#242)
Summary of Local Dev Features for Documentation This PR enhances the render ea tasks dev experience with clearer status reporting and a cleaner terminal output. New Features 1. Startup Checklist When starting the dev server, users now see a clean startup flow: ✓ Dev server starting... ✓ Loaded tasks from main.go • my-task • another-task ✓ Server ready at http://localhost:8120 ✓ Watching for tasks... 2. Task Status Updates Real-time status updates for task execution with colored output: Running - Task is currently executing Completed (green) - Task finished successfully, includes duration Failed (red) - Task failed, shows error details Example output: Task running: tsk-abc123 (my-task) input=[...] Task Completed: tsk-abc123 (my-task) duration=1.23s 3. Multi-Step Workflow Visualization Subtasks are indented with visual hierarchy showing parent relationships: Task running: tsk-parent (parent-task) input=[...] ↳ Subtask (parent: parent-task) running: tsk-child (child-task) input=[...] ↳ Subtask (parent: parent-task) Completed: tsk-child (child-task) duration=500ms Task Completed: tsk-parent (parent-task) duration=2.5s 4. Debug Mode Use --debug flag for detailed execution events, including: Timestamps on every message Task enqueued events Detailed workflow service logs render ea tasks dev --debug -- "go run main.go" Documentation Highlights Default behavior: Clean output showing only running/completed/failed states Debug mode: Use --debug for timestamps and enqueued events Port configuration: Default port is 8120, configurable with --port Local testing: Use --local flag with other task commands to target the dev server
1 parent 1085718 commit 3a060c4

7 files changed

Lines changed: 381 additions & 32 deletions

File tree

‎cmd/task.go‎

Lines changed: 92 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,9 @@ import (
55
"fmt"
66
"net/http"
77
"os"
8+
"path/filepath"
9+
"sort"
10+
"strings"
811

912
"github.com/gorilla/websocket"
1013
"github.com/render-oss/cli/pkg/client"
@@ -58,7 +61,6 @@ Examples:
5861
render ea taskruns start my-task --local --input='["arg1"]'
5962
`,
6063
RunE: func(cmd *cobra.Command, args []string) error {
61-
6264
ctx := cmd.Context()
6365
var commandArgs []string
6466
if cmd.ArgsLenAtDash() >= 0 {
@@ -69,6 +71,13 @@ Examples:
6971
return errors.New("command is required")
7072
}
7173

74+
command.Println(cmd, "✓ Dev server starting...")
75+
76+
debugMode, err := cmd.Flags().GetBool("debug")
77+
if err != nil {
78+
return fmt.Errorf("failed to get debug flag: %w", err)
79+
}
80+
7281
socketTracker, err := orchestrator.NewSocketTracker(ctx)
7382
if err != nil {
7483
return err
@@ -78,7 +87,32 @@ Examples:
7887

7988
logs := logstore.NewLogStore()
8089
store := store.NewTaskStore()
81-
coordinator := orchestrator.NewCoordinator(ctx, store, orchestrator.NewExec(logs, commandArgs[0], commandArgs[1:]...), socketTracker, taskServerFactory)
90+
var (
91+
pending []string
92+
ready bool
93+
)
94+
95+
statusReporter := orchestrator.NewPrintStatusReporter(
96+
func(format string, args ...any) {
97+
message := fmt.Sprintf(format, args...)
98+
if !ready {
99+
pending = append(pending, message)
100+
return
101+
}
102+
command.Println(cmd, "%s", message)
103+
},
104+
orchestrator.WithStatusReporterTimestamps(debugMode),
105+
orchestrator.WithStatusReporterTaskEnqueued(debugMode),
106+
orchestrator.WithStatusReporterIncludeInputs(true),
107+
)
108+
coordinator := orchestrator.NewCoordinator(
109+
ctx,
110+
store,
111+
orchestrator.NewExec(logs, debugMode, commandArgs[0], commandArgs[1:]...),
112+
socketTracker,
113+
taskServerFactory,
114+
statusReporter,
115+
)
82116

83117
upgrader := &websocket.Upgrader{
84118
Error: func(w http.ResponseWriter, r *http.Request, status int, reason error) {
@@ -95,6 +129,28 @@ Examples:
95129
apiSrv := apiserver.Start(api, port)
96130
logs.Start(ctx)
97131

132+
registeredTasks, err := coordinator.PopulateTasks(ctx)
133+
if err != nil {
134+
return fmt.Errorf("failed to load tasks: %w", err)
135+
}
136+
137+
if source := describeWorkflowSource(commandArgs); source != "" {
138+
command.Println(cmd, "✓ Loaded tasks from %s", source)
139+
} else {
140+
command.Println(cmd, "✓ Loaded tasks")
141+
}
142+
143+
command.Println(cmd, "%s", formatTaskSummary(registeredTasks))
144+
command.Println(cmd, "✓ Server ready at http://localhost:%d", port)
145+
command.Println(cmd, "✓ Watching for tasks...")
146+
command.Println(cmd, "")
147+
148+
ready = true
149+
for _, line := range pending {
150+
command.Println(cmd, "%s", line)
151+
}
152+
pending = nil
153+
98154
<-ctx.Done()
99155

100156
apiSrv.Shutdown(ctx)
@@ -184,6 +240,7 @@ func init() {
184240
taskCmd.PersistentFlags().Int("port", defaultTaskAPIPort, "Port of the local task server (8120 when not specified)")
185241

186242
taskDevCmd.Flags().Int("port", defaultTaskAPIPort, "Port of the local task server (8120 when not specified)")
243+
taskDevCmd.Flags().Bool("debug", false, "Print detailed workflow task execution events")
187244

188245
EarlyAccessCmd.AddCommand(taskCmd)
189246
taskCmd.AddCommand(taskDevCmd)
@@ -238,3 +295,36 @@ func getLocalDeps(cmd *cobra.Command, deps flows.WorkflowDeps) (flows.WorkflowDe
238295
}
239296
return deps, local, nil
240297
}
298+
299+
func describeWorkflowSource(commandArgs []string) string {
300+
if len(commandArgs) == 0 {
301+
return ""
302+
}
303+
304+
last := commandArgs[len(commandArgs)-1]
305+
base := filepath.Base(last)
306+
if strings.Contains(base, ".") {
307+
return base
308+
}
309+
310+
return strings.Join(commandArgs, " ")
311+
}
312+
313+
func formatTaskSummary(tasks []*store.Task) string {
314+
if len(tasks) == 0 {
315+
return "✓ Found 0 tasks (waiting for registration)"
316+
}
317+
318+
names := make([]string, 0, len(tasks))
319+
for _, task := range tasks {
320+
names = append(names, task.Name)
321+
}
322+
sort.Strings(names)
323+
324+
plural := ""
325+
if len(names) > 1 {
326+
plural = "s"
327+
}
328+
329+
return fmt.Sprintf("✓ Found %d task%s: %s", len(names), plural, strings.Join(names, ", "))
330+
}

‎pkg/workflows/logs/logs.go‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ package logs
33
import (
44
"context"
55
"fmt"
6-
"os"
6+
"io"
77
"regexp"
88
"slices"
99
"strings"
@@ -151,21 +151,25 @@ func (l *LogStore) Start(ctx context.Context) {
151151
}
152152

153153
type LogInterceptor struct {
154-
file *os.File
154+
writer io.Writer
155155
taskRunID string
156156
logs *LogStore
157157
}
158158

159-
func NewLogInterceptor(taskRunID string, file *os.File, logs *LogStore) *LogInterceptor {
159+
func NewLogInterceptor(taskRunID string, writer io.Writer, logs *LogStore) *LogInterceptor {
160160
return &LogInterceptor{
161-
file: file,
161+
writer: writer,
162162
taskRunID: taskRunID,
163163
logs: logs,
164164
}
165165
}
166166

167167
func (l *LogInterceptor) Write(p []byte) (n int, err error) {
168-
l.file.Write(p)
168+
if l.writer != nil {
169+
if _, err := l.writer.Write(p); err != nil {
170+
return 0, err
171+
}
172+
}
169173
l.logs.AddLog(&Log{
170174
TaskRunID: l.taskRunID,
171175
Message: string(p),

‎pkg/workflows/orchestrator/coordinator.go‎

Lines changed: 31 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,6 @@ import (
99
"github.com/render-oss/cli/pkg/workflows/taskserver"
1010
)
1111

12-
const (
13-
minPort = 10000
14-
maxPort = 20000
15-
)
16-
1712
type TaskNotFoundError struct {
1813
TaskIdentifier string
1914
}
@@ -22,6 +17,13 @@ func (e *TaskNotFoundError) Error() string {
2217
return fmt.Sprintf("task not found: %s", e.TaskIdentifier)
2318
}
2419

20+
type StatusReporter interface {
21+
TaskEnqueued(taskRun *store.TaskRun)
22+
TaskRunning(taskRun *store.TaskRun)
23+
TaskCompleted(taskRun *store.TaskRun)
24+
TaskFailed(taskRun *store.TaskRun)
25+
}
26+
2527
type Coordinator struct {
2628
store *store.TaskStore
2729
sdkExec sdkExec
@@ -32,6 +34,7 @@ type Coordinator struct {
3234
serverFactory serverFactory
3335

3436
topLevelContext context.Context
37+
statusReporter StatusReporter
3538
}
3639

3740
type sdkExec interface {
@@ -47,14 +50,17 @@ type serverFactory interface {
4750
) *taskserver.ServerHandler
4851
}
4952

50-
func NewCoordinator(ctx context.Context, store *store.TaskStore, sdkExec sdkExec, socketTracker *SocketTracker, serverFactory serverFactory) *Coordinator {
51-
return &Coordinator{
53+
func NewCoordinator(ctx context.Context, store *store.TaskStore, sdkExec sdkExec, socketTracker *SocketTracker, serverFactory serverFactory, statusReporter StatusReporter) *Coordinator {
54+
coordinator := &Coordinator{
5255
store: store,
5356
sdkExec: sdkExec,
5457
socketTracker: socketTracker,
5558
serverFactory: serverFactory,
59+
statusReporter: statusReporter,
5660
topLevelContext: ctx,
5761
}
62+
63+
return coordinator
5864
}
5965

6066
func (c *Coordinator) GetSubtaskResult(taskRunID string) (taskserver.PostGetSubtaskResultResponseObject, error) {
@@ -78,7 +84,7 @@ func (c *Coordinator) GetSubtaskResult(taskRunID string) (taskserver.PostGetSubt
7884

7985
return taskserver.PostGetSubtaskResult200JSONResponse{
8086
Complete: complete,
81-
StillRunning: taskRun.Status != store.TaskRunStatusComplete,
87+
StillRunning: taskRun.Status == store.TaskRunStatusRunning,
8288
Error: taskError,
8389
}, nil
8490
}
@@ -102,8 +108,7 @@ func (c *Coordinator) StartSubtaskFunc(parentTaskRunID string) taskserver.StartS
102108

103109
func (c *Coordinator) StartTask(ctx context.Context, taskIdentifier string, input []byte, parentTaskRunID *string) (*store.TaskRun, error) {
104110
// Ensure tasks are up to date
105-
_, err := c.PopulateTasks(ctx)
106-
if err != nil {
111+
if _, err := c.PopulateTasks(ctx); err != nil {
107112
return nil, err
108113
}
109114

@@ -121,6 +126,8 @@ func (c *Coordinator) StartTask(ctx context.Context, taskIdentifier string, inpu
121126

122127
taskRun := c.store.StartTaskRun(taskName, input, parentTaskRunID)
123128

129+
c.statusReporter.TaskEnqueued(taskRun)
130+
124131
server := c.serverFactory.NewHandler(socket, taskserver.GetInput200JSONResponse{
125132
TaskName: taskName,
126133
Input: input,
@@ -134,12 +141,14 @@ func (c *Coordinator) StartTask(ctx context.Context, taskIdentifier string, inpu
134141
return nil, err
135142
}
136143

144+
c.statusReporter.TaskRunning(taskRun)
145+
137146
go func() {
138147
defer cleanupFunc()
139148

140149
callback := <-server.Channels.PostCallback
141150

142-
err := c.completeTask(c.topLevelContext, callback.Body, taskRun.ID)
151+
err := c.completeTask(callback.Body, taskRun.ID)
143152
if err != nil {
144153
fmt.Println("error completing task", err)
145154
}
@@ -176,16 +185,24 @@ func (c *Coordinator) PopulateTasks(ctx context.Context) ([]*store.Task, error)
176185
}
177186
}
178187

179-
func (c *Coordinator) completeTask(ctx context.Context, completeBody *taskserver.CallbackRequest, taskRunID string) error {
188+
func (c *Coordinator) completeTask(completeBody *taskserver.CallbackRequest, taskRunID string) error {
180189
taskRun := c.store.GetTaskRun(taskRunID)
181190
if taskRun == nil {
182191
return fmt.Errorf("task run not found")
183192
}
184193

185194
if completeBody.Complete != nil {
186-
c.store.CompleteTaskRun(taskRun.ID, completeBody.Complete.Output)
195+
updated, err := c.store.CompleteTaskRun(taskRun.ID, completeBody.Complete.Output)
196+
if err != nil {
197+
return err
198+
}
199+
c.statusReporter.TaskCompleted(updated)
187200
} else if completeBody.Error != nil {
188-
c.store.FailTaskRun(taskRun.ID, completeBody.Error.Details)
201+
updated, err := c.store.FailTaskRun(taskRun.ID, completeBody.Error.Details)
202+
if err != nil {
203+
return err
204+
}
205+
c.statusReporter.TaskFailed(updated)
189206
}
190207

191208
return nil

‎pkg/workflows/orchestrator/coordinator_test.go‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,15 @@ type fakeServerFactory struct {
2525
newHandler func(socket net.Listener, input taskserver.GetInput200JSONResponse, getSubtaskResultFunc taskserver.GetSubtaskResultFunc, startSubtaskFunc taskserver.StartSubtaskFunc) *taskserver.ServerHandler
2626
}
2727

28+
type noopStatusReporter struct{}
29+
30+
func (noopStatusReporter) Ready() {}
31+
func (noopStatusReporter) TasksRegistered(taskNames []string) {}
32+
func (noopStatusReporter) TaskEnqueued(taskRun *store.TaskRun) {}
33+
func (noopStatusReporter) TaskRunning(taskRun *store.TaskRun) {}
34+
func (noopStatusReporter) TaskCompleted(taskRun *store.TaskRun) {}
35+
func (noopStatusReporter) TaskFailed(taskRun *store.TaskRun) {}
36+
2837
func (f *fakeServerFactory) NewHandler(socket net.Listener, input taskserver.GetInput200JSONResponse, getSubtaskResultFunc taskserver.GetSubtaskResultFunc, startSubtaskFunc taskserver.StartSubtaskFunc) *taskserver.ServerHandler {
2938
return f.newHandler(socket, input, getSubtaskResultFunc, startSubtaskFunc)
3039
}
@@ -89,7 +98,9 @@ func TestStartTask(t *testing.T) {
8998
},
9099
}
91100
},
92-
})
101+
},
102+
&noopStatusReporter{},
103+
)
93104

94105
_, err = coordinator.StartTask(ctx, "fake-workflow/test-task", []byte{}, nil)
95106
require.NoError(t, err)
@@ -193,7 +204,9 @@ func TestStartTaskWithSubtask(t *testing.T) {
193204
},
194205
}
195206
},
196-
})
207+
},
208+
&noopStatusReporter{},
209+
)
197210

198211
// Trigger a task and then we will simulate a subtask
199212
go func() {

‎pkg/workflows/orchestrator/exec.go‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package orchestrator
33
import (
44
"context"
55
"fmt"
6+
"io"
67
"os"
78
"os/exec"
89

@@ -14,6 +15,7 @@ type Exec struct {
1415
logsStore *logs.LogStore
1516
command string
1617
args []string
18+
debug bool
1719
}
1820

1921
type Mode string
@@ -30,20 +32,27 @@ const (
3032

3133
type CleanupFunc func() error
3234

33-
func NewExec(logsStore *logs.LogStore, command string, args ...string) *Exec {
35+
func NewExec(logsStore *logs.LogStore, debug bool, command string, args ...string) *Exec {
3436
return &Exec{
3537
logsStore: logsStore,
3638
command: command,
3739
args: args,
40+
debug: debug,
3841
}
3942
}
4043

4144
func (e *Exec) StartService(ctx context.Context, taskRunID string, socketPath string, mode Mode) (CleanupFunc, error) {
4245
cmd := exec.CommandContext(ctx, e.command, e.args...)
4346
cmd.Env = append(os.Environ(), fmt.Sprintf(SocketPathEnv+"=%s", socketPath), fmt.Sprintf(ModeEnv+"=%s", mode))
4447

45-
stdOutInterceptor := logs.NewLogInterceptor(taskRunID, os.Stdout, e.logsStore)
46-
stdErrInterceptor := logs.NewLogInterceptor(taskRunID, os.Stderr, e.logsStore)
48+
stdoutWriter := io.Writer(os.Stdout)
49+
stderrWriter := io.Writer(os.Stderr)
50+
if !e.debug {
51+
stdoutWriter = io.Discard
52+
stderrWriter = io.Discard
53+
}
54+
stdOutInterceptor := logs.NewLogInterceptor(taskRunID, stdoutWriter, e.logsStore)
55+
stdErrInterceptor := logs.NewLogInterceptor(taskRunID, stderrWriter, e.logsStore)
4756

4857
cmd.Stdout = stdOutInterceptor
4958
cmd.Stderr = stdErrInterceptor

0 commit comments

Comments
 (0)