act_runner/runtime/reporter.go

261 lines
5.5 KiB
Go

package runtime
import (
"context"
"fmt"
"strings"
"sync"
"time"
"gitea.com/gitea/act_runner/client"
runnerv1 "gitea.com/gitea/proto-go/runner/v1"
"github.com/avast/retry-go/v4"
"github.com/bufbuild/connect-go"
log "github.com/sirupsen/logrus"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
)
type Reporter struct {
ctx context.Context
closed bool
client client.Client
clientM sync.Mutex
logOffset int
logRows []*runnerv1.LogRow
state *runnerv1.TaskState
stateM sync.RWMutex
}
func NewReporter(ctx context.Context, client client.Client, taskID int64) *Reporter {
return &Reporter{
ctx: ctx,
client: client,
state: &runnerv1.TaskState{
Id: taskID,
},
}
}
func (r *Reporter) ResetSteps(l int) {
r.stateM.Lock()
defer r.stateM.Unlock()
for i := 0; i < l; i++ {
r.state.Steps = append(r.state.Steps, &runnerv1.StepState{
Id: int64(i),
})
}
}
func (r *Reporter) Levels() []log.Level {
return log.AllLevels
}
func (r *Reporter) Fire(entry *log.Entry) error {
r.stateM.Lock()
defer r.stateM.Unlock()
timestamp := entry.Time
if r.state.StartedAt == nil {
r.state.StartedAt = timestamppb.New(timestamp)
}
var step *runnerv1.StepState
if v, ok := entry.Data["stepNumber"]; ok {
if v, ok := v.(int); ok {
step = r.state.Steps[v]
}
}
if step == nil {
if v, ok := entry.Data["jobResult"]; ok {
if v, ok := v.(string); ok {
if jobResult := r.parseResult(v); jobResult != runnerv1.Result_RESULT_UNSPECIFIED {
r.state.Result = jobResult
r.state.StoppedAt = timestamppb.New(timestamp)
for _, s := range r.state.Steps {
if s.Result == runnerv1.Result_RESULT_UNSPECIFIED {
s.Result = runnerv1.Result_RESULT_CANCELLED
}
}
}
}
}
if !r.duringSteps() {
r.logRows = append(r.logRows, r.parseLogRow(entry))
}
return nil
}
if step.StartedAt == nil {
step.StartedAt = timestamppb.New(timestamp)
}
if v, ok := entry.Data["raw_output"]; ok {
if rawOutput, ok := v.(bool); ok && rawOutput {
if step.LogLength == 0 {
step.LogIndex = int64(r.logOffset + len(r.logRows))
}
step.LogLength++
r.logRows = append(r.logRows, r.parseLogRow(entry))
return nil
}
}
log.Info(entry.Data)
if v, ok := entry.Data["stepResult"]; ok {
if v, ok := v.(string); ok {
if stepResult := r.parseResult(v); stepResult != runnerv1.Result_RESULT_UNSPECIFIED {
step.Result = stepResult
step.StoppedAt = timestamppb.New(timestamp)
}
}
}
return nil
}
func (r *Reporter) RunDaemon() {
if r.closed {
return
}
if r.ctx.Err() != nil {
return
}
_ = r.ReportLog(false)
_ = r.ReportState()
time.AfterFunc(time.Second, r.RunDaemon)
}
func (r *Reporter) Logf(format string, a ...interface{}) {
r.stateM.Lock()
defer r.stateM.Unlock()
if !r.duringSteps() {
r.logRows = append(r.logRows, &runnerv1.LogRow{
Time: timestamppb.Now(),
Content: fmt.Sprintf(format, a...),
})
}
}
func (r *Reporter) Close(lastWords string) error {
r.closed = true
r.stateM.Lock()
if r.state.Result == runnerv1.Result_RESULT_UNSPECIFIED {
if lastWords == "" {
lastWords = "Early termination"
}
for _, v := range r.state.Steps {
if v.Result == runnerv1.Result_RESULT_UNSPECIFIED {
v.Result = runnerv1.Result_RESULT_CANCELLED
}
}
r.logRows = append(r.logRows, &runnerv1.LogRow{
Time: timestamppb.Now(),
Content: lastWords,
})
return nil
} else if lastWords != "" {
r.logRows = append(r.logRows, &runnerv1.LogRow{
Time: timestamppb.Now(),
Content: lastWords,
})
}
r.stateM.Unlock()
return retry.Do(func() error {
if err := r.ReportLog(true); err != nil {
return err
}
return r.ReportState()
}, retry.Context(r.ctx))
}
func (r *Reporter) ReportLog(noMore bool) error {
r.clientM.Lock()
defer r.clientM.Unlock()
r.stateM.RLock()
rows := r.logRows
r.stateM.RUnlock()
resp, err := r.client.UpdateLog(r.ctx, connect.NewRequest(&runnerv1.UpdateLogRequest{
TaskId: r.state.Id,
Index: int64(r.logOffset),
Rows: rows,
NoMore: noMore,
}))
if err != nil {
return err
}
ack := int(resp.Msg.AckIndex)
if ack < r.logOffset {
return fmt.Errorf("submitted logs are lost")
}
r.stateM.Lock()
r.logRows = r.logRows[ack-r.logOffset:]
r.logOffset = ack
r.stateM.Unlock()
if noMore && ack < r.logOffset+len(rows) {
return fmt.Errorf("not all logs are submitted")
}
return nil
}
func (r *Reporter) ReportState() error {
r.clientM.Lock()
defer r.clientM.Unlock()
r.stateM.RLock()
state := proto.Clone(r.state).(*runnerv1.TaskState)
r.stateM.RUnlock()
_, err := r.client.UpdateTask(r.ctx, connect.NewRequest(&runnerv1.UpdateTaskRequest{
State: state,
}))
return err
}
func (r *Reporter) duringSteps() bool {
if steps := r.state.Steps; len(steps) == 0 {
return false
} else if first := steps[0]; first.Result == runnerv1.Result_RESULT_UNSPECIFIED && first.LogLength == 0 {
return false
} else if last := steps[len(steps)-1]; last.Result != runnerv1.Result_RESULT_UNSPECIFIED {
return false
}
return true
}
func (r *Reporter) parseResult(s string) runnerv1.Result {
switch s {
case "success":
return runnerv1.Result_RESULT_SUCCESS
case "failure":
return runnerv1.Result_RESULT_FAILURE
case "skipped":
return runnerv1.Result_RESULT_SKIPPED
case "cancelled":
return runnerv1.Result_RESULT_CANCELLED
}
return runnerv1.Result_RESULT_UNSPECIFIED
}
func (r *Reporter) parseLogRow(entry *log.Entry) *runnerv1.LogRow {
return &runnerv1.LogRow{
Time: timestamppb.New(entry.Time),
Content: strings.TrimSuffix(entry.Message, "\r\n"),
}
}