1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.mapreduce;
19
20 import static org.junit.Assert.assertEquals;
21 import static org.junit.Assert.assertTrue;
22
23 import java.io.IOException;
24 import java.security.PrivilegedExceptionAction;
25 import java.util.ArrayList;
26 import java.util.Arrays;
27 import java.util.HashSet;
28 import java.util.List;
29 import java.util.Set;
30 import java.util.UUID;
31
32 import org.apache.commons.logging.Log;
33 import org.apache.commons.logging.LogFactory;
34 import org.apache.hadoop.conf.Configurable;
35 import org.apache.hadoop.conf.Configuration;
36 import org.apache.hadoop.fs.FSDataOutputStream;
37 import org.apache.hadoop.fs.FileStatus;
38 import org.apache.hadoop.fs.FileSystem;
39 import org.apache.hadoop.fs.Path;
40 import org.apache.hadoop.hbase.Cell;
41 import org.apache.hadoop.hbase.CellUtil;
42 import org.apache.hadoop.hbase.HBaseTestingUtility;
43 import org.apache.hadoop.hbase.HConstants;
44 import org.apache.hadoop.hbase.LargeTests;
45 import org.apache.hadoop.hbase.client.Delete;
46 import org.apache.hadoop.hbase.client.HBaseAdmin;
47 import org.apache.hadoop.hbase.client.HTable;
48 import org.apache.hadoop.hbase.client.Result;
49 import org.apache.hadoop.hbase.client.ResultScanner;
50 import org.apache.hadoop.hbase.client.Scan;
51 import org.apache.hadoop.hbase.protobuf.generated.VisibilityLabelsProtos.VisibilityLabelsResponse;
52 import org.apache.hadoop.hbase.security.User;
53 import org.apache.hadoop.hbase.security.visibility.Authorizations;
54 import org.apache.hadoop.hbase.security.visibility.CellVisibility;
55 import org.apache.hadoop.hbase.security.visibility.ScanLabelGenerator;
56 import org.apache.hadoop.hbase.security.visibility.SimpleScanLabelGenerator;
57 import org.apache.hadoop.hbase.security.visibility.VisibilityClient;
58 import org.apache.hadoop.hbase.security.visibility.VisibilityConstants;
59 import org.apache.hadoop.hbase.security.visibility.VisibilityController;
60 import org.apache.hadoop.hbase.security.visibility.VisibilityUtils;
61 import org.apache.hadoop.hbase.util.Bytes;
62 import org.apache.hadoop.mapred.Utils.OutputFileUtils.OutputFilesFilter;
63 import org.apache.hadoop.util.Tool;
64 import org.apache.hadoop.util.ToolRunner;
65 import org.junit.AfterClass;
66 import org.junit.BeforeClass;
67 import org.junit.Test;
68 import org.junit.experimental.categories.Category;
69
70 @Category(LargeTests.class)
71 public class TestImportTSVWithVisibilityLabels implements Configurable {
72
73 protected static final Log LOG = LogFactory.getLog(TestImportTSVWithVisibilityLabels.class);
74 protected static final String NAME = TestImportTsv.class.getSimpleName();
75 protected static HBaseTestingUtility util = new HBaseTestingUtility();
76
77
78
79
80
81 protected static final String DELETE_AFTER_LOAD_CONF = NAME + ".deleteAfterLoad";
82
83
84
85
86 protected static final String FORCE_COMBINER_CONF = NAME + ".forceCombiner";
87
88 private final String FAMILY = "FAM";
89 private final static String TOPSECRET = "topsecret";
90 private final static String PUBLIC = "public";
91 private final static String PRIVATE = "private";
92 private final static String CONFIDENTIAL = "confidential";
93 private final static String SECRET = "secret";
94 private static User SUPERUSER;
95 private static Configuration conf;
96
97 @Override
98 public Configuration getConf() {
99 return util.getConfiguration();
100 }
101
102 @Override
103 public void setConf(Configuration conf) {
104 throw new IllegalArgumentException("setConf not supported");
105 }
106
107 @BeforeClass
108 public static void provisionCluster() throws Exception {
109 conf = util.getConfiguration();
110 SUPERUSER = User.createUserForTesting(conf, "admin", new String[] { "supergroup" });
111 conf.set("hbase.superuser", "admin,"+User.getCurrent().getName());
112 conf.setInt("hfile.format.version", 3);
113 conf.set("hbase.coprocessor.master.classes", VisibilityController.class.getName());
114 conf.set("hbase.coprocessor.region.classes", VisibilityController.class.getName());
115 conf.setClass(VisibilityUtils.VISIBILITY_LABEL_GENERATOR_CLASS, SimpleScanLabelGenerator.class,
116 ScanLabelGenerator.class);
117 util.startMiniCluster();
118
119 util.waitTableEnabled(VisibilityConstants.LABELS_TABLE_NAME.getName(), 50000);
120 createLabels();
121 HBaseAdmin admin = new HBaseAdmin(util.getConfiguration());
122 util.startMiniMapReduceCluster();
123 }
124
125 private static void createLabels() throws IOException, InterruptedException {
126 PrivilegedExceptionAction<VisibilityLabelsResponse> action =
127 new PrivilegedExceptionAction<VisibilityLabelsResponse>() {
128 @Override
129 public VisibilityLabelsResponse run() throws Exception {
130 String[] labels = { SECRET, TOPSECRET, CONFIDENTIAL, PUBLIC, PRIVATE };
131 try {
132 VisibilityClient.addLabels(conf, labels);
133 LOG.info("Added labels ");
134 } catch (Throwable t) {
135 LOG.error("Error in adding labels" , t);
136 throw new IOException(t);
137 }
138 return null;
139 }
140 };
141 SUPERUSER.runAs(action);
142 }
143
144 @AfterClass
145 public static void releaseCluster() throws Exception {
146 util.shutdownMiniMapReduceCluster();
147 util.shutdownMiniCluster();
148 }
149
150 @Test
151 public void testMROnTable() throws Exception {
152 String tableName = "test-" + UUID.randomUUID();
153
154
155 String[] args = new String[] {
156 "-D" + ImportTsv.MAPPER_CONF_KEY
157 + "=org.apache.hadoop.hbase.mapreduce.TsvImporterMapper",
158 "-D" + ImportTsv.COLUMNS_CONF_KEY + "=HBASE_ROW_KEY,FAM:A,FAM:B,HBASE_CELL_VISIBILITY",
159 "-D" + ImportTsv.SEPARATOR_CONF_KEY + "=\u001b", tableName };
160 String data = "KEY\u001bVALUE1\u001bVALUE2\u001bsecret&private\n";
161 util.createTable(tableName, FAMILY);
162 doMROnTableTest(util, FAMILY, data, args, 1);
163 util.deleteTable(tableName);
164 }
165
166 @Test
167 public void testMROnTableWithDeletes() throws Exception {
168 String tableName = "test-" + UUID.randomUUID();
169
170
171 String[] args = new String[] {
172 "-D" + ImportTsv.MAPPER_CONF_KEY + "=org.apache.hadoop.hbase.mapreduce.TsvImporterMapper",
173 "-D" + ImportTsv.COLUMNS_CONF_KEY + "=HBASE_ROW_KEY,FAM:A,FAM:B,HBASE_CELL_VISIBILITY",
174 "-D" + ImportTsv.SEPARATOR_CONF_KEY + "=\u001b", tableName };
175 String data = "KEY\u001bVALUE1\u001bVALUE2\u001bsecret&private\n";
176 util.createTable(tableName, FAMILY);
177 doMROnTableTest(util, FAMILY, data, args, 1);
178 issueDeleteAndVerifyData(tableName);
179 util.deleteTable(tableName);
180 }
181
182 private void issueDeleteAndVerifyData(String tableName) throws IOException {
183 LOG.debug("Validating table after delete.");
184 HTable table = new HTable(conf, tableName);
185 boolean verified = false;
186 long pause = conf.getLong("hbase.client.pause", 5 * 1000);
187 int numRetries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 5);
188 for (int i = 0; i < numRetries; i++) {
189 try {
190 Delete d = new Delete(Bytes.toBytes("KEY"));
191 d.deleteFamily(Bytes.toBytes(FAMILY));
192 d.setCellVisibility(new CellVisibility("private&secret"));
193 table.delete(d);
194
195 Scan scan = new Scan();
196
197 scan.addFamily(Bytes.toBytes(FAMILY));
198 scan.setAuthorizations(new Authorizations("secret", "private"));
199 ResultScanner resScanner = table.getScanner(scan);
200 Result[] next = resScanner.next(5);
201 assertEquals(0, next.length);
202 verified = true;
203 break;
204 } catch (NullPointerException e) {
205
206
207 }
208 try {
209 Thread.sleep(pause);
210 } catch (InterruptedException e) {
211
212 }
213 }
214 table.close();
215 assertTrue(verified);
216 }
217
218 @Test
219 public void testMROnTableWithBulkload() throws Exception {
220 String tableName = "test-" + UUID.randomUUID();
221 Path hfiles = new Path(util.getDataTestDirOnTestFS(tableName), "hfiles");
222
223 String[] args = new String[] {
224 "-D" + ImportTsv.BULK_OUTPUT_CONF_KEY + "=" + hfiles.toString(),
225 "-D" + ImportTsv.COLUMNS_CONF_KEY
226 + "=HBASE_ROW_KEY,FAM:A,FAM:B,HBASE_CELL_VISIBILITY",
227 "-D" + ImportTsv.SEPARATOR_CONF_KEY + "=\u001b", tableName };
228 String data = "KEY\u001bVALUE1\u001bVALUE2\u001bsecret&private\n";
229 util.createTable(tableName, FAMILY);
230 doMROnTableTest(util, FAMILY, data, args, 1);
231 util.deleteTable(tableName);
232 }
233
234 @Test
235 public void testBulkOutputWithTsvImporterTextMapper() throws Exception {
236 String table = "test-" + UUID.randomUUID();
237 String FAMILY = "FAM";
238 Path bulkOutputPath = new Path(util.getDataTestDirOnTestFS(table),"hfiles");
239
240 String[] args =
241 new String[] {
242 "-D" + ImportTsv.MAPPER_CONF_KEY
243 + "=org.apache.hadoop.hbase.mapreduce.TsvImporterTextMapper",
244 "-D" + ImportTsv.COLUMNS_CONF_KEY
245 + "=HBASE_ROW_KEY,FAM:A,FAM:B,HBASE_CELL_VISIBILITY",
246 "-D" + ImportTsv.SEPARATOR_CONF_KEY + "=\u001b",
247 "-D" + ImportTsv.BULK_OUTPUT_CONF_KEY + "=" + bulkOutputPath.toString(), table
248 };
249 String data = "KEY\u001bVALUE4\u001bVALUE8\u001bsecret&private\n";
250 doMROnTableTest(util, FAMILY, data, args, 4);
251 util.deleteTable(table);
252 }
253
254 @Test
255 public void testMRWithOutputFormat() throws Exception {
256 String tableName = "test-" + UUID.randomUUID();
257 Path hfiles = new Path(util.getDataTestDirOnTestFS(tableName), "hfiles");
258
259 String[] args = new String[] {
260 "-D" + ImportTsv.MAPPER_CONF_KEY
261 + "=org.apache.hadoop.hbase.mapreduce.TsvImporterMapper",
262 "-D" + ImportTsv.BULK_OUTPUT_CONF_KEY + "=" + hfiles.toString(),
263 "-D" + ImportTsv.COLUMNS_CONF_KEY + "=HBASE_ROW_KEY,FAM:A,FAM:B,HBASE_CELL_VISIBILITY",
264 "-D" + ImportTsv.SEPARATOR_CONF_KEY + "=\u001b", tableName };
265 String data = "KEY\u001bVALUE4\u001bVALUE8\u001bsecret&private\n";
266 util.createTable(tableName, FAMILY);
267 doMROnTableTest(util, FAMILY, data, args, 1);
268 util.deleteTable(tableName);
269 }
270
271
272
273
274
275
276
277
278
279
280
281 protected static Tool doMROnTableTest(HBaseTestingUtility util, String family, String data,
282 String[] args, int valueMultiplier) throws Exception {
283 String table = args[args.length - 1];
284 Configuration conf = new Configuration(util.getConfiguration());
285
286
287 FileSystem fs = FileSystem.get(conf);
288 Path inputPath = fs.makeQualified(new Path(util.getDataTestDirOnTestFS(table), "input.dat"));
289 FSDataOutputStream op = fs.create(inputPath, true);
290 if (data == null) {
291 data = "KEY\u001bVALUE1\u001bVALUE2\n";
292 }
293 op.write(Bytes.toBytes(data));
294 op.close();
295 LOG.debug(String.format("Wrote test data to file: %s", inputPath));
296
297 if (conf.getBoolean(FORCE_COMBINER_CONF, true)) {
298 LOG.debug("Forcing combiner.");
299 conf.setInt("min.num.spills.for.combine", 1);
300 }
301
302
303 List<String> argv = new ArrayList<String>(Arrays.asList(args));
304 argv.add(inputPath.toString());
305 Tool tool = new ImportTsv();
306 LOG.debug("Running ImportTsv with arguments: " + argv);
307 assertEquals(0, ToolRunner.run(conf, tool, argv.toArray(args)));
308
309
310
311
312 boolean createdHFiles = false;
313 String outputPath = null;
314 for (String arg : argv) {
315 if (arg.contains(ImportTsv.BULK_OUTPUT_CONF_KEY)) {
316 createdHFiles = true;
317
318 outputPath = arg.split("=")[1];
319 break;
320 }
321 }
322 LOG.debug("validating the table " + createdHFiles);
323 if (createdHFiles)
324 validateHFiles(fs, outputPath, family);
325 else
326 validateTable(conf, table, family, valueMultiplier);
327
328 if (conf.getBoolean(DELETE_AFTER_LOAD_CONF, true)) {
329 LOG.debug("Deleting test subdirectory");
330 util.cleanupDataTestDirOnTestFS(table);
331 }
332 return tool;
333 }
334
335
336
337
338 private static void validateHFiles(FileSystem fs, String outputPath, String family)
339 throws IOException {
340
341
342 LOG.debug("Validating HFiles.");
343 Set<String> configFamilies = new HashSet<String>();
344 configFamilies.add(family);
345 Set<String> foundFamilies = new HashSet<String>();
346 for (FileStatus cfStatus : fs.listStatus(new Path(outputPath), new OutputFilesFilter())) {
347 LOG.debug("The output path has files");
348 String[] elements = cfStatus.getPath().toString().split(Path.SEPARATOR);
349 String cf = elements[elements.length - 1];
350 foundFamilies.add(cf);
351 assertTrue(String.format(
352 "HFile ouput contains a column family (%s) not present in input families (%s)", cf,
353 configFamilies), configFamilies.contains(cf));
354 for (FileStatus hfile : fs.listStatus(cfStatus.getPath())) {
355 assertTrue(String.format("HFile %s appears to contain no data.", hfile.getPath()),
356 hfile.getLen() > 0);
357 }
358 }
359 }
360
361
362
363
364 private static void validateTable(Configuration conf, String tableName, String family,
365 int valueMultiplier) throws IOException {
366
367 LOG.debug("Validating table.");
368 HTable table = new HTable(conf, tableName);
369 boolean verified = false;
370 long pause = conf.getLong("hbase.client.pause", 5 * 1000);
371 int numRetries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 5);
372 for (int i = 0; i < numRetries; i++) {
373 try {
374 Scan scan = new Scan();
375
376 scan.addFamily(Bytes.toBytes(family));
377 scan.setAuthorizations(new Authorizations("secret","private"));
378 ResultScanner resScanner = table.getScanner(scan);
379 Result[] next = resScanner.next(5);
380 assertEquals(1, next.length);
381 for (Result res : resScanner) {
382 LOG.debug("Getting results " + res.size());
383 assertTrue(res.size() == 2);
384 List<Cell> kvs = res.listCells();
385 assertTrue(CellUtil.matchingRow(kvs.get(0), Bytes.toBytes("KEY")));
386 assertTrue(CellUtil.matchingRow(kvs.get(1), Bytes.toBytes("KEY")));
387 assertTrue(CellUtil.matchingValue(kvs.get(0), Bytes.toBytes("VALUE" + valueMultiplier)));
388 assertTrue(CellUtil.matchingValue(kvs.get(1),
389 Bytes.toBytes("VALUE" + 2 * valueMultiplier)));
390
391 }
392 verified = true;
393 break;
394 } catch (NullPointerException e) {
395
396
397 }
398 try {
399 Thread.sleep(pause);
400 } catch (InterruptedException e) {
401
402 }
403 }
404 table.close();
405 assertTrue(verified);
406 }
407
408 }