You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by fp...@apache.org on 2014/08/02 20:02:07 UTC
svn commit: r1615338 [2/2] - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-server/conf/
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/
bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/
bookkeeper-server/src/main/java/org/apache...
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java?rev=1615338&r1=1615337&r2=1615338&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java Sat Aug 2 18:02:06 2014
@@ -23,7 +23,6 @@ package org.apache.bookkeeper.proto;
import java.io.IOException;
import java.nio.ByteBuffer;
-import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperProtocol.AddRequest;
@@ -31,6 +30,7 @@ import org.apache.bookkeeper.proto.Bookk
import org.apache.bookkeeper.proto.BookkeeperProtocol.Request;
import org.apache.bookkeeper.proto.BookkeeperProtocol.Response;
import org.apache.bookkeeper.proto.BookkeeperProtocol.StatusCode;
+import org.apache.bookkeeper.util.MathUtils;
import org.jboss.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,12 +38,14 @@ import org.slf4j.LoggerFactory;
class WriteEntryProcessorV3 extends PacketProcessorBaseV3 implements Runnable {
private final static Logger logger = LoggerFactory.getLogger(WriteEntryProcessorV3.class);
- public WriteEntryProcessorV3(Request request, Channel channel, Bookie bookie) {
- super(request, channel, bookie);
+ public WriteEntryProcessorV3(Request request, Channel channel,
+ BookieRequestProcessor requestProcessor) {
+ super(request, channel, requestProcessor);
}
// Returns null if there is no exception thrown
private AddResponse getAddResponse() {
+ final long startTimeNanos = MathUtils.nowInNano();
AddRequest addRequest = request.getAddRequest();
long ledgerId = addRequest.getLedgerId();
long entryId = addRequest.getEntryId();
@@ -57,7 +59,7 @@ class WriteEntryProcessorV3 extends Pack
return addResponse.build();
}
- if (bookie.isReadOnly()) {
+ if (requestProcessor.bookie.isReadOnly()) {
logger.warn("BookieServer is running as readonly mode, so rejecting the request from the client!");
addResponse.setStatus(StatusCode.EREADONLY);
return addResponse.build();
@@ -67,6 +69,12 @@ class WriteEntryProcessorV3 extends Pack
@Override
public void writeComplete(int rc, long ledgerId, long entryId,
BookieSocketAddress addr, Object ctx) {
+ if (BookieProtocol.EOK == rc) {
+ requestProcessor.addEntryStats.registerSuccessfulEvent(MathUtils.elapsedMSec(startTimeNanos));
+ } else {
+ requestProcessor.addEntryStats.registerFailedEvent(MathUtils.elapsedMSec(startTimeNanos));
+ }
+
Channel conn = (Channel) ctx;
StatusCode status;
switch (rc) {
@@ -86,7 +94,7 @@ class WriteEntryProcessorV3 extends Pack
.setStatus(addResponse.getStatus())
.setAddResponse(addResponse);
Response resp = response.build();
- conn.write(resp);
+ sendResponse(status, resp, requestProcessor.addRequestStats);
}
};
StatusCode status = null;
@@ -94,9 +102,9 @@ class WriteEntryProcessorV3 extends Pack
ByteBuffer entryToAdd = addRequest.getBody().asReadOnlyByteBuffer();
try {
if (addRequest.hasFlag() && addRequest.getFlag().equals(AddRequest.Flag.RECOVERY_ADD)) {
- bookie.recoveryAddEntry(entryToAdd, wcb, channel, masterKey);
+ requestProcessor.bookie.recoveryAddEntry(entryToAdd, wcb, channel, masterKey);
} else {
- bookie.addEntry(entryToAdd, wcb, channel, masterKey);
+ requestProcessor.bookie.addEntry(entryToAdd, wcb, channel, masterKey);
}
status = StatusCode.EOK;
} catch (IOException e) {
@@ -137,7 +145,8 @@ class WriteEntryProcessorV3 extends Pack
.setStatus(addResponse.getStatus())
.setAddResponse(addResponse);
Response resp = response.build();
- channel.write(resp);
+ sendResponse(addResponse.getStatus(), resp,
+ requestProcessor.addRequestStats);
}
}
}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/DiskChecker.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/DiskChecker.java?rev=1615338&r1=1615337&r2=1615338&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/DiskChecker.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/DiskChecker.java Sat Aug 2 18:02:06 2014
@@ -22,11 +22,16 @@ import java.io.File;
import java.io.IOException;
import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Class that provides utility functions for checking disk problems
*/
public class DiskChecker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DiskChecker.class);
+
private float diskUsageThreshold;
private float diskUsageWarnThreshold;
@@ -47,16 +52,30 @@ public class DiskChecker {
public static class DiskOutOfSpaceException extends DiskException {
private static final long serialVersionUID = 160898797915906860L;
- public DiskOutOfSpaceException(String msg) {
+ private final float usage;
+
+ public DiskOutOfSpaceException(String msg, float usage) {
super(msg);
+ this.usage = usage;
+ }
+
+ public float getUsage() {
+ return usage;
}
}
public static class DiskWarnThresholdException extends DiskException {
private static final long serialVersionUID = -1629284987500841657L;
- public DiskWarnThresholdException(String msg) {
+ private final float usage;
+
+ public DiskWarnThresholdException(String msg, float usage) {
super(msg);
+ this.usage = usage;
+ }
+
+ public float getUsage() {
+ return usage;
}
}
@@ -106,9 +125,9 @@ public class DiskChecker {
* less than threshhold.
*/
@VisibleForTesting
- void checkDiskFull(File dir) throws DiskOutOfSpaceException, DiskWarnThresholdException {
+ float checkDiskFull(File dir) throws DiskOutOfSpaceException, DiskWarnThresholdException {
if (null == dir) {
- return;
+ return 0f;
}
if (dir.exists()) {
long usableSpace = dir.getUsableSpace();
@@ -116,16 +135,22 @@ public class DiskChecker {
float free = (float) usableSpace / (float) totalSpace;
float used = 1f - free;
if (used > diskUsageThreshold) {
+ LOG.error("Space left on device {} : {}, Used space fraction: {} < threshold {}.",
+ new Object[] { dir, usableSpace, used, diskUsageThreshold });
throw new DiskOutOfSpaceException("Space left on device "
- + usableSpace + " Used space fraction:" + used + " < threshhold " + diskUsageThreshold);
+ + usableSpace + " Used space fraction:" + used + " < threshold " + diskUsageThreshold, used);
}
// Warn should be triggered only if disk usage threshold doesn't trigger first.
if (used > diskUsageWarnThreshold) {
+ LOG.warn("Space left on device {} : {}, Used space fraction: {} < WarnThreshold {}.",
+ new Object[] { dir, usableSpace, used, diskUsageThreshold });
throw new DiskWarnThresholdException("Space left on device:"
- + usableSpace + " Used space fraction:" + used +" < WarnThreshold:" + diskUsageWarnThreshold);
+ + usableSpace + " Used space fraction:" + used +" < WarnThreshold:" + diskUsageWarnThreshold,
+ used);
}
+ return used;
} else {
- checkDiskFull(dir.getParentFile());
+ return checkDiskFull(dir.getParentFile());
}
}
@@ -141,9 +166,9 @@ public class DiskChecker {
* @throws DiskOutOfSpaceException
* If disk is full or having less space than threshhold
*/
- public void checkDir(File dir) throws DiskErrorException,
+ public float checkDir(File dir) throws DiskErrorException,
DiskOutOfSpaceException, DiskWarnThresholdException {
- checkDiskFull(dir);
+ float usage = checkDiskFull(dir);
if (!mkdirsWithExistsCheck(dir))
throw new DiskErrorException("can not create directory: "
+ dir.toString());
@@ -158,6 +183,7 @@ public class DiskChecker {
if (!dir.canWrite())
throw new DiskErrorException("directory is not writable: "
+ dir.toString());
+ return usage;
}
/**