feat: scanning broker logs to discover information on clients #35

glenn-smith-0 · 2025-08-11T16:24:37Z

with s3 broker logging enabled
AND
kafka.server.KafkaApis=TRACE enabled on each broker
AND
policy on user

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::<BUCKET>",
                "arn:aws:s3:::<BUCKET>/*"
            ]
        }
    ]
}

given a folder in s3 for particular hour range

run scan

kcp scan broker-logs --region eu-west-1 --s3-uri s3://glenns-kcp-testing-logs/AWSLogs/635910096382/KafkaBrokerLogs/eu-west-1/kcp-pub-cluster-90a919bc-5967-4805-8a47-09dad9019d9b-5/2025-08-13-14/

will result in scan of all unique clients it can find in logs, uniqueness based on
composite key of

clientID + topic + role + auth + principal

and sorted by timestamp

for example IAM

my-client + orders + producer + IAM + arn:aws:sts::635910096382:assumed-role/AWSReservedSSO_nonprod-administrator_b3955bd58a347b7b/[email protected]

or SASL_SCRAM

my-client + orders + producer + SASL_SCRAM + User:kafka-user-2

or TLS

my-client + orders + producer + TLS + User:CN=kcp_tls_testing

or UNAUTHENTICATED

my-client + orders + producer + TLS + User:ANONYMOUS

ie

Client ID,Role,Topic,Auth,Principal,Timestamp
sarama,Producer,test-topic-1,IAM,arn:aws:sts::635910096382:assumed-role/kcp-testing-private-ec2-msk-role/i-0cc053268fead9442,2025-08-18 10:09:08
sarama,Consumer,test-topic-1,IAM,arn:aws:sts::635910096382:assumed-role/kcp-testing-private-ec2-msk-role/i-0cc053268fead9442,2025-08-18 10:09:10
sarama,Consumer,test-topic-1,SASL_SCRAM,User:kafka-user-2,2025-08-18 10:15:16
sarama,Producer,test-topic-1,SASL_SCRAM,User:kafka-user-2,2025-08-18 10:15:18
sarama,Consumer,test-topic-1,UNAUTHENTICATED,User:ANONYMOUS,2025-08-18 10:18:22
sarama,Producer,test-topic-1,UNAUTHENTICATED,User:ANONYMOUS,2025-08-18 10:18:24

Copilot

Pull Request Overview

This PR introduces a new feature for scanning Kafka broker logs stored in S3 to discover information about clients. The implementation adds the ability to parse log files, extract client activity metadata, and generate CSV reports of client connections.

Key changes:

Added S3 service integration for downloading and decompressing broker log files
Implemented a Kafka API trace line parser to extract client metadata from log entries
Created a broker logs scanner that processes multiple log files and generates CSV output

Reviewed Changes

Copilot reviewed 8 out of 9 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
`internal/services/s3/s3_service.go`	S3 service for downloading and decompressing log files
`internal/generators/scan/broker_logs/kafka_trace_line_parser.go`	Parser for extracting client metadata from Kafka API trace logs
`internal/generators/scan/broker_logs/kafka_trace_line_parser_test.go`	Comprehensive test coverage for the log parser
`internal/generators/scan/broker_logs/broker_logs_scanner.go`	Main scanner that orchestrates log processing and CSV generation
`internal/cli/scan/broker_logs/scan_broker_logs.go`	CLI command implementation for broker logs scanning
`internal/client/s3.go`	S3 client factory function
`internal/cli/scan/scan.go`	Added broker logs command to scan CLI
`go.mod`	Added S3 SDK dependency

internal/generators/scan/broker_logs/broker_logs_scanner.go

internal/generators/scan/broker_logs/kafka_trace_line_parser.go

adrian-januzi · 2025-08-12T08:55:52Z

Not to be that guy but we should add an entry in the README.md for the command.

Copilot

Pull Request Overview

This PR adds a new broker-logs subcommand to the kcp scan command that analyzes Kafka broker logs from S3 to discover client activity. The feature scans compressed log files to identify unique clients based on their composite key (clientID + topic + role + auth + principal) and generates a CSV report sorted by timestamp.

Key changes:

New broker logs scanning functionality with S3 integration
Kafka trace log parsing for PRODUCE and FETCH operations
Support for multiple authentication types (IAM, SASL_SCRAM, TLS, UNAUTHENTICATED)

