back round 0 view raw

spindle: stream logs from disk #261

merged

opened by

anirudh.fi 5 months ago targeting master from push-vwyomovpppwp

Signed-off-by: Anirudh Oppiliappan anirudh@tangled.sh

options

unified split

Changed files

+85 -21

spindle

server.go

stream.go

spindle/server.go

···

       148
       148
        
       		w.Write([]byte(s.cfg.Server.Owner))

     

       149
       149
        
       	})

     

       150
       150
        
       	mux.HandleFunc("/logs/{knot}/{rkey}/{name}", s.Logs)

     

       151
       151
       +
       	mux.HandleFunc("/logs/{knot}/{rkey}/{name}/{idx}", s.StepLogs)

     

       151
       152
        
       	return mux

     

       152
       153
        
       }

     

       153
       154

+84 -21

spindle/stream.go

···

       1
       1
        
       package spindle

     

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
       +
       	"bufio"

     

       4
       5
        
       	"context"

     

       5
       6
        
       	"encoding/json"

     

       6
       7
        
       	"fmt"

     

       7
       8
        
       	"net/http"

     

       8
       9
        
       	"strconv"

     

       10
       10
       +
       	"strings"

     

       9
       11
        
       	"time"

     

       10
       12
        
       

     

       13
       13
       +
       	"tangled.sh/tangled.sh/core/spindle/engine"

     

       11
       14
        
       	"tangled.sh/tangled.sh/core/spindle/models"

     

       12
       15
        
       

     

       13
       16
        
       	"github.com/go-chi/chi/v5"

     
