perf(aws): GetLogEvents を時間範囲チャンキングで並列化 (#26)

internal/aws/client.go

@@ -14,8 +14,19 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 )
 
-// maxConcurrentStreams limits the number of concurrent GetLogEvents API calls.
-const maxConcurrentStreams = 5
+// maxConcurrent caps the number of in-flight GetLogEvents API calls across all
+// streams and time chunks. Tuned to stay well under the CloudWatch Logs
+// 25 TPS account-level limit while still saturating typical fetches.
+const maxConcurrent = 8
+
+// timeChunks is the number of equal time slices each stream's [first, last]
+// range is split into for parallel GetLogEvents fetches.
+const timeChunks = 8
+
+// minChunkDurationMs is the smallest time slice we will create. If a stream's
+// range is shorter than timeChunks * minChunkDurationMs, fewer chunks are
+// used (chunking a 10-second stream into 8 slices is pointless overhead).
+const minChunkDurationMs = 1000
 
 // defaultPageSize is the default number of items to fetch per API page.
 const defaultPageSize = 50
@@ -38,7 +49,8 @@ type LogGroup struct {
 
 // LogStream represents a CloudWatch Logs log stream.
 type LogStream struct {
-	Name               string
+	Name                string
+	FirstEventTimestamp time.Time
 	LastEventTimestamp  time.Time
 }
 
@@ -168,35 +180,79 @@ func (c *Client) ListLogStreamsPage(ctx context.Context, logGroupName string, ne
 
 	streams := make([]LogStream, 0, len(out.LogStreams))
 	for _, s := range out.LogStreams {
-		var lastEvent time.Time
-		if s.LastEventTimestamp != nil {
-			lastEvent = time.UnixMilli(*s.LastEventTimestamp)
-		}
-		streams = append(streams, LogStream{
-			Name:              awssdk.ToString(s.LogStreamName),
-			LastEventTimestamp: lastEvent,
-		})
+		streams = append(streams, toLogStream(s))
 	}
 	return streams, out.NextToken, nil
 }
 
+func toLogStream(s types.LogStream) LogStream {
+	var first, last time.Time
+	if s.FirstEventTimestamp != nil {
+		first = time.UnixMilli(*s.FirstEventTimestamp)
+	}
+	if s.LastEventTimestamp != nil {
+		last = time.UnixMilli(*s.LastEventTimestamp)
+	}
+	return LogStream{
+		Name:                awssdk.ToString(s.LogStreamName),
+		FirstEventTimestamp: first,
+		LastEventTimestamp:  last,
+	}
+}
+
+// DescribeLogStream returns metadata (including event time range) for a single
+// log stream. The lookup uses LogStreamNamePrefix and filters for an exact
+// name match, since CloudWatch Logs has no single-stream describe API.
+func (c *Client) DescribeLogStream(ctx context.Context, logGroupName, logStreamName string) (LogStream, error) {
+	out, err := c.api.DescribeLogStreams(ctx, &cloudwatchlogs.DescribeLogStreamsInput{
+		LogGroupName:        awssdk.String(logGroupName),
+		LogStreamNamePrefix: awssdk.String(logStreamName),
+		Limit:               awssdk.Int32(defaultPageSize),
+	})
+	if err != nil {
+		return LogStream{}, fmt.Errorf("describing log stream %q: %w", logStreamName, err)
+	}
+	for _, s := range out.LogStreams {
+		if awssdk.ToString(s.LogStreamName) == logStreamName {
+			return toLogStream(s), nil
+		}
+	}
+	return LogStream{}, fmt.Errorf("log stream %q not found in group %q", logStreamName, logGroupName)
+}
+
 // ListLogStreams returns log streams for a given log group (first page, descending).
 func (c *Client) ListLogStreams(ctx context.Context, logGroupName string) ([]LogStream, error) {
 	streams, _, err := c.ListLogStreamsPage(ctx, logGroupName, nil, true)
 	return streams, err
 }
 
 // GetLogEvents returns all log events for a given log group and stream,
-// handling pagination automatically.
+// handling NextForwardToken pagination automatically.
 func (c *Client) GetLogEvents(ctx context.Context, logGroupName, logStreamName string) ([]LogEvent, error) {
-	var allEvents []LogEvent
-	var prevToken string
+	return c.paginateGetLogEvents(ctx, &cloudwatchlogs.GetLogEventsInput{
+		LogGroupName:  awssdk.String(logGroupName),
+		LogStreamName: awssdk.String(logStreamName),
+		StartFromHead: awssdk.Bool(true),
+	})
+}
 
-	input := &cloudwatchlogs.GetLogEventsInput{
+// GetLogEventsByTimeRange paginates events for a single stream restricted to
+// [startTimeMs, endTimeMs). StartTime is inclusive, EndTime is exclusive —
+// callers can split a stream's range into adjacent windows without overlap
+// or gaps and fetch the windows in parallel.
+func (c *Client) GetLogEventsByTimeRange(ctx context.Context, logGroupName, logStreamName string, startTimeMs, endTimeMs int64) ([]LogEvent, error) {
+	return c.paginateGetLogEvents(ctx, &cloudwatchlogs.GetLogEventsInput{
 		LogGroupName:  awssdk.String(logGroupName),
 		LogStreamName: awssdk.String(logStreamName),
 		StartFromHead: awssdk.Bool(true),
-	}
+		StartTime:     awssdk.Int64(startTimeMs),
+		EndTime:       awssdk.Int64(endTimeMs),
+	})
+}
+
+func (c *Client) paginateGetLogEvents(ctx context.Context, input *cloudwatchlogs.GetLogEventsInput) ([]LogEvent, error) {
+	var allEvents []LogEvent
+	var prevToken string
 
 	for {
 		out, err := c.api.GetLogEvents(ctx, input)
@@ -216,8 +272,7 @@ func (c *Client) GetLogEvents(ctx context.Context, logGroupName, logStreamName s
 		}
 
 		// GetLogEvents signals end-of-stream by returning the same
-		// NextForwardToken as the previous call, or an empty page with
-		// no new token.
+		// NextForwardToken as the previous call, or an empty page.
 		nextToken := awssdk.ToString(out.NextForwardToken)
 		if nextToken == "" || nextToken == prevToken || len(out.Events) == 0 {
 			break
@@ -229,30 +284,33 @@ func (c *Client) GetLogEvents(ctx context.Context, logGroupName, logStreamName s
 	return allEvents, nil
 }
 
-// GetMultiStreamLogEvents fetches log events from multiple streams concurrently,
-// limiting parallelism to maxConcurrentStreams.
+// GetMultiStreamLogEvents fetches log events from one or more streams. For
+// each stream it first describes the stream to learn its event time range,
+// then splits that range into time chunks and issues parallel GetLogEvents
+// calls — turning a long serial pagination chain into mostly-parallel work.
+// All API calls share one semaphore (maxConcurrent) to stay under
+// CloudWatch Logs throttling limits.
 func (c *Client) GetMultiStreamLogEvents(ctx context.Context, logGroupName string, streamNames []string) ([]LogEvent, error) {
+	sem := make(chan struct{}, maxConcurrent)
 	results := make([][]LogEvent, len(streamNames))
 	var mu sync.Mutex
 	var firstErr error
-	var wg sync.WaitGroup
-
-	sem := make(chan struct{}, maxConcurrentStreams)
+	recordErr := func(err error) {
+		mu.Lock()
+		if firstErr == nil {
+			firstErr = err
+		}
+		mu.Unlock()
+	}
 
+	var wg sync.WaitGroup
 	for i, name := range streamNames {
 		wg.Add(1)
 		go func(idx int, streamName string) {
 			defer wg.Done()
-			sem <- struct{}{}
-			defer func() { <-sem }()
-
-			events, err := c.GetLogEvents(ctx, logGroupName, streamName)
+			events, err := c.fetchStreamEvents(ctx, logGroupName, streamName, sem)
 			if err != nil {
-				mu.Lock()
-				if firstErr == nil {
-					firstErr = err
-				}
-				mu.Unlock()
+				recordErr(err)
 				return
 			}
 			results[idx] = events
@@ -270,3 +328,122 @@ func (c *Client) GetMultiStreamLogEvents(ctx context.Context, logGroupName strin
 	}
 	return allEvents, nil
 }
+
+// fetchStreamEvents pulls every event for one stream, preferring the
+// time-chunked parallel path when the stream's range is known. When the
+// describe call fails or the range is missing/degenerate, it falls back to a
+// single sequential GetLogEvents that still benefits from the global
+// semaphore.
+func (c *Client) fetchStreamEvents(ctx context.Context, logGroupName, logStreamName string, sem chan struct{}) ([]LogEvent, error) {
+	stream, err := c.DescribeLogStream(ctx, logGroupName, logStreamName)
+	if err != nil || stream.FirstEventTimestamp.IsZero() || stream.LastEventTimestamp.IsZero() {
+		return c.fetchAllSequential(ctx, logGroupName, logStreamName, sem)
+	}
+	first := stream.FirstEventTimestamp.UnixMilli()
+	last := stream.LastEventTimestamp.UnixMilli()
+	if last < first {
+		return c.fetchAllSequential(ctx, logGroupName, logStreamName, sem)
+	}
+	return c.fetchChunked(ctx, logGroupName, logStreamName, first, last, sem)
+}
+
+func (c *Client) fetchAllSequential(ctx context.Context, logGroupName, logStreamName string, sem chan struct{}) ([]LogEvent, error) {
+	if err := acquire(ctx, sem); err != nil {
+		return nil, err
+	}
+	defer func() { <-sem }()
+	return c.GetLogEvents(ctx, logGroupName, logStreamName)
+}
+
+// fetchChunked splits [firstMs, lastMs] into equal time windows and fetches
+// them in parallel. Each chunk does its own NextForwardToken pagination
+// internally; the global semaphore bounds total API parallelism.
+func (c *Client) fetchChunked(ctx context.Context, logGroupName, logStreamName string, firstMs, lastMs int64, sem chan struct{}) ([]LogEvent, error) {
+	chunks := planTimeChunks(firstMs, lastMs)
+	if len(chunks) == 1 {
+		if err := acquire(ctx, sem); err != nil {
+			return nil, err
+		}
+		defer func() { <-sem }()
+		return c.GetLogEventsByTimeRange(ctx, logGroupName, logStreamName, chunks[0][0], chunks[0][1])
+	}
+
+	results := make([][]LogEvent, len(chunks))
+	var mu sync.Mutex
+	var firstErr error
+	recordErr := func(err error) {
+		mu.Lock()
+		if firstErr == nil {
+			firstErr = err
+		}
+		mu.Unlock()
+	}
+
+	var wg sync.WaitGroup
+	for i, window := range chunks {
+		wg.Add(1)
+		go func(idx int, start, end int64) {
+			defer wg.Done()
+			if err := acquire(ctx, sem); err != nil {
+				recordErr(err)
+				return
+			}
+			defer func() { <-sem }()
+			events, err := c.GetLogEventsByTimeRange(ctx, logGroupName, logStreamName, start, end)
+			if err != nil {
+				recordErr(err)
+				return
+			}
+			results[idx] = events
+		}(i, window[0], window[1])
+	}
+	wg.Wait()
+
+	if firstErr != nil {
+		return nil, firstErr
+	}
+
+	// Chunks are time-ordered and non-overlapping, and events within each
+	// chunk arrive oldest-first (StartFromHead=true), so plain concatenation
+	// yields a timestamp-sorted result.
+	var all []LogEvent
+	for _, r := range results {
+		all = append(all, r...)
+	}
+	return all, nil
+}
+
+// planTimeChunks returns adjacent [start, end) windows covering [firstMs, lastMs].
+// The final window's end is lastMs+1 because GetLogEvents.EndTime is exclusive.
+// Streams shorter than timeChunks * minChunkDurationMs use fewer chunks so we
+// don't pay describe + scatter overhead for already-tiny streams.
+func planTimeChunks(firstMs, lastMs int64) [][2]int64 {
+	totalMs := lastMs - firstMs + 1
+	chunks := timeChunks
+	if minSpan := int64(chunks) * minChunkDurationMs; totalMs < minSpan {
+		chunks = int(totalMs / minChunkDurationMs)
+		if chunks < 1 {
+			chunks = 1
+		}
+	}
+	windows := make([][2]int64, 0, chunks)
+	span := totalMs / int64(chunks)
+	for i := 0; i < chunks; i++ {
+		start := firstMs + int64(i)*span
+		end := start + span
+		if i == chunks-1 {
+			end = lastMs + 1
+		}
+		windows = append(windows, [2]int64{start, end})
+	}
+	return windows
+}
+
+func acquire(ctx context.Context, sem chan struct{}) error {
+	select {
+	case sem <- struct{}{}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}