001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.test; 019 020import java.io.IOException; 021import java.security.PrivilegedExceptionAction; 022import java.util.Arrays; 023import java.util.Iterator; 024import java.util.UUID; 025 026import org.apache.hadoop.conf.Configuration; 027import org.apache.hadoop.conf.Configured; 028import org.apache.hadoop.fs.Path; 029import org.apache.hadoop.hbase.Cell; 030import org.apache.hadoop.hbase.CellUtil; 031import org.apache.hadoop.hbase.HBaseConfiguration; 032import org.apache.hadoop.hbase.HColumnDescriptor; 033import org.apache.hadoop.hbase.HRegionLocation; 034import org.apache.hadoop.hbase.HTableDescriptor; 035import org.apache.hadoop.hbase.IntegrationTestingUtility; 036import org.apache.hadoop.hbase.TableName; 037import org.apache.hadoop.hbase.chaos.factories.MonkeyFactory; 038import org.apache.hadoop.hbase.client.Admin; 039import org.apache.hadoop.hbase.client.BufferedMutator; 040import org.apache.hadoop.hbase.client.BufferedMutatorParams; 041import org.apache.hadoop.hbase.client.ClusterConnection; 042import org.apache.hadoop.hbase.client.Connection; 043import org.apache.hadoop.hbase.client.ConnectionFactory; 044import org.apache.hadoop.hbase.client.Delete; 045import org.apache.hadoop.hbase.client.Put; 046import org.apache.hadoop.hbase.client.Result; 047import org.apache.hadoop.hbase.client.Scan; 048import org.apache.hadoop.hbase.io.ImmutableBytesWritable; 049import org.apache.hadoop.hbase.log.HBaseMarkers; 050import org.apache.hadoop.hbase.mapreduce.Import; 051import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; 052import org.apache.hadoop.hbase.security.User; 053import org.apache.hadoop.hbase.security.access.AccessControlClient; 054import org.apache.hadoop.hbase.security.access.Permission; 055import org.apache.hadoop.hbase.security.visibility.Authorizations; 056import org.apache.hadoop.hbase.security.visibility.CellVisibility; 057import org.apache.hadoop.hbase.security.visibility.VisibilityClient; 058import org.apache.hadoop.hbase.security.visibility.VisibilityTestUtil; 059import org.apache.hadoop.hbase.testclassification.IntegrationTests; 060import org.apache.hadoop.hbase.util.AbstractHBaseTool; 061import org.apache.hadoop.hbase.util.Bytes; 062import org.apache.hadoop.io.BytesWritable; 063import org.apache.hadoop.mapreduce.Counter; 064import org.apache.hadoop.mapreduce.CounterGroup; 065import org.apache.hadoop.mapreduce.Counters; 066import org.apache.hadoop.mapreduce.Job; 067import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 068import org.apache.hadoop.util.Tool; 069import org.apache.hadoop.util.ToolRunner; 070import org.junit.Test; 071import org.junit.experimental.categories.Category; 072import org.slf4j.Logger; 073import org.slf4j.LoggerFactory; 074import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine; 075 076/** 077 * IT test used to verify the deletes with visibility labels. 078 * The test creates three tables tablename_0, tablename_1 and tablename_2 and each table 079 * is associated with a unique pair of labels. 080 * Another common table with the name 'commontable' is created and it has the data combined 081 * from all these 3 tables such that there are 3 versions of every row but the visibility label 082 * in every row corresponds to the table from which the row originated. 083 * Then deletes are issued to the common table by selecting the visibility label 084 * associated with each of the smaller tables. 085 * After the delete is issued with one set of visibility labels we try to scan the common table 086 * with each of the visibility pairs defined for the 3 tables. 087 * So after the first delete is issued, a scan with the first set of visibility labels would 088 * return zero result whereas the scan issued with the other two sets of visibility labels 089 * should return all the rows corresponding to that set of visibility labels. The above 090 * process of delete and scan is repeated until after the last set of visibility labels are 091 * used for the deletes the common table should not return any row. 092 * 093 * To use this 094 * ./hbase org.apache.hadoop.hbase.test.IntegrationTestBigLinkedListWithVisibility Loop 1 1 20000 /tmp 1 10000 095 * or 096 * ./hbase org.apache.hadoop.hbase.IntegrationTestsDriver -r .*IntegrationTestBigLinkedListWithVisibility.* 097 */ 098@Category(IntegrationTests.class) 099public class IntegrationTestBigLinkedListWithVisibility extends IntegrationTestBigLinkedList { 100 101 private static final String CONFIDENTIAL = "confidential"; 102 private static final String TOPSECRET = "topsecret"; 103 private static final String SECRET = "secret"; 104 private static final String PUBLIC = "public"; 105 private static final String PRIVATE = "private"; 106 private static final String EVERYONE = "everyone"; 107 private static final String RESTRICTED = "restricted"; 108 private static final String GROUP = "group"; 109 private static final String PREVILIGED = "previliged"; 110 private static final String OPEN = "open"; 111 public static String labels = CONFIDENTIAL + "," + TOPSECRET + "," + SECRET + "," + RESTRICTED 112 + "," + PRIVATE + "," + PREVILIGED + "," + GROUP + "," + OPEN + "," + PUBLIC + "," + EVERYONE; 113 private static final String COMMA = ","; 114 private static final String UNDER_SCORE = "_"; 115 public static int DEFAULT_TABLES_COUNT = 3; 116 public static String tableName = "tableName"; 117 public static final String COMMON_TABLE_NAME = "commontable"; 118 public static final String LABELS_KEY = "LABELS"; 119 public static final String INDEX_KEY = "INDEX"; 120 private static User USER; 121 private static final String OR = "|"; 122 private static String USER_OPT = "user"; 123 private static String userName = "user1"; 124 125 static class VisibilityGenerator extends Generator { 126 private static final Logger LOG = LoggerFactory.getLogger(VisibilityGenerator.class); 127 128 @Override 129 protected void createSchema() throws IOException { 130 LOG.info("Creating tables"); 131 // Create three tables 132 boolean acl = AccessControlClient.isAccessControllerRunning(ConnectionFactory 133 .createConnection(getConf())); 134 if(!acl) { 135 LOG.info("No ACL available."); 136 } 137 try (Connection conn = ConnectionFactory.createConnection(getConf()); 138 Admin admin = conn.getAdmin()) { 139 for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) { 140 TableName tableName = IntegrationTestBigLinkedListWithVisibility.getTableName(i); 141 createTable(admin, tableName, false, acl); 142 } 143 TableName tableName = TableName.valueOf(COMMON_TABLE_NAME); 144 createTable(admin, tableName, true, acl); 145 } 146 } 147 148 private void createTable(Admin admin, TableName tableName, boolean setVersion, 149 boolean acl) throws IOException { 150 if (!admin.tableExists(tableName)) { 151 HTableDescriptor htd = new HTableDescriptor(tableName); 152 HColumnDescriptor family = new HColumnDescriptor(FAMILY_NAME); 153 if (setVersion) { 154 family.setMaxVersions(DEFAULT_TABLES_COUNT); 155 } 156 htd.addFamily(family); 157 admin.createTable(htd); 158 if (acl) { 159 LOG.info("Granting permissions for user " + USER.getShortName()); 160 Permission.Action[] actions = { Permission.Action.READ }; 161 try { 162 AccessControlClient.grant(ConnectionFactory.createConnection(getConf()), tableName, 163 USER.getShortName(), null, null, actions); 164 } catch (Throwable e) { 165 LOG.error(HBaseMarkers.FATAL, "Error in granting permission for the user " + 166 USER.getShortName(), e); 167 throw new IOException(e); 168 } 169 } 170 } 171 } 172 173 @Override 174 protected void setMapperForGenerator(Job job) { 175 job.setMapperClass(VisibilityGeneratorMapper.class); 176 } 177 178 static class VisibilityGeneratorMapper extends GeneratorMapper { 179 BufferedMutator[] tables = new BufferedMutator[DEFAULT_TABLES_COUNT]; 180 181 @Override 182 protected void setup(org.apache.hadoop.mapreduce.Mapper.Context context) throws IOException, 183 InterruptedException { 184 super.setup(context); 185 } 186 187 @Override 188 protected void instantiateHTable() throws IOException { 189 for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) { 190 BufferedMutatorParams params = new BufferedMutatorParams(getTableName(i)); 191 params.writeBufferSize(4 * 1024 * 1024); 192 BufferedMutator table = connection.getBufferedMutator(params); 193 this.tables[i] = table; 194 } 195 } 196 197 @Override 198 protected void cleanup(org.apache.hadoop.mapreduce.Mapper.Context context) 199 throws IOException, InterruptedException { 200 for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) { 201 if (tables[i] != null) { 202 tables[i].close(); 203 } 204 } 205 } 206 207 @Override 208 protected void persist(org.apache.hadoop.mapreduce.Mapper.Context output, long count, 209 byte[][] prev, byte[][] current, byte[] id) throws IOException { 210 String visibilityExps = ""; 211 String[] split = labels.split(COMMA); 212 for (int i = 0; i < current.length; i++) { 213 for (int j = 0; j < DEFAULT_TABLES_COUNT; j++) { 214 Put put = new Put(current[i]); 215 byte[] value = prev == null ? NO_KEY : prev[i]; 216 put.addColumn(FAMILY_NAME, COLUMN_PREV, value); 217 218 if (count >= 0) { 219 put.addColumn(FAMILY_NAME, COLUMN_COUNT, Bytes.toBytes(count + i)); 220 } 221 if (id != null) { 222 put.addColumn(FAMILY_NAME, COLUMN_CLIENT, id); 223 } 224 visibilityExps = split[j * 2] + OR + split[(j * 2) + 1]; 225 put.setCellVisibility(new CellVisibility(visibilityExps)); 226 tables[j].mutate(put); 227 try { 228 Thread.sleep(1); 229 } catch (InterruptedException e) { 230 throw new IOException(); 231 } 232 } 233 if (i % 1000 == 0) { 234 // Tickle progress every so often else maprunner will think us hung 235 output.progress(); 236 } 237 } 238 } 239 } 240 } 241 242 static class Copier extends Configured implements Tool { 243 private static final Logger LOG = LoggerFactory.getLogger(Copier.class); 244 private TableName tableName; 245 private int labelIndex; 246 private boolean delete; 247 248 public Copier(TableName tableName, int index, boolean delete) { 249 this.tableName = tableName; 250 this.labelIndex = index; 251 this.delete = delete; 252 } 253 254 public int runCopier(String outputDir) throws Exception { 255 Job job = null; 256 Scan scan = null; 257 job = new Job(getConf()); 258 job.setJobName("Data copier"); 259 job.getConfiguration().setInt("INDEX", labelIndex); 260 job.getConfiguration().set("LABELS", labels); 261 job.setJarByClass(getClass()); 262 scan = new Scan(); 263 scan.setCacheBlocks(false); 264 scan.setRaw(true); 265 266 String[] split = labels.split(COMMA); 267 scan.setAuthorizations(new Authorizations(split[this.labelIndex * 2], 268 split[(this.labelIndex * 2) + 1])); 269 if (delete) { 270 LOG.info("Running deletes"); 271 } else { 272 LOG.info("Running copiers"); 273 } 274 if (delete) { 275 TableMapReduceUtil.initTableMapperJob(tableName.getNameAsString(), scan, 276 VisibilityDeleteImport.class, null, null, job); 277 } else { 278 TableMapReduceUtil.initTableMapperJob(tableName.getNameAsString(), scan, 279 VisibilityImport.class, null, null, job); 280 } 281 job.getConfiguration().setBoolean("mapreduce.map.speculative", false); 282 job.getConfiguration().setBoolean("mapreduce.reduce.speculative", false); 283 TableMapReduceUtil.initTableReducerJob(COMMON_TABLE_NAME, null, job, null, null, null, null); 284 TableMapReduceUtil.addDependencyJars(job); 285 TableMapReduceUtil.addDependencyJars(job.getConfiguration(), AbstractHBaseTool.class); 286 TableMapReduceUtil.initCredentials(job); 287 job.setNumReduceTasks(0); 288 boolean success = job.waitForCompletion(true); 289 return success ? 0 : 1; 290 } 291 292 @Override 293 public int run(String[] arg0) throws Exception { 294 // TODO Auto-generated method stub 295 return 0; 296 } 297 } 298 299 static class VisibilityImport extends Import.Importer { 300 private int index; 301 private String labels; 302 private String[] split; 303 304 @Override 305 public void setup(org.apache.hadoop.mapreduce.Mapper.Context context) { 306 index = context.getConfiguration().getInt(INDEX_KEY, -1); 307 labels = context.getConfiguration().get(LABELS_KEY); 308 split = labels.split(COMMA); 309 super.setup(context); 310 } 311 312 @Override 313 protected void addPutToKv(Put put, Cell kv) throws IOException { 314 String visibilityExps = split[index * 2] + OR + split[(index * 2) + 1]; 315 put.setCellVisibility(new CellVisibility(visibilityExps)); 316 super.addPutToKv(put, kv); 317 } 318 } 319 320 static class VisibilityDeleteImport extends Import.Importer { 321 private int index; 322 private String labels; 323 private String[] split; 324 325 @Override 326 public void setup(org.apache.hadoop.mapreduce.Mapper.Context context) { 327 index = context.getConfiguration().getInt(INDEX_KEY, -1); 328 labels = context.getConfiguration().get(LABELS_KEY); 329 split = labels.split(COMMA); 330 super.setup(context); 331 } 332 333 // Creating delete here 334 @Override 335 protected void processKV(ImmutableBytesWritable key, Result result, 336 org.apache.hadoop.mapreduce.Mapper.Context context, Put put, 337 org.apache.hadoop.hbase.client.Delete delete) throws 338 IOException, InterruptedException { 339 String visibilityExps = split[index * 2] + OR + split[(index * 2) + 1]; 340 for (Cell kv : result.rawCells()) { 341 // skip if we filter it out 342 if (kv == null) 343 continue; 344 // Create deletes here 345 if (delete == null) { 346 delete = new Delete(key.get()); 347 } 348 delete.setCellVisibility(new CellVisibility(visibilityExps)); 349 delete.addFamily(CellUtil.cloneFamily(kv)); 350 } 351 if (delete != null) { 352 context.write(key, delete); 353 } 354 } 355 } 356 357 @Override 358 protected void addOptions() { 359 super.addOptions(); 360 addOptWithArg("u", USER_OPT, "User name"); 361 } 362 363 @Override 364 protected void processOptions(CommandLine cmd) { 365 super.processOptions(cmd); 366 if (cmd.hasOption(USER_OPT)) { 367 userName = cmd.getOptionValue(USER_OPT); 368 } 369 370 } 371 @Override 372 public void setUpCluster() throws Exception { 373 util = getTestingUtil(null); 374 Configuration conf = util.getConfiguration(); 375 VisibilityTestUtil.enableVisiblityLabels(conf); 376 conf.set("hbase.superuser", User.getCurrent().getName()); 377 conf.setBoolean("dfs.permissions", false); 378 USER = User.createUserForTesting(conf, userName, new String[] {}); 379 super.setUpCluster(); 380 addLabels(); 381 } 382 383 static TableName getTableName(int i) { 384 return TableName.valueOf(tableName + UNDER_SCORE + i); 385 } 386 387 private void addLabels() throws Exception { 388 try { 389 VisibilityClient.addLabels(util.getConnection(), labels.split(COMMA)); 390 VisibilityClient.setAuths(util.getConnection(), labels.split(COMMA), USER.getName()); 391 } catch (Throwable t) { 392 throw new IOException(t); 393 } 394 } 395 396 static class VisibilityVerify extends Verify { 397 private static final Logger LOG = LoggerFactory.getLogger(VisibilityVerify.class); 398 private TableName tableName; 399 private int labelIndex; 400 401 public VisibilityVerify(String tableName, int index) { 402 this.tableName = TableName.valueOf(tableName); 403 this.labelIndex = index; 404 } 405 406 @Override 407 public int run(final Path outputDir, final int numReducers) throws Exception { 408 LOG.info("Running Verify with outputDir=" + outputDir + ", numReducers=" + numReducers); 409 PrivilegedExceptionAction<Integer> scanAction = new PrivilegedExceptionAction<Integer>() { 410 @Override 411 public Integer run() throws Exception { 412 return doVerify(outputDir, numReducers); 413 } 414 }; 415 return USER.runAs(scanAction); 416 } 417 418 private int doVerify(Path outputDir, int numReducers) throws IOException, InterruptedException, 419 ClassNotFoundException { 420 job = new Job(getConf()); 421 422 job.setJobName("Link Verifier"); 423 job.setNumReduceTasks(numReducers); 424 job.setJarByClass(getClass()); 425 426 setJobScannerConf(job); 427 428 Scan scan = new Scan(); 429 scan.addColumn(FAMILY_NAME, COLUMN_PREV); 430 scan.setCaching(10000); 431 scan.setCacheBlocks(false); 432 String[] split = labels.split(COMMA); 433 434 scan.setAuthorizations(new Authorizations(split[this.labelIndex * 2], 435 split[(this.labelIndex * 2) + 1])); 436 437 TableMapReduceUtil.initTableMapperJob(tableName.getName(), scan, VerifyMapper.class, 438 BytesWritable.class, BytesWritable.class, job); 439 TableMapReduceUtil.addDependencyJars(job.getConfiguration(), AbstractHBaseTool.class); 440 441 job.getConfiguration().setBoolean("mapreduce.map.speculative", false); 442 443 job.setReducerClass(VerifyReducer.class); 444 job.setOutputFormatClass(TextOutputFormat.class); 445 TextOutputFormat.setOutputPath(job, outputDir); 446 boolean success = job.waitForCompletion(true); 447 448 return success ? 0 : 1; 449 } 450 451 @Override 452 protected void handleFailure(Counters counters) throws IOException { 453 Configuration conf = job.getConfiguration(); 454 ClusterConnection conn = (ClusterConnection) ConnectionFactory.createConnection(conf); 455 TableName tableName = TableName.valueOf(COMMON_TABLE_NAME); 456 CounterGroup g = counters.getGroup("undef"); 457 Iterator<Counter> it = g.iterator(); 458 while (it.hasNext()) { 459 String keyString = it.next().getName(); 460 byte[] key = Bytes.toBytes(keyString); 461 HRegionLocation loc = conn.relocateRegion(tableName, key); 462 LOG.error("undefined row " + keyString + ", " + loc); 463 } 464 g = counters.getGroup("unref"); 465 it = g.iterator(); 466 while (it.hasNext()) { 467 String keyString = it.next().getName(); 468 byte[] key = Bytes.toBytes(keyString); 469 HRegionLocation loc = conn.relocateRegion(tableName, key); 470 LOG.error("unreferred row " + keyString + ", " + loc); 471 } 472 } 473 } 474 475 static class VisibilityLoop extends Loop { 476 private static final int SLEEP_IN_MS = 5000; 477 private static final Logger LOG = LoggerFactory.getLogger(VisibilityLoop.class); 478 IntegrationTestBigLinkedListWithVisibility it; 479 480 @Override 481 protected void runGenerator(int numMappers, long numNodes, String outputDir, Integer width, 482 Integer wrapMultiplier, Integer numWalkers) throws Exception { 483 Path outputPath = new Path(outputDir); 484 UUID uuid = UUID.randomUUID(); // create a random UUID. 485 Path generatorOutput = new Path(outputPath, uuid.toString()); 486 487 Generator generator = new VisibilityGenerator(); 488 generator.setConf(getConf()); 489 int retCode = generator.run(numMappers, numNodes, generatorOutput, width, wrapMultiplier, 490 numWalkers); 491 if (retCode > 0) { 492 throw new RuntimeException("Generator failed with return code: " + retCode); 493 } 494 } 495 496 protected void runDelete(int numMappers, long numNodes, String outputDir, Integer width, 497 Integer wrapMultiplier, int tableIndex) throws Exception { 498 LOG.info("Running copier on table "+IntegrationTestBigLinkedListWithVisibility.getTableName(tableIndex)); 499 Copier copier = new Copier( 500 IntegrationTestBigLinkedListWithVisibility.getTableName(tableIndex), tableIndex, true); 501 copier.setConf(getConf()); 502 copier.runCopier(outputDir); 503 Thread.sleep(SLEEP_IN_MS); 504 } 505 506 protected void runVerify(String outputDir, int numReducers, long expectedNumNodes, 507 boolean allTables) throws Exception { 508 Path outputPath = new Path(outputDir); 509 510 if (allTables) { 511 for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) { 512 LOG.info("Verifying table " + i); 513 sleep(SLEEP_IN_MS); 514 UUID uuid = UUID.randomUUID(); // create a random UUID. 515 Path iterationOutput = new Path(outputPath, uuid.toString()); 516 Verify verify = new VisibilityVerify(getTableName(i).getNameAsString(), i); 517 verify(numReducers, expectedNumNodes, iterationOutput, verify); 518 } 519 } 520 for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) { 521 runVerifyCommonTable(outputDir, numReducers, expectedNumNodes, i); 522 } 523 } 524 525 private void runVerify(String outputDir, int numReducers, long expectedNodes, int tableIndex) 526 throws Exception { 527 long temp = expectedNodes; 528 for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) { 529 if (i <= tableIndex) { 530 expectedNodes = 0; 531 } else { 532 expectedNodes = temp; 533 } 534 LOG.info("Verifying data in the table with index "+i+ " and expected nodes is "+expectedNodes); 535 runVerifyCommonTable(outputDir, numReducers, expectedNodes, i); 536 } 537 } 538 539 private void sleep(long ms) throws InterruptedException { 540 Thread.sleep(ms); 541 } 542 543 protected void runVerifyCommonTable(String outputDir, int numReducers, long expectedNumNodes, 544 int index) throws Exception { 545 LOG.info("Verifying common table with index " + index); 546 sleep(SLEEP_IN_MS); 547 Path outputPath = new Path(outputDir); 548 UUID uuid = UUID.randomUUID(); // create a random UUID. 549 Path iterationOutput = new Path(outputPath, uuid.toString()); 550 Verify verify = new VisibilityVerify(TableName.valueOf(COMMON_TABLE_NAME).getNameAsString(), 551 index); 552 verify(numReducers, expectedNumNodes, iterationOutput, verify); 553 } 554 555 protected void runCopier(String outputDir) throws Exception { 556 for (int i = 0; i < DEFAULT_TABLES_COUNT; i++) { 557 LOG.info("Running copier " + IntegrationTestBigLinkedListWithVisibility.getTableName(i)); 558 sleep(SLEEP_IN_MS); 559 Copier copier = new Copier(IntegrationTestBigLinkedListWithVisibility.getTableName(i), i, 560 false); 561 copier.setConf(getConf()); 562 copier.runCopier(outputDir); 563 } 564 } 565 566 private void verify(int numReducers, long expectedNumNodes, 567 Path iterationOutput, Verify verify) throws Exception { 568 verify.setConf(getConf()); 569 int retCode = verify.run(iterationOutput, numReducers); 570 if (retCode > 0) { 571 throw new RuntimeException("Verify.run failed with return code: " + retCode); 572 } 573 574 if (!verify.verify(expectedNumNodes)) { 575 throw new RuntimeException("Verify.verify failed"); 576 } 577 578 LOG.info("Verify finished with succees. Total nodes=" + expectedNumNodes); 579 } 580 581 @Override 582 public int run(String[] args) throws Exception { 583 if (args.length < 5) { 584 System.err 585 .println("Usage: Loop <num iterations> " + 586 "<num mappers> <num nodes per mapper> <output dir> " + 587 "<num reducers> [<width> <wrap multiplier>]"); 588 return 1; 589 } 590 LOG.info("Running Loop with args:" + Arrays.deepToString(args)); 591 592 int numIterations = Integer.parseInt(args[0]); 593 int numMappers = Integer.parseInt(args[1]); 594 long numNodes = Long.parseLong(args[2]); 595 String outputDir = args[3]; 596 int numReducers = Integer.parseInt(args[4]); 597 Integer width = (args.length < 6) ? null : Integer.parseInt(args[5]); 598 Integer wrapMultiplier = (args.length < 7) ? null : Integer.parseInt(args[6]); 599 long expectedNumNodes = 0; 600 601 if (numIterations < 0) { 602 numIterations = Integer.MAX_VALUE; // run indefinitely (kind of) 603 } 604 605 for (int i = 0; i < numIterations; i++) { 606 LOG.info("Starting iteration = " + i); 607 LOG.info("Generating data"); 608 // By default run no concurrent walkers for test with visibility 609 runGenerator(numMappers, numNodes, outputDir, width, wrapMultiplier, 0); 610 expectedNumNodes += numMappers * numNodes; 611 // Copying wont work because expressions are not returned back to the 612 // client 613 LOG.info("Running copier"); 614 sleep(SLEEP_IN_MS); 615 runCopier(outputDir); 616 LOG.info("Verifying copied data"); 617 sleep(SLEEP_IN_MS); 618 runVerify(outputDir, numReducers, expectedNumNodes, true); 619 sleep(SLEEP_IN_MS); 620 for (int j = 0; j < DEFAULT_TABLES_COUNT; j++) { 621 LOG.info("Deleting data on table with index: "+j); 622 runDelete(numMappers, numNodes, outputDir, width, wrapMultiplier, j); 623 sleep(SLEEP_IN_MS); 624 LOG.info("Verifying common table after deleting"); 625 runVerify(outputDir, numReducers, expectedNumNodes, j); 626 sleep(SLEEP_IN_MS); 627 } 628 } 629 return 0; 630 } 631 } 632 633 @Override 634 @Test 635 public void testContinuousIngest() throws IOException, Exception { 636 // Loop <num iterations> <num mappers> <num nodes per mapper> <output dir> 637 // <num reducers> 638 int ret = ToolRunner.run( 639 getTestingUtil(getConf()).getConfiguration(), 640 new VisibilityLoop(), 641 new String[] { "1", "1", "20000", 642 util.getDataTestDirOnTestFS("IntegrationTestBigLinkedListWithVisibility").toString(), 643 "1", "10000" }); 644 org.junit.Assert.assertEquals(0, ret); 645 } 646 647 public static void main(String[] args) throws Exception { 648 Configuration conf = HBaseConfiguration.create(); 649 IntegrationTestingUtility.setUseDistributedCluster(conf); 650 int ret = ToolRunner.run(conf, new IntegrationTestBigLinkedListWithVisibility(), args); 651 System.exit(ret); 652 } 653 654 @Override 655 protected MonkeyFactory getDefaultMonkeyFactory() { 656 return MonkeyFactory.getFactory(MonkeyFactory.CALM); 657 } 658 659 @Override 660 public int runTestFromCommandLine() throws Exception { 661 Tool tool = null; 662 Loop loop = new VisibilityLoop(); 663 loop.it = this; 664 tool = loop; 665 return ToolRunner.run(getConf(), tool, otherArgs); 666 } 667}