Rework log streaming and related functions

agent/logger.go

@@ -50,9 +50,9 @@ func (r *Runner) createLogger(_ context.Context, logger zerolog.Logger, uploads
 
 		loglogger.Debug().Msg("log stream opened")
 
-		limitedPart := io.LimitReader(part, maxLogsUpload)
-		logStream := rpc.NewLineWriter(r.client, work.ID, step.Alias, secrets...)
-		if _, err := io.Copy(logStream, limitedPart); err != nil {
+		// TODO: use step-id instead of work-id
+		logStream := rpc.NewLineWriter(r.client, work.ID, secrets...)
+		if _, err := io.Copy(logStream, part); err != nil {
 			log.Error().Err(err).Msg("copy limited logStream part")
 		}
 

agent/rpc/client_grpc.go

@@ -17,6 +17,7 @@ package rpc
 import (
 	"context"
 	"encoding/json"
+	"strconv"
 	"strings"
 	"time"
 
@@ -278,14 +279,14 @@ func (c *client) Update(ctx context.Context, id string, state rpc.State) (err er
 }
 
 // Log writes the pipeline log entry.
-func (c *client) Log(ctx context.Context, id string, line *rpc.Line) (err error) {
+func (c *client) Log(ctx context.Context, logEntry *rpc.LogEntry) (err error) {
 	req := new(proto.LogRequest)
-	req.Id = id
-	req.Line = new(proto.Line)
-	req.Line.Out = line.Out
-	req.Line.Pos = int32(line.Pos)
-	req.Line.Step = line.Step
-	req.Line.Time = line.Time
+	req.Id = strconv.FormatInt(logEntry.StepID, 10)
+	req.LogEntry = new(proto.LogEntry)
+	req.LogEntry.StepId = logEntry.StepID
+	req.LogEntry.Data = logEntry.Data
+	req.LogEntry.Line = int32(logEntry.Line)
+	req.LogEntry.Time = logEntry.Time
 	for {
 		_, err = c.client.Log(ctx, req)
 		if err == nil {

agent/runner.go

@@ -31,14 +31,6 @@ import (
 	"github.com/woodpecker-ci/woodpecker/shared/utils"
 )
 
-// TODO: Implement log streaming.
-// Until now we need to limit the size of the logs and files that we upload.
-// The maximum grpc payload size is 4194304. So we need to set these limits below the maximum.
-const (
-	maxLogsUpload = 2000000 // this is per step
-	maxFileUpload = 1000000
-)
-
 type Runner struct {
 	client   rpc.Peer
 	filter   rpc.Filter

cli/exec/line.go

@@ -19,60 +19,44 @@ import (
 	"os"
 	"strings"
 	"time"
-)
 
-// Identifies the type of line in the logs.
-const (
-	LineStdout int = iota
-	LineStderr
-	LineExitCode
-	LineMetadata
-	LineProgress
+	"github.com/woodpecker-ci/woodpecker/pipeline/rpc"
 )
 
-// Line is a line of console output.
-type Line struct {
-	Step string `json:"step,omitempty"`
-	Time int64  `json:"time,omitempty"`
-	Type int    `json:"type,omitempty"`
-	Pos  int    `json:"pos,omitempty"`
-	Out  string `json:"out,omitempty"`
-}
-
 // LineWriter sends logs to the client.
 type LineWriter struct {
-	name  string
-	num   int
-	now   time.Time
-	rep   *strings.Replacer
-	lines []*Line
+	stepID int64
+	num    int
+	now    time.Time
+	rep    *strings.Replacer
+	lines  []*rpc.LogEntry
 }
 
 // NewLineWriter returns a new line reader.
-func NewLineWriter(name string) *LineWriter {
+func NewLineWriter(stepID int64) *LineWriter {
 	w := new(LineWriter)
-	w.name = name
+	w.stepID = stepID
 	w.num = 0
 	w.now = time.Now().UTC()
 
 	return w
 }
 
 func (w *LineWriter) Write(p []byte) (n int, err error) {
-	out := string(p)
+	data := string(p)
 	if w.rep != nil {
-		out = w.rep.Replace(out)
+		data = w.rep.Replace(data)
 	}
 
-	line := &Line{
-		Out:  out,
-		Step: w.name,
-		Pos:  w.num,
-		Time: int64(time.Since(w.now).Seconds()),
-		Type: LineStdout,
+	line := &rpc.LogEntry{
+		Data:   data,
+		StepID: w.stepID,
+		Line:   w.num,
+		Time:   int64(time.Since(w.now).Seconds()),
+		Type:   rpc.LogEntryStdout,
 	}
 
-	fmt.Fprintf(os.Stderr, "[%s:L%d:%ds] %s", w.name, w.num, int64(time.Since(w.now).Seconds()), out)
+	fmt.Fprintf(os.Stderr, "[%d:L%d:%ds] %s", w.stepID, w.num, int64(time.Since(w.now).Seconds()), data)
 
 	w.num++
 

pipeline/rpc/line.go → pipeline/rpc/log_entry.go

@@ -28,70 +28,67 @@ import (
 
 // Identifies the type of line in the logs.
 const (
-	LineStdout int = iota
-	LineStderr
-	LineExitCode
-	LineMetadata
-	LineProgress
+	LogEntryStdout int = iota
+	LogEntryStderr
+	LogEntryExitCode
+	LogEntryMetadata
+	LogEntryProgress
 )
 
 // Line is a line of console output.
-type Line struct {
-	Step string `json:"step,omitempty"`
-	Time int64  `json:"time,omitempty"`
-	Type int    `json:"type,omitempty"`
-	Pos  int    `json:"pos,omitempty"`
-	Out  string `json:"out,omitempty"`
+type LogEntry struct {
+	StepID int64  `json:"step_id,omitempty"`
+	Time   int64  `json:"time,omitempty"`
+	Type   int    `json:"type,omitempty"`
+	Line   int    `json:"line,omitempty"`
+	Data   string `json:"data,omitempty"`
 }
 
-func (l *Line) String() string {
+func (l *LogEntry) String() string {
 	switch l.Type {
-	case LineExitCode:
-		return fmt.Sprintf("[%s] exit code %s", l.Step, l.Out)
+	case LogEntryExitCode:
+		return fmt.Sprintf("[%d] exit code %s", l.StepID, l.Data)
 	default:
-		return fmt.Sprintf("[%s:L%v:%vs] %s", l.Step, l.Pos, l.Time, l.Out)
+		return fmt.Sprintf("[%d:L%v:%vs] %s", l.StepID, l.Line, l.Time, l.Data)
 	}
 }
 
 // LineWriter sends logs to the client.
 type LineWriter struct {
-	peer  Peer
-	id    string
-	name  string
-	num   int
-	now   time.Time
-	rep   *strings.Replacer
-	lines []*Line
+	peer   Peer
+	stepID int64
+	num    int
+	now    time.Time
+	rep    *strings.Replacer
+	lines  []*LogEntry
 }
 
 // NewLineWriter returns a new line reader.
-func NewLineWriter(peer Peer, id, name string, secret ...string) *LineWriter {
+func NewLineWriter(peer Peer, stepID int64, secret ...string) *LineWriter {
 	return &LineWriter{
-		peer:  peer,
-		id:    id,
-		name:  name,
-		now:   time.Now().UTC(),
-		rep:   shared.NewSecretsReplacer(secret),
-		lines: nil,
+		peer:   peer,
+		stepID: stepID,
+		now:    time.Now().UTC(),
+		rep:    shared.NewSecretsReplacer(secret),
+		lines:  nil,
 	}
 }
 
 func (w *LineWriter) Write(p []byte) (n int, err error) {
-	out := string(p)
+	data := string(p)
 	if w.rep != nil {
-		out = w.rep.Replace(out)
+		data = w.rep.Replace(data)
 	}
-	log.Trace().Str("name", w.name).Str("ID", w.id).Msgf("grpc write line: %s", out)
+	log.Trace().Int64("step-id", w.stepID).Msgf("grpc write line: %s", data)
 
-	line := &Line{
-		Out:  out,
-		Step: w.name,
-		Pos:  w.num,
-		Time: int64(time.Since(w.now).Seconds()),
-		Type: LineStdout,
+	line := &LogEntry{
+		Data:   data,
+		StepID: w.stepID,
+		Time:   int64(time.Since(w.now).Seconds()),
+		Type:   LogEntryStdout,
 	}
-	if err := w.peer.Log(context.Background(), w.id, line); err != nil {
-		log.Error().Err(err).Msgf("fail to write pipeline log to peer '%s'", w.id)
+	if err := w.peer.Log(context.Background(), line); err != nil {
+		log.Error().Err(err).Msgf("fail to write pipeline log to peer '%d'", w.stepID)
 	}
 	w.num++
 
@@ -111,7 +108,7 @@ func (w *LineWriter) Write(p []byte) (n int, err error) {
 }
 
 // Lines returns the line history
-func (w *LineWriter) Lines() []*Line {
+func (w *LineWriter) Lines() []*LogEntry {
 	return w.lines
 }
 

pipeline/rpc/line_test.go → pipeline/rpc/log_entry_test.go

@@ -18,12 +18,12 @@ import (
 	"testing"
 )
 
-func TestLine(t *testing.T) {
-	line := Line{
-		Step: "redis",
-		Time: 60,
-		Pos:  1,
-		Out:  "starting redis server",
+func TestLogEntry(t *testing.T) {
+	line := LogEntry{
+		StepID: int64(123),
+		Time:   60,
+		Line:   1,
+		Data:   "starting redis server",
 	}
 	got, want := line.String(), "[redis:L1:60s] starting redis server"
 	if got != want {

pipeline/rpc/peer.go

@@ -74,7 +74,7 @@ type Peer interface {
 	Update(c context.Context, id string, state State) error
 
 	// Log writes the pipeline log entry.
-	Log(c context.Context, id string, line *Line) error
+	Log(c context.Context, logEntry *LogEntry) error
 
 	// RegisterAgent register our agent to the server
 	RegisterAgent(ctx context.Context, platform, backend, version string, capacity int) (int64, error)

pipeline/rpc/proto/version.go

@@ -16,4 +16,4 @@ package proto
 
 // Version is the version of the woodpecker.proto file,
 // !IMPORTANT! increased by 1 each time it get changed !IMPORTANT!
-const Version int32 = 2
+const Version int32 = 3