Merge pull request #408 from alltilla/s3-dest-content-type

S3 dest: `content-type()`

modules/python-modules/syslogng/modules/s3/s3_destination.py

@@ -69,6 +69,7 @@ def __init_options(self, options: Dict[str, Any]) -> None:
             self.kms_key = str(options["kms_key"])
             self.storage_class = str(options["storage_class"]).upper().replace("-", "_")
             self.canned_acl = str(options["canned_acl"]).lower().replace("_", "-")
+            self.content_type = str(options["content_type"])
         except KeyError:
             assert False, (
                 f"S3: {str(exc_info()[1])[1:-1]}() option is missing. "
@@ -392,6 +393,7 @@ def __create_initial_s3_object(self, target_key: str, timestamp: str) -> S3Objec
             compresslevel=self.compresslevel,
             chunk_size=self.chunk_size,
             canned_acl=self.canned_acl,
+            content_type=self.content_type,
             client=self.client,
             logger=self.logger,
             exit_requested=self.exit_requested,

modules/python-modules/syslogng/modules/s3/s3_object.py

@@ -121,6 +121,7 @@ def __init__(
         kms_key: Optional[str] = None,
         storage_class: Optional[str] = None,
         canned_acl: Optional[str] = None,
+        content_type: Optional[str] = None,
     ):
         if path is not None:
             self.__path = path
@@ -138,6 +139,7 @@ def __init__(
                 and server_side_encryption is not None
                 and kms_key is not None
                 and canned_acl is not None
+                and content_type is not None
             )
         else:
             assert False
@@ -163,6 +165,7 @@ def __init__(
                 "canned-acl",
                 "kms-key",
                 "server-side-encryption",
+                "content-type",
             }:
                 cache[upgrade_field] = cache.get(upgrade_field, "")
 
@@ -207,6 +210,7 @@ def __init__(
         self.__canned_acl: str = cache.get("canned-acl", canned_acl)
         self.__kms_key: str = cache.get("kms-key", kms_key)
         self.__server_side_encryption: str = cache.get("server-side-encryption", server_side_encryption)
+        self.__content_type: str = cache.get("content-type", content_type)
 
         self.__flush()
 
@@ -232,6 +236,7 @@ def __flush(self) -> None:
             "upload-id": self.__upload_id,
             "uploaded-parts": self.__uploaded_parts,
             "pending-parts": self.__pending_parts,
+            "content-type": self.__content_type,
         }
 
         with self.__path.open("w", encoding="utf-8") as f:
@@ -303,6 +308,10 @@ def storage_class(self) -> str:
     def canned_acl(self) -> str:
         return self.__canned_acl
 
+    @property
+    def content_type(self) -> str:
+        return self.__content_type
+
     @property
     def finished(self) -> bool:
         with self.__lock:
@@ -380,6 +389,7 @@ def __init__(
         kms_key: Optional[str] = None,
         storage_class: Optional[str] = None,
         canned_acl: Optional[str] = None,
+        content_type: Optional[str] = None,
         persist: Optional[S3ObjectPersist] = None,
         persist_name: Optional[str] = None,
         working_dir: Optional[Path] = None,
@@ -414,6 +424,7 @@ def __init__(
                 and kms_key is not None
                 and storage_class is not None
                 and canned_acl is not None
+                and content_type is not None
             )
             self.__persist = S3ObjectPersist(
                 working_dir=working_dir,
@@ -428,6 +439,7 @@ def __init__(
                 kms_key=kms_key,
                 storage_class=storage_class,
                 canned_acl=canned_acl,
+                content_type=content_type,
             )
             self.__persist.index = self.__get_next_index(target_index)
 
@@ -472,6 +484,7 @@ def create_initial(
         kms_key: str,
         storage_class: str,
         canned_acl: str,
+        content_type: str,
         persist_name: str,
         chunk_size: int,
         executor: ThreadPoolExecutor,
@@ -494,6 +507,7 @@ def create_initial(
             kms_key=kms_key,
             storage_class=storage_class,
             canned_acl=canned_acl,
+            content_type=content_type,
             chunk_size=chunk_size,
             persist_name=persist_name,
             exit_requested=exit_requested,
@@ -526,6 +540,7 @@ def create_next(self) -> S3Object:
             storage_class=self.__persist.storage_class,
             canned_acl=self.__persist.canned_acl,
             chunk_size=self.__persist.chunk_size,
+            content_type=self.__persist.content_type,
             persist_name=self.__persist.persist_name,
             exit_requested=self.__exit_requested,
         )
@@ -589,29 +604,27 @@ def __ensure_multipart_upload_started(self) -> bool:
         if self.__persist.upload_id != "":
             return True
 
-        sse_arguments = {}
+        extra_args = {}
+
         if self.__persist.server_side_encryption != "":
-            sse_arguments["ServerSideEncryption"] = self.__persist.server_side_encryption
+            extra_args["ServerSideEncryption"] = self.__persist.server_side_encryption
 
         if self.__persist.kms_key != "":
-            sse_arguments["SSEKMSKeyId"] = self.__persist.kms_key
+            extra_args["SSEKMSKeyId"] = self.__persist.kms_key
+
+        if self.__persist.canned_acl != "":
+            extra_args["ACL"] = self.__persist.canned_acl,
+
+        if self.__persist.content_type != "":
+            extra_args["ContentType"] = self.__persist.content_type
 
         try:
-            if self.__persist.canned_acl != "":
-                response = self.__client.create_multipart_upload(
-                    Bucket=self.bucket,
-                    Key=self.key,
-                    StorageClass=self.__persist.storage_class,
-                    ACL=self.__persist.canned_acl,
-                    **sse_arguments,
-                )
-            else:
-                response = self.__client.create_multipart_upload(
-                    Bucket=self.bucket,
-                    Key=self.key,
-                    StorageClass=self.__persist.storage_class,
-                    **sse_arguments,
-                )
+            response = self.__client.create_multipart_upload(
+                Bucket=self.bucket,
+                Key=self.key,
+                StorageClass=self.__persist.storage_class,
+                **extra_args,
+            )
         except (ClientError, EndpointConnectionError) as e:
             self.__logger.error(f"Failed to create multipart upload: {self.bucket}/{self.key} => {e}")
             return False

modules/python-modules/syslogng/modules/s3/scl/s3.conf

@@ -41,6 +41,7 @@ block destination s3(
     kms_key("")
     storage_class("STANDARD")
     canned_acl("")
+    content-type("application/octet-stream")
     ...
 )
 {
@@ -67,6 +68,7 @@ block destination s3(
             "kms_key" => "`kms_key`"
             "storage_class" => "`storage_class`"
             "canned_acl" => "`canned_acl`"
+            "content_type" => "`content_type`"
         )
         `__VARARGS__`
     );

news/feature-408.md

@@ -0,0 +1 @@
+`s3` destination: Added `content-type()` option.