From dd292db1186149c26f04f0961fa3242fc1f7b069 Mon Sep 17 00:00:00 2001 From: myuron Date: Fri, 22 May 2026 04:16:27 +0000 Subject: [PATCH] =?UTF-8?q?perf(aws):=20GetLogEvents=20=E3=82=92=E6=99=82?= =?UTF-8?q?=E9=96=93=E7=AF=84=E5=9B=B2=E3=83=81=E3=83=A3=E3=83=B3=E3=82=AD?= =?UTF-8?q?=E3=83=B3=E3=82=B0=E3=81=A7=E4=B8=A6=E5=88=97=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #26 対応。単一ストリームで NextForwardToken による逐次 ページネーションが大規模ログ取得のボトルネックだったのを、 DescribeLogStreams で得た first/last event 時刻範囲を timeChunks 等分し、各窓を StartTime/EndTime 指定で並列に GetLogEvents する 方式に置き換える。 - LogStream に FirstEventTimestamp 追加 + toLogStream ヘルパー - DescribeLogStream / GetLogEventsByTimeRange を追加 - GetMultiStreamLogEvents を時間チャンク並列に再実装 - グローバルセマフォ maxConcurrent=8 で AWS スロットルを回避 - range 不明 / 極小ストリームは逐次 GetLogEvents へフォールバック Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/aws/client.go | 241 ++++++++++++++++++++++---- internal/aws/client_test.go | 336 ++++++++++++++++++++++++++++++++++-- todo.md | 9 + 3 files changed, 537 insertions(+), 49 deletions(-) diff --git a/internal/aws/client.go b/internal/aws/client.go index e653cec..6418860 100644 --- a/internal/aws/client.go +++ b/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,18 +180,46 @@ 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) @@ -187,16 +227,32 @@ func (c *Client) ListLogStreams(ctx context.Context, logGroupName string) ([]Log } // 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() + } +} diff --git a/internal/aws/client_test.go b/internal/aws/client_test.go index 8ba2ea8..77b5b8b 100644 --- a/internal/aws/client_test.go +++ b/internal/aws/client_test.go @@ -3,6 +3,7 @@ package aws import ( "context" "fmt" + "sync" "sync/atomic" "testing" "time" @@ -150,12 +151,12 @@ func TestClient_ListLogStreams(t *testing.T) { return &cloudwatchlogs.DescribeLogStreamsOutput{ LogStreams: []types.LogStream{ { - LogStreamName: aws.String("stream-001"), - LastEventTimestamp: aws.Int64(nowMs), + LogStreamName: aws.String("stream-001"), + LastEventTimestamp: aws.Int64(nowMs), }, { - LogStreamName: aws.String("stream-002"), - LastEventTimestamp: aws.Int64(nowMs - 60000), + LogStreamName: aws.String("stream-002"), + LastEventTimestamp: aws.Int64(nowMs - 60000), }, }, }, nil @@ -198,7 +199,7 @@ func TestClient_ListLogStreamsPage(t *testing.T) { return &cloudwatchlogs.DescribeLogStreamsOutput{ LogStreams: []types.LogStream{ { - LogStreamName: aws.String("stream-001"), + LogStreamName: aws.String("stream-001"), LastEventTimestamp: aws.Int64(nowMs), }, }, @@ -239,11 +240,11 @@ func TestClient_ListLogStreamsPage_Ascending(t *testing.T) { return &cloudwatchlogs.DescribeLogStreamsOutput{ LogStreams: []types.LogStream{ { - LogStreamName: aws.String("stream-old"), + LogStreamName: aws.String("stream-old"), LastEventTimestamp: aws.Int64(nowMs - 60000), }, { - LogStreamName: aws.String("stream-new"), + LogStreamName: aws.String("stream-new"), LastEventTimestamp: aws.Int64(nowMs), }, }, @@ -462,21 +463,29 @@ func TestClient_GetLogEvents_ContextCancellation(t *testing.T) { } func TestFetchMultiLogEvents_ConcurrencyLimit(t *testing.T) { - // Verify that concurrent goroutines are limited. - var maxConcurrent atomic.Int32 + // Verify that concurrent GetLogEvents goroutines are bounded by the + // global semaphore. DescribeLogStreams returns no event-time range, so + // each stream falls back to a single sequential GetLogEvents call — + // keeping this test focused on the multi-stream concurrency cap. + var peakConcurrent atomic.Int32 var current atomic.Int32 mock := &mockLogsAPI{ + describeLogStreamsFn: func(ctx context.Context, params *cloudwatchlogs.DescribeLogStreamsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogStreamsOutput, error) { + return &cloudwatchlogs.DescribeLogStreamsOutput{ + LogStreams: []types.LogStream{ + {LogStreamName: params.LogStreamNamePrefix}, + }, + }, nil + }, getLogEventsFn: func(ctx context.Context, params *cloudwatchlogs.GetLogEventsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.GetLogEventsOutput, error) { cur := current.Add(1) - // Track the peak concurrency for { - old := maxConcurrent.Load() - if cur <= old || maxConcurrent.CompareAndSwap(old, cur) { + old := peakConcurrent.Load() + if cur <= old || peakConcurrent.CompareAndSwap(old, cur) { break } } - // Simulate some work time.Sleep(10 * time.Millisecond) current.Add(-1) return &cloudwatchlogs.GetLogEventsOutput{ @@ -489,7 +498,6 @@ func TestFetchMultiLogEvents_ConcurrencyLimit(t *testing.T) { } client := &Client{api: mock} - // Request 20 streams to ensure the limit is tested streams := make([]string, 20) for i := range streams { streams[i] = fmt.Sprintf("stream-%d", i) @@ -501,11 +509,305 @@ func TestFetchMultiLogEvents_ConcurrencyLimit(t *testing.T) { t.Fatalf("unexpected error: %v", err) } - peak := maxConcurrent.Load() - if peak > maxConcurrentStreams { - t.Errorf("peak concurrency %d exceeded limit %d", peak, maxConcurrentStreams) + peak := peakConcurrent.Load() + if peak > maxConcurrent { + t.Errorf("peak concurrency %d exceeded limit %d", peak, maxConcurrent) } if peak == 0 { t.Error("expected at least 1 concurrent call") } } + +func TestClient_DescribeLogStream(t *testing.T) { + nowMs := time.Now().UnixMilli() + mock := &mockLogsAPI{ + describeLogStreamsFn: func(ctx context.Context, params *cloudwatchlogs.DescribeLogStreamsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogStreamsOutput, error) { + if aws.ToString(params.LogGroupName) != "/aws/lambda/func-a" { + t.Errorf("unexpected log group: %s", aws.ToString(params.LogGroupName)) + } + if aws.ToString(params.LogStreamNamePrefix) != "stream-001" { + t.Errorf("expected LogStreamNamePrefix=stream-001, got %s", aws.ToString(params.LogStreamNamePrefix)) + } + return &cloudwatchlogs.DescribeLogStreamsOutput{ + LogStreams: []types.LogStream{ + { + LogStreamName: aws.String("stream-001-other"), + FirstEventTimestamp: aws.Int64(nowMs - 200000), + LastEventTimestamp: aws.Int64(nowMs - 100000), + }, + { + LogStreamName: aws.String("stream-001"), + FirstEventTimestamp: aws.Int64(nowMs - 60000), + LastEventTimestamp: aws.Int64(nowMs), + }, + }, + }, nil + }, + } + + client := &Client{api: mock} + stream, err := client.DescribeLogStream(context.Background(), "/aws/lambda/func-a", "stream-001") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if stream.Name != "stream-001" { + t.Errorf("expected exact-match Name=stream-001, got %s", stream.Name) + } + if got := stream.FirstEventTimestamp.UnixMilli(); got != nowMs-60000 { + t.Errorf("expected FirstEventTimestamp=%d, got %d", nowMs-60000, got) + } + if got := stream.LastEventTimestamp.UnixMilli(); got != nowMs { + t.Errorf("expected LastEventTimestamp=%d, got %d", nowMs, got) + } +} + +func TestClient_DescribeLogStream_NotFound(t *testing.T) { + mock := &mockLogsAPI{ + describeLogStreamsFn: func(ctx context.Context, params *cloudwatchlogs.DescribeLogStreamsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogStreamsOutput, error) { + return &cloudwatchlogs.DescribeLogStreamsOutput{LogStreams: nil}, nil + }, + } + client := &Client{api: mock} + if _, err := client.DescribeLogStream(context.Background(), "g", "missing"); err == nil { + t.Fatal("expected error when stream is not found") + } +} + +func TestClient_GetLogEventsByTimeRange(t *testing.T) { + calls := 0 + mock := &mockLogsAPI{ + getLogEventsFn: func(ctx context.Context, params *cloudwatchlogs.GetLogEventsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.GetLogEventsOutput, error) { + calls++ + if got := aws.ToInt64(params.StartTime); got != 1000 { + t.Errorf("expected StartTime=1000, got %d", got) + } + if got := aws.ToInt64(params.EndTime); got != 5000 { + t.Errorf("expected EndTime=5000, got %d", got) + } + if !aws.ToBool(params.StartFromHead) { + t.Error("expected StartFromHead=true") + } + // Empty events on the second call terminates pagination. + if calls == 1 { + return &cloudwatchlogs.GetLogEventsOutput{ + Events: []types.OutputLogEvent{ + {Timestamp: aws.Int64(1000), Message: aws.String("a")}, + {Timestamp: aws.Int64(4000), Message: aws.String("b")}, + }, + NextForwardToken: aws.String("next"), + }, nil + } + return &cloudwatchlogs.GetLogEventsOutput{ + Events: []types.OutputLogEvent{}, + NextForwardToken: aws.String("next"), + }, nil + }, + } + client := &Client{api: mock} + events, err := client.GetLogEventsByTimeRange(context.Background(), "g", "s", 1000, 5000) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(events) != 2 { + t.Fatalf("expected 2 events, got %d", len(events)) + } +} + +func TestClient_GetLogEventsByTimeRange_Pagination(t *testing.T) { + // Within a single time window, pagination still uses NextForwardToken. + calls := 0 + mock := &mockLogsAPI{ + getLogEventsFn: func(ctx context.Context, params *cloudwatchlogs.GetLogEventsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.GetLogEventsOutput, error) { + calls++ + if aws.ToInt64(params.StartTime) != 100 || aws.ToInt64(params.EndTime) != 200 { + t.Errorf("unexpected time window: start=%d end=%d", aws.ToInt64(params.StartTime), aws.ToInt64(params.EndTime)) + } + switch calls { + case 1: + return &cloudwatchlogs.GetLogEventsOutput{ + Events: []types.OutputLogEvent{{Timestamp: aws.Int64(100), Message: aws.String("e1")}}, + NextForwardToken: aws.String("t2"), + }, nil + case 2: + if aws.ToString(params.NextToken) != "t2" { + t.Errorf("expected NextToken=t2, got %s", aws.ToString(params.NextToken)) + } + return &cloudwatchlogs.GetLogEventsOutput{ + Events: []types.OutputLogEvent{{Timestamp: aws.Int64(150), Message: aws.String("e2")}}, + NextForwardToken: aws.String("t2"), + }, nil + } + t.Fatal("too many calls") + return nil, nil + }, + } + client := &Client{api: mock} + events, err := client.GetLogEventsByTimeRange(context.Background(), "g", "s", 100, 200) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(events) != 2 || calls != 2 { + t.Errorf("expected 2 events from 2 calls, got %d events / %d calls", len(events), calls) + } +} + +func TestClient_GetMultiStreamLogEvents_TimeChunksInParallel(t *testing.T) { + // When DescribeLogStreams returns a wide time range, GetMultiStreamLogEvents + // should split the range into multiple chunks and fetch them in parallel, + // reducing wall-clock time vs. sequential pagination. + var getCalls atomic.Int32 + var peakConcurrent atomic.Int32 + var current atomic.Int32 + + var startsMu sync.Mutex + seenWindows := make(map[[2]int64]bool) + + const firstMs = int64(1_000_000) + const lastMs = int64(9_000_000) + + mock := &mockLogsAPI{ + describeLogStreamsFn: func(ctx context.Context, params *cloudwatchlogs.DescribeLogStreamsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogStreamsOutput, error) { + return &cloudwatchlogs.DescribeLogStreamsOutput{ + LogStreams: []types.LogStream{ + { + LogStreamName: params.LogStreamNamePrefix, + FirstEventTimestamp: aws.Int64(firstMs), + LastEventTimestamp: aws.Int64(lastMs), + }, + }, + }, nil + }, + getLogEventsFn: func(ctx context.Context, params *cloudwatchlogs.GetLogEventsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.GetLogEventsOutput, error) { + getCalls.Add(1) + cur := current.Add(1) + for { + old := peakConcurrent.Load() + if cur <= old || peakConcurrent.CompareAndSwap(old, cur) { + break + } + } + time.Sleep(10 * time.Millisecond) + current.Add(-1) + + startsMu.Lock() + seenWindows[[2]int64{aws.ToInt64(params.StartTime), aws.ToInt64(params.EndTime)}] = true + startsMu.Unlock() + + return &cloudwatchlogs.GetLogEventsOutput{ + Events: []types.OutputLogEvent{ + {Timestamp: params.StartTime, Message: aws.String("e")}, + }, + NextForwardToken: aws.String("end"), + }, nil + }, + } + + client := &Client{api: mock} + events, err := client.GetMultiStreamLogEvents(context.Background(), "g", []string{"stream-001"}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if getCalls.Load() < 2 { + t.Errorf("expected multiple chunked GetLogEvents calls, got %d", getCalls.Load()) + } + if peakConcurrent.Load() < 2 { + t.Errorf("expected concurrent execution across chunks, got peak=%d", peakConcurrent.Load()) + } + if len(events) < 2 { + t.Errorf("expected at least 2 events from chunked fetch, got %d", len(events)) + } + + // Time windows must be non-overlapping and cover [firstMs, lastMs+1). + startsMu.Lock() + defer startsMu.Unlock() + type window struct{ start, end int64 } + windows := make([]window, 0, len(seenWindows)) + for k := range seenWindows { + windows = append(windows, window{k[0], k[1]}) + } + // Sort by start ascending. + for i := 0; i < len(windows); i++ { + for j := i + 1; j < len(windows); j++ { + if windows[j].start < windows[i].start { + windows[i], windows[j] = windows[j], windows[i] + } + } + } + if windows[0].start != firstMs { + t.Errorf("first window start=%d, expected %d", windows[0].start, firstMs) + } + if windows[len(windows)-1].end != lastMs+1 { + t.Errorf("last window end=%d, expected %d", windows[len(windows)-1].end, lastMs+1) + } + for i := 1; i < len(windows); i++ { + if windows[i].start != windows[i-1].end { + t.Errorf("window gap or overlap: prev=[%d,%d) next=[%d,%d)", + windows[i-1].start, windows[i-1].end, windows[i].start, windows[i].end) + } + } +} + +func TestClient_GetMultiStreamLogEvents_FallbackWhenNoTimeRange(t *testing.T) { + // DescribeLogStreams returns a stream with no FirstEventTimestamp; the + // client must fall back to a plain sequential GetLogEvents (no time + // chunking) for that stream and still return its events. + var startTimesSeen atomic.Int32 + mock := &mockLogsAPI{ + describeLogStreamsFn: func(ctx context.Context, params *cloudwatchlogs.DescribeLogStreamsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogStreamsOutput, error) { + return &cloudwatchlogs.DescribeLogStreamsOutput{ + LogStreams: []types.LogStream{ + {LogStreamName: params.LogStreamNamePrefix}, + }, + }, nil + }, + getLogEventsFn: func(ctx context.Context, params *cloudwatchlogs.GetLogEventsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.GetLogEventsOutput, error) { + if params.StartTime != nil || params.EndTime != nil { + startTimesSeen.Add(1) + } + return &cloudwatchlogs.GetLogEventsOutput{ + Events: []types.OutputLogEvent{{Timestamp: aws.Int64(1), Message: aws.String("e")}}, + NextForwardToken: aws.String("same"), + }, nil + }, + } + client := &Client{api: mock} + events, err := client.GetMultiStreamLogEvents(context.Background(), "g", []string{"s"}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(events) == 0 { + t.Error("expected at least one event from fallback path") + } + if startTimesSeen.Load() != 0 { + t.Errorf("expected sequential fallback (no StartTime/EndTime), got %d windowed calls", startTimesSeen.Load()) + } +} + +func TestClient_ListLogStreamsPage_PopulatesFirstEventTimestamp(t *testing.T) { + nowMs := time.Now().UnixMilli() + mock := &mockLogsAPI{ + describeLogStreamsFn: func(ctx context.Context, params *cloudwatchlogs.DescribeLogStreamsInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogStreamsOutput, error) { + return &cloudwatchlogs.DescribeLogStreamsOutput{ + LogStreams: []types.LogStream{ + { + LogStreamName: aws.String("s"), + FirstEventTimestamp: aws.Int64(nowMs - 60000), + LastEventTimestamp: aws.Int64(nowMs), + }, + }, + }, nil + }, + } + client := &Client{api: mock} + streams, _, err := client.ListLogStreamsPage(context.Background(), "g", nil, true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(streams) != 1 { + t.Fatalf("expected 1 stream, got %d", len(streams)) + } + if got := streams[0].FirstEventTimestamp.UnixMilli(); got != nowMs-60000 { + t.Errorf("expected FirstEventTimestamp=%d, got %d", nowMs-60000, got) + } +} diff --git a/todo.md b/todo.md index 09c2983..e7a2129 100644 --- a/todo.md +++ b/todo.md @@ -132,3 +132,12 @@ - [x] 2026-05-04 起動時エラー表示のテスト追加3件（Err()アクセサ / View()のwidth折り返し / width=0フォールバック） - [x] 2026-05-04 Model.Err() アクセサ追加 + preview.go の renderErrorView() で lipgloss 折り返し - [x] 2026-05-04 main.go で p.Run() 終了後に Model.Err() を stderr へ出力 + +## ログ取得の並列化 (Issue #26) +- [x] 2026-05-22 LogStream に FirstEventTimestamp フィールド追加 + toLogStream ヘルパー切り出し +- [x] 2026-05-22 DescribeLogStream 単一ストリーム metadata 取得を追加（exact-match filter） +- [x] 2026-05-22 GetLogEventsByTimeRange を追加（StartTime/EndTime 指定 + ページネーション） +- [x] 2026-05-22 GetMultiStreamLogEvents を時間チャンク並列に再実装（planTimeChunks で [first,last+1) を timeChunks 等分） +- [x] 2026-05-22 グローバルセマフォ maxConcurrent=8 で全 GetLogEvents 呼び出しの並列度を抑制 +- [x] 2026-05-22 describe 失敗 / range 不正 / 極小ストリームは逐次 GetLogEvents にフォールバック +- [x] 2026-05-22 テスト追加 7件（Describe / TimeRange / 並列チャンク / フォールバック / FirstEventTimestamp 伝播）