[filebeat][awss3] - Added support for parquet decoding and decoder config

CHANGELOG.next.asciidoc

@@ -287,6 +287,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Add nginx ingress_controller parsing if one of upstreams fails to return response {pull}34787[34787]
 - Allow neflow v9 and ipfix templates to be shared between source addresses. {pull}35036[35036]
 - Add support for collecting IPv6 metrics. {pull}35123[35123]
+- Added support for apache parquet files to aws-s3 input. {issue}34662[34662] {pull}35578[35578]
 - Add oracle authentication messages parsing {pull}35127[35127]
 - Add sanitization capabilities to azure-eventhub input {pull}34874[34874]
 - Add support for CRC validation in Filebeat's HTTP endpoint input. {pull}35204[35204]

x-pack/filebeat/docs/inputs/input-aws-s3.asciidoc

@@ -115,6 +115,43 @@ it was uploaded. For example: `application/json`.
 The file encoding to use for reading data that contains international
 characters. This only applies to non-JSON logs. See <<_encoding_3>>.
 
+[id="input-{type}-decoding"]
+[float]
+==== `decoding`
+
+The file decoding option is used to specify a codec that will be used to
+decode the file contents. This can apply to any file stream data.  
+An example config is shown below:
+
+[source,yaml]
+----
+  decoding.codec: "parquet"
+----
+
+Currently supported codecs are given below:-
+
+    1. <<attrib-decoding-parquet,Parquet>>: This codec decodes parquet compressed data streams.
+
+[id="attrib-decoding-parquet"]
+[float]
+==== `the parquet codec`
+The `parquet` codec is used to decode parquet compressed data streams.
+Only specifying the codec name will use the default codec options. The parquet codec supports
+two sub attributes which can make parquet decoding more efficient. The `batch_size` attribute and 
+the `process_parallel` attribute. The `batch_size` attribute can be used to specify the number of
+records to read from the parquet stream at a time. By default the `batch size` is set to `1` and 
+`process_parallel` is set to `false`. If the `process_parallel` attribute is set to `true` then functions 
+which read multiple columns will read those columns in parallel from the parquet stream with a 
+number of readers equal to the number of columns. Setting `process_parallel` to `true` will greatly
+increase the rate of processing at the cost of increased memory usage. Having a larger `batch_size`
+also helps to increase the rate of processing. An example config is shown below:
+
+[source,yaml]
+----
+  decoding.codec: "parquet"
+  decoding.codec.parquet.process_parallel = true
+  decoding.codec.parquet.batch_size = 1000
+----
 
 [float]
 ==== `expand_event_list_from_field`

x-pack/filebeat/input/awss3/config.go

