feat(accesslog): export HTTP access logs to S3 as gzipped JSONL #406
+599
−6
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| // Package accesslog exports structured HTTP access log events to S3 as | ||
| // batched JSONL objects (gzip-compressed by default), suitable for ingestion | ||
| // by log analysis and security monitoring pipelines. | ||
| package accesslog | ||
|
|
||
| import ( | ||
| "net/http" | ||
| "time" | ||
|
|
||
| "github.com/alecthomas/errors" | ||
| ) | ||
|
|
||
| // Compression modes for exported objects. | ||
| const ( | ||
| CompressionGzip = "gzip" | ||
| CompressionNone = "none" | ||
| ) | ||
|
|
||
| // Config configures access log export to S3. Export is enabled when Bucket is | ||
| // non-empty. Connection parameters (endpoint, region, credentials) come from | ||
| // the global s3 block. | ||
| type Config struct { | ||
| Bucket string `hcl:"bucket" help:"S3 bucket to export access log events to."` | ||
| Prefix string `hcl:"prefix,optional" default:"access-logs" help:"Object key prefix for exported batches."` | ||
| FlushInterval time.Duration `hcl:"flush-interval,optional" default:"1m" help:"How often buffered events are flushed to S3."` | ||
| MaxBufferedEvents int `hcl:"max-buffered-events,optional" default:"65536" help:"Maximum events held in memory; new events are dropped when the buffer is full."` | ||
| Compression string `hcl:"compression,optional" default:"gzip" help:"Compression for exported objects: gzip or none."` | ||
| Headers map[string]string `hcl:"headers,optional" help:"Record these inbound request headers as the given event field."` | ||
| } | ||
|
|
||
| // Validate checks the configuration for invalid values. | ||
| func (c Config) Validate() error { | ||
| if c.FlushInterval <= 0 { | ||
|
There was a problem hiding this comment.
When Useful? React with 👍 / 👎. |
||
| return errors.Errorf("invalid access log flush-interval %s: must be positive", c.FlushInterval) | ||
| } | ||
| switch c.Compression { | ||
| case "", CompressionGzip, CompressionNone: | ||
| return nil | ||
|
worstell marked this conversation as resolved.
|
||
| default: | ||
| return errors.Errorf("invalid access log compression %q: must be %q or %q", c.Compression, CompressionGzip, CompressionNone) | ||
| } | ||
| } | ||
|
|
||
| // Event is a single access log record. It is serialised as one JSON object | ||
| // per line (JSONL). | ||
| type Event struct { | ||
| Timestamp time.Time `json:"timestamp"` | ||
| Method string `json:"method"` | ||
| Path string `json:"path"` | ||
| Query string `json:"query,omitempty"` | ||
| Status int `json:"status"` | ||
| BytesSent int64 `json:"bytes_sent"` | ||
| DurationMS float64 `json:"duration_ms"` | ||
| RemoteAddr string `json:"remote_addr"` | ||
| Host string `json:"host,omitempty"` | ||
| UserAgent string `json:"user_agent,omitempty"` | ||
| Headers map[string]string `json:"headers,omitempty"` | ||
| } | ||
|
|
||
| // Recorder accepts access log events. Implemented by *Writer. | ||
| type Recorder interface { | ||
| Record(event Event) | ||
| } | ||
|
|
||
| // Middleware records one Event per request to the given Recorder. | ||
| func Middleware(next http.Handler, recorder Recorder, config Config) http.Handler { | ||
| return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | ||
| rec := &responseRecorder{ResponseWriter: w} | ||
| start := time.Now() | ||
| next.ServeHTTP(rec, r) | ||
|
There was a problem hiding this comment.
If an OPA or strategy handler panics, control unwinds past this call and never reaches Useful? React with 👍 / 👎. |
||
|
|
||
| event := Event{ | ||
| Timestamp: start.UTC(), | ||
| Method: r.Method, | ||
| Path: r.URL.Path, | ||
| Query: r.URL.RawQuery, | ||
| Status: rec.statusCode(), | ||
| BytesSent: rec.bytes, | ||
| DurationMS: float64(time.Since(start)) / float64(time.Millisecond), | ||
| RemoteAddr: r.RemoteAddr, | ||
| Host: r.Host, | ||
| UserAgent: r.UserAgent(), | ||
| } | ||
| for header, field := range config.Headers { | ||
| if v := r.Header.Get(header); v != "" { | ||
| if event.Headers == nil { | ||
| event.Headers = map[string]string{} | ||
| } | ||
| event.Headers[field] = v | ||
| } | ||
| } | ||
| recorder.Record(event) | ||
| }) | ||
| } | ||
|
|
||
| type responseRecorder struct { | ||
| http.ResponseWriter | ||
| status int | ||
| bytes int64 | ||
| } | ||
|
|
||
| func (r *responseRecorder) WriteHeader(status int) { | ||
| if r.status == 0 { | ||
| r.status = status | ||
| } | ||
| r.ResponseWriter.WriteHeader(status) | ||
| } | ||
|
|
||
| func (r *responseRecorder) Write(b []byte) (int, error) { | ||
| if r.status == 0 { | ||
| r.status = http.StatusOK | ||
| } | ||
| n, err := r.ResponseWriter.Write(b) | ||
| r.bytes += int64(n) | ||
| return n, err //nolint:wrapcheck | ||
| } | ||
|
|
||
| // Flush is implemented explicitly because streaming handlers type-assert | ||
| // http.Flusher directly on the ResponseWriter they receive. | ||
| func (r *responseRecorder) Flush() { | ||
| if f, ok := r.ResponseWriter.(http.Flusher); ok { | ||
| f.Flush() | ||
| } | ||
| } | ||
|
|
||
| // Unwrap supports http.ResponseController. | ||
| func (r *responseRecorder) Unwrap() http.ResponseWriter { return r.ResponseWriter } | ||
|
|
||
| func (r *responseRecorder) statusCode() int { | ||
| if r.status == 0 { | ||
| return http.StatusOK | ||
| } | ||
| return r.status | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| package accesslog_test | ||
|
|
||
| import ( | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "testing" | ||
|
|
||
| "github.com/alecthomas/assert/v2" | ||
|
|
||
| "github.com/block/cachew/internal/accesslog" | ||
| ) | ||
|
|
||
| type recorder struct { | ||
| events []accesslog.Event | ||
| } | ||
|
|
||
| func (r *recorder) Record(event accesslog.Event) { r.events = append(r.events, event) } | ||
|
|
||
| func TestMiddleware(t *testing.T) { | ||
| rec := &recorder{} | ||
| handler := accesslog.Middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | ||
| w.WriteHeader(http.StatusTeapot) | ||
| _, _ = w.Write([]byte("hello")) //nolint:errcheck | ||
| }), rec, accesslog.Config{Headers: map[string]string{"X-Client-Id": "client_id"}}) | ||
|
|
||
| req := httptest.NewRequest(http.MethodGet, "/git/github.com/org/repo?service=git-upload-pack", nil) | ||
| req.Header.Set("X-Client-Id", "abc-123") | ||
| req.Header.Set("User-Agent", "git/2.44.0") | ||
| handler.ServeHTTP(httptest.NewRecorder(), req) | ||
|
|
||
| assert.Equal(t, 1, len(rec.events)) | ||
| event := rec.events[0] | ||
| assert.Equal(t, http.MethodGet, event.Method) | ||
| assert.Equal(t, "/git/github.com/org/repo", event.Path) | ||
| assert.Equal(t, "service=git-upload-pack", event.Query) | ||
| assert.Equal(t, http.StatusTeapot, event.Status) | ||
| assert.Equal(t, 5, int(event.BytesSent)) | ||
| assert.Equal(t, "git/2.44.0", event.UserAgent) | ||
| assert.Equal(t, map[string]string{"client_id": "abc-123"}, event.Headers) | ||
| assert.False(t, event.Timestamp.IsZero()) | ||
| } | ||
|
|
||
| func TestMiddlewareImplicitOK(t *testing.T) { | ||
| rec := &recorder{} | ||
| handler := accesslog.Middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { | ||
| _, _ = w.Write([]byte("ok")) //nolint:errcheck | ||
| }), rec, accesslog.Config{}) | ||
|
|
||
| handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) | ||
|
|
||
| assert.Equal(t, 1, len(rec.events)) | ||
| assert.Equal(t, http.StatusOK, rec.events[0].Status) | ||
| assert.Zero(t, rec.events[0].Headers) | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When access-log validation fails, this factory logs and terminates the process internally through
fatalIfErrorinstead of returning the error, preventing its caller from controlling reporting or cleanup. Return an error from the helper and handle it inmain, as required by the repository's error-handling convention.AGENTS.md reference: AGENTS.md:L24-L25
Useful? React with 👍 / 👎.