1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.master.cleaner;
19
20 import static org.apache.hadoop.hbase.HConstants.HBASE_MASTER_LOGCLEANER_PLUGINS;
21
22 import com.google.common.annotations.VisibleForTesting;
23
24 import java.io.IOException;
25 import java.util.ArrayList;
26 import java.util.LinkedList;
27 import java.util.List;
28 import java.util.concurrent.LinkedBlockingQueue;
29 import java.util.concurrent.TimeUnit;
30
31 import org.apache.commons.logging.Log;
32 import org.apache.commons.logging.LogFactory;
33 import org.apache.hadoop.fs.FileStatus;
34 import org.apache.hadoop.hbase.classification.InterfaceAudience;
35 import org.apache.hadoop.conf.Configuration;
36 import org.apache.hadoop.fs.FileSystem;
37 import org.apache.hadoop.fs.Path;
38 import org.apache.hadoop.hbase.Stoppable;
39 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
40
41
42
43
44
45
46 @InterfaceAudience.Private
47 public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
48 private static final Log LOG = LogFactory.getLog(LogCleaner.class.getName());
49
50 public static final String OLD_WALS_CLEANER_THREAD_SIZE = "hbase.oldwals.cleaner.thread.size";
51 public static final int DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE = 2;
52
53 public static final String OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC =
54 "hbase.oldwals.cleaner.thread.timeout.msec";
55 @VisibleForTesting
56 static final long DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC = 60 * 1000L;
57
58 public static final String OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC =
59 "hbase.oldwals.cleaner.thread.check.interval.msec";
60 @VisibleForTesting
61 static final long DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC = 500L;
62
63
64 private final LinkedBlockingQueue<CleanerContext> pendingDelete;
65 private List<Thread> oldWALsCleaner;
66 private long cleanerThreadTimeoutMsec;
67 private long cleanerThreadCheckIntervalMsec;
68
69
70
71
72
73
74
75
76 public LogCleaner(final int p, final Stoppable s, Configuration conf, FileSystem fs,
77 Path oldLogDir) {
78 super("LogsCleaner", p, s, conf, fs, oldLogDir, HBASE_MASTER_LOGCLEANER_PLUGINS);
79 this.pendingDelete = new LinkedBlockingQueue<>();
80 int size = conf.getInt(OLD_WALS_CLEANER_THREAD_SIZE, DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE);
81 this.oldWALsCleaner = createOldWalsCleaner(size);
82 this.cleanerThreadTimeoutMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
83 DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC);
84 this.cleanerThreadCheckIntervalMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
85 DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC);
86 }
87
88 @Override
89 protected boolean validate(Path file) {
90 return DefaultWALProvider.validateWALFilename(file.getName());
91 }
92
93 @Override
94 public void onConfigurationChange(Configuration conf) {
95 super.onConfigurationChange(conf);
96
97 int newSize = conf.getInt(OLD_WALS_CLEANER_THREAD_SIZE, DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE);
98 if (newSize == oldWALsCleaner.size()) {
99 if (LOG.isDebugEnabled()) {
100 LOG.debug("Size from configuration is the same as previous which is " +
101 newSize + ", no need to update.");
102 }
103 return;
104 }
105 interruptOldWALsCleaner();
106 oldWALsCleaner = createOldWalsCleaner(newSize);
107 cleanerThreadTimeoutMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
108 DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC);
109 cleanerThreadCheckIntervalMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
110 DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC);
111 }
112
113 @Override
114 protected int deleteFiles(Iterable<FileStatus> filesToDelete) {
115 List<CleanerContext> results = new LinkedList<>();
116 for (FileStatus toDelete : filesToDelete) {
117 CleanerContext context = CleanerContext.createCleanerContext(toDelete,
118 cleanerThreadTimeoutMsec);
119 if (context != null) {
120 pendingDelete.add(context);
121 results.add(context);
122 }
123 }
124
125 int deletedFiles = 0;
126 for (CleanerContext res : results) {
127 deletedFiles += res.getResult(cleanerThreadCheckIntervalMsec) ? 1 : 0;
128 }
129 return deletedFiles;
130 }
131
132 @Override
133 public synchronized void cleanup() {
134 super.cleanup();
135 interruptOldWALsCleaner();
136 }
137
138 @VisibleForTesting
139 int getSizeOfCleaners() {
140 return oldWALsCleaner.size();
141 }
142
143 @VisibleForTesting
144 long getCleanerThreadTimeoutMsec() {
145 return cleanerThreadTimeoutMsec;
146 }
147
148 @VisibleForTesting
149 long getCleanerThreadCheckIntervalMsec() {
150 return cleanerThreadCheckIntervalMsec;
151 }
152
153 private List<Thread> createOldWalsCleaner(int size) {
154 LOG.info("Creating OldWALs cleaners with size=" + size);
155
156 List<Thread> oldWALsCleaner = new ArrayList<>(size);
157 for (int i = 0; i < size; i++) {
158 Thread cleaner = new Thread(new Runnable() {
159 @Override
160 public void run() {
161 deleteFile();
162 }
163 });
164 cleaner.setName("OldWALsCleaner-" + i);
165 cleaner.setDaemon(true);
166 cleaner.start();
167 oldWALsCleaner.add(cleaner);
168 }
169 return oldWALsCleaner;
170 }
171
172 private void interruptOldWALsCleaner() {
173 for (Thread cleaner : oldWALsCleaner) {
174 cleaner.interrupt();
175 }
176 oldWALsCleaner.clear();
177 }
178
179 private void deleteFile() {
180 while (true) {
181 CleanerContext context = null;
182 boolean succeed = false;
183 boolean interrupted = false;
184 try {
185 context = pendingDelete.take();
186 if (context != null) {
187 FileStatus toClean = context.getTargetToClean();
188 succeed = this.fs.delete(toClean.getPath(), false);
189 }
190 } catch (InterruptedException ite) {
191
192 if (context != null) {
193 LOG.warn("Interrupted while cleaning oldWALs " +
194 context.getTargetToClean() + ", try to clean it next round.");
195 }
196 interrupted = true;
197 } catch (IOException e) {
198
199 LOG.warn("Failed to clean oldwals with exception: " + e);
200 succeed = false;
201 } finally {
202 if (context != null) {
203 context.setResult(succeed);
204 }
205 if (interrupted) {
206
207 Thread.currentThread().interrupt();
208 break;
209 }
210 }
211 }
212 if (LOG.isDebugEnabled()) {
213 LOG.debug("Exiting cleaner.");
214 }
215 }
216
217 @Override
218 public synchronized void cancel(boolean mayInterruptIfRunning) {
219 super.cancel(mayInterruptIfRunning);
220 for (Thread t : oldWALsCleaner) {
221 t.interrupt();
222 }
223 }
224
225 private static final class CleanerContext {
226
227 final FileStatus target;
228 volatile boolean result;
229 volatile boolean setFromCleaner = false;
230 long timeoutMsec;
231
232 static CleanerContext createCleanerContext(FileStatus status, long timeoutMsec) {
233 return status != null ? new CleanerContext(status, timeoutMsec) : null;
234 }
235
236 private CleanerContext(FileStatus status, long timeoutMsec) {
237 this.target = status;
238 this.result = false;
239 this.timeoutMsec = timeoutMsec;
240 }
241
242 synchronized void setResult(boolean res) {
243 this.result = res;
244 this.setFromCleaner = true;
245 notify();
246 }
247
248 synchronized boolean getResult(long waitIfNotFinished) {
249 long totalTimeMsec = 0;
250 try {
251 while (!setFromCleaner) {
252 long startTimeNanos = System.nanoTime();
253 wait(waitIfNotFinished);
254 totalTimeMsec += TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTimeNanos,
255 TimeUnit.NANOSECONDS);
256 if (totalTimeMsec >= timeoutMsec) {
257 LOG.warn("Spend too much time " + totalTimeMsec + " ms to delete oldwals " + target);
258 return result;
259 }
260 }
261 } catch (InterruptedException e) {
262 LOG.warn("Interrupted while waiting deletion of " + target);
263 return result;
264 }
265 return result;
266 }
267
268 FileStatus getTargetToClean() {
269 return target;
270 }
271 }
272 }