|
@@ -23,15 +23,17 @@ import java.util.Queue;
|
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
|
import java.util.concurrent.atomic.AtomicLong;
|
|
import java.util.concurrent.atomic.AtomicLong;
|
|
|
|
|
|
|
|
+import com.google.common.annotations.VisibleForTesting;
|
|
|
|
+import org.slf4j.Logger;
|
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
|
+
|
|
import org.apache.hadoop.ozone.om.OMMetadataManager;
|
|
import org.apache.hadoop.ozone.om.OMMetadataManager;
|
|
import org.apache.hadoop.ozone.om.ratis.helpers.DoubleBufferEntry;
|
|
import org.apache.hadoop.ozone.om.ratis.helpers.DoubleBufferEntry;
|
|
|
|
+import org.apache.hadoop.ozone.om.ratis.metrics.OzoneManagerDoubleBufferMetrics;
|
|
import org.apache.hadoop.ozone.om.response.OMClientResponse;
|
|
import org.apache.hadoop.ozone.om.response.OMClientResponse;
|
|
import org.apache.hadoop.util.Daemon;
|
|
import org.apache.hadoop.util.Daemon;
|
|
import org.apache.hadoop.utils.db.BatchOperation;
|
|
import org.apache.hadoop.utils.db.BatchOperation;
|
|
-
|
|
|
|
import org.apache.ratis.util.ExitUtils;
|
|
import org.apache.ratis.util.ExitUtils;
|
|
-import org.slf4j.Logger;
|
|
|
|
-import org.slf4j.LoggerFactory;
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
* This class implements DoubleBuffer implementation of OMClientResponse's. In
|
|
* This class implements DoubleBuffer implementation of OMClientResponse's. In
|
|
@@ -63,6 +65,8 @@ public class OzoneManagerDoubleBuffer {
|
|
private final AtomicLong flushedTransactionCount = new AtomicLong(0);
|
|
private final AtomicLong flushedTransactionCount = new AtomicLong(0);
|
|
private final AtomicLong flushIterations = new AtomicLong(0);
|
|
private final AtomicLong flushIterations = new AtomicLong(0);
|
|
private volatile boolean isRunning;
|
|
private volatile boolean isRunning;
|
|
|
|
+ private OzoneManagerDoubleBufferMetrics ozoneManagerDoubleBufferMetrics;
|
|
|
|
+ private long maxFlushedTransactionsInOneIteration;
|
|
|
|
|
|
private final OzoneManagerRatisSnapshot ozoneManagerRatisSnapShot;
|
|
private final OzoneManagerRatisSnapshot ozoneManagerRatisSnapShot;
|
|
|
|
|
|
@@ -71,8 +75,9 @@ public class OzoneManagerDoubleBuffer {
|
|
this.currentBuffer = new ConcurrentLinkedQueue<>();
|
|
this.currentBuffer = new ConcurrentLinkedQueue<>();
|
|
this.readyBuffer = new ConcurrentLinkedQueue<>();
|
|
this.readyBuffer = new ConcurrentLinkedQueue<>();
|
|
this.omMetadataManager = omMetadataManager;
|
|
this.omMetadataManager = omMetadataManager;
|
|
-
|
|
|
|
this.ozoneManagerRatisSnapShot = ozoneManagerRatisSnapShot;
|
|
this.ozoneManagerRatisSnapShot = ozoneManagerRatisSnapShot;
|
|
|
|
+ this.ozoneManagerDoubleBufferMetrics =
|
|
|
|
+ OzoneManagerDoubleBufferMetrics.create();
|
|
|
|
|
|
isRunning = true;
|
|
isRunning = true;
|
|
// Daemon thread which runs in back ground and flushes transactions to DB.
|
|
// Daemon thread which runs in back ground and flushes transactions to DB.
|
|
@@ -80,7 +85,6 @@ public class OzoneManagerDoubleBuffer {
|
|
daemon.setName("OMDoubleBufferFlushThread");
|
|
daemon.setName("OMDoubleBufferFlushThread");
|
|
daemon.start();
|
|
daemon.start();
|
|
|
|
|
|
-
|
|
|
|
}
|
|
}
|
|
|
|
|
|
/**
|
|
/**
|
|
@@ -120,6 +124,7 @@ public class OzoneManagerDoubleBuffer {
|
|
.max(Long::compareTo).get();
|
|
.max(Long::compareTo).get();
|
|
|
|
|
|
readyBuffer.clear();
|
|
readyBuffer.clear();
|
|
|
|
+
|
|
// cleanup cache.
|
|
// cleanup cache.
|
|
cleanupCache(lastRatisTransactionIndex);
|
|
cleanupCache(lastRatisTransactionIndex);
|
|
|
|
|
|
@@ -129,6 +134,9 @@ public class OzoneManagerDoubleBuffer {
|
|
// update the last updated index in OzoneManagerStateMachine.
|
|
// update the last updated index in OzoneManagerStateMachine.
|
|
ozoneManagerRatisSnapShot.updateLastAppliedIndex(
|
|
ozoneManagerRatisSnapShot.updateLastAppliedIndex(
|
|
lastRatisTransactionIndex);
|
|
lastRatisTransactionIndex);
|
|
|
|
+
|
|
|
|
+ // set metrics.
|
|
|
|
+ updateMetrics(flushedTransactionsSize);
|
|
}
|
|
}
|
|
} catch (InterruptedException ex) {
|
|
} catch (InterruptedException ex) {
|
|
Thread.currentThread().interrupt();
|
|
Thread.currentThread().interrupt();
|
|
@@ -162,6 +170,23 @@ public class OzoneManagerDoubleBuffer {
|
|
omMetadataManager.getUserTable().cleanupCache(lastRatisTransactionIndex);
|
|
omMetadataManager.getUserTable().cleanupCache(lastRatisTransactionIndex);
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
+ * Update OzoneManagerDoubleBuffer metrics values.
|
|
|
|
+ * @param flushedTransactionsSize
|
|
|
|
+ */
|
|
|
|
+ private void updateMetrics(
|
|
|
|
+ long flushedTransactionsSize) {
|
|
|
|
+ ozoneManagerDoubleBufferMetrics.incrTotalNumOfFlushOperations();
|
|
|
|
+ ozoneManagerDoubleBufferMetrics.incrTotalSizeOfFlushedTransactions(
|
|
|
|
+ flushedTransactionsSize);
|
|
|
|
+ if (maxFlushedTransactionsInOneIteration < flushedTransactionsSize) {
|
|
|
|
+ maxFlushedTransactionsInOneIteration = flushedTransactionsSize;
|
|
|
|
+ ozoneManagerDoubleBufferMetrics
|
|
|
|
+ .setMaxNumberOfTransactionsFlushedInOneIteration(
|
|
|
|
+ flushedTransactionsSize);
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+
|
|
/**
|
|
/**
|
|
* Stop OM DoubleBuffer flush thread.
|
|
* Stop OM DoubleBuffer flush thread.
|
|
*/
|
|
*/
|
|
@@ -170,6 +195,9 @@ public class OzoneManagerDoubleBuffer {
|
|
LOG.info("Stopping OMDoubleBuffer flush thread");
|
|
LOG.info("Stopping OMDoubleBuffer flush thread");
|
|
isRunning = false;
|
|
isRunning = false;
|
|
daemon.interrupt();
|
|
daemon.interrupt();
|
|
|
|
+
|
|
|
|
+ // stop metrics.
|
|
|
|
+ ozoneManagerDoubleBufferMetrics.unRegister();
|
|
} else {
|
|
} else {
|
|
LOG.info("OMDoubleBuffer flush thread is not running.");
|
|
LOG.info("OMDoubleBuffer flush thread is not running.");
|
|
}
|
|
}
|
|
@@ -236,5 +264,10 @@ public class OzoneManagerDoubleBuffer {
|
|
readyBuffer = temp;
|
|
readyBuffer = temp;
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+ @VisibleForTesting
|
|
|
|
+ public OzoneManagerDoubleBufferMetrics getOzoneManagerDoubleBufferMetrics() {
|
|
|
|
+ return ozoneManagerDoubleBufferMetrics;
|
|
|
|
+ }
|
|
|
|
+
|
|
}
|
|
}
|
|
|
|
|