[8.17](backport #44824) [filebeat][ABS] - Added missing "text/csv" content-type filter support (#44826)

Currently content-type "text/csv" did not have official filtering
support even though we support  csv decoding. This made blobs having this
content type defined were not picked up for processing. This PR as a "bugfix",
adds support for explicit "text/csv" content-type filtering in the
scheduler.

(cherry picked from commit fba3749)

---------

Co-authored-by: Shourie Ganguly <shourie.ganguly@elastic.co>

Author: mergify[bot]

CHANGELOG.next.asciidoc

@@ -119,6 +119,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix a bug in Salesforce input to only handle responses with 200 status code {pull}41015[41015]
 - If a Filestream input fails to be created, its ID is removed from the list of running input IDs {pull}44697[44697]
 - Ensure DEPROVISIONED Okta entities are published by Okta entityanalytics provider. {issue}12658[12658] {pull}44719[44719]
+- Added missing "text/csv" content-type filter support in azureblobsortorage input. {issue}44596[44596] {pull}44824[44824]
 - Fix unexpected EOF detection and improve memory usage. {pull}44813[44813]
 
 *Heartbeat*

x-pack/filebeat/input/azureblobstorage/input_test.go

@@ -36,6 +36,7 @@ const (
 	beatsNdJSONContainer        = "beatsndjsoncontainer"
 	beatsGzJSONContainer        = "beatsgzjsoncontainer"
 	beatsJSONWithArrayContainer = "beatsjsonwitharraycontainer"
+	beatsCSVContainer           = "beatscsvcontainer"
 )
 
 func Test_StorageClient(t *testing.T) {
@@ -471,6 +472,28 @@ func Test_StorageClient(t *testing.T) {
 				mock.Beatscontainer_2_blob_data3_json:  true,
 			},
 		},
+		{
+			name: "ReadCSV",
+			baseConfig: map[string]interface{}{
+				"account_name":                        "beatsblobnew",
+				"auth.shared_credentials.account_key": "7pfLm1betGiRyyABEM/RFrLYlafLZHbLtGhB52LkWVeBxE7la9mIvk6YYAbQKYE/f0GdhiaOZeV8+AStsAdr/Q==",
+				"max_workers":                         1,
+				"poll":                                true,
+				"poll_interval":                       "10s",
+				"decoding.codec.csv.enabled":          true,
+				"decoding.codec.csv.comma":            " ",
+				"containers": []map[string]interface{}{
+					{
+						"name": beatsCSVContainer,
+					},
+				},
+			},
+			mockHandler: mock.AzureStorageFileServer,
+			expected: map[string]bool{
+				mock.BeatsFilesContainer_csv[0]: true,
+				mock.BeatsFilesContainer_csv[1]: true,
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -540,7 +563,7 @@ func Test_StorageClient(t *testing.T) {
 					var err error
 					val, err = got.Fields.GetValue("message")
 					assert.NoError(t, err)
-					assert.True(t, tt.expected[val.(string)])
+					assert.True(t, tt.expected[strings.ReplaceAll(val.(string), "\r\n", "\n")])
 					assert.Equal(t, tt.expectedError, err)
 					receivedCount += 1
 					if receivedCount == len(tt.expected) {

x-pack/filebeat/input/azureblobstorage/mock/data_files.go

@@ -10,6 +10,7 @@ const (
 	beatsNdJSONContainer        = "beatsndjsoncontainer"
 	beatsGzJSONContainer        = "beatsgzjsoncontainer"
 	beatsJSONWithArrayContainer = "beatsjsonwitharraycontainer"
+	beatsCSVContainer           = "beatscsvcontainer"
 )
 
 var fileContainers = map[string]bool{
@@ -18,6 +19,7 @@ var fileContainers = map[string]bool{
 	beatsNdJSONContainer:        true,
 	beatsGzJSONContainer:        true,
 	beatsJSONWithArrayContainer: true,
+	beatsCSVContainer:           true,
 }
 
 var availableFileBlobs = map[string]map[string]bool{
@@ -38,6 +40,9 @@ var availableFileBlobs = map[string]map[string]bool{
 		"array-at-root.json": true,
 		"nested-arrays.json": true,
 	},
+	beatsCSVContainer: {
+		"txn1.csv": true,
+	},
 }
 
 var fetchFilesContainer = map[string]string{
@@ -197,6 +202,30 @@ var fetchFilesContainer = map[string]string{
 						</Blobs>
 						<NextMarker />
 					</EnumerationResults>`,
+	beatsCSVContainer: `<?xml version="1.0" encoding="utf-8"?>
+	<EnumerationResults ServiceEndpoint="https://127.0.0.1/" ContainerName="beatscsvcontainer">
+		<Blobs>
+			<Blob>
+				<Name>txn1.csv</Name>
+				<Properties>
+					<Last-Modified>Wed, 14 Sep 2022 12:12:28 GMT</Last-Modified>
+					<Etag>0x8DA964A64516C82</Etag>
+					<Content-Length>643</Content-Length>
+					<Content-Type>text/csv</Content-Type>
+					<Content-Encoding />
+					<Content-Language />
+					<Content-MD5>UjQX73kQRTHx+UyXZDvVkg==</Content-MD5>
+					<Cache-Control />
+					<Content-Disposition />
+					<BlobType>BlockBlob</BlobType>
+					<LeaseStatus>unlocked</LeaseStatus>
+					<LeaseState>available</LeaseState>
+				</Properties>
+				<Metadata />
+			</Blob>
+		</Blobs>
+		<NextMarker />
+	</EnumerationResults>`,
 }
 
 var BeatsFilesContainer_multiline_json = []string{
@@ -231,3 +260,8 @@ var BeatsFilesContainer_multiline_json_gz = []string{
 	"{\n    \"@timestamp\": \"2021-05-25T17:25:42.806Z\",\n    \"log.level\": \"error\",\n    \"message\": \"error making http request\"\n}",
 	"{\n    \"@timestamp\": \"2021-05-25T17:25:51.391Z\",\n    \"log.level\": \"info\",\n    \"message\": \"available disk space 44.3gb\"\n}",
 }
+
+var BeatsFilesContainer_csv = []string{
+	"{\"id\":\"1\",\"name\":\"Alice\",\"email\":\"alice@example.com\",\"status\":\"active\"}",
+	"{\"id\":\"2\",\"name\":\"Bob\",\"email\":\"bob@example.com\",\"status\":\"inactive\"}",
+}

x-pack/filebeat/input/azureblobstorage/mock/mock.go

@@ -78,6 +78,8 @@ func AzureStorageFileServer() http.Handler {
 						w.Header().Set(contentType, jsonType)
 					case "log.ndjson":
 						w.Header().Set(contentType, "application/x-ndjson")
+					case "txn1.csv":
+						w.Header().Set(contentType, "text/csv")
 					}
 					w.Write(data)
 					return

x-pack/filebeat/input/azureblobstorage/mock/testdata/txn1.csv

@@ -0,0 +1,3 @@
+id name email status
+1 Alice alice@example.com active
+2 Bob bob@example.com inactive

x-pack/filebeat/input/azureblobstorage/types.go

@@ -37,6 +37,7 @@ const (
 	octetType            = "application/octet-stream"
 	ndJsonType           = "application/x-ndjson"
 	gzType               = "application/x-gzip"
+	csvType              = "text/csv"
 	encodingGzip         = "gzip"
 )
 
@@ -59,4 +60,5 @@ var allowedContentTypes = map[string]bool{
 	octetType:  true,
 	ndJsonType: true,
 	gzType:     true,
+	csvType:    true,
 }