Switch from iter.Seq to iter.Seq2 without error in s3Record struct

Author: ndakkoune

aws/logs_monitoring_go/internal/handling/cloudtrail.go

@@ -30,45 +30,45 @@ var ec2InstanceRegexp = sync.OnceValue(func() *regexp.Regexp {
 	return regexp.MustCompile(`^arn:aws:sts::.*?:assumed-role/(?P<role>.*?)/(?P<host>i-([0-9a-f]{8}|[0-9a-f]{17}))$`)
 })
 
-func decodeCloudTrail(r io.Reader) iter.Seq[s3Record] {
-	return func(yield func(s3Record) bool) {
+func decodeCloudTrail(r io.Reader) iter.Seq2[s3Record, error] {
+	return func(yield func(s3Record, error) bool) {
 		gz, err := gzip.NewReader(r)
 		if err != nil {
-			yield(s3Record{Err: fmt.Errorf("decode cloudtrail gzip: %w", err)})
+			yield(s3Record{}, fmt.Errorf("decode cloudtrail gzip: %w", err))
 			return
 		}
 		defer gz.Close() //nolint:errcheck
 
 		dec := json.NewDecoder(gz)
 
 		if t, err := dec.Token(); err != nil || t != json.Delim('{') {
-			yield(s3Record{Err: errors.New("decode cloudtrail: expected '{' at start of JSON")})
+			yield(s3Record{}, errors.New("decode cloudtrail: expected '{' at start of JSON"))
 			return
 		}
 		if t, err := dec.Token(); err != nil || t != "Records" {
-			yield(s3Record{Err: errors.New("decode cloudtrail: expected 'Records' key")})
+			yield(s3Record{}, errors.New("decode cloudtrail: expected 'Records' key"))
 			return
 		}
 		if t, err := dec.Token(); err != nil || t != json.Delim('[') {
-			yield(s3Record{Err: errors.New("decode cloudtrail: expected '[' at start of Records array")})
+			yield(s3Record{}, errors.New("decode cloudtrail: expected '[' at start of Records array"))
 			return
 		}
 
 		for dec.More() {
 			var record map[string]any
 			if err := dec.Decode(&record); err != nil {
-				yield(s3Record{Err: fmt.Errorf("decode cloudtrail record: %w", err)})
+				yield(s3Record{}, fmt.Errorf("decode cloudtrail record: %w", err))
 				return
 			}
 
 			msg, err := json.Marshal(record)
 			if err != nil {
-				yield(s3Record{Err: fmt.Errorf("marshal cloudtrail record: %w", err)})
+				yield(s3Record{}, fmt.Errorf("marshal cloudtrail record: %w", err))
 				return
 			}
 
 			host := cloudtrailHost(record)
-			if !yield(s3Record{Message: string(msg), Host: host}) {
+			if !yield(s3Record{Message: string(msg), Host: host}, nil) {
 				return
 			}
 		}
@@ -98,7 +98,7 @@ func cloudtrailHost(record map[string]any) string {
 	re := ec2InstanceRegexp()
 	matches := re.FindStringSubmatch(arn)
 	if matches == nil {
-		slog.Debug(arn + " arn did not match, cloudtrail host extraction skipped")
+		slog.Debug(arn + " arn did not match an EC2 host instance, cloudtrail host extraction skipped")
 		return ""
 	}
 

aws/logs_monitoring_go/internal/handling/cloudtrail_test.go

@@ -198,10 +198,10 @@ func TestDecodeCloudTrail(t *testing.T) {
 			t.Parallel()
 
 			var got []s3Record
-			for rec := range decodeCloudTrail(bytes.NewReader(tc.input)) {
-				if rec.Err != nil {
+			for rec, err := range decodeCloudTrail(bytes.NewReader(tc.input)) {
+				if err != nil {
 					if !tc.wantErr {
-						t.Fatalf("unexpected error: %v", rec.Err)
+						t.Fatalf("unexpected error: %v", err)
 					}
 					return
 				}

aws/logs_monitoring_go/internal/handling/s3.go

@@ -31,7 +31,6 @@ const (
 type s3Record struct {
 	Message string
 	Host    string
-	Err     error
 }
 
 type S3Handler struct {
@@ -82,16 +81,16 @@ func (h S3Handler) processRecord(ctx context.Context, client S3APIClient, out ch
 		}
 	}()
 
-	var records iter.Seq[s3Record]
+	var records iter.Seq2[s3Record, error]
 	if cloudTrailRegex().MatchString(key) {
 		records = decodeCloudTrail(body)
 	} else {
 		records = scan(body, h.cfg.S3MultilineLogRegex)
 	}
 
-	for rec := range records {
-		if rec.Err != nil {
-			return rec.Err
+	for rec, err := range records {
+		if err != nil {
+			return err
 		}
 		entry := h.newS3LogEntry(eventRecord, rec.Message, lambdaOrigin)
 		if h.cfg.Filter.ShouldExclude(entry.Message) {

aws/logs_monitoring_go/internal/handling/scanner.go

@@ -80,17 +80,17 @@ func (s *Scanner) splitOnLines(data []byte, atEOF bool) (advance int, token []by
 	return 0, nil, nil
 }
 
-func scan(r io.Reader, re *regexp.Regexp) iter.Seq[s3Record] {
-	return func(yield func(s3Record) bool) {
+func scan(r io.Reader, re *regexp.Regexp) iter.Seq2[s3Record, error] {
+	return func(yield func(s3Record, error) bool) {
 		scanner := NewScanner(r, re)
 		for scanner.Scan() {
 			message := strings.ToValidUTF8(scanner.Text(), "")
-			if !yield(s3Record{Message: message}) {
+			if !yield(s3Record{Message: message}, nil) {
 				return
 			}
 		}
 		if err := scanner.Err(); err != nil {
-			yield(s3Record{Err: err})
+			yield(s3Record{}, err)
 		}
 	}
 }