diff --git a/beedb-gateway/pom.xml b/beedb-gateway/pom.xml index 770d0e2..c85a896 100644 --- a/beedb-gateway/pom.xml +++ b/beedb-gateway/pom.xml @@ -54,22 +54,25 @@ org.codehaus.mojo exec-maven-plugin 3.0.0 + + com.memcache.gateway.GatewayApplication + + + + org.springframework.boot + spring-boot-maven-plugin + ${spring.boot.version} - java + repackage - com.memcache.gateway.client.BeedbClient + com.memcache.gateway.GatewayApplication - - org.springframework.boot - spring-boot-maven-plugin - ${spring.boot.version} - diff --git a/beedb-gateway/src/main/java/com/memcache/gateway/api/ApiExceptionHandler.java b/beedb-gateway/src/main/java/com/memcache/gateway/api/ApiExceptionHandler.java index 92e77f8..2890208 100644 --- a/beedb-gateway/src/main/java/com/memcache/gateway/api/ApiExceptionHandler.java +++ b/beedb-gateway/src/main/java/com/memcache/gateway/api/ApiExceptionHandler.java @@ -6,7 +6,7 @@ import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.RestControllerAdvice; - +import java.io.IOException; import com.memcache.gateway.client.beedbexception.BeedbException; import com.memcache.gateway.client.beedbexception.BeedbTimeoutException; import com.memcache.gateway.client.beedbexception.NoLeaderException; @@ -39,4 +39,9 @@ else if(ex instanceof BeedbTimeoutException){ LOGGER.error("Exception occurred in the controller", ex); return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(ex.getMessage()); } + + @ExceptionHandler (IOException.class) + public void handleClientDisconnect(IOException ex){ + LOGGER.debug("client disconnected: {}", ex.getMessage()); + } } diff --git a/beedb-gateway/src/main/java/com/memcache/gateway/api/KvController.java b/beedb-gateway/src/main/java/com/memcache/gateway/api/KvController.java index 9a3f2e5..47bd89e 100644 --- a/beedb-gateway/src/main/java/com/memcache/gateway/api/KvController.java +++ b/beedb-gateway/src/main/java/com/memcache/gateway/api/KvController.java @@ -4,6 +4,7 @@ import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.List; import org.slf4j.Logger; @@ -19,6 +20,7 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import com.memcache.gateway.client.BeedbClient; @@ -94,8 +96,18 @@ private ResponseEntity reservedKey(String key) { * keys the current browser happened to write itself. */ @GetMapping - public ResponseEntity> keys(){ - return ResponseEntity.ok(keyRegistry.list()); + public ResponseEntity> keys( + @RequestParam(value = "limit", defaultValue = "100") int limit, + @RequestHeader(value = SESSION_HEADER, required = false) String sessionId){ + List all = keyRegistry.list(); // newest first + int n = Math.max(0, Math.min(limit, 200)); + List out = new ArrayList<>(all.subList(0, Math.min(n, all.size()))); + if(sessionId != null){ + for(KeyRegistry.Entry e : all){ + if(sessionId.equals(e.sessionId()) && !out.contains(e)) out.add(e); + } + } + return ResponseEntity.ok(out); } @GetMapping ("/{key}") diff --git a/beedb-gateway/src/main/java/com/memcache/gateway/api/QuotaController.java b/beedb-gateway/src/main/java/com/memcache/gateway/api/QuotaController.java new file mode 100644 index 0000000..01968c3 --- /dev/null +++ b/beedb-gateway/src/main/java/com/memcache/gateway/api/QuotaController.java @@ -0,0 +1,33 @@ +package com.memcache.gateway.api; + +import java.util.Map; + +import jakarta.servlet.http.HttpServletRequest; + +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import com.memcache.gateway.kv.WriteRateLimiter; + +/** + * How many writes this client has left this hour, so the page shows the server's real count + * instead of guessing + */ +@RestController +@RequestMapping("/api") +public class QuotaController { + + private final WriteRateLimiter rateLimiter; + + public QuotaController(WriteRateLimiter rateLimiter) { + this.rateLimiter = rateLimiter; + } + + @GetMapping("/quota") + public Map quota(HttpServletRequest request) { + String client = request.getRemoteAddr(); + return Map.of("remaining", rateLimiter.remaining(client), + "limit", rateLimiter.maxWritesPerHour()); + } +} diff --git a/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbClient.java b/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbClient.java index f423de6..508ffaf 100644 --- a/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbClient.java +++ b/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbClient.java @@ -68,6 +68,27 @@ public String delete(String key){ } + /* + * Nodes whose last stats request failed. The status poll asks every node once a second, + * so a node that is down for a 15-second chaos round used to log the same WARN 15 times. + * Logging only the two transitions -- gone, then back -- says the same thing. + */ + private final java.util.Set unreachableNodes = java.util.concurrent.ConcurrentHashMap.newKeySet(); + + private void statsFailed(String operation, String address, Exception e){ + if(unreachableNodes.add(address)){ + logFailure(operation, address, e); + }else{ + LOGGER.debug("{} still failing on node {} ({})", operation, address, e.toString()); + } + } + + private void statsAnswered(String address){ + if(unreachableNodes.remove(address)){ + LOGGER.info("node {} is answering again", address); + } + } + public NodeStatus getStats(String nodeId, String address){ String response = null; BeedbConnection connection = null; @@ -78,9 +99,10 @@ public NodeStatus getStats(String nodeId, String address){ if(response == null) { return NodeStatus.unreachable(nodeId, address); } - else return createNodeStatus(response , nodeId , address); + statsAnswered(address); + return createNodeStatus(response , nodeId , address); }catch(Exception e){ - logFailure("stats", address, e); + statsFailed("stats", address, e); pool.closeConnection(connection); return NodeStatus.unreachable(nodeId, address); } @@ -135,6 +157,13 @@ private void logFailure(String operation, String address, Exception e){ LOGGER.warn("{} failed: node {} is unreachable ({})", operation, address, e.getMessage()); }else if(e instanceof java.net.SocketTimeoutException){ LOGGER.warn("{} failed: node {} timed out ({})", operation, address, e.getMessage()); + }else if(e instanceof NoLeaderException){ + // Expected for the second or two an election takes -- every chaos round causes one. + LOGGER.warn("{} failed: no leader right now ({})", operation, e.getMessage()); + }else if(e instanceof java.util.concurrent.TimeoutException){ + // The status poll gives each node a fixed wait and records a slow one as unreachable + // for that tick + LOGGER.warn("{}: node {} did not answer in time", operation, address); }else{ LOGGER.error("{} failed unexpectedly against node {}", operation, address, e); } @@ -198,7 +227,7 @@ public List getClusterStatus(){ NodeStatus result = future.get(1000, TimeUnit.MILLISECONDS); nodeStatuses.add(result); }catch(Exception e){ - logFailure("exception in getclusterstatus ", entries.get(i).getValue(), e); + statsFailed("stats poll", entries.get(i).getValue(), e); nodeStatuses.add(NodeStatus.unreachable(entries.get(i).getKey() , entries.get(i).getValue())); } } diff --git a/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbConnection.java b/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbConnection.java index 48a8316..d85f75f 100644 --- a/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbConnection.java +++ b/beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbConnection.java @@ -3,6 +3,7 @@ import java.io.BufferedInputStream; import java.io.BufferedOutputStream; import java.io.IOException; +import java.net.SocketException; import java.net.Socket; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -113,18 +114,18 @@ public void close(){ LOGGER.error("something is wrong ",e); } } - - public String readLine(BufferedInputStream bufferedInputStream) { + + public String readLine(BufferedInputStream bufferedInputStream) throws IOException { ByteArrayOutputStream baos = new ByteArrayOutputStream(); while(true){ - try{ - int c = bufferedInputStream.read(); - if(c == -1 || c == '\n')break; - baos.write(c); - }catch(Exception e){ - LOGGER.error("readLine threw exception ", e); - break; + int c = bufferedInputStream.read(); + if(c == -1){ + // the node closed the connection mid-reply: it died or restarted. Same outcome as + // a reset , this socket is finished and must not go back into the pool. + throw new SocketException("connection closed by node before the reply ended"); } + if(c == '\n')break; + baos.write(c); } byte[] buffer = baos.toByteArray(); if(buffer.length > 0 && buffer[buffer.length-1] == '\r')return new String(buffer, 0, buffer.length-1, StandardCharsets.UTF_8); diff --git a/beedb-gateway/src/main/java/com/memcache/gateway/cluster/ClusterStatusService.java b/beedb-gateway/src/main/java/com/memcache/gateway/cluster/ClusterStatusService.java index 5da60f8..9819570 100644 --- a/beedb-gateway/src/main/java/com/memcache/gateway/cluster/ClusterStatusService.java +++ b/beedb-gateway/src/main/java/com/memcache/gateway/cluster/ClusterStatusService.java @@ -54,6 +54,8 @@ public SseEmitter registerEmitter(){ emitters.add(emitter); try { emitter.send(clusterView); + } catch (IOException e) { + clientGone(emitter, e); } catch (Exception e) { logFailure(e, "failed to send cluster status"); } @@ -66,12 +68,19 @@ public void sendClusterStatus(){ try { emitter.send(clusterView); LOGGER.debug("sent cluster status"); + } catch (IOException e) { + clientGone(emitter, e); } catch (Exception e) { emitters.remove(emitter); logFailure(e , "failed to send cluster status"); } }); } + + private void clientGone(SseEmitter emitter, IOException e){ + emitters.remove(emitter); + LOGGER.debug("stream client disconnected ({})", e.getMessage()); + } public void updateClusterStatus(){ try{ diff --git a/beedb-gateway/src/main/java/com/memcache/gateway/demo/DemoWriter.java b/beedb-gateway/src/main/java/com/memcache/gateway/demo/DemoWriter.java index 73890f9..20a9019 100644 --- a/beedb-gateway/src/main/java/com/memcache/gateway/demo/DemoWriter.java +++ b/beedb-gateway/src/main/java/com/memcache/gateway/demo/DemoWriter.java @@ -42,7 +42,7 @@ public DemoWriter(BeedbClient beedbClient, KeyRegistry keyRegistry){ this.keyRegistry = keyRegistry; } - @Scheduled (fixedDelay = 1000) + @Scheduled (fixedDelay = 2000) public void write(){ long n = attempts++; String key = KEY_PREFIX + (n % RING_SIZE); diff --git a/beedb-server/src/main/java/com/memcache/Server.java b/beedb-server/src/main/java/com/memcache/Server.java index 8f87c1e..1c5a805 100644 --- a/beedb-server/src/main/java/com/memcache/Server.java +++ b/beedb-server/src/main/java/com/memcache/Server.java @@ -6,7 +6,6 @@ */ package com.memcache; -import java.io.ByteArrayOutputStream; import java.io.FileInputStream; import java.io.IOException; import java.net.InetSocketAddress; @@ -55,6 +54,8 @@ public class Server implements AutoCloseable{ private RaftTransport raftTransport; private static final int INITIAL_BUFFER_SIZE = 4 * 1024; private static final int MAX_BUFFER_SIZE = 8 * 1024; + // memcached: an expiry above 30 days is an absolute Unix time, not seconds-from-now. + private static final long MAX_RELATIVE_EXPIRY_SECONDS = 30L * 24 * 60 * 60; public static void main(String[] args) throws IOException{ // read file name String fileName = args[0]; @@ -114,6 +115,19 @@ private static int intProperty(Properties properties, String key, int defaultVal return Integer.parseInt(raw.trim()); } + /** + * the client's memcached expiry, as an absolute deadline in epoch milliseconds. + * 0 never expires -> 0 + * 1 .. 30 days seconds from now -> now + seconds + * above 30 days an absolute Unix time -> that time, in milliseconds + * negative already expired -> a moment in the past + */ + static long absoluteExpiryMillis(long expiry, long nowMillis){ + if(expiry == 0) return 0; + if(expiry > MAX_RELATIVE_EXPIRY_SECONDS) return expiry * 1000L; + return nowMillis + expiry * 1000L; + } + public void start(ArrayList peers, String nodeId, int clientPort,int raftPort, String stateDir, String snapshotDir, String tmpDir, String walDir, int snapShotLimit, int snapShotThreshold, int minElectionTimeout, int maxElectionTimeout, int heartbeatInterval , int peerRetryBackoffInitialMs, int peerRetryBackoffMaxMs) throws IOException, InterruptedException{ // selector to notify about new connections @@ -380,6 +394,7 @@ else if(command.getType() == CommandType.GET){ return response.toProtocolString().getBytes(); } else{ + command.setExpiry(absoluteExpiryMillis(command.getExpiry(), System.currentTimeMillis())); LOGGER.info("command type is {}", command.getType()); LOGGER.info("command is {}", command); // we have to propose this command to raft node diff --git a/beedb-server/src/main/java/com/memcache/cache/Cache.java b/beedb-server/src/main/java/com/memcache/cache/Cache.java index 62c0177..ce56fd2 100644 --- a/beedb-server/src/main/java/com/memcache/cache/Cache.java +++ b/beedb-server/src/main/java/com/memcache/cache/Cache.java @@ -23,7 +23,7 @@ public CacheItem get(String key) { return null; } - public void put(String key, byte[] value, int flags, int expiry) { + public void put(String key, byte[] value, int flags, long expiry) { map.put(key, new CacheItem(key, value, flags, expiry)); } diff --git a/beedb-server/src/main/java/com/memcache/cache/CacheItem.java b/beedb-server/src/main/java/com/memcache/cache/CacheItem.java index d48f5e5..ab9581c 100644 --- a/beedb-server/src/main/java/com/memcache/cache/CacheItem.java +++ b/beedb-server/src/main/java/com/memcache/cache/CacheItem.java @@ -12,11 +12,14 @@ public class CacheItem { private boolean infiniteExpiry; private int flags; - public CacheItem(String key, byte[] value, int flags, int expiry) { + public CacheItem(String key, byte[] value, int flags, long expiry) { this.key = key; this.value = value; this.flags = flags; - this.expiresAt = (expiry == 0) ? -1 : System.currentTimeMillis() + (expiry * 1000L); + // expiry is already an absolute epoch-millisecond deadline, set once by the leader. + // Never read the clock here: applying an entry must give the same result on every node + // and on every replay. + this.expiresAt = (expiry == 0) ? -1 : expiry; this.infiniteExpiry = (expiry == 0); } diff --git a/beedb-server/src/main/java/com/memcache/command/Command.java b/beedb-server/src/main/java/com/memcache/command/Command.java index d6ecfbb..d288edd 100644 --- a/beedb-server/src/main/java/com/memcache/command/Command.java +++ b/beedb-server/src/main/java/com/memcache/command/Command.java @@ -7,11 +7,14 @@ public class Command { private CommandType type; private String key; private int flags; - private int expiry; + // From the client: memcached seconds (from now, or an absolute Unix time above 30 days). + // After the leader converts it (Server.absoluteExpiryMillis), and in every log entry: an + // absolute deadline in epoch milliseconds, or 0 for "never". + private long expiry; private int byteLength; private byte[] value; - public Command(CommandType type, String key, int flags, int expiry, int byteLength) { + public Command(CommandType type, String key, int flags, long expiry, int byteLength) { this.type = type; this.key = key; this.flags = flags; @@ -31,7 +34,7 @@ public int getFlags() { return flags; } - public int getExpiry() { + public long getExpiry() { return expiry; } @@ -55,7 +58,7 @@ public void setFlags(int flags) { this.flags = flags; } - public void setExpiry(int expiry) { + public void setExpiry(long expiry) { this.expiry = expiry; } @@ -69,7 +72,7 @@ public static Command deserialize(String s) { CommandType type = CommandType.valueOf(parts[0]); String key = parts[1]; int flags = Integer.parseInt(parts[2]); - int expiry = Integer.parseInt(parts[3]); + long expiry = Long.parseLong(parts[3]); int byteLength = Integer.parseInt(parts[4]); Command cmd = new Command(type, key, flags, expiry, byteLength); if (parts.length == 6 && !parts[5].isEmpty()) { diff --git a/beedb-server/src/main/java/com/memcache/raft/FailStop.java b/beedb-server/src/main/java/com/memcache/raft/FailStop.java new file mode 100644 index 0000000..3440269 --- /dev/null +++ b/beedb-server/src/main/java/com/memcache/raft/FailStop.java @@ -0,0 +1,25 @@ +package com.memcache.raft; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import ch.qos.logback.classic.LoggerContext; + +public final class FailStop { + + private FailStop() { } + + public static void halt(Logger logger, String what, Throwable cause) { + try { + logger.error("FAIL-STOP: {} -- halting this node", what, cause); + System.err.println("FAIL-STOP: " + what + ": " + cause); + cause.printStackTrace(); + if (LoggerFactory.getILoggerFactory() instanceof LoggerContext context) { + // drains the AsyncAppender queue, which halt() would drop + context.stop(); + } + } finally { + Runtime.getRuntime().halt(1); + } + } +} diff --git a/beedb-server/src/main/java/com/memcache/raft/RaftNode.java b/beedb-server/src/main/java/com/memcache/raft/RaftNode.java index 21bbb3c..8bf9d95 100644 --- a/beedb-server/src/main/java/com/memcache/raft/RaftNode.java +++ b/beedb-server/src/main/java/com/memcache/raft/RaftNode.java @@ -6,7 +6,6 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; @@ -185,11 +184,11 @@ public synchronized void start(){ // and still the problem of deadlock is not solved // so we are using synchronized blocks whenever required private void startElection(){ - LOGGER.info("startElection for term {} by {}", currentTerm, nodeId); synchronized(this){ if(isLeader()) return; leaderId = null; transitionToCandidate(); + LOGGER.info("startElection for term {} by {}", currentTerm, nodeId); resetElectionTimer(); } // below method is synchronized internally and so we dont need to add this in synchronized block @@ -933,8 +932,20 @@ public CompletableFuture propose(String command){ return future; } - // backgroud loop which applied commited entires to our cache + /** + * backgroud loop which applied commited entires to our cache + * a dead apply loop means this node can never make progress again, so it fail-stops instead. + */ private void applyCommitedEntries(){ + try{ + applyCommitedEntriesLoop(); + }catch(Throwable t){ + if(stopping) return; // stop() interrupts this thread; that is not a failure + FailStop.halt(LOGGER, "apply loop died on " + nodeId + " at lastApplied=" + lastApplied, t); + } + } + + private void applyCommitedEntriesLoop(){ MDC.put("nodeId", nodeId); while(!Thread.currentThread().isInterrupted()){ @@ -1004,8 +1015,11 @@ private void applyCommitedEntries(){ Response response = null; LOGGER.debug("term {} node id {} applying entry {} at index {} ", currentTerm, nodeId, entry, lastApplied); if(!entry.isNoOp()) { - Command command = Command.deserialize(entry.getCommand()); + // deserialize sits inside the try too. Every node parses the same bytes and fails + // the same way, so each skips the same entry and they stay identical. Outside the + // try, one entry it could not parse killed this loop on every node at once. try{ + Command command = Command.deserialize(entry.getCommand()); response = CommandProcessor.process(command, cache); } catch (Exception e) { LOGGER.error("failed to apply command at index {} : {} - {}", diff --git a/beedb-server/src/main/java/com/memcache/raft/wal/WalService.java b/beedb-server/src/main/java/com/memcache/raft/wal/WalService.java index a8e93d1..e28b3b5 100644 --- a/beedb-server/src/main/java/com/memcache/raft/wal/WalService.java +++ b/beedb-server/src/main/java/com/memcache/raft/wal/WalService.java @@ -1,5 +1,7 @@ package com.memcache.raft.wal; import com.memcache.raft.LogEntry; + + import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; @@ -26,6 +28,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.memcache.raft.FailStop; + import com.fasterxml.jackson.databind.ObjectMapper; public class WalService { @@ -67,11 +71,12 @@ public void start(){ try{ writeToFile(); }catch(Exception e){ - // LOGGER.error("something is wrong {}",e); - LOGGER.error("file in which write failed is ", walFilePath); - LOGGER.error("error while writing to file, please check stack trace ", e); - if(!stopping) shutdown(); - break; + // shutdown() interrupts this thread and closes the channel; a write or fsync caught + // mid-flight then throws ClosedByInterruptException or AsynchronousCloseException. + if(stopping) break; + // real write/fsync failure. Exit instead of lingering: a leader whose WAL cannot + // persist keeps heartbeating + FailStop.halt(LOGGER, "WAL write failed on " + walFilePath, e); } } }