1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.regionserver.throttle;
19
20 import org.apache.commons.logging.Log;
21 import org.apache.commons.logging.LogFactory;
22 import org.apache.hadoop.conf.Configuration;
23 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
24 import org.apache.hadoop.hbase.ScheduledChore;
25 import org.apache.hadoop.hbase.classification.InterfaceAudience;
26 import org.apache.hadoop.hbase.regionserver.RegionServerServices;
27 import org.apache.hadoop.hbase.regionserver.compactions.OffPeakHours;
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
43 public class PressureAwareCompactionThroughputController extends PressureAwareThroughputController {
44
45 private final static Log LOG = LogFactory
46 .getLog(PressureAwareCompactionThroughputController.class);
47
48 public static final String HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_HIGHER_BOUND =
49 "hbase.hstore.compaction.throughput.higher.bound";
50
51 private static final long DEFAULT_HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_HIGHER_BOUND =
52 20L * 1024 * 1024;
53
54 public static final String HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_LOWER_BOUND =
55 "hbase.hstore.compaction.throughput.lower.bound";
56
57 private static final long DEFAULT_HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_LOWER_BOUND =
58 10L * 1024 * 1024;
59
60 public static final String HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_OFFPEAK =
61 "hbase.hstore.compaction.throughput.offpeak";
62
63 private static final long DEFAULT_HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_OFFPEAK = Long.MAX_VALUE;
64
65 public static final String HBASE_HSTORE_COMPACTION_THROUGHPUT_TUNE_PERIOD =
66 "hbase.hstore.compaction.throughput.tune.period";
67
68 private static final int DEFAULT_HSTORE_COMPACTION_THROUGHPUT_TUNE_PERIOD = 60 * 1000;
69
70
71 private static final String HBASE_HSTORE_COMPACTION_THROUGHPUT_CONTROL_CHECK_INTERVAL =
72 "hbase.hstore.compaction.throughput.control.check.interval";
73
74 private long maxThroughputOffpeak;
75
76 @Override
77 public void setup(final RegionServerServices server) {
78 server.getChoreService().scheduleChore(
79 new ScheduledChore("CompactionThroughputTuner", this, tuningPeriod) {
80
81 @Override
82 protected void chore() {
83 tune(server.getCompactionPressure());
84 }
85 });
86 }
87
88 private void tune(double compactionPressure) {
89 double maxThroughputToSet;
90 if (compactionPressure > 1.0) {
91
92 maxThroughputToSet = Double.MAX_VALUE;
93 } else if (offPeakHours.isOffPeakHour()) {
94 maxThroughputToSet = maxThroughputOffpeak;
95 } else {
96
97
98 maxThroughputToSet =
99 maxThroughputLowerBound + (maxThroughputUpperBound - maxThroughputLowerBound)
100 * compactionPressure;
101 }
102 if (LOG.isDebugEnabled()) {
103 LOG.debug("compactionPressure is " + compactionPressure + ", tune compaction throughput to "
104 + throughputDesc(maxThroughputToSet));
105 }
106 this.setMaxThroughput(maxThroughputToSet);
107 }
108
109 @Override
110 public void setConf(Configuration conf) {
111 super.setConf(conf);
112 if (conf == null) {
113 return;
114 }
115 this.maxThroughputUpperBound =
116 conf.getLong(HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_HIGHER_BOUND,
117 DEFAULT_HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_HIGHER_BOUND);
118 this.maxThroughputLowerBound =
119 conf.getLong(HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_LOWER_BOUND,
120 DEFAULT_HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_LOWER_BOUND);
121 this.maxThroughputOffpeak =
122 conf.getLong(HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_OFFPEAK,
123 DEFAULT_HBASE_HSTORE_COMPACTION_MAX_THROUGHPUT_OFFPEAK);
124 this.offPeakHours = OffPeakHours.getInstance(conf);
125 this.controlPerSize =
126 conf.getLong(HBASE_HSTORE_COMPACTION_THROUGHPUT_CONTROL_CHECK_INTERVAL,
127 this.maxThroughputLowerBound);
128 this.setMaxThroughput(this.maxThroughputLowerBound);
129 this.tuningPeriod =
130 getConf().getInt(HBASE_HSTORE_COMPACTION_THROUGHPUT_TUNE_PERIOD,
131 DEFAULT_HSTORE_COMPACTION_THROUGHPUT_TUNE_PERIOD);
132 LOG.info("Compaction throughput configurations, higher bound: "
133 + throughputDesc(maxThroughputUpperBound) + ", lower bound "
134 + throughputDesc(maxThroughputLowerBound) + ", off peak: "
135 + throughputDesc(maxThroughputOffpeak) + ", tuning period: " + tuningPeriod + " ms");
136 }
137
138 @Override
139 public String toString() {
140 return "DefaultCompactionThroughputController [maxThroughput="
141 + throughputDesc(getMaxThroughput()) + ", activeCompactions=" + activeOperations.size()
142 + "]";
143 }
144
145 @Override
146 protected boolean skipControl(long deltaSize, long controlSize) {
147 if (deltaSize < controlSize) {
148 return true;
149 } else {
150 return false;
151 }
152 }
153 }