mirror of https://github.com/grafana/loki
feat(lambda-promtail): add cloudtrail log ingestion support (#9497)
**What this PR does / why we need it**: ### Add support for AWS CloudTrail audit logs ingestion using lambda-promtail Calls to AWS APIs are logged in AWS Cloudtrail and are helpful for security and debugging purposes. However, I've experienced difficulties with it: + The AWS CloudTrail service is not well integrated with Prometheus (no metrics, no alerts) and I don't want to manage alerts in CloudWatch Alerts + The search experience is painful with CloudTrail via the AWS Console (I will not elaborate 😅). This PR allows ingesting CloudTrail audit logs sent to an S3 bucket using the same approach as VPC flow logs or Load Balancer logs. **Special notes for your reviewer**: + Because the Cloudtrail file format is not text but json, we stream the json CloudTrail records instead of using the already existing scanner. + Because the Cloudtrail filename format is not the same as for the Flow log or the Load balancer log files, we need to split the regexes by service (although many AWS services seem to share the same `defaultFilenameRegex`). + In the `getLabels` function, we expect the `type` parameter to be found in the filename using the Regex. For some log files (ex: Cloudfront log files). The file name has no reference to the service name. This is why, as a default, when no type is found in the name of the file, I set it to use the key of the matching Regex expression. **Checklist** - [x] Reviewed the [`CONTRIBUTING.md`](https://github.com/grafana/loki/blob/main/CONTRIBUTING.md) guide (**required**) - [x] Documentation added - [x] Tests updated - [x] `CHANGELOG.md` updated - [x] Changes that require user attention or interaction to upgrade are documented in `docs/sources/upgrading/_index.md`pull/9558/head^2
Christophe Collot
2 years ago
committed by
GitHub
parent
a7976b584f
commit
b709b32d6a
@ -0,0 +1,32 @@ |
||||
package main |
||||
|
||||
import ( |
||||
"encoding/json" |
||||
"time" |
||||
|
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
// Parses a Cloudtrail Record and returns a logproto.Entry
|
||||
func parseCloudtrailRecord(record Record) (logproto.Entry, error) { |
||||
timestamp := time.Now() |
||||
if record.Error != nil { |
||||
return logproto.Entry{}, record.Error |
||||
} |
||||
document, err := json.Marshal(record.Content) |
||||
if err != nil { |
||||
return logproto.Entry{}, err |
||||
} |
||||
if val, ok := record.Content["eventTime"]; ok { |
||||
time, err := time.Parse(time.RFC3339, val.(string)) |
||||
if err != nil { |
||||
return logproto.Entry{}, err |
||||
} else { |
||||
timestamp = time |
||||
} |
||||
} |
||||
return logproto.Entry{ |
||||
Line: string(document), |
||||
Timestamp: timestamp, |
||||
}, nil |
||||
} |
||||
@ -0,0 +1,31 @@ |
||||
package main |
||||
|
||||
import ( |
||||
"compress/gzip" |
||||
"os" |
||||
"testing" |
||||
) |
||||
|
||||
func TestParseJson(t *testing.T) { |
||||
records := make(chan Record) |
||||
jsonStream := NewJSONStream(records) |
||||
file, err := os.Open("../testdata/cloudtrail-log-file.json.gz") |
||||
if err != nil { |
||||
t.Error(err) |
||||
} |
||||
gzipReader,err := gzip.NewReader(file) |
||||
if err != nil { |
||||
t.Error(err) |
||||
} |
||||
go jsonStream.Start(gzipReader, 3) |
||||
|
||||
for record := range jsonStream.records { |
||||
if record.Error != nil { |
||||
t.Error(record.Error) |
||||
} |
||||
_, err := parseCloudtrailRecord(record) |
||||
if err != nil { |
||||
t.Error(err) |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,54 @@ |
||||
package main |
||||
|
||||
import ( |
||||
"encoding/json" |
||||
"fmt" |
||||
"io" |
||||
) |
||||
|
||||
// Stream helps transmit each recordss withing a channel.
|
||||
type Stream struct { |
||||
records chan Record |
||||
} |
||||
|
||||
// Represents a Record in the Json Stream. If there is not error, error is nil
|
||||
type Record struct { |
||||
Error error |
||||
Content map[string]any |
||||
} |
||||
|
||||
// Creates a new Stream of records
|
||||
func NewJSONStream(recordChan chan Record) Stream { |
||||
return Stream{ |
||||
records: recordChan, |
||||
} |
||||
} |
||||
|
||||
// Streams the JSON file starting from the target token, record by record.
|
||||
func (s Stream) Start(r io.ReadCloser, tokenCountToTarget int) { |
||||
defer r.Close() |
||||
defer close(s.records) |
||||
var decoder *json.Decoder |
||||
decoder = json.NewDecoder(r) |
||||
|
||||
// Skip the provided count of JSON tokens to get the the target array, ex: "{" "Record"
|
||||
for i := 0; i < tokenCountToTarget; i++ { |
||||
_, err := decoder.Token() |
||||
if err != nil { |
||||
s.records <- Record{Error: fmt.Errorf("failed decoding beginning token: %w", err)} |
||||
return |
||||
} |
||||
} |
||||
|
||||
// Read the JSON token content
|
||||
i := 1 |
||||
for decoder.More() { |
||||
var content map[string]any |
||||
if err := decoder.Decode(&content); err != nil { |
||||
s.records <- Record{Error: fmt.Errorf("failed decoding record %d: %w", i, err)} |
||||
return |
||||
} |
||||
s.records <- Record{Content: content} |
||||
i++ |
||||
} |
||||
} |
||||
Binary file not shown.
Loading…
Reference in new issue