feat(logs): expose request timing breakdown

common/request_timing.go

@@ -0,0 +1,170 @@
+package common
+
+import (
+	"sync"
+	"time"
+
+	"github.com/gin-gonic/gin"
+)
+
+const requestTimingContextKey = "request_timing_session"
+
+type RequestTiming struct {
+	TotalMs             int64  `json:"total_ms"`
+	GatewayMs           *int64 `json:"gateway_ms,omitempty"`
+	UpstreamFirstDataMs *int64 `json:"upstream_first_data_ms,omitempty"`
+	FirstDataToClientMs *int64 `json:"first_data_to_client_ms,omitempty"`
+	ClientStreamMs      *int64 `json:"client_stream_ms,omitempty"`
+	UpstreamResponseMs  *int64 `json:"upstream_response_ms,omitempty"`
+	ResponseWriteMs     *int64 `json:"response_write_ms,omitempty"`
+	UpstreamErrorMs     *int64 `json:"upstream_error_ms,omitempty"`
+	FinalizeMs          *int64 `json:"finalize_ms,omitempty"`
+}
+
+type RequestTimingSession struct {
+	mu                   sync.Mutex
+	start                time.Time
+	firstUpstreamAttempt time.Time
+	firstUpstreamData    time.Time
+	upstreamComplete     time.Time
+	firstClientWrite     time.Time
+	lastClientWrite      time.Time
+	stream               bool
+}
+
+func NewRequestTimingSession(start time.Time) *RequestTimingSession {
+	return &RequestTimingSession{start: start}
+}
+
+func SetRequestTimingSession(c *gin.Context, session *RequestTimingSession) {
+	if c == nil || session == nil {
+		return
+	}
+	c.Set(requestTimingContextKey, session)
+}
+
+func GetRequestTimingSession(c *gin.Context) *RequestTimingSession {
+	if c == nil {
+		return nil
+	}
+	value, exists := c.Get(requestTimingContextKey)
+	if !exists {
+		return nil
+	}
+	session, _ := value.(*RequestTimingSession)
+	return session
+}
+
+func (s *RequestTimingSession) MarkUpstreamAttempt(at time.Time, stream bool) bool {
+	if s == nil {
+		return false
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	resetFirstData := false
+	if s.firstUpstreamAttempt.IsZero() {
+		s.firstUpstreamAttempt = at
+	} else if s.firstClientWrite.IsZero() {
+		s.firstUpstreamData = time.Time{}
+		resetFirstData = true
+	}
+	s.upstreamComplete = time.Time{}
+	s.stream = stream
+	return resetFirstData
+}
+
+func (s *RequestTimingSession) SetStream(stream bool) {
+	if s == nil {
+		return
+	}
+	s.mu.Lock()
+	s.stream = stream
+	s.mu.Unlock()
+}
+
+func (s *RequestTimingSession) MarkFirstUpstreamData(at time.Time) {
+	if s == nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.stream = true
+	if s.firstUpstreamData.IsZero() {
+		s.firstUpstreamData = at
+	}
+}
+
+func (s *RequestTimingSession) MarkClientWrite(startedAt time.Time, completedAt time.Time) {
+	if s == nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.stream && s.firstUpstreamData.IsZero() {
+		return
+	}
+	if !s.stream && s.upstreamComplete.IsZero() {
+		s.upstreamComplete = startedAt
+	}
+	if s.firstClientWrite.IsZero() {
+		s.firstClientWrite = startedAt
+	}
+	s.lastClientWrite = completedAt
+}
+
+func (s *RequestTimingSession) Snapshot(at time.Time, failed bool) *RequestTiming {
+	if s == nil {
+		return nil
+	}
+	s.mu.Lock()
+	start := s.start
+	firstUpstreamAttempt := s.firstUpstreamAttempt
+	firstUpstreamData := s.firstUpstreamData
+	upstreamComplete := s.upstreamComplete
+	firstClientWrite := s.firstClientWrite
+	lastClientWrite := s.lastClientWrite
+	stream := s.stream
+	s.mu.Unlock()
+	if start.IsZero() || at.Before(start) {
+		return nil
+	}
+
+	timing := &RequestTiming{TotalMs: at.Sub(start).Milliseconds()}
+	if firstUpstreamAttempt.IsZero() {
+		if failed {
+			timing.GatewayMs = millisecondsBetween(start, at)
+		}
+		return timing
+	}
+
+	timing.GatewayMs = millisecondsBetween(start, firstUpstreamAttempt)
+	if failed && upstreamComplete.IsZero() && firstUpstreamData.IsZero() {
+		timing.UpstreamErrorMs = millisecondsBetween(firstUpstreamAttempt, at)
+		return timing
+	}
+
+	if stream {
+		timing.UpstreamFirstDataMs = millisecondsBetween(firstUpstreamAttempt, firstUpstreamData)
+		timing.FirstDataToClientMs = millisecondsBetween(firstUpstreamData, firstClientWrite)
+		timing.ClientStreamMs = millisecondsBetween(firstClientWrite, lastClientWrite)
+		timing.FinalizeMs = millisecondsBetween(lastClientWrite, at)
+		return timing
+	}
+
+	timing.UpstreamResponseMs = millisecondsBetween(firstUpstreamAttempt, upstreamComplete)
+	timing.ResponseWriteMs = millisecondsBetween(upstreamComplete, lastClientWrite)
+	if lastClientWrite.IsZero() {
+		timing.FinalizeMs = millisecondsBetween(upstreamComplete, at)
+	} else {
+		timing.FinalizeMs = millisecondsBetween(lastClientWrite, at)
+	}
+	return timing
+}
+
+func millisecondsBetween(start time.Time, end time.Time) *int64 {
+	if start.IsZero() || end.IsZero() || end.Before(start) {
+		return nil
+	}
+	value := end.Sub(start).Milliseconds()
+	return &value
+}

common/request_timing_test.go

@@ -0,0 +1,160 @@
+package common
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func assertMilliseconds(t *testing.T, expected int64, actual *int64) {
+	t.Helper()
+	require.NotNil(t, actual)
+	assert.Equal(t, expected, *actual)
+}
+
+func TestRequestTimingSnapshotForStreamingRequest(t *testing.T) {
+	start := time.Unix(100, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start.Add(10*time.Millisecond), true)
+	session.MarkClientWrite(start.Add(15*time.Millisecond), start.Add(16*time.Millisecond))
+	session.MarkFirstUpstreamData(start.Add(30 * time.Millisecond))
+	session.MarkClientWrite(start.Add(35*time.Millisecond), start.Add(36*time.Millisecond))
+	session.MarkClientWrite(start.Add(60*time.Millisecond), start.Add(61*time.Millisecond))
+
+	timing := session.Snapshot(start.Add(70*time.Millisecond), false)
+
+	require.NotNil(t, timing)
+	assert.Equal(t, int64(70), timing.TotalMs)
+	assertMilliseconds(t, 10, timing.GatewayMs)
+	assertMilliseconds(t, 20, timing.UpstreamFirstDataMs)
+	assertMilliseconds(t, 5, timing.FirstDataToClientMs)
+	assertMilliseconds(t, 26, timing.ClientStreamMs)
+	assertMilliseconds(t, 9, timing.FinalizeMs)
+	assert.Nil(t, timing.UpstreamResponseMs)
+	assert.Nil(t, timing.ResponseWriteMs)
+}
+
+func TestRequestTimingSnapshotForNonStreamingRequest(t *testing.T) {
+	start := time.Unix(200, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start.Add(10*time.Millisecond), false)
+	session.MarkClientWrite(start.Add(40*time.Millisecond), start.Add(45*time.Millisecond))
+	session.MarkClientWrite(start.Add(46*time.Millisecond), start.Add(47*time.Millisecond))
+
+	timing := session.Snapshot(start.Add(50*time.Millisecond), false)
+
+	require.NotNil(t, timing)
+	assert.Equal(t, int64(50), timing.TotalMs)
+	assertMilliseconds(t, 10, timing.GatewayMs)
+	assertMilliseconds(t, 30, timing.UpstreamResponseMs)
+	assertMilliseconds(t, 7, timing.ResponseWriteMs)
+	assertMilliseconds(t, 3, timing.FinalizeMs)
+	assert.Nil(t, timing.UpstreamFirstDataMs)
+	assert.Nil(t, timing.FirstDataToClientMs)
+	assert.Nil(t, timing.ClientStreamMs)
+}
+
+func TestRequestTimingSnapshotForUpstreamErrorBeforeClientWrite(t *testing.T) {
+	start := time.Unix(250, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start.Add(10*time.Millisecond), false)
+
+	timing := session.Snapshot(start.Add(45*time.Millisecond), true)
+
+	require.NotNil(t, timing)
+	assertMilliseconds(t, 10, timing.GatewayMs)
+	assertMilliseconds(t, 35, timing.UpstreamErrorMs)
+	assert.Nil(t, timing.UpstreamResponseMs)
+	assert.Nil(t, timing.FinalizeMs)
+}
+
+func TestRequestTimingSnapshotForUpstreamFailure(t *testing.T) {
+	start := time.Unix(300, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start.Add(10*time.Millisecond), false)
+
+	timing := session.Snapshot(start.Add(45*time.Millisecond), true)
+
+	require.NotNil(t, timing)
+	assert.Equal(t, int64(45), timing.TotalMs)
+	assertMilliseconds(t, 10, timing.GatewayMs)
+	assertMilliseconds(t, 35, timing.UpstreamErrorMs)
+	assert.Nil(t, timing.UpstreamResponseMs)
+	assert.Nil(t, timing.FinalizeMs)
+}
+
+func TestRequestTimingSnapshotForGatewayFailure(t *testing.T) {
+	start := time.Unix(400, 0)
+	session := NewRequestTimingSession(start)
+
+	timing := session.Snapshot(start.Add(20*time.Millisecond), true)
+
+	require.NotNil(t, timing)
+	assert.Equal(t, int64(20), timing.TotalMs)
+	assertMilliseconds(t, 20, timing.GatewayMs)
+	assert.Nil(t, timing.UpstreamErrorMs)
+}
+
+func TestRequestTimingSnapshotIncludesRetriesInUpstreamPhase(t *testing.T) {
+	start := time.Unix(500, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start.Add(10*time.Millisecond), true)
+	session.MarkFirstUpstreamData(start.Add(20 * time.Millisecond))
+	session.MarkUpstreamAttempt(start.Add(30*time.Millisecond), true)
+	session.MarkFirstUpstreamData(start.Add(50 * time.Millisecond))
+	session.MarkClientWrite(start.Add(55*time.Millisecond), start.Add(56*time.Millisecond))
+
+	timing := session.Snapshot(start.Add(60*time.Millisecond), false)
+
+	require.NotNil(t, timing)
+	assertMilliseconds(t, 10, timing.GatewayMs)
+	assertMilliseconds(t, 40, timing.UpstreamFirstDataMs)
+}
+
+func TestRequestTimingFirstDataPromotesResponseToStreaming(t *testing.T) {
+	start := time.Unix(550, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start.Add(10*time.Millisecond), false)
+	session.MarkFirstUpstreamData(start.Add(30 * time.Millisecond))
+	session.MarkClientWrite(start.Add(35*time.Millisecond), start.Add(36*time.Millisecond))
+
+	timing := session.Snapshot(start.Add(40*time.Millisecond), false)
+
+	require.NotNil(t, timing)
+	assertMilliseconds(t, 20, timing.UpstreamFirstDataMs)
+	assert.Nil(t, timing.UpstreamResponseMs)
+}
+
+func TestRequestTimingStreamPromotionIgnoresWritesBeforeFirstData(t *testing.T) {
+	start := time.Unix(575, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start.Add(10*time.Millisecond), false)
+	session.SetStream(true)
+	session.MarkClientWrite(start.Add(15*time.Millisecond), start.Add(16*time.Millisecond))
+	session.MarkFirstUpstreamData(start.Add(30 * time.Millisecond))
+	session.MarkClientWrite(start.Add(35*time.Millisecond), start.Add(36*time.Millisecond))
+
+	timing := session.Snapshot(start.Add(40*time.Millisecond), false)
+
+	require.NotNil(t, timing)
+	assertMilliseconds(t, 5, timing.FirstDataToClientMs)
+	assert.Nil(t, timing.UpstreamResponseMs)
+}
+
+func TestRequestTimingSnapshotPreservesZeroMillisecondPhases(t *testing.T) {
+	start := time.Unix(600, 0)
+	session := NewRequestTimingSession(start)
+	session.MarkUpstreamAttempt(start, false)
+	session.MarkClientWrite(start, start)
+
+	timing := session.Snapshot(start, false)
+
+	require.NotNil(t, timing)
+	assert.Equal(t, int64(0), timing.TotalMs)
+	assertMilliseconds(t, 0, timing.GatewayMs)
+	assertMilliseconds(t, 0, timing.UpstreamResponseMs)
+	assertMilliseconds(t, 0, timing.ResponseWriteMs)
+	assertMilliseconds(t, 0, timing.FinalizeMs)
+}