@@ -162,6 +162,15 @@ type readerConfig struct {
 	LineTerminator           readfile.LineTerminator `config:"line_terminator"`
 	MaxBytes                 cfgtype.ByteSize        `config:"max_bytes"`
 	Parsers                  parser.Config           `config:",inline"`
+	Decoding                 decoderConfig           `config:"decoding"`
+}
+type decoderConfig struct {
+	Codec   string             `config:"codec"`
+	Parquet parquetCodecConfig `config:"codec.parquet"`
+}
+type parquetCodecConfig struct {
+	ProcessParallel bool `config:"process_parallel"`
+	BatchSize       int  `config:"batch_size" default:"1"`
 }
 
 func (rc *readerConfig) Validate() error {

x-pack/filebeat/input/awss3/decoding.go

@@ -0,0 +1,85 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package awss3
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/elastic/beats/v7/x-pack/libbeat/reader/parquet"
+)
+
+const (
+	codecParquet = "parquet"
+)
+
+// decoder is an interface for decoding data from an io reader.
+type decoder interface {
+	// decode reads and decodes data from an io reader based on the codec type.
+	decode() error
+}
+
+// newDecoder creates a new decoder based on the codec type.
+// It returns a decoder type and an error if the codec type is not supported.
+// If the reader config codec option is not set, it returns a nil decoder and nil error.
+func newDecoder(p *s3ObjectProcessor, r io.Reader) (decoder, error) {
+	switch p.readerConfig.Decoding.Codec {
+	case "":
+		return nil, nil
+	case codecParquet:
+		return newParquetDecoder(p, r)
+	default:
+		return nil, fmt.Errorf("unsupported codec type: %s", p.readerConfig.Decoding.Codec)
+	}
+}
+
+// parquetDecoder is a decoder for parquet data.
+type parquetDecoder struct {
+	p      *s3ObjectProcessor
+	reader *parquet.BufferedReader
+}
+
+// newParquetDecoder creates a new parquet decoder. It uses the libbeat parquet reader under the hood.
+// It returns an error if the parquet reader cannot be created.
+func newParquetDecoder(p *s3ObjectProcessor, r io.Reader) (decoder, error) {
+	reader, err := parquet.NewBufferedReader(r, &parquet.Config{
+		ProcessParallel: p.readerConfig.Decoding.Parquet.ProcessParallel,
+		BatchSize:       p.readerConfig.Decoding.Parquet.BatchSize,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create parquet decoder: %w", err)
+	}
+	return &parquetDecoder{
+		p:      p,
+		reader: reader,
+	}, nil
+}
+
+// decode reads and decodes a parquet data stream. After reading the parquet data it decodes
+// the output to JSON and sends the decoded data to the readJSONSlice method to
+// process further as individual JSON objects.
+func (pd *parquetDecoder) decode() error {
+	// releases the reader resources once processing is done
+	defer pd.reader.Close()
+
+	for pd.reader.Next() {
+		data, err := pd.reader.Record()
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return nil
+			}
+			return fmt.Errorf("failed to read records from parquet record reader: %w", err)
+		}
+
+		err = pd.p.readJSONSlice(bytes.NewReader(data))
+		if err != nil {
+			return fmt.Errorf("failed to read JSON data from arrow record: %w", err)
+		}
+	}
+
+	return nil
+}

x-pack/filebeat/input/awss3/decoding_test.go

@@ -0,0 +1,124 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package awss3
+
+import (
+	"bufio"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// all test files are read from the "testdata" directory
+const testDataPath = "testdata"
+
+func TestParquetDecoding(t *testing.T) {
+	testCases := []struct {
+		name          string
+		file          string
+		contentType   string
+		numEvents     int
+		assertAgainst string
+		config        *readerConfig
+	}{
+		{
+			name:      "test decoding of a parquet file and compare the number of events with batch size 1",
+			file:      "vpc-flow.gz.parquet",
+			numEvents: 1304,
+			config: &readerConfig{
+				Decoding: decoderConfig{
+					Codec: "parquet",
+					Parquet: parquetCodecConfig{
+						ProcessParallel: true,
+						BatchSize:       1,
+					},
+				},
+			},
+		},
+		{
+			name:      "test decoding of a parquet file and compare the number of events with batch size 100",
+			file:      "vpc-flow.gz.parquet",
+			numEvents: 1304,
+			config: &readerConfig{
+				Decoding: decoderConfig{
+					Codec: "parquet",
+					Parquet: parquetCodecConfig{
+						ProcessParallel: true,
+						BatchSize:       100,
+					},
+				},
+			},
+		},
+		{
+			name:      "test decoding of a parquet file and compare the number of events with default parquet config",
+			file:      "vpc-flow.gz.parquet",
+			numEvents: 1304,
+			config: &readerConfig{
+				Decoding: decoderConfig{
+					Codec: "parquet",
+				},
+			},
+		},
+		{
+			name:          "test decoding of a parquet file and compare the number of events along with the content",
+			file:          "cloudtrail.parquet",
+			numEvents:     1,
+			assertAgainst: "cloudtrail.ndjson",
+			config: &readerConfig{
+				Decoding: decoderConfig{
+					Codec: "parquet",
+					Parquet: parquetCodecConfig{
+						ProcessParallel: true,
+						BatchSize:       1,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			file := filepath.Join(testDataPath, tc.file)
+			sel := fileSelectorConfig{ReaderConfig: *tc.config}
+			if tc.contentType == "" {
+				tc.contentType = "application/octet-stream"
+			}
+			// uses the s3_objects test method to perform the test
+			events := testProcessS3Object(t, file, tc.contentType, tc.numEvents, sel)
+			// if assertAgainst is not empty, then compare the events with the target file
+			// there is a chance for this comparison to become flaky if number of events > 1 as
+			// the order of events are not guaranteed by beats
+			if tc.assertAgainst != "" {
+				targetFile, err := os.Open(filepath.Join(testDataPath, tc.assertAgainst))
+				assert.NoError(t, err)
+				defer targetFile.Close()
+				targetData := readJSONFromFile(t, targetFile)
+				assert.Equal(t, len(targetData), len(events))
+
+				for i, event := range events {
+					msg, err := event.Fields.GetValue("message")
+					assert.NoError(t, err)
+					assert.JSONEq(t, targetData[i], msg.(string))
+				}
+			}
+		})
+	}
+}
+
+// readJSONFromFile reads the json file and returns the data as a slice of strings
+func readJSONFromFile(t *testing.T, file *os.File) []string {
+	var data []string
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		data = append(data, scanner.Text())
+	}
+	if err := scanner.Err(); err != nil {
+		t.Fatalf("failed to read ndjson file: %v", err)
+	}
+
+	return data
+}

x-pack/filebeat/input/awss3/s3_objects.go

@@ -15,7 +15,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"reflect"
 	"strings"
 	"time"
@@ -152,14 +151,23 @@ func (p *s3ObjectProcessor) ProcessS3Object() error {
 		contentType = p.readerConfig.ContentType
 	}
 
-	// Process object content stream.
-	switch {
-	case strings.HasPrefix(contentType, contentTypeJSON) || strings.HasPrefix(contentType, contentTypeNDJSON):
-		err = p.readJSON(reader)
-	default:
-		err = p.readFile(reader)
+	// try to create a decoder from the using the codec config
+	decoder, err := newDecoder(p, reader)
+	if err != nil {
+		return err
+	}
+	if decoder != nil {
+		err = decoder.decode()
+	} else {
+		// This is the legacy path. It will be removed in future and clubbed together with the decoder.
+		// Process object content stream.
+		switch {
+		case strings.HasPrefix(contentType, contentTypeJSON) || strings.HasPrefix(contentType, contentTypeNDJSON):
+			err = p.readJSON(reader)
+		default:
+			err = p.readFile(reader)
+		}
 	}
-
 	if err != nil {
 		return fmt.Errorf("failed reading s3 object (elapsed_time_ns=%d): %w",
 			time.Since(start).Nanoseconds(), err)
@@ -233,6 +241,37 @@ func (p *s3ObjectProcessor) readJSON(r io.Reader) error {
 	return nil
 }
 
+// readJSONSlice uses a slice of json.RawMessage to process JSON slice data
+// as individual JSON objects.
+func (p *s3ObjectProcessor) readJSONSlice(r io.Reader) error {
+	dec := json.NewDecoder(r)
+	dec.UseNumber()
+
+	for dec.More() && p.ctx.Err() == nil {
+		offset := dec.InputOffset()
+
+		var items []json.RawMessage
+		if err := dec.Decode(&items); err != nil {
+			return fmt.Errorf("failed to decode json: %w", err)
+		}
+
+		for _, item := range items {
+			if p.readerConfig.ExpandEventListFromField != "" {
+				if err := p.splitEventList(p.readerConfig.ExpandEventListFromField, item, offset, p.s3ObjHash); err != nil {
+					return err
+				}
+				continue
+			}
+
+			data, _ := item.MarshalJSON()
+			evt := p.createEvent(string(data), offset)
+			p.publish(p.acker, &evt)
+		}
+	}
+
+	return nil
+}
+
 func (p *s3ObjectProcessor) splitEventList(key string, raw json.RawMessage, offset int64, objHash string) error {
 	var jsonObject map[string]json.RawMessage
 	if err := json.Unmarshal(raw, &jsonObject); err != nil {
@@ -285,7 +324,7 @@ func (p *s3ObjectProcessor) readFile(r io.Reader) error {
 	}
 
 	var reader reader.Reader
-	reader, err = readfile.NewEncodeReader(ioutil.NopCloser(r), readfile.Config{
+	reader, err = readfile.NewEncodeReader(io.NopCloser(r), readfile.Config{
 		Codec:        enc,
 		BufferSize:   int(p.readerConfig.BufferSize),
 		Terminator:   p.readerConfig.LineTerminator,

x-pack/filebeat/input/awss3/testdata/cloudtrail.ndjson

@@ -0,0 +1 @@
+{"activity_id":1,"activity_name":"Create","actor":{"idp":{"name":null},"invoked_by":null,"session":{"created_time":null,"issuer":null,"mfa":null},"user":{"account_uid":"123456789012","credential_uid":"AKIAIOSFODNN7EXAMPLE","name":"Alice","type":"IAMUser","uid":"123456789012","uuid":"arn:aws:iam::123456789012:user/Alice"}},"api":{"operation":"CreateLoadBalancer","request":{"uid":"b9960276-b9b2-11e3-8a13-f1ef1EXAMPLE"},"response":{"error":null,"message":null},"service":{"name":"elasticloadbalancing.amazonaws.com"},"version":"2015-12-01"},"category_name":"Audit Activity","category_uid":3,"class_name":"API Activity","class_uid":3005,"cloud":{"provider":"AWS","region":"us-west-2"},"http_request":{"user_agent":"aws-cli/1.10.10 Python/2.7.9 Windows/7 botocore/1.4.1"},"metadata":{"product":{"feature":{"name":"Management, Data, and Insights"},"name":"CloudTrail","vendor_name":"AWS","version":"1.03"},"profiles":["cloud"],"uid":"6f4ab5bd-2daa-4d00-be14-d92efEXAMPLE","version":"1.0.0-rc.2"},"resources":null,"severity":"Informational","severity_id":1,"src_endpoint":{"domain":null,"ip":"198.51.100.1","uid":null},"status":"Success","status_id":1,"time":1459524708000,"type_name":"API Activity: Create","type_uid":300501,"unmapped":[{"key":"responseElements","value":"{\"loadBalancers\":[{\"type\":\"application\",\"loadBalancerName\":\"my-load-balancer\",\"vpcId\":\"vpc-3ac0fb5f\",\"securityGroups\":[\"sg-5943793c\"],\"state\":{\"code\":\"provisioning\"},\"availabilityZones\":[{\"subnetId\":\"subnet-8360a9e7\",\"zoneName\":\"us-west-2a\"},{\"subnetId\":\"subnet-b7d581c0\",\"zoneName\":\"us-west-2b\"}],\"dNSName\":\"my-load-balancer-1836718677.us-west-2.elb.amazonaws.com\",\"canonicalHostedZoneId\":\"Z2P70J7HTTTPLU\",\"createdTime\":\"Apr 11, 2016 5:23:50 PM\",\"loadBalancerArn\":\"arn:aws:elasticloadbalancing:us-west-2:123456789012:loadbalancer/app/my-load-balancer/ffcddace1759e1d0\",\"scheme\":\"internet-facing\"}]}"},{"key":"requestParameters","value":"{\"subnets\":[\"subnet-8360a9e7\",\"subnet-b7d581c0\"],\"securityGroups\":[\"sg-5943793c\"],\"name\":\"my-load-balancer\",\"scheme\":\"internet-facing\"}"},{"key":"recipientAccountId","value":"123456789012"},{"key":"eventType","value":"AwsApiCall"}]}