feat: Add timestamp to S3 uploaded JDBC log filenames

kentoyoshida · kentoyoshida · commit 394440d998fd · 2025-06-30T02:34:51.000Z
- Add automatic timestamp formatting (yyyyMMdd_HHmmss) to uploaded filenames
- Example: snowflake_jdbc.log -&gt; snowflake_jdbc_20250630_021500.log
- Update README with timestamp feature documentation
- Clarify s3_prefix as optional parameter in documentation
- Improve S3 path handling for empty prefix cases
diff --git a/README.md b/README.md
@@ -32,7 +32,7 @@ Snowflake output plugin for Embulk loads records to Snowflake.
 - **match_by_column_name**: specify whether to load semi-structured data into columns in the target table that match corresponding columns represented in the data. ("case_sensitive", "case_insensitive", "none", default: "none")
 - **upload_jdbc_log_to_s3**: enable automatic upload of JDBC driver logs to S3 when communication errors occur (boolean, default: false)
 - **s3_bucket**: S3 bucket name for JDBC log upload (string, required when upload_jdbc_log_to_s3 is true)
-- **s3_prefix**: S3 key prefix for JDBC log upload (string, required when upload_jdbc_log_to_s3 is true)
+- **s3_prefix**: S3 key prefix for JDBC log upload (string, optional - if empty, files are uploaded to bucket root)
 - **s3_region**: AWS region for S3 bucket (string, required when upload_jdbc_log_to_s3 is true)
 - **s3_access_key_id**: AWS access key ID for S3 access (string, optional - uses default AWS credentials provider chain if not specified)
 - **s3_secret_access_key**: AWS secret access key for S3 access (string, optional - uses default AWS credentials provider chain if not specified)
@@ -89,7 +89,7 @@ out:
   # JDBC log upload configuration
   upload_jdbc_log_to_s3: true
   s3_bucket: your-log-bucket
-  s3_prefix: snowflake-jdbc-logs
+  s3_prefix: snowflake-jdbc-logs  # Optional: omit to upload to bucket root
   s3_region: us-east-1
   
   # Optional: Explicit AWS credentials (uses IAM role if not specified)
@@ -107,7 +107,8 @@ out:
 
 - JDBC logs are only uploaded when communication errors occur during Snowflake operations
 - The plugin automatically finds the latest `snowflake_jdbc*.log` file in the system temp directory
-- Logs are uploaded to `s3://{bucket}/{prefix}/{filename}`
+- **Automatic timestamping**: Upload timestamp is automatically added to the filename (format: `yyyyMMdd_HHmmss`)
+  - Example: `snowflake_jdbc0.log.0` → `snowflake_jdbc0.log_20250630_021500.0`
 - If S3 upload fails, a warning is logged but the original error is still thrown
 - If required S3 configuration is missing, a warning is logged and log upload is skipped
 
