S3Uploader implements Runnable { String uploadFile(File fileToUpload, String logType) { String bucketName = s3OutputConfiguration.getS3BucketName(); String s3AccessKey = s3OutputConfiguration.getS3AccessKey(); String s3SecretKey = s3OutputConfiguration.getS3SecretKey(); String compressionAlgo = s3OutputConfiguration.getCompressionAlgo(); String keySuffix = fileToUpload.getName() + "." + compressionAlgo; String s3Path = new S3LogPathResolver().getResolvedPath( s3OutputConfiguration.getS3Path() + LogFeederConstants.S3_PATH_SEPARATOR + logType, keySuffix, s3OutputConfiguration.getCluster()); LOG.info(String.format("keyPrefix=%s, keySuffix=%s, s3Path=%s", s3OutputConfiguration.getS3Path(), keySuffix, s3Path)); File sourceFile = createCompressedFileForUpload(fileToUpload, compressionAlgo); LOG.info("Starting S3 upload " + sourceFile + " -> " + bucketName + ", " + s3Path); uploadFileToS3(bucketName, s3Path, sourceFile, s3AccessKey, s3SecretKey); sourceFile.delete(); if (deleteOnEnd) { LOG.info("Deleting input file as required"); if (!fileToUpload.delete()) { LOG.error("Could not delete file " + fileToUpload.getAbsolutePath() + " after upload to S3"); } } return s3Path; } S3Uploader(S3OutputConfiguration s3OutputConfiguration, boolean deleteOnEnd, String logType); @Override void run(); static final String POISON_PILL; }
@Test public void shouldUploadToS3ToRightBucket() { File fileToUpload = mock(File.class); String fileName = "hdfs_namenode.log.123343493473948"; expect(fileToUpload.getName()).andReturn(fileName); final File compressedFile = mock(File.class); Map<String, Object> configs = setupS3Configs(); S3OutputConfiguration s3OutputConfiguration = new S3OutputConfiguration(configs); expect(compressedFile.delete()).andReturn(true); expect(fileToUpload.delete()).andReturn(true); replay(fileToUpload, compressedFile); S3Uploader s3Uploader = new S3Uploader(s3OutputConfiguration, true, LOG_TYPE) { @Override protected File createCompressedFileForUpload(File fileToUpload, String compressionAlgo) { return compressedFile; } protected void uploadFileToS3(String bucketName, String s3Key, File localFile, String accessKey, String secretKey) { } }; String resolvedPath = s3Uploader.uploadFile(fileToUpload, LOG_TYPE); assertEquals("test_path/hdfs_namenode/hdfs_namenode.log.123343493473948.gz", resolvedPath); }
@Test public void shouldCleanupLocalFilesOnSuccessfulUpload() { File fileToUpload = mock(File.class); String fileName = "hdfs_namenode.log.123343493473948"; expect(fileToUpload.getName()).andReturn(fileName); final File compressedFile = mock(File.class); Map<String, Object> configs = setupS3Configs(); S3OutputConfiguration s3OutputConfiguration = new S3OutputConfiguration(configs); expect(compressedFile.delete()).andReturn(true); expect(fileToUpload.delete()).andReturn(true); replay(fileToUpload, compressedFile); S3Uploader s3Uploader = new S3Uploader(s3OutputConfiguration, true, LOG_TYPE) { @Override protected File createCompressedFileForUpload(File fileToUpload, String compressionAlgo) { return compressedFile; } protected void uploadFileToS3(String bucketName, String s3Key, File localFile, String accessKey, String secretKey) { } }; s3Uploader.uploadFile(fileToUpload, LOG_TYPE); verify(fileToUpload); verify(compressedFile); }
@Test public void shouldNotCleanupUncompressedFileIfNotRequired() { File fileToUpload = mock(File.class); String fileName = "hdfs_namenode.log.123343493473948"; expect(fileToUpload.getName()).andReturn(fileName); final File compressedFile = mock(File.class); Map<String, Object> configs = setupS3Configs(); S3OutputConfiguration s3OutputConfiguration = new S3OutputConfiguration(configs); expect(compressedFile.delete()).andReturn(true); replay(fileToUpload, compressedFile); S3Uploader s3Uploader = new S3Uploader(s3OutputConfiguration, false, LOG_TYPE) { @Override protected File createCompressedFileForUpload(File fileToUpload, String compressionAlgo) { return compressedFile; } protected void uploadFileToS3(String bucketName, String s3Key, File localFile, String accessKey, String secretKey) { } }; s3Uploader.uploadFile(fileToUpload, LOG_TYPE); verify(fileToUpload); verify(compressedFile); }
@Test public void shouldExpandVariablesInPath() { File fileToUpload = mock(File.class); String fileName = "hdfs_namenode.log.123343493473948"; expect(fileToUpload.getName()).andReturn(fileName); final File compressedFile = mock(File.class); Map<String, Object> configs = setupS3Configs(); configs.put(S3OutputConfiguration.S3_LOG_DIR_KEY, "$cluster/"+TEST_PATH); S3OutputConfiguration s3OutputConfiguration = new S3OutputConfiguration(configs); expect(compressedFile.delete()).andReturn(true); expect(fileToUpload.delete()).andReturn(true); replay(fileToUpload, compressedFile); S3Uploader s3Uploader = new S3Uploader(s3OutputConfiguration, true, LOG_TYPE) { @Override protected File createCompressedFileForUpload(File fileToUpload, String compressionAlgo) { return compressedFile; } protected void uploadFileToS3(String bucketName, String s3Key, File localFile, String accessKey, String secretKey) { } }; s3Uploader.uploadFile(fileToUpload, LOG_TYPE); }