1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.wal;
19
20 import static org.junit.Assert.assertFalse;
21 import static org.junit.Assert.assertTrue;
22
23 import java.io.IOException;
24
25 import org.apache.commons.io.IOUtils;
26 import org.apache.commons.logging.LogFactory;
27 import org.apache.commons.logging.impl.Log4JLogger;
28 import org.apache.hadoop.conf.Configuration;
29 import org.apache.hadoop.fs.FSDataInputStream;
30 import org.apache.hadoop.fs.FileStatus;
31 import org.apache.hadoop.fs.FileSystem;
32 import org.apache.hadoop.fs.Path;
33 import org.apache.hadoop.hbase.HBaseTestingUtility;
34 import org.apache.hadoop.hbase.HColumnDescriptor;
35 import org.apache.hadoop.hbase.HConstants;
36 import org.apache.hadoop.hbase.HRegionInfo;
37 import org.apache.hadoop.hbase.HTableDescriptor;
38 import org.apache.hadoop.hbase.KeyValue;
39 import org.apache.hadoop.hbase.TableName;
40 import org.apache.hadoop.hbase.io.crypto.KeyProviderForTesting;
41 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
42 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
43
44 import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogReader;
45 import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogWriter;
46 import org.apache.hadoop.hbase.regionserver.wal.SecureProtobufLogReader;
47 import org.apache.hadoop.hbase.regionserver.wal.SecureProtobufLogWriter;
48 import org.apache.hadoop.hbase.regionserver.wal.SecureWALCellCodec;
49 import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
50 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
51 import org.apache.hadoop.hbase.testclassification.MediumTests;
52 import org.apache.hadoop.hbase.util.Bytes;
53 import org.apache.hadoop.hbase.util.FSUtils;
54 import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
55 import org.apache.log4j.Level;
56 import org.junit.BeforeClass;
57 import org.junit.Rule;
58 import org.junit.Test;
59 import org.junit.experimental.categories.Category;
60 import org.junit.rules.TestName;
61
62
63
64
65 @Category(MediumTests.class)
66 public class TestWALReaderOnSecureWAL {
67 static {
68 ((Log4JLogger)LogFactory.getLog("org.apache.hadoop.hbase.regionserver.wal"))
69 .getLogger().setLevel(Level.ALL);
70 };
71 static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
72 final byte[] value = Bytes.toBytes("Test value");
73
74 private static final String WAL_ENCRYPTION = "hbase.regionserver.wal.encryption";
75
76 @Rule
77 public TestName currentTest = new TestName();
78
79 @BeforeClass
80 public static void setUpBeforeClass() throws Exception {
81 Configuration conf = TEST_UTIL.getConfiguration();
82 conf.set(HConstants.CRYPTO_KEYPROVIDER_CONF_KEY, KeyProviderForTesting.class.getName());
83 conf.set(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY, "hbase");
84 conf.setBoolean("hbase.hlog.split.skip.errors", true);
85 conf.setBoolean(HConstants.ENABLE_WAL_ENCRYPTION, true);
86 FSUtils.setRootDir(conf, TEST_UTIL.getDataTestDir());
87 }
88
89 private Path writeWAL(final WALFactory wals, final String tblName) throws IOException {
90 Configuration conf = TEST_UTIL.getConfiguration();
91 String clsName = conf.get(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, WALCellCodec.class.getName());
92 conf.setClass(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, SecureWALCellCodec.class,
93 WALCellCodec.class);
94 try {
95 TableName tableName = TableName.valueOf(tblName);
96 HTableDescriptor htd = new HTableDescriptor(tableName);
97 htd.addFamily(new HColumnDescriptor(tableName.getName()));
98 HRegionInfo regioninfo = new HRegionInfo(tableName,
99 HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, false);
100 final int total = 10;
101 final byte[] row = Bytes.toBytes("row");
102 final byte[] family = Bytes.toBytes("family");
103 final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl(1);
104
105
106 WAL wal = wals.getWAL(regioninfo.getEncodedNameAsBytes());
107 for (int i = 0; i < total; i++) {
108 WALEdit kvs = new WALEdit();
109 kvs.add(new KeyValue(row, family, Bytes.toBytes(i), value));
110 wal.append(htd, regioninfo, new WALKey(regioninfo.getEncodedNameAsBytes(), tableName,
111 System.currentTimeMillis(), mvcc), kvs, true);
112 }
113 wal.sync();
114 final Path walPath = DefaultWALProvider.getCurrentFileName(wal);
115 wal.shutdown();
116
117 return walPath;
118 } finally {
119
120 conf.set(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, clsName);
121 }
122 }
123
124 @Test()
125 public void testWALReaderOnSecureWAL() throws Exception {
126 Configuration conf = TEST_UTIL.getConfiguration();
127 conf.setClass("hbase.regionserver.hlog.reader.impl", ProtobufLogReader.class,
128 WAL.Reader.class);
129 conf.setClass("hbase.regionserver.hlog.writer.impl", SecureProtobufLogWriter.class,
130 WALProvider.Writer.class);
131 conf.setBoolean(WAL_ENCRYPTION, true);
132 FileSystem fs = TEST_UTIL.getTestFileSystem();
133 final WALFactory wals = new WALFactory(conf, null, currentTest.getMethodName());
134 Path walPath = writeWAL(wals, currentTest.getMethodName());
135
136
137 long length = fs.getFileStatus(walPath).getLen();
138 FSDataInputStream in = fs.open(walPath);
139 byte[] fileData = new byte[(int)length];
140 IOUtils.readFully(in, fileData);
141 in.close();
142 assertFalse("Cells appear to be plaintext", Bytes.contains(fileData, value));
143
144
145 try {
146 wals.createReader(TEST_UTIL.getTestFileSystem(), walPath);
147 assertFalse(true);
148 } catch (IOException ioe) {
149
150 }
151
152 FileStatus[] listStatus = fs.listStatus(walPath.getParent());
153 RecoveryMode mode = (conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false) ?
154 RecoveryMode.LOG_REPLAY : RecoveryMode.LOG_SPLITTING);
155 Path rootdir = FSUtils.getRootDir(conf);
156 try {
157 WALSplitter s = new WALSplitter(wals, conf, rootdir, fs, null, null, mode);
158 s.splitLogFile(listStatus[0], null);
159 Path file = new Path(ZKSplitLog.getSplitLogDir(rootdir, listStatus[0].getPath().getName()),
160 "corrupt");
161 assertTrue(fs.exists(file));
162
163 } catch (IOException ioe) {
164 assertTrue("WAL should have been sidelined", false);
165 }
166 wals.close();
167 }
168
169 @Test()
170 public void testSecureWALReaderOnWAL() throws Exception {
171 Configuration conf = TEST_UTIL.getConfiguration();
172 conf.setClass("hbase.regionserver.hlog.reader.impl", SecureProtobufLogReader.class,
173 WAL.Reader.class);
174 conf.setClass("hbase.regionserver.hlog.writer.impl", ProtobufLogWriter.class,
175 WALProvider.Writer.class);
176 conf.setBoolean(WAL_ENCRYPTION, false);
177 FileSystem fs = TEST_UTIL.getTestFileSystem();
178 final WALFactory wals = new WALFactory(conf, null, currentTest.getMethodName());
179 Path walPath = writeWAL(wals, currentTest.getMethodName());
180
181
182 long length = fs.getFileStatus(walPath).getLen();
183 FSDataInputStream in = fs.open(walPath);
184 byte[] fileData = new byte[(int)length];
185 IOUtils.readFully(in, fileData);
186 in.close();
187 assertTrue("Cells should be plaintext", Bytes.contains(fileData, value));
188
189
190 try {
191 WAL.Reader reader = wals.createReader(TEST_UTIL.getTestFileSystem(), walPath);
192 reader.close();
193 } catch (IOException ioe) {
194 assertFalse(true);
195 }
196
197 FileStatus[] listStatus = fs.listStatus(walPath.getParent());
198 RecoveryMode mode = (conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false) ?
199 RecoveryMode.LOG_REPLAY : RecoveryMode.LOG_SPLITTING);
200 Path rootdir = FSUtils.getRootDir(conf);
201 try {
202 WALSplitter s = new WALSplitter(wals, conf, rootdir, fs, null, null, mode);
203 s.splitLogFile(listStatus[0], null);
204 Path file = new Path(ZKSplitLog.getSplitLogDir(rootdir, listStatus[0].getPath().getName()),
205 "corrupt");
206 assertTrue(!fs.exists(file));
207 } catch (IOException ioe) {
208 assertTrue("WAL should have been processed", false);
209 }
210 wals.close();
211 }
212 }