1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.tool;
21
22 import java.io.Closeable;
23 import java.io.IOException;
24 import java.util.ArrayList;
25 import java.util.Arrays;
26 import java.util.HashMap;
27 import java.util.HashSet;
28 import java.util.LinkedList;
29 import java.util.List;
30 import java.util.Map;
31 import java.util.Random;
32 import java.util.Set;
33 import java.util.TreeSet;
34 import java.util.concurrent.atomic.AtomicLong;
35 import java.util.concurrent.Callable;
36 import java.util.concurrent.ExecutionException;
37 import java.util.concurrent.ExecutorService;
38 import java.util.concurrent.Future;
39 import java.util.concurrent.ScheduledThreadPoolExecutor;
40 import java.util.regex.Matcher;
41 import java.util.regex.Pattern;
42
43 import org.apache.commons.lang.time.StopWatch;
44 import org.apache.commons.logging.Log;
45 import org.apache.commons.logging.LogFactory;
46 import org.apache.hadoop.conf.Configuration;
47 import org.apache.hadoop.hbase.AuthUtil;
48 import org.apache.hadoop.hbase.ChoreService;
49 import org.apache.hadoop.hbase.DoNotRetryIOException;
50 import org.apache.hadoop.hbase.HBaseConfiguration;
51 import org.apache.hadoop.hbase.HColumnDescriptor;
52 import org.apache.hadoop.hbase.HConstants;
53 import org.apache.hadoop.hbase.HRegionInfo;
54 import org.apache.hadoop.hbase.HRegionLocation;
55 import org.apache.hadoop.hbase.HTableDescriptor;
56 import org.apache.hadoop.hbase.NamespaceDescriptor;
57 import org.apache.hadoop.hbase.ScheduledChore;
58 import org.apache.hadoop.hbase.ServerName;
59 import org.apache.hadoop.hbase.TableName;
60 import org.apache.hadoop.hbase.TableNotEnabledException;
61 import org.apache.hadoop.hbase.TableNotFoundException;
62 import org.apache.hadoop.hbase.client.Admin;
63 import org.apache.hadoop.hbase.client.Connection;
64 import org.apache.hadoop.hbase.client.ConnectionFactory;
65 import org.apache.hadoop.hbase.client.Get;
66 import org.apache.hadoop.hbase.client.Put;
67 import org.apache.hadoop.hbase.client.RegionLocator;
68 import org.apache.hadoop.hbase.client.ResultScanner;
69 import org.apache.hadoop.hbase.client.Scan;
70 import org.apache.hadoop.hbase.client.Table;
71 import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
72 import org.apache.hadoop.hbase.tool.Canary.RegionTask.TaskType;
73 import org.apache.hadoop.hbase.util.Bytes;
74 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
75 import org.apache.hadoop.hbase.util.ReflectionUtils;
76 import org.apache.hadoop.hbase.util.RegionSplitter;
77 import org.apache.hadoop.util.GenericOptionsParser;
78 import org.apache.hadoop.util.Tool;
79 import org.apache.hadoop.util.ToolRunner;
80
81
82
83
84
85
86
87
88
89
90
91
92 public final class Canary implements Tool {
93
94 public interface Sink {
95 public void publishReadFailure(HRegionInfo region, Exception e);
96 public void publishReadFailure(HRegionInfo region, HColumnDescriptor column, Exception e);
97 public void publishReadTiming(HRegionInfo region, HColumnDescriptor column, long msTime);
98 public void publishWriteFailure(HRegionInfo region, Exception e);
99 public void publishWriteFailure(HRegionInfo region, HColumnDescriptor column, Exception e);
100 public void publishWriteTiming(HRegionInfo region, HColumnDescriptor column, long msTime);
101 }
102
103
104 public interface ExtendedSink extends Sink {
105 public void publishReadFailure(String table, String server);
106 public void publishReadTiming(String table, String server, long msTime);
107 }
108
109
110
111 public static class StdOutSink implements Sink {
112 @Override
113 public void publishReadFailure(HRegionInfo region, Exception e) {
114 LOG.error(String.format("read from region %s failed", region.getRegionNameAsString()), e);
115 }
116
117 @Override
118 public void publishReadFailure(HRegionInfo region, HColumnDescriptor column, Exception e) {
119 LOG.error(String.format("read from region %s column family %s failed",
120 region.getRegionNameAsString(), column.getNameAsString()), e);
121 }
122
123 @Override
124 public void publishReadTiming(HRegionInfo region, HColumnDescriptor column, long msTime) {
125 LOG.info(String.format("read from region %s column family %s in %dms",
126 region.getRegionNameAsString(), column.getNameAsString(), msTime));
127 }
128
129 @Override
130 public void publishWriteFailure(HRegionInfo region, Exception e) {
131 LOG.error(String.format("write to region %s failed", region.getRegionNameAsString()), e);
132 }
133
134 @Override
135 public void publishWriteFailure(HRegionInfo region, HColumnDescriptor column, Exception e) {
136 LOG.error(String.format("write to region %s column family %s failed",
137 region.getRegionNameAsString(), column.getNameAsString()), e);
138 }
139
140 @Override
141 public void publishWriteTiming(HRegionInfo region, HColumnDescriptor column, long msTime) {
142 LOG.info(String.format("write to region %s column family %s in %dms",
143 region.getRegionNameAsString(), column.getNameAsString(), msTime));
144 }
145 }
146
147 public static class RegionServerStdOutSink extends StdOutSink implements ExtendedSink {
148
149 @Override
150 public void publishReadFailure(String table, String server) {
151 LOG.error(String.format("Read from table:%s on region server:%s", table, server));
152 }
153
154 @Override
155 public void publishReadTiming(String table, String server, long msTime) {
156 LOG.info(String.format("Read from table:%s on region server:%s in %dms",
157 table, server, msTime));
158 }
159 }
160
161
162
163
164
165 static class RegionTask implements Callable<Void> {
166 public enum TaskType{
167 READ, WRITE
168 }
169 private Connection connection;
170 private HRegionInfo region;
171 private Sink sink;
172 private TaskType taskType;
173
174 RegionTask(Connection connection, HRegionInfo region, Sink sink, TaskType taskType) {
175 this.connection = connection;
176 this.region = region;
177 this.sink = sink;
178 this.taskType = taskType;
179 }
180
181 @Override
182 public Void call() {
183 switch (taskType) {
184 case READ:
185 return read();
186 case WRITE:
187 return write();
188 default:
189 return read();
190 }
191 }
192
193 public Void read() {
194 Table table = null;
195 HTableDescriptor tableDesc = null;
196 try {
197 table = connection.getTable(region.getTable());
198 tableDesc = table.getTableDescriptor();
199 } catch (IOException e) {
200 LOG.debug("sniffRegion failed", e);
201 sink.publishReadFailure(region, e);
202 if (table != null) {
203 try {
204 table.close();
205 } catch (IOException ioe) {
206 LOG.error("Close table failed", e);
207 }
208 }
209 return null;
210 }
211
212 byte[] startKey = null;
213 Get get = null;
214 Scan scan = null;
215 ResultScanner rs = null;
216 StopWatch stopWatch = new StopWatch();
217 for (HColumnDescriptor column : tableDesc.getColumnFamilies()) {
218 stopWatch.reset();
219 startKey = region.getStartKey();
220
221 if (startKey.length > 0) {
222 get = new Get(startKey);
223 get.setCacheBlocks(false);
224 get.setFilter(new FirstKeyOnlyFilter());
225 get.addFamily(column.getName());
226 } else {
227 scan = new Scan();
228 scan.setCaching(1);
229 scan.setCacheBlocks(false);
230 scan.setFilter(new FirstKeyOnlyFilter());
231 scan.addFamily(column.getName());
232 scan.setMaxResultSize(1L);
233 }
234
235 try {
236 if (startKey.length > 0) {
237 stopWatch.start();
238 table.get(get);
239 stopWatch.stop();
240 sink.publishReadTiming(region, column, stopWatch.getTime());
241 } else {
242 stopWatch.start();
243 rs = table.getScanner(scan);
244 stopWatch.stop();
245 sink.publishReadTiming(region, column, stopWatch.getTime());
246 }
247 } catch (Exception e) {
248 sink.publishReadFailure(region, column, e);
249 } finally {
250 if (rs != null) {
251 rs.close();
252 }
253 scan = null;
254 get = null;
255 startKey = null;
256 }
257 }
258 try {
259 table.close();
260 } catch (IOException e) {
261 LOG.error("Close table failed", e);
262 }
263 return null;
264 }
265
266
267
268
269
270 private Void write() {
271 Table table = null;
272 HTableDescriptor tableDesc = null;
273 try {
274 table = connection.getTable(region.getTable());
275 tableDesc = table.getTableDescriptor();
276 byte[] rowToCheck = region.getStartKey();
277 if (rowToCheck.length == 0) {
278 rowToCheck = new byte[]{0x0};
279 }
280 int writeValueSize =
281 connection.getConfiguration().getInt(HConstants.HBASE_CANARY_WRITE_VALUE_SIZE_KEY, 10);
282 for (HColumnDescriptor column : tableDesc.getColumnFamilies()) {
283 Put put = new Put(rowToCheck);
284 byte[] value = new byte[writeValueSize];
285 Bytes.random(value);
286 put.addColumn(column.getName(), HConstants.EMPTY_BYTE_ARRAY, value);
287 try {
288 long startTime = System.currentTimeMillis();
289 table.put(put);
290 long time = System.currentTimeMillis() - startTime;
291 sink.publishWriteTiming(region, column, time);
292 } catch (Exception e) {
293 sink.publishWriteFailure(region, column, e);
294 }
295 }
296 table.close();
297 } catch (IOException e) {
298 sink.publishWriteFailure(region, e);
299 }
300 return null;
301 }
302 }
303
304
305
306
307 static class RegionServerTask implements Callable<Void> {
308 private Connection connection;
309 private String serverName;
310 private HRegionInfo region;
311 private ExtendedSink sink;
312 private AtomicLong successes;
313
314 RegionServerTask(Connection connection, String serverName, HRegionInfo region,
315 ExtendedSink sink, AtomicLong successes) {
316 this.connection = connection;
317 this.serverName = serverName;
318 this.region = region;
319 this.sink = sink;
320 this.successes = successes;
321 }
322
323 @Override
324 public Void call() {
325 TableName tableName = null;
326 Table table = null;
327 Get get = null;
328 byte[] startKey = null;
329 Scan scan = null;
330 StopWatch stopWatch = new StopWatch();
331
332 stopWatch.reset();
333 try {
334 tableName = region.getTable();
335 table = connection.getTable(tableName);
336 startKey = region.getStartKey();
337
338 if (startKey.length > 0) {
339 get = new Get(startKey);
340 get.setCacheBlocks(false);
341 get.setFilter(new FirstKeyOnlyFilter());
342 stopWatch.start();
343 table.get(get);
344 stopWatch.stop();
345 } else {
346 scan = new Scan();
347 scan.setCacheBlocks(false);
348 scan.setFilter(new FirstKeyOnlyFilter());
349 scan.setCaching(1);
350 scan.setMaxResultSize(1L);
351 stopWatch.start();
352 ResultScanner s = table.getScanner(scan);
353 s.close();
354 stopWatch.stop();
355 }
356 successes.incrementAndGet();
357 sink.publishReadTiming(tableName.getNameAsString(), serverName, stopWatch.getTime());
358 } catch (TableNotFoundException tnfe) {
359 LOG.error("Table may be deleted", tnfe);
360
361 } catch (TableNotEnabledException tnee) {
362
363 successes.incrementAndGet();
364 LOG.debug("The targeted table was disabled. Assuming success.");
365 } catch (DoNotRetryIOException dnrioe) {
366 sink.publishReadFailure(tableName.getNameAsString(), serverName);
367 LOG.error(dnrioe);
368 } catch (IOException e) {
369 sink.publishReadFailure(tableName.getNameAsString(), serverName);
370 LOG.error(e);
371 } finally {
372 if (table != null) {
373 try {
374 table.close();
375 } catch (IOException e) {
376 LOG.error("Close table failed", e);
377 }
378 }
379 scan = null;
380 get = null;
381 startKey = null;
382 }
383 return null;
384 }
385 }
386
387 private static final int USAGE_EXIT_CODE = 1;
388 private static final int INIT_ERROR_EXIT_CODE = 2;
389 private static final int TIMEOUT_ERROR_EXIT_CODE = 3;
390 private static final int ERROR_EXIT_CODE = 4;
391
392 private static final long DEFAULT_INTERVAL = 6000;
393
394 private static final long DEFAULT_TIMEOUT = 600000;
395 private static final int MAX_THREADS_NUM = 16;
396
397 private static final Log LOG = LogFactory.getLog(Canary.class);
398
399 public static final TableName DEFAULT_WRITE_TABLE_NAME = TableName.valueOf(
400 NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR, "canary");
401
402 private static final String CANARY_TABLE_FAMILY_NAME = "Test";
403
404 private Configuration conf = null;
405 private long interval = 0;
406 private Sink sink = null;
407
408 private boolean useRegExp;
409 private long timeout = DEFAULT_TIMEOUT;
410 private boolean failOnError = true;
411 private boolean regionServerMode = false;
412 private boolean regionServerAllRegions = false;
413 private boolean writeSniffing = false;
414 private TableName writeTableName = DEFAULT_WRITE_TABLE_NAME;
415
416 private ExecutorService executor;
417
418 public Canary() {
419 this(new ScheduledThreadPoolExecutor(1), new RegionServerStdOutSink());
420 }
421
422 public Canary(ExecutorService executor, Sink sink) {
423 this.executor = executor;
424 this.sink = sink;
425 }
426
427 @Override
428 public Configuration getConf() {
429 return conf;
430 }
431
432 @Override
433 public void setConf(Configuration conf) {
434 this.conf = conf;
435 }
436
437 private int parseArgs(String[] args) {
438 int index = -1;
439
440 for (int i = 0; i < args.length; i++) {
441 String cmd = args[i];
442
443 if (cmd.startsWith("-")) {
444 if (index >= 0) {
445
446 System.err.println("Invalid command line options");
447 printUsageAndExit();
448 }
449
450 if (cmd.equals("-help")) {
451
452 printUsageAndExit();
453 } else if (cmd.equals("-daemon") && interval == 0) {
454
455 interval = DEFAULT_INTERVAL;
456 } else if (cmd.equals("-interval")) {
457
458 i++;
459
460 if (i == args.length) {
461 System.err.println("-interval needs a numeric value argument.");
462 printUsageAndExit();
463 }
464
465 try {
466 interval = Long.parseLong(args[i]) * 1000;
467 } catch (NumberFormatException e) {
468 System.err.println("-interval needs a numeric value argument.");
469 printUsageAndExit();
470 }
471 } else if(cmd.equals("-regionserver")) {
472 this.regionServerMode = true;
473 } else if(cmd.equals("-allRegions")) {
474 this.regionServerAllRegions = true;
475 } else if(cmd.equals("-writeSniffing")) {
476 this.writeSniffing = true;
477 } else if (cmd.equals("-e")) {
478 this.useRegExp = true;
479 } else if (cmd.equals("-t")) {
480 i++;
481
482 if (i == args.length) {
483 System.err.println("-t needs a numeric value argument.");
484 printUsageAndExit();
485 }
486
487 try {
488 this.timeout = Long.parseLong(args[i]);
489 } catch (NumberFormatException e) {
490 System.err.println("-t needs a numeric value argument.");
491 printUsageAndExit();
492 }
493 } else if (cmd.equals("-writeTable")) {
494 i++;
495
496 if (i == args.length) {
497 System.err.println("-writeTable needs a string value argument.");
498 printUsageAndExit();
499 }
500 this.writeTableName = TableName.valueOf(args[i]);
501 } else if (cmd.equals("-f")) {
502 i++;
503
504 if (i == args.length) {
505 System.err
506 .println("-f needs a boolean value argument (true|false).");
507 printUsageAndExit();
508 }
509
510 this.failOnError = Boolean.parseBoolean(args[i]);
511 } else {
512
513 System.err.println(cmd + " options is invalid.");
514 printUsageAndExit();
515 }
516 } else if (index < 0) {
517
518 index = i;
519 }
520 }
521 if (this.regionServerAllRegions && !this.regionServerMode) {
522 System.err.println("-allRegions can only be specified in regionserver mode.");
523 printUsageAndExit();
524 }
525 return index;
526 }
527
528 @Override
529 public int run(String[] args) throws Exception {
530 int index = parseArgs(args);
531 ChoreService choreService = null;
532
533
534
535
536 final ScheduledChore authChore = AuthUtil.getAuthChore(conf);
537 if (authChore != null) {
538 choreService = new ChoreService("CANARY_TOOL");
539 choreService.scheduleChore(authChore);
540 }
541
542
543 Monitor monitor = null;
544 Thread monitorThread = null;
545 long startTime = 0;
546 long currentTimeLength = 0;
547
548
549
550 try (Connection connection = ConnectionFactory.createConnection(this.conf)) {
551 do {
552
553 try {
554 monitor = this.newMonitor(connection, index, args);
555 monitorThread = new Thread(monitor);
556 startTime = System.currentTimeMillis();
557 monitorThread.start();
558 while (!monitor.isDone()) {
559
560 Thread.sleep(1000);
561
562 if (this.failOnError && monitor.hasError()) {
563 monitorThread.interrupt();
564 if (monitor.initialized) {
565 return monitor.errorCode;
566 } else {
567 return INIT_ERROR_EXIT_CODE;
568 }
569 }
570 currentTimeLength = System.currentTimeMillis() - startTime;
571 if (currentTimeLength > this.timeout) {
572 LOG.error("The monitor is running too long (" + currentTimeLength
573 + ") after timeout limit:" + this.timeout
574 + " will be killed itself !!");
575 if (monitor.initialized) {
576 return TIMEOUT_ERROR_EXIT_CODE;
577 } else {
578 return INIT_ERROR_EXIT_CODE;
579 }
580 }
581 }
582
583 if (this.failOnError && monitor.hasError()) {
584 monitorThread.interrupt();
585 return monitor.errorCode;
586 }
587 } finally {
588 if (monitor != null) monitor.close();
589 }
590
591 Thread.sleep(interval);
592 } while (interval > 0);
593 }
594
595 if (choreService != null) {
596 choreService.shutdown();
597 }
598 return monitor.errorCode;
599 }
600
601 private void printUsageAndExit() {
602 System.err.printf(
603 "Usage: bin/hbase %s [opts] [table1 [table2]...] | [regionserver1 [regionserver2]..]%n",
604 getClass().getName());
605 System.err.println(" where [opts] are:");
606 System.err.println(" -help Show this help and exit.");
607 System.err.println(" -regionserver replace the table argument to regionserver,");
608 System.err.println(" which means to enable regionserver mode");
609 System.err.println(" -allRegions Tries all regions on a regionserver,");
610 System.err.println(" only works in regionserver mode.");
611 System.err.println(" -daemon Continuous check at defined intervals.");
612 System.err.println(" -interval <N> Interval between checks (sec)");
613 System.err.println(" -e Use region/regionserver as regular expression");
614 System.err.println(" which means the region/regionserver is regular expression pattern");
615 System.err.println(" -f <B> stop whole program if first error occurs," +
616 " default is true");
617 System.err.println(" -t <N> timeout for a check, default is 600000 (milisecs)");
618 System.err.println(" -writeSniffing enable the write sniffing in canary");
619 System.err.println(" -writeTable The table used for write sniffing."
620 + " Default is hbase:canary");
621 System.err
622 .println(" -D<configProperty>=<value> assigning or override the configuration params");
623 System.exit(USAGE_EXIT_CODE);
624 }
625
626
627
628
629
630
631
632
633 public Monitor newMonitor(final Connection connection, int index, String[] args) {
634 Monitor monitor = null;
635 String[] monitorTargets = null;
636
637 if(index >= 0) {
638 int length = args.length - index;
639 monitorTargets = new String[length];
640 System.arraycopy(args, index, monitorTargets, 0, length);
641 }
642
643 if (this.regionServerMode) {
644 monitor =
645 new RegionServerMonitor(connection, monitorTargets, this.useRegExp,
646 (ExtendedSink) this.sink, this.executor, this.regionServerAllRegions);
647 } else {
648 monitor =
649 new RegionMonitor(connection, monitorTargets, this.useRegExp, this.sink, this.executor,
650 this.writeSniffing, this.writeTableName);
651 }
652 return monitor;
653 }
654
655
656 public static abstract class Monitor implements Runnable, Closeable {
657
658 protected Connection connection;
659 protected Admin admin;
660 protected String[] targets;
661 protected boolean useRegExp;
662 protected boolean initialized = false;
663
664 protected boolean done = false;
665 protected int errorCode = 0;
666 protected Sink sink;
667 protected ExecutorService executor;
668
669 public boolean isDone() {
670 return done;
671 }
672
673 public boolean hasError() {
674 return errorCode != 0;
675 }
676
677 @Override
678 public void close() throws IOException {
679 if (this.admin != null) this.admin.close();
680 }
681
682 protected Monitor(Connection connection, String[] monitorTargets, boolean useRegExp, Sink sink,
683 ExecutorService executor) {
684 if (null == connection) throw new IllegalArgumentException("connection shall not be null");
685
686 this.connection = connection;
687 this.targets = monitorTargets;
688 this.useRegExp = useRegExp;
689 this.sink = sink;
690 this.executor = executor;
691 }
692
693 public abstract void run();
694
695 protected boolean initAdmin() {
696 if (null == this.admin) {
697 try {
698 this.admin = this.connection.getAdmin();
699 } catch (Exception e) {
700 LOG.error("Initial HBaseAdmin failed...", e);
701 this.errorCode = INIT_ERROR_EXIT_CODE;
702 }
703 } else if (admin.isAborted()) {
704 LOG.error("HBaseAdmin aborted");
705 this.errorCode = INIT_ERROR_EXIT_CODE;
706 }
707 return !this.hasError();
708 }
709 }
710
711
712 private static class RegionMonitor extends Monitor {
713
714 private static final int DEFAULT_WRITE_TABLE_CHECK_PERIOD = 10 * 60 * 1000;
715
716 private static final int DEFAULT_WRITE_DATA_TTL = 24 * 60 * 60;
717
718 private long lastCheckTime = -1;
719 private boolean writeSniffing;
720 private TableName writeTableName;
721 private int writeDataTTL;
722 private float regionsLowerLimit;
723 private float regionsUpperLimit;
724 private int checkPeriod;
725
726 public RegionMonitor(Connection connection, String[] monitorTargets, boolean useRegExp,
727 Sink sink, ExecutorService executor, boolean writeSniffing, TableName writeTableName) {
728 super(connection, monitorTargets, useRegExp, sink, executor);
729 Configuration conf = connection.getConfiguration();
730 this.writeSniffing = writeSniffing;
731 this.writeTableName = writeTableName;
732 this.writeDataTTL =
733 conf.getInt(HConstants.HBASE_CANARY_WRITE_DATA_TTL_KEY, DEFAULT_WRITE_DATA_TTL);
734 this.regionsLowerLimit =
735 conf.getFloat(HConstants.HBASE_CANARY_WRITE_PERSERVER_REGIONS_LOWERLIMIT_KEY, 1.0f);
736 this.regionsUpperLimit =
737 conf.getFloat(HConstants.HBASE_CANARY_WRITE_PERSERVER_REGIONS_UPPERLIMIT_KEY, 1.5f);
738 this.checkPeriod =
739 conf.getInt(HConstants.HBASE_CANARY_WRITE_TABLE_CHECK_PERIOD_KEY,
740 DEFAULT_WRITE_TABLE_CHECK_PERIOD);
741 }
742
743 @Override
744 public void run() {
745 if (this.initAdmin()) {
746 try {
747 List<Future<Void>> taskFutures = new LinkedList<Future<Void>>();
748 if (this.targets != null && this.targets.length > 0) {
749 String[] tables = generateMonitorTables(this.targets);
750 this.initialized = true;
751 for (String table : tables) {
752 taskFutures.addAll(Canary.sniff(admin, sink, table, executor, TaskType.READ));
753 }
754 } else {
755 taskFutures.addAll(sniff(TaskType.READ));
756 }
757
758 if (writeSniffing) {
759 if (EnvironmentEdgeManager.currentTime() - lastCheckTime > checkPeriod) {
760 try {
761 checkWriteTableDistribution();
762 } catch (IOException e) {
763 LOG.error("Check canary table distribution failed!", e);
764 }
765 lastCheckTime = EnvironmentEdgeManager.currentTime();
766 }
767
768 taskFutures.addAll(Canary.sniff(admin, sink,
769 admin.getTableDescriptor(writeTableName), executor, TaskType.WRITE));
770 }
771
772 for (Future<Void> future : taskFutures) {
773 try {
774 future.get();
775 } catch (ExecutionException e) {
776 LOG.error("Sniff region failed!", e);
777 }
778 }
779 } catch (Exception e) {
780 LOG.error("Run regionMonitor failed", e);
781 this.errorCode = ERROR_EXIT_CODE;
782 }
783 }
784 this.done = true;
785 }
786
787 private String[] generateMonitorTables(String[] monitorTargets) throws IOException {
788 String[] returnTables = null;
789
790 if (this.useRegExp) {
791 Pattern pattern = null;
792 HTableDescriptor[] tds = null;
793 Set<String> tmpTables = new TreeSet<String>();
794 try {
795 for (String monitorTarget : monitorTargets) {
796 pattern = Pattern.compile(monitorTarget);
797 tds = this.admin.listTables(pattern);
798 if (tds != null) {
799 for (HTableDescriptor td : tds) {
800 tmpTables.add(td.getNameAsString());
801 }
802 }
803 }
804 } catch (IOException e) {
805 LOG.error("Communicate with admin failed", e);
806 throw e;
807 }
808
809 if (tmpTables.size() > 0) {
810 returnTables = tmpTables.toArray(new String[tmpTables.size()]);
811 } else {
812 String msg = "No HTable found, tablePattern:" + Arrays.toString(monitorTargets);
813 LOG.error(msg);
814 this.errorCode = INIT_ERROR_EXIT_CODE;
815 throw new TableNotFoundException(msg);
816 }
817 } else {
818 returnTables = monitorTargets;
819 }
820
821 return returnTables;
822 }
823
824
825
826
827 private List<Future<Void>> sniff(TaskType taskType) throws Exception {
828 List<Future<Void>> taskFutures = new LinkedList<Future<Void>>();
829 for (HTableDescriptor table : admin.listTables()) {
830 if (admin.isTableEnabled(table.getTableName())
831 && (!table.getTableName().equals(writeTableName))) {
832 taskFutures.addAll(Canary.sniff(admin, sink, table, executor, taskType));
833 }
834 }
835 return taskFutures;
836 }
837
838 private void checkWriteTableDistribution() throws IOException {
839 if (!admin.tableExists(writeTableName)) {
840 int numberOfServers = admin.getClusterStatus().getServers().size();
841 if (numberOfServers == 0) {
842 throw new IllegalStateException("No live regionservers");
843 }
844 createWriteTable(numberOfServers);
845 }
846
847 if (!admin.isTableEnabled(writeTableName)) {
848 admin.enableTable(writeTableName);
849 }
850
851 int numberOfServers = admin.getClusterStatus().getServers().size();
852 List<HRegionLocation> locations;
853 RegionLocator locator = connection.getRegionLocator(writeTableName);
854 try {
855 locations = locator.getAllRegionLocations();
856 } finally {
857 locator.close();
858 }
859 int numberOfRegions = locations.size();
860 if (numberOfRegions < numberOfServers * regionsLowerLimit
861 || numberOfRegions > numberOfServers * regionsUpperLimit) {
862 admin.disableTable(writeTableName);
863 admin.deleteTable(writeTableName);
864 createWriteTable(numberOfServers);
865 }
866 HashSet<ServerName> serverSet = new HashSet<ServerName>();
867 for (HRegionLocation location: locations) {
868 serverSet.add(location.getServerName());
869 }
870 int numberOfCoveredServers = serverSet.size();
871 if (numberOfCoveredServers < numberOfServers) {
872 admin.balancer();
873 }
874 }
875
876 private void createWriteTable(int numberOfServers) throws IOException {
877 int numberOfRegions = (int)(numberOfServers * regionsLowerLimit);
878 LOG.info("Number of live regionservers: " + numberOfServers + ", "
879 + "pre-splitting the canary table into " + numberOfRegions + " regions "
880 + "(current lower limi of regions per server is " + regionsLowerLimit
881 + " and you can change it by config: "
882 + HConstants.HBASE_CANARY_WRITE_PERSERVER_REGIONS_LOWERLIMIT_KEY + " )");
883 HTableDescriptor desc = new HTableDescriptor(writeTableName);
884 HColumnDescriptor family = new HColumnDescriptor(CANARY_TABLE_FAMILY_NAME);
885 family.setMaxVersions(1);
886 family.setTimeToLive(writeDataTTL);
887
888 desc.addFamily(family);
889 byte[][] splits = new RegionSplitter.HexStringSplit().split(numberOfRegions);
890 admin.createTable(desc, splits);
891 }
892 }
893
894
895
896
897
898 public static void sniff(final Admin admin, TableName tableName)
899 throws Exception {
900 sniff(admin, tableName, TaskType.READ);
901 }
902
903
904
905
906
907 public static void sniff(final Admin admin, TableName tableName, TaskType taskType)
908 throws Exception {
909 List<Future<Void>> taskFutures =
910 Canary.sniff(admin, new StdOutSink(), tableName.getNameAsString(),
911 new ScheduledThreadPoolExecutor(1), taskType);
912 for (Future<Void> future : taskFutures) {
913 future.get();
914 }
915 }
916
917
918
919
920
921 private static List<Future<Void>> sniff(final Admin admin, final Sink sink, String tableName,
922 ExecutorService executor, TaskType taskType) throws Exception {
923 if (admin.isTableEnabled(TableName.valueOf(tableName))) {
924 return Canary.sniff(admin, sink, admin.getTableDescriptor(TableName.valueOf(tableName)),
925 executor, taskType);
926 } else {
927 LOG.warn(String.format("Table %s is not enabled", tableName));
928 }
929 return new LinkedList<Future<Void>>();
930 }
931
932
933
934
935 private static List<Future<Void>> sniff(final Admin admin, final Sink sink,
936 HTableDescriptor tableDesc, ExecutorService executor, TaskType taskType) throws Exception {
937 Table table = null;
938 try {
939 table = admin.getConnection().getTable(tableDesc.getTableName());
940 } catch (TableNotFoundException e) {
941 return new ArrayList<Future<Void>>();
942 }
943 List<RegionTask> tasks = new ArrayList<RegionTask>();
944 try {
945 for (HRegionInfo region : admin.getTableRegions(tableDesc.getTableName())) {
946 tasks.add(new RegionTask(admin.getConnection(), region, sink, taskType));
947 }
948 } finally {
949 table.close();
950 }
951 return executor.invokeAll(tasks);
952 }
953
954
955
956
957
958 private static void sniffRegion(
959 final Admin admin,
960 final Sink sink,
961 HRegionInfo region,
962 Table table) throws Exception {
963 HTableDescriptor tableDesc = table.getTableDescriptor();
964 byte[] startKey = null;
965 Get get = null;
966 Scan scan = null;
967 ResultScanner rs = null;
968 StopWatch stopWatch = new StopWatch();
969 for (HColumnDescriptor column : tableDesc.getColumnFamilies()) {
970 stopWatch.reset();
971 startKey = region.getStartKey();
972
973 if (startKey.length > 0) {
974 get = new Get(startKey);
975 get.setCacheBlocks(false);
976 get.setFilter(new FirstKeyOnlyFilter());
977 get.addFamily(column.getName());
978 } else {
979 scan = new Scan();
980 scan.setRaw(true);
981 scan.setCaching(1);
982 scan.setCacheBlocks(false);
983 scan.setFilter(new FirstKeyOnlyFilter());
984 scan.addFamily(column.getName());
985 scan.setMaxResultSize(1L);
986 }
987
988 try {
989 if (startKey.length > 0) {
990 stopWatch.start();
991 table.get(get);
992 stopWatch.stop();
993 sink.publishReadTiming(region, column, stopWatch.getTime());
994 } else {
995 stopWatch.start();
996 rs = table.getScanner(scan);
997 stopWatch.stop();
998 sink.publishReadTiming(region, column, stopWatch.getTime());
999 }
1000 } catch (Exception e) {
1001 sink.publishReadFailure(region, column, e);
1002 } finally {
1003 if (rs != null) {
1004 rs.close();
1005 }
1006 scan = null;
1007 get = null;
1008 startKey = null;
1009 }
1010 }
1011 }
1012
1013 private static class RegionServerMonitor extends Monitor {
1014
1015 private boolean allRegions;
1016
1017 public RegionServerMonitor(Connection connection, String[] monitorTargets, boolean useRegExp,
1018 ExtendedSink sink, ExecutorService executor, boolean allRegions) {
1019 super(connection, monitorTargets, useRegExp, sink, executor);
1020 this.allRegions = allRegions;
1021 }
1022
1023 private ExtendedSink getSink() {
1024 return (ExtendedSink) this.sink;
1025 }
1026
1027 @Override
1028 public void run() {
1029 if (this.initAdmin() && this.checkNoTableNames()) {
1030 Map<String, List<HRegionInfo>> rsAndRMap = this.filterRegionServerByName();
1031 this.initialized = true;
1032 this.monitorRegionServers(rsAndRMap);
1033 }
1034 this.done = true;
1035 }
1036
1037 private boolean checkNoTableNames() {
1038 List<String> foundTableNames = new ArrayList<String>();
1039 TableName[] tableNames = null;
1040
1041 try {
1042 tableNames = this.admin.listTableNames();
1043 } catch (IOException e) {
1044 LOG.error("Get listTableNames failed", e);
1045 this.errorCode = INIT_ERROR_EXIT_CODE;
1046 return false;
1047 }
1048
1049 if (this.targets == null || this.targets.length == 0) return true;
1050
1051 for (String target : this.targets) {
1052 for (TableName tableName : tableNames) {
1053 if (target.equals(tableName.getNameAsString())) {
1054 foundTableNames.add(target);
1055 }
1056 }
1057 }
1058
1059 if (foundTableNames.size() > 0) {
1060 System.err.println("Cannot pass a tablename when using the -regionserver " +
1061 "option, tablenames:" + foundTableNames.toString());
1062 this.errorCode = USAGE_EXIT_CODE;
1063 }
1064 return foundTableNames.size() == 0;
1065 }
1066
1067 private void monitorRegionServers(Map<String, List<HRegionInfo>> rsAndRMap) {
1068 List<RegionServerTask> tasks = new ArrayList<RegionServerTask>();
1069 Map<String, AtomicLong> successMap = new HashMap<String, AtomicLong>();
1070 Random rand = new Random();
1071 for (Map.Entry<String, List<HRegionInfo>> entry : rsAndRMap.entrySet()) {
1072 String serverName = entry.getKey();
1073 AtomicLong successes = new AtomicLong(0);
1074 successMap.put(serverName, successes);
1075 if (this.allRegions) {
1076 for (HRegionInfo region : entry.getValue()) {
1077 tasks.add(new RegionServerTask(this.connection,
1078 serverName,
1079 region,
1080 getSink(),
1081 successes));
1082 }
1083 } else {
1084
1085 HRegionInfo region = entry.getValue().get(rand.nextInt(entry.getValue().size()));
1086 tasks.add(new RegionServerTask(this.connection,
1087 serverName,
1088 region,
1089 getSink(),
1090 successes));
1091 }
1092 }
1093 try {
1094 for (Future<Void> future : this.executor.invokeAll(tasks)) {
1095 try {
1096 future.get();
1097 } catch (ExecutionException e) {
1098 LOG.error("Sniff regionserver failed!", e);
1099 this.errorCode = ERROR_EXIT_CODE;
1100 }
1101 }
1102 if (this.allRegions) {
1103 for (Map.Entry<String, List<HRegionInfo>> entry : rsAndRMap.entrySet()) {
1104 String serverName = entry.getKey();
1105 LOG.info("Successfully read " + successMap.get(serverName) + " regions out of "
1106 + entry.getValue().size() + " on regionserver:" + serverName);
1107 }
1108 }
1109 } catch (InterruptedException e) {
1110 this.errorCode = ERROR_EXIT_CODE;
1111 LOG.error("Sniff regionserver failed!", e);
1112 }
1113 }
1114
1115 private Map<String, List<HRegionInfo>> filterRegionServerByName() {
1116 Map<String, List<HRegionInfo>> regionServerAndRegionsMap = this.getAllRegionServerByName();
1117 regionServerAndRegionsMap = this.doFilterRegionServerByName(regionServerAndRegionsMap);
1118 return regionServerAndRegionsMap;
1119 }
1120
1121 private Map<String, List<HRegionInfo>> getAllRegionServerByName() {
1122 Map<String, List<HRegionInfo>> rsAndRMap = new HashMap<String, List<HRegionInfo>>();
1123 Table table = null;
1124 RegionLocator regionLocator = null;
1125 try {
1126 HTableDescriptor[] tableDescs = this.admin.listTables();
1127 List<HRegionInfo> regions = null;
1128 for (HTableDescriptor tableDesc : tableDescs) {
1129 table = this.admin.getConnection().getTable(tableDesc.getTableName());
1130 regionLocator = this.admin.getConnection().getRegionLocator(tableDesc.getTableName());
1131
1132 for (HRegionLocation location : regionLocator.getAllRegionLocations()) {
1133 ServerName rs = location.getServerName();
1134 String rsName = rs.getHostname();
1135 HRegionInfo r = location.getRegionInfo();
1136
1137 if (rsAndRMap.containsKey(rsName)) {
1138 regions = rsAndRMap.get(rsName);
1139 } else {
1140 regions = new ArrayList<HRegionInfo>();
1141 rsAndRMap.put(rsName, regions);
1142 }
1143 regions.add(r);
1144 }
1145 table.close();
1146 }
1147
1148 } catch (IOException e) {
1149 String msg = "Get HTables info failed";
1150 LOG.error(msg, e);
1151 this.errorCode = INIT_ERROR_EXIT_CODE;
1152 } finally {
1153 if (table != null) {
1154 try {
1155 table.close();
1156 } catch (IOException e) {
1157 LOG.warn("Close table failed", e);
1158 }
1159 }
1160 }
1161
1162 return rsAndRMap;
1163 }
1164
1165 private Map<String, List<HRegionInfo>> doFilterRegionServerByName(
1166 Map<String, List<HRegionInfo>> fullRsAndRMap) {
1167
1168 Map<String, List<HRegionInfo>> filteredRsAndRMap = null;
1169
1170 if (this.targets != null && this.targets.length > 0) {
1171 filteredRsAndRMap = new HashMap<String, List<HRegionInfo>>();
1172 Pattern pattern = null;
1173 Matcher matcher = null;
1174 boolean regExpFound = false;
1175 for (String rsName : this.targets) {
1176 if (this.useRegExp) {
1177 regExpFound = false;
1178 pattern = Pattern.compile(rsName);
1179 for (Map.Entry<String, List<HRegionInfo>> entry : fullRsAndRMap.entrySet()) {
1180 matcher = pattern.matcher(entry.getKey());
1181 if (matcher.matches()) {
1182 filteredRsAndRMap.put(entry.getKey(), entry.getValue());
1183 regExpFound = true;
1184 }
1185 }
1186 if (!regExpFound) {
1187 LOG.info("No RegionServerInfo found, regionServerPattern:" + rsName);
1188 }
1189 } else {
1190 if (fullRsAndRMap.containsKey(rsName)) {
1191 filteredRsAndRMap.put(rsName, fullRsAndRMap.get(rsName));
1192 } else {
1193 LOG.info("No RegionServerInfo found, regionServerName:" + rsName);
1194 }
1195 }
1196 }
1197 } else {
1198 filteredRsAndRMap = fullRsAndRMap;
1199 }
1200 return filteredRsAndRMap;
1201 }
1202 }
1203
1204 public static void main(String[] args) throws Exception {
1205 final Configuration conf = HBaseConfiguration.create();
1206 final ChoreService choreService = new ChoreService("CANARY_TOOL");
1207 final ScheduledChore authChore = AuthUtil.getAuthChore(conf);
1208 if (authChore != null) {
1209 choreService.scheduleChore(authChore);
1210 }
1211
1212
1213 new GenericOptionsParser(conf, args);
1214
1215 int numThreads = conf.getInt("hbase.canary.threads.num", MAX_THREADS_NUM);
1216 LOG.info("Number of exection threads " + numThreads);
1217
1218 ExecutorService executor = new ScheduledThreadPoolExecutor(numThreads);
1219
1220 Class<? extends Sink> sinkClass =
1221 conf.getClass("hbase.canary.sink.class", RegionServerStdOutSink.class, Sink.class);
1222 Sink sink = ReflectionUtils.newInstance(sinkClass);
1223
1224 int exitCode = ToolRunner.run(conf, new Canary(executor, sink), args);
1225 choreService.shutdown();
1226 executor.shutdown();
1227 System.exit(exitCode);
1228 }
1229 }