View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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   * This Chore, every time it runs, will attempt to delete the WALs in the old logs folder. The WAL
43   * is only deleted if none of the cleaner delegates says otherwise.
44   * @see BaseLogCleanerDelegate
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     * @param p the period of time to sleep between each run
71     * @param s the stopper
72     * @param conf configuration to use
73     * @param fs handle to the FS
74     * @param oldLogDir the path to the archived logs
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         // It's most likely from configuration changing request
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         // fs.delete() fails.
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           // Restore interrupt status
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 }