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  
19  package org.apache.hadoop.hbase.snapshot;
20  
21  import com.google.common.annotations.VisibleForTesting;
22  import com.google.protobuf.CodedInputStream;
23  import com.google.protobuf.InvalidProtocolBufferException;
24  
25  import java.io.FileNotFoundException;
26  import java.io.IOException;
27  import java.util.ArrayList;
28  import java.util.Collection;
29  import java.util.HashMap;
30  import java.util.List;
31  import java.util.Map;
32  import java.util.concurrent.ThreadPoolExecutor;
33  import java.util.concurrent.TimeUnit;
34  
35  import org.apache.commons.logging.Log;
36  import org.apache.commons.logging.LogFactory;
37  import org.apache.hadoop.conf.Configuration;
38  import org.apache.hadoop.fs.FSDataInputStream;
39  import org.apache.hadoop.fs.FSDataOutputStream;
40  import org.apache.hadoop.fs.FileSystem;
41  import org.apache.hadoop.fs.Path;
42  import org.apache.hadoop.hbase.HRegionInfo;
43  import org.apache.hadoop.hbase.HTableDescriptor;
44  import org.apache.hadoop.hbase.classification.InterfaceAudience;
45  import org.apache.hadoop.hbase.errorhandling.ForeignExceptionSnare;
46  import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
47  import org.apache.hadoop.hbase.protobuf.generated.SnapshotProtos.SnapshotDataManifest;
48  import org.apache.hadoop.hbase.protobuf.generated.SnapshotProtos.SnapshotRegionManifest;
49  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
50  import org.apache.hadoop.hbase.regionserver.HRegion;
51  import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
52  import org.apache.hadoop.hbase.regionserver.Store;
53  import org.apache.hadoop.hbase.regionserver.StoreFile;
54  import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
55  import org.apache.hadoop.hbase.util.Bytes;
56  import org.apache.hadoop.hbase.util.FSUtils;
57  import org.apache.hadoop.hbase.util.FSTableDescriptors;
58  import org.apache.hadoop.hbase.util.Threads;
59  
60  /**
61   * Utility class to help read/write the Snapshot Manifest.
62   *
63   * The snapshot format is transparent for the users of this class,
64   * once the snapshot is written, it will never be modified.
65   * On open() the snapshot will be loaded to the current in-memory format.
66   */
67  @InterfaceAudience.Private
68  public final class SnapshotManifest {
69    private static final Log LOG = LogFactory.getLog(SnapshotManifest.class);
70  
71    public static final String SNAPSHOT_MANIFEST_SIZE_LIMIT_CONF_KEY = "snapshot.manifest.size.limit";
72  
73    public static final String DATA_MANIFEST_NAME = "data.manifest";
74  
75    private List<SnapshotRegionManifest> regionManifests;
76    private SnapshotDescription desc;
77    private HTableDescriptor htd;
78  
79    private final ForeignExceptionSnare monitor;
80    private final Configuration conf;
81    private final Path workingDir;
82    private final FileSystem fs;
83    private int manifestSizeLimit;
84  
85    private SnapshotManifest(final Configuration conf, final FileSystem fs,
86        final Path workingDir, final SnapshotDescription desc,
87        final ForeignExceptionSnare monitor) {
88      this.monitor = monitor;
89      this.desc = desc;
90      this.workingDir = workingDir;
91      this.conf = conf;
92      this.fs = fs;
93  
94      this.manifestSizeLimit = conf.getInt(SNAPSHOT_MANIFEST_SIZE_LIMIT_CONF_KEY, 64 * 1024 * 1024);
95    }
96  
97    /**
98     * Return a SnapshotManifest instance, used for writing a snapshot.
99     *
100    * There are two usage pattern:
101    *  - The Master will create a manifest, add the descriptor, offline regions
102    *    and consolidate the snapshot by writing all the pending stuff on-disk.
103    *      manifest = SnapshotManifest.create(...)
104    *      manifest.addRegion(tableDir, hri)
105    *      manifest.consolidate()
106    *  - The RegionServer will create a single region manifest
107    *      manifest = SnapshotManifest.create(...)
108    *      manifest.addRegion(region)
109    */
110   public static SnapshotManifest create(final Configuration conf, final FileSystem fs,
111       final Path workingDir, final SnapshotDescription desc,
112       final ForeignExceptionSnare monitor) {
113     return new SnapshotManifest(conf, fs, workingDir, desc, monitor);
114 
115   }
116 
117   /**
118    * Return a SnapshotManifest instance with the information already loaded in-memory.
119    *    SnapshotManifest manifest = SnapshotManifest.open(...)
120    *    HTableDescriptor htd = manifest.getTableDescriptor()
121    *    for (SnapshotRegionManifest regionManifest: manifest.getRegionManifests())
122    *      hri = regionManifest.getRegionInfo()
123    *      for (regionManifest.getFamilyFiles())
124    *        ...
125    */
126   public static SnapshotManifest open(final Configuration conf, final FileSystem fs,
127       final Path workingDir, final SnapshotDescription desc) throws IOException {
128     SnapshotManifest manifest = new SnapshotManifest(conf, fs, workingDir, desc, null);
129     manifest.load();
130     return manifest;
131   }
132 
133 
134   /**
135    * Add the table descriptor to the snapshot manifest
136    */
137   public void addTableDescriptor(final HTableDescriptor htd) throws IOException {
138     this.htd = htd;
139   }
140 
141   interface RegionVisitor<TRegion, TFamily> {
142     TRegion regionOpen(final HRegionInfo regionInfo) throws IOException;
143     void regionClose(final TRegion region) throws IOException;
144 
145     TFamily familyOpen(final TRegion region, final byte[] familyName) throws IOException;
146     void familyClose(final TRegion region, final TFamily family) throws IOException;
147 
148     void storeFile(final TRegion region, final TFamily family, final StoreFileInfo storeFile)
149       throws IOException;
150   }
151 
152   private RegionVisitor createRegionVisitor(final SnapshotDescription desc) throws IOException {
153     switch (getSnapshotFormat(desc)) {
154       case SnapshotManifestV1.DESCRIPTOR_VERSION:
155         return new SnapshotManifestV1.ManifestBuilder(conf, fs, workingDir);
156       case SnapshotManifestV2.DESCRIPTOR_VERSION:
157         return new SnapshotManifestV2.ManifestBuilder(conf, fs, workingDir);
158       default:
159         throw new CorruptedSnapshotException("Invalid Snapshot version: "+ desc.getVersion(), desc);
160     }
161   }
162 
163   /**
164    * Creates a 'manifest' for the specified region, by reading directly from the HRegion object.
165    * This is used by the "online snapshot" when the table is enabled.
166    */
167   public void addRegion(final HRegion region) throws IOException {
168     // Get the ManifestBuilder/RegionVisitor
169     RegionVisitor visitor = createRegionVisitor(desc);
170 
171     // Visit the region and add it to the manifest
172     addRegion(region, visitor);
173   }
174 
175   @VisibleForTesting
176   protected void addRegion(final HRegion region, RegionVisitor visitor) throws IOException {
177     // 1. dump region meta info into the snapshot directory
178     LOG.debug("Storing '" + region + "' region-info for snapshot.");
179     Object regionData = visitor.regionOpen(region.getRegionInfo());
180     monitor.rethrowException();
181 
182     // 2. iterate through all the stores in the region
183     LOG.debug("Creating references for hfiles");
184 
185     for (Store store : region.getStores()) {
186       // 2.1. build the snapshot reference for the store
187       Object familyData = visitor.familyOpen(regionData, store.getFamily().getName());
188       monitor.rethrowException();
189 
190       List<StoreFile> storeFiles = new ArrayList<StoreFile>(store.getStorefiles());
191       if (LOG.isDebugEnabled()) {
192         LOG.debug("Adding snapshot references for " + storeFiles  + " hfiles");
193       }
194 
195       // 2.2. iterate through all the store's files and create "references".
196       for (int i = 0, sz = storeFiles.size(); i < sz; i++) {
197         StoreFile storeFile = storeFiles.get(i);
198         monitor.rethrowException();
199 
200         // create "reference" to this store file.
201         LOG.debug("Adding reference for file (" + (i+1) + "/" + sz + "): " + storeFile.getPath());
202         visitor.storeFile(regionData, familyData, storeFile.getFileInfo());
203       }
204       visitor.familyClose(regionData, familyData);
205     }
206     visitor.regionClose(regionData);
207   }
208 
209   /**
210    * Creates a 'manifest' for the specified region, by reading directly from the disk.
211    * This is used by the "offline snapshot" when the table is disabled.
212    */
213   public void addRegion(final Path tableDir, final HRegionInfo regionInfo) throws IOException {
214     // Get the ManifestBuilder/RegionVisitor
215     RegionVisitor visitor = createRegionVisitor(desc);
216 
217     // Visit the region and add it to the manifest
218     addRegion(tableDir, regionInfo, visitor);
219   }
220 
221   @VisibleForTesting
222   protected void addRegion(final Path tableDir, final HRegionInfo regionInfo, RegionVisitor visitor)
223       throws IOException {
224 
225     // Open the RegionFS
226     HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(conf, fs,
227         tableDir, regionInfo, true);
228     monitor.rethrowException();
229 
230     // 1. dump region meta info into the snapshot directory
231     LOG.debug("Storing region-info for snapshot.");
232     Object regionData = visitor.regionOpen(regionInfo);
233     monitor.rethrowException();
234 
235     // 2. iterate through all the stores in the region
236     LOG.debug("Creating references for hfiles");
237 
238     // This ensures that we have an atomic view of the directory as long as we have < ls limit
239     // (batch size of the files in a directory) on the namenode. Otherwise, we get back the files in
240     // batches and may miss files being added/deleted. This could be more robust (iteratively
241     // checking to see if we have all the files until we are sure), but the limit is currently 1000
242     // files/batch, far more than the number of store files under a single column family.
243     Collection<String> familyNames = regionFs.getFamilies();
244     if (familyNames != null) {
245       for (String familyName: familyNames) {
246         Object familyData = visitor.familyOpen(regionData, Bytes.toBytes(familyName));
247         monitor.rethrowException();
248 
249         Collection<StoreFileInfo> storeFiles = regionFs.getStoreFiles(familyName);
250         if (storeFiles == null) {
251           LOG.debug("No files under family: " + familyName);
252           continue;
253         }
254 
255         // 2.1. build the snapshot reference for the store
256         if (LOG.isDebugEnabled()) {
257           LOG.debug("Adding snapshot references for " + storeFiles  + " hfiles");
258         }
259 
260         // 2.2. iterate through all the store's files and create "references".
261         int i = 0;
262         int sz = storeFiles.size();
263         for (StoreFileInfo storeFile: storeFiles) {
264           monitor.rethrowException();
265 
266           // create "reference" to this store file.
267           LOG.debug("Adding reference for file ("+ (++i) +"/" + sz + "): " + storeFile.getPath());
268           visitor.storeFile(regionData, familyData, storeFile);
269         }
270         visitor.familyClose(regionData, familyData);
271       }
272     }
273     visitor.regionClose(regionData);
274   }
275 
276   /**
277    * Load the information in the SnapshotManifest. Called by SnapshotManifest.open()
278    *
279    * If the format is v2 and there is no data-manifest, means that we are loading an
280    * in-progress snapshot. Since we support rolling-upgrades, we loook for v1 and v2
281    * regions format.
282    */
283   private void load() throws IOException {
284     switch (getSnapshotFormat(desc)) {
285       case SnapshotManifestV1.DESCRIPTOR_VERSION: {
286         this.htd = FSTableDescriptors.getTableDescriptorFromFs(fs, workingDir);
287         ThreadPoolExecutor tpool = createExecutor("SnapshotManifestLoader");
288         try {
289           this.regionManifests =
290             SnapshotManifestV1.loadRegionManifests(conf, tpool, fs, workingDir, desc);
291         } finally {
292           tpool.shutdown();
293         }
294         break;
295       }
296       case SnapshotManifestV2.DESCRIPTOR_VERSION: {
297         SnapshotDataManifest dataManifest = readDataManifest();
298         if (dataManifest != null) {
299           htd = HTableDescriptor.convert(dataManifest.getTableSchema());
300           regionManifests = dataManifest.getRegionManifestsList();
301         } else {
302           // Compatibility, load the v1 regions
303           // This happens only when the snapshot is in-progress and the cache wants to refresh.
304           List<SnapshotRegionManifest> v1Regions, v2Regions;
305           ThreadPoolExecutor tpool = createExecutor("SnapshotManifestLoader");
306           try {
307             v1Regions = SnapshotManifestV1.loadRegionManifests(conf, tpool, fs, workingDir, desc);
308             v2Regions = SnapshotManifestV2.loadRegionManifests(conf, tpool, fs, workingDir, desc,
309                 manifestSizeLimit);
310           } catch (InvalidProtocolBufferException e) {
311             throw new CorruptedSnapshotException("unable to parse region manifest " +
312                 e.getMessage(), e);
313           } finally {
314             tpool.shutdown();
315           }
316           if (v1Regions != null && v2Regions != null) {
317             regionManifests =
318               new ArrayList<SnapshotRegionManifest>(v1Regions.size() + v2Regions.size());
319             regionManifests.addAll(v1Regions);
320             regionManifests.addAll(v2Regions);
321           } else if (v1Regions != null) {
322             regionManifests = v1Regions;
323           } else /* if (v2Regions != null) */ {
324             regionManifests = v2Regions;
325           }
326         }
327         break;
328       }
329       default:
330         throw new CorruptedSnapshotException("Invalid Snapshot version: "+ desc.getVersion(), desc);
331     }
332   }
333 
334   /**
335    * Get the current snapshot working dir
336    */
337   public Path getSnapshotDir() {
338     return this.workingDir;
339   }
340 
341   /**
342    * Get the SnapshotDescription
343    */
344   public SnapshotDescription getSnapshotDescription() {
345     return this.desc;
346   }
347 
348   /**
349    * Get the table descriptor from the Snapshot
350    */
351   public HTableDescriptor getTableDescriptor() {
352     return this.htd;
353   }
354 
355   /**
356    * Get all the Region Manifest from the snapshot
357    */
358   public List<SnapshotRegionManifest> getRegionManifests() {
359     return this.regionManifests;
360   }
361 
362   /**
363    * Get all the Region Manifest from the snapshot.
364    * This is an helper to get a map with the region encoded name
365    */
366   public Map<String, SnapshotRegionManifest> getRegionManifestsMap() {
367     if (regionManifests == null || regionManifests.size() == 0) return null;
368 
369     HashMap<String, SnapshotRegionManifest> regionsMap =
370         new HashMap<String, SnapshotRegionManifest>(regionManifests.size());
371     for (SnapshotRegionManifest manifest: regionManifests) {
372       String regionName = getRegionNameFromManifest(manifest);
373       regionsMap.put(regionName, manifest);
374     }
375     return regionsMap;
376   }
377 
378   public void consolidate() throws IOException {
379     if (getSnapshotFormat(desc) == SnapshotManifestV1.DESCRIPTOR_VERSION) {
380       Path rootDir = FSUtils.getRootDir(conf);
381       LOG.info("Using old Snapshot Format");
382       // write a copy of descriptor to the snapshot directory
383       new FSTableDescriptors(conf, fs, rootDir)
384         .createTableDescriptorForTableDirectory(workingDir, htd, false);
385     } else {
386       LOG.debug("Convert to Single Snapshot Manifest");
387       convertToV2SingleManifest();
388     }
389   }
390 
391   /*
392    * In case of rolling-upgrade, we try to read all the formats and build
393    * the snapshot with the latest format.
394    */
395   private void convertToV2SingleManifest() throws IOException {
396     // Try to load v1 and v2 regions
397     List<SnapshotRegionManifest> v1Regions, v2Regions;
398     ThreadPoolExecutor tpool = createExecutor("SnapshotManifestLoader");
399     try {
400       v1Regions = SnapshotManifestV1.loadRegionManifests(conf, tpool, fs, workingDir, desc);
401       v2Regions = SnapshotManifestV2.loadRegionManifests(conf, tpool, fs, workingDir, desc,
402           manifestSizeLimit);
403     } finally {
404       tpool.shutdown();
405     }
406 
407     SnapshotDataManifest.Builder dataManifestBuilder = SnapshotDataManifest.newBuilder();
408     dataManifestBuilder.setTableSchema(htd.convert());
409 
410     if (v1Regions != null && v1Regions.size() > 0) {
411       dataManifestBuilder.addAllRegionManifests(v1Regions);
412     }
413     if (v2Regions != null && v2Regions.size() > 0) {
414       dataManifestBuilder.addAllRegionManifests(v2Regions);
415     }
416 
417     // Write the v2 Data Manifest.
418     // Once the data-manifest is written, the snapshot can be considered complete.
419     // Currently snapshots are written in a "temporary" directory and later
420     // moved to the "complated" snapshot directory.
421     SnapshotDataManifest dataManifest = dataManifestBuilder.build();
422     writeDataManifest(dataManifest);
423     this.regionManifests = dataManifest.getRegionManifestsList();
424 
425     // Remove the region manifests. Everything is now in the data-manifest.
426     // The delete operation is "relaxed", unless we get an exception we keep going.
427     // The extra files in the snapshot directory will not give any problem,
428     // since they have the same content as the data manifest, and even by re-reading
429     // them we will get the same information.
430     if (v1Regions != null && v1Regions.size() > 0) {
431       for (SnapshotRegionManifest regionManifest: v1Regions) {
432         SnapshotManifestV1.deleteRegionManifest(fs, workingDir, regionManifest);
433       }
434     }
435     if (v2Regions != null && v2Regions.size() > 0) {
436       for (SnapshotRegionManifest regionManifest: v2Regions) {
437         SnapshotManifestV2.deleteRegionManifest(fs, workingDir, regionManifest);
438       }
439     }
440   }
441 
442   /*
443    * Write the SnapshotDataManifest file
444    */
445   private void writeDataManifest(final SnapshotDataManifest manifest)
446       throws IOException {
447     FSDataOutputStream stream = fs.create(new Path(workingDir, DATA_MANIFEST_NAME));
448     try {
449       manifest.writeTo(stream);
450     } finally {
451       stream.close();
452     }
453   }
454 
455   /*
456    * Read the SnapshotDataManifest file
457    */
458   private SnapshotDataManifest readDataManifest() throws IOException {
459     FSDataInputStream in = null;
460     try {
461       in = fs.open(new Path(workingDir, DATA_MANIFEST_NAME));
462       CodedInputStream cin = CodedInputStream.newInstance(in);
463       cin.setSizeLimit(manifestSizeLimit);
464       return SnapshotDataManifest.parseFrom(cin);
465     } catch (FileNotFoundException e) {
466       return null;
467     } catch (InvalidProtocolBufferException e) {
468       throw new CorruptedSnapshotException("unable to parse data manifest " + e.getMessage(), e);
469     } finally {
470       if (in != null) in.close();
471     }
472   }
473 
474   private ThreadPoolExecutor createExecutor(final String name) {
475     return createExecutor(conf, name);
476   }
477 
478   public static ThreadPoolExecutor createExecutor(final Configuration conf, final String name) {
479     int maxThreads = conf.getInt("hbase.snapshot.thread.pool.max", 8);
480     return Threads.getBoundedCachedThreadPool(maxThreads, 30L, TimeUnit.SECONDS,
481               Threads.getNamedThreadFactory(name));
482   }
483 
484   /**
485    * Extract the region encoded name from the region manifest
486    */
487   static String getRegionNameFromManifest(final SnapshotRegionManifest manifest) {
488     byte[] regionName = HRegionInfo.createRegionName(
489             ProtobufUtil.toTableName(manifest.getRegionInfo().getTableName()),
490             manifest.getRegionInfo().getStartKey().toByteArray(),
491             manifest.getRegionInfo().getRegionId(), true);
492     return HRegionInfo.encodeRegionName(regionName);
493   }
494 
495   /*
496    * Return the snapshot format
497    */
498   private static int getSnapshotFormat(final SnapshotDescription desc) {
499     return desc.hasVersion() ? desc.getVersion() : SnapshotManifestV1.DESCRIPTOR_VERSION;
500   }
501 }