1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package org.apache.hadoop.hbase.util;
18
19 import java.io.IOException;
20 import java.io.InterruptedIOException;
21 import java.lang.reflect.Constructor;
22 import java.net.InetAddress;
23 import java.security.SecureRandom;
24 import java.util.ArrayList;
25 import java.util.Arrays;
26 import java.util.List;
27 import java.util.Properties;
28 import java.util.Random;
29 import java.util.concurrent.atomic.AtomicReference;
30
31 import javax.crypto.spec.SecretKeySpec;
32
33 import org.apache.commons.cli.CommandLine;
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.conf.Configuration;
37 import org.apache.hadoop.hbase.HBaseConfiguration;
38 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
39 import org.apache.hadoop.hbase.HBaseTestingUtility;
40 import org.apache.hadoop.hbase.HColumnDescriptor;
41 import org.apache.hadoop.hbase.HConstants;
42 import org.apache.hadoop.hbase.HTableDescriptor;
43 import org.apache.hadoop.hbase.TableName;
44 import org.apache.hadoop.hbase.classification.InterfaceAudience;
45 import org.apache.log4j.Level;
46 import org.apache.log4j.LogManager;
47 import org.apache.zookeeper.ZooKeeper;
48 import org.slf4j.Logger;
49 import org.slf4j.LoggerFactory;
50 import org.apache.hadoop.hbase.client.Admin;
51 import org.apache.hadoop.hbase.client.ConnectionFactory;
52 import org.apache.hadoop.hbase.client.Durability;
53 import org.apache.hadoop.hbase.client.HBaseAdmin;
54 import org.apache.hadoop.hbase.io.compress.Compression;
55 import org.apache.hadoop.hbase.io.crypto.Cipher;
56 import org.apache.hadoop.hbase.io.crypto.Encryption;
57 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
58 import org.apache.hadoop.hbase.regionserver.BloomType;
59 import org.apache.hadoop.hbase.security.EncryptionUtil;
60 import org.apache.hadoop.hbase.security.User;
61 import org.apache.hadoop.hbase.security.access.AccessControlClient;
62 import org.apache.hadoop.hbase.security.access.Permission;
63 import org.apache.hadoop.hbase.util.test.LoadTestDataGenerator;
64 import org.apache.hadoop.hbase.util.test.LoadTestDataGeneratorWithACL;
65 import org.apache.hadoop.security.SecurityUtil;
66 import org.apache.hadoop.security.UserGroupInformation;
67 import org.apache.hadoop.util.ToolRunner;
68
69
70
71
72
73
74 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
75 public class LoadTestTool extends AbstractHBaseTool {
76
77 private static final Log LOG = LogFactory.getLog(LoadTestTool.class);
78 private static final String COLON = ":";
79
80
81 private TableName tableName;
82
83
84 private byte[][] families;
85
86
87 protected static final String DEFAULT_TABLE_NAME = "cluster_test";
88
89
90 public static byte[] DEFAULT_COLUMN_FAMILY = Bytes.toBytes("test_cf");
91
92
93 public static final byte[][] DEFAULT_COLUMN_FAMILIES = { DEFAULT_COLUMN_FAMILY };
94
95
96 protected static final int DEFAULT_DATA_SIZE = 64;
97
98
99 protected static final int DEFAULT_NUM_THREADS = 20;
100
101
102 protected static final String OPT_USAGE_LOAD =
103 "<avg_cols_per_key>:<avg_data_size>" +
104 "[:<#threads=" + DEFAULT_NUM_THREADS + ">]";
105
106
107 protected static final String OPT_USAGE_READ =
108 "<verify_percent>[:<#threads=" + DEFAULT_NUM_THREADS + ">]";
109
110
111 protected static final String OPT_USAGE_UPDATE =
112 "<update_percent>[:<#threads=" + DEFAULT_NUM_THREADS
113 + ">][:<#whether to ignore nonce collisions=0>]";
114
115 protected static final String OPT_USAGE_BLOOM = "Bloom filter type, one of " +
116 Arrays.toString(BloomType.values());
117
118 protected static final String OPT_USAGE_COMPRESSION = "Compression type, " +
119 "one of " + Arrays.toString(Compression.Algorithm.values());
120
121 public static final String OPT_DATA_BLOCK_ENCODING_USAGE =
122 "Encoding algorithm (e.g. prefix "
123 + "compression) to use for data blocks in the test column family, "
124 + "one of " + Arrays.toString(DataBlockEncoding.values()) + ".";
125
126 protected static final String OPT_VERBOSE = "verbose";
127
128 public static final String OPT_BLOOM = "bloom";
129 public static final String OPT_COMPRESSION = "compression";
130 public static final String OPT_DEFERRED_LOG_FLUSH = "deferredlogflush";
131 public static final String OPT_DEFERRED_LOG_FLUSH_USAGE = "Enable deferred log flush.";
132
133 public static final String OPT_DATA_BLOCK_ENCODING =
134 HColumnDescriptor.DATA_BLOCK_ENCODING.toLowerCase();
135
136 public static final String OPT_INMEMORY = "in_memory";
137 public static final String OPT_USAGE_IN_MEMORY = "Tries to keep the HFiles of the CF " +
138 "inmemory as far as possible. Not guaranteed that reads are always served from inmemory";
139
140 public static final String OPT_GENERATOR = "generator";
141 public static final String OPT_GENERATOR_USAGE = "The class which generates load for the tool."
142 + " Any args for this class can be passed as colon separated after class name";
143
144 public static final String OPT_WRITER = "writer";
145 public static final String OPT_WRITER_USAGE = "The class for executing the write requests";
146
147 public static final String OPT_UPDATER = "updater";
148 public static final String OPT_UPDATER_USAGE = "The class for executing the update requests";
149
150 public static final String OPT_READER = "reader";
151 public static final String OPT_READER_USAGE = "The class for executing the read requests";
152
153 protected static final String OPT_KEY_WINDOW = "key_window";
154 protected static final String OPT_WRITE = "write";
155 protected static final String OPT_MAX_READ_ERRORS = "max_read_errors";
156 public static final String OPT_MULTIPUT = "multiput";
157 public static final String OPT_MULTIGET = "multiget_batchsize";
158 protected static final String OPT_NUM_KEYS = "num_keys";
159 protected static final String OPT_READ = "read";
160 protected static final String OPT_START_KEY = "start_key";
161 public static final String OPT_TABLE_NAME = "tn";
162 public static final String OPT_COLUMN_FAMILIES = "families";
163 protected static final String OPT_ZK_QUORUM = "zk";
164 protected static final String OPT_ZK_PARENT_NODE = "zk_root";
165 protected static final String OPT_SKIP_INIT = "skip_init";
166 protected static final String OPT_INIT_ONLY = "init_only";
167 protected static final String NUM_TABLES = "num_tables";
168 protected static final String OPT_BATCHUPDATE = "batchupdate";
169 protected static final String OPT_UPDATE = "update";
170
171 public static final String OPT_ENCRYPTION = "encryption";
172 protected static final String OPT_ENCRYPTION_USAGE =
173 "Enables transparent encryption on the test table, one of " +
174 Arrays.toString(Encryption.getSupportedCiphers());
175
176 public static final String OPT_NUM_REGIONS_PER_SERVER = "num_regions_per_server";
177 protected static final String OPT_NUM_REGIONS_PER_SERVER_USAGE
178 = "Desired number of regions per region server. Defaults to 5.";
179 public static int DEFAULT_NUM_REGIONS_PER_SERVER = 5;
180
181 public static final String OPT_REGION_REPLICATION = "region_replication";
182 protected static final String OPT_REGION_REPLICATION_USAGE =
183 "Desired number of replicas per region";
184
185 public static final String OPT_REGION_REPLICA_ID = "region_replica_id";
186 protected static final String OPT_REGION_REPLICA_ID_USAGE =
187 "Region replica id to do the reads from";
188
189 protected static final long DEFAULT_START_KEY = 0;
190
191
192 protected CommandLine cmd;
193
194 protected MultiThreadedWriter writerThreads = null;
195 protected MultiThreadedReader readerThreads = null;
196 protected MultiThreadedUpdater updaterThreads = null;
197
198 protected long startKey, endKey;
199
200 protected boolean isVerbose, isWrite, isRead, isUpdate;
201 protected boolean deferredLogFlush;
202
203
204 protected DataBlockEncoding dataBlockEncodingAlgo;
205 protected Compression.Algorithm compressAlgo;
206 protected BloomType bloomType;
207 private boolean inMemoryCF;
208
209 private User userOwner;
210
211 protected int numWriterThreads = DEFAULT_NUM_THREADS;
212 protected int minColsPerKey, maxColsPerKey;
213 protected int minColDataSize = DEFAULT_DATA_SIZE, maxColDataSize = DEFAULT_DATA_SIZE;
214 protected boolean isMultiPut;
215
216
217 protected int numUpdaterThreads = DEFAULT_NUM_THREADS;
218 protected int updatePercent;
219 protected boolean ignoreConflicts = false;
220 protected boolean isBatchUpdate;
221
222
223 private int numReaderThreads = DEFAULT_NUM_THREADS;
224 private int keyWindow = MultiThreadedReader.DEFAULT_KEY_WINDOW;
225 private int multiGetBatchSize = MultiThreadedReader.DEFAULT_BATCH_SIZE;
226 private int maxReadErrors = MultiThreadedReader.DEFAULT_MAX_ERRORS;
227 private int verifyPercent;
228
229 private int numTables = 1;
230
231 private String superUser;
232
233 private String userNames;
234
235 private String authnFileName;
236
237 private int numRegionsPerServer = DEFAULT_NUM_REGIONS_PER_SERVER;
238 private int regionReplication = -1;
239 private int regionReplicaId = -1;
240
241
242
243 protected boolean isSkipInit = false;
244 protected boolean isInitOnly = false;
245
246 protected Cipher cipher = null;
247
248 protected String[] splitColonSeparated(String option,
249 int minNumCols, int maxNumCols) {
250 String optVal = cmd.getOptionValue(option);
251 String[] cols = optVal.split(COLON);
252 if (cols.length < minNumCols || cols.length > maxNumCols) {
253 throw new IllegalArgumentException("Expected at least "
254 + minNumCols + " columns but no more than " + maxNumCols +
255 " in the colon-separated value '" + optVal + "' of the " +
256 "-" + option + " option");
257 }
258 return cols;
259 }
260
261 protected int getNumThreads(String numThreadsStr) {
262 return parseInt(numThreadsStr, 1, Short.MAX_VALUE);
263 }
264
265 public byte[][] getColumnFamilies() {
266 return families;
267 }
268
269
270
271
272
273 protected void applyColumnFamilyOptions(TableName tableName,
274 byte[][] columnFamilies) throws IOException {
275 Admin admin = new HBaseAdmin(conf);
276 HTableDescriptor tableDesc = admin.getTableDescriptor(tableName);
277 LOG.info("Disabling table " + tableName);
278 admin.disableTable(tableName);
279 for (byte[] cf : columnFamilies) {
280 HColumnDescriptor columnDesc = tableDesc.getFamily(cf);
281 boolean isNewCf = columnDesc == null;
282 if (isNewCf) {
283 columnDesc = new HColumnDescriptor(cf);
284 }
285 if (bloomType != null) {
286 columnDesc.setBloomFilterType(bloomType);
287 }
288 if (compressAlgo != null) {
289 columnDesc.setCompressionType(compressAlgo);
290 }
291 if (dataBlockEncodingAlgo != null) {
292 columnDesc.setDataBlockEncoding(dataBlockEncodingAlgo);
293 }
294 if (inMemoryCF) {
295 columnDesc.setInMemory(inMemoryCF);
296 }
297 if (cipher != null) {
298 byte[] keyBytes = new byte[cipher.getKeyLength()];
299 new SecureRandom().nextBytes(keyBytes);
300 columnDesc.setEncryptionType(cipher.getName());
301 columnDesc.setEncryptionKey(EncryptionUtil.wrapKey(conf,
302 User.getCurrent().getShortName(),
303 new SecretKeySpec(keyBytes, cipher.getName())));
304 }
305 if (isNewCf) {
306 admin.addColumn(tableName, columnDesc);
307 } else {
308 admin.modifyColumn(tableName, columnDesc);
309 }
310 }
311 LOG.info("Enabling table " + tableName);
312 admin.enableTable(tableName);
313 admin.close();
314 }
315
316 @Override
317 protected void addOptions() {
318 addOptNoArg("v", OPT_VERBOSE, "Will display a full readout of logs, including ZooKeeper");
319 addOptWithArg(OPT_ZK_QUORUM, "ZK quorum as comma-separated host names " +
320 "without port numbers");
321 addOptWithArg(OPT_ZK_PARENT_NODE, "name of parent znode in zookeeper");
322 addOptWithArg(OPT_TABLE_NAME, "The name of the table to read or write");
323 addOptWithArg(OPT_COLUMN_FAMILIES, "The name of the column families to use separated by comma");
324 addOptWithArg(OPT_WRITE, OPT_USAGE_LOAD);
325 addOptWithArg(OPT_READ, OPT_USAGE_READ);
326 addOptWithArg(OPT_UPDATE, OPT_USAGE_UPDATE);
327 addOptNoArg(OPT_INIT_ONLY, "Initialize the test table only, don't do any loading");
328 addOptWithArg(OPT_BLOOM, OPT_USAGE_BLOOM);
329 addOptWithArg(OPT_COMPRESSION, OPT_USAGE_COMPRESSION);
330 addOptWithArg(OPT_DATA_BLOCK_ENCODING, OPT_DATA_BLOCK_ENCODING_USAGE);
331 addOptWithArg(OPT_MAX_READ_ERRORS, "The maximum number of read errors " +
332 "to tolerate before terminating all reader threads. The default is " +
333 MultiThreadedReader.DEFAULT_MAX_ERRORS + ".");
334 addOptWithArg(OPT_MULTIGET, "Whether to use multi-gets as opposed to " +
335 "separate gets for every column in a row");
336 addOptWithArg(OPT_KEY_WINDOW, "The 'key window' to maintain between " +
337 "reads and writes for concurrent write/read workload. The default " +
338 "is " + MultiThreadedReader.DEFAULT_KEY_WINDOW + ".");
339
340 addOptNoArg(OPT_MULTIPUT, "Whether to use multi-puts as opposed to " +
341 "separate puts for every column in a row");
342 addOptNoArg(OPT_BATCHUPDATE, "Whether to use batch as opposed to " +
343 "separate updates for every column in a row");
344 addOptNoArg(OPT_INMEMORY, OPT_USAGE_IN_MEMORY);
345 addOptWithArg(OPT_GENERATOR, OPT_GENERATOR_USAGE);
346 addOptWithArg(OPT_WRITER, OPT_WRITER_USAGE);
347 addOptWithArg(OPT_UPDATER, OPT_UPDATER_USAGE);
348 addOptWithArg(OPT_READER, OPT_READER_USAGE);
349
350 addOptWithArg(OPT_NUM_KEYS, "The number of keys to read/write");
351 addOptWithArg(OPT_START_KEY, "The first key to read/write " +
352 "(a 0-based index). The default value is " +
353 DEFAULT_START_KEY + ".");
354 addOptNoArg(OPT_SKIP_INIT, "Skip the initialization; assume test table "
355 + "already exists");
356
357 addOptWithArg(NUM_TABLES,
358 "A positive integer number. When a number n is speicfied, load test "
359 + "tool will load n table parallely. -tn parameter value becomes "
360 + "table name prefix. Each table name is in format <tn>_1...<tn>_n");
361
362 addOptWithArg(OPT_ENCRYPTION, OPT_ENCRYPTION_USAGE);
363 addOptNoArg(OPT_DEFERRED_LOG_FLUSH, OPT_DEFERRED_LOG_FLUSH_USAGE);
364 addOptWithArg(OPT_NUM_REGIONS_PER_SERVER, OPT_NUM_REGIONS_PER_SERVER_USAGE);
365 addOptWithArg(OPT_REGION_REPLICATION, OPT_REGION_REPLICATION_USAGE);
366 addOptWithArg(OPT_REGION_REPLICA_ID, OPT_REGION_REPLICA_ID_USAGE);
367 }
368
369 @Override
370 protected void processOptions(CommandLine cmd) {
371 this.cmd = cmd;
372
373 tableName = TableName.valueOf(cmd.getOptionValue(OPT_TABLE_NAME,
374 DEFAULT_TABLE_NAME));
375
376 if (cmd.hasOption(OPT_COLUMN_FAMILIES)) {
377 String[] list = cmd.getOptionValue(OPT_COLUMN_FAMILIES).split(",");
378 families = new byte[list.length][];
379 for (int i = 0; i < list.length; i++) {
380 families[i] = Bytes.toBytes(list[i]);
381 }
382 } else {
383 families = DEFAULT_COLUMN_FAMILIES;
384 }
385
386 isVerbose = cmd.hasOption(OPT_VERBOSE);
387 isWrite = cmd.hasOption(OPT_WRITE);
388 isRead = cmd.hasOption(OPT_READ);
389 isUpdate = cmd.hasOption(OPT_UPDATE);
390 isInitOnly = cmd.hasOption(OPT_INIT_ONLY);
391 deferredLogFlush = cmd.hasOption(OPT_DEFERRED_LOG_FLUSH);
392
393 if (!isWrite && !isRead && !isUpdate && !isInitOnly) {
394 throw new IllegalArgumentException("Either -" + OPT_WRITE + " or " +
395 "-" + OPT_UPDATE + " or -" + OPT_READ + " has to be specified");
396 }
397
398 if (isInitOnly && (isRead || isWrite || isUpdate)) {
399 throw new IllegalArgumentException(OPT_INIT_ONLY + " cannot be specified with"
400 + " either -" + OPT_WRITE + " or -" + OPT_UPDATE + " or -" + OPT_READ);
401 }
402
403 if (!isInitOnly) {
404 if (!cmd.hasOption(OPT_NUM_KEYS)) {
405 throw new IllegalArgumentException(OPT_NUM_KEYS + " must be specified in "
406 + "read or write mode");
407 }
408 startKey = parseLong(cmd.getOptionValue(OPT_START_KEY,
409 String.valueOf(DEFAULT_START_KEY)), 0, Long.MAX_VALUE);
410 long numKeys = parseLong(cmd.getOptionValue(OPT_NUM_KEYS), 1,
411 Long.MAX_VALUE - startKey);
412 endKey = startKey + numKeys;
413 isSkipInit = cmd.hasOption(OPT_SKIP_INIT);
414 System.out.println("Key range: [" + startKey + ".." + (endKey - 1) + "]");
415 }
416
417 parseColumnFamilyOptions(cmd);
418
419 if (isWrite) {
420 String[] writeOpts = splitColonSeparated(OPT_WRITE, 2, 3);
421
422 int colIndex = 0;
423 minColsPerKey = 1;
424 maxColsPerKey = 2 * Integer.parseInt(writeOpts[colIndex++]);
425 int avgColDataSize =
426 parseInt(writeOpts[colIndex++], 1, Integer.MAX_VALUE);
427 minColDataSize = avgColDataSize / 2;
428 maxColDataSize = avgColDataSize * 3 / 2;
429
430 if (colIndex < writeOpts.length) {
431 numWriterThreads = getNumThreads(writeOpts[colIndex++]);
432 }
433
434 isMultiPut = cmd.hasOption(OPT_MULTIPUT);
435
436 System.out.println("Multi-puts: " + isMultiPut);
437 System.out.println("Columns per key: " + minColsPerKey + ".."
438 + maxColsPerKey);
439 System.out.println("Data size per column: " + minColDataSize + ".."
440 + maxColDataSize);
441 }
442
443 if (isUpdate) {
444 String[] mutateOpts = splitColonSeparated(OPT_UPDATE, 1, 3);
445 int colIndex = 0;
446 updatePercent = parseInt(mutateOpts[colIndex++], 0, 100);
447 if (colIndex < mutateOpts.length) {
448 numUpdaterThreads = getNumThreads(mutateOpts[colIndex++]);
449 }
450 if (colIndex < mutateOpts.length) {
451 ignoreConflicts = parseInt(mutateOpts[colIndex++], 0, 1) == 1;
452 }
453
454 isBatchUpdate = cmd.hasOption(OPT_BATCHUPDATE);
455
456 System.out.println("Batch updates: " + isBatchUpdate);
457 System.out.println("Percent of keys to update: " + updatePercent);
458 System.out.println("Updater threads: " + numUpdaterThreads);
459 System.out.println("Ignore nonce conflicts: " + ignoreConflicts);
460 }
461
462 if (isRead) {
463 String[] readOpts = splitColonSeparated(OPT_READ, 1, 2);
464 int colIndex = 0;
465 verifyPercent = parseInt(readOpts[colIndex++], 0, 100);
466 if (colIndex < readOpts.length) {
467 numReaderThreads = getNumThreads(readOpts[colIndex++]);
468 }
469
470 if (cmd.hasOption(OPT_MAX_READ_ERRORS)) {
471 maxReadErrors = parseInt(cmd.getOptionValue(OPT_MAX_READ_ERRORS),
472 0, Integer.MAX_VALUE);
473 }
474
475 if (cmd.hasOption(OPT_KEY_WINDOW)) {
476 keyWindow = parseInt(cmd.getOptionValue(OPT_KEY_WINDOW),
477 0, Integer.MAX_VALUE);
478 }
479
480 if (cmd.hasOption(OPT_MULTIGET)) {
481 multiGetBatchSize = parseInt(cmd.getOptionValue(OPT_MULTIGET),
482 0, Integer.MAX_VALUE);
483 }
484
485 System.out.println("Multi-gets (value of 1 means no multigets): " + multiGetBatchSize);
486 System.out.println("Percent of keys to verify: " + verifyPercent);
487 System.out.println("Reader threads: " + numReaderThreads);
488 }
489
490 numTables = 1;
491 if (cmd.hasOption(NUM_TABLES)) {
492 numTables = parseInt(cmd.getOptionValue(NUM_TABLES), 1, Short.MAX_VALUE);
493 }
494
495 numRegionsPerServer = DEFAULT_NUM_REGIONS_PER_SERVER;
496 if (cmd.hasOption(OPT_NUM_REGIONS_PER_SERVER)) {
497 numRegionsPerServer = Integer.parseInt(cmd.getOptionValue(OPT_NUM_REGIONS_PER_SERVER));
498 }
499
500 regionReplication = 1;
501 if (cmd.hasOption(OPT_REGION_REPLICATION)) {
502 regionReplication = Integer.parseInt(cmd.getOptionValue(OPT_REGION_REPLICATION));
503 }
504
505 regionReplicaId = -1;
506 if (cmd.hasOption(OPT_REGION_REPLICA_ID)) {
507 regionReplicaId = Integer.parseInt(cmd.getOptionValue(OPT_REGION_REPLICA_ID));
508 }
509 }
510
511 private void parseColumnFamilyOptions(CommandLine cmd) {
512 String dataBlockEncodingStr = cmd.getOptionValue(OPT_DATA_BLOCK_ENCODING);
513 dataBlockEncodingAlgo = dataBlockEncodingStr == null ? null :
514 DataBlockEncoding.valueOf(dataBlockEncodingStr);
515
516 String compressStr = cmd.getOptionValue(OPT_COMPRESSION);
517 compressAlgo = compressStr == null ? Compression.Algorithm.NONE :
518 Compression.Algorithm.valueOf(compressStr);
519
520 String bloomStr = cmd.getOptionValue(OPT_BLOOM);
521 bloomType = bloomStr == null ? BloomType.ROW :
522 BloomType.valueOf(bloomStr);
523
524 inMemoryCF = cmd.hasOption(OPT_INMEMORY);
525 if (cmd.hasOption(OPT_ENCRYPTION)) {
526 cipher = Encryption.getCipher(conf, cmd.getOptionValue(OPT_ENCRYPTION));
527 }
528
529 }
530
531 public void initTestTable() throws IOException {
532 Durability durability = Durability.USE_DEFAULT;
533 if (deferredLogFlush) {
534 durability = Durability.ASYNC_WAL;
535 }
536
537 HBaseTestingUtility.createPreSplitLoadTestTable(conf, tableName,
538 getColumnFamilies(), compressAlgo, dataBlockEncodingAlgo, numRegionsPerServer,
539 regionReplication, durability);
540 applyColumnFamilyOptions(tableName, getColumnFamilies());
541 }
542
543 @Override
544 protected int doWork() throws IOException {
545 if (!isVerbose) {
546 LogManager.getLogger(ZooKeeper.class.getName()).setLevel(Level.WARN);
547 }
548 if (numTables > 1) {
549 return parallelLoadTables();
550 } else {
551 return loadTable();
552 }
553 }
554
555 protected int loadTable() throws IOException {
556 if (cmd.hasOption(OPT_ZK_QUORUM)) {
557 conf.set(HConstants.ZOOKEEPER_QUORUM, cmd.getOptionValue(OPT_ZK_QUORUM));
558 }
559 if (cmd.hasOption(OPT_ZK_PARENT_NODE)) {
560 conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, cmd.getOptionValue(OPT_ZK_PARENT_NODE));
561 }
562
563 if (isInitOnly) {
564 LOG.info("Initializing only; no reads or writes");
565 initTestTable();
566 return 0;
567 }
568
569 if (!isSkipInit) {
570 initTestTable();
571 }
572 LoadTestDataGenerator dataGen = null;
573 if (cmd.hasOption(OPT_GENERATOR)) {
574 String[] clazzAndArgs = cmd.getOptionValue(OPT_GENERATOR).split(COLON);
575 dataGen = getLoadGeneratorInstance(clazzAndArgs[0]);
576 String[] args;
577 if (dataGen instanceof LoadTestDataGeneratorWithACL) {
578 LOG.info("Using LoadTestDataGeneratorWithACL");
579 if (User.isHBaseSecurityEnabled(conf)) {
580 LOG.info("Security is enabled");
581 authnFileName = clazzAndArgs[1];
582 superUser = clazzAndArgs[2];
583 userNames = clazzAndArgs[3];
584 args = Arrays.copyOfRange(clazzAndArgs, 2, clazzAndArgs.length);
585 Properties authConfig = new Properties();
586 authConfig.load(this.getClass().getClassLoader().getResourceAsStream(authnFileName));
587 try {
588 addAuthInfoToConf(authConfig, conf, superUser, userNames);
589 } catch (IOException exp) {
590 LOG.error(exp);
591 return EXIT_FAILURE;
592 }
593 userOwner = User.create(loginAndReturnUGI(conf, superUser));
594 } else {
595 superUser = clazzAndArgs[1];
596 userNames = clazzAndArgs[2];
597 args = Arrays.copyOfRange(clazzAndArgs, 1, clazzAndArgs.length);
598 userOwner = User.createUserForTesting(conf, superUser, new String[0]);
599 }
600 } else {
601 args = clazzAndArgs.length == 1 ? new String[0] : Arrays.copyOfRange(clazzAndArgs, 1,
602 clazzAndArgs.length);
603 }
604 dataGen.initialize(args);
605 } else {
606
607 dataGen = new MultiThreadedAction.DefaultDataGenerator(minColDataSize, maxColDataSize,
608 minColsPerKey, maxColsPerKey, families);
609 }
610
611 if (userOwner != null) {
612 LOG.info("Granting permissions for user " + userOwner.getShortName());
613 Permission.Action[] actions = {
614 Permission.Action.ADMIN, Permission.Action.CREATE,
615 Permission.Action.READ, Permission.Action.WRITE };
616 try {
617 AccessControlClient.grant(ConnectionFactory.createConnection(conf),
618 tableName, userOwner.getShortName(), null, null, actions);
619 } catch (Throwable e) {
620 LOG.fatal("Error in granting permission for the user " + userOwner.getShortName(), e);
621 return EXIT_FAILURE;
622 }
623 }
624
625 if (userNames != null) {
626
627 String users[] = userNames.split(",");
628 User user = null;
629 for (String userStr : users) {
630 if (User.isHBaseSecurityEnabled(conf)) {
631 user = User.create(loginAndReturnUGI(conf, userStr));
632 } else {
633 user = User.createUserForTesting(conf, userStr, new String[0]);
634 }
635 }
636 }
637
638 if (isWrite) {
639 if (userOwner != null) {
640 writerThreads = new MultiThreadedWriterWithACL(dataGen, conf, tableName, userOwner);
641 } else {
642 String writerClass = null;
643 if (cmd.hasOption(OPT_WRITER)) {
644 writerClass = cmd.getOptionValue(OPT_WRITER);
645 } else {
646 writerClass = MultiThreadedWriter.class.getCanonicalName();
647 }
648
649 writerThreads = getMultiThreadedWriterInstance(writerClass, dataGen);
650 }
651 writerThreads.setMultiPut(isMultiPut);
652 }
653
654 if (isUpdate) {
655 if (userOwner != null) {
656 updaterThreads = new MultiThreadedUpdaterWithACL(dataGen, conf, tableName, updatePercent,
657 userOwner, userNames);
658 } else {
659 String updaterClass = null;
660 if (cmd.hasOption(OPT_UPDATER)) {
661 updaterClass = cmd.getOptionValue(OPT_UPDATER);
662 } else {
663 updaterClass = MultiThreadedUpdater.class.getCanonicalName();
664 }
665 updaterThreads = getMultiThreadedUpdaterInstance(updaterClass, dataGen);
666 }
667 updaterThreads.setBatchUpdate(isBatchUpdate);
668 updaterThreads.setIgnoreNonceConflicts(ignoreConflicts);
669 }
670
671 if (isRead) {
672 if (userOwner != null) {
673 readerThreads = new MultiThreadedReaderWithACL(dataGen, conf, tableName, verifyPercent,
674 userNames);
675 } else {
676 String readerClass = null;
677 if (cmd.hasOption(OPT_READER)) {
678 readerClass = cmd.getOptionValue(OPT_READER);
679 } else {
680 readerClass = MultiThreadedReader.class.getCanonicalName();
681 }
682 readerThreads = getMultiThreadedReaderInstance(readerClass, dataGen);
683 }
684 readerThreads.setMaxErrors(maxReadErrors);
685 readerThreads.setKeyWindow(keyWindow);
686 readerThreads.setMultiGetBatchSize(multiGetBatchSize);
687 readerThreads.setRegionReplicaId(regionReplicaId);
688 }
689
690 if (isUpdate && isWrite) {
691 LOG.info("Concurrent write/update workload: making updaters aware of the " +
692 "write point");
693 updaterThreads.linkToWriter(writerThreads);
694 }
695
696 if (isRead && (isUpdate || isWrite)) {
697 LOG.info("Concurrent write/read workload: making readers aware of the " +
698 "write point");
699 readerThreads.linkToWriter(isUpdate ? updaterThreads : writerThreads);
700 }
701
702 if (isWrite) {
703 System.out.println("Starting to write data...");
704 writerThreads.start(startKey, endKey, numWriterThreads);
705 }
706
707 if (isUpdate) {
708 LOG.info("Starting to mutate data...");
709 System.out.println("Starting to mutate data...");
710
711
712 updaterThreads.start(startKey, endKey, numUpdaterThreads);
713 }
714
715 if (isRead) {
716 System.out.println("Starting to read data...");
717 readerThreads.start(startKey, endKey, numReaderThreads);
718 }
719
720 if (isWrite) {
721 writerThreads.waitForFinish();
722 }
723
724 if (isUpdate) {
725 updaterThreads.waitForFinish();
726 }
727
728 if (isRead) {
729 readerThreads.waitForFinish();
730 }
731
732 boolean success = true;
733 if (isWrite) {
734 success = success && writerThreads.getNumWriteFailures() == 0;
735 }
736 if (isUpdate) {
737 success = success && updaterThreads.getNumWriteFailures() == 0;
738 }
739 if (isRead) {
740 success = success && readerThreads.getNumReadErrors() == 0
741 && readerThreads.getNumReadFailures() == 0;
742 }
743 return success ? EXIT_SUCCESS : EXIT_FAILURE;
744 }
745
746 private LoadTestDataGenerator getLoadGeneratorInstance(String clazzName) throws IOException {
747 try {
748 Class<?> clazz = Class.forName(clazzName);
749 Constructor<?> constructor = clazz.getConstructor(int.class, int.class, int.class, int.class,
750 byte[][].class);
751 return (LoadTestDataGenerator) constructor.newInstance(minColDataSize, maxColDataSize,
752 minColsPerKey, maxColsPerKey, families);
753 } catch (Exception e) {
754 throw new IOException(e);
755 }
756 }
757
758 private MultiThreadedWriter getMultiThreadedWriterInstance(String clazzName
759 , LoadTestDataGenerator dataGen) throws IOException {
760 try {
761 Class<?> clazz = Class.forName(clazzName);
762 Constructor<?> constructor = clazz.getConstructor(
763 LoadTestDataGenerator.class, Configuration.class, TableName.class);
764 return (MultiThreadedWriter) constructor.newInstance(dataGen, conf, tableName);
765 } catch (Exception e) {
766 throw new IOException(e);
767 }
768 }
769
770 private MultiThreadedUpdater getMultiThreadedUpdaterInstance(String clazzName
771 , LoadTestDataGenerator dataGen) throws IOException {
772 try {
773 Class<?> clazz = Class.forName(clazzName);
774 Constructor<?> constructor = clazz.getConstructor(
775 LoadTestDataGenerator.class, Configuration.class, TableName.class, double.class);
776 return (MultiThreadedUpdater) constructor.newInstance(
777 dataGen, conf, tableName, updatePercent);
778 } catch (Exception e) {
779 throw new IOException(e);
780 }
781 }
782
783 private MultiThreadedReader getMultiThreadedReaderInstance(String clazzName
784 , LoadTestDataGenerator dataGen) throws IOException {
785 try {
786 Class<?> clazz = Class.forName(clazzName);
787 Constructor<?> constructor = clazz.getConstructor(
788 LoadTestDataGenerator.class, Configuration.class, TableName.class, double.class);
789 return (MultiThreadedReader) constructor.newInstance(dataGen, conf, tableName, verifyPercent);
790 } catch (Exception e) {
791 throw new IOException(e);
792 }
793 }
794
795 public static byte[] generateData(final Random r, int length) {
796 byte [] b = new byte [length];
797 int i = 0;
798
799 for(i = 0; i < (length-8); i += 8) {
800 b[i] = (byte) (65 + r.nextInt(26));
801 b[i+1] = b[i];
802 b[i+2] = b[i];
803 b[i+3] = b[i];
804 b[i+4] = b[i];
805 b[i+5] = b[i];
806 b[i+6] = b[i];
807 b[i+7] = b[i];
808 }
809
810 byte a = (byte) (65 + r.nextInt(26));
811 for(; i < length; i++) {
812 b[i] = a;
813 }
814 return b;
815 }
816 public static void main(String[] args) {
817 new LoadTestTool().doStaticMain(args);
818 }
819
820
821
822
823
824
825
826
827
828 private int parallelLoadTables()
829 throws IOException {
830
831 String tableName = cmd.getOptionValue(OPT_TABLE_NAME, DEFAULT_TABLE_NAME);
832 String[] newArgs = null;
833 if (!cmd.hasOption(LoadTestTool.OPT_TABLE_NAME)) {
834 newArgs = new String[cmdLineArgs.length + 2];
835 newArgs[0] = "-" + LoadTestTool.OPT_TABLE_NAME;
836 newArgs[1] = LoadTestTool.DEFAULT_TABLE_NAME;
837 System.arraycopy(cmdLineArgs, 0, newArgs, 2, cmdLineArgs.length);
838 } else {
839 newArgs = cmdLineArgs;
840 }
841
842 int tableNameValueIndex = -1;
843 for (int j = 0; j < newArgs.length; j++) {
844 if (newArgs[j].endsWith(OPT_TABLE_NAME)) {
845 tableNameValueIndex = j + 1;
846 } else if (newArgs[j].endsWith(NUM_TABLES)) {
847
848 newArgs[j + 1] = "1";
849 }
850 }
851
852
853 List<WorkerThread> workers = new ArrayList<WorkerThread>();
854 for (int i = 0; i < numTables; i++) {
855 String[] workerArgs = newArgs.clone();
856 workerArgs[tableNameValueIndex] = tableName + "_" + (i+1);
857 WorkerThread worker = new WorkerThread(i, workerArgs);
858 workers.add(worker);
859 LOG.info(worker + " starting");
860 worker.start();
861 }
862
863
864 LOG.info("Waiting for worker threads to finish");
865 for (WorkerThread t : workers) {
866 try {
867 t.join();
868 } catch (InterruptedException ie) {
869 IOException iie = new InterruptedIOException();
870 iie.initCause(ie);
871 throw iie;
872 }
873 checkForErrors();
874 }
875
876 return EXIT_SUCCESS;
877 }
878
879
880
881 protected AtomicReference<Throwable> thrown = new AtomicReference<Throwable>();
882
883 private void workerThreadError(Throwable t) {
884 thrown.compareAndSet(null, t);
885 }
886
887
888
889
890 private void checkForErrors() throws IOException {
891 Throwable thrown = this.thrown.get();
892 if (thrown == null) return;
893 if (thrown instanceof IOException) {
894 throw (IOException) thrown;
895 } else {
896 throw new RuntimeException(thrown);
897 }
898 }
899
900 class WorkerThread extends Thread {
901 private String[] workerArgs;
902
903 WorkerThread(int i, String[] args) {
904 super("WorkerThread-" + i);
905 workerArgs = args;
906 }
907
908 @Override
909 public void run() {
910 try {
911 int ret = ToolRunner.run(HBaseConfiguration.create(), new LoadTestTool(), workerArgs);
912 if (ret != 0) {
913 throw new RuntimeException("LoadTestTool exit with non-zero return code.");
914 }
915 } catch (Exception ex) {
916 LOG.error("Error in worker thread", ex);
917 workerThreadError(ex);
918 }
919 }
920 }
921
922 private void addAuthInfoToConf(Properties authConfig, Configuration conf, String owner,
923 String userList) throws IOException {
924 List<String> users = new ArrayList(Arrays.asList(userList.split(",")));
925 users.add(owner);
926 for (String user : users) {
927 String keyTabFileConfKey = "hbase." + user + ".keytab.file";
928 String principalConfKey = "hbase." + user + ".kerberos.principal";
929 if (!authConfig.containsKey(keyTabFileConfKey) || !authConfig.containsKey(principalConfKey)) {
930 throw new IOException("Authentication configs missing for user : " + user);
931 }
932 }
933 for (String key : authConfig.stringPropertyNames()) {
934 conf.set(key, authConfig.getProperty(key));
935 }
936 LOG.debug("Added authentication properties to config successfully.");
937 }
938
939 public static UserGroupInformation loginAndReturnUGI(Configuration conf, String username)
940 throws IOException {
941 String hostname = InetAddress.getLocalHost().getHostName();
942 String keyTabFileConfKey = "hbase." + username + ".keytab.file";
943 String keyTabFileLocation = conf.get(keyTabFileConfKey);
944 String principalConfKey = "hbase." + username + ".kerberos.principal";
945 String principal = SecurityUtil.getServerPrincipal(conf.get(principalConfKey), hostname);
946 if (keyTabFileLocation == null || principal == null) {
947 LOG.warn("Principal or key tab file null for : " + principalConfKey + ", "
948 + keyTabFileConfKey);
949 }
950 UserGroupInformation ugi =
951 UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keyTabFileLocation);
952 return ugi;
953 }
954 }