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}