feat: [1/3] refactor pipeline into shared package with step-aware log streaming by alexvcodesphere · Pull Request #182 · codesphere-cloud/cs-go

alexvcodesphere · 2026-02-20T18:33:44Z

PR 1/3: Pipeline Refactor + Streaming

Extract pipeline stage execution from cli/cmd/start_pipeline.go into a reusable pkg/pipeline package.

Changes

pkg/pipeline/pipeline.go — Runner orchestrates stage execution, discovers sub-steps from IDE server's pipeline status API, streams logs per-step via SSE
api/workspace.go — StreamLogs SSE method for real-time log output
cli/cmd/start_pipeline.go — Refactored to delegate to pipeline.Runner
cli/cmd/client.go — Added StreamLogs to Client interface
pkg/pipeline/pipeline_test.go — 4 streaming unit tests

Merge Order

Merge this first, then PR 2/3, then PR 3/3.

PR 1/3 ← this PR — pipeline refactor + streaming
PR 2/3 — pkg/deploy package
PR 3/3 — cs deploy github CLI command

…ming Extract pipeline stage execution from cli/cmd/start_pipeline.go into a reusable pkg/pipeline package with a Runner that orchestrates stage execution, discovers sub-steps from the IDE server's pipeline status API, and streams logs per-step via SSE. - Add pkg/pipeline with Runner, Client interface, and step discovery - Add StreamLogs SSE method to api/workspace.go for real-time log output - Add StreamLogs to cli/cmd Client interface - Refactor start_pipeline.go to delegate to pipeline.Runner - Add pipeline streaming unit tests (single/multi-step, no-stream cases) - Update .mockery.yml for pipeline mock generation Signed-off-by: Alex <132889147+alexvcodesphere@users.noreply.github.com>

NautiluX

Thanks for opening the PR, I think it should orient itself a bit better on the rest of the codebase, e.g. reusing existing functionality for log streaming instead of implementing it again.
In general I think functions tend to be lengthy and overly complex, I recommend trying to simplify them for improved readability. The handling of the log streaming with timeout specifically looks very complex. I know it's not easy to get parallel code working without losing readability, but I think you can achieve a bit more simplicity.

NautiluX · 2026-03-05T11:48:42Z

api/workspace.go

