From 4d35546dc05a1e9d3fd942ac812596066ab196bb Mon Sep 17 00:00:00 2001 From: Nitin Kumar Date: Fri, 23 Oct 2026 09:15:04 +0530 Subject: [PATCH 1/2] fix(aws-cloudwatchlogs): StartLiveTail event stream --- docs/coverage/aws/cloudwatchlogs.md | 12 + docs/coverage/coverage.json | 22 ++ .../aws/cloudwatchlogs/cloudwatchlogs.go | 5 + providers/aws/cloudwatchlogs/ingest.go | 3 +- providers/aws/cloudwatchlogs/live_tail.go | 374 ++++++++++++++++++ .../aws/cloudwatchlogs/live_tail_test.go | 280 +++++++++++++ server/aws/cloudwatchlogs/export_tasks.go | 6 +- server/aws/cloudwatchlogs/handler.go | 2 + server/aws/cloudwatchlogs/live_tail.go | 142 +++++++ .../aws/cloudwatchlogs/live_tail_sdk_test.go | 278 +++++++++++++ server/aws/kinesis/streaming_ops.go | 39 +- server/serverkit/middleware.go | 4 + server/serverkit/serverkit.go | 53 ++- server/wire/awseventstream/awseventstream.go | 75 ++++ server/wire/shutdown.go | 32 ++ server/wire/shutdown_test.go | 31 ++ services/logging/driver/aws.go | 41 ++ 17 files changed, 1341 insertions(+), 58 deletions(-) create mode 100644 providers/aws/cloudwatchlogs/live_tail.go create mode 100644 providers/aws/cloudwatchlogs/live_tail_test.go create mode 100644 server/aws/cloudwatchlogs/live_tail.go create mode 100644 server/aws/cloudwatchlogs/live_tail_sdk_test.go create mode 100644 server/wire/awseventstream/awseventstream.go create mode 100644 server/wire/shutdown.go create mode 100644 server/wire/shutdown_test.go diff --git a/docs/coverage/aws/cloudwatchlogs.md b/docs/coverage/aws/cloudwatchlogs.md index 0947e2d66..7c81eecc6 100644 --- a/docs/coverage/aws/cloudwatchlogs.md +++ b/docs/coverage/aws/cloudwatchlogs.md @@ -58,10 +58,22 @@ AWSLogging is an optional interface implemented by the AWS CloudWatch Logs | `PutDataProtectionPolicy` | Data protection policies take a log group name or ARN. Masking applies | | `PutLogEventsWithResult` | | | `PutRetentionPolicy` | | +| `StartLiveTail` | StartLiveTail opens a Live Tail session over events ingested from now | | `StartQuery` | Logs Insights. A query is Running until its first GetQueryResults, | | `StopQuery` | | | `TestMetricFilter` | TestMetricFilter runs a filter pattern over sample log messages. | +### LiveTailSession + +LiveTailSession is a running Live Tail session. Updates delivers one update + +| Operation | Description | +| --- | --- | +| `Close` | | +| `SessionID` | | +| `TimedOut` | | +| `Updates` | | + ## Not in scope - AWS: `CreateExportTask` checks the destination bucket's policy the way CloudWatch Logs does before it accepts a task. The policy must allow the `logs.amazonaws.com` or `logs..amazonaws.com` service principal (or `*`) to call `s3:GetBucketAcl` on the bucket and `s3:PutObject` under the prefix, and a matching `Deny` wins. Policy conditions (`aws:SourceAccount`, `aws:SourceArn`, `s3:x-amz-acl`), `NotPrincipal`/`NotAction`/`NotResource`, the Region named in the principal, SSE-KMS key policies and cross-account buckets are not evaluated. diff --git a/docs/coverage/coverage.json b/docs/coverage/coverage.json index 804792a90..6d7bbeda2 100644 --- a/docs/coverage/coverage.json +++ b/docs/coverage/coverage.json @@ -11639,6 +11639,10 @@ { "name": "PutRetentionPolicy" }, + { + "name": "StartLiveTail", + "doc": "StartLiveTail opens a Live Tail session over events ingested from now" + }, { "name": "StartQuery", "doc": "Logs Insights. A query is Running until its first GetQueryResults," @@ -11702,6 +11706,24 @@ "name": "UpdateSink" } ] + }, + { + "name": "LiveTailSession", + "doc": "LiveTailSession is a running Live Tail session. Updates delivers one update", + "operations": [ + { + "name": "Close" + }, + { + "name": "SessionID" + }, + { + "name": "TimedOut" + }, + { + "name": "Updates" + } + ] } ], "providers": { diff --git a/providers/aws/cloudwatchlogs/cloudwatchlogs.go b/providers/aws/cloudwatchlogs/cloudwatchlogs.go index 3697b5faa..8d4d4224c 100644 --- a/providers/aws/cloudwatchlogs/cloudwatchlogs.go +++ b/providers/aws/cloudwatchlogs/cloudwatchlogs.go @@ -83,6 +83,10 @@ type Mock struct { // step, and serializes claiming tasks for writing. It is never held // across an S3 call. exportMu sync.Mutex + // liveTails holds the running Live Tail sessions by ID, under liveTailMu. + // Sessions are ephemeral and never persisted. + liveTails map[string]*liveTail + liveTailMu sync.Mutex } // SetMonitoring sets the monitoring backend for auto-metric generation. @@ -117,6 +121,7 @@ func New(opts *config.Options) *Mock { periods: map[string]filterPeriod{}, accountPolicies: map[string]*accountPolicy{}, + liveTails: map[string]*liveTail{}, } } diff --git a/providers/aws/cloudwatchlogs/ingest.go b/providers/aws/cloudwatchlogs/ingest.go index 1bf80b82d..604dd7173 100644 --- a/providers/aws/cloudwatchlogs/ingest.go +++ b/providers/aws/cloudwatchlogs/ingest.go @@ -148,7 +148,7 @@ func firstAfter(events []driver.LogEvent, cutoff time.Time) int { } // storeEvents appends accepted events to the stream, updates the group's -// counters, and runs metric filters and subscriptions. It returns the +// counters, and runs metric filters, subscriptions and Live Tail. It returns the // stream's next sequence token. Every upload, even one whose events were all // rejected, advances the token. func (m *Mock) storeEvents( @@ -211,6 +211,7 @@ func (m *Mock) storeEvents( m.evaluateMetricFilters(g, events) m.deliverToSubscriptions(ctx, g, streamName, copied) + m.publishLiveTail(groupName, streamName, copied) return token } diff --git a/providers/aws/cloudwatchlogs/live_tail.go b/providers/aws/cloudwatchlogs/live_tail.go new file mode 100644 index 000000000..adb6504cc --- /dev/null +++ b/providers/aws/cloudwatchlogs/live_tail.go @@ -0,0 +1,374 @@ +package cloudwatchlogs + +import ( + "context" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/stackshy/cloudemu/v2/errors" + "github.com/stackshy/cloudemu/v2/internal/idgen" + "github.com/stackshy/cloudemu/v2/providers/aws/cloudwatchlogs/filterpattern" + "github.com/stackshy/cloudemu/v2/services/logging/driver" +) + +// Live Tail limits, per the StartLiveTail API reference and the CloudWatch +// Logs quotas. +const ( + // liveTailSessionTimeout: a session ends with SessionTimeoutException + // after it has been open for three hours. + liveTailSessionTimeout = 3 * time.Hour + // liveTailUpdateInterval: a sessionUpdate is sent every second, empty when + // nothing new matched. + liveTailUpdateInterval = time.Second + // liveTailMaxEventsPerUpdate: one update carries at most 500 events; more + // are sampled down to 500. + liveTailMaxEventsPerUpdate = 500 + // liveTailBufferedUpdates and liveTailBufferedEvents: for a slow client, + // up to 10 updates or 5000 events are buffered, then the oldest drop. + liveTailBufferedUpdates = 10 + liveTailBufferedEvents = 5000 + maxLiveTailLogGroups = 10 + maxLiveTailStreams = 100 + maxLiveTailStreamName = 512 + maxLiveTailPatternLen = 1024 + // maxLiveTailSessions is the account quota of concurrent sessions. + maxLiveTailSessions = 15 +) + +// liveTail is one running session. The fan-out in PutLogEvents only appends +// to pending under mu; the session goroutine owns out. +type liveTail struct { + id string + groups map[string]string // log group name -> identifier as requested + streams map[string]bool // nil means every stream + prefixes []string + pattern *filterpattern.Pattern + started time.Time + + out chan driver.LiveTailUpdate + done chan struct{} + closeOnce sync.Once + timedOut atomic.Bool + + mu sync.Mutex + pending []driver.LiveTailEvent + matched int +} + +func (t *liveTail) SessionID() string { return t.id } +func (t *liveTail) Updates() <-chan driver.LiveTailUpdate { return t.out } +func (t *liveTail) TimedOut() bool { return t.timedOut.Load() } + +// Close ends the session. It is safe to call more than once. +func (t *liveTail) Close() { + t.closeOnce.Do(func() { close(t.done) }) +} + +// StartLiveTail opens a Live Tail session. Events ingested after it starts, +// in the selected groups and streams and matching the filter pattern, are +// delivered in one update per second. Masked events stay masked. +func (m *Mock) StartLiveTail(ctx context.Context, in *driver.StartLiveTailInput) (driver.LiveTailSession, error) { + t, err := m.newLiveTail(in) + if err != nil { + return nil, err + } + + m.liveTailMu.Lock() + + if len(m.liveTails) >= maxLiveTailSessions { + m.liveTailMu.Unlock() + + return nil, errors.Newf(errors.ResourceExhausted, + "The maximum number of concurrent Live Tail sessions (%d) has been reached", maxLiveTailSessions) + } + + m.liveTails[t.id] = t + m.liveTailMu.Unlock() + + go m.runLiveTail(ctx, t) + + return t, nil +} + +// newLiveTail validates a request and builds its session. +func (m *Mock) newLiveTail(in *driver.StartLiveTailInput) (*liveTail, error) { + if err := validateLiveTail(in); err != nil { + return nil, err + } + + pattern, err := filterpattern.Parse(in.LogEventFilterPattern) + if err != nil { + return nil, err + } + + groups := make(map[string]string, len(in.LogGroupIdentifiers)) + + for _, id := range in.LogGroupIdentifiers { + name := logGroupFromIdentifier(id) + if !m.groups.Has(name) { + return nil, errors.Newf(errors.NotFound, "The specified log group does not exist: %s", id) + } + + groups[name] = id + } + + t := &liveTail{ + id: idgen.UUID(), + groups: groups, + prefixes: in.LogStreamNamePrefixes, + pattern: pattern, + started: m.opts.Clock.Now(), + out: make(chan driver.LiveTailUpdate, liveTailBufferedUpdates), + done: make(chan struct{}), + } + + if len(in.LogStreamNames) > 0 { + t.streams = make(map[string]bool, len(in.LogStreamNames)) + for _, s := range in.LogStreamNames { + t.streams[s] = true + } + } + + return t, nil +} + +func validateLiveTail(in *driver.StartLiveTailInput) error { + if err := validateLiveTailShape(in); err != nil { + return err + } + + for _, id := range in.LogGroupIdentifiers { + if err := validateLiveTailGroup(id); err != nil { + return err + } + } + + if err := validateLiveTailStreams("logStreamNames", in.LogStreamNames); err != nil { + return err + } + + return validateLiveTailStreams("logStreamNamePrefixes", in.LogStreamNamePrefixes) +} + +// validateLiveTailShape checks the item counts, the stream selector rules and +// the pattern length. +func validateLiveTailShape(in *driver.StartLiveTailInput) error { + switch { + case len(in.LogGroupIdentifiers) == 0 || len(in.LogGroupIdentifiers) > maxLiveTailLogGroups: + return errors.Newf(errors.InvalidArgument, + "logGroupIdentifiers must contain between 1 and %d log group ARNs", maxLiveTailLogGroups) + case len(in.LogStreamNames) > 0 && len(in.LogStreamNamePrefixes) > 0: + return errors.New(errors.InvalidArgument, + "logStreamNames and logStreamNamePrefixes cannot both be specified") + case (len(in.LogStreamNames) > 0 || len(in.LogStreamNamePrefixes) > 0) && len(in.LogGroupIdentifiers) != 1: + return errors.New(errors.InvalidArgument, + "logStreamNames and logStreamNamePrefixes can only be specified with exactly one log group") + case len(in.LogEventFilterPattern) > maxLiveTailPatternLen: + return errors.Newf(errors.InvalidArgument, + "logEventFilterPattern must be at most %d characters", maxLiveTailPatternLen) + } + + return nil +} + +// validateLiveTailGroup checks one log group identifier: an ARN that does not +// end with an asterisk. +func validateLiveTailGroup(id string) error { + if len(id) > maxLogGroupIdentifierLength || !logGroupIdentifierRE.MatchString(id) { + return errors.New(errors.InvalidArgument, + "1 validation error detected: Value at 'logGroupIdentifiers' failed to satisfy constraint: "+ + `Member must satisfy regular expression pattern: [\w#+=/:,.@-]*`) + } + + if !strings.HasPrefix(id, "arn:") || !strings.Contains(id, logGroupARNMarker) { + return errors.Newf(errors.InvalidArgument, "Specify each log group by its ARN: %s", id) + } + + if strings.HasSuffix(id, "*") { + return errors.Newf(errors.InvalidArgument, "The log group ARN cannot end with an asterisk: %s", id) + } + + return nil +} + +func validateLiveTailStreams(field string, names []string) error { + if len(names) > maxLiveTailStreams { + return errors.Newf(errors.InvalidArgument, "%s can contain at most %d items", field, maxLiveTailStreams) + } + + for _, n := range names { + if n == "" || len(n) > maxLiveTailStreamName || strings.ContainsAny(n, ":*") { + return errors.Newf(errors.InvalidArgument, + "1 validation error detected: Value at '%s' failed to satisfy constraint: "+ + "Member must have length between 1 and %d and satisfy regular expression pattern: [^:*]*", + field, maxLiveTailStreamName) + } + } + + return nil +} + +// runLiveTail sends one update per interval until the session's context +// ends, it is closed, or it reaches the session timeout. Session age is read +// from opts.Clock; the cadence is wall-clock time, since a fake clock's After +// fires at once. +func (m *Mock) runLiveTail(ctx context.Context, t *liveTail) { + ticker := time.NewTicker(liveTailUpdateInterval) + + defer func() { + ticker.Stop() + m.liveTailMu.Lock() + delete(m.liveTails, t.id) + m.liveTailMu.Unlock() + close(t.out) + }() + + for { + select { + case <-ctx.Done(): + return + case <-t.done: + return + case <-ticker.C: + if m.opts.Clock.Since(t.started) >= liveTailSessionTimeout { + t.timedOut.Store(true) + + return + } + + t.send(t.drain()) + } + } +} + +// send queues an update. When the client has fallen 10 updates behind, the +// oldest queued update is dropped. Only the session goroutine sends, so the +// second attempt always has room. +func (t *liveTail) send(u driver.LiveTailUpdate) { + select { + case t.out <- u: + return + default: + } + + select { + case <-t.out: + default: + } + + select { + case t.out <- u: + default: + } +} + +// drain takes the events matched since the last update, sampled down to the +// per-update limit. +func (t *liveTail) drain() driver.LiveTailUpdate { + t.mu.Lock() + events, matched := t.pending, t.matched + t.pending, t.matched = nil, 0 + t.mu.Unlock() + + u := driver.LiveTailUpdate{Results: events, Sampled: matched > liveTailMaxEventsPerUpdate} + if len(events) <= liveTailMaxEventsPerUpdate { + if u.Results == nil { + u.Results = []driver.LiveTailEvent{} + } + + return u + } + + sampled := make([]driver.LiveTailEvent, liveTailMaxEventsPerUpdate) + for i := range sampled { + sampled[i] = events[i*len(events)/liveTailMaxEventsPerUpdate] + } + + u.Results = sampled + + return u +} + +// wants reports whether the session tails the given stream of a group, and +// the group identifier it reports events under. +func (t *liveTail) wants(group, stream string) (string, bool) { + id, ok := t.groups[group] + if !ok { + return "", false + } + + if t.streams != nil { + return id, t.streams[stream] + } + + if len(t.prefixes) == 0 { + return id, true + } + + for _, p := range t.prefixes { + if strings.HasPrefix(stream, p) { + return id, true + } + } + + return "", false +} + +// offer adds the matching events of one stored batch to the session's +// pending buffer, keeping at most the newest liveTailBufferedEvents. +func (t *liveTail) offer(group, stream string, events []driver.LogEvent) { + id, ok := t.wants(group, stream) + if !ok { + return + } + + var hits []driver.LiveTailEvent + + for i := range events { + if !t.pattern.Match(events[i].Message) { + continue + } + + hits = append(hits, driver.LiveTailEvent{ + LogGroupIdentifier: id, + LogStreamName: stream, + Message: events[i].Message, + Timestamp: events[i].Timestamp, + IngestionTime: events[i].IngestionTime, + }) + } + + if len(hits) == 0 { + return + } + + t.mu.Lock() + defer t.mu.Unlock() + + t.matched += len(hits) + t.pending = append(t.pending, hits...) + + if over := len(t.pending) - liveTailBufferedEvents; over > 0 { + t.pending = append(t.pending[:0:0], t.pending[over:]...) + } +} + +// publishLiveTail fans a stored batch out to the running sessions. The +// registry lock is held only to copy the session list; offer takes each +// session's own lock briefly and never blocks on a client. +func (m *Mock) publishLiveTail(group, stream string, events []driver.LogEvent) { + m.liveTailMu.Lock() + + sessions := make([]*liveTail, 0, len(m.liveTails)) + for _, t := range m.liveTails { + sessions = append(sessions, t) + } + + m.liveTailMu.Unlock() + + for _, t := range sessions { + t.offer(group, stream, events) + } +} diff --git a/providers/aws/cloudwatchlogs/live_tail_test.go b/providers/aws/cloudwatchlogs/live_tail_test.go new file mode 100644 index 000000000..b7097c48d --- /dev/null +++ b/providers/aws/cloudwatchlogs/live_tail_test.go @@ -0,0 +1,280 @@ +package cloudwatchlogs + +import ( + "context" + "fmt" + "runtime" + "strings" + "testing" + "time" + + "github.com/stackshy/cloudemu/v2/config" + "github.com/stackshy/cloudemu/v2/errors" + "github.com/stackshy/cloudemu/v2/services/logging/driver" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const liveTailTestARN = "arn:aws:logs:us-east-1:123456789012:log-group:" + +// liveTailWait bounds how long a test waits for the next one-second update. +const liveTailWait = 5 * time.Second + +func newLiveTailMock(t *testing.T) (*Mock, *config.FakeClock) { + t.Helper() + + fc := config.NewFakeClock(testClockStart()) + m := New(config.NewOptions(config.WithClock(fc), config.WithRegion("us-east-1"))) + + ctx := context.Background() + + for _, g := range []string{"lt-a", "lt-b"} { + _, err := m.CreateLogGroup(ctx, driver.LogGroupConfig{Name: g}) + require.NoError(t, err) + + for _, s := range []string{"app-1", "other"} { + _, err = m.CreateLogStream(ctx, g, s) + require.NoError(t, err) + } + } + + return m, fc +} + +// nextResults returns the first update that carries events. +func nextResults(t *testing.T, s driver.LiveTailSession) driver.LiveTailUpdate { + t.Helper() + + deadline := time.After(liveTailWait) + + for { + select { + case u, ok := <-s.Updates(): + require.True(t, ok, "session ended before any results") + + if len(u.Results) > 0 { + return u + } + case <-deadline: + t.Fatal("no Live Tail update with results") + } + } +} + +// waitEnded waits for the session's update channel to close. +func waitEnded(t *testing.T, s driver.LiveTailSession) { + t.Helper() + + deadline := time.After(liveTailWait) + + for { + select { + case _, ok := <-s.Updates(): + if !ok { + return + } + case <-deadline: + t.Fatal("Live Tail session did not end") + } + } +} + +func TestStartLiveTailValidation(t *testing.T) { + m, _ := newLiveTailMock(t) + arn := liveTailTestARN + "lt-a" + + eleven := make([]string, maxLiveTailLogGroups+1) + for i := range eleven { + eleven[i] = arn + } + + tests := []struct { + name string + in driver.StartLiveTailInput + code errors.Code + }{ + {"no groups", driver.StartLiveTailInput{}, errors.InvalidArgument}, + {"eleven groups", driver.StartLiveTailInput{LogGroupIdentifiers: eleven}, errors.InvalidArgument}, + {"name not ARN", driver.StartLiveTailInput{LogGroupIdentifiers: []string{"lt-a"}}, errors.InvalidArgument}, + {"ARN ends with asterisk", driver.StartLiveTailInput{LogGroupIdentifiers: []string{arn + ":*"}}, errors.InvalidArgument}, + {"names and prefixes", driver.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn}, LogStreamNames: []string{"a"}, LogStreamNamePrefixes: []string{"a"}, + }, errors.InvalidArgument}, + {"streams with two groups", driver.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn, liveTailTestARN + "lt-b"}, LogStreamNames: []string{"a"}, + }, errors.InvalidArgument}, + {"bad stream name", driver.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn}, LogStreamNamePrefixes: []string{"a:b"}, + }, errors.InvalidArgument}, + {"pattern too long", driver.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn}, LogEventFilterPattern: strings.Repeat("a", maxLiveTailPatternLen+1), + }, errors.InvalidArgument}, + {"bad pattern", driver.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn}, LogEventFilterPattern: "{ $.a = ", + }, errors.InvalidArgument}, + {"unknown group", driver.StartLiveTailInput{LogGroupIdentifiers: []string{liveTailTestARN + "nope"}}, errors.NotFound}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := m.StartLiveTail(context.Background(), &tt.in) + require.Error(t, err) + assert.Equal(t, tt.code, errors.GetCode(err), err.Error()) + }) + } +} + +func TestLiveTailDeliversMatchingEvents(t *testing.T) { + m, fc := newLiveTailMock(t) + ctx := context.Background() + + sess, err := m.StartLiveTail(ctx, &driver.StartLiveTailInput{ + LogGroupIdentifiers: []string{liveTailTestARN + "lt-a"}, + LogStreamNamePrefixes: []string{"app"}, + LogEventFilterPattern: "ERROR", + }) + require.NoError(t, err) + + defer sess.Close() + + ts := fc.Now() + for _, target := range []struct{ group, stream, msg string }{ + {"lt-a", "app-1", "ERROR disk full"}, + {"lt-a", "app-1", "INFO all good"}, + {"lt-a", "other", "ERROR other stream"}, + {"lt-b", "app-1", "ERROR other group"}, + } { + require.NoError(t, m.PutLogEvents(ctx, target.group, target.stream, + []driver.LogEvent{{Timestamp: ts, Message: target.msg}})) + } + + u := nextResults(t, sess) + require.Len(t, u.Results, 1) + + got := u.Results[0] + assert.Equal(t, liveTailTestARN+"lt-a", got.LogGroupIdentifier) + assert.Equal(t, "app-1", got.LogStreamName) + assert.Equal(t, "ERROR disk full", got.Message) + assert.Equal(t, ts, got.Timestamp) + assert.False(t, got.IngestionTime.IsZero()) + assert.False(t, u.Sampled) +} + +func TestLiveTailStreamNamesAndMasking(t *testing.T) { + m, fc := newLiveTailMock(t) + ctx := context.Background() + + _, err := m.PutDataProtectionPolicy(ctx, "lt-a", dpPolicy("EmailAddress")) + require.NoError(t, err) + + sess, err := m.StartLiveTail(ctx, &driver.StartLiveTailInput{ + LogGroupIdentifiers: []string{liveTailTestARN + "lt-a"}, + LogStreamNames: []string{"other"}, + }) + require.NoError(t, err) + + defer sess.Close() + + require.NoError(t, m.PutLogEvents(ctx, "lt-a", "app-1", []driver.LogEvent{{Timestamp: fc.Now(), Message: "skip"}})) + require.NoError(t, m.PutLogEvents(ctx, "lt-a", "other", []driver.LogEvent{{Timestamp: fc.Now(), Message: "mail a@b.io"}})) + + u := nextResults(t, sess) + require.Len(t, u.Results, 1) + assert.Equal(t, "other", u.Results[0].LogStreamName) + assert.Equal(t, "mail ******", u.Results[0].Message) +} + +func TestLiveTailSamplesAndBuffers(t *testing.T) { + m, _ := newLiveTailMock(t) + + tail, err := m.newLiveTail(&driver.StartLiveTailInput{LogGroupIdentifiers: []string{liveTailTestARN + "lt-a"}}) + require.NoError(t, err) + + events := make([]driver.LogEvent, liveTailBufferedEvents+100) + for i := range events { + events[i] = driver.LogEvent{Message: fmt.Sprintf("m%d", i)} + } + + tail.offer("lt-a", "s", events) + + u := tail.drain() + assert.True(t, u.Sampled) + assert.Len(t, u.Results, liveTailMaxEventsPerUpdate) + assert.Equal(t, "m100", u.Results[0].Message, "the oldest events beyond the buffer are dropped") + + empty := tail.drain() + assert.False(t, empty.Sampled) + assert.NotNil(t, empty.Results) + assert.Empty(t, empty.Results) + + for i := range liveTailBufferedUpdates + 1 { + tail.send(driver.LiveTailUpdate{Results: []driver.LiveTailEvent{{Message: fmt.Sprint(i)}}}) + } + + first := <-tail.out + assert.Equal(t, "1", first.Results[0].Message, "the oldest queued update is dropped") + assert.Len(t, tail.out, liveTailBufferedUpdates-1) +} + +func TestLiveTailEndsAndReleases(t *testing.T) { + m, fc := newLiveTailMock(t) + in := &driver.StartLiveTailInput{LogGroupIdentifiers: []string{liveTailTestARN + "lt-a"}} + before := runtime.NumGoroutine() + + // Timeout after three hours on the mock clock. + timed, err := m.StartLiveTail(context.Background(), in) + require.NoError(t, err) + fc.Advance(liveTailSessionTimeout) + waitEnded(t, timed) + assert.True(t, timed.TimedOut()) + + // Context cancellation (client disconnect) and Close. + ctx, cancel := context.WithCancel(context.Background()) + byCtx, err := m.StartLiveTail(ctx, in) + require.NoError(t, err) + + closed, err := m.StartLiveTail(context.Background(), in) + require.NoError(t, err) + + cancel() + closed.Close() + closed.Close() + waitEnded(t, byCtx) + waitEnded(t, closed) + assert.False(t, byCtx.TimedOut()) + assert.False(t, closed.TimedOut()) + + // Polled by hand: require.Eventually runs its condition on a goroutine of + // its own, which would skew the count. + deadline := time.Now().Add(liveTailWait) + for time.Now().Before(deadline) { + m.liveTailMu.Lock() + n := len(m.liveTails) + m.liveTailMu.Unlock() + + if n == 0 && runtime.NumGoroutine() <= before { + return + } + + time.Sleep(10 * time.Millisecond) + } + + t.Fatalf("sessions or goroutines left: goroutines %d, before %d", runtime.NumGoroutine(), before) +} + +func TestLiveTailConcurrentSessionLimit(t *testing.T) { + m, _ := newLiveTailMock(t) + in := &driver.StartLiveTailInput{LogGroupIdentifiers: []string{liveTailTestARN + "lt-a"}} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + for range maxLiveTailSessions { + _, err := m.StartLiveTail(ctx, in) + require.NoError(t, err) + } + + _, err := m.StartLiveTail(ctx, in) + require.Error(t, err) + assert.Equal(t, errors.ResourceExhausted, errors.GetCode(err)) +} diff --git a/server/aws/cloudwatchlogs/export_tasks.go b/server/aws/cloudwatchlogs/export_tasks.go index 76b152eae..fbc014af3 100644 --- a/server/aws/cloudwatchlogs/export_tasks.go +++ b/server/aws/cloudwatchlogs/export_tasks.go @@ -11,8 +11,8 @@ import ( // maxDescribeExportTasks is the DescribeExportTasks limit maximum and default. const maxDescribeExportTasks = 50 -// dispatchExport handles the export task operations, delegating the policy -// operations to dispatchPolicies. +// dispatchExport handles the export task operations and StartLiveTail, +// delegating the policy operations to dispatchPolicies. func (h *Handler) dispatchExport(w http.ResponseWriter, r *http.Request, op string) { switch op { case "CreateExportTask": @@ -21,6 +21,8 @@ func (h *Handler) dispatchExport(w http.ResponseWriter, r *http.Request, op stri h.describeExportTasks(w, r) case "CancelExportTask": h.cancelExportTask(w, r) + case "StartLiveTail": + h.startLiveTail(w, r) default: h.dispatchPolicies(w, r, op) } diff --git a/server/aws/cloudwatchlogs/handler.go b/server/aws/cloudwatchlogs/handler.go index 7c695e290..2aa0163c6 100644 --- a/server/aws/cloudwatchlogs/handler.go +++ b/server/aws/cloudwatchlogs/handler.go @@ -22,6 +22,8 @@ // CreateExportTask DescribeExportTasks CancelExportTask // PutDataProtectionPolicy GetDataProtectionPolicy DeleteDataProtectionPolicy // PutAccountPolicy DescribeAccountPolicies DeleteAccountPolicy +// StartLiveTail (an event stream; SDKs send it to the stream-logs. host prefix, +// and dispatch is by X-Amz-Target, so the prefixed host needs no extra route) // TagResource / UntagResource / ListTagsForResource (and the legacy *LogGroup forms) // // CloudWatch Logs-only operations go through the optional driver.AWSLogging diff --git a/server/aws/cloudwatchlogs/live_tail.go b/server/aws/cloudwatchlogs/live_tail.go new file mode 100644 index 000000000..419970ec4 --- /dev/null +++ b/server/aws/cloudwatchlogs/live_tail.go @@ -0,0 +1,142 @@ +package cloudwatchlogs + +import ( + "encoding/json" + "net/http" + + "github.com/stackshy/cloudemu/v2/internal/idgen" + "github.com/stackshy/cloudemu/v2/server/wire" + "github.com/stackshy/cloudemu/v2/server/wire/awseventstream" + logdriver "github.com/stackshy/cloudemu/v2/services/logging/driver" +) + +// Live Tail event-stream event and exception types. +const ( + eventSessionStart = "sessionStart" + eventSessionUpdate = "sessionUpdate" + sessionTimeoutException = "SessionTimeoutException" + sessionTimeoutMessage = "The Live Tail session has timed out after three hours." +) + +type startLiveTailRequest struct { + LogGroupIdentifiers []string `json:"logGroupIdentifiers"` + LogStreamNames []string `json:"logStreamNames"` + LogStreamNamePrefixes []string `json:"logStreamNamePrefixes"` + LogEventFilterPattern string `json:"logEventFilterPattern"` +} + +type liveTailSessionStartJSON struct { + RequestID string `json:"requestId"` + SessionID string `json:"sessionId"` + LogGroupIdentifiers []string `json:"logGroupIdentifiers"` + LogStreamNames []string `json:"logStreamNames,omitempty"` + LogStreamNamePrefixes []string `json:"logStreamNamePrefixes,omitempty"` + LogEventFilterPattern string `json:"logEventFilterPattern,omitempty"` +} + +type liveTailLogEventJSON struct { + LogGroupIdentifier string `json:"logGroupIdentifier"` + LogStreamName string `json:"logStreamName"` + Message string `json:"message"` + Timestamp int64 `json:"timestamp"` + IngestionTime int64 `json:"ingestionTime"` +} + +type liveTailExceptionJSON struct { + Message string `json:"message"` +} + +type liveTailSessionUpdateJSON struct { + SessionMetadata struct { + Sampled bool `json:"sampled"` + } `json:"sessionMetadata"` + SessionResults []liveTailLogEventJSON `json:"sessionResults"` +} + +// startLiveTail handles StartLiveTail. Request errors are ordinary JSON +// errors. Once the session is open the response is an event stream: the +// initial-response, a sessionStart event, then the session's updates until +// the client disconnects, the server shuts down, or the session times out. +func (h *Handler) startLiveTail(w http.ResponseWriter, r *http.Request) { + ext, ok := h.awsLogs(w, "StartLiveTail") + if !ok { + return + } + + var req startLiveTailRequest + if !wire.DecodeJSON(w, r, &req) { + return + } + + ctx, cancel := wire.StreamContext(r.Context()) + defer cancel() + + sess, err := ext.StartLiveTail(ctx, &logdriver.StartLiveTailInput{ + LogGroupIdentifiers: req.LogGroupIdentifiers, + LogStreamNames: req.LogStreamNames, + LogStreamNamePrefixes: req.LogStreamNamePrefixes, + LogEventFilterPattern: req.LogEventFilterPattern, + }) + if err != nil { + writeErr(w, err) + + return + } + + defer sess.Close() + + stream := awseventstream.Start(w) + + if stream.Event(awseventstream.InitialResponse, []byte("{}")) != nil { + return + } + + start, _ := json.Marshal(liveTailSessionStartJSON{ + RequestID: idgen.UUID(), + SessionID: sess.SessionID(), + LogGroupIdentifiers: req.LogGroupIdentifiers, + LogStreamNames: req.LogStreamNames, + LogStreamNamePrefixes: req.LogStreamNamePrefixes, + LogEventFilterPattern: req.LogEventFilterPattern, + }) + if stream.Event(eventSessionStart, start) != nil { + return + } + + streamLiveTail(stream, sess) +} + +// streamLiveTail writes each update until the session ends or a write fails. +// A timed-out session ends with SessionTimeoutException. +func streamLiveTail(stream *awseventstream.Writer, sess logdriver.LiveTailSession) { + for u := range sess.Updates() { + payload, _ := json.Marshal(toLiveTailUpdate(u)) + if stream.Event(eventSessionUpdate, payload) != nil { + return + } + } + + if sess.TimedOut() { + payload, _ := json.Marshal(liveTailExceptionJSON{Message: sessionTimeoutMessage}) + _ = stream.Exception(sessionTimeoutException, payload) + } +} + +func toLiveTailUpdate(u logdriver.LiveTailUpdate) liveTailSessionUpdateJSON { + var out liveTailSessionUpdateJSON + + out.SessionMetadata.Sampled = u.Sampled + out.SessionResults = make([]liveTailLogEventJSON, 0, len(u.Results)) + + for _, e := range u.Results { + out.SessionResults = append(out.SessionResults, liveTailLogEventJSON{ + LogGroupIdentifier: e.LogGroupIdentifier, + LogStreamName: e.LogStreamName, + Message: e.Message, + Timestamp: epochMillis(e.Timestamp), + IngestionTime: epochMillis(e.IngestionTime), + }) + } + + return out +} diff --git a/server/aws/cloudwatchlogs/live_tail_sdk_test.go b/server/aws/cloudwatchlogs/live_tail_sdk_test.go new file mode 100644 index 000000000..ac72f35dd --- /dev/null +++ b/server/aws/cloudwatchlogs/live_tail_sdk_test.go @@ -0,0 +1,278 @@ +package cloudwatchlogs_test + +import ( + "context" + "errors" + "net" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + cwl "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs" + cwltypes "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types" + + "github.com/stackshy/cloudemu/v2" + "github.com/stackshy/cloudemu/v2/config" + awsserver "github.com/stackshy/cloudemu/v2/server/aws" +) + +const liveTailSDKWait = 5 * time.Second + +// liveTailEnv is a CloudWatch Logs server and a client whose transport dials +// it whatever the host, so the SDK's stream- host prefix stays in place. +type liveTailEnv struct { + client *cwl.Client + transport *http.Transport + clock *config.FakeClock + + mu sync.Mutex + hosts []string +} + +func newLiveTailEnv(t *testing.T) *liveTailEnv { + t.Helper() + + env := &liveTailEnv{clock: config.NewFakeClock(time.Now())} + cloud := cloudemu.NewAWS(config.WithClock(env.clock)) + srv := awsserver.New(awsserver.Drivers{CloudWatchLogs: cloud.CloudWatchLogs}) + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + env.mu.Lock() + env.hosts = append(env.hosts, r.Host) + env.mu.Unlock() + srv.ServeHTTP(w, r) + })) + t.Cleanup(ts.Close) + + addr := ts.Listener.Addr().String() + env.transport = &http.Transport{ + DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, addr) + }, + } + t.Cleanup(env.transport.CloseIdleConnections) + + cfg, err := awsconfig.LoadDefaultConfig(context.Background(), + awsconfig.WithRegion("us-east-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")), + awsconfig.WithHTTPClient(&http.Client{Transport: env.transport}), + ) + if err != nil { + t.Fatalf("aws config: %v", err) + } + + env.client = cwl.NewFromConfig(cfg, func(o *cwl.Options) { + o.BaseEndpoint = aws.String("http://logs.localhost.test") + }) + + return env +} + +func (e *liveTailEnv) sawHost(host string) bool { + e.mu.Lock() + defer e.mu.Unlock() + + for _, h := range e.hosts { + if h == host { + return true + } + } + + return false +} + +// groupARN creates a group and stream and returns the group's ARN. +func (e *liveTailEnv) groupARN(t *testing.T, group, stream string) string { + t.Helper() + + mustLogGroupStream(t, e.client, group, stream) + + out, err := e.client.DescribeLogGroups(context.Background(), &cwl.DescribeLogGroupsInput{ + LogGroupNamePrefix: aws.String(group), + }) + if err != nil || len(out.LogGroups) != 1 { + t.Fatalf("DescribeLogGroups: %v %+v", err, out) + } + + return aws.ToString(out.LogGroups[0].LogGroupArn) +} + +func (e *liveTailEnv) put(t *testing.T, group, stream string, msgs ...string) { + t.Helper() + + events := make([]cwltypes.InputLogEvent, 0, len(msgs)) + for _, m := range msgs { + events = append(events, cwltypes.InputLogEvent{ + Timestamp: aws.Int64(e.clock.Now().UnixMilli()), Message: aws.String(m), + }) + } + + mustSDK(t, "PutLogEvents", func() error { + _, err := e.client.PutLogEvents(context.Background(), &cwl.PutLogEventsInput{ + LogGroupName: aws.String(group), LogStreamName: aws.String(stream), LogEvents: events, + }) + + return err + }) +} + +// nextEvent reads the stream's next event, failing on timeout or stream end. +func nextEvent(t *testing.T, ev <-chan cwltypes.StartLiveTailResponseStream) cwltypes.StartLiveTailResponseStream { + t.Helper() + + select { + case e, ok := <-ev: + if !ok { + t.Fatal("Live Tail stream ended early") + } + + return e + case <-time.After(liveTailSDKWait): + t.Fatal("no Live Tail event") + } + + return nil +} + +func TestSDKStartLiveTailStreamsMatchingEvents(t *testing.T) { + env := newLiveTailEnv(t) + arn := env.groupARN(t, "/lt/app", "web-1") + before := runtime.NumGoroutine() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + out, err := env.client.StartLiveTail(ctx, &cwl.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn}, + LogStreamNamePrefixes: []string{"web"}, + LogEventFilterPattern: aws.String("ERROR"), + }) + if err != nil { + t.Fatalf("StartLiveTail: %v", err) + } + + stream := out.GetStream() + ev := stream.Events() + + start, ok := nextEvent(t, ev).(*cwltypes.StartLiveTailResponseStreamMemberSessionStart) + if !ok || aws.ToString(start.Value.SessionId) == "" || aws.ToString(start.Value.RequestId) == "" || + len(start.Value.LogGroupIdentifiers) != 1 || aws.ToString(start.Value.LogEventFilterPattern) != "ERROR" { + t.Fatalf("sessionStart = %+v", start) + } + + if !env.sawHost("stream-logs.localhost.test") { + t.Fatal("StartLiveTail did not reach the stream-logs. host") + } + + env.put(t, "/lt/app", "web-1", "ERROR boom", "INFO fine") + + for { + up, ok := nextEvent(t, ev).(*cwltypes.StartLiveTailResponseStreamMemberSessionUpdate) + if !ok { + t.Fatal("expected a sessionUpdate") + } + + if len(up.Value.SessionResults) == 0 { + continue + } + + r := up.Value.SessionResults + if len(r) != 1 || aws.ToString(r[0].Message) != "ERROR boom" || aws.ToString(r[0].LogStreamName) != "web-1" || + aws.ToString(r[0].LogGroupIdentifier) != arn || aws.ToInt64(r[0].IngestionTime) == 0 || + up.Value.SessionMetadata == nil || up.Value.SessionMetadata.Sampled { + t.Fatalf("sessionUpdate = %+v", up.Value) + } + + break + } + + // Ending the stream from the client ends the session on the server. + _ = stream.Close() + cancel() + env.transport.CloseIdleConnections() + + deadline := time.Now().Add(liveTailSDKWait) + for runtime.NumGoroutine() > before && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + + if n := runtime.NumGoroutine(); n > before { + t.Fatalf("goroutines after the session = %d, before = %d", n, before) + } +} + +func TestSDKStartLiveTailTimeout(t *testing.T) { + env := newLiveTailEnv(t) + arn := env.groupARN(t, "/lt/timeout", "s") + + out, err := env.client.StartLiveTail(context.Background(), &cwl.StartLiveTailInput{LogGroupIdentifiers: []string{arn}}) + if err != nil { + t.Fatalf("StartLiveTail: %v", err) + } + + stream := out.GetStream() + defer stream.Close() + + if _, ok := nextEvent(t, stream.Events()).(*cwltypes.StartLiveTailResponseStreamMemberSessionStart); !ok { + t.Fatal("expected sessionStart first") + } + + env.clock.Advance(3 * time.Hour) + + deadline := time.After(liveTailSDKWait) + + for { + select { + case _, ok := <-stream.Events(): + if ok { + continue + } + + var te *cwltypes.SessionTimeoutException + if !errors.As(stream.Err(), &te) { + t.Fatalf("stream error = %v, want SessionTimeoutException", stream.Err()) + } + + return + case <-deadline: + t.Fatal("session did not time out") + } + } +} + +func TestSDKStartLiveTailErrors(t *testing.T) { + env := newLiveTailEnv(t) + arn := env.groupARN(t, "/lt/err", "s") + missing := strings.Replace(arn, "/lt/err", "/lt/missing", 1) + + tests := []struct { + name string + in *cwl.StartLiveTailInput + want string + }{ + {"name not ARN", &cwl.StartLiveTailInput{LogGroupIdentifiers: []string{"/lt/err"}}, "InvalidParameterException"}, + {"names and prefixes", &cwl.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn}, LogStreamNames: []string{"s"}, LogStreamNamePrefixes: []string{"s"}, + }, "InvalidParameterException"}, + {"bad pattern", &cwl.StartLiveTailInput{ + LogGroupIdentifiers: []string{arn}, LogEventFilterPattern: aws.String("{ $.a = "), + }, "InvalidParameterException"}, + {"unknown group", &cwl.StartLiveTailInput{LogGroupIdentifiers: []string{missing}}, "ResourceNotFoundException"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := env.client.StartLiveTail(context.Background(), tt.in) + if err == nil || !strings.Contains(err.Error(), tt.want) { + t.Fatalf("err = %v, want %s", err, tt.want) + } + }) + } +} diff --git a/server/aws/kinesis/streaming_ops.go b/server/aws/kinesis/streaming_ops.go index 477a27497..b5d4435e9 100644 --- a/server/aws/kinesis/streaming_ops.go +++ b/server/aws/kinesis/streaming_ops.go @@ -5,22 +5,12 @@ import ( "net/http" "time" - "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream" - "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream/eventstreamapi" - "github.com/stackshy/cloudemu/v2/server/wire" + "github.com/stackshy/cloudemu/v2/server/wire/awseventstream" kinesisdriver "github.com/stackshy/cloudemu/v2/services/kinesis/driver" ) -const ( - contentTypeEventStream = "application/vnd.amazon.eventstream" - contentTypeJSON = "application/json" - eventTypeSubscribe = "SubscribeToShardEvent" - // eventTypeInitial is the first frame Kinesis sends on a SubscribeToShard - // connection. The SDK blocks its SubscribeToShard call until it arrives, so it - // must precede any data event. - eventTypeInitial = "initial-response" -) +const eventTypeSubscribe = "SubscribeToShardEvent" // startingPositionJSON is the {Type,SequenceNumber,Timestamp} wire shape of a // SubscribeToShard StartingPosition. Timestamp is epoch seconds. @@ -82,21 +72,17 @@ func (h *Handler) subscribeToShard(w http.ResponseWriter, r *http.Request) { // writeSubscribeEvent frames res as a single SubscribeToShardEvent over the // eventstream protocol. func writeSubscribeEvent(w http.ResponseWriter, res *kinesisdriver.SubscribeToShardResult) { - w.Header().Set("Content-Type", contentTypeEventStream) // Deliberately do NOT force "Connection: close" here. Closing the connection // as the handler returns races the client's in-flight read of the final // frame, intermittently dropping it under CI load; normal keep-alive chunked // termination delivers every frame first. Any later teardown of the pooled // connection surfaces as a benign "use of closed network connection" that the // stream tests tolerate, never as lost data. - w.WriteHeader(http.StatusOK) - - enc := eventstream.NewEncoder() - flusher, _ := w.(http.Flusher) + stream := awseventstream.Start(w) // The SDK's SubscribeToShard call blocks until it reads the initial-response // event, so emit it (empty SubscribeToShardOutput body) before any data. - writeEvent(w, enc, flusher, eventTypeInitial, []byte("{}")) + _ = stream.Event(awseventstream.InitialResponse, []byte("{}")) payload, err := json.Marshal(subscribeToShardEventJSON{ Records: recordsToWire(res.Records), @@ -107,20 +93,5 @@ func writeSubscribeEvent(w http.ResponseWriter, res *kinesisdriver.SubscribeToSh return } - writeEvent(w, enc, flusher, eventTypeSubscribe, payload) -} - -// writeEvent frames one JSON event of the given type and flushes it. -func writeEvent(w http.ResponseWriter, enc *eventstream.Encoder, flusher http.Flusher, eventType string, payload []byte) { - var hdr eventstream.Headers - - hdr.Set(eventstreamapi.MessageTypeHeader, eventstream.StringValue(eventstreamapi.EventMessageType)) - hdr.Set(eventstreamapi.EventTypeHeader, eventstream.StringValue(eventType)) - hdr.Set(eventstreamapi.ContentTypeHeader, eventstream.StringValue(contentTypeJSON)) - - _ = enc.Encode(w, eventstream.Message{Headers: hdr, Payload: payload}) - - if flusher != nil { - flusher.Flush() - } + _ = stream.Event(eventTypeSubscribe, payload) } diff --git a/server/serverkit/middleware.go b/server/serverkit/middleware.go index 6220372c3..c368cb3ae 100644 --- a/server/serverkit/middleware.go +++ b/server/serverkit/middleware.go @@ -97,3 +97,7 @@ func (w *statusWriter) WriteHeader(code int) { w.ResponseWriter.WriteHeader(code) } + +// Unwrap exposes the underlying writer to http.ResponseController, so a +// streaming handler can still flush through request logging. +func (w *statusWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter } diff --git a/server/serverkit/serverkit.go b/server/serverkit/serverkit.go index 6bf0f794b..ae8166969 100644 --- a/server/serverkit/serverkit.go +++ b/server/serverkit/serverkit.go @@ -49,6 +49,7 @@ import ( cgrpc "github.com/stackshy/cloudemu/v2/server/grpc" bigtableadmingrpc "github.com/stackshy/cloudemu/v2/server/grpc/bigtableadmin" ociserver "github.com/stackshy/cloudemu/v2/server/oci" + "github.com/stackshy/cloudemu/v2/server/wire" "github.com/stackshy/cloudemu/v2/server/wire/awsidentity" btdriver "github.com/stackshy/cloudemu/v2/services/bigtable/driver" "github.com/stackshy/cloudemu/v2/services/kubernetes" @@ -1110,16 +1111,12 @@ func (a *App) buildServers() ([]listenerServer, endpointSet, error) { eps.Azure = fmt.Sprintf("https://%s", addr) } - servers = append(servers, &httpServer{ - label: p, - srv: &http.Server{ - Addr: addr, - Handler: a.handlerFor(a.backends[p], a.seedFor(p)), - TLSConfig: tlsCfg, - ReadHeaderTimeout: serverReadHeaderTimeout, - }, - tls: isTLS, - }) + servers = append(servers, newHTTPServer(p, &http.Server{ + Addr: addr, + Handler: a.handlerFor(a.backends[p], a.seedFor(p)), + TLSConfig: tlsCfg, + ReadHeaderTimeout: serverReadHeaderTimeout, + }, isTLS)) } if a.k8sBackend != nil { @@ -1133,16 +1130,12 @@ func (a *App) buildServers() ([]listenerServer, endpointSet, error) { return nil, eps, fmt.Errorf("kubernetes data-plane TLS: %w", err) } - servers = append(servers, &httpServer{ - label: "kubernetes", - srv: &http.Server{ - Addr: addr, - Handler: a.handlerFor(a.k8sBackend, nil), - TLSConfig: k8sTLS, - ReadHeaderTimeout: serverReadHeaderTimeout, - }, - tls: true, - }) + servers = append(servers, newHTTPServer("kubernetes", &http.Server{ + Addr: addr, + Handler: a.handlerFor(a.k8sBackend, nil), + TLSConfig: k8sTLS, + ReadHeaderTimeout: serverReadHeaderTimeout, + }, true)) // Show the reachable (advertised) endpoint, not the bind address. eps.Kubernetes = fmt.Sprintf("https://%s", net.JoinHostPort(a.advertiseHost, a.k8sPort)) } @@ -1310,6 +1303,20 @@ type httpServer struct { label string srv *http.Server tls bool + // stopStreams ends the long-lived streaming responses (such as a Live + // Tail session) when shutdown starts, so they do not hold the graceful + // shutdown until its timeout. + stopStreams context.CancelFunc +} + +// newHTTPServer wraps srv, giving its connections a base context that carries +// the server's stream-shutdown signal (see wire.StreamContext). +func newHTTPServer(label string, srv *http.Server, isTLS bool) *httpServer { + streams, stop := context.WithCancel(context.Background()) + base := wire.WithShutdown(context.Background(), streams) + srv.BaseContext = func(net.Listener) context.Context { return base } + + return &httpServer{label: label, srv: srv, tls: isTLS, stopStreams: stop} } func (h *httpServer) name() string { return h.label } @@ -1323,7 +1330,11 @@ func (h *httpServer) serve(ln net.Listener) error { return h.srv.Serve(ln) } -func (h *httpServer) shutdown(ctx context.Context) error { return h.srv.Shutdown(ctx) } +func (h *httpServer) shutdown(ctx context.Context) error { + h.stopStreams() + + return h.srv.Shutdown(ctx) +} // grpcServer adapts the server/grpc transport wrapper to listenerServer, so the // optional gRPC endpoint binds and drains through the same loop as the HTTP diff --git a/server/wire/awseventstream/awseventstream.go b/server/wire/awseventstream/awseventstream.go new file mode 100644 index 000000000..038a53195 --- /dev/null +++ b/server/wire/awseventstream/awseventstream.go @@ -0,0 +1,75 @@ +// Package awseventstream writes AWS event-stream responses +// (application/vnd.amazon.eventstream): JSON events and modeled exceptions, +// each flushed as it is written, using the SDK's own frame encoder. +package awseventstream + +import ( + "errors" + "net/http" + + "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream" + "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream/eventstreamapi" +) + +const ( + // ContentType is the response Content-Type of an event stream. + ContentType = "application/vnd.amazon.eventstream" + contentTypeJSON = "application/json" + // InitialResponse is the event type of the first frame of an AWS JSON + // event stream. SDKs block the operation call until it arrives. + InitialResponse = "initial-response" +) + +// Writer frames events onto one HTTP response. +type Writer struct { + w http.ResponseWriter + rc *http.ResponseController + enc *eventstream.Encoder +} + +// Start writes the 200 event-stream response header and returns a Writer +// for its frames. +func Start(w http.ResponseWriter) *Writer { + w.Header().Set("Content-Type", ContentType) + w.WriteHeader(http.StatusOK) + + return &Writer{w: w, rc: http.NewResponseController(w), enc: eventstream.NewEncoder()} +} + +// Event writes one JSON event of the given :event-type and flushes it. The +// error is non-nil once the client has gone away. +func (s *Writer) Event(eventType string, payload []byte) error { + var hdr eventstream.Headers + + hdr.Set(eventstreamapi.MessageTypeHeader, eventstream.StringValue(eventstreamapi.EventMessageType)) + hdr.Set(eventstreamapi.EventTypeHeader, eventstream.StringValue(eventType)) + hdr.Set(eventstreamapi.ContentTypeHeader, eventstream.StringValue(contentTypeJSON)) + + return s.write(hdr, payload) +} + +// Exception writes one modeled exception of the given :exception-type, with +// a JSON payload, and flushes it. +func (s *Writer) Exception(exceptionType string, payload []byte) error { + var hdr eventstream.Headers + + hdr.Set(eventstreamapi.MessageTypeHeader, eventstream.StringValue(eventstreamapi.ExceptionMessageType)) + hdr.Set(eventstreamapi.ExceptionTypeHeader, eventstream.StringValue(exceptionType)) + hdr.Set(eventstreamapi.ContentTypeHeader, eventstream.StringValue(contentTypeJSON)) + + return s.write(hdr, payload) +} + +func (s *Writer) write(hdr eventstream.Headers, payload []byte) error { + if err := s.enc.Encode(s.w, eventstream.Message{Headers: hdr, Payload: payload}); err != nil { + return err + } + + // A writer without flush support still delivers the frames when the + // response ends. + if err := s.rc.Flush(); err != nil && !errors.Is(err, http.ErrNotSupported) { + return err + } + + return nil +} diff --git a/server/wire/shutdown.go b/server/wire/shutdown.go new file mode 100644 index 000000000..baae85f0a --- /dev/null +++ b/server/wire/shutdown.go @@ -0,0 +1,32 @@ +package wire + +import "context" + +type shutdownKey struct{} + +// WithShutdown returns base carrying shutdown, a context the server cancels +// when it starts shutting down. Servers set it as each connection's base +// context so long-lived streaming responses can end promptly; a graceful +// shutdown otherwise waits for them until its timeout. +func WithShutdown(base, shutdown context.Context) context.Context { + return context.WithValue(base, shutdownKey{}, shutdown) +} + +// StreamContext returns a context for a long-lived streaming response. It +// ends with the request (client disconnect) or when the server starts +// shutting down, whichever comes first. The caller must call cancel. +func StreamContext(req context.Context) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(req) + + shutdown, ok := req.Value(shutdownKey{}).(context.Context) + if !ok { + return ctx, cancel + } + + stop := context.AfterFunc(shutdown, cancel) + + return ctx, func() { + stop() + cancel() + } +} diff --git a/server/wire/shutdown_test.go b/server/wire/shutdown_test.go new file mode 100644 index 000000000..acd04c5f3 --- /dev/null +++ b/server/wire/shutdown_test.go @@ -0,0 +1,31 @@ +package wire + +import ( + "context" + "testing" +) + +func TestStreamContextEndsOnShutdown(t *testing.T) { + shutdown, stop := context.WithCancel(context.Background()) + req := WithShutdown(context.Background(), shutdown) + + ctx, cancel := StreamContext(req) + defer cancel() + + if ctx.Err() != nil { + t.Fatal("stream context ended before shutdown") + } + + stop() + <-ctx.Done() +} + +func TestStreamContextEndsWithRequest(t *testing.T) { + req, end := context.WithCancel(context.Background()) + + ctx, cancel := StreamContext(req) + defer cancel() + + end() + <-ctx.Done() +} diff --git a/services/logging/driver/aws.go b/services/logging/driver/aws.go index 0be28e016..a2612a3ec 100644 --- a/services/logging/driver/aws.go +++ b/services/logging/driver/aws.go @@ -245,6 +245,43 @@ type DescribeAccountPoliciesInput struct { AccountIdentifiers []string } +// StartLiveTailInput is a StartLiveTail request. LogGroupIdentifiers are log +// group ARNs. LogStreamNames and LogStreamNamePrefixes are mutually exclusive +// and need exactly one log group. +type StartLiveTailInput struct { + LogGroupIdentifiers []string + LogStreamNames []string + LogStreamNamePrefixes []string + LogEventFilterPattern string +} + +// LiveTailEvent is one log event a Live Tail session delivers. +type LiveTailEvent struct { + LogGroupIdentifier string + LogStreamName string + Message string + Timestamp time.Time + IngestionTime time.Time +} + +// LiveTailUpdate is one sessionUpdate: the events ingested since the previous +// update. Sampled is set when more events matched than one update carries. +type LiveTailUpdate struct { + Results []LiveTailEvent + Sampled bool +} + +// LiveTailSession is a running Live Tail session. Updates delivers one update +// per interval and is closed when the session ends: on Close, when the +// StartLiveTail context ends, or on timeout, after which TimedOut reports +// true. +type LiveTailSession interface { + SessionID() string + Updates() <-chan LiveTailUpdate + TimedOut() bool + Close() +} + // AWSLogging is an optional interface implemented by the AWS CloudWatch Logs // backend for operations with no cross-provider equivalent. The CloudWatch // Logs wire handler type-asserts for it; Azure, GCP and OCI backends do not @@ -307,4 +344,8 @@ type AWSLogging interface { PutAccountPolicy(ctx context.Context, input *PutAccountPolicyInput) (*AccountPolicy, error) DescribeAccountPolicies(ctx context.Context, input DescribeAccountPoliciesInput) ([]AccountPolicy, error) DeleteAccountPolicy(ctx context.Context, policyName, policyType string) error + + // StartLiveTail opens a Live Tail session over events ingested from now + // on. The session runs until ctx ends, Close is called, or it times out. + StartLiveTail(ctx context.Context, input *StartLiveTailInput) (LiveTailSession, error) } From 67cf0eb50470cedd512a8910c58cf268ab0fe283 Mon Sep 17 00:00:00 2001 From: Nitin Kumar Date: Fri, 23 Oct 2026 09:39:34 +0530 Subject: [PATCH 2/2] docs(aws-cloudwatchlogs): Live Tail host prefix usage note --- docs/coverage/aws/cloudwatchlogs.md | 3 +- docs/coverage/azure/loganalytics.md | 3 +- docs/coverage/gcp/cloudlogging.md | 3 +- docs/coverage/nongoals/logging.md | 3 +- docs/coverage/oci/logging.md | 3 +- server/aws/cloudwatchlogs/handler.go | 5 +- .../aws/cloudwatchlogs/live_tail_sdk_test.go | 88 +++++++++++++++++++ 7 files changed, 101 insertions(+), 7 deletions(-) diff --git a/docs/coverage/aws/cloudwatchlogs.md b/docs/coverage/aws/cloudwatchlogs.md index 7c81eecc6..82d814eb5 100644 --- a/docs/coverage/aws/cloudwatchlogs.md +++ b/docs/coverage/aws/cloudwatchlogs.md @@ -81,7 +81,8 @@ LiveTailSession is a running Live Tail session. Updates delivers one update - AWS: an export task writes each log stream to a single `000000.gz` object, where CloudWatch Logs splits large streams across `000000.gz`, `000001.gz`, and so on. The events are written when the task settles to `COMPLETED` and an export-task API call observes it: at once without `--async-settle`, or on the first `DescribeExportTasks` after the settle window with it. `PENDING_CANCEL` is never reported; `CancelExportTask` moves a task straight to `CANCELLED`. - AWS: data protection detects four managed data identifiers: `EmailAddress`, `IpAddress` (IPv4 and IPv6), `AwsSecretKey` (a 40-character key next to a keyword such as `aws_secret_access_key`) and `CreditCardNumber` (13-19 digits that pass the Luhn check), plus custom data identifiers. Any other well-formed managed identifier ARN is accepted and stored but masks nothing, and the machine-learning and keyword-proximity detection AWS uses for the other identifiers is not modelled. Each masked character becomes one `*`. - AWS: data protection audit findings are not produced. No findings report goes to the `FindingsDestination`, and no `LogEventsWithFindings` metric is emitted. A CloudWatch Logs findings destination must exist, but S3 and Firehose destinations are not checked. -- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported, and Live Tail is not modelled. +- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported. Live Tail always shows the masked text. - AWS: an account data protection policy applies to existing groups at once, without AWS's eventual-consistency delay of up to 5 minutes. `DescribeLogGroups` reports `dataProtectionStatus` `ACTIVATED` only for a group-level policy, and the `DELETED`, `ARCHIVED` and `DISABLED` statuses are never reported. - AWS: `FIELD_INDEX_POLICY`, `TRANSFORMER_POLICY` and `METRIC_EXTRACTION_POLICY` account policies are validated, stored and returned, but they have no data-plane effect: no field indexes, no transformation at ingest, and no change to EMF metric extraction. Transformer and metric extraction documents only have to be valid JSON. A `SUBSCRIPTION_FILTER_POLICY` delivers only to Lambda destinations, the same as group subscription filters. - AWS: account policy quotas are one data protection policy and one subscription filter policy per account, and, for each of the other types, one unscoped policy plus up to 20 scoped policies (5 for metric extraction). Scoped policies of one type can't share overlapping `LogGroupNamePrefix` values or an identical data source, and a type can have only one `NOT IN` policy. AWS's finer `IN`/`NOT IN` containment rules for metric extraction policies are not modelled. `DescribeAccountPolicies` returns every match on one page and never returns a `nextToken`. +- AWS: `StartLiveTail` is sent to a prefixed host. aws-sdk-go-v2 adds `stream-` and botocore (the AWS CLI, boto3) adds `streaming-`, so a client pointed at `http://localhost:4566` dials `stream-localhost` or `streaming-localhost`, which do not resolve. cloudemu routes on `X-Amz-Target`, so any host that reaches the port works. With the AWS CLI, use a wildcard-DNS host such as `--endpoint-url http://logs.localtest.me:4566`, or add `127.0.0.1 streaming-localhost` to `/etc/hosts`. With boto3, pass `botocore.config.Config(inject_host_prefix=False)`. With aws-sdk-go-v2, set the context flag in an API option: `o.APIOptions = append(o.APIOptions, func(s *middleware.Stack) error { return s.Initialize.Add(middleware.InitializeMiddlewareFunc("noHostPrefix", func(ctx context.Context, in middleware.InitializeInput, next middleware.InitializeHandler) (middleware.InitializeOutput, middleware.Metadata, error) { return next.HandleInitialize(smithyhttp.DisableEndpointHostPrefix(ctx, true), in) }), middleware.Before) })`. Or use the same wildcard-DNS host, or an `http.Transport` whose dialer ignores the host. diff --git a/docs/coverage/azure/loganalytics.md b/docs/coverage/azure/loganalytics.md index 532cef441..21c105d7a 100644 --- a/docs/coverage/azure/loganalytics.md +++ b/docs/coverage/azure/loganalytics.md @@ -32,7 +32,8 @@ Azure's `logging` service · portable interface `driver.Logging` · [Azure index - AWS: an export task writes each log stream to a single `000000.gz` object, where CloudWatch Logs splits large streams across `000000.gz`, `000001.gz`, and so on. The events are written when the task settles to `COMPLETED` and an export-task API call observes it: at once without `--async-settle`, or on the first `DescribeExportTasks` after the settle window with it. `PENDING_CANCEL` is never reported; `CancelExportTask` moves a task straight to `CANCELLED`. - AWS: data protection detects four managed data identifiers: `EmailAddress`, `IpAddress` (IPv4 and IPv6), `AwsSecretKey` (a 40-character key next to a keyword such as `aws_secret_access_key`) and `CreditCardNumber` (13-19 digits that pass the Luhn check), plus custom data identifiers. Any other well-formed managed identifier ARN is accepted and stored but masks nothing, and the machine-learning and keyword-proximity detection AWS uses for the other identifiers is not modelled. Each masked character becomes one `*`. - AWS: data protection audit findings are not produced. No findings report goes to the `FindingsDestination`, and no `LogEventsWithFindings` metric is emitted. A CloudWatch Logs findings destination must exist, but S3 and Firehose destinations are not checked. -- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported, and Live Tail is not modelled. +- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported. Live Tail always shows the masked text. - AWS: an account data protection policy applies to existing groups at once, without AWS's eventual-consistency delay of up to 5 minutes. `DescribeLogGroups` reports `dataProtectionStatus` `ACTIVATED` only for a group-level policy, and the `DELETED`, `ARCHIVED` and `DISABLED` statuses are never reported. - AWS: `FIELD_INDEX_POLICY`, `TRANSFORMER_POLICY` and `METRIC_EXTRACTION_POLICY` account policies are validated, stored and returned, but they have no data-plane effect: no field indexes, no transformation at ingest, and no change to EMF metric extraction. Transformer and metric extraction documents only have to be valid JSON. A `SUBSCRIPTION_FILTER_POLICY` delivers only to Lambda destinations, the same as group subscription filters. - AWS: account policy quotas are one data protection policy and one subscription filter policy per account, and, for each of the other types, one unscoped policy plus up to 20 scoped policies (5 for metric extraction). Scoped policies of one type can't share overlapping `LogGroupNamePrefix` values or an identical data source, and a type can have only one `NOT IN` policy. AWS's finer `IN`/`NOT IN` containment rules for metric extraction policies are not modelled. `DescribeAccountPolicies` returns every match on one page and never returns a `nextToken`. +- AWS: `StartLiveTail` is sent to a prefixed host. aws-sdk-go-v2 adds `stream-` and botocore (the AWS CLI, boto3) adds `streaming-`, so a client pointed at `http://localhost:4566` dials `stream-localhost` or `streaming-localhost`, which do not resolve. cloudemu routes on `X-Amz-Target`, so any host that reaches the port works. With the AWS CLI, use a wildcard-DNS host such as `--endpoint-url http://logs.localtest.me:4566`, or add `127.0.0.1 streaming-localhost` to `/etc/hosts`. With boto3, pass `botocore.config.Config(inject_host_prefix=False)`. With aws-sdk-go-v2, set the context flag in an API option: `o.APIOptions = append(o.APIOptions, func(s *middleware.Stack) error { return s.Initialize.Add(middleware.InitializeMiddlewareFunc("noHostPrefix", func(ctx context.Context, in middleware.InitializeInput, next middleware.InitializeHandler) (middleware.InitializeOutput, middleware.Metadata, error) { return next.HandleInitialize(smithyhttp.DisableEndpointHostPrefix(ctx, true), in) }), middleware.Before) })`. Or use the same wildcard-DNS host, or an `http.Transport` whose dialer ignores the host. diff --git a/docs/coverage/gcp/cloudlogging.md b/docs/coverage/gcp/cloudlogging.md index befa4d3bf..5bd60b845 100644 --- a/docs/coverage/gcp/cloudlogging.md +++ b/docs/coverage/gcp/cloudlogging.md @@ -58,7 +58,8 @@ GCPLogging is an optional interface implemented by the GCP logging backend for - AWS: an export task writes each log stream to a single `000000.gz` object, where CloudWatch Logs splits large streams across `000000.gz`, `000001.gz`, and so on. The events are written when the task settles to `COMPLETED` and an export-task API call observes it: at once without `--async-settle`, or on the first `DescribeExportTasks` after the settle window with it. `PENDING_CANCEL` is never reported; `CancelExportTask` moves a task straight to `CANCELLED`. - AWS: data protection detects four managed data identifiers: `EmailAddress`, `IpAddress` (IPv4 and IPv6), `AwsSecretKey` (a 40-character key next to a keyword such as `aws_secret_access_key`) and `CreditCardNumber` (13-19 digits that pass the Luhn check), plus custom data identifiers. Any other well-formed managed identifier ARN is accepted and stored but masks nothing, and the machine-learning and keyword-proximity detection AWS uses for the other identifiers is not modelled. Each masked character becomes one `*`. - AWS: data protection audit findings are not produced. No findings report goes to the `FindingsDestination`, and no `LogEventsWithFindings` metric is emitted. A CloudWatch Logs findings destination must exist, but S3 and Firehose destinations are not checked. -- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported, and Live Tail is not modelled. +- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported. Live Tail always shows the masked text. - AWS: an account data protection policy applies to existing groups at once, without AWS's eventual-consistency delay of up to 5 minutes. `DescribeLogGroups` reports `dataProtectionStatus` `ACTIVATED` only for a group-level policy, and the `DELETED`, `ARCHIVED` and `DISABLED` statuses are never reported. - AWS: `FIELD_INDEX_POLICY`, `TRANSFORMER_POLICY` and `METRIC_EXTRACTION_POLICY` account policies are validated, stored and returned, but they have no data-plane effect: no field indexes, no transformation at ingest, and no change to EMF metric extraction. Transformer and metric extraction documents only have to be valid JSON. A `SUBSCRIPTION_FILTER_POLICY` delivers only to Lambda destinations, the same as group subscription filters. - AWS: account policy quotas are one data protection policy and one subscription filter policy per account, and, for each of the other types, one unscoped policy plus up to 20 scoped policies (5 for metric extraction). Scoped policies of one type can't share overlapping `LogGroupNamePrefix` values or an identical data source, and a type can have only one `NOT IN` policy. AWS's finer `IN`/`NOT IN` containment rules for metric extraction policies are not modelled. `DescribeAccountPolicies` returns every match on one page and never returns a `nextToken`. +- AWS: `StartLiveTail` is sent to a prefixed host. aws-sdk-go-v2 adds `stream-` and botocore (the AWS CLI, boto3) adds `streaming-`, so a client pointed at `http://localhost:4566` dials `stream-localhost` or `streaming-localhost`, which do not resolve. cloudemu routes on `X-Amz-Target`, so any host that reaches the port works. With the AWS CLI, use a wildcard-DNS host such as `--endpoint-url http://logs.localtest.me:4566`, or add `127.0.0.1 streaming-localhost` to `/etc/hosts`. With boto3, pass `botocore.config.Config(inject_host_prefix=False)`. With aws-sdk-go-v2, set the context flag in an API option: `o.APIOptions = append(o.APIOptions, func(s *middleware.Stack) error { return s.Initialize.Add(middleware.InitializeMiddlewareFunc("noHostPrefix", func(ctx context.Context, in middleware.InitializeInput, next middleware.InitializeHandler) (middleware.InitializeOutput, middleware.Metadata, error) { return next.HandleInitialize(smithyhttp.DisableEndpointHostPrefix(ctx, true), in) }), middleware.Before) })`. Or use the same wildcard-DNS host, or an `http.Transport` whose dialer ignores the host. diff --git a/docs/coverage/nongoals/logging.md b/docs/coverage/nongoals/logging.md index 6ca70512b..76e1f1936 100644 --- a/docs/coverage/nongoals/logging.md +++ b/docs/coverage/nongoals/logging.md @@ -3,7 +3,8 @@ - AWS: an export task writes each log stream to a single `000000.gz` object, where CloudWatch Logs splits large streams across `000000.gz`, `000001.gz`, and so on. The events are written when the task settles to `COMPLETED` and an export-task API call observes it: at once without `--async-settle`, or on the first `DescribeExportTasks` after the settle window with it. `PENDING_CANCEL` is never reported; `CancelExportTask` moves a task straight to `CANCELLED`. - AWS: data protection detects four managed data identifiers: `EmailAddress`, `IpAddress` (IPv4 and IPv6), `AwsSecretKey` (a 40-character key next to a keyword such as `aws_secret_access_key`) and `CreditCardNumber` (13-19 digits that pass the Luhn check), plus custom data identifiers. Any other well-formed managed identifier ARN is accepted and stored but masks nothing, and the machine-learning and keyword-proximity detection AWS uses for the other identifiers is not modelled. Each masked character becomes one `*`. - AWS: data protection audit findings are not produced. No findings report goes to the `FindingsDestination`, and no `LogEventsWithFindings` metric is emitted. A CloudWatch Logs findings destination must exist, but S3 and Firehose destinations are not checked. -- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported, and Live Tail is not modelled. +- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported. Live Tail always shows the masked text. - AWS: an account data protection policy applies to existing groups at once, without AWS's eventual-consistency delay of up to 5 minutes. `DescribeLogGroups` reports `dataProtectionStatus` `ACTIVATED` only for a group-level policy, and the `DELETED`, `ARCHIVED` and `DISABLED` statuses are never reported. - AWS: `FIELD_INDEX_POLICY`, `TRANSFORMER_POLICY` and `METRIC_EXTRACTION_POLICY` account policies are validated, stored and returned, but they have no data-plane effect: no field indexes, no transformation at ingest, and no change to EMF metric extraction. Transformer and metric extraction documents only have to be valid JSON. A `SUBSCRIPTION_FILTER_POLICY` delivers only to Lambda destinations, the same as group subscription filters. - AWS: account policy quotas are one data protection policy and one subscription filter policy per account, and, for each of the other types, one unscoped policy plus up to 20 scoped policies (5 for metric extraction). Scoped policies of one type can't share overlapping `LogGroupNamePrefix` values or an identical data source, and a type can have only one `NOT IN` policy. AWS's finer `IN`/`NOT IN` containment rules for metric extraction policies are not modelled. `DescribeAccountPolicies` returns every match on one page and never returns a `nextToken`. +- AWS: `StartLiveTail` is sent to a prefixed host. aws-sdk-go-v2 adds `stream-` and botocore (the AWS CLI, boto3) adds `streaming-`, so a client pointed at `http://localhost:4566` dials `stream-localhost` or `streaming-localhost`, which do not resolve. cloudemu routes on `X-Amz-Target`, so any host that reaches the port works. With the AWS CLI, use a wildcard-DNS host such as `--endpoint-url http://logs.localtest.me:4566`, or add `127.0.0.1 streaming-localhost` to `/etc/hosts`. With boto3, pass `botocore.config.Config(inject_host_prefix=False)`. With aws-sdk-go-v2, set the context flag in an API option: `o.APIOptions = append(o.APIOptions, func(s *middleware.Stack) error { return s.Initialize.Add(middleware.InitializeMiddlewareFunc("noHostPrefix", func(ctx context.Context, in middleware.InitializeInput, next middleware.InitializeHandler) (middleware.InitializeOutput, middleware.Metadata, error) { return next.HandleInitialize(smithyhttp.DisableEndpointHostPrefix(ctx, true), in) }), middleware.Before) })`. Or use the same wildcard-DNS host, or an `http.Transport` whose dialer ignores the host. diff --git a/docs/coverage/oci/logging.md b/docs/coverage/oci/logging.md index d938f6db4..0524053f8 100644 --- a/docs/coverage/oci/logging.md +++ b/docs/coverage/oci/logging.md @@ -32,7 +32,8 @@ OCI's `logging` service · portable interface `driver.Logging` · [OCI index](./ - AWS: an export task writes each log stream to a single `000000.gz` object, where CloudWatch Logs splits large streams across `000000.gz`, `000001.gz`, and so on. The events are written when the task settles to `COMPLETED` and an export-task API call observes it: at once without `--async-settle`, or on the first `DescribeExportTasks` after the settle window with it. `PENDING_CANCEL` is never reported; `CancelExportTask` moves a task straight to `CANCELLED`. - AWS: data protection detects four managed data identifiers: `EmailAddress`, `IpAddress` (IPv4 and IPv6), `AwsSecretKey` (a 40-character key next to a keyword such as `aws_secret_access_key`) and `CreditCardNumber` (13-19 digits that pass the Luhn check), plus custom data identifiers. Any other well-formed managed identifier ARN is accepted and stored but masks nothing, and the machine-learning and keyword-proximity detection AWS uses for the other identifiers is not modelled. Each masked character becomes one `*`. - AWS: data protection audit findings are not produced. No findings report goes to the `FindingsDestination`, and no `LogEventsWithFindings` metric is emitted. A CloudWatch Logs findings destination must exist, but S3 and Firehose destinations are not checked. -- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported, and Live Tail is not modelled. +- AWS: `logs:Unmask` is not enforced, so any caller can pass `unmask` to `GetLogEvents`, `FilterLogEvents` and `GetLogRecord`. A `FilterLogEvents` pattern always matches the masked text. The Logs Insights `unmask` command is not supported. Live Tail always shows the masked text. - AWS: an account data protection policy applies to existing groups at once, without AWS's eventual-consistency delay of up to 5 minutes. `DescribeLogGroups` reports `dataProtectionStatus` `ACTIVATED` only for a group-level policy, and the `DELETED`, `ARCHIVED` and `DISABLED` statuses are never reported. - AWS: `FIELD_INDEX_POLICY`, `TRANSFORMER_POLICY` and `METRIC_EXTRACTION_POLICY` account policies are validated, stored and returned, but they have no data-plane effect: no field indexes, no transformation at ingest, and no change to EMF metric extraction. Transformer and metric extraction documents only have to be valid JSON. A `SUBSCRIPTION_FILTER_POLICY` delivers only to Lambda destinations, the same as group subscription filters. - AWS: account policy quotas are one data protection policy and one subscription filter policy per account, and, for each of the other types, one unscoped policy plus up to 20 scoped policies (5 for metric extraction). Scoped policies of one type can't share overlapping `LogGroupNamePrefix` values or an identical data source, and a type can have only one `NOT IN` policy. AWS's finer `IN`/`NOT IN` containment rules for metric extraction policies are not modelled. `DescribeAccountPolicies` returns every match on one page and never returns a `nextToken`. +- AWS: `StartLiveTail` is sent to a prefixed host. aws-sdk-go-v2 adds `stream-` and botocore (the AWS CLI, boto3) adds `streaming-`, so a client pointed at `http://localhost:4566` dials `stream-localhost` or `streaming-localhost`, which do not resolve. cloudemu routes on `X-Amz-Target`, so any host that reaches the port works. With the AWS CLI, use a wildcard-DNS host such as `--endpoint-url http://logs.localtest.me:4566`, or add `127.0.0.1 streaming-localhost` to `/etc/hosts`. With boto3, pass `botocore.config.Config(inject_host_prefix=False)`. With aws-sdk-go-v2, set the context flag in an API option: `o.APIOptions = append(o.APIOptions, func(s *middleware.Stack) error { return s.Initialize.Add(middleware.InitializeMiddlewareFunc("noHostPrefix", func(ctx context.Context, in middleware.InitializeInput, next middleware.InitializeHandler) (middleware.InitializeOutput, middleware.Metadata, error) { return next.HandleInitialize(smithyhttp.DisableEndpointHostPrefix(ctx, true), in) }), middleware.Before) })`. Or use the same wildcard-DNS host, or an `http.Transport` whose dialer ignores the host. diff --git a/server/aws/cloudwatchlogs/handler.go b/server/aws/cloudwatchlogs/handler.go index 2aa0163c6..4b122af28 100644 --- a/server/aws/cloudwatchlogs/handler.go +++ b/server/aws/cloudwatchlogs/handler.go @@ -22,8 +22,9 @@ // CreateExportTask DescribeExportTasks CancelExportTask // PutDataProtectionPolicy GetDataProtectionPolicy DeleteDataProtectionPolicy // PutAccountPolicy DescribeAccountPolicies DeleteAccountPolicy -// StartLiveTail (an event stream; SDKs send it to the stream-logs. host prefix, -// and dispatch is by X-Amz-Target, so the prefixed host needs no extra route) +// StartLiveTail (an event stream; aws-sdk-go-v2 sends it to a stream- host prefix +// and botocore to streaming-, and dispatch is by X-Amz-Target, so either +// prefixed host needs no extra route) // TagResource / UntagResource / ListTagsForResource (and the legacy *LogGroup forms) // // CloudWatch Logs-only operations go through the optional driver.AWSLogging diff --git a/server/aws/cloudwatchlogs/live_tail_sdk_test.go b/server/aws/cloudwatchlogs/live_tail_sdk_test.go index ac72f35dd..f4d264cc2 100644 --- a/server/aws/cloudwatchlogs/live_tail_sdk_test.go +++ b/server/aws/cloudwatchlogs/live_tail_sdk_test.go @@ -13,10 +13,14 @@ import ( "time" "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream" + "github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream/eventstreamapi" awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" cwl "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs" cwltypes "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types" + "github.com/aws/smithy-go/middleware" + smithyhttp "github.com/aws/smithy-go/transport/http" "github.com/stackshy/cloudemu/v2" "github.com/stackshy/cloudemu/v2/config" @@ -31,6 +35,8 @@ type liveTailEnv struct { client *cwl.Client transport *http.Transport clock *config.FakeClock + url string + cfg aws.Config mu sync.Mutex hosts []string @@ -51,6 +57,7 @@ func newLiveTailEnv(t *testing.T) *liveTailEnv { })) t.Cleanup(ts.Close) + env.url = ts.URL addr := ts.Listener.Addr().String() env.transport = &http.Transport{ DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) { @@ -68,6 +75,7 @@ func newLiveTailEnv(t *testing.T) *liveTailEnv { t.Fatalf("aws config: %v", err) } + env.cfg = cfg env.client = cwl.NewFromConfig(cfg, func(o *cwl.Options) { o.BaseEndpoint = aws.String("http://logs.localhost.test") }) @@ -276,3 +284,83 @@ func TestSDKStartLiveTailErrors(t *testing.T) { }) } } + +// noHostPrefix is the documented way to reach a localhost endpoint from +// aws-sdk-go-v2: it turns off the stream- host prefix for every call. +func noHostPrefix(s *middleware.Stack) error { + return s.Initialize.Add(middleware.InitializeMiddlewareFunc("noHostPrefix", + func(ctx context.Context, in middleware.InitializeInput, next middleware.InitializeHandler) ( + middleware.InitializeOutput, middleware.Metadata, error, + ) { + return next.HandleInitialize(smithyhttp.DisableEndpointHostPrefix(ctx, true), in) + }), middleware.Before) +} + +func TestSDKStartLiveTailWithoutHostPrefix(t *testing.T) { + env := newLiveTailEnv(t) + arn := env.groupARN(t, "/lt/plain", "s") + + // A plain 127.0.0.1 endpoint with the default transport, as a user would + // configure it. + cfg := env.cfg.Copy() + cfg.HTTPClient = http.DefaultClient + client := cwl.NewFromConfig(cfg, func(o *cwl.Options) { + o.BaseEndpoint = aws.String(env.url) + o.APIOptions = append(o.APIOptions, noHostPrefix) + }) + + out, err := client.StartLiveTail(context.Background(), &cwl.StartLiveTailInput{LogGroupIdentifiers: []string{arn}}) + if err != nil { + t.Fatalf("StartLiveTail: %v", err) + } + + stream := out.GetStream() + defer stream.Close() + + if _, ok := nextEvent(t, stream.Events()).(*cwltypes.StartLiveTailResponseStreamMemberSessionStart); !ok { + t.Fatal("expected sessionStart") + } +} + +// TestStartLiveTailStreamingHostPrefix sends the request the way botocore +// does, to the streaming- host prefix, and reads the frames. +func TestStartLiveTailStreamingHostPrefix(t *testing.T) { + env := newLiveTailEnv(t) + arn := env.groupARN(t, "/lt/botocore", "s") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + body := `{"logGroupIdentifiers":["` + arn + `"]}` + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://streaming-logs.localhost.test/", strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + + req.Header.Set("X-Amz-Target", "Logs_20140328.StartLiveTail") + req.Header.Set("Content-Type", "application/x-amz-json-1.1") + + resp, err := (&http.Client{Transport: env.transport}).Do(req) + if err != nil { + t.Fatalf("StartLiveTail: %v", err) + } + defer resp.Body.Close() + + if !env.sawHost("streaming-logs.localhost.test") { + t.Fatal("StartLiveTail did not reach the streaming-logs. host") + } + + dec := eventstream.NewDecoder() + + for _, want := range []string{"initial-response", "sessionStart", "sessionUpdate"} { + msg, err := dec.Decode(resp.Body, nil) + if err != nil { + t.Fatalf("decode %s: %v", want, err) + } + + if got := msg.Headers.Get(eventstreamapi.EventTypeHeader); got == nil || got.String() != want { + t.Fatalf("event type = %v, want %s", got, want) + } + } +}