URL decode s3 key in s3:ObjectCreated (#5692)

When a file includes `:`, it gets converted to `%3A`, and doing
`GetObject` fails.

Author: tontinton

quickwit/Cargo.lock

@@ -32,6 +32,7 @@ libz-sys = { workspace = true, optional = true }
 once_cell = { workspace = true }
 oneshot = { workspace = true }
 openssl = { workspace = true, optional = true }
+percent-encoding = { workspace = true }
 pulsar = { workspace = true, optional = true }
 quickwit-query = { workspace = true }
 regex = { workspace = true }

quickwit/quickwit-indexing/Cargo.toml

@@ -157,7 +157,10 @@ fn uri_from_s3_notification(message: &[u8], ack_id: &str) -> Result<Uri, PreProc
     let bucket = value["Records"][0]["s3"]["bucket"]["name"]
         .as_str()
         .context("invalid S3 notification: Records[0].s3.bucket.name not found")?;
-    Uri::from_str(&format!("s3://{}/{}", bucket, key)).map_err(|e| e.into())
+    let encoded_key = percent_encoding::percent_decode(key.as_bytes())
+        .decode_utf8()
+        .context("invalid S3 notification: Records[0].s3.object.key could not be url decoded")?;
+    Uri::from_str(&format!("s3://{}/{}", bucket, encoded_key)).map_err(|e| e.into())
 }
 
 /// A message for which we know as much of the global processing status as
@@ -344,4 +347,51 @@ mod tests {
             panic!("Expected skippable error");
         }
     }
+
+    #[test]
+    fn test_uri_from_s3_notification_url_decode() {
+        let test_message = r#"
+        {
+            "Records": [
+                {
+                "eventVersion": "2.1",
+                "eventSource": "aws:s3",
+                "awsRegion": "us-west-2",
+                "eventTime": "2021-05-22T09:22:41.789Z",
+                "eventName": "ObjectCreated:Put",
+                "userIdentity": {
+                    "principalId": "AWS:AIDAJDPLRKLG7UEXAMPLE"
+                },
+                "requestParameters": {
+                    "sourceIPAddress": "127.0.0.1"
+                },
+                "responseElements": {
+                    "x-amz-request-id": "C3D13FE58DE4C810",
+                    "x-amz-id-2": "FMyUVURIx7Zv2cPi/IZb9Fk1/U4QfTaVK5fahHPj/"
+                },
+                "s3": {
+                    "s3SchemaVersion": "1.0",
+                    "configurationId": "testConfigRule",
+                    "bucket": {
+                        "name": "mybucket",
+                        "ownerIdentity": {
+                            "principalId": "A3NL1KOZZKExample"
+                        },
+                        "arn": "arn:aws:s3:::mybucket"
+                    },
+                    "object": {
+                        "key": "hello%3A%3Aworld%3A%3Alogs.json",
+                        "size": 1024,
+                        "eTag": "d41d8cd98f00b204e9800998ecf8427e",
+                        "versionId": "096fKKXTRTtl3on89fVO.nfljtsv6qko",
+                        "sequencer": "0055AED6DCD90281E5"
+                    }
+                }
+                }
+            ]
+        }"#;
+        let actual_uri = uri_from_s3_notification(test_message.as_bytes(), "myackid").unwrap();
+        let expected_uri = Uri::from_str("s3://mybucket/hello::world::logs.json").unwrap();
+        assert_eq!(actual_uri, expected_uri);
+    }
 }