You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by mp...@apache.org on 2012/09/08 02:05:43 UTC
[1/2] git commit: FLUME-1424. File Channel should support encryption.
Updated Branches:
refs/heads/trunk 680c2d7cd -> d3c85b1d7
FLUME-1424. File Channel should support encryption.
(Brock Noland via Mike Percy)
Project: http://git-wip-us.apache.org/repos/asf/flume/repo
Commit: http://git-wip-us.apache.org/repos/asf/flume/commit/d3c85b1d
Tree: http://git-wip-us.apache.org/repos/asf/flume/tree/d3c85b1d
Diff: http://git-wip-us.apache.org/repos/asf/flume/diff/d3c85b1d
Branch: refs/heads/trunk
Commit: d3c85b1d77584205b82b13126330cf17ce0f5ed9
Parents: 680c2d7
Author: Mike Percy <mp...@cloudera.com>
Authored: Fri Sep 7 17:04:53 2012 -0700
Committer: Mike Percy <mp...@cloudera.com>
Committed: Fri Sep 7 17:04:53 2012 -0700
----------------------------------------------------------------------
.../flume/channel/file/CheckpointRebuilder.java | 2 +-
.../org/apache/flume/channel/file/FileChannel.java | 32 ++-
.../java/org/apache/flume/channel/file/Log.java | 70 +++-
.../org/apache/flume/channel/file/LogFile.java | 67 +++-
.../apache/flume/channel/file/LogFileFactory.java | 21 +-
.../org/apache/flume/channel/file/LogFileV2.java | 12 +-
.../org/apache/flume/channel/file/LogFileV3.java | 86 ++++-
.../apache/flume/channel/file/ReplayHandler.java | 20 +-
.../flume/channel/file/TransactionEventRecord.java | 37 ++-
.../file/encryption/AESCTRNoPaddingProvider.java | 124 ++++++
.../channel/file/encryption/CipherProvider.java | 68 ++++
.../file/encryption/CipherProviderFactory.java | 89 +++++
.../file/encryption/CipherProviderType.java | 35 ++
.../file/encryption/EncryptionConfiguration.java | 60 +++
.../file/encryption/JCEFileKeyProvider.java | 109 ++++++
.../flume/channel/file/encryption/KeyProvider.java | 32 ++
.../file/encryption/KeyProviderFactory.java | 82 ++++
.../channel/file/encryption/KeyProviderType.java | 35 ++
.../src/main/proto/filechannel.proto | 7 +
.../apache/flume/channel/file/TestFileChannel.java | 21 +-
.../org/apache/flume/channel/file/TestLogFile.java | 11 +-
.../channel/file/TestTransactionEventRecordV3.java | 17 +-
.../org/apache/flume/channel/file/TestUtils.java | 31 ++
.../file/encryption/CipherProviderTestSuite.java | 69 ++++
.../file/encryption/EncryptionTestUtils.java | 124 ++++++
.../encryption/TestAESCTRNoPaddingProvider.java | 49 +++
.../file/encryption/TestFileChannelEncryption.java | 298 +++++++++++++++
.../file/encryption/TestJCEFileKeyProvider.java | 112 ++++++
.../src/test/resources/test.keystore | Bin 0 -> 980 bytes
29 files changed, 1623 insertions(+), 97 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/CheckpointRebuilder.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/CheckpointRebuilder.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/CheckpointRebuilder.java
index 1ba5d1c..6e1d2fc 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/CheckpointRebuilder.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/CheckpointRebuilder.java
@@ -69,7 +69,7 @@ public class CheckpointRebuilder {
LOG.info("Attempting to fast replay the log files.");
List<LogFile.SequentialReader> logReaders = Lists.newArrayList();
for (File logFile : logFiles) {
- logReaders.add(LogFileFactory.getSequentialReader(logFile));
+ logReaders.add(LogFileFactory.getSequentialReader(logFile, null));
}
long transactionIDSeed = 0;
long writeOrderIDSeed = 0;
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
index 5d588ea..f2ccd3a 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
@@ -33,6 +33,9 @@ import org.apache.flume.Event;
import org.apache.flume.channel.BasicChannelSemantics;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.flume.channel.file.Log.Builder;
+import org.apache.flume.channel.file.encryption.EncryptionConfiguration;
+import org.apache.flume.channel.file.encryption.KeyProvider;
+import org.apache.flume.channel.file.encryption.KeyProviderFactory;
import org.apache.flume.instrumentation.ChannelCounter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,6 +90,9 @@ public class FileChannel extends BasicChannelSemantics {
private ChannelCounter channelCounter;
private boolean useLogReplayV1;
private boolean useFastReplay = false;
+ private KeyProvider encryptionKeyProvider;
+ private String encryptionKeyAlias;
+ private String encryptionCipherProvider;
@Override
public synchronized void setName(String name) {
@@ -198,6 +204,27 @@ public class FileChannel extends BasicChannelSemantics {
FileChannelConfiguration.USE_FAST_REPLAY,
FileChannelConfiguration.DEFAULT_USE_FAST_REPLAY);
+ Context encryptionContext = new Context(
+ context.getSubProperties(EncryptionConfiguration.ENCRYPTION_PREFIX +
+ "."));
+ String encryptionKeyProviderName = encryptionContext.getString(
+ EncryptionConfiguration.KEY_PROVIDER);
+ encryptionKeyAlias = encryptionContext.getString(
+ EncryptionConfiguration.KEY_ALIAS);
+ encryptionCipherProvider = encryptionContext.getString(
+ EncryptionConfiguration.CIPHER_PROVIDER);
+ if(encryptionKeyProviderName != null) {
+ Preconditions.checkNotNull(encryptionKeyAlias, "encryptionKeyAlias");
+ Preconditions.checkNotNull(encryptionCipherProvider,
+ "encryptionCipherProvider");
+ encryptionKeyProvider = KeyProviderFactory.
+ getInstance(encryptionKeyProviderName, encryptionContext);
+ } else {
+ Preconditions.checkState(encryptionKeyAlias == null, "encryptionKeyAlias");
+ Preconditions.checkState(encryptionCipherProvider == null,
+ "encryptionCipherProvider");
+ }
+
if(queueRemaining == null) {
queueRemaining = new Semaphore(capacity, true);
}
@@ -226,6 +253,9 @@ public class FileChannel extends BasicChannelSemantics {
builder.setCheckpointWriteTimeout(checkpointWriteTimeout);
builder.setUseLogReplayV1(useLogReplayV1);
builder.setUseFastReplay(useFastReplay);
+ builder.setEncryptionKeyProvider(encryptionKeyProvider);
+ builder.setEncryptionKeyAlias(encryptionKeyAlias);
+ builder.setEncryptionCipherProvider(encryptionCipherProvider);
log = builder.build();
log.replay();
open = true;
@@ -300,7 +330,7 @@ public class FileChannel extends BasicChannelSemantics {
}
}
- boolean isOpen() {
+ public boolean isOpen() {
return open;
}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
index 6e8e3d0..f834148 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
@@ -24,6 +24,7 @@ import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileLock;
import java.nio.channels.OverlappingFileLockException;
+import java.security.Key;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
@@ -39,7 +40,10 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
+import javax.annotation.Nullable;
+
import org.apache.flume.Event;
+import org.apache.flume.channel.file.encryption.KeyProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,6 +98,10 @@ class Log {
private final String channelNameDescriptor;
private int checkpointWriteTimeout;
private boolean useLogReplayV1;
+ private KeyProvider encryptionKeyProvider;
+ private String encryptionCipherProvider;
+ private String encryptionKeyAlias;
+ private Key encryptionKey;
static class Builder {
private long bCheckpointInterval;
@@ -108,6 +116,9 @@ class Log {
FileChannelConfiguration.DEFAULT_CHECKPOINT_WRITE_TIMEOUT;
private boolean useLogReplayV1;
private boolean useFastReplay;
+ private KeyProvider bEncryptionKeyProvider;
+ private String bEncryptionKeyAlias;
+ private String bEncryptionCipherProvider;
Builder setCheckpointInterval(long interval) {
bCheckpointInterval = interval;
@@ -148,6 +159,7 @@ class Log {
bCheckpointWriteTimeout = checkpointTimeout;
return this;
}
+
Builder setUseLogReplayV1(boolean useLogReplayV1){
this.useLogReplayV1 = useLogReplayV1;
return this;
@@ -158,17 +170,35 @@ class Log {
return this;
}
+ Builder setEncryptionKeyProvider(KeyProvider encryptionKeyProvider) {
+ bEncryptionKeyProvider = encryptionKeyProvider;
+ return this;
+ }
+
+ Builder setEncryptionKeyAlias(String encryptionKeyAlias) {
+ bEncryptionKeyAlias = encryptionKeyAlias;
+ return this;
+ }
+
+ Builder setEncryptionCipherProvider(String encryptionCipherProvider) {
+ bEncryptionCipherProvider = encryptionCipherProvider;
+ return this;
+ }
+
Log build() throws IOException {
return new Log(bCheckpointInterval, bMaxFileSize, bQueueCapacity,
bLogWriteTimeout, bCheckpointWriteTimeout, bCheckpointDir, bName,
- useLogReplayV1, useFastReplay, bLogDirs);
+ useLogReplayV1, useFastReplay, bEncryptionKeyProvider,
+ bEncryptionKeyAlias, bEncryptionCipherProvider, bLogDirs);
}
}
private Log(long checkpointInterval, long maxFileSize, int queueCapacity,
int logWriteTimeout, int checkpointWriteTimeout, File checkpointDir,
String name, boolean useLogReplayV1, boolean useFastReplay,
- File... logDirs)
+ @Nullable KeyProvider encryptionKeyProvider,
+ @Nullable String encryptionKeyAlias,
+ @Nullable String encryptionCipherProvider, File... logDirs)
throws IOException {
Preconditions.checkArgument(checkpointInterval > 0,
"checkpointInterval <= 0");
@@ -203,6 +233,25 @@ class Log {
}
throw e;
}
+ if(encryptionKeyProvider != null && encryptionKeyAlias != null &&
+ encryptionCipherProvider != null) {
+ LOGGER.info("Encryption is enabled with encryptionKeyProvider = " +
+ encryptionKeyProvider + ", encryptionKeyAlias = " + encryptionKeyAlias
+ + ", encryptionCipherProvider = " + encryptionCipherProvider);
+ this.encryptionKeyProvider = encryptionKeyProvider;
+ this.encryptionKeyAlias = encryptionKeyAlias;
+ this.encryptionCipherProvider = encryptionCipherProvider;
+ this.encryptionKey = encryptionKeyProvider.getKey(encryptionKeyAlias);
+ } else if (encryptionKeyProvider == null && encryptionKeyAlias == null &&
+ encryptionCipherProvider == null) {
+ LOGGER.info("Encryption is not enabled");
+ } else {
+ throw new IllegalArgumentException("Encryption configuration must all " +
+ "null or all not null: encryptionKeyProvider = " +
+ encryptionKeyProvider + ", encryptionKeyAlias = " +
+ encryptionKeyAlias + ", encryptionCipherProvider = " +
+ encryptionCipherProvider);
+ }
open = false;
this.checkpointInterval = checkpointInterval;
this.maxFileSize = maxFileSize;
@@ -247,8 +296,8 @@ class Log {
int id = LogUtils.getIDForFile(file);
dataFiles.add(file);
nextFileID.set(Math.max(nextFileID.get(), id));
- idLogFileMap.put(id, LogFileFactory.getRandomReader(new File(logDir, PREFIX
- + id)));
+ idLogFileMap.put(id, LogFileFactory.getRandomReader(new File(logDir,
+ PREFIX + id), encryptionKeyProvider));
}
}
LOGGER.info("Found NextFileID " + nextFileID +
@@ -280,8 +329,8 @@ class Log {
* the queue, the timestamp the queue was written to disk, and
* the list of data files.
*/
- ReplayHandler replayHandler = new ReplayHandler(queue, useFastReplay,
- checkpointFile);
+ ReplayHandler replayHandler = new ReplayHandler(queue,
+ encryptionKeyProvider, useFastReplay, checkpointFile);
if(useLogReplayV1) {
LOGGER.info("Replaying logs with v1 replay logic");
replayHandler.replayLogv1(dataFiles);
@@ -632,9 +681,12 @@ class Log {
"File already exists " + file);
Preconditions.checkState(file.createNewFile(),
"File could not be created " + file);
- idLogFileMap.put(fileID, LogFileFactory.getRandomReader(file));
+ idLogFileMap.put(fileID, LogFileFactory.getRandomReader(file,
+ encryptionKeyProvider));
// writer from this point on will get new reference
- logFiles.set(index, LogFileFactory.getWriter(file, fileID, maxFileSize));
+ logFiles.set(index, LogFileFactory.getWriter(file, fileID,
+ maxFileSize, encryptionKey, encryptionKeyAlias,
+ encryptionCipherProvider));
// close out old log
if (oldLogFile != null) {
oldLogFile.close();
@@ -713,7 +765,7 @@ class Log {
} finally {
writer.close();
}
- reader = LogFileFactory.getRandomReader(file);
+ reader = LogFileFactory.getRandomReader(file, encryptionKeyProvider);
idLogFileMap.put(id, reader);
LOGGER.debug("Updated checkpoint for file: " + file
+ "logWriteOrderID " + logWriteOrderID);
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java
index c4e0424..8071140 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java
@@ -28,6 +28,10 @@ import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
+import javax.annotation.Nullable;
+
+import org.apache.flume.channel.file.encryption.CipherProvider;
+import org.apache.flume.channel.file.encryption.KeyProvider;
import org.apache.flume.tools.DirectMemoryUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -118,14 +122,17 @@ abstract class LogFile {
private final long maxFileSize;
private final RandomAccessFile writeFileHandle;
private final FileChannel writeFileChannel;
+ private final CipherProvider.Encryptor encryptor;
private volatile boolean open;
- Writer(File file, int logFileID, long maxFileSize)
+ Writer(File file, int logFileID, long maxFileSize,
+ CipherProvider.Encryptor encryptor)
throws IOException {
this.file = file;
this.logFileID = logFileID;
this.maxFileSize = Math.min(maxFileSize,
FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE);
+ this.encryptor = encryptor;
writeFileHandle = new RandomAccessFile(file, "rw");
writeFileChannel = writeFileHandle.getChannel();
LOG.info("Opened " + file);
@@ -134,6 +141,9 @@ abstract class LogFile {
abstract int getVersion();
+ protected CipherProvider.Encryptor getEncryptor() {
+ return encryptor;
+ }
int getLogFileID() {
return logFileID;
}
@@ -151,38 +161,51 @@ abstract class LogFile {
return getFileChannel().position();
}
synchronized FlumeEventPointer put(ByteBuffer buffer) throws IOException {
+ if(encryptor != null) {
+ buffer = ByteBuffer.wrap(encryptor.encrypt(buffer.array()));
+ }
Pair<Integer, Integer> pair = write(buffer);
return new FlumeEventPointer(pair.getLeft(), pair.getRight());
}
synchronized void take(ByteBuffer buffer) throws IOException {
+ if(encryptor != null) {
+ buffer = ByteBuffer.wrap(encryptor.encrypt(buffer.array()));
+ }
write(buffer);
}
synchronized void rollback(ByteBuffer buffer) throws IOException {
+ if(encryptor != null) {
+ buffer = ByteBuffer.wrap(encryptor.encrypt(buffer.array()));
+ }
write(buffer);
}
synchronized void commit(ByteBuffer buffer) throws IOException {
+ if(encryptor != null) {
+ buffer = ByteBuffer.wrap(encryptor.encrypt(buffer.array()));
+ }
write(buffer);
sync();
}
private Pair<Integer, Integer> write(ByteBuffer buffer) throws IOException {
Preconditions.checkState(isOpen(), "File closed");
long length = position();
- long expectedLength = length + (long) buffer.capacity();
+ long expectedLength = length + (long) buffer.limit();
Preconditions.checkArgument(expectedLength < (long) Integer.MAX_VALUE);
int offset = (int)length;
Preconditions.checkState(offset >= 0, String.valueOf(offset));
- int recordLength = 1 + buffer.capacity();
+ // OP_RECORD + size + buffer
+ int recordLength = 1 + (int)Serialization.SIZE_OF_INT + buffer.limit();
preallocate(recordLength);
ByteBuffer toWrite = ByteBuffer.allocate(recordLength);
toWrite.put(OP_RECORD);
- toWrite.put(buffer);
+ writeDelimitedBuffer(toWrite, buffer);
toWrite.position(0);
int wrote = getFileChannel().write(toWrite);
Preconditions.checkState(wrote == toWrite.limit());
return Pair.of(getLogFileID(), offset);
}
synchronized boolean isRollRequired(ByteBuffer buffer) throws IOException {
- return isOpen() && position() + (long) buffer.capacity() > getMaxSize();
+ return isOpen() && position() + (long) buffer.limit() > getMaxSize();
}
private void sync() throws IOException {
Preconditions.checkState(isOpen(), "File closed");
@@ -233,10 +256,12 @@ abstract class LogFile {
private final File file;
private final BlockingQueue<RandomAccessFile> readFileHandles =
new ArrayBlockingQueue<RandomAccessFile>(50, true);
-
+ private final KeyProvider encryptionKeyProvider;
private volatile boolean open;
- public RandomReader(File file) throws IOException {
+ public RandomReader(File file, @Nullable KeyProvider encryptionKeyProvider)
+ throws IOException {
this.file = file;
+ this.encryptionKeyProvider = encryptionKeyProvider;
readFileHandles.add(open());
open = true;
}
@@ -250,6 +275,10 @@ abstract class LogFile {
return file;
}
+ protected KeyProvider getKeyProvider() {
+ return encryptionKeyProvider;
+ }
+
FlumeEvent get(int offset) throws IOException, InterruptedException {
Preconditions.checkState(open, "File closed");
RandomAccessFile fileHandle = checkOut();
@@ -340,6 +369,7 @@ abstract class LogFile {
private final RandomAccessFile fileHandle;
private final FileChannel fileChannel;
private final File file;
+ private final KeyProvider encryptionKeyProvider;
private int logFileID;
private long lastCheckpointPosition;
@@ -351,8 +381,10 @@ abstract class LogFile {
* @throws IOException if an I/O error occurs
* @throws EOFException if the file is empty
*/
- SequentialReader(File file) throws IOException, EOFException {
+ SequentialReader(File file, @Nullable KeyProvider encryptionKeyProvider)
+ throws IOException, EOFException {
this.file = file;
+ this.encryptionKeyProvider = encryptionKeyProvider;
fileHandle = new RandomAccessFile(file, "r");
fileChannel = fileHandle.getChannel();
}
@@ -372,6 +404,9 @@ abstract class LogFile {
+ Integer.toHexString(logFileID));
}
+ protected KeyProvider getKeyProvider() {
+ return encryptionKeyProvider;
+ }
protected RandomAccessFile getFileHandle() {
return fileHandle;
}
@@ -431,11 +466,25 @@ abstract class LogFile {
}
}
}
+ protected static void writeDelimitedBuffer(ByteBuffer output, ByteBuffer buffer)
+ throws IOException {
+ output.putInt(buffer.limit());
+ output.put(buffer);
+ }
+ protected static byte[] readDelimitedBuffer(RandomAccessFile fileHandle)
+ throws IOException {
+ int length = fileHandle.readInt();
+ Preconditions.checkState(length >= 0, Integer.toHexString(length));
+ byte[] buffer = new byte[length];
+ fileHandle.readFully(buffer);
+ return buffer;
+ }
+
public static void main(String[] args) throws EOFException, IOException {
File file = new File(args[0]);
LogFile.SequentialReader reader = null;
try {
- reader = LogFileFactory.getSequentialReader(file);
+ reader = LogFileFactory.getSequentialReader(file, null);
LogRecord entry;
FlumeEventPointer ptr;
// for puts the fileId is the fileID of the file they exist in
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileFactory.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileFactory.java
index 8de37b8..7bf6de4 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileFactory.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileFactory.java
@@ -21,7 +21,11 @@ package org.apache.flume.channel.file;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
+import java.security.Key;
+import javax.annotation.Nullable;
+
+import org.apache.flume.channel.file.encryption.KeyProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,14 +62,18 @@ class LogFileFactory {
}
static LogFile.Writer getWriter(File file, int logFileID,
- long maxFileSize) throws IOException {
+ long maxFileSize, @Nullable Key encryptionKey,
+ @Nullable String encryptionKeyAlias,
+ @Nullable String encryptionCipherProvider) throws IOException {
if(!(file.exists() || file.createNewFile())) {
throw new IOException("Cannot create " + file);
}
- return new LogFileV3.Writer(file, logFileID, maxFileSize);
+ return new LogFileV3.Writer(file, logFileID, maxFileSize, encryptionKey,
+ encryptionKeyAlias, encryptionCipherProvider);
}
- static LogFile.RandomReader getRandomReader(File file)
+ static LogFile.RandomReader getRandomReader(File file,
+ @Nullable KeyProvider encryptionKeyProvider)
throws IOException {
RandomAccessFile logFile = new RandomAccessFile(file, "r");
try {
@@ -73,7 +81,7 @@ class LogFileFactory {
// either this is a rr for a just created file or
// the metadata file exists and as such it's V3
if(logFile.length() == 0L || metaDataFile.exists()) {
- return new LogFileV3.RandomReader(file);
+ return new LogFileV3.RandomReader(file, encryptionKeyProvider);
}
int version = logFile.readInt();
if(Serialization.VERSION_2 == version) {
@@ -92,13 +100,14 @@ class LogFileFactory {
}
}
- static LogFile.SequentialReader getSequentialReader(File file)
+ static LogFile.SequentialReader getSequentialReader(File file,
+ @Nullable KeyProvider encryptionKeyProvider)
throws IOException {
RandomAccessFile logFile = null;
try {
File metaDataFile = Serialization.getMetaDataFile(file);
if(metaDataFile.exists()) {
- return new LogFileV3.SequentialReader(file);
+ return new LogFileV3.SequentialReader(file, encryptionKeyProvider);
}
logFile = new RandomAccessFile(file, "r");
int version = logFile.readInt();
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV2.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV2.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV2.java
index c81b290..4c593a4 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV2.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV2.java
@@ -95,7 +95,7 @@ class LogFileV2 extends LogFile {
Writer(File file, int logFileID, long maxFileSize)
throws IOException {
- super(file, logFileID, maxFileSize);
+ super(file, logFileID, maxFileSize, null);
RandomAccessFile writeFileHandle = getFileHandle();
writeFileHandle.writeInt(getVersion());
writeFileHandle.writeInt(logFileID);
@@ -113,8 +113,9 @@ class LogFileV2 extends LogFile {
}
static class RandomReader extends LogFile.RandomReader {
- RandomReader(File file) throws IOException {
- super(file);
+ RandomReader(File file)
+ throws IOException {
+ super(file, null);
}
@Override
int getVersion() {
@@ -129,8 +130,9 @@ class LogFileV2 extends LogFile {
static class SequentialReader extends LogFile.SequentialReader {
- SequentialReader(File file) throws EOFException, IOException {
- super(file);
+ SequentialReader(File file)
+ throws EOFException, IOException {
+ super(file, null);
RandomAccessFile fileHandle = getFileHandle();
int version = fileHandle.readInt();
if(version != getVersion()) {
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV3.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV3.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV3.java
index 502dfbd..414d6d9 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV3.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFileV3.java
@@ -23,14 +23,21 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.InputStream;
import java.io.RandomAccessFile;
-import java.nio.channels.Channels;
+import java.security.Key;
+
+import javax.annotation.Nullable;
import org.apache.flume.chanel.file.proto.ProtosFactory;
+import org.apache.flume.channel.file.encryption.CipherProvider;
+import org.apache.flume.channel.file.encryption.CipherProviderFactory;
+import org.apache.flume.channel.file.encryption.KeyProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Preconditions;
+import com.google.protobuf.ByteString;
+
/**
* Represents a single data file on disk. Has methods to write,
* read sequentially (replay), and read randomly (channel takes).
@@ -42,7 +49,7 @@ class LogFileV3 extends LogFile {
private LogFileV3() {}
static class MetaDataWriter extends LogFile.MetaDataWriter {
- private final ProtosFactory.LogFileMetaData logFileMetaData;
+ private ProtosFactory.LogFileMetaData logFileMetaData;
private final File metaDataFile;
protected MetaDataWriter(File logFile, int logFileID) throws IOException {
super(logFile, logFileID);
@@ -71,11 +78,12 @@ class LogFileV3 extends LogFile {
ProtosFactory.LogFileMetaData.newBuilder(logFileMetaData);
metaDataBuilder.setCheckpointPosition(currentPosition);
metaDataBuilder.setCheckpointWriteOrderID(logWriteOrderID);
+ logFileMetaData = metaDataBuilder.build();
LOGGER.info("Updating " + metaDataFile.getName() + " currentPosition = "
+ currentPosition + ", logWriteOrderID = " + logWriteOrderID);
FileOutputStream outputStream = new FileOutputStream(metaDataFile);
try {
- metaDataBuilder.build().writeDelimitedTo(outputStream);
+ logFileMetaData.writeDelimitedTo(outputStream);
outputStream.getChannel().force(true);
} finally {
try {
@@ -119,11 +127,27 @@ class LogFileV3 extends LogFile {
}
static class Writer extends LogFile.Writer {
- Writer(File file, int logFileID, long maxFileSize)
+ Writer(File file, int logFileID, long maxFileSize,
+ @Nullable Key encryptionKey,
+ @Nullable String encryptionKeyAlias,
+ @Nullable String encryptionCipherProvider)
throws IOException {
- super(file, logFileID, maxFileSize);
+ super(file, logFileID, maxFileSize, CipherProviderFactory.
+ getEncrypter(encryptionCipherProvider, encryptionKey));
ProtosFactory.LogFileMetaData.Builder metaDataBuilder =
ProtosFactory.LogFileMetaData.newBuilder();
+ if(encryptionKey != null) {
+ Preconditions.checkNotNull(encryptionKeyAlias, "encryptionKeyAlias");
+ Preconditions.checkNotNull(encryptionCipherProvider,
+ "encryptionCipherProvider");
+ ProtosFactory.LogFileEncryption.Builder logFileEncryptionBuilder =
+ ProtosFactory.LogFileEncryption.newBuilder();
+ logFileEncryptionBuilder.setCipherProvider(encryptionCipherProvider);
+ logFileEncryptionBuilder.setKeyAlias(encryptionKeyAlias);
+ logFileEncryptionBuilder.setParameters(
+ ByteString.copyFrom(getEncryptor().getParameters()));
+ metaDataBuilder.setEncryption(logFileEncryptionBuilder);
+ }
metaDataBuilder.setVersion(getVersion());
metaDataBuilder.setLogFileID(logFileID);
metaDataBuilder.setCheckpointPosition(0L);
@@ -149,8 +173,10 @@ class LogFileV3 extends LogFile {
static class RandomReader extends LogFile.RandomReader {
private volatile boolean initialized;
- RandomReader(File file) throws IOException {
- super(file);
+ private CipherProvider.Decryptor decryptor;
+ RandomReader(File file, @Nullable KeyProvider encryptionKeyProvider)
+ throws IOException {
+ super(file, encryptionKeyProvider);
}
private void initialize() throws IOException {
File metaDataFile = Serialization.getMetaDataFile(getFile());
@@ -164,6 +190,17 @@ class LogFileV3 extends LogFile {
" expected " + Integer.toHexString(getVersion())
+ " file: " + getFile().getCanonicalPath());
}
+ if(metaData.hasEncryption()) {
+ if(getKeyProvider() == null) {
+ throw new IllegalStateException("Data file is encrypted but no " +
+ " provider was specified");
+ }
+ ProtosFactory.LogFileEncryption encryption = metaData.getEncryption();
+ Key key = getKeyProvider().getKey(encryption.getKeyAlias());
+ decryptor = CipherProviderFactory.
+ getDecrypter(encryption.getCipherProvider(), key,
+ encryption.getParameters().toByteArray());
+ }
} finally {
try {
inputStream.close();
@@ -186,16 +223,19 @@ class LogFileV3 extends LogFile {
initialized = true;
initialize();
}
- return TransactionEventRecord.
- fromInputStream(Channels.newInputStream(fileHandle.getChannel()));
+ byte[] buffer = readDelimitedBuffer(fileHandle);
+ if(decryptor != null) {
+ buffer = decryptor.decrypt(buffer);
+ }
+ return TransactionEventRecord.fromByteArray(buffer);
}
}
static class SequentialReader extends LogFile.SequentialReader {
-
- private InputStream inputStream;
- SequentialReader(File file) throws EOFException, IOException {
- super(file);
+ private CipherProvider.Decryptor decryptor;
+ SequentialReader(File file, @Nullable KeyProvider encryptionKeyProvider)
+ throws EOFException, IOException {
+ super(file, encryptionKeyProvider);
File metaDataFile = Serialization.getMetaDataFile(file);
FileInputStream inputStream = new FileInputStream(metaDataFile);
try {
@@ -207,6 +247,17 @@ class LogFileV3 extends LogFile {
" expected " + Integer.toHexString(getVersion())
+ " file: " + file.getCanonicalPath());
}
+ if(metaData.hasEncryption()) {
+ if(getKeyProvider() == null) {
+ throw new IllegalStateException("Data file is encrypted but no " +
+ " provider was specified");
+ }
+ ProtosFactory.LogFileEncryption encryption = metaData.getEncryption();
+ Key key = getKeyProvider().getKey(encryption.getKeyAlias());
+ decryptor = CipherProviderFactory.
+ getDecrypter(encryption.getCipherProvider(), key,
+ encryption.getParameters().toByteArray());
+ }
setLogFileID(metaData.getLogFileID());
setLastCheckpointPosition(metaData.getCheckpointPosition());
setLastCheckpointWriteOrderID(metaData.getCheckpointWriteOrderID());
@@ -217,7 +268,6 @@ class LogFileV3 extends LogFile {
LOGGER.warn("Unable to close " + metaDataFile, e);
}
}
- this.inputStream = Channels.newInputStream(getFileHandle().getChannel());
}
@Override
@@ -226,8 +276,12 @@ class LogFileV3 extends LogFile {
}
@Override
LogRecord doNext(int offset) throws IOException {
+ byte[] buffer = readDelimitedBuffer(getFileHandle());
+ if(decryptor != null) {
+ buffer = decryptor.decrypt(buffer);
+ }
TransactionEventRecord event =
- TransactionEventRecord.fromInputStream(inputStream);
+ TransactionEventRecord.fromByteArray(buffer);
return new LogRecord(getLogFileID(), offset, event);
}
}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/ReplayHandler.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/ReplayHandler.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/ReplayHandler.java
index 17550ec..7c32526 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/ReplayHandler.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/ReplayHandler.java
@@ -22,10 +22,17 @@ import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
+import java.util.Set;
+
+import javax.annotation.Nullable;
+import org.apache.commons.collections.MultiMap;
+import org.apache.commons.collections.map.MultiValueMap;
+import org.apache.flume.channel.file.encryption.KeyProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,10 +41,6 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
-import java.util.Iterator;
-import java.util.Set;
-import org.apache.commons.collections.MultiMap;
-import org.apache.commons.collections.map.MultiValueMap;
/**
* Processes a set of data logs, replaying said logs into the queue.
@@ -49,6 +52,7 @@ class ReplayHandler {
private final long lastCheckpoint;
private final Map<Integer, LogFile.SequentialReader> readers;
private final PriorityQueue<LogRecord> logRecordBuffer;
+ private final KeyProvider encryptionKeyProvider;
/**
* This data structure stores takes for which we found a commit in the log
* files before we found a commit for the put. This can happen if the channel
@@ -68,7 +72,8 @@ class ReplayHandler {
private final boolean useFastReplay;
private final File cpDir;
- ReplayHandler(FlumeEventQueue queue, boolean useFastReplay, File cpDir) {
+ ReplayHandler(FlumeEventQueue queue, @Nullable KeyProvider encryptionKeyProvider,
+ boolean useFastReplay, File cpDir) {
this.queue = queue;
this.useFastReplay = useFastReplay;
this.lastCheckpoint = queue.getLogWriteOrderID();
@@ -76,6 +81,7 @@ class ReplayHandler {
pendingTakes = Lists.newArrayList();
readers = Maps.newHashMap();
logRecordBuffer = new PriorityQueue<LogRecord>();
+ this.encryptionKeyProvider = encryptionKeyProvider;
}
/**
* Replay logic from Flume1.2 which can be activated if the v2 logic
@@ -109,7 +115,7 @@ class ReplayHandler {
LOG.info("Replaying " + log);
LogFile.SequentialReader reader = null;
try {
- reader = LogFileFactory.getSequentialReader(log);
+ reader = LogFileFactory.getSequentialReader(log, encryptionKeyProvider);
reader.skipToLastCheckpointPosition(queue.getLogWriteOrderID());
LogRecord entry;
FlumeEventPointer ptr;
@@ -250,7 +256,7 @@ class ReplayHandler {
LOG.info("Replaying " + log);
try {
LogFile.SequentialReader reader =
- LogFileFactory.getSequentialReader(log);
+ LogFileFactory.getSequentialReader(log, encryptionKeyProvider);
reader.skipToLastCheckpointPosition(queue.getLogWriteOrderID());
Preconditions.checkState(!readers.containsKey(reader.getLogFileID()),
"Readers " + readers + " already contains "
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/TransactionEventRecord.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/TransactionEventRecord.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/TransactionEventRecord.java
index 0488e90..f78147f 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/TransactionEventRecord.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/TransactionEventRecord.java
@@ -18,6 +18,7 @@
*/
package org.apache.flume.channel.file;
+import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataOutput;
@@ -184,20 +185,30 @@ abstract class TransactionEventRecord implements Writable {
}
}
- static TransactionEventRecord fromInputStream(InputStream in)
+
+ static TransactionEventRecord fromByteArray(byte[] buffer)
throws IOException {
- ProtosFactory.TransactionEventHeader header =
- ProtosFactory.TransactionEventHeader.parseDelimitedFrom(in);
- short type = (short)header.getType();
- long transactionID = header.getTransactionID();
- long writeOrderID = header.getWriteOrderID();
- TransactionEventRecord transactionEvent =
- newRecordForType(type, transactionID, writeOrderID);
- transactionEvent.readProtos(in);
- @SuppressWarnings("unused")
- ProtosFactory.TransactionEventFooter footer =
- ProtosFactory.TransactionEventFooter.parseDelimitedFrom(in);
- return transactionEvent;
+ ByteArrayInputStream in = new ByteArrayInputStream(buffer);
+ try {
+ ProtosFactory.TransactionEventHeader header =
+ ProtosFactory.TransactionEventHeader.parseDelimitedFrom(in);
+ short type = (short)header.getType();
+ long transactionID = header.getTransactionID();
+ long writeOrderID = header.getWriteOrderID();
+ TransactionEventRecord transactionEvent =
+ newRecordForType(type, transactionID, writeOrderID);
+ transactionEvent.readProtos(in);
+ @SuppressWarnings("unused")
+ ProtosFactory.TransactionEventFooter footer =
+ ProtosFactory.TransactionEventFooter.parseDelimitedFrom(in);
+ return transactionEvent;
+ } finally {
+ try {
+ in.close();
+ } catch (IOException e) {
+ LOG.warn("Error closing byte array input stream", e);
+ }
+ }
}
static String getName(short type) {
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/AESCTRNoPaddingProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/AESCTRNoPaddingProvider.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/AESCTRNoPaddingProvider.java
new file mode 100644
index 0000000..b77e10c
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/AESCTRNoPaddingProvider.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.nio.ByteBuffer;
+import java.security.Key;
+import java.security.SecureRandom;
+
+import javax.crypto.Cipher;
+import javax.crypto.spec.IvParameterSpec;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Throwables;
+
+public class AESCTRNoPaddingProvider extends CipherProvider {
+ private static final Logger LOG = LoggerFactory
+ .getLogger(AESCTRNoPaddingProvider.class);
+ static final String TYPE = "AES/CTR/NoPadding";
+
+
+ public Encryptor.Builder<AESCTRNoPaddingEncryptor> newEncryptorBuilder() {
+ return new EncryptorBuilder();
+ }
+ public Decryptor.Builder<AESCTRNoPaddingDecryptor> newDecryptorBuilder() {
+ return new DecryptorBuilder();
+ }
+
+ public static class EncryptorBuilder
+ extends CipherProvider.Encryptor.Builder<AESCTRNoPaddingEncryptor> {
+ @Override
+ public AESCTRNoPaddingEncryptor build() {
+ ByteBuffer buffer = ByteBuffer.allocate(16);
+ byte[] seed = new byte[12];
+ SecureRandom random = new SecureRandom();
+ random.nextBytes(seed);
+ buffer.put(seed).putInt(1);
+ return new AESCTRNoPaddingEncryptor(key, buffer.array());
+ }
+ }
+
+
+ public static class DecryptorBuilder
+ extends CipherProvider.Decryptor.Builder<AESCTRNoPaddingDecryptor> {
+ @Override
+ public AESCTRNoPaddingDecryptor build() {
+ return new AESCTRNoPaddingDecryptor(key, parameters);
+ }
+ }
+
+ private static class AESCTRNoPaddingEncryptor extends Encryptor {
+ private byte[] parameters;
+ private Cipher cipher;
+ private AESCTRNoPaddingEncryptor(Key key, byte[] parameters) {
+ this.parameters = parameters;
+ cipher = getCipher(key, Cipher.ENCRYPT_MODE, parameters);
+ }
+ @Override
+ public byte[] getParameters() {
+ return parameters;
+ }
+ @Override
+ public String getCodec() {
+ return TYPE;
+ }
+ @Override
+ public byte[] encrypt(byte[] clearText) {
+ return doFinal(cipher, clearText);
+ }
+ }
+
+ private static class AESCTRNoPaddingDecryptor extends Decryptor {
+ private Cipher cipher;
+ private AESCTRNoPaddingDecryptor(Key key, byte[] parameters) {
+ cipher = getCipher(key, Cipher.DECRYPT_MODE, parameters);
+ }
+ @Override
+ public byte[] decrypt(byte[] cipherText) {
+ return doFinal(cipher, cipherText);
+ }
+ @Override
+ public String getCodec() {
+ return TYPE;
+ }
+ }
+
+ private static byte[] doFinal(Cipher cipher, byte[] input) {
+ try {
+ return cipher.doFinal(input);
+ } catch (Exception e) {
+ String msg = "Unable to encrypt or decrypt data " + TYPE
+ + " input.length " + input.length;
+ LOG.error(msg, e);
+ throw Throwables.propagate(e);
+ }
+ }
+ private static Cipher getCipher(Key key, int mode, byte[] parameters) {
+ try {
+ Cipher cipher = Cipher.getInstance(TYPE);
+ cipher.init(mode, key, new IvParameterSpec(parameters));
+ return cipher;
+ } catch (Exception e) {
+ LOG.error("Unable to instaniate " + TYPE, e);
+ throw Throwables.propagate(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProvider.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProvider.java
new file mode 100644
index 0000000..3b8d6c3
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProvider.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.security.Key;
+
+public abstract class CipherProvider {
+
+ public abstract Encryptor.Builder<?> newEncryptorBuilder();
+ public abstract Decryptor.Builder<?> newDecryptorBuilder();
+
+ public static abstract class Encryptor {
+
+ public abstract byte[] encrypt(byte[] clearText);
+ public abstract byte[] getParameters();
+ public abstract String getCodec();
+
+ /** Builder implementations MUST have a no-arg constructor */
+ public static abstract class Builder<T extends Encryptor> {
+ protected Key key;
+ public Builder<T> setKey(Key key) {
+ this.key = key;
+ return this;
+ }
+ public abstract T build();
+ }
+
+ }
+
+ public static abstract class Decryptor {
+ public abstract byte[] decrypt(byte[] cipherText);
+ public abstract String getCodec();
+
+ /** Builder implementations MUST have a no-arg constructor */
+ public static abstract class Builder<T extends Decryptor> {
+ protected byte[] parameters;
+ protected Key key;
+ public Builder<T> setKey(Key key) {
+ this.key = key;
+ return this;
+ }
+ public Builder<T> setParameters(byte[] parameters) {
+ this.parameters = parameters;
+ return this;
+ }
+ public abstract T build();
+ }
+
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderFactory.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderFactory.java
new file mode 100644
index 0000000..d5e9353
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderFactory.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.security.Key;
+
+import org.apache.flume.FlumeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+
+public class CipherProviderFactory {
+ private static final Logger logger =
+ LoggerFactory.getLogger(CipherProviderFactory.class);
+
+ public static CipherProvider.Encryptor getEncrypter(String cipherProviderType,
+ Key key) {
+ if(cipherProviderType == null) {
+ return null;
+ }
+ CipherProvider provider = getProvider(cipherProviderType);
+ return provider.newEncryptorBuilder().setKey(key).build();
+ }
+ public static CipherProvider.Decryptor getDecrypter(String cipherProviderType,
+ Key key, byte[] parameters) {
+ if(cipherProviderType == null) {
+ return null;
+ }
+ CipherProvider provider = getProvider(cipherProviderType);
+ return provider.newDecryptorBuilder().setKey(key).setParameters(parameters)
+ .build();
+ }
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ private static CipherProvider getProvider(String cipherProviderType) {
+ Preconditions.checkNotNull(cipherProviderType,
+ "cipher provider type must not be null");
+ // try to find builder class in enum of known providers
+ CipherProviderType type;
+ try {
+ type = CipherProviderType.valueOf(cipherProviderType.toUpperCase());
+ } catch (IllegalArgumentException e) {
+ logger.debug("Not in enum, loading provider class: {}", cipherProviderType);
+ type = CipherProviderType.OTHER;
+ }
+ Class<? extends CipherProvider> providerClass = type.getProviderClass();
+
+ // handle the case where they have specified their own builder in the config
+ if (providerClass == null) {
+ try {
+ Class c = Class.forName(cipherProviderType);
+ if (c != null && CipherProvider.class.isAssignableFrom(c)) {
+ providerClass = (Class<? extends CipherProvider>) c;
+ } else {
+ String errMessage = "Unable to instantiate provider from " +
+ cipherProviderType;
+ logger.error(errMessage);
+ throw new FlumeException(errMessage);
+ }
+ } catch (ClassNotFoundException ex) {
+ logger.error("Class not found: " + cipherProviderType, ex);
+ throw new FlumeException(ex);
+ }
+ }
+ try {
+ return providerClass.newInstance();
+ } catch (Exception ex) {
+ String errMessage = "Cannot instantiate provider: " + cipherProviderType;
+ logger.error(errMessage, ex);
+ throw new FlumeException(errMessage, ex);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderType.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderType.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderType.java
new file mode 100644
index 0000000..87834d7
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/CipherProviderType.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+
+public enum CipherProviderType {
+ AESCTRNOPADDING(AESCTRNoPaddingProvider.class),
+ OTHER(null);
+
+ private final Class<? extends CipherProvider> providerClass;
+
+ CipherProviderType(Class<? extends CipherProvider> providerClass) {
+ this.providerClass = providerClass;
+ }
+
+ public Class<? extends CipherProvider> getProviderClass() {
+ return providerClass;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/EncryptionConfiguration.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/EncryptionConfiguration.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/EncryptionConfiguration.java
new file mode 100644
index 0000000..909d6bf
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/EncryptionConfiguration.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+public class EncryptionConfiguration {
+ private EncryptionConfiguration() {}
+
+ // prefix before all encryption options
+ public static final String ENCRYPTION_PREFIX = "encryption";
+ /**
+ * Encryption key provider, default is null.
+ */
+ public static final String KEY_PROVIDER = "keyProvider";
+ /**
+ * Encryption key alias, default is null.
+ */
+ public static final String KEY_ALIAS = "keyAlias";
+ /**
+ * Encryption cipher provider, default is null.
+ */
+ public static final String CIPHER_PROVIDER = "cipherProvider";
+
+ /**
+ * Space separated list of keys which are needed for the current set of logs
+ * plus the one specified in keyAlias
+ */
+ public static final String KEYS = "keys";
+
+ /**
+ * Path to key password file is:
+ * keys.aliasName.passwordFile
+ */
+ public static final String JCE_FILE_KEY_PASSWORD_FILE = "passwordFile";
+
+ /**
+ * Path to a jceks key store
+ */
+ public static final String JCE_FILE_KEY_STORE_FILE = "keyStoreFile";
+ /**
+ * Path to a jceks key store password file
+ */
+ public static final String JCE_FILE_KEY_STORE_PASSWORD_FILE = "keyStorePasswordFile";
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/JCEFileKeyProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/JCEFileKeyProvider.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/JCEFileKeyProvider.java
new file mode 100644
index 0000000..fff78e0
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/JCEFileKeyProvider.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.security.Key;
+import java.security.KeyStore;
+import java.util.Map;
+
+import org.apache.flume.Context;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Maps;
+import com.google.common.io.Files;
+
+public class JCEFileKeyProvider extends KeyProvider {
+ private static final Logger logger =
+ LoggerFactory.getLogger(JCEFileKeyProvider.class);
+
+ private Map<String, File> aliasPasswordFileMap;
+ private KeyStore ks;
+ private char[] keyStorePassword;
+
+ public JCEFileKeyProvider(File keyStoreFile, File keyStorePasswordFile,
+ Map<String, File> aliasPasswordFileMap) {
+ super();
+ this.aliasPasswordFileMap = aliasPasswordFileMap;
+ try {
+ ks = KeyStore.getInstance("jceks");
+ keyStorePassword = Files.toString(keyStorePasswordFile, Charsets.UTF_8)
+ .toCharArray();
+ ks.load(new FileInputStream(keyStoreFile), keyStorePassword);
+ } catch(Exception ex) {
+ throw Throwables.propagate(ex);
+ }
+ }
+
+ @Override
+ public Key getKey(String alias) {
+ try {
+ char[] keyPassword = keyStorePassword;
+ if(aliasPasswordFileMap.containsKey(alias)) {
+ keyPassword = Files.toString(aliasPasswordFileMap.get(alias),
+ Charsets.UTF_8).toCharArray();
+ }
+ Key key = ks.getKey(alias, keyPassword);
+ return key;
+ } catch (Exception e) {
+ throw Throwables.propagate(e);
+ }
+ }
+
+ public static class Builder implements KeyProvider.Builder {
+ @Override
+ public KeyProvider build(Context context) {
+ String keyStoreFileName = context.getString(
+ EncryptionConfiguration.JCE_FILE_KEY_STORE_FILE);
+ String keyStorePasswordFileName = context.getString(
+ EncryptionConfiguration.JCE_FILE_KEY_STORE_PASSWORD_FILE);
+ Preconditions.checkNotNull(keyStoreFileName, "KeyStore file not specified");
+ Preconditions.checkNotNull(keyStorePasswordFileName, "KeyStore password " +
+ "file not specified");
+ Map<String, File> aliasPasswordFileMap = Maps.newHashMap();
+ String passwordProtectedKeys = context.getString(
+ EncryptionConfiguration.KEYS);
+ if(passwordProtectedKeys != null) {
+ for(String passwordName : passwordProtectedKeys.trim().split("\\s+")) {
+ String propertyName = EncryptionConfiguration.KEYS + "." +
+ passwordName + "." +
+ EncryptionConfiguration.JCE_FILE_KEY_PASSWORD_FILE;
+ String passwordFileName = context.getString(propertyName,
+ keyStorePasswordFileName);
+ File passwordFile = new File(passwordFileName.trim());
+ if(passwordFile.isFile()) {
+ aliasPasswordFileMap.put(passwordName, passwordFile);
+ } else {
+ logger.warn("Password file for alias " + passwordName +
+ " does not exist");
+ }
+ }
+ }
+ File keyStoreFile = new File(keyStoreFileName.trim());
+ File keyStorePasswordFile = new File(keyStorePasswordFileName.trim());
+ return new JCEFileKeyProvider(keyStoreFile, keyStorePasswordFile,
+ aliasPasswordFileMap);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProvider.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProvider.java
new file mode 100644
index 0000000..e0877d1
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProvider.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.security.Key;
+
+import org.apache.flume.Context;
+
+public abstract class KeyProvider {
+
+ public abstract Key getKey(String alias);
+
+ public interface Builder {
+ public abstract KeyProvider build(Context context);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderFactory.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderFactory.java
new file mode 100644
index 0000000..fef0367
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderFactory.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import org.apache.flume.Context;
+import org.apache.flume.FlumeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+
+public class KeyProviderFactory {
+ private static final Logger logger =
+ LoggerFactory.getLogger(KeyProviderFactory.class);
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public static KeyProvider getInstance(String keyProviderType, Context context) {
+
+ Preconditions.checkNotNull(keyProviderType,
+ "provider type must not be null");
+
+ // try to find builder class in enum of known providers
+ KeyProviderType type;
+ try {
+ type = KeyProviderType.valueOf(keyProviderType.toUpperCase());
+ } catch (IllegalArgumentException e) {
+ logger.debug("Not in enum, loading provider class: {}", keyProviderType);
+ type = KeyProviderType.OTHER;
+ }
+ Class<? extends KeyProvider.Builder> providerClass =
+ type.getBuilderClass();
+
+ // handle the case where they have specified their own builder in the config
+ if (providerClass == null) {
+ try {
+ Class c = Class.forName(keyProviderType);
+ if (c != null && CipherProvider.class.isAssignableFrom(c)) {
+ providerClass = (Class<? extends KeyProvider.Builder>) c;
+ } else {
+ String errMessage = "Unable to instantiate Builder from " +
+ keyProviderType;
+ logger.error(errMessage);
+ throw new FlumeException(errMessage);
+ }
+ } catch (ClassNotFoundException ex) {
+ logger.error("Class not found: " + keyProviderType, ex);
+ throw new FlumeException(ex);
+ }
+ }
+
+ // build the builder
+ KeyProvider.Builder provider;
+ try {
+ provider = providerClass.newInstance();
+ } catch (InstantiationException ex) {
+ String errMessage = "Cannot instantiate builder: " + keyProviderType;
+ logger.error(errMessage, ex);
+ throw new FlumeException(errMessage, ex);
+ } catch (IllegalAccessException ex) {
+ String errMessage = "Cannot instantiate builder: " + keyProviderType;
+ logger.error(errMessage, ex);
+ throw new FlumeException(errMessage, ex);
+ }
+ return provider.build(context);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderType.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderType.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderType.java
new file mode 100644
index 0000000..0fef6dc
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/encryption/KeyProviderType.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+
+public enum KeyProviderType {
+ JCEKSFILE(JCEFileKeyProvider.Builder.class),
+ OTHER(null);
+
+ private final Class<? extends KeyProvider.Builder> keyProviderClass;
+
+ KeyProviderType(Class<? extends KeyProvider.Builder> keyStoreProviderClass) {
+ this.keyProviderClass = keyStoreProviderClass;
+ }
+
+ public Class<? extends KeyProvider.Builder> getBuilderClass() {
+ return keyProviderClass;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/main/proto/filechannel.proto
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/proto/filechannel.proto b/flume-ng-channels/flume-file-channel/src/main/proto/filechannel.proto
index 91567a6..850ec39 100644
--- a/flume-ng-channels/flume-file-channel/src/main/proto/filechannel.proto
+++ b/flume-ng-channels/flume-file-channel/src/main/proto/filechannel.proto
@@ -38,6 +38,13 @@ message LogFileMetaData {
required sfixed32 logFileID = 2;
required sfixed64 checkpointPosition = 3;
required sfixed64 checkpointWriteOrderID = 4;
+ optional LogFileEncryption encryption = 5;
+}
+
+message LogFileEncryption {
+ required string cipherProvider = 1;
+ required string keyAlias = 2;
+ optional bytes parameters = 3;
}
message TransactionEventHeader {
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java
index 43c7ae3..e2abc27 100644
--- a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java
@@ -31,7 +31,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@@ -96,25 +95,14 @@ public class TestFileChannel {
return createContext(new HashMap<String, String>());
}
private Context createContext(Map<String, String> overrides) {
- Context context = new Context();
- context.put(FileChannelConfiguration.CHECKPOINT_DIR,
- checkpointDir.getAbsolutePath());
- context.put(FileChannelConfiguration.DATA_DIRS, dataDir);
- context.put(FileChannelConfiguration.CAPACITY, String.valueOf(10000));
- // Set checkpoint for 5 seconds otherwise test will run out of memory
- context.put(FileChannelConfiguration.CHECKPOINT_INTERVAL, "5000");
- context.putAll(overrides);
- return context;
+ return TestUtils.createFileChannelContext(checkpointDir.getAbsolutePath(),
+ dataDir, overrides);
}
private FileChannel createFileChannel() {
return createFileChannel(new HashMap<String, String>());
}
private FileChannel createFileChannel(Map<String, String> overrides) {
- FileChannel channel = new FileChannel();
- channel.setName("FileChannel-" + UUID.randomUUID());
- Context context = createContext(overrides);
- Configurables.configure(channel, context);
- return channel;
+ return TestUtils.createFileChannel(checkpointDir.getAbsolutePath(), dataDir, overrides);
}
@Test
public void testFailAfterTakeBeforeCommit() throws Throwable {
@@ -728,7 +716,6 @@ public class TestFileChannel {
@Test
public void testPutForceCheckpointCommitReplay() throws Exception{
- Set<String> set = Sets.newHashSet();
Map<String, String> overrides = Maps.newHashMap();
overrides.put(FileChannelConfiguration.CAPACITY, String.valueOf(2));
overrides.put(FileChannelConfiguration.CHECKPOINT_INTERVAL, "10000");
@@ -782,7 +769,7 @@ public class TestFileChannel {
overrides.put(FileChannelConfiguration.MAX_FILE_SIZE, "20");
final FileChannel channel = createFileChannel(overrides);
channel.start();
- Set<String> in = putEvents(channel, "testing-reference-counting", 1, 15);
+ putEvents(channel, "testing-reference-counting", 1, 15);
Transaction tx = channel.getTransaction();
takeWithoutCommit(channel, tx, 10);
forceCheckpoint(channel);
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
index 11d0be0..1efdb80 100644
--- a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
@@ -52,12 +52,15 @@ public class TestLogFile {
dataDir = Files.createTempDir();
dataFile = new File(dataDir, String.valueOf(fileID));
Assert.assertTrue(dataDir.isDirectory());
- logFileWriter = LogFileFactory.getWriter(dataFile, fileID, 1000);
+ logFileWriter = LogFileFactory.getWriter(dataFile, fileID, 1000, null, null,
+ null);
}
@After
public void cleanup() throws IOException {
try {
- logFileWriter.close();
+ if(logFileWriter != null) {
+ logFileWriter.close();
+ }
} finally {
FileUtils.deleteQuietly(dataDir);
}
@@ -68,7 +71,7 @@ public class TestLogFile {
Collections.synchronizedList(new ArrayList<Throwable>());
ExecutorService executorService = Executors.newFixedThreadPool(10);
final LogFile.RandomReader logFileReader =
- LogFileFactory.getRandomReader(dataFile);
+ LogFileFactory.getRandomReader(dataFile, null);
for (int i = 0; i < 1000; i++) {
// first try and throw failures
synchronized (errors) {
@@ -123,7 +126,7 @@ public class TestLogFile {
puts.put(ptr.getOffset(), put);
}
LogFile.SequentialReader reader =
- LogFileFactory.getSequentialReader(dataFile);
+ LogFileFactory.getSequentialReader(dataFile, null);
LogRecord entry;
while((entry = reader.next()) != null) {
Integer offset = entry.getOffset();
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestTransactionEventRecordV3.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestTransactionEventRecordV3.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestTransactionEventRecordV3.java
index a9866a0..274ee7b 100644
--- a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestTransactionEventRecordV3.java
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestTransactionEventRecordV3.java
@@ -61,7 +61,7 @@ public class TestTransactionEventRecordV3 {
Put in = new Put(System.currentTimeMillis(),
WriteOrderOracle.next(),
new FlumeEvent(headers, new byte[0]));
- Put out = (Put)TransactionEventRecord.fromInputStream(toInputStream(in));
+ Put out = (Put)TransactionEventRecord.fromByteArray(toByteArray(in));
Assert.assertEquals(in.getClass(), out.getClass());
Assert.assertEquals(in.getRecordType(), out.getRecordType());
Assert.assertEquals(in.getTransactionID(), out.getTransactionID());
@@ -76,7 +76,7 @@ public class TestTransactionEventRecordV3 {
Put in = new Put(System.currentTimeMillis(),
WriteOrderOracle.next(),
new FlumeEvent(null, new byte[0]));
- Put out = (Put)TransactionEventRecord.fromInputStream(toInputStream(in));
+ Put out = (Put)TransactionEventRecord.fromByteArray(toByteArray(in));
Assert.assertEquals(in.getClass(), out.getClass());
Assert.assertEquals(in.getRecordType(), out.getRecordType());
Assert.assertEquals(in.getTransactionID(), out.getTransactionID());
@@ -89,7 +89,7 @@ public class TestTransactionEventRecordV3 {
public void testTakeSerialization() throws IOException {
Take in = new Take(System.currentTimeMillis(),
WriteOrderOracle.next(), 10, 20);
- Take out = (Take)TransactionEventRecord.fromInputStream(toInputStream(in));
+ Take out = (Take)TransactionEventRecord.fromByteArray(toByteArray(in));
Assert.assertEquals(in.getClass(), out.getClass());
Assert.assertEquals(in.getRecordType(), out.getRecordType());
Assert.assertEquals(in.getTransactionID(), out.getTransactionID());
@@ -102,7 +102,7 @@ public class TestTransactionEventRecordV3 {
public void testRollbackSerialization() throws IOException {
Rollback in = new Rollback(System.currentTimeMillis(),
WriteOrderOracle.next());
- Rollback out = (Rollback)TransactionEventRecord.fromInputStream(toInputStream(in));
+ Rollback out = (Rollback)TransactionEventRecord.fromByteArray(toByteArray(in));
Assert.assertEquals(in.getClass(), out.getClass());
Assert.assertEquals(in.getRecordType(), out.getRecordType());
Assert.assertEquals(in.getTransactionID(), out.getTransactionID());
@@ -113,7 +113,7 @@ public class TestTransactionEventRecordV3 {
public void testCommitSerialization() throws IOException {
Commit in = new Commit(System.currentTimeMillis(),
WriteOrderOracle.next());
- Commit out = (Commit)TransactionEventRecord.fromInputStream(toInputStream(in));
+ Commit out = (Commit)TransactionEventRecord.fromByteArray(toByteArray(in));
Assert.assertEquals(in.getClass(), out.getClass());
Assert.assertEquals(in.getRecordType(), out.getRecordType());
Assert.assertEquals(in.getTransactionID(), out.getTransactionID());
@@ -125,16 +125,15 @@ public class TestTransactionEventRecordV3 {
TransactionEventRecord in = mock(TransactionEventRecord.class);
when(in.getRecordType()).thenReturn(Short.MIN_VALUE);
try {
- TransactionEventRecord.fromInputStream(toInputStream(in));
+ TransactionEventRecord.fromByteArray(toByteArray(in));
Assert.fail();
} catch(NullPointerException e) {
Assert.assertEquals("Unknown action ffff8000", e.getMessage());
}
}
- private InputStream toInputStream(TransactionEventRecord record) throws IOException {
+ private byte[] toByteArray(TransactionEventRecord record) throws IOException {
ByteBuffer buffer = TransactionEventRecord.toByteBuffer(record);
- ByteArrayInputStream byteInput = new ByteArrayInputStream(buffer.array());
- return byteInput;
+ return buffer.array();
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
index 48948e4..8a9f10f 100644
--- a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
@@ -35,8 +35,10 @@ import java.util.UUID;
import java.util.zip.GZIPInputStream;
import org.apache.flume.Channel;
+import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.conf.Configurables;
import org.apache.flume.event.EventBuilder;
import org.apache.hadoop.io.Writable;
import org.junit.Assert;
@@ -45,6 +47,7 @@ import com.google.common.base.Charsets;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.io.ByteStreams;
+import com.google.common.io.Files;
import com.google.common.io.Resources;
public class TestUtils {
@@ -175,4 +178,32 @@ public class TestUtils {
ByteStreams.copy(new GZIPInputStream(input.openStream()),
new FileOutputStream(output));
}
+
+ public static Context createFileChannelContext(String checkpointDir,
+ String dataDir, Map<String, String> overrides) {
+ Context context = new Context();
+ context.put(FileChannelConfiguration.CHECKPOINT_DIR,
+ checkpointDir);
+ context.put(FileChannelConfiguration.DATA_DIRS, dataDir);
+ context.put(FileChannelConfiguration.CAPACITY, String.valueOf(10000));
+ // Set checkpoint for 5 seconds otherwise test will run out of memory
+ context.put(FileChannelConfiguration.CHECKPOINT_INTERVAL, "5000");
+ context.putAll(overrides);
+ return context;
+ }
+ public static FileChannel createFileChannel(String checkpointDir,
+ String dataDir, Map<String, String> overrides) {
+ FileChannel channel = new FileChannel();
+ channel.setName("FileChannel-" + UUID.randomUUID());
+ Context context = createFileChannelContext(checkpointDir, dataDir, overrides);
+ Configurables.configure(channel, context);
+ return channel;
+ }
+ public static File writeStringToFile(File baseDir, String name,
+ String text) throws IOException {
+ File passwordFile = new File(baseDir, name);
+ Files.write(text, passwordFile, Charsets.UTF_8);
+ return passwordFile;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/CipherProviderTestSuite.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/CipherProviderTestSuite.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/CipherProviderTestSuite.java
new file mode 100644
index 0000000..530ccf6
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/CipherProviderTestSuite.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import junit.framework.Assert;
+
+import com.google.common.base.Charsets;
+
+public class CipherProviderTestSuite {
+
+ private final CipherProvider.Encryptor encryptor;
+ private final CipherProvider.Decryptor decryptor;
+
+ public CipherProviderTestSuite(CipherProvider.Encryptor encryptor,
+ CipherProvider.Decryptor decryptor) {
+ this.encryptor = encryptor;
+ this.decryptor = decryptor;
+ }
+ public void test() throws Exception {
+ testBasic();
+ testEmpty();
+ testNullPlainText();
+ testNullCipherText();
+ }
+ public void testBasic() throws Exception {
+ String expected = "mn state fair is the place to be";
+ byte[] cipherText = encryptor.encrypt(expected.getBytes(Charsets.UTF_8));
+ byte[] clearText = decryptor.decrypt(cipherText);
+ Assert.assertEquals(expected, new String(clearText, Charsets.UTF_8));
+ }
+ public void testEmpty() throws Exception {
+ String expected = "";
+ byte[] cipherText = encryptor.encrypt(new byte[]{});
+ byte[] clearText = decryptor.decrypt(cipherText);
+ Assert.assertEquals(expected, new String(clearText));
+ }
+ public void testNullPlainText() throws Exception {
+ try {
+ encryptor.encrypt(null);
+ Assert.fail();
+ } catch (NullPointerException e) {
+ // expected
+ }
+ }
+ public void testNullCipherText() throws Exception {
+ try {
+ decryptor.decrypt(null);
+ Assert.fail();
+ } catch (NullPointerException e) {
+ // expected
+ }
+ }
+}