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;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertTrue;
022
023import java.io.IOException;
024import java.util.Arrays;
025import java.util.Set;
026import org.apache.hadoop.conf.Configuration;
027import org.apache.hadoop.fs.FileStatus;
028import org.apache.hadoop.fs.FileSystem;
029import org.apache.hadoop.fs.Path;
030import org.apache.hadoop.hbase.client.Admin;
031import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
032import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
033import org.apache.hadoop.hbase.client.Put;
034import org.apache.hadoop.hbase.client.Result;
035import org.apache.hadoop.hbase.client.ResultScanner;
036import org.apache.hadoop.hbase.client.Table;
037import org.apache.hadoop.hbase.client.TableDescriptor;
038import org.apache.hadoop.hbase.master.cleaner.TimeToLiveHFileCleaner;
039import org.apache.hadoop.hbase.mob.FaultyMobStoreCompactor;
040import org.apache.hadoop.hbase.mob.MobConstants;
041import org.apache.hadoop.hbase.mob.MobFileCleanerChore;
042import org.apache.hadoop.hbase.mob.MobStoreEngine;
043import org.apache.hadoop.hbase.mob.MobUtils;
044import org.apache.hadoop.hbase.testclassification.IntegrationTests;
045import org.apache.hadoop.hbase.util.Bytes;
046import org.apache.hadoop.util.ToolRunner;
047import org.junit.After;
048import org.junit.Before;
049import org.junit.Test;
050import org.junit.experimental.categories.Category;
051import org.slf4j.Logger;
052import org.slf4j.LoggerFactory;
053
054import org.apache.hbase.thirdparty.com.google.common.base.MoreObjects;
055import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
056
057/**
058 * An integration test to detect regressions in HBASE-22749. Test creates MOB-enabled table, and
059 * runs in parallel, the following tasks: loads data, runs MOB compactions, runs MOB cleaning chore.
060 * The failure injections into MOB compaction cycle is implemented via specific sub-class of
061 * DefaultMobStoreCompactor - FaultyMobStoreCompactor. The probability of failure is controlled by
062 * command-line argument 'failprob'.
063 * @see <a href="https://issues.apache.org/jira/browse/HBASE-22749">HBASE-22749</a>
064 *      <p>
065 *      Sample usage:
066 *
067 *      <pre>
068 * hbase org.apache.hadoop.hbase.IntegrationTestMobCompaction -Dservers=10 -Drows=1000000
069 * -Dfailprob=0.2
070 *      </pre>
071 */
072@SuppressWarnings("deprecation")
073
074@Category(IntegrationTests.class)
075public class IntegrationTestMobCompaction extends IntegrationTestBase {
076  protected static final Logger LOG = LoggerFactory.getLogger(IntegrationTestMobCompaction.class);
077
078  protected static final String REGIONSERVER_COUNT_KEY = "servers";
079  protected static final String ROWS_COUNT_KEY = "rows";
080  protected static final String FAILURE_PROB_KEY = "failprob";
081
082  protected static final int DEFAULT_REGIONSERVER_COUNT = 3;
083  protected static final int DEFAULT_ROWS_COUNT = 5000000;
084  protected static final double DEFAULT_FAILURE_PROB = 0.1;
085
086  protected static int regionServerCount = DEFAULT_REGIONSERVER_COUNT;
087  protected static long rowsToLoad = DEFAULT_ROWS_COUNT;
088  protected static double failureProb = DEFAULT_FAILURE_PROB;
089
090  protected static String famStr = "f1";
091  protected static byte[] fam = Bytes.toBytes(famStr);
092  protected static byte[] qualifier = Bytes.toBytes("q1");
093  protected static long mobLen = 10;
094  protected static byte[] mobVal = Bytes
095    .toBytes("01234567890123456789012345678901234567890123456789012345678901234567890123456789");
096
097  private static Configuration conf;
098  private static TableDescriptor tableDescriptor;
099  private static ColumnFamilyDescriptor familyDescriptor;
100  private static Admin admin;
101  private static Table table = null;
102  private static MobFileCleanerChore chore;
103
104  private static volatile boolean run = true;
105
106  @Override
107  @Before
108  public void setUp() throws Exception {
109    util = getTestingUtil(getConf());
110    conf = util.getConfiguration();
111    // Initialize with test-specific configuration values
112    initConf(conf);
113    regionServerCount = conf.getInt(REGIONSERVER_COUNT_KEY, DEFAULT_REGIONSERVER_COUNT);
114    LOG.info("Initializing cluster with {} region servers.", regionServerCount);
115    util.initializeCluster(regionServerCount);
116    admin = util.getAdmin();
117
118    createTestTable();
119
120    LOG.info("Cluster initialized and ready");
121  }
122
123  private void createTestTable() throws IOException {
124    // Create test table
125    familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(fam).setMobEnabled(true)
126      .setMobThreshold(mobLen).setMaxVersions(1).build();
127    tableDescriptor = util.createModifyableTableDescriptor("testMobCompactTable")
128      .setColumnFamily(familyDescriptor).build();
129    table = util.createTable(tableDescriptor, null);
130  }
131
132  @After
133  public void tearDown() throws IOException {
134    LOG.info("Cleaning up after test.");
135    if (util.isDistributedCluster()) {
136      deleteTablesIfAny();
137      // TODO
138    }
139    LOG.info("Restoring cluster.");
140    util.restoreCluster();
141    LOG.info("Cluster restored.");
142  }
143
144  @Override
145  public void setUpMonkey() throws Exception {
146    // Sorry, no Monkey
147    String msg = "Chaos monkey is not supported";
148    LOG.warn(msg);
149    throw new IOException(msg);
150  }
151
152  private void deleteTablesIfAny() throws IOException {
153    if (table != null) {
154      util.deleteTableIfAny(table.getName());
155    }
156  }
157
158  @Override
159  public void setUpCluster() throws Exception {
160    util = getTestingUtil(getConf());
161    LOG.debug("Initializing/checking cluster has {} servers", regionServerCount);
162    util.initializeCluster(regionServerCount);
163    LOG.debug("Done initializing/checking cluster");
164  }
165
166  /**
167   * @return status of CLI execution
168   */
169  @Override
170  public int runTestFromCommandLine() throws Exception {
171    testMobCompaction();
172    return 0;
173  }
174
175  @Override
176  public TableName getTablename() {
177    // That is only valid when Monkey is CALM (no monkey)
178    return null;
179  }
180
181  @Override
182  protected Set<String> getColumnFamilies() {
183    // That is only valid when Monkey is CALM (no monkey)
184    return null;
185  }
186
187  @Override
188  protected void addOptions() {
189    addOptWithArg(REGIONSERVER_COUNT_KEY,
190      "Total number of region servers. Default: '" + DEFAULT_REGIONSERVER_COUNT + "'");
191    addOptWithArg(ROWS_COUNT_KEY,
192      "Total number of data rows to load. Default: '" + DEFAULT_ROWS_COUNT + "'");
193    addOptWithArg(FAILURE_PROB_KEY,
194      "Probability of a failure of a region MOB compaction request. Default: '"
195        + DEFAULT_FAILURE_PROB + "'");
196  }
197
198  @Override
199  protected void processOptions(CommandLine cmd) {
200    super.processOptions(cmd);
201
202    regionServerCount = Integer.parseInt(
203      cmd.getOptionValue(REGIONSERVER_COUNT_KEY, Integer.toString(DEFAULT_REGIONSERVER_COUNT)));
204    rowsToLoad =
205      Long.parseLong(cmd.getOptionValue(ROWS_COUNT_KEY, Long.toString(DEFAULT_ROWS_COUNT)));
206    failureProb = Double
207      .parseDouble(cmd.getOptionValue(FAILURE_PROB_KEY, Double.toString(DEFAULT_FAILURE_PROB)));
208
209    LOG.info(
210      MoreObjects.toStringHelper("Parsed Options").add(REGIONSERVER_COUNT_KEY, regionServerCount)
211        .add(ROWS_COUNT_KEY, rowsToLoad).add(FAILURE_PROB_KEY, failureProb).toString());
212  }
213
214  private static void initConf(Configuration conf) {
215
216    conf.setInt("hfile.format.version", 3);
217    conf.setLong(TimeToLiveHFileCleaner.TTL_CONF_KEY, 0);
218    conf.setInt("hbase.client.retries.number", 100);
219    conf.setInt("hbase.hregion.max.filesize", 200000000);
220    conf.setInt("hbase.hregion.memstore.flush.size", 800000);
221    conf.setInt("hbase.hstore.blockingStoreFiles", 150);
222    conf.setInt("hbase.hstore.compaction.throughput.lower.bound", 52428800);
223    conf.setInt("hbase.hstore.compaction.throughput.higher.bound", 2 * 52428800);
224    conf.setDouble("hbase.mob.compaction.fault.probability", failureProb);
225    conf.set(MobStoreEngine.MOB_COMPACTOR_CLASS_KEY, FaultyMobStoreCompactor.class.getName());
226    conf.setBoolean("hbase.table.sanity.checks", false);
227    conf.setLong(MobConstants.MIN_AGE_TO_ARCHIVE_KEY, 20000);
228
229  }
230
231  class MajorCompaction implements Runnable {
232
233    @Override
234    public void run() {
235      while (run) {
236        try {
237          admin.majorCompact(tableDescriptor.getTableName(), fam);
238          Thread.sleep(120000);
239        } catch (Exception e) {
240          LOG.error("MOB Stress Test FAILED", e);
241          System.exit(-1);
242        }
243      }
244    }
245  }
246
247  class CleanMobAndArchive implements Runnable {
248
249    @Override
250    public void run() {
251      while (run) {
252        try {
253          LOG.info("MOB cleanup chore started ...");
254          if (chore == null) {
255            chore = new MobFileCleanerChore();
256          }
257          chore.cleanupObsoleteMobFiles(conf, table.getName());
258          LOG.info("MOB cleanup chore finished");
259
260          Thread.sleep(130000);
261        } catch (Exception e) {
262          e.printStackTrace();
263        }
264      }
265    }
266  }
267
268  class WriteData implements Runnable {
269
270    private long rows = -1;
271
272    public WriteData(long rows) {
273      this.rows = rows;
274    }
275
276    @Override
277    public void run() {
278      try {
279
280        // BufferedMutator bm = admin.getConnection().getBufferedMutator(table.getName());
281        // Put Operation
282        for (int i = 0; i < rows; i++) {
283          Put p = new Put(Bytes.toBytes(i));
284          p.addColumn(fam, qualifier, mobVal);
285          table.put(p);
286
287          // bm.mutate(p);
288          if (i % 10000 == 0) {
289            LOG.info("LOADED=" + i);
290            try {
291              Thread.sleep(500);
292            } catch (InterruptedException ee) {
293
294            }
295          }
296          if (i % 100000 == 0) {
297            printStats(i);
298          }
299        }
300        // bm.flush();
301        admin.flush(table.getName());
302        run = false;
303      } catch (Exception e) {
304        LOG.error("MOB Stress Test FAILED", e);
305        System.exit(-1);
306      }
307    }
308  }
309
310  @Test
311  public void testMobCompaction() throws InterruptedException, IOException {
312
313    try {
314
315      Thread writeData = new Thread(new WriteData(rowsToLoad));
316      writeData.start();
317
318      Thread majorcompact = new Thread(new MajorCompaction());
319      majorcompact.start();
320
321      Thread cleaner = new Thread(new CleanMobAndArchive());
322      cleaner.start();
323
324      while (run) {
325        Thread.sleep(1000);
326      }
327
328      getNumberOfMobFiles(conf, table.getName(), new String(fam));
329      LOG.info("Waiting for write thread to finish ...");
330      writeData.join();
331      // Cleanup again
332      chore.cleanupObsoleteMobFiles(conf, table.getName());
333
334      if (util != null) {
335        LOG.info("Archive cleaner started ...");
336        // Call archive cleaner again
337        util.getMiniHBaseCluster().getMaster().getHFileCleaner().choreForTesting();
338        LOG.info("Archive cleaner finished");
339      }
340
341      scanTable();
342
343    } finally {
344
345      admin.disableTable(tableDescriptor.getTableName());
346      admin.deleteTable(tableDescriptor.getTableName());
347    }
348    LOG.info("MOB Stress Test finished OK");
349    printStats(rowsToLoad);
350
351  }
352
353  private long getNumberOfMobFiles(Configuration conf, TableName tableName, String family)
354    throws IOException {
355    FileSystem fs = FileSystem.get(conf);
356    Path dir = MobUtils.getMobFamilyPath(conf, tableName, family);
357    FileStatus[] stat = fs.listStatus(dir);
358    for (FileStatus st : stat) {
359      LOG.debug("MOB Directory content: {}", st.getPath());
360    }
361    LOG.debug("MOB Directory content total files: {}", stat.length);
362
363    return stat.length;
364  }
365
366  public void printStats(long loaded) {
367    LOG.info("MOB Stress Test: loaded=" + loaded + " compactions="
368      + FaultyMobStoreCompactor.totalCompactions.get() + " major="
369      + FaultyMobStoreCompactor.totalMajorCompactions.get() + " mob="
370      + FaultyMobStoreCompactor.mobCounter.get() + " injected failures="
371      + FaultyMobStoreCompactor.totalFailures.get());
372  }
373
374  private void scanTable() {
375    try {
376
377      Result result;
378      ResultScanner scanner = table.getScanner(fam);
379      int counter = 0;
380      while ((result = scanner.next()) != null) {
381        assertTrue(Arrays.equals(result.getValue(fam, qualifier), mobVal));
382        if (counter % 10000 == 0) {
383          LOG.info("GET=" + counter);
384        }
385        counter++;
386      }
387      assertEquals(rowsToLoad, counter);
388    } catch (Exception e) {
389      e.printStackTrace();
390      LOG.error("MOB Stress Test FAILED");
391      if (util != null) {
392        assertTrue(false);
393      } else {
394        System.exit(-1);
395      }
396    }
397  }
398
399  public static void main(String[] args) throws Exception {
400    Configuration conf = HBaseConfiguration.create();
401    initConf(conf);
402    IntegrationTestingUtility.setUseDistributedCluster(conf);
403    int status = ToolRunner.run(conf, new IntegrationTestMobCompaction(), args);
404    System.exit(status);
405  }
406}