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