diff --git a/build.gradle b/build.gradle
@@ -53,23 +53,36 @@ dependencies {
     compile "org.embulk:embulk-output-jdbc:0.10.2"
     compile "net.snowflake:snowflake-jdbc:3.13.26"
 
-    implementation platform("software.amazon.awssdk:bom:2.25.20")
+    compile platform("software.amazon.awssdk:bom:2.25.20")
 
-    implementation("software.amazon.awssdk:s3") {
+    compile("software.amazon.awssdk:s3") {
         exclude group: "com.fasterxml.jackson.core", module: "jackson-annotations"
         exclude group: "com.fasterxml.jackson.core", module: "jackson-core"
         exclude group: "com.fasterxml.jackson.core", module: "jackson-databind"
         exclude group: "joda-time", module: "joda-time"
         exclude group: "commons-logging", module: "commons-logging"
+        exclude group: "org.slf4j", module: "slf4j-api"
     }
 
-    implementation("software.amazon.awssdk:regions") {
+    compile("software.amazon.awssdk:regions") {
         exclude group: "com.fasterxml.jackson.core", module: "jackson-annotations"
         exclude group: "com.fasterxml.jackson.core", module: "jackson-core"
         exclude group: "com.fasterxml.jackson.core", module: "jackson-databind"
         exclude group: "joda-time", module: "joda-time"
         exclude group: "commons-logging", module: "commons-logging"
+        exclude group: "org.slf4j", module: "slf4j-api"
     }
+
+    compile("software.amazon.awssdk:auth") {
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-annotations"
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-core"
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-databind"
+        exclude group: "joda-time", module: "joda-time"
+        exclude group: "commons-logging", module: "commons-logging"
+        exclude group: "org.slf4j", module: "slf4j-api"
+    }
+
+    compile "commons-logging:commons-logging:1.2"
 }
 embulkPlugin {
     mainClass = "org.embulk.output.SnowflakeOutputPlugin"
diff --git a/gradle/dependency-locks/embulkPluginRuntime.lockfile b/gradle/dependency-locks/embulkPluginRuntime.lockfile
@@ -5,11 +5,53 @@ com.fasterxml.jackson.core:jackson-annotations:2.6.7
 com.fasterxml.jackson.core:jackson-core:2.6.7
 com.fasterxml.jackson.core:jackson-databind:2.6.7
 com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.6.7
+commons-codec:commons-codec:1.15
+commons-logging:commons-logging:1.2
+io.netty:netty-buffer:4.1.108.Final
+io.netty:netty-codec-http2:4.1.108.Final
+io.netty:netty-codec-http:4.1.108.Final
+io.netty:netty-codec:4.1.108.Final
+io.netty:netty-common:4.1.108.Final
+io.netty:netty-handler:4.1.108.Final
+io.netty:netty-resolver:4.1.108.Final
+io.netty:netty-transport-classes-epoll:4.1.108.Final
+io.netty:netty-transport-native-unix-common:4.1.108.Final
+io.netty:netty-transport:4.1.108.Final
 javax.validation:validation-api:1.1.0.Final
 net.snowflake:snowflake-jdbc:3.13.26
+org.apache.httpcomponents:httpclient:4.5.13
+org.apache.httpcomponents:httpcore:4.4.13
 org.embulk:embulk-output-jdbc:0.10.2
 org.embulk:embulk-util-config:0.3.0
 org.embulk:embulk-util-json:0.1.1
 org.embulk:embulk-util-retryhelper:0.8.2
 org.embulk:embulk-util-rubytime:0.3.2
 org.embulk:embulk-util-timestamp:0.2.1
+org.reactivestreams:reactive-streams:1.0.4
+software.amazon.awssdk:annotations:2.25.20
+software.amazon.awssdk:apache-client:2.25.20
+software.amazon.awssdk:arns:2.25.20
+software.amazon.awssdk:auth:2.25.20
+software.amazon.awssdk:aws-core:2.25.20
+software.amazon.awssdk:aws-query-protocol:2.25.20
+software.amazon.awssdk:aws-xml-protocol:2.25.20
+software.amazon.awssdk:checksums-spi:2.25.20
+software.amazon.awssdk:checksums:2.25.20
+software.amazon.awssdk:crt-core:2.25.20
+software.amazon.awssdk:endpoints-spi:2.25.20
+software.amazon.awssdk:http-auth-aws:2.25.20
+software.amazon.awssdk:http-auth-spi:2.25.20
+software.amazon.awssdk:http-auth:2.25.20
+software.amazon.awssdk:http-client-spi:2.25.20
+software.amazon.awssdk:identity-spi:2.25.20
+software.amazon.awssdk:json-utils:2.25.20
+software.amazon.awssdk:metrics-spi:2.25.20
+software.amazon.awssdk:netty-nio-client:2.25.20
+software.amazon.awssdk:profiles:2.25.20
+software.amazon.awssdk:protocol-core:2.25.20
+software.amazon.awssdk:regions:2.25.20
+software.amazon.awssdk:s3:2.25.20
+software.amazon.awssdk:sdk-core:2.25.20
+software.amazon.awssdk:third-party-jackson-core:2.25.20
+software.amazon.awssdk:utils:2.25.20
+software.amazon.eventstream:eventstream:1.0.1
diff --git a/src/main/java/org/embulk/output/SnowflakeOutputPlugin.java b/src/main/java/org/embulk/output/SnowflakeOutputPlugin.java
@@ -250,6 +250,7 @@ public ConfigDiff transaction(
     try {
       snowflakeCon = (SnowflakeOutputConnection) getConnector(task, true).connect(true);
       snowflakeCon.runCreateStage(stageIdentifier);
+
       configDiff = super.transaction(config, schema, taskCount, control);
       if (t.getDeleteStage()) {
         runDropStageWithRecovery(snowflakeCon, stageIdentifier, task);
@@ -265,18 +266,22 @@ public ConfigDiff transaction(
           final Optional<String> s3Region = t.getS3Region();
           final Optional<String> s3AccessKeyId = t.getS3AccessKeyId();
           final Optional<String> s3SecretAccessKey = t.getS3SecretAccessKey();
-          if (!s3Bucket.isPresent() || !s3Prefix.isPresent() || !s3Region.isPresent()) {
-            logger.warn("s3_bucket, s3_prefix, and s3_region must be set when upload_jdbc_log_to_s3 is true");
+          if (!s3Bucket.isPresent() || !s3Region.isPresent()) {
+            logger.warn("s3_bucket, and s3_region must be set when upload_jdbc_log_to_s3 is true");
           } else {
-            try (JdbcLogUploader jdbcLogUploader = new JdbcLogUploader(s3Bucket.get(), s3Prefix.get(), s3Region.get(), s3AccessKeyId.orElse(null), s3SecretAccessKey.orElse(null))) {
-              // snowflake_jdbc*.log で最新のファイルを探してアップロード
+            try (JdbcLogUploader jdbcLogUploader =
+                new JdbcLogUploader(
+                    s3Bucket.get(),
+                    s3Prefix.orElse(""),
+                    s3Region.get(),
+                    s3AccessKeyId.orElse(null),
+                    s3SecretAccessKey.orElse(null))) {
               String tmpDir = System.getProperty("java.io.tmpdir", "/tmp");
               File logDir = new File(tmpDir);
               File[] logFiles =
                   logDir.listFiles(
-                      (dir, name) -> name.startsWith("snowflake_jdbc") && name.endsWith(".log"));
+                      (dir, name) -> name.startsWith("snowflake_jdbc"));
               if (logFiles != null && logFiles.length > 0) {
-                // 最終更新日時が新しいファイルを選択
                 Optional<File> latest =
                     Arrays.stream(logFiles).max(Comparator.comparingLong(File::lastModified));
                 if (latest.isPresent()) {
diff --git a/src/main/java/org/embulk/output/s3/JdbcLogUploader.java b/src/main/java/org/embulk/output/s3/JdbcLogUploader.java
@@ -1,11 +1,13 @@
 package org.embulk.output.s3;
 
 import java.io.File;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
-import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.core.sync.RequestBody;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3Client;
@@ -19,24 +21,27 @@ public class JdbcLogUploader implements AutoCloseable {
   private final String prefix;
   private final Region region;
 
-  public JdbcLogUploader(String bucket, String prefix, String region, String accessKeyId, String secretAccessKey) {
+  public JdbcLogUploader(
+      String bucket, String prefix, String region, String accessKeyId, String secretAccessKey) {
     this.bucket = bucket;
     this.prefix = prefix;
     this.region = Region.of(region);
-    
+
     if (accessKeyId != null && secretAccessKey != null) {
       // Use explicit credentials
       AwsBasicCredentials awsCreds = AwsBasicCredentials.create(accessKeyId, secretAccessKey);
-      this.s3Client = S3Client.builder()
-        .region(this.region)
-        .credentialsProvider(StaticCredentialsProvider.create(awsCreds))
-        .build();
+      this.s3Client =
+          S3Client.builder()
+              .region(this.region)
+              .credentialsProvider(StaticCredentialsProvider.create(awsCreds))
+              .build();
     } else {
       // Use default credentials provider (IAM role, environment variables, etc.)
-      this.s3Client = S3Client.builder()
-        .region(this.region)
-        .credentialsProvider(DefaultCredentialsProvider.create())
-        .build();
+      this.s3Client =
+          S3Client.builder()
+              .region(this.region)
+              .credentialsProvider(DefaultCredentialsProvider.create())
+              .build();
     }
   }
 
@@ -46,15 +51,30 @@ public void uploadIfExists(File file) {
       return;
     }
 
-    String key = prefix + "/" + file.getName();
+    // Add timestamp to filename
+    String timestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss"));
+    String originalFileName = file.getName();
+    String fileNameWithTimestamp;
+    
+    // Insert timestamp before file extension
+    int lastDotIndex = originalFileName.lastIndexOf('.');
+    if (lastDotIndex > 0) {
+      String nameWithoutExt = originalFileName.substring(0, lastDotIndex);
+      String extension = originalFileName.substring(lastDotIndex);
+      fileNameWithTimestamp = nameWithoutExt + "_" + timestamp + extension;
+    } else {
+      fileNameWithTimestamp = originalFileName + "_" + timestamp;
+    }
+    
+    String key = prefix.isEmpty() ? fileNameWithTimestamp : prefix + "/" + fileNameWithTimestamp;
     try {
       PutObjectRequest putObjectRequest =
           PutObjectRequest.builder().bucket(bucket).key(key).build();
 
       s3Client.putObject(putObjectRequest, RequestBody.fromFile(file));
-      logger.info("Uploaded {} to s3://{}/{}", file.getAbsolutePath(), bucket, key);
+      logger.info("Uploaded {}", file.getAbsolutePath());
     } catch (Exception e) {
-      logger.error("Failed to upload {} to S3: {}", file.getAbsolutePath(), e.getMessage(), e);
+      logger.error("Failed to upload {}", file.getAbsolutePath());
     }
   }
 