+// log entries to the provided writer until the context is cancelled or the
+// stream ends. This is used during pipeline execution to provide real-time
+// log output.
+func (c *Client) StreamLogs(ctx context.Context, apiUrl string, wsId int, stage string, step int, w io.Writer) error {


We already have a pretty similar function here:

cs-go/cli/cmd/log.go

Line 171 in 900cc71

func printLogsOfEndpoint(prefix string, endpoint string) error {

it's not beautiful either but we shouldn't reimplement it but reuse/refactor.

NautiluX · 2026-03-05T11:50:28Z

pkg/pipeline/pipeline.go

+	for _, stage := range stages {
+		// Sync the landscape before the run stage
+		if stage == "run" {
+			fmt.Println("  🔄 Syncing landscape...")


please use the log framework instead of fmt

NautiluX · 2026-03-05T11:51:50Z

pkg/pipeline/pipeline.go

+	// Step-aware log streaming for non-run stages.
+	// Each step gets its own context; when a new step is discovered the
+	// previous step's stream is cancelled and drained before moving on.
+	streamEnabled := stage != "run" && cfg.ApiUrl != ""


how is the apiurl related to the log streaming?

NautiluX · 2026-03-05T11:53:25Z

pkg/pipeline/pipeline.go

+	err := r.waitForStageWithStepCallback(wsId, stage, cfg, startStreamForStep)
+
+	// Drain final step's logs
+	drainStream()


why not defer?

NautiluX · 2026-03-05T11:55:27Z

pkg/pipeline/pipeline_test.go

+				mockClient.EXPECT().GetPipelineState(wsId, "prepare").Return([]api.PipelineStatus{
+					statusWithSteps("codesphere-ide", "success", "success"),
+				}, nil).NotBefore(startCall)
+				// StreamLogs should NOT be called — mockery will fail if it is


Suggested change

// StreamLogs should NOT be called — mockery will fail if it is

// StreamLogs should NOT be called

NautiluX · 2026-03-05T11:57:25Z

pkg/pipeline/pipeline.go

+		go func() {
+			defer stepWg.Done()
+			if err := r.Client.StreamLogs(ctx, cfg.ApiUrl, wsId, stage, step, os.Stdout); err != nil {
+				_, _ = fmt.Fprintf(os.Stderr, "⚠ log stream error (step %d): %v\n", step, err)


what is the expected behavior of this? We just ignore the error and go to the next step?

NautiluX · 2026-03-05T12:00:03Z

pkg/pipeline/pipeline.go

+	startStreamForStep := func(step int, totalSteps int) {
+		if !streamEnabled || step <= streamingStep {
+			return
+		}
+
+		// Drain previous step before starting next
+		drainStream()
+
+		streamingStep = step
+		fmt.Printf("\n  📋 Step %d/%d\n", step+1, totalSteps)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		stepCancel = cancel
+		stepWg.Add(1)
+		go func() {
+			defer stepWg.Done()
+			if err := r.Client.StreamLogs(ctx, cfg.ApiUrl, wsId, stage, step, os.Stdout); err != nil {
+				_, _ = fmt.Fprintf(os.Stderr, "⚠ log stream error (step %d): %v\n", step, err)
+			}
+		}()
+	}


why is this function defined inside of the function? I think it deserves a private but real function for improved readability of the runStage function.

NautiluX · 2026-03-05T12:03:32Z

pkg/pipeline/pipeline.go

+		if onStep != nil {
+			for _, s := range status {
+				if s.Server == IdeServer {
+					total := len(s.Steps)
+					for i, step := range s.Steps {
+						if step.State == "running" || step.State == "success" {
+							onStep(i, total)
+						}
+					}
+					break
+				}
+			}
+		}


I wonder if we can find a simpler way of implementing this functionality.

NautiluX · 2026-03-05T12:05:51Z

pkg/pipeline/pipeline.go

+		ctx, cancel := context.WithCancel(context.Background())
+		stepCancel = cancel
+		stepWg.Add(1)
+		go func() {
+			defer stepWg.Done()
+			if err := r.Client.StreamLogs(ctx, cfg.ApiUrl, wsId, stage, step, os.Stdout); err != nil {
+				_, _ = fmt.Fprintf(os.Stderr, "⚠ log stream error (step %d): %v\n", step, err)
+			}
+		}()


so there are 2 things happening in parallel here, on one hand we start the stream and for the timeout we have the waiting function that would cancel this go func when the timeout is reached, right?
I'm wondering if we can not achieve the intended behavior without this go func, or hiding the cancel after time X inside the streamlogs function that anyways takes a context here.

NautiluX · 2026-03-05T12:08:40Z

pkg/pipeline/pipeline.go

+		stepWg.Add(1)
+		go func() {
+			defer stepWg.Done()
+			if err := r.Client.StreamLogs(ctx, cfg.ApiUrl, wsId, stage, step, os.Stdout); err != nil {


why does this endpoint need an API endpoint but none of the others?

alexvcodesphere requested a review from a team as a code owner February 20, 2026 18:33

This was referenced Feb 20, 2026

feat: [2/3] add provider-agnostic deploy package for preview environments #183

Open

feat(cli): [3/3] add cs deploy github command for PR deployments #184

Open

alexvcodesphere changed the title ~~feat: refactor pipeline into shared package with step-aware log streaming~~ feat: [1/3] refactor pipeline into shared package with step-aware log streaming Feb 20, 2026

NautiluX reviewed Mar 5, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: [1/3] refactor pipeline into shared package with step-aware log streaming#182

feat: [1/3] refactor pipeline into shared package with step-aware log streaming#182
alexvcodesphere wants to merge 1 commit intocodesphere-cloud:mainfrom
alexvcodesphere:feat/1-pipeline-refactor-streaming

alexvcodesphere commented Feb 20, 2026

Uh oh!

NautiluX left a comment

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

NautiluX Mar 5, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	// StreamLogs should NOT be called — mockery will fail if it is
	// StreamLogs should NOT be called

Conversation

alexvcodesphere commented Feb 20, 2026

PR 1/3: Pipeline Refactor + Streaming

Changes

Merge Order

Uh oh!

NautiluX left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants