Add integration tests for out_s3

integ/integ.sh

@@ -78,6 +78,50 @@ test_firehose() {
 	fi
 }
 
+test_s3() {
+	# different S3 prefix for each test
+	export ARCHITECTURE=$(uname -m)
+	export S3_PREFIX_PUT_OBJECT="logs/${ARCHITECTURE}/putobject"
+	export S3_PREFIX_MULTIPART="logs/${ARCHITECTURE}/multipart"
+	# Tag is used in the s3 keys; each test run has a unique (random) tag
+	export TAG=$(head /dev/urandom | tr -dc A-Za-z0-9 | head -c 10)
+	# Generates log data which will be stored on the s3 bucket
+	docker-compose --file ./integ/test_s3/docker-compose.test.yml build
+	docker-compose --file ./integ/test_s3/docker-compose.test.yml up --abort-on-container-exit
+
+    # Giving a pause before running the validation test
+	sleep 20
+
+	# Creates a file as a flag for the validation failure
+	mkdir -p ./integ/out
+	touch ./integ/out/s3-test
+
+	export S3_ACTION="validate"
+	docker-compose --file ./integ/test_s3/docker-compose.validate-s3-multipart.yml build
+	docker-compose --file ./integ/test_s3/docker-compose.validate-s3-multipart.yml up --abort-on-container-exit
+
+	if [ -f ./integ/out/s3-test ]; then
+		# if the file still exists, test failed
+		echo "Test failed for S3."
+		exit 1
+	fi
+
+	# Creates a file as a flag for the validation failure
+	mkdir -p ./integ/out
+	touch ./integ/out/s3-test
+
+	export S3_ACTION="validate"
+	docker-compose --file ./integ/test_s3/docker-compose.validate-s3-putobject.yml build
+	docker-compose --file ./integ/test_s3/docker-compose.validate-s3-putobject.yml up --abort-on-container-exit
+
+	if [ -f ./integ/out/s3-test ]; then
+		# if the file still exists, test failed
+		echo "Test failed for S3."
+		exit 1
+	fi
+}
+
+
 clean_s3() {
 	validate_or_clean_s3 clean
 }
@@ -113,6 +157,7 @@ fi
 if [ "${1}" = "kinesis" ]; then
 	export S3_PREFIX="kinesis-test"
 	export TEST_FILE="kinesis-test"
+	export EXPECTED_EVENTS_LEN="1000"
 	source ./integ/resources/create_test_resources.sh
 	source ./integ/resources/setup_test_environment.sh
 
@@ -122,12 +167,23 @@ fi
 if [ "${1}" = "firehose" ]; then
 	export S3_PREFIX="firehose-test"
 	export TEST_FILE="firehose-test"
+	export EXPECTED_EVENTS_LEN="1000"
 	source ./integ/resources/create_test_resources.sh
 	source ./integ/resources/setup_test_environment.sh
 
 	clean_s3 && test_firehose
 fi
 
+if [ "${1}" = "s3" ]; then
+	export S3_PREFIX="logs"
+	export TEST_FILE="s3-test"
+	export EXPECTED_EVENTS_LEN="7717"
+	source ./integ/resources/create_test_resources.sh
+	source ./integ/resources/setup_test_environment.sh
+
+	clean_s3 && test_s3
+fi
+
 if [ "${1}" = "clean-s3" ]; then
 	source ./integ/resources/setup_test_environment.sh
 	clean_s3
@@ -155,10 +211,19 @@ if [ "${1}" = "cicd" ]; then
 	source ./integ/resources/setup_test_environment.sh
 	export S3_PREFIX="kinesis-test"
 	export TEST_FILE="kinesis-test"
+	export EXPECTED_EVENTS_LEN="1000"
 	clean_s3 && test_kinesis
 	export S3_PREFIX="firehose-test"
 	export TEST_FILE="firehose-test"
 	clean_s3 && test_firehose
+
+	export S3_PREFIX="logs"
+	export TEST_FILE="s3-test"
+	export EXPECTED_EVENTS_LEN="7717"
+	source ./integ/resources/create_test_resources.sh
+	source ./integ/resources/setup_test_environment.sh
+
+	clean_s3 && test_s3
 fi
 
 if [ "${1}" = "delete" ]; then

integ/out/expected-metric-name

@@ -0,0 +1 @@
+fluent-bit-integ-test-1264

integ/s3-logger/Dockerfile

@@ -0,0 +1,8 @@
+FROM amazonlinux
+
+RUN yum upgrade -y
+RUN yum install -y openssl
+
+COPY logscript.sh /
+
+CMD ["bash", "/logscript.sh"]