···

       88
       91
        
       }

     

       89
       92
        
       

     

       90
       93
        
       func (s *Spindle) Logs(w http.ResponseWriter, r *http.Request) {

     

       91
       91
       -
       	l := s.l.With("handler", "Logs")

     

       92
       92
       -
       

     

       93
       93
       -
       	knot := chi.URLParam(r, "knot")

     

       94
       94
       -
       	if knot == "" {

     

       95
       95
       -
       		http.Error(w, "knot required", http.StatusBadRequest)

     

       94
       94
       +
       	wid, err := getWorkflowID(r)

     

       95
       95
       +
       	if err != nil {

     

       96
       96
       +
       		http.Error(w, err.Error(), http.StatusBadRequest)

     

       96
       97
        
       		return

     

       97
       98
        
       	}

     

       98
       99
        
       

     

       99
       99
       -
       	rkey := chi.URLParam(r, "rkey")

     

       100
       100
       -
       	if rkey == "" {

     

       101
       101
       -
       		http.Error(w, "rkey required", http.StatusBadRequest)

     

       100
       100
       +
       	s.handleLogStream(w, r, func(ctx context.Context, conn *websocket.Conn) error {

     

       101
       101
       +
       		return s.streamLogs(ctx, conn, wid)

     

       102
       102
       +
       	})

     

       103
       103
       +
       }

     

       104
       104
       +
       

     

       105
       105
       +
       func (s *Spindle) StepLogs(w http.ResponseWriter, r *http.Request) {

     

       106
       106
       +
       	wid, err := getWorkflowID(r)

     

       107
       107
       +
       	if err != nil {

     

       108
       108
       +
       		http.Error(w, err.Error(), http.StatusBadRequest)

     

       102
       109
        
       		return

     

       103
       110
        
       	}

     

       104
       111
        
       

     

       105
       105
       -
       	name := chi.URLParam(r, "name")

     

       106
       106
       -
       	if name == "" {

     

       107
       107
       -
       		http.Error(w, "name required", http.StatusBadRequest)

     

       112
       112
       +
       	idxStr := chi.URLParam(r, "idx")

     

       113
       113
       +
       	if idxStr == "" {

     

       114
       114
       +
       		http.Error(w, "step index required", http.StatusBadRequest)

     

       108
       115
        
       		return

     

       109
       116
        
       	}

     

       110
       110
       -
       

     

       111
       111
       -
       	wid := models.WorkflowId{

     

       112
       112
       -
       		PipelineId: models.PipelineId{

     

       113
       113
       -
       			Knot: knot,

     

       114
       114
       -
       			Rkey: rkey,

     

       115
       115
       -
       		},

     

       116
       116
       -
       		Name: name,

     

       117
       117
       +
       	idx, err := strconv.Atoi(idxStr)

     

       118
       118
       +
       	if err != nil {

     

       119
       119
       +
       		http.Error(w, "bad step index", http.StatusBadRequest)

     

       120
       120
       +
       		return

     

       117
       121
        
       	}

     

       118
       122
        
       

     

       119
       119
       -
       	l = l.With("knot", knot, "rkey", rkey, "name", name)

     

       123
       123
       +
       	s.handleLogStream(w, r, func(ctx context.Context, conn *websocket.Conn) error {

     

       124
       124
       +
       		return s.streamLogFromDisk(ctx, conn, wid, idx)

     

       125
       125
       +
       	})

     

       126
       126
       +
       }

     

       127
       127
       +
       

     

       128
       128
       +
       func (s *Spindle) handleLogStream(w http.ResponseWriter, r *http.Request, streamFn func(ctx context.Context, conn *websocket.Conn) error) {

     

       129
       129
       +
       	l := s.l.With("handler", "Logs")

     

       120
       130
        
       

     

       121
       131
        
       	conn, err := upgrader.Upgrade(w, r, nil)

     

       122
       132
        
       	if err != nil {

     
···

       140
       150
        
       		}

     

       141
       151
        
       	}()

     

       142
       152
        
       

     

       143
       143
       -
       	if err := s.streamLogs(ctx, conn, wid); err != nil {

     

       144
       144
       -
       		l.Error("streamLogs failed", "err", err)

     

       153
       153
       +
       	if err := streamFn(ctx, conn); err != nil {

     

       154
       154
       +
       		l.Error("log stream failed", "err", err)

     

       145
       155
        
       	}

     

       146
       156
        
       	l.Debug("logs connection closed")

     

       147
       157
        
       }

     
···

       206
       216
        
       	return nil

     

       207
       217
        
       }

     

       208
       218
        
       

     

       219
       219
       +
       func (s *Spindle) streamLogFromDisk(ctx context.Context, conn *websocket.Conn, wid models.WorkflowId, stepIdx int) error {

     

       220
       220
       +
       	streams := []string{"stdout", "stderr"}

     

       221
       221
       +
       

     

       222
       222
       +
       	for _, stream := range streams {

     

       223
       223
       +
       		data, err := engine.ReadStepLog(s.cfg.Pipelines.LogDir, wid.String(), stream, stepIdx)

     

       224
       224
       +
       		if err != nil {

     

       225
       225
       +
       			// log but continue to next stream

     

       226
       226
       +
       			s.l.Error("failed to read step log", "stream", stream, "step", stepIdx, "wid", wid.String(), "err", err)

     

       227
       227
       +
       			continue

     

       228
       228
       +
       		}

     

       229
       229
       +
       

     

       230
       230
       +
       		scanner := bufio.NewScanner(strings.NewReader(data))

     

       231
       231
       +
       		for scanner.Scan() {

     

       232
       232
       +
       			select {

     

       233
       233
       +
       			case <-ctx.Done():

     

       234
       234
       +
       				return ctx.Err()

     

       235
       235
       +
       			default:

     

       236
       236
       +
       				msg := map[string]string{

     

       237
       237
       +
       					"type": stream,

     

       238
       238
       +
       					"data": scanner.Text(),

     

       239
       239
       +
       				}

     

       240
       240
       +
       				if err := conn.WriteJSON(msg); err != nil {

     

       241
       241
       +
       					return err

     

       242
       242
       +
       				}

     

       243
       243
       +
       			}

     

       244
       244
       +
       		}

     

       245
       245
       +
       

     

       246
       246
       +
       		if err := scanner.Err(); err != nil {

     

       247
       247
       +
       			return fmt.Errorf("error scanning %s log: %w", stream, err)

     

       248
       248
       +
       		}

     

       249
       249
       +
       	}

     

       250
       250
       +
       

     

       251
       251
       +
       	return nil

     

       252
       252
       +
       }

     

       253
       253
       +
       

     

       209
       254
        
       func (s *Spindle) streamPipelines(conn *websocket.Conn, cursor *int64) error {

     

       210
       255
        
       	events, err := s.db.GetEvents(*cursor)

     

       211
       256
        
       	if err != nil {

     
···

       242
       287
        
       

     

       243
       288
        
       	return nil

     

       244
       289
        
       }

     

       290
       290
       +
       

     

       291
       291
       +
       func getWorkflowID(r *http.Request) (models.WorkflowId, error) {

     

       292
       292
       +
       	knot := chi.URLParam(r, "knot")

     

       293
       293
       +
       	rkey := chi.URLParam(r, "rkey")

     

       294
       294
       +
       	name := chi.URLParam(r, "name")

     

       295
       295
       +
       

     

       296
       296
       +
       	if knot == "" || rkey == "" || name == "" {

     

       297
       297
       +
       		return models.WorkflowId{}, fmt.Errorf("missing required parameters")

     

       298
       298
       +
       	}

     

       299
       299
       +
       

     

       300
       300
       +
       	return models.WorkflowId{

     

       301
       301
       +
       		PipelineId: models.PipelineId{

     

       302
       302
       +
       			Knot: knot,

     

       303
       303
       +
       			Rkey: rkey,

     

       304
       304
       +
       		},

     

       305
       305
       +
       		Name: name,

     

       306
       306
       +
       	}, nil

     

       307
       307
       +
       }