Merge pull request #490 from travis-ci/igor-log-part-received-at

measure log processing SLI

amqp_job.go

@@ -134,7 +134,7 @@ func (j *amqpJob) LogWriter(ctx gocontext.Context, defaultLogTimeout time.Durati
 		logTimeout = defaultLogTimeout
 	}
 
-	return newAMQPLogWriter(ctx, j.logWriterChan, j.payload.Job.ID, logTimeout, j.withLogSharding)
+	return newAMQPLogWriter(ctx, j.logWriterChan, j.payload.Job.ID, j.Payload().Job.QueuedAt, logTimeout, j.withLogSharding)
 }
 
 func (j *amqpJob) createStateUpdateBody(ctx gocontext.Context, state string) map[string]interface{} {
@@ -186,7 +186,9 @@ func (j *amqpJob) sendStateUpdate(ctx gocontext.Context, event, state string) er
 	return err.(error)
 }
 
-func (j *amqpJob) SetupContext(ctx gocontext.Context) gocontext.Context { return ctx }
+func (j *amqpJob) SetupContext(ctx gocontext.Context) gocontext.Context {
+	return ctx
+}
 
 func (j *amqpJob) Name() string { return "amqp" }
 

amqp_log_writer.go

@@ -16,23 +16,26 @@ import (
 )
 
 type amqpLogPart struct {
-	JobID   uint64 `json:"id"`
-	Content string `json:"log"`
-	Number  int    `json:"number"`
-	UUID    string `json:"uuid"`
-	Final   bool   `json:"final"`
+	JobID    uint64     `json:"id"`
+	Content  string     `json:"log"`
+	Number   int        `json:"number"`
+	UUID     string     `json:"uuid"`
+	Final    bool       `json:"final"`
+	QueuedAt *time.Time `json:"queued_at,omitempty"`
 }
 
 type amqpLogWriter struct {
-	ctx     gocontext.Context
-	jobID   uint64
-	sharded bool
+	ctx         gocontext.Context
+	jobID       uint64
+	jobQueuedAt *time.Time
+	sharded     bool
 
 	closeChan chan struct{}
 
 	bufferMutex   sync.Mutex
 	buffer        *bytes.Buffer
 	logPartNumber int
+	jobStarted    bool
 
 	bytesWritten int
 	maxLength    int
@@ -44,17 +47,18 @@ type amqpLogWriter struct {
 	timeout time.Duration
 }
 
-func newAMQPLogWriter(ctx gocontext.Context, logWriterChan *amqp.Channel, jobID uint64, timeout time.Duration, sharded bool) (*amqpLogWriter, error) {
+func newAMQPLogWriter(ctx gocontext.Context, logWriterChan *amqp.Channel, jobID uint64, jobQueuedAt *time.Time, timeout time.Duration, sharded bool) (*amqpLogWriter, error) {
 
 	writer := &amqpLogWriter{
-		ctx:       context.FromComponent(ctx, "log_writer"),
-		amqpChan:  logWriterChan,
-		jobID:     jobID,
-		closeChan: make(chan struct{}),
-		buffer:    new(bytes.Buffer),
-		timer:     time.NewTimer(time.Hour),
-		timeout:   timeout,
-		sharded:   sharded,
+		ctx:         context.FromComponent(ctx, "log_writer"),
+		amqpChan:    logWriterChan,
+		jobID:       jobID,
+		jobQueuedAt: jobQueuedAt,
+		closeChan:   make(chan struct{}),
+		buffer:      new(bytes.Buffer),
+		timer:       time.NewTimer(time.Hour),
+		timeout:     timeout,
+		sharded:     sharded,
 	}
 
 	context.LoggerFromContext(ctx).WithFields(logrus.Fields{
@@ -127,6 +131,10 @@ func (w *amqpLogWriter) SetMaxLogLength(bytes int) {
 	w.maxLength = bytes
 }
 
+func (w *amqpLogWriter) SetJobStarted() {
+	w.jobStarted = true
+}
+
 // WriteAndClose works like a Write followed by a Close, but ensures that no
 // other Writes are allowed in between.
 func (w *amqpLogWriter) WriteAndClose(p []byte) (int, error) {
@@ -222,6 +230,15 @@ func (w *amqpLogWriter) flush() {
 func (w *amqpLogWriter) publishLogPart(part amqpLogPart) error {
 	part.UUID, _ = context.UUIDFromContext(w.ctx)
 
+	// we emit the queued_at field on the log part to indicate that
+	// this is when the job started running. downstream consumers of
+	// the log parts (travis-logs) can then use the timestamp to compute
+	// a "time to first log line" metric.
+	if w.jobStarted {
+		part.QueuedAt = w.jobQueuedAt
+		w.jobStarted = false
+	}
+
 	partBody, err := json.Marshal(part)
 	if err != nil {
 		return err

amqp_log_writer_factory.go

@@ -50,7 +50,7 @@ func (l *AMQPLogWriterFactory) LogWriter(ctx gocontext.Context, defaultLogTimeou
 		logTimeout = defaultLogTimeout
 	}
 
-	return newAMQPLogWriter(ctx, l.logWriterChan, job.Payload().Job.ID, logTimeout, l.withLogSharding)
+	return newAMQPLogWriter(ctx, l.logWriterChan, job.Payload().Job.ID, job.Payload().Job.QueuedAt, logTimeout, l.withLogSharding)
 }
 
 func (l *AMQPLogWriterFactory) Cleanup() error {

amqp_log_writer_test.go

@@ -17,9 +17,10 @@ func TestAMQPLogWriterWrite(t *testing.T) {
 	defer amqpChan.Close()
 
 	uuid := uuid.NewRandom()
+	queuedAt := time.Now()
 	ctx := workerctx.FromUUID(context.TODO(), uuid.String())
 
-	logWriter, err := newAMQPLogWriter(ctx, amqpChan, 4, time.Hour, false)
+	logWriter, err := newAMQPLogWriter(ctx, amqpChan, 4, &queuedAt, time.Hour, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -74,9 +75,10 @@ func TestAMQPLogWriterClose(t *testing.T) {
 	defer amqpChan.Close()
 
 	uuid := uuid.NewRandom()
+	queuedAt := time.Now()
 	ctx := workerctx.FromUUID(context.TODO(), uuid.String())
 
-	logWriter, err := newAMQPLogWriter(ctx, amqpChan, 4, time.Hour, false)
+	logWriter, err := newAMQPLogWriter(ctx, amqpChan, 4, &queuedAt, time.Hour, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -122,9 +124,10 @@ func TestAMQPMaxLogLength(t *testing.T) {
 	defer amqpChan.Close()
 
 	uuid := uuid.NewRandom()
+	queuedAt := time.Now()
 	ctx := workerctx.FromUUID(context.TODO(), uuid.String())
 
-	logWriter, err := newAMQPLogWriter(ctx, amqpChan, 4, time.Hour, false)
+	logWriter, err := newAMQPLogWriter(ctx, amqpChan, 4, &queuedAt, time.Hour, false)
 	if err != nil {
 		t.Fatal(err)
 	}

file_log_writer.go

@@ -44,6 +44,8 @@ func (w *fileLogWriter) SetMaxLogLength(n int) {
 	return
 }
 
+func (w *fileLogWriter) SetJobStarted() {}
+
 func (w *fileLogWriter) Timeout() <-chan time.Time {
 	return w.timer.C
 }

http_log_writer.go

@@ -130,6 +130,8 @@ func (w *httpLogWriter) SetMaxLogLength(bytes int) {
 	w.maxLength = bytes
 }
 
+func (w *httpLogWriter) SetJobStarted() {}
+
 func (w *httpLogWriter) WriteAndClose(p []byte) (int, error) {
 	if w.closed() {
 		return 0, fmt.Errorf("log already closed")

log_writer.go

@@ -45,4 +45,5 @@ type LogWriter interface {
 	WriteAndClose([]byte) (int, error)
 	Timeout() <-chan time.Time
 	SetMaxLogLength(int)
+	SetJobStarted()
 }

package_test.go

@@ -146,3 +146,5 @@ func (flw *fakeLogWriter) Timeout() <-chan time.Time {
 }
 
 func (flw *fakeLogWriter) SetMaxLogLength(_ int) {}
+
+func (flw *fakeLogWriter) SetJobStarted() {}

step_update_state.go

@@ -17,6 +17,7 @@ func (s *stepUpdateState) Run(state multistep.StateBag) multistep.StepAction {
 	buildJob := state.Get("buildJob").(Job)
 	instance := state.Get("instance").(backend.Instance)
 	processedAt := state.Get("processedAt").(time.Time)
+	logWriter := state.Get("logWriter").(LogWriter)
 
 	logger := context.LoggerFromContext(ctx).WithField("self", "step_update_state")
 
@@ -26,6 +27,8 @@ func (s *stepUpdateState) Run(state multistep.StateBag) multistep.StepAction {
 		state.Put("ctx", ctx)
 	}
 
+	logWriter.SetJobStarted()
+
 	err := buildJob.Started(ctx)
 	if err != nil {
 		context.LoggerFromContext(ctx).WithFields(logrus.Fields{

step_write_worker_info_test.go

@@ -32,6 +32,9 @@ func (w *byteBufferLogWriter) Timeout() <-chan time.Time {
 func (w *byteBufferLogWriter) SetMaxLogLength(m int) {
 }
 
+func (w *byteBufferLogWriter) SetJobStarted() {
+}
+
 func setupStepWriteWorkerInfo() (*stepWriteWorkerInfo, *byteBufferLogWriter, multistep.StateBag) {
 	s := &stepWriteWorkerInfo{}