integ/s3-logger/logscript.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+
+# Writes 7717 unique log lines
+# intermixed with 1 KB lines of random data
+# these logs are written over the course of a little over 1 minute
+# because we want to test S3's local buffering
+# Why 7717? Its a prime number. And I like that.
+# Finally we sleep for 90s- ensuring the upload timeout hits for the last chunk
+# then exit
+
+for i in {0..7716}
+do
+    echo $i
+    openssl rand -base64 1000 | tr '\n' '-' && echo ""
+    sleep 0.0001
+done
+
+sleep 200
+
+exit 0

integ/s3/validate-and-clean.go

@@ -16,11 +16,12 @@ import (
 )
 
 const (
-	envAWSRegion = "AWS_REGION"
-	envS3Bucket  = "S3_BUCKET_NAME"
-	envS3Action  = "S3_ACTION"
-	envS3Prefix  = "S3_PREFIX"
-	envTestFile  = "TEST_FILE"
+	envAWSRegion       = "AWS_REGION"
+	envS3Bucket        = "S3_BUCKET_NAME"
+	envS3Action        = "S3_ACTION"
+	envS3Prefix        = "S3_PREFIX"
+	envTestFile        = "TEST_FILE"
+	envExpectedLogsLen = "EXPECTED_EVENTS_LEN"
 )
 
 type Message struct {
@@ -48,6 +49,15 @@ func main() {
 		exitErrorf("[TEST FAILURE] test verfication file name required. Set the value for environment variable- %s", envTestFile)
 	}
 
+	expectedEventsLen := os.Getenv(envExpectedLogsLen)
+	if expectedEventsLen == "" {
+		exitErrorf("[TEST FAILURE] number of expected log events required. Set the value for environment variable- %s", envExpectedLogsLen)
+	}
+	numEvents, convertionError := strconv.Atoi(expectedEventsLen)
+	if convertionError != nil {
+		exitErrorf("[TEST FAILURE] String to Int convertion Error for EXPECTED_EVENTS_LEN:", convertionError)
+	}
+
 	s3Client, err := getS3Client(region)
 	if err != nil {
 		exitErrorf("[TEST FAILURE] Unable to create new S3 client: %v", err)
@@ -57,7 +67,7 @@ func main() {
 	if s3Action == "validate" {
 		// Validate the data on the s3 bucket
 		getS3ObjectsResponse := getS3Objects(s3Client, bucket, prefix)
-		validate(s3Client, getS3ObjectsResponse, bucket, testFile)
+		validate(s3Client, getS3ObjectsResponse, bucket, testFile, numEvents)
 	} else {
 		// Clean the s3 bucket-- delete all objects
 		deleteS3Objects(s3Client, bucket, prefix)
@@ -97,8 +107,8 @@ func getS3Objects(s3Client *s3.S3, bucket string, prefix string) *s3.ListObjects
 // Validates the log messages. Our log producer is designed to send 1000 integers [0 - 999].
 // Both of the Kinesis Streams and Kinesis Firehose try to send each log maintaining the "at least once" policy.
 // To validate, we need to make sure all the valid numbers [0 - 999] are stored at least once.
-func validate(s3Client *s3.S3, response *s3.ListObjectsV2Output, bucket string, testFile string) {
-	logCounter := make([]int, 1000)
+func validate(s3Client *s3.S3, response *s3.ListObjectsV2Output, bucket string, testFile string, numEvents int) {
+	logCounter := make([]int, numEvents)
 	for index := range logCounter {
 		logCounter[index] = 1
 	}
@@ -121,6 +131,9 @@ func validate(s3Client *s3.S3, response *s3.ListObjectsV2Output, bucket string,
 			if d == "" {
 				continue
 			}
+			if len(d) > 500 {
+				continue
+			}
 
 			var message Message
 
@@ -134,8 +147,8 @@ func validate(s3Client *s3.S3, response *s3.ListObjectsV2Output, bucket string,
 				exitErrorf("[TEST FAILURE] String to Int convertion Error:", convertionError)
 			}
 
-			if number < 0 || number >= 1000 {
-				exitErrorf("[TEST FAILURE] Invalid number: %d found. Expected value in range (0 - 999)", number)
+			if number < 0 || number >= numEvents {
+				exitErrorf("[TEST FAILURE] Invalid number: %d found. Expected value in range (0 - %d)", number, numEvents)
 			}
 
 			logCounter[number] = 0

integ/test_firehose/docker-compose.validate-and-clean-s3.yml

@@ -12,5 +12,6 @@ services:
             - "S3_ACTION=${S3_ACTION}"
             - "S3_PREFIX=${S3_PREFIX}"
             - "TEST_FILE=${TEST_FILE}"
+            - "EXPECTED_EVENTS_LEN=${EXPECTED_EVENTS_LEN}"
         volumes:
             - ${PROJECT_ROOT}/integ/out:/out

integ/test_kinesis/docker-compose.validate-and-clean-s3.yml

@@ -12,5 +12,6 @@ services:
             - "S3_ACTION=${S3_ACTION}"
             - "S3_PREFIX=${S3_PREFIX}"
             - "TEST_FILE=${TEST_FILE}"
+            - "EXPECTED_EVENTS_LEN=${EXPECTED_EVENTS_LEN}"
         volumes:
             - ${PROJECT_ROOT}/integ/out:/out

integ/test_s3/docker-compose.test.yml

@@ -0,0 +1,35 @@
+version: "2"
+
+services:
+    fluent-bit:
+        image: amazon/aws-for-fluent-bit:latest
+        environment:
+            - "AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}"
+            - "AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}"
+            - "AWS_SESSION_TOKEN=${AWS_SESSION_TOKEN}"
+            - "S3_BUCKET_NAME=${S3_BUCKET_NAME}"
+            - "ARCHITECTURE=${ARCHITECTURE}"
+            - FLB_LOG_LEVEL=debug
+        volumes:
+            - /var/run/:/var/run
+            - ./:/fluent-bit/etc/
+    logger-multipart-test:
+        build: ${PROJECT_ROOT}/integ/s3-logger
+        depends_on:
+            - fluent-bit
+        logging:
+            driver: fluentd
+            options:
+                tag: "multipart-upload-test-${TAG}"
+                fluentd-address: unix:///var/run/fluent.sock
+                fluentd-async-connect: "true"
+    logger-put-object-test:
+        build: ${PROJECT_ROOT}/integ/s3-logger
+        depends_on:
+            - fluent-bit
+        logging:
+            driver: fluentd
+            options:
+                tag: "put-object-test-${TAG}"
+                fluentd-address: unix:///var/run/fluent.sock
+                fluentd-async-connect: "true"

integ/test_s3/docker-compose.validate-s3-multipart.yml

@@ -0,0 +1,17 @@
+version: "2"
+
+services:
+    validate-s3-multipart:
+        build: ${PROJECT_ROOT}/integ/s3
+        environment:
+            - "AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}"
+            - "AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}"
+            - "AWS_SESSION_TOKEN=${AWS_SESSION_TOKEN}"
+            - "AWS_REGION=${AWS_REGION}"
+            - "S3_BUCKET_NAME=${S3_BUCKET_NAME}"
+            - "S3_ACTION=${S3_ACTION}"
+            - "S3_PREFIX=${S3_PREFIX_MULTIPART}"
+            - "EXPECTED_EVENTS_LEN=${EXPECTED_EVENTS_LEN}"
+            - "TEST_FILE=${TEST_FILE}"
+        volumes:
+            - ${PROJECT_ROOT}/integ/out:/out

integ/test_s3/docker-compose.validate-s3-putobject.yml

@@ -0,0 +1,17 @@
+version: "2"
+
+services:
+    validate-s3-put-object:
+        build: ${PROJECT_ROOT}/integ/s3
+        environment:
+            - "AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}"
+            - "AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}"
+            - "AWS_SESSION_TOKEN=${AWS_SESSION_TOKEN}"
+            - "AWS_REGION=${AWS_REGION}"
+            - "S3_BUCKET_NAME=${S3_BUCKET_NAME}"
+            - "S3_ACTION=${S3_ACTION}"
+            - "S3_PREFIX=${S3_PREFIX_PUT_OBJECT}"
+            - "EXPECTED_EVENTS_LEN=${EXPECTED_EVENTS_LEN}"
+            - "TEST_FILE=${TEST_FILE}"
+        volumes:
+            - ${PROJECT_ROOT}/integ/out:/out

integ/test_s3/fluent-bit.conf

@@ -0,0 +1,29 @@
+[SERVICE]
+    Log_Level debug
+
+[INPUT]
+    Name forward
+    unix_path /var/run/fluent.sock
+
+[OUTPUT]
+    Name s3
+    Match multipart*
+    bucket ${S3_BUCKET_NAME}
+    region us-west-2
+    store_dir /fluent-bit/buffer
+    total_file_size 100M
+    s3_key_format /logs/${ARCHITECTURE}/multipart/$TAG[1]/logs/$TAG/%Y/%m/%d/%H/%M/%S
+    s3_key_format_tag_delimiters .-
+    upload_timeout 2m
+
+[OUTPUT]
+    Name s3
+    Match put-object*
+    bucket ${S3_BUCKET_NAME}
+    region us-west-2
+    store_dir /fluent-bit/buffer
+    total_file_size 1M
+    s3_key_format /logs/${ARCHITECTURE}/put$TAG[1]/logs/$TAG/%Y/%m/%d/%H/%M/%S
+    s3_key_format_tag_delimiters .-
+    use_put_object On
+    upload_timeout 30s

tests/.gitignore

@@ -0,0 +1,2 @@
+bin
+integ/out