001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.procedure2.store.wal; 019 020import java.io.FileNotFoundException; 021import java.io.IOException; 022import java.util.ArrayList; 023import java.util.Arrays; 024import java.util.Comparator; 025import java.util.HashSet; 026import java.util.Iterator; 027import java.util.LinkedList; 028import java.util.Set; 029import java.util.concurrent.LinkedTransferQueue; 030import java.util.concurrent.TimeUnit; 031import java.util.concurrent.atomic.AtomicBoolean; 032import java.util.concurrent.atomic.AtomicLong; 033import java.util.concurrent.atomic.AtomicReference; 034import java.util.concurrent.locks.Condition; 035import java.util.concurrent.locks.ReentrantLock; 036import org.apache.hadoop.conf.Configuration; 037import org.apache.hadoop.fs.FSDataOutputStream; 038import org.apache.hadoop.fs.FileAlreadyExistsException; 039import org.apache.hadoop.fs.FileStatus; 040import org.apache.hadoop.fs.FileSystem; 041import org.apache.hadoop.fs.Path; 042import org.apache.hadoop.fs.PathFilter; 043import org.apache.hadoop.hbase.HBaseConfiguration; 044import org.apache.hadoop.hbase.HConstants; 045import org.apache.hadoop.hbase.log.HBaseMarkers; 046import org.apache.hadoop.hbase.procedure2.Procedure; 047import org.apache.hadoop.hbase.procedure2.ProcedureExecutor; 048import org.apache.hadoop.hbase.procedure2.store.ProcedureStore; 049import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreBase; 050import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreTracker; 051import org.apache.hadoop.hbase.procedure2.util.ByteSlot; 052import org.apache.hadoop.hbase.procedure2.util.StringUtils; 053import org.apache.hadoop.hbase.util.CommonFSUtils; 054import org.apache.hadoop.hbase.util.Threads; 055import org.apache.hadoop.ipc.RemoteException; 056import org.apache.yetus.audience.InterfaceAudience; 057import org.slf4j.Logger; 058import org.slf4j.LoggerFactory; 059 060import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 061import org.apache.hbase.thirdparty.org.apache.commons.collections4.queue.CircularFifoQueue; 062 063import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureWALHeader; 064 065/** 066 * WAL implementation of the ProcedureStore. 067 * <p/> 068 * When starting, the upper layer will first call {@link #start(int)}, then {@link #recoverLease()}, 069 * then {@link #load(ProcedureLoader)}. 070 * <p/> 071 * In {@link #recoverLease()}, we will get the lease by closing all the existing wal files(by 072 * calling recoverFileLease), and creating a new wal writer. And we will also get the list of all 073 * the old wal files. 074 * <p/> 075 * FIXME: notice that the current recover lease implementation is problematic, it can not deal with 076 * the races if there are two master both wants to acquire the lease... 077 * <p/> 078 * In {@link #load(ProcedureLoader)} method, we will load all the active procedures. See the 079 * comments of this method for more details. 080 * <p/> 081 * The actual logging way is a bit like our FileSystem based WAL implementation as RS side. There is 082 * a {@link #slots}, which is more like the ring buffer, and in the insert, update and delete 083 * methods we will put thing into the {@link #slots} and wait. And there is a background sync 084 * thread(see the {@link #syncLoop()} method) which get data from the {@link #slots} and write them 085 * to the FileSystem, and notify the caller that we have finished. 086 * <p/> 087 * TODO: try using disruptor to increase performance and simplify the logic? 088 * <p/> 089 * The {@link #storeTracker} keeps track of the modified procedures in the newest wal file, which is 090 * also the one being written currently. And the deleted bits in it are for all the procedures, not 091 * only the ones in the newest wal file. And when rolling a log, we will first store it in the 092 * trailer of the current wal file, and then reset its modified bits, so that it can start to track 093 * the modified procedures for the new wal file. 094 * <p/> 095 * The {@link #holdingCleanupTracker} is used to test whether we are safe to delete the oldest wal 096 * file. When there are log rolling and there are more than 1 wal files, we will make use of it. It 097 * will first be initialized to the oldest file's tracker(which is stored in the trailer), using the 098 * method {@link ProcedureStoreTracker#resetTo(ProcedureStoreTracker, boolean)}, and then merge it 099 * with the tracker of every newer wal files, using the 100 * {@link ProcedureStoreTracker#setDeletedIfModifiedInBoth(ProcedureStoreTracker)}. 101 * If we find out 102 * that all the modified procedures for the oldest wal file are modified or deleted in newer wal 103 * files, then we can delete it. This is because that, every time we call 104 * {@link ProcedureStore#insert(Procedure[])} or {@link ProcedureStore#update(Procedure)}, we will 105 * persist the full state of a Procedure, so the earlier wal records for this procedure can all be 106 * deleted. 107 * @see ProcedureWALPrettyPrinter for printing content of a single WAL. 108 * @see #main(String[]) to parse a directory of MasterWALProcs. 109 */ 110@InterfaceAudience.Private 111public class WALProcedureStore extends ProcedureStoreBase { 112 private static final Logger LOG = LoggerFactory.getLogger(WALProcedureStore.class); 113 public static final String LOG_PREFIX = "pv2-"; 114 /** Used to construct the name of the log directory for master procedures */ 115 public static final String MASTER_PROCEDURE_LOGDIR = "MasterProcWALs"; 116 117 118 public interface LeaseRecovery { 119 void recoverFileLease(FileSystem fs, Path path) throws IOException; 120 } 121 122 public static final String WAL_COUNT_WARN_THRESHOLD_CONF_KEY = 123 "hbase.procedure.store.wal.warn.threshold"; 124 private static final int DEFAULT_WAL_COUNT_WARN_THRESHOLD = 10; 125 126 public static final String EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY = 127 "hbase.procedure.store.wal.exec.cleanup.on.load"; 128 private static final boolean DEFAULT_EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY = true; 129 130 public static final String MAX_RETRIES_BEFORE_ROLL_CONF_KEY = 131 "hbase.procedure.store.wal.max.retries.before.roll"; 132 private static final int DEFAULT_MAX_RETRIES_BEFORE_ROLL = 3; 133 134 public static final String WAIT_BEFORE_ROLL_CONF_KEY = 135 "hbase.procedure.store.wal.wait.before.roll"; 136 private static final int DEFAULT_WAIT_BEFORE_ROLL = 500; 137 138 public static final String ROLL_RETRIES_CONF_KEY = 139 "hbase.procedure.store.wal.max.roll.retries"; 140 private static final int DEFAULT_ROLL_RETRIES = 3; 141 142 public static final String MAX_SYNC_FAILURE_ROLL_CONF_KEY = 143 "hbase.procedure.store.wal.sync.failure.roll.max"; 144 private static final int DEFAULT_MAX_SYNC_FAILURE_ROLL = 3; 145 146 public static final String PERIODIC_ROLL_CONF_KEY = 147 "hbase.procedure.store.wal.periodic.roll.msec"; 148 private static final int DEFAULT_PERIODIC_ROLL = 60 * 60 * 1000; // 1h 149 150 public static final String SYNC_WAIT_MSEC_CONF_KEY = "hbase.procedure.store.wal.sync.wait.msec"; 151 private static final int DEFAULT_SYNC_WAIT_MSEC = 100; 152 153 public static final String USE_HSYNC_CONF_KEY = "hbase.procedure.store.wal.use.hsync"; 154 private static final boolean DEFAULT_USE_HSYNC = true; 155 156 public static final String ROLL_THRESHOLD_CONF_KEY = "hbase.procedure.store.wal.roll.threshold"; 157 private static final long DEFAULT_ROLL_THRESHOLD = 32 * 1024 * 1024; // 32M 158 159 public static final String STORE_WAL_SYNC_STATS_COUNT = 160 "hbase.procedure.store.wal.sync.stats.count"; 161 private static final int DEFAULT_SYNC_STATS_COUNT = 10; 162 163 private final LinkedList<ProcedureWALFile> logs = new LinkedList<>(); 164 private final ProcedureStoreTracker holdingCleanupTracker = new ProcedureStoreTracker(); 165 private final ProcedureStoreTracker storeTracker = new ProcedureStoreTracker(); 166 private final ReentrantLock lock = new ReentrantLock(); 167 private final Condition waitCond = lock.newCondition(); 168 private final Condition slotCond = lock.newCondition(); 169 private final Condition syncCond = lock.newCondition(); 170 171 private final LeaseRecovery leaseRecovery; 172 private final Configuration conf; 173 private final FileSystem fs; 174 private final Path walDir; 175 private final Path walArchiveDir; 176 private final boolean enforceStreamCapability; 177 178 private final AtomicReference<Throwable> syncException = new AtomicReference<>(); 179 private final AtomicBoolean loading = new AtomicBoolean(true); 180 private final AtomicBoolean inSync = new AtomicBoolean(false); 181 private final AtomicLong totalSynced = new AtomicLong(0); 182 private final AtomicLong lastRollTs = new AtomicLong(0); 183 private final AtomicLong syncId = new AtomicLong(0); 184 185 private LinkedTransferQueue<ByteSlot> slotsCache = null; 186 private Set<ProcedureWALFile> corruptedLogs = null; 187 private FSDataOutputStream stream = null; 188 private int runningProcCount = 1; 189 private long flushLogId = 0; 190 private int syncMaxSlot = 1; 191 private int slotIndex = 0; 192 private Thread syncThread; 193 private ByteSlot[] slots; 194 195 private int walCountWarnThreshold; 196 private int maxRetriesBeforeRoll; 197 private int maxSyncFailureRoll; 198 private int waitBeforeRoll; 199 private int rollRetries; 200 private int periodicRollMsec; 201 private long rollThreshold; 202 private boolean useHsync; 203 private int syncWaitMsec; 204 205 // Variables used for UI display 206 private CircularFifoQueue<SyncMetrics> syncMetricsQueue; 207 208 public static class SyncMetrics { 209 private long timestamp; 210 private long syncWaitMs; 211 private long totalSyncedBytes; 212 private int syncedEntries; 213 private float syncedPerSec; 214 215 public long getTimestamp() { 216 return timestamp; 217 } 218 219 public long getSyncWaitMs() { 220 return syncWaitMs; 221 } 222 223 public long getTotalSyncedBytes() { 224 return totalSyncedBytes; 225 } 226 227 public long getSyncedEntries() { 228 return syncedEntries; 229 } 230 231 public float getSyncedPerSec() { 232 return syncedPerSec; 233 } 234 } 235 236 public WALProcedureStore(final Configuration conf, final LeaseRecovery leaseRecovery) 237 throws IOException { 238 this(conf, 239 new Path(CommonFSUtils.getWALRootDir(conf), MASTER_PROCEDURE_LOGDIR), 240 new Path(CommonFSUtils.getWALRootDir(conf), HConstants.HREGION_OLDLOGDIR_NAME), 241 leaseRecovery); 242 } 243 244 @VisibleForTesting 245 public WALProcedureStore(final Configuration conf, final Path walDir, final Path walArchiveDir, 246 final LeaseRecovery leaseRecovery) throws IOException { 247 this.conf = conf; 248 this.leaseRecovery = leaseRecovery; 249 this.walDir = walDir; 250 this.walArchiveDir = walArchiveDir; 251 this.fs = walDir.getFileSystem(conf); 252 this.enforceStreamCapability = conf.getBoolean(CommonFSUtils.UNSAFE_STREAM_CAPABILITY_ENFORCE, true); 253 254 // Create the log directory for the procedure store 255 if (!fs.exists(walDir)) { 256 if (!fs.mkdirs(walDir)) { 257 throw new IOException("Unable to mkdir " + walDir); 258 } 259 } 260 // Now that it exists, set the log policy 261 String storagePolicy = 262 conf.get(HConstants.WAL_STORAGE_POLICY, HConstants.DEFAULT_WAL_STORAGE_POLICY); 263 CommonFSUtils.setStoragePolicy(fs, walDir, storagePolicy); 264 265 // Create archive dir up front. Rename won't work w/o it up on HDFS. 266 if (this.walArchiveDir != null && !this.fs.exists(this.walArchiveDir)) { 267 if (this.fs.mkdirs(this.walArchiveDir)) { 268 LOG.debug("Created Procedure Store WAL archive dir {}", this.walArchiveDir); 269 } else { 270 LOG.warn("Failed create of {}", this.walArchiveDir); 271 } 272 } 273 } 274 275 @Override 276 public void start(int numSlots) throws IOException { 277 if (!setRunning(true)) { 278 return; 279 } 280 281 // Init buffer slots 282 loading.set(true); 283 runningProcCount = numSlots; 284 syncMaxSlot = numSlots; 285 slots = new ByteSlot[numSlots]; 286 slotsCache = new LinkedTransferQueue<>(); 287 while (slotsCache.size() < numSlots) { 288 slotsCache.offer(new ByteSlot()); 289 } 290 291 // Tunings 292 walCountWarnThreshold = 293 conf.getInt(WAL_COUNT_WARN_THRESHOLD_CONF_KEY, DEFAULT_WAL_COUNT_WARN_THRESHOLD); 294 maxRetriesBeforeRoll = 295 conf.getInt(MAX_RETRIES_BEFORE_ROLL_CONF_KEY, DEFAULT_MAX_RETRIES_BEFORE_ROLL); 296 maxSyncFailureRoll = conf.getInt(MAX_SYNC_FAILURE_ROLL_CONF_KEY, DEFAULT_MAX_SYNC_FAILURE_ROLL); 297 waitBeforeRoll = conf.getInt(WAIT_BEFORE_ROLL_CONF_KEY, DEFAULT_WAIT_BEFORE_ROLL); 298 rollRetries = conf.getInt(ROLL_RETRIES_CONF_KEY, DEFAULT_ROLL_RETRIES); 299 rollThreshold = conf.getLong(ROLL_THRESHOLD_CONF_KEY, DEFAULT_ROLL_THRESHOLD); 300 periodicRollMsec = conf.getInt(PERIODIC_ROLL_CONF_KEY, DEFAULT_PERIODIC_ROLL); 301 syncWaitMsec = conf.getInt(SYNC_WAIT_MSEC_CONF_KEY, DEFAULT_SYNC_WAIT_MSEC); 302 useHsync = conf.getBoolean(USE_HSYNC_CONF_KEY, DEFAULT_USE_HSYNC); 303 304 // WebUI 305 syncMetricsQueue = new CircularFifoQueue<>( 306 conf.getInt(STORE_WAL_SYNC_STATS_COUNT, DEFAULT_SYNC_STATS_COUNT)); 307 308 // Init sync thread 309 syncThread = new Thread("WALProcedureStoreSyncThread") { 310 @Override 311 public void run() { 312 try { 313 syncLoop(); 314 } catch (Throwable e) { 315 LOG.error("Got an exception from the sync-loop", e); 316 if (!isSyncAborted()) { 317 sendAbortProcessSignal(); 318 } 319 } 320 } 321 }; 322 syncThread.start(); 323 } 324 325 @Override 326 public void stop(final boolean abort) { 327 if (!setRunning(false)) { 328 return; 329 } 330 331 LOG.info("Stopping the WAL Procedure Store, isAbort=" + abort + 332 (isSyncAborted() ? " (self aborting)" : "")); 333 sendStopSignal(); 334 if (!isSyncAborted()) { 335 try { 336 while (syncThread.isAlive()) { 337 sendStopSignal(); 338 syncThread.join(250); 339 } 340 } catch (InterruptedException e) { 341 LOG.warn("join interrupted", e); 342 Thread.currentThread().interrupt(); 343 } 344 } 345 346 // Close the writer 347 closeCurrentLogStream(abort); 348 349 // Close the old logs 350 // they should be already closed, this is just in case the load fails 351 // and we call start() and then stop() 352 for (ProcedureWALFile log: logs) { 353 log.close(); 354 } 355 logs.clear(); 356 loading.set(true); 357 } 358 359 private void sendStopSignal() { 360 if (lock.tryLock()) { 361 try { 362 waitCond.signalAll(); 363 syncCond.signalAll(); 364 } finally { 365 lock.unlock(); 366 } 367 } 368 } 369 370 @Override 371 public int getNumThreads() { 372 return slots == null ? 0 : slots.length; 373 } 374 375 @Override 376 public int setRunningProcedureCount(final int count) { 377 this.runningProcCount = count > 0 ? Math.min(count, slots.length) : slots.length; 378 return this.runningProcCount; 379 } 380 381 public ProcedureStoreTracker getStoreTracker() { 382 return storeTracker; 383 } 384 385 public ArrayList<ProcedureWALFile> getActiveLogs() { 386 lock.lock(); 387 try { 388 return new ArrayList<>(logs); 389 } finally { 390 lock.unlock(); 391 } 392 } 393 394 public Set<ProcedureWALFile> getCorruptedLogs() { 395 return corruptedLogs; 396 } 397 398 @Override 399 public void recoverLease() throws IOException { 400 lock.lock(); 401 try { 402 LOG.debug("Starting WAL Procedure Store lease recovery"); 403 boolean afterFirstAttempt = false; 404 while (isRunning()) { 405 // Don't sleep before first attempt 406 if (afterFirstAttempt) { 407 LOG.trace("Sleep {} ms after first lease recovery attempt.", 408 waitBeforeRoll); 409 Threads.sleepWithoutInterrupt(waitBeforeRoll); 410 } else { 411 afterFirstAttempt = true; 412 } 413 FileStatus[] oldLogs = getLogFiles(); 414 // Get Log-MaxID and recover lease on old logs 415 try { 416 flushLogId = initOldLogs(oldLogs); 417 } catch (FileNotFoundException e) { 418 LOG.warn("Someone else is active and deleted logs. retrying.", e); 419 continue; 420 } 421 422 // Create new state-log 423 if (!rollWriter(flushLogId + 1)) { 424 // someone else has already created this log 425 LOG.debug("Someone else has already created log {}. Retrying.", flushLogId); 426 continue; 427 } 428 429 // We have the lease on the log 430 oldLogs = getLogFiles(); 431 if (getMaxLogId(oldLogs) > flushLogId) { 432 LOG.debug("Someone else created new logs. Expected maxLogId < {}", flushLogId); 433 logs.getLast().removeFile(this.walArchiveDir); 434 continue; 435 } 436 437 LOG.debug("Lease acquired for flushLogId={}", flushLogId); 438 break; 439 } 440 } finally { 441 lock.unlock(); 442 } 443 } 444 445 @Override 446 public void load(ProcedureLoader loader) throws IOException { 447 lock.lock(); 448 try { 449 if (logs.isEmpty()) { 450 throw new IllegalStateException("recoverLease() must be called before loading data"); 451 } 452 453 // Nothing to do, If we have only the current log. 454 if (logs.size() == 1) { 455 LOG.debug("No state logs to replay."); 456 loader.setMaxProcId(0); 457 loading.set(false); 458 return; 459 } 460 461 // Load the old logs 462 Iterator<ProcedureWALFile> it = logs.descendingIterator(); 463 it.next(); // Skip the current log 464 465 ProcedureWALFormat.load(it, storeTracker, new ProcedureWALFormat.Loader() { 466 467 @Override 468 public void setMaxProcId(long maxProcId) { 469 loader.setMaxProcId(maxProcId); 470 } 471 472 @Override 473 public void load(ProcedureIterator procIter) throws IOException { 474 loader.load(procIter); 475 } 476 477 @Override 478 public void handleCorrupted(ProcedureIterator procIter) throws IOException { 479 loader.handleCorrupted(procIter); 480 } 481 482 @Override 483 public void markCorruptedWAL(ProcedureWALFile log, IOException e) { 484 if (corruptedLogs == null) { 485 corruptedLogs = new HashSet<>(); 486 } 487 corruptedLogs.add(log); 488 // TODO: sideline corrupted log 489 } 490 }); 491 // if we fail when loading, we should prevent persisting the storeTracker later in the stop 492 // method. As it may happen that, we have finished constructing the modified and deleted bits, 493 // but before we call resetModified, we fail, then if we persist the storeTracker then when 494 // restarting, we will consider that all procedures have been included in this file and delete 495 // all the previous files. Obviously this not correct. So here we will only set loading to 496 // false when we successfully loaded all the procedures, and when closing we will skip 497 // persisting the store tracker. And also, this will prevent the sync thread to do 498 // periodicRoll, where we may also clean old logs. 499 loading.set(false); 500 // try to cleanup inactive wals and complete the operation 501 buildHoldingCleanupTracker(); 502 tryCleanupLogsOnLoad(); 503 } finally { 504 lock.unlock(); 505 } 506 } 507 508 private void tryCleanupLogsOnLoad() { 509 // nothing to cleanup. 510 if (logs.size() <= 1) { 511 return; 512 } 513 514 // the config says to not cleanup wals on load. 515 if (!conf.getBoolean(EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY, 516 DEFAULT_EXEC_WAL_CLEANUP_ON_LOAD_CONF_KEY)) { 517 LOG.debug("WALs cleanup on load is not enabled: " + getActiveLogs()); 518 return; 519 } 520 521 try { 522 periodicRoll(); 523 } catch (IOException e) { 524 LOG.warn("Unable to cleanup logs on load: " + e.getMessage(), e); 525 } 526 } 527 528 @Override 529 public void insert(Procedure<?> proc, Procedure<?>[] subprocs) { 530 if (LOG.isTraceEnabled()) { 531 LOG.trace("Insert " + proc + ", subproc=" + Arrays.toString(subprocs)); 532 } 533 534 ByteSlot slot = acquireSlot(); 535 try { 536 // Serialize the insert 537 long[] subProcIds = null; 538 if (subprocs != null) { 539 ProcedureWALFormat.writeInsert(slot, proc, subprocs); 540 subProcIds = new long[subprocs.length]; 541 for (int i = 0; i < subprocs.length; ++i) { 542 subProcIds[i] = subprocs[i].getProcId(); 543 } 544 } else { 545 assert !proc.hasParent(); 546 ProcedureWALFormat.writeInsert(slot, proc); 547 } 548 549 // Push the transaction data and wait until it is persisted 550 pushData(PushType.INSERT, slot, proc.getProcId(), subProcIds); 551 } catch (IOException e) { 552 // We are not able to serialize the procedure. 553 // this is a code error, and we are not able to go on. 554 LOG.error(HBaseMarkers.FATAL, "Unable to serialize one of the procedure: proc=" + 555 proc + ", subprocs=" + Arrays.toString(subprocs), e); 556 throw new RuntimeException(e); 557 } finally { 558 releaseSlot(slot); 559 } 560 } 561 562 @Override 563 public void insert(Procedure<?>[] procs) { 564 if (LOG.isTraceEnabled()) { 565 LOG.trace("Insert " + Arrays.toString(procs)); 566 } 567 568 ByteSlot slot = acquireSlot(); 569 try { 570 // Serialize the insert 571 long[] procIds = new long[procs.length]; 572 for (int i = 0; i < procs.length; ++i) { 573 assert !procs[i].hasParent(); 574 procIds[i] = procs[i].getProcId(); 575 ProcedureWALFormat.writeInsert(slot, procs[i]); 576 } 577 578 // Push the transaction data and wait until it is persisted 579 pushData(PushType.INSERT, slot, Procedure.NO_PROC_ID, procIds); 580 } catch (IOException e) { 581 // We are not able to serialize the procedure. 582 // this is a code error, and we are not able to go on. 583 LOG.error(HBaseMarkers.FATAL, "Unable to serialize one of the procedure: " + 584 Arrays.toString(procs), e); 585 throw new RuntimeException(e); 586 } finally { 587 releaseSlot(slot); 588 } 589 } 590 591 @Override 592 public void update(Procedure<?> proc) { 593 if (LOG.isTraceEnabled()) { 594 LOG.trace("Update " + proc); 595 } 596 597 ByteSlot slot = acquireSlot(); 598 try { 599 // Serialize the update 600 ProcedureWALFormat.writeUpdate(slot, proc); 601 602 // Push the transaction data and wait until it is persisted 603 pushData(PushType.UPDATE, slot, proc.getProcId(), null); 604 } catch (IOException e) { 605 // We are not able to serialize the procedure. 606 // this is a code error, and we are not able to go on. 607 LOG.error(HBaseMarkers.FATAL, "Unable to serialize the procedure: " + proc, e); 608 throw new RuntimeException(e); 609 } finally { 610 releaseSlot(slot); 611 } 612 } 613 614 @Override 615 public void delete(long procId) { 616 LOG.trace("Delete {}", procId); 617 ByteSlot slot = acquireSlot(); 618 try { 619 // Serialize the delete 620 ProcedureWALFormat.writeDelete(slot, procId); 621 622 // Push the transaction data and wait until it is persisted 623 pushData(PushType.DELETE, slot, procId, null); 624 } catch (IOException e) { 625 // We are not able to serialize the procedure. 626 // this is a code error, and we are not able to go on. 627 LOG.error(HBaseMarkers.FATAL, "Unable to serialize the procedure: " + procId, e); 628 throw new RuntimeException(e); 629 } finally { 630 releaseSlot(slot); 631 } 632 } 633 634 @Override 635 public void delete(Procedure<?> proc, long[] subProcIds) { 636 assert proc != null : "expected a non-null procedure"; 637 assert subProcIds != null && subProcIds.length > 0 : "expected subProcIds"; 638 if (LOG.isTraceEnabled()) { 639 LOG.trace("Update " + proc + " and Delete " + Arrays.toString(subProcIds)); 640 } 641 642 ByteSlot slot = acquireSlot(); 643 try { 644 // Serialize the delete 645 ProcedureWALFormat.writeDelete(slot, proc, subProcIds); 646 647 // Push the transaction data and wait until it is persisted 648 pushData(PushType.DELETE, slot, proc.getProcId(), subProcIds); 649 } catch (IOException e) { 650 // We are not able to serialize the procedure. 651 // this is a code error, and we are not able to go on. 652 LOG.error(HBaseMarkers.FATAL, "Unable to serialize the procedure: " + proc, e); 653 throw new RuntimeException(e); 654 } finally { 655 releaseSlot(slot); 656 } 657 } 658 659 @Override 660 public void delete(final long[] procIds, final int offset, final int count) { 661 if (count == 0) return; 662 if (offset == 0 && count == procIds.length) { 663 delete(procIds); 664 } else if (count == 1) { 665 delete(procIds[offset]); 666 } else { 667 delete(Arrays.copyOfRange(procIds, offset, offset + count)); 668 } 669 } 670 671 private void delete(long[] procIds) { 672 if (LOG.isTraceEnabled()) { 673 LOG.trace("Delete " + Arrays.toString(procIds)); 674 } 675 676 final ByteSlot slot = acquireSlot(); 677 try { 678 // Serialize the delete 679 for (int i = 0; i < procIds.length; ++i) { 680 ProcedureWALFormat.writeDelete(slot, procIds[i]); 681 } 682 683 // Push the transaction data and wait until it is persisted 684 pushData(PushType.DELETE, slot, Procedure.NO_PROC_ID, procIds); 685 } catch (IOException e) { 686 // We are not able to serialize the procedure. 687 // this is a code error, and we are not able to go on. 688 LOG.error("Unable to serialize the procedures: " + Arrays.toString(procIds), e); 689 throw new RuntimeException(e); 690 } finally { 691 releaseSlot(slot); 692 } 693 } 694 695 private ByteSlot acquireSlot() { 696 ByteSlot slot = slotsCache.poll(); 697 return slot != null ? slot : new ByteSlot(); 698 } 699 700 private void releaseSlot(final ByteSlot slot) { 701 slot.reset(); 702 slotsCache.offer(slot); 703 } 704 705 private enum PushType { INSERT, UPDATE, DELETE }; 706 707 private long pushData(final PushType type, final ByteSlot slot, 708 final long procId, final long[] subProcIds) { 709 if (!isRunning()) { 710 throw new RuntimeException("the store must be running before inserting data"); 711 } 712 if (logs.isEmpty()) { 713 throw new RuntimeException("recoverLease() must be called before inserting data"); 714 } 715 716 long logId = -1; 717 lock.lock(); 718 try { 719 // Wait for the sync to be completed 720 while (true) { 721 if (!isRunning()) { 722 throw new RuntimeException("store no longer running"); 723 } else if (isSyncAborted()) { 724 throw new RuntimeException("sync aborted", syncException.get()); 725 } else if (inSync.get()) { 726 syncCond.await(); 727 } else if (slotIndex >= syncMaxSlot) { 728 slotCond.signal(); 729 syncCond.await(); 730 } else { 731 break; 732 } 733 } 734 735 final long pushSyncId = syncId.get(); 736 updateStoreTracker(type, procId, subProcIds); 737 slots[slotIndex++] = slot; 738 logId = flushLogId; 739 740 // Notify that there is new data 741 if (slotIndex == 1) { 742 waitCond.signal(); 743 } 744 745 // Notify that the slots are full 746 if (slotIndex == syncMaxSlot) { 747 waitCond.signal(); 748 slotCond.signal(); 749 } 750 751 while (pushSyncId == syncId.get() && isRunning()) { 752 syncCond.await(); 753 } 754 } catch (InterruptedException e) { 755 Thread.currentThread().interrupt(); 756 sendAbortProcessSignal(); 757 throw new RuntimeException(e); 758 } finally { 759 lock.unlock(); 760 if (isSyncAborted()) { 761 throw new RuntimeException("sync aborted", syncException.get()); 762 } 763 } 764 return logId; 765 } 766 767 private void updateStoreTracker(final PushType type, 768 final long procId, final long[] subProcIds) { 769 switch (type) { 770 case INSERT: 771 if (subProcIds == null) { 772 storeTracker.insert(procId); 773 } else if (procId == Procedure.NO_PROC_ID) { 774 storeTracker.insert(subProcIds); 775 } else { 776 storeTracker.insert(procId, subProcIds); 777 holdingCleanupTracker.setDeletedIfModified(procId); 778 } 779 break; 780 case UPDATE: 781 storeTracker.update(procId); 782 holdingCleanupTracker.setDeletedIfModified(procId); 783 break; 784 case DELETE: 785 if (subProcIds != null && subProcIds.length > 0) { 786 storeTracker.delete(subProcIds); 787 holdingCleanupTracker.setDeletedIfModified(subProcIds); 788 } else { 789 storeTracker.delete(procId); 790 holdingCleanupTracker.setDeletedIfModified(procId); 791 } 792 break; 793 default: 794 throw new RuntimeException("invalid push type " + type); 795 } 796 } 797 798 private boolean isSyncAborted() { 799 return syncException.get() != null; 800 } 801 802 private void syncLoop() throws Throwable { 803 long totalSyncedToStore = 0; 804 inSync.set(false); 805 lock.lock(); 806 try { 807 while (isRunning()) { 808 try { 809 // Wait until new data is available 810 if (slotIndex == 0) { 811 if (!loading.get()) { 812 periodicRoll(); 813 } 814 815 if (LOG.isTraceEnabled()) { 816 float rollTsSec = getMillisFromLastRoll() / 1000.0f; 817 LOG.trace(String.format("Waiting for data. flushed=%s (%s/sec)", 818 StringUtils.humanSize(totalSynced.get()), 819 StringUtils.humanSize(totalSynced.get() / rollTsSec))); 820 } 821 822 waitCond.await(getMillisToNextPeriodicRoll(), TimeUnit.MILLISECONDS); 823 if (slotIndex == 0) { 824 // no data.. probably a stop() or a periodic roll 825 continue; 826 } 827 } 828 // Wait SYNC_WAIT_MSEC or the signal of "slots full" before flushing 829 syncMaxSlot = runningProcCount; 830 assert syncMaxSlot > 0 : "unexpected syncMaxSlot=" + syncMaxSlot; 831 final long syncWaitSt = System.currentTimeMillis(); 832 if (slotIndex != syncMaxSlot) { 833 slotCond.await(syncWaitMsec, TimeUnit.MILLISECONDS); 834 } 835 836 final long currentTs = System.currentTimeMillis(); 837 final long syncWaitMs = currentTs - syncWaitSt; 838 final float rollSec = getMillisFromLastRoll() / 1000.0f; 839 final float syncedPerSec = totalSyncedToStore / rollSec; 840 if (LOG.isTraceEnabled() && (syncWaitMs > 10 || slotIndex < syncMaxSlot)) { 841 LOG.trace(String.format("Sync wait %s, slotIndex=%s , totalSynced=%s (%s/sec)", 842 StringUtils.humanTimeDiff(syncWaitMs), slotIndex, 843 StringUtils.humanSize(totalSyncedToStore), 844 StringUtils.humanSize(syncedPerSec))); 845 } 846 847 // update webui circular buffers (TODO: get rid of allocations) 848 final SyncMetrics syncMetrics = new SyncMetrics(); 849 syncMetrics.timestamp = currentTs; 850 syncMetrics.syncWaitMs = syncWaitMs; 851 syncMetrics.syncedEntries = slotIndex; 852 syncMetrics.totalSyncedBytes = totalSyncedToStore; 853 syncMetrics.syncedPerSec = syncedPerSec; 854 syncMetricsQueue.add(syncMetrics); 855 856 // sync 857 inSync.set(true); 858 long slotSize = syncSlots(); 859 logs.getLast().addToSize(slotSize); 860 totalSyncedToStore = totalSynced.addAndGet(slotSize); 861 slotIndex = 0; 862 inSync.set(false); 863 syncId.incrementAndGet(); 864 } catch (InterruptedException e) { 865 Thread.currentThread().interrupt(); 866 syncException.compareAndSet(null, e); 867 sendAbortProcessSignal(); 868 throw e; 869 } catch (Throwable t) { 870 syncException.compareAndSet(null, t); 871 sendAbortProcessSignal(); 872 throw t; 873 } finally { 874 syncCond.signalAll(); 875 } 876 } 877 } finally { 878 lock.unlock(); 879 } 880 } 881 882 public ArrayList<SyncMetrics> getSyncMetrics() { 883 lock.lock(); 884 try { 885 return new ArrayList<>(syncMetricsQueue); 886 } finally { 887 lock.unlock(); 888 } 889 } 890 891 private long syncSlots() throws Throwable { 892 int retry = 0; 893 int logRolled = 0; 894 long totalSynced = 0; 895 do { 896 try { 897 totalSynced = syncSlots(stream, slots, 0, slotIndex); 898 break; 899 } catch (Throwable e) { 900 LOG.warn("unable to sync slots, retry=" + retry); 901 if (++retry >= maxRetriesBeforeRoll) { 902 if (logRolled >= maxSyncFailureRoll && isRunning()) { 903 LOG.error("Sync slots after log roll failed, abort.", e); 904 throw e; 905 } 906 907 if (!rollWriterWithRetries()) { 908 throw e; 909 } 910 911 logRolled++; 912 retry = 0; 913 } 914 } 915 } while (isRunning()); 916 return totalSynced; 917 } 918 919 protected long syncSlots(final FSDataOutputStream stream, final ByteSlot[] slots, 920 final int offset, final int count) throws IOException { 921 long totalSynced = 0; 922 for (int i = 0; i < count; ++i) { 923 final ByteSlot data = slots[offset + i]; 924 data.writeTo(stream); 925 totalSynced += data.size(); 926 } 927 928 syncStream(stream); 929 sendPostSyncSignal(); 930 931 if (LOG.isTraceEnabled()) { 932 LOG.trace("Sync slots=" + count + '/' + syncMaxSlot + 933 ", flushed=" + StringUtils.humanSize(totalSynced)); 934 } 935 return totalSynced; 936 } 937 938 protected void syncStream(final FSDataOutputStream stream) throws IOException { 939 if (useHsync) { 940 stream.hsync(); 941 } else { 942 stream.hflush(); 943 } 944 } 945 946 private boolean rollWriterWithRetries() { 947 for (int i = 0; i < rollRetries && isRunning(); ++i) { 948 if (i > 0) Threads.sleepWithoutInterrupt(waitBeforeRoll * i); 949 950 try { 951 if (rollWriter()) { 952 return true; 953 } 954 } catch (IOException e) { 955 LOG.warn("Unable to roll the log, attempt=" + (i + 1), e); 956 } 957 } 958 LOG.error(HBaseMarkers.FATAL, "Unable to roll the log"); 959 return false; 960 } 961 962 private boolean tryRollWriter() { 963 try { 964 return rollWriter(); 965 } catch (IOException e) { 966 LOG.warn("Unable to roll the log", e); 967 return false; 968 } 969 } 970 971 public long getMillisToNextPeriodicRoll() { 972 if (lastRollTs.get() > 0 && periodicRollMsec > 0) { 973 return periodicRollMsec - getMillisFromLastRoll(); 974 } 975 return Long.MAX_VALUE; 976 } 977 978 public long getMillisFromLastRoll() { 979 return (System.currentTimeMillis() - lastRollTs.get()); 980 } 981 982 @VisibleForTesting 983 void periodicRollForTesting() throws IOException { 984 lock.lock(); 985 try { 986 periodicRoll(); 987 } finally { 988 lock.unlock(); 989 } 990 } 991 992 @VisibleForTesting 993 public boolean rollWriterForTesting() throws IOException { 994 lock.lock(); 995 try { 996 return rollWriter(); 997 } finally { 998 lock.unlock(); 999 } 1000 } 1001 1002 @VisibleForTesting 1003 void removeInactiveLogsForTesting() throws Exception { 1004 lock.lock(); 1005 try { 1006 removeInactiveLogs(); 1007 } finally { 1008 lock.unlock(); 1009 } 1010 } 1011 1012 private void periodicRoll() throws IOException { 1013 if (storeTracker.isEmpty()) { 1014 LOG.trace("no active procedures"); 1015 tryRollWriter(); 1016 removeAllLogs(flushLogId - 1, "no active procedures"); 1017 } else { 1018 if (storeTracker.isAllModified()) { 1019 LOG.trace("all the active procedures are in the latest log"); 1020 removeAllLogs(flushLogId - 1, "all the active procedures are in the latest log"); 1021 } 1022 1023 // if the log size has exceeded the roll threshold 1024 // or the periodic roll timeout is expired, try to roll the wal. 1025 if (totalSynced.get() > rollThreshold || getMillisToNextPeriodicRoll() <= 0) { 1026 tryRollWriter(); 1027 } 1028 1029 removeInactiveLogs(); 1030 } 1031 } 1032 1033 private boolean rollWriter() throws IOException { 1034 if (!isRunning()) { 1035 return false; 1036 } 1037 1038 // Create new state-log 1039 if (!rollWriter(flushLogId + 1)) { 1040 LOG.warn("someone else has already created log {}", flushLogId); 1041 return false; 1042 } 1043 1044 // We have the lease on the log, 1045 // but we should check if someone else has created new files 1046 if (getMaxLogId(getLogFiles()) > flushLogId) { 1047 LOG.warn("Someone else created new logs. Expected maxLogId < {}", flushLogId); 1048 logs.getLast().removeFile(this.walArchiveDir); 1049 return false; 1050 } 1051 1052 // We have the lease on the log 1053 return true; 1054 } 1055 1056 @VisibleForTesting 1057 boolean rollWriter(long logId) throws IOException { 1058 assert logId > flushLogId : "logId=" + logId + " flushLogId=" + flushLogId; 1059 assert lock.isHeldByCurrentThread() : "expected to be the lock owner. " + lock.isLocked(); 1060 1061 ProcedureWALHeader header = ProcedureWALHeader.newBuilder() 1062 .setVersion(ProcedureWALFormat.HEADER_VERSION) 1063 .setType(ProcedureWALFormat.LOG_TYPE_STREAM) 1064 .setMinProcId(storeTracker.getActiveMinProcId()) 1065 .setLogId(logId) 1066 .build(); 1067 1068 FSDataOutputStream newStream = null; 1069 Path newLogFile = null; 1070 long startPos = -1; 1071 newLogFile = getLogFilePath(logId); 1072 try { 1073 newStream = CommonFSUtils.createForWal(fs, newLogFile, false); 1074 } catch (FileAlreadyExistsException e) { 1075 LOG.error("Log file with id={} already exists", logId, e); 1076 return false; 1077 } catch (RemoteException re) { 1078 LOG.warn("failed to create log file with id={}", logId, re); 1079 return false; 1080 } 1081 // After we create the stream but before we attempt to use it at all 1082 // ensure that we can provide the level of data safety we're configured 1083 // to provide. 1084 final String durability = useHsync ? "hsync" : "hflush"; 1085 if (enforceStreamCapability && !(CommonFSUtils.hasCapability(newStream, durability))) { 1086 throw new IllegalStateException("The procedure WAL relies on the ability to " + durability + 1087 " for proper operation during component failures, but the underlying filesystem does " + 1088 "not support doing so. Please check the config value of '" + USE_HSYNC_CONF_KEY + 1089 "' to set the desired level of robustness and ensure the config value of '" + 1090 CommonFSUtils.HBASE_WAL_DIR + "' points to a FileSystem mount that can provide it."); 1091 } 1092 try { 1093 ProcedureWALFormat.writeHeader(newStream, header); 1094 startPos = newStream.getPos(); 1095 } catch (IOException ioe) { 1096 LOG.warn("Encountered exception writing header", ioe); 1097 newStream.close(); 1098 return false; 1099 } 1100 1101 closeCurrentLogStream(false); 1102 1103 storeTracker.resetModified(); 1104 stream = newStream; 1105 flushLogId = logId; 1106 totalSynced.set(0); 1107 long rollTs = System.currentTimeMillis(); 1108 lastRollTs.set(rollTs); 1109 logs.add(new ProcedureWALFile(fs, newLogFile, header, startPos, rollTs)); 1110 1111 // if it's the first next WAL being added, build the holding cleanup tracker 1112 if (logs.size() == 2) { 1113 buildHoldingCleanupTracker(); 1114 } else if (logs.size() > walCountWarnThreshold) { 1115 LOG.warn("procedure WALs count={} above the warning threshold {}. check running procedures" + 1116 " to see if something is stuck.", logs.size(), walCountWarnThreshold); 1117 // This is just like what we have done at RS side when there are too many wal files. For RS, 1118 // if there are too many wal files, we will find out the wal entries in the oldest file, and 1119 // tell the upper layer to flush these regions so the wal entries will be useless and then we 1120 // can delete the wal file. For WALProcedureStore, the assumption is that, if all the 1121 // procedures recorded in a proc wal file are modified or deleted in a new proc wal file, then 1122 // we are safe to delete it. So here if there are too many proc wal files, we will find out 1123 // the procedure ids in the oldest file, which are neither modified nor deleted in newer proc 1124 // wal files, and tell upper layer to update the state of these procedures to the newest proc 1125 // wal file(by calling ProcedureStore.update), then we are safe to delete the oldest proc wal 1126 // file. 1127 sendForceUpdateSignal(holdingCleanupTracker.getAllActiveProcIds()); 1128 } 1129 1130 LOG.info("Rolled new Procedure Store WAL, id={}", logId); 1131 return true; 1132 } 1133 1134 private void closeCurrentLogStream(boolean abort) { 1135 if (stream == null || logs.isEmpty()) { 1136 return; 1137 } 1138 1139 try { 1140 ProcedureWALFile log = logs.getLast(); 1141 // If the loading flag is true, it usually means that we fail when loading procedures, so we 1142 // should not persist the store tracker, as its state may not be correct. 1143 if (!loading.get()) { 1144 log.setProcIds(storeTracker.getModifiedMinProcId(), storeTracker.getModifiedMaxProcId()); 1145 log.updateLocalTracker(storeTracker); 1146 if (!abort) { 1147 long trailerSize = ProcedureWALFormat.writeTrailer(stream, storeTracker); 1148 log.addToSize(trailerSize); 1149 } 1150 } 1151 } catch (IOException e) { 1152 LOG.warn("Unable to write the trailer", e); 1153 } 1154 try { 1155 stream.close(); 1156 } catch (IOException e) { 1157 LOG.error("Unable to close the stream", e); 1158 } 1159 stream = null; 1160 } 1161 1162 // ========================================================================== 1163 // Log Files cleaner helpers 1164 // ========================================================================== 1165 private void removeInactiveLogs() throws IOException { 1166 // We keep track of which procedures are holding the oldest WAL in 'holdingCleanupTracker'. 1167 // once there is nothing olding the oldest WAL we can remove it. 1168 while (logs.size() > 1 && holdingCleanupTracker.isEmpty()) { 1169 LOG.info("Remove the oldest log {}", logs.getFirst()); 1170 removeLogFile(logs.getFirst(), walArchiveDir); 1171 buildHoldingCleanupTracker(); 1172 } 1173 1174 // TODO: In case we are holding up a lot of logs for long time we should 1175 // rewrite old procedures (in theory parent procs) to the new WAL. 1176 } 1177 1178 private void buildHoldingCleanupTracker() { 1179 if (logs.size() <= 1) { 1180 // we only have one wal, so nothing to do 1181 holdingCleanupTracker.reset(); 1182 return; 1183 } 1184 1185 // compute the holding tracker. 1186 // - the first WAL is used for the 'updates' 1187 // - the global tracker will be used to determine whether a procedure has been deleted 1188 // - other trackers will be used to determine whether a procedure has been updated, as a deleted 1189 // procedure can always be detected by checking the global tracker, we can save the deleted 1190 // checks when applying other trackers 1191 holdingCleanupTracker.resetTo(logs.getFirst().getTracker(), true); 1192 holdingCleanupTracker.setDeletedIfDeletedByThem(storeTracker); 1193 // the logs is a linked list, so avoid calling get(index) on it. 1194 Iterator<ProcedureWALFile> iter = logs.iterator(); 1195 // skip the tracker for the first file when creating the iterator. 1196 iter.next(); 1197 ProcedureStoreTracker tracker = iter.next().getTracker(); 1198 // testing iter.hasNext after calling iter.next to skip applying the tracker for last file, 1199 // which is just the storeTracker above. 1200 while (iter.hasNext()) { 1201 holdingCleanupTracker.setDeletedIfModifiedInBoth(tracker); 1202 if (holdingCleanupTracker.isEmpty()) { 1203 break; 1204 } 1205 tracker = iter.next().getTracker(); 1206 } 1207 } 1208 1209 /** 1210 * Remove all logs with logId <= {@code lastLogId}. 1211 */ 1212 private void removeAllLogs(long lastLogId, String why) { 1213 if (logs.size() <= 1) { 1214 return; 1215 } 1216 1217 LOG.info("Remove all state logs with ID less than {}, since {}", lastLogId, why); 1218 1219 boolean removed = false; 1220 while (logs.size() > 1) { 1221 ProcedureWALFile log = logs.getFirst(); 1222 if (lastLogId < log.getLogId()) { 1223 break; 1224 } 1225 removeLogFile(log, walArchiveDir); 1226 removed = true; 1227 } 1228 1229 if (removed) { 1230 buildHoldingCleanupTracker(); 1231 } 1232 } 1233 1234 private boolean removeLogFile(final ProcedureWALFile log, final Path walArchiveDir) { 1235 try { 1236 LOG.trace("Removing log={}", log); 1237 log.removeFile(walArchiveDir); 1238 logs.remove(log); 1239 LOG.debug("Removed log={}, activeLogs={}", log, logs); 1240 assert logs.size() > 0 : "expected at least one log"; 1241 } catch (IOException e) { 1242 LOG.error("Unable to remove log: " + log, e); 1243 return false; 1244 } 1245 return true; 1246 } 1247 1248 // ========================================================================== 1249 // FileSystem Log Files helpers 1250 // ========================================================================== 1251 public Path getWALDir() { 1252 return this.walDir; 1253 } 1254 1255 @VisibleForTesting 1256 Path getWalArchiveDir() { 1257 return this.walArchiveDir; 1258 } 1259 1260 public FileSystem getFileSystem() { 1261 return this.fs; 1262 } 1263 1264 protected Path getLogFilePath(final long logId) throws IOException { 1265 return new Path(walDir, String.format(LOG_PREFIX + "%020d.log", logId)); 1266 } 1267 1268 private static long getLogIdFromName(final String name) { 1269 int end = name.lastIndexOf(".log"); 1270 int start = name.lastIndexOf('-') + 1; 1271 return Long.parseLong(name.substring(start, end)); 1272 } 1273 1274 private static final PathFilter WALS_PATH_FILTER = new PathFilter() { 1275 @Override 1276 public boolean accept(Path path) { 1277 String name = path.getName(); 1278 return name.startsWith(LOG_PREFIX) && name.endsWith(".log"); 1279 } 1280 }; 1281 1282 private static final Comparator<FileStatus> FILE_STATUS_ID_COMPARATOR = 1283 new Comparator<FileStatus>() { 1284 @Override 1285 public int compare(FileStatus a, FileStatus b) { 1286 final long aId = getLogIdFromName(a.getPath().getName()); 1287 final long bId = getLogIdFromName(b.getPath().getName()); 1288 return Long.compare(aId, bId); 1289 } 1290 }; 1291 1292 private FileStatus[] getLogFiles() throws IOException { 1293 try { 1294 FileStatus[] files = fs.listStatus(walDir, WALS_PATH_FILTER); 1295 Arrays.sort(files, FILE_STATUS_ID_COMPARATOR); 1296 return files; 1297 } catch (FileNotFoundException e) { 1298 LOG.warn("Log directory not found: " + e.getMessage()); 1299 return null; 1300 } 1301 } 1302 1303 /** 1304 * Make sure that the file set are gotten by calling {@link #getLogFiles()}, where we will sort 1305 * the file set by log id. 1306 * @return Max-LogID of the specified log file set 1307 */ 1308 private static long getMaxLogId(FileStatus[] logFiles) { 1309 if (logFiles == null || logFiles.length == 0) { 1310 return 0L; 1311 } 1312 return getLogIdFromName(logFiles[logFiles.length - 1].getPath().getName()); 1313 } 1314 1315 /** 1316 * Make sure that the file set are gotten by calling {@link #getLogFiles()}, where we will sort 1317 * the file set by log id. 1318 * @return Max-LogID of the specified log file set 1319 */ 1320 private long initOldLogs(FileStatus[] logFiles) throws IOException { 1321 if (logFiles == null || logFiles.length == 0) { 1322 return 0L; 1323 } 1324 long maxLogId = 0; 1325 for (int i = 0; i < logFiles.length; ++i) { 1326 final Path logPath = logFiles[i].getPath(); 1327 leaseRecovery.recoverFileLease(fs, logPath); 1328 if (!isRunning()) { 1329 throw new IOException("wal aborting"); 1330 } 1331 1332 maxLogId = Math.max(maxLogId, getLogIdFromName(logPath.getName())); 1333 ProcedureWALFile log = initOldLog(logFiles[i], this.walArchiveDir); 1334 if (log != null) { 1335 this.logs.add(log); 1336 } 1337 } 1338 initTrackerFromOldLogs(); 1339 return maxLogId; 1340 } 1341 1342 /** 1343 * If last log's tracker is not null, use it as {@link #storeTracker}. Otherwise, set storeTracker 1344 * as partial, and let {@link ProcedureWALFormatReader} rebuild it using entries in the log. 1345 */ 1346 private void initTrackerFromOldLogs() { 1347 if (logs.isEmpty() || !isRunning()) { 1348 return; 1349 } 1350 ProcedureWALFile log = logs.getLast(); 1351 if (!log.getTracker().isPartial()) { 1352 storeTracker.resetTo(log.getTracker()); 1353 } else { 1354 storeTracker.reset(); 1355 storeTracker.setPartialFlag(true); 1356 } 1357 } 1358 1359 /** 1360 * Loads given log file and it's tracker. 1361 */ 1362 private ProcedureWALFile initOldLog(final FileStatus logFile, final Path walArchiveDir) 1363 throws IOException { 1364 final ProcedureWALFile log = new ProcedureWALFile(fs, logFile); 1365 if (logFile.getLen() == 0) { 1366 LOG.warn("Remove uninitialized log: {}", logFile); 1367 log.removeFile(walArchiveDir); 1368 return null; 1369 } 1370 LOG.debug("Opening Pv2 {}", logFile); 1371 try { 1372 log.open(); 1373 } catch (ProcedureWALFormat.InvalidWALDataException e) { 1374 LOG.warn("Remove uninitialized log: {}", logFile, e); 1375 log.removeFile(walArchiveDir); 1376 return null; 1377 } catch (IOException e) { 1378 String msg = "Unable to read state log: " + logFile; 1379 LOG.error(msg, e); 1380 throw new IOException(msg, e); 1381 } 1382 1383 try { 1384 log.readTracker(); 1385 } catch (IOException e) { 1386 log.getTracker().reset(); 1387 log.getTracker().setPartialFlag(true); 1388 LOG.warn("Unable to read tracker for {}", log, e); 1389 } 1390 1391 log.close(); 1392 return log; 1393 } 1394 1395 /** 1396 * Parses a directory of WALs building up ProcedureState. 1397 * For testing parse and profiling. 1398 * @param args Include pointer to directory of WAL files for a store instance to parse & load. 1399 */ 1400 public static void main(String [] args) throws IOException { 1401 Configuration conf = HBaseConfiguration.create(); 1402 if (args == null || args.length != 1) { 1403 System.out.println("ERROR: Empty arguments list; pass path to MASTERPROCWALS_DIR."); 1404 System.out.println("Usage: WALProcedureStore MASTERPROCWALS_DIR"); 1405 System.exit(-1); 1406 } 1407 WALProcedureStore store = new WALProcedureStore(conf, new Path(args[0]), null, 1408 new WALProcedureStore.LeaseRecovery() { 1409 @Override 1410 public void recoverFileLease(FileSystem fs, Path path) throws IOException { 1411 // no-op 1412 } 1413 }); 1414 try { 1415 store.start(16); 1416 ProcedureExecutor<?> pe = new ProcedureExecutor<>(conf, new Object()/*Pass anything*/, store); 1417 pe.init(1, true); 1418 } finally { 1419 store.stop(true); 1420 } 1421 } 1422}