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