Reviewed Changes

Copilot reviewed 10 out of 11 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
internal/services/s3/s3_service.go	S3 service for downloading and decompressing log files
internal/generators/scan/broker_logs/kafka_trace_line_parser.go	Parser for Kafka API trace log lines with regex patterns
internal/generators/scan/broker_logs/kafka_trace_line_parser_test.go	Comprehensive test suite for log parsing functionality
internal/generators/scan/broker_logs/broker_logs_scanner.go	Main scanner that coordinates log processing and CSV generation
internal/cli/scan/broker_logs/scan_broker_logs.go	CLI command implementation
internal/client/s3.go	S3 client factory
internal/cli/scan/scan.go	Registration of new broker-logs subcommand
go.mod	Addition of S3 SDK dependency
README.md	Documentation for the new broker-logs command
internal/generators/init/assets/README.md	Template documentation for the broker-logs command

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

internal/generators/scan/broker_logs/kafka_trace_line_parser.go

Copilot · 2025-08-18T16:32:24Z

internal/generators/scan/broker_logs/broker_logs_scanner.go

+	requestMetadataByCompositeKey := bs.handleLogFiles(ctx, bucket, logFiles)
+
+	if err := bs.generateCSV(requestMetadataByCompositeKey); err != nil {
+		slog.Error("failed to write CSV file", "error", err)


The error message on line 82 'failed to write CSV file' is generic and doesn't provide enough context about what specifically failed during CSV generation. Consider making it more specific, such as 'failed to generate CSV output file'.

Suggested change

slog.Error("failed to write CSV file", "error", err)

slog.Error("failed to generate CSV output file", "error", err)

Copilot · 2025-08-18T16:32:24Z

internal/generators/scan/broker_logs/broker_logs_scanner.go

+}
+
+func (bs *BrokerLogsScanner) generateCSV(requestMetadataByCompositeKey map[string]*RequestMetadata) error {
+	fileName := "broker_logs_scan_results.csv"


The CSV filename is hardcoded as 'broker_logs_scan_results.csv' on line 163. This could cause issues if multiple scans are run in the same directory. Consider making the filename configurable or include a timestamp to make it unique.

Suggested change

fileName := "broker_logs_scan_results.csv"

fileName := fmt.Sprintf("broker_logs_scan_results_%s.csv", time.Now().Format("20060102_150405"))

Copilot · 2025-08-18T16:32:25Z

internal/services/s3/s3_service.go

+}
+
+// DownloadAndDecompressLogFile downloads and decompresses a log file from S3
+func (s *S3Service) DownloadAndDecompressLogFile(ctx context.Context, bucket, key string) ([]byte, error) {


The function loads entire log files into memory as byte arrays. For large log files, this could cause memory issues. Consider using streaming or chunked processing for better memory efficiency, especially when processing multiple large files.

Suggested change

func (s *S3Service) DownloadAndDecompressLogFile(ctx context.Context, bucket, key string) ([]byte, error) {

// DownloadAndDecompressLogFile downloads and decompresses a log file from S3.

// It returns an io.ReadCloser for streaming the decompressed content.

func (s *S3Service) DownloadAndDecompressLogFile(ctx context.Context, bucket, key string) (io.ReadCloser, error) {

Copilot · 2025-08-18T16:32:25Z

internal/generators/scan/broker_logs/kafka_trace_line_parser.go

+		return nil, ErrorUnableToParseKafkaApiLine
+	}
+
+	timestamp, err := time.Parse("2006-01-02 15:04:05,000", timestampMatches[1])


The timestamp parsing uses a hardcoded UTC timezone assumption. The parsed time should explicitly be set to UTC timezone to ensure consistency, as time.Parse returns a time in UTC when no timezone is specified in the input, but this should be made explicit.

Suggested change

timestamp, err := time.Parse("2006-01-02 15:04:05,000", timestampMatches[1])

timestamp, err := time.ParseInLocation("2006-01-02 15:04:05,000", timestampMatches[1], time.UTC)

adrian-januzi

LGTM

Though I think we should eventually look into parsing the logs in parallel

glenn-smith-0 · 2025-08-19T14:54:41Z

LGTM

Though I think we should eventually look into parsing the logs in parallel

I had this originally but then it was becoming too complicated, but in reality what i have now is a much stripped back version of what i originally had. So may not be too hard to bring it back.

glenn-smith-0 added 13 commits July 31, 2025 15:57

feat: broker logs scanning

77175be

adding s3uri flag

ed2f812

listing files in folder

755236c

wip

b78fa9a

attempt to parse info from relevant logs

7c6c894

wip

8067819

wip

8791d06

wip

5ce40ab

working example;

1e1df8b

Merge remote-tracking branch 'origin/main' into feat-client-discovery

f8e04c3

complete?

2930385

remove unused const

9f70422

tidy

2f198c9

Copilot AI review requested due to automatic review settings August 11, 2025 16:24

glenn-smith-0 requested a review from a team as a code owner August 11, 2025 16:24

remove unused code

0df6177

This comment was marked as outdated.

Sign in to view

glenn-smith-0 requested a review from Copilot August 11, 2025 16:27

Copilot AI reviewed Aug 11, 2025

View reviewed changes

internal/generators/scan/broker_logs/broker_logs_scanner.go Outdated Show resolved Hide resolved

internal/generators/scan/broker_logs/kafka_trace_line_parser.go Outdated Show resolved Hide resolved

internal/generators/scan/broker_logs/kafka_trace_line_parser.go Outdated Show resolved Hide resolved

glenn-smith-0 added 7 commits August 12, 2025 10:21

Merge remote-tracking branch 'origin/main' into feat-client-discovery

822336c

adding in tls matching

f0a96da

no longer capture IP

2e71da0

Merge remote-tracking branch 'origin/main' into feat-client-discovery

8d46e6e

remove debug info

c78a000

update readmes

f278713

readme;

1a0f3ff

glenn-smith-0 requested a review from Copilot August 18, 2025 14:37

This comment was marked as outdated.

Sign in to view

glenn-smith-0 requested a review from Copilot August 18, 2025 16:30

Copilot AI reviewed Aug 18, 2025

View reviewed changes

adrian-januzi approved these changes Aug 19, 2025

View reviewed changes

glenn-smith-0 merged commit 3b0a4db into main Aug 19, 2025
2 checks passed

glenn-smith-0 deleted the feat-client-discovery branch August 19, 2025 14:54

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat: scanning broker logs to discover information on clients #35

feat: scanning broker logs to discover information on clients #35

Uh oh!

glenn-smith-0 commented Aug 11, 2025 •

edited

Loading

Uh oh!

This comment was marked as outdated.

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

adrian-januzi commented Aug 12, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI Aug 18, 2025

Uh oh!

Copilot AI Aug 18, 2025

Uh oh!

Copilot AI Aug 18, 2025

Uh oh!

Copilot AI Aug 18, 2025

Uh oh!

adrian-januzi left a comment

Uh oh!

glenn-smith-0 commented Aug 19, 2025

Uh oh!

Uh oh!

Uh oh!

	slog.Error("failed to write CSV file", "error", err)
	slog.Error("failed to generate CSV output file", "error", err)

	fileName := "broker_logs_scan_results.csv"
	fileName := fmt.Sprintf("broker_logs_scan_results_%s.csv", time.Now().Format("20060102_150405"))

-func (s *S3Service) DownloadAndDecompressLogFile(ctx context.Context, bucket, key string) ([]byte, error) {
+// DownloadAndDecompressLogFile downloads and decompresses a log file from S3.
+// It returns an io.ReadCloser for streaming the decompressed content.
+func (s *S3Service) DownloadAndDecompressLogFile(ctx context.Context, bucket, key string) (io.ReadCloser, error) {

	timestamp, err := time.Parse("2006-01-02 15:04:05,000", timestampMatches[1])
	timestamp, err := time.ParseInLocation("2006-01-02 15:04:05,000", timestampMatches[1], time.UTC)

feat: scanning broker logs to discover information on clients #35

feat: scanning broker logs to discover information on clients #35

Uh oh!

Conversation

glenn-smith-0 commented Aug 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Key changes:

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

adrian-januzi commented Aug 12, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Copilot AI Aug 18, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Aug 18, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Aug 18, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Aug 18, 2025

Choose a reason for hiding this comment

Uh oh!

adrian-januzi left a comment

Choose a reason for hiding this comment

Uh oh!

glenn-smith-0 commented Aug 19, 2025

Uh oh!

Uh oh!

Uh oh!

glenn-smith-0 commented Aug 11, 2025 •

edited

Loading