Rework log streaming and related functions

agent/logger.go

@@ -50,9 +50,8 @@ 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 {
+		logStream := rpc.NewLineWriter(r.client, step.UUID, secrets...)
+		if _, err := io.Copy(logStream, part); err != nil {
 			log.Error().Err(err).Msg("copy limited logStream part")
 		}
 

agent/rpc/client_grpc.go

@@ -278,14 +278,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.LogEntry = new(proto.LogEntry)
+	req.LogEntry.StepUuid = logEntry.StepUUID
+	req.LogEntry.Data = logEntry.Data
+	req.LogEntry.Line = int32(logEntry.Line)
+	req.LogEntry.Time = logEntry.Time
+	req.LogEntry.Type = int32(logEntry.Type)
 	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/exec.go

@@ -248,7 +248,7 @@ var defaultLogger = pipeline.LogFunc(func(step *backendTypes.Step, rc multipart.
 		return err
 	}
 
-	logStream := NewLineWriter(step.Alias)
+	logStream := NewLineWriter(step.Alias, step.UUID)
 	_, err = io.Copy(logStream, part)
 	return err
 })

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
+	stepName string
+	stepUUID string
+	num      int
+	now      time.Time
+	rep      *strings.Replacer
+	lines    []*rpc.LogEntry
 }
 
 // NewLineWriter returns a new line reader.
-func NewLineWriter(name string) *LineWriter {
-	w := new(LineWriter)
-	w.name = name
-	w.num = 0
-	w.now = time.Now().UTC()
-
-	return w
+func NewLineWriter(stepName, stepUUID string) *LineWriter {
+	return &LineWriter{
+		stepName: stepName,
+		stepUUID: stepUUID,
+		now:      time.Now().UTC(),
+	}
 }
 
 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,
+		StepUUID: w.stepUUID,
+		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, "[%s:L%d:%ds] %s", w.stepName, w.num, int64(time.Since(w.now).Seconds()), data)
 
 	w.num++
 

cli/pipeline/logs.go

@@ -27,7 +27,7 @@ import (
 var pipelineLogsCmd = &cli.Command{
 	Name:      "logs",
 	Usage:     "show pipeline logs",
-	ArgsUsage: "<repo/name> [pipeline] [step]",
+	ArgsUsage: "<repo/name> [pipeline] [stepID]",
 	Action:    pipelineLogs,
 	Flags:     common.GlobalFlags,
 }
@@ -54,13 +54,13 @@ func pipelineLogs(c *cli.Context) error {
 		return err
 	}
 
-	logs, err := client.PipelineLogs(owner, name, number, step)
+	logs, err := client.StepLogEntries(owner, name, number, step)
 	if err != nil {
 		return err
 	}
 
 	for _, log := range logs {
-		fmt.Print(log.Output)
+		fmt.Print(string(log.Data))
 	}
 
 	return nil