@Test public void testReadStepLogsFromS3_whenLogFileExistsInS3() throws Exception { FileInputStream inputStream = null; S3ObjectInputStream s3ObjectInputStream = null; String stagingBucketName = "alzhk"; String hadoopJobFlowId = "j-11WRZQW6NIQOA"; String stepId = "s-15PK2NMVIPRPF"; PowerMockito.doReturn( "" ) .when( s3Client, "readLogFromS3", stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + "/controller.gz" ); PowerMockito.doReturn( "" ) .when( s3Client, "readLogFromS3", stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + "/stdout.gz" ); PowerMockito.doReturn( "" ) .when( s3Client, "readLogFromS3", stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + "/syslog.gz" ); try { inputStream = new FileInputStream( gzArchName ); s3ObjectInputStream = new S3ObjectInputStream( inputStream, null, false ); S3Object s3Object = new S3Object(); s3Object.setObjectContent( s3ObjectInputStream ); Mockito.when( awsS3Client.getObject( Matchers.anyString(), Matchers.anyString() ) ).thenReturn( s3Object ); Mockito.when( awsS3Client.doesObjectExist( Matchers.anyString(), Matchers.anyString() ) ).thenReturn( true ); String lineFromStepLogFile = s3Client.readStepLogsFromS3( stagingBucketName, hadoopJobFlowId, stepId ); lineFromStepLogFile = lineFromStepLogFile.replace( "\n", "" ).replace( "\r", "" ); Assert.assertEquals( "all bootstrap actions complete and instance ready", lineFromStepLogFile ); } finally { if ( inputStream != null ) { inputStream.close(); } if ( s3ObjectInputStream != null ) { s3ObjectInputStream.close(); } } }
@Override public String readStepLogsFromS3( String stagingBucketName, String hadoopJobFlowId, String stepId ) { String lineSeparator = System.getProperty( "line.separator" ); String[] logArchives = { "/controller.gz", "/stdout.gz", "/syslog.gz", "/stderr.gz" }; StringBuilder logContents = new StringBuilder(); String logFromS3File = ""; String pathToStepLogs = ""; for ( String gzLogFile : logArchives ) { logFromS3File = readLogFromS3( stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + gzLogFile ); if ( logFromS3File != null && !logFromS3File.isEmpty() ) { logContents.append( logFromS3File + lineSeparator ); } } if ( logContents.length() == 0 ) { pathToStepLogs = "s3: logContents.append( "Step " + stepId + " failed. See logs here: " + pathToStepLogs + lineSeparator ); } return logContents.toString(); }
S3ClientImpl implements S3Client { @Override public String readStepLogsFromS3( String stagingBucketName, String hadoopJobFlowId, String stepId ) { String lineSeparator = System.getProperty( "line.separator" ); String[] logArchives = { "/controller.gz", "/stdout.gz", "/syslog.gz", "/stderr.gz" }; StringBuilder logContents = new StringBuilder(); String logFromS3File = ""; String pathToStepLogs = ""; for ( String gzLogFile : logArchives ) { logFromS3File = readLogFromS3( stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + gzLogFile ); if ( logFromS3File != null && !logFromS3File.isEmpty() ) { logContents.append( logFromS3File + lineSeparator ); } } if ( logContents.length() == 0 ) { pathToStepLogs = "s3: logContents.append( "Step " + stepId + " failed. See logs here: " + pathToStepLogs + lineSeparator ); } return logContents.toString(); } }
S3ClientImpl implements S3Client { @Override public String readStepLogsFromS3( String stagingBucketName, String hadoopJobFlowId, String stepId ) { String lineSeparator = System.getProperty( "line.separator" ); String[] logArchives = { "/controller.gz", "/stdout.gz", "/syslog.gz", "/stderr.gz" }; StringBuilder logContents = new StringBuilder(); String logFromS3File = ""; String pathToStepLogs = ""; for ( String gzLogFile : logArchives ) { logFromS3File = readLogFromS3( stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + gzLogFile ); if ( logFromS3File != null && !logFromS3File.isEmpty() ) { logContents.append( logFromS3File + lineSeparator ); } } if ( logContents.length() == 0 ) { pathToStepLogs = "s3: logContents.append( "Step " + stepId + " failed. See logs here: " + pathToStepLogs + lineSeparator ); } return logContents.toString(); } S3ClientImpl( AmazonS3 s3Client ); }
S3ClientImpl implements S3Client { @Override public String readStepLogsFromS3( String stagingBucketName, String hadoopJobFlowId, String stepId ) { String lineSeparator = System.getProperty( "line.separator" ); String[] logArchives = { "/controller.gz", "/stdout.gz", "/syslog.gz", "/stderr.gz" }; StringBuilder logContents = new StringBuilder(); String logFromS3File = ""; String pathToStepLogs = ""; for ( String gzLogFile : logArchives ) { logFromS3File = readLogFromS3( stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + gzLogFile ); if ( logFromS3File != null && !logFromS3File.isEmpty() ) { logContents.append( logFromS3File + lineSeparator ); } } if ( logContents.length() == 0 ) { pathToStepLogs = "s3: logContents.append( "Step " + stepId + " failed. See logs here: " + pathToStepLogs + lineSeparator ); } return logContents.toString(); } S3ClientImpl( AmazonS3 s3Client ); @Override void createBucketIfNotExists( String stagingBucketName ); @Override void deleteObjectFromBucket( String stagingBucketName, String key ); @Override void putObjectInBucket( String stagingBucketName, String key, File tmpFile ); @Override String readStepLogsFromS3( String stagingBucketName, String hadoopJobFlowId, String stepId ); }
S3ClientImpl implements S3Client { @Override public String readStepLogsFromS3( String stagingBucketName, String hadoopJobFlowId, String stepId ) { String lineSeparator = System.getProperty( "line.separator" ); String[] logArchives = { "/controller.gz", "/stdout.gz", "/syslog.gz", "/stderr.gz" }; StringBuilder logContents = new StringBuilder(); String logFromS3File = ""; String pathToStepLogs = ""; for ( String gzLogFile : logArchives ) { logFromS3File = readLogFromS3( stagingBucketName, hadoopJobFlowId + "/steps/" + stepId + gzLogFile ); if ( logFromS3File != null && !logFromS3File.isEmpty() ) { logContents.append( logFromS3File + lineSeparator ); } } if ( logContents.length() == 0 ) { pathToStepLogs = "s3: logContents.append( "Step " + stepId + " failed. See logs here: " + pathToStepLogs + lineSeparator ); } return logContents.toString(); } S3ClientImpl( AmazonS3 s3Client ); @Override void createBucketIfNotExists( String stagingBucketName ); @Override void deleteObjectFromBucket( String stagingBucketName, String key ); @Override void putObjectInBucket( String stagingBucketName, String key, File tmpFile ); @Override String readStepLogsFromS3( String stagingBucketName, String hadoopJobFlowId, String stepId );  }