chaos: wired chaos service, demo-writer, setsid restore and updates o…

beedb-gateway/src/main/java/com/memcache/gateway/api/ChaosController.java

@@ -0,0 +1,41 @@
+package com.memcache.gateway.api;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.memcache.gateway.chaos.ChaosKillState;
+import com.memcache.gateway.chaos.ChaosService;
+import com.memcache.gateway.cluster.ClusterSnapshot;
+import com.memcache.gateway.cluster.ClusterStatusService;
+
+@RestController 
+@RequestMapping ("/api/chaos")
+public class ChaosController {
+
+    private final ClusterStatusService clusterStatusService;
+    private final ChaosService chaosService;
+    public ChaosController(ClusterStatusService clusterStatusService, ChaosService chaosService){
+        this.clusterStatusService = clusterStatusService;
+        this.chaosService = chaosService;
+    }
+
+    @PostMapping ("/kill")
+    public ResponseEntity<String> kill(){
+        ClusterSnapshot clusterSnapshot = clusterStatusService.getClusterSnapshot();
+        ChaosKillState state = chaosService.requestKill(clusterSnapshot.leaderId(), clusterSnapshot);
+        if(state == ChaosKillState.SUCCESS){
+            return ResponseEntity.status(HttpStatus.ACCEPTED).body(state.toString());
+        } else if (state == ChaosKillState.UNDER_COOLDOWN) {
+            long secondsUntilKill = chaosService.secondsUntilEligible();
+            return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
+                                  .header("Retry-After", String.valueOf(secondsUntilKill))
+                                  .body(state.toString());
+        } else {
+            return ResponseEntity.status(HttpStatus.CONFLICT).body(state.toString());
+        }
+
+    }
+}

beedb-gateway/src/main/java/com/memcache/gateway/api/ClusterController.java

@@ -1,41 +1,43 @@
 package com.memcache.gateway.api;
 
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
-import com.memcache.gateway.cluster.ClusterSnapshot;
 import com.memcache.gateway.cluster.ClusterStatusService;
+import com.memcache.gateway.cluster.ClusterView;
 
 @RestController
 @RequestMapping("/api")
 public class ClusterController {
 
     private final ClusterStatusService clusterStatusService;
+    private static final Logger LOGGER = LoggerFactory.getLogger(ClusterController.class);
 
     public ClusterController(ClusterStatusService clusterStatusService){
         this.clusterStatusService = clusterStatusService;
     }
 
     @GetMapping ("/cluster")
-    public ResponseEntity<ClusterSnapshot> getClusterStatus(){
-        // System.out.println("getting cluster status");
-        ClusterSnapshot clusterSnapshot = clusterStatusService.getClusterSnapshot();
-        // System.out.println("cluster snapshot is " + clusterSnapshot);
-        if(clusterSnapshot == null){
-            return ResponseEntity.notFound().build();
-        }
-        return ResponseEntity.ok(clusterSnapshot);
+    public ResponseEntity<ClusterView> getClusterStatus(){
+        ClusterView clusterView = clusterStatusService.getClusterView();
+        return ResponseEntity.ok(clusterView);
     }
 
     @GetMapping (value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
     public SseEmitter getClusterStatusEvents(){
-        System.out.println("getting cluster status events");
-        SseEmitter emitter = clusterStatusService.registerEmitter();
-        return emitter;
+        try{
+            SseEmitter emitter = clusterStatusService.registerEmitter();
+            return emitter;
+        }catch(Exception e){
+            LOGGER.error("failed to register emitter", e);
+            return null;
+        }
     }
 }

beedb-gateway/src/main/java/com/memcache/gateway/api/KvController.java

@@ -2,7 +2,6 @@
 
 import jakarta.servlet.http.HttpServletRequest;
 
-import java.net.http.HttpHeaders;
 import java.nio.charset.StandardCharsets;
 
 import java.util.List;
@@ -24,6 +23,7 @@
 
 import com.memcache.gateway.client.BeedbClient;
 import com.memcache.gateway.client.BeedbConfig;
+import com.memcache.gateway.demo.DemoWriter;
 import com.memcache.gateway.kv.KeyRegistry;
 import com.memcache.gateway.kv.WriteRateLimiter;
 
@@ -73,6 +73,19 @@ private ResponseEntity<String> rateLimited(HttpServletRequest request) {
                       + " writes per hour. This is a shared demo.");
     }
 
+    /**
+     * 403 for keys the demo writer owns,checked before
+     * the rate limit, so a refused request does not also cost the visitor a write.
+     */
+    private ResponseEntity<String> reservedKey(String key) {
+        if (!key.startsWith(DemoWriter.KEY_PREFIX)) {
+            return null;
+        }
+        LOGGER.debug("rejected write to reserved demo key {}", key);
+        return ResponseEntity.status(HttpStatus.FORBIDDEN)
+                .body("key '" + key + "' is reserved for the demo writer");
+    }
+
     /**
      * Every key the gateway has written and who owns it.
      *
@@ -101,18 +114,16 @@ public ResponseEntity<String> set(@PathVariable("key") String key,
         // Size check FIRST, before the rate limit: an oversized body is rejected
         // outright, so it should not also cost the client one of their ten writes.
         //
-        // This is the one check that is load-bearing for safety rather than fairness.
-        // The server reads a command into a single 4096-byte buffer and indexes the
-        // value out of it; a larger value throws BufferUnderflowException, and the
-        // ClosedChannelException that follows has been seen killing the node. Until
-        // the server read path is fixed, this is what stops a stranger with curl
-        // taking the cluster down.
+        // The server accepts up to 8192 bytes per command, line and value together, so
+        // its real ceiling depends on how long the key is
         int valueBytes = value == null ? 0 : value.getBytes(StandardCharsets.UTF_8).length;
         if(valueBytes > maxValueBytes){
             LOGGER.warn("rejected {}-byte value for key {} (max {})", valueBytes, key, maxValueBytes);
             return ResponseEntity.status(HttpStatus.PAYLOAD_TOO_LARGE)
                     .body("value is " + valueBytes + " bytes; the limit is " + maxValueBytes);
         }
+        ResponseEntity<String> reserved = reservedKey(key);
+        if(reserved != null) return reserved;
         ResponseEntity<String> limited = rateLimited(request);
         if(limited != null) return limited;
         // 403 rather than 404: the key exists, you simply are not its owner. The UI
@@ -136,6 +147,8 @@ public ResponseEntity<String> set(@PathVariable("key") String key,
     public ResponseEntity<String> delete(@PathVariable("key") String key,
                                          @RequestHeader(value = SESSION_HEADER, required = false) String sessionId,
                                          HttpServletRequest request){
+        ResponseEntity<String> reserved = reservedKey(key);
+        if(reserved != null) return reserved;
         ResponseEntity<String> limited = rateLimited(request);
         if(limited != null) return limited;
         if(!keyRegistry.mayWrite(key, sessionId)){

beedb-gateway/src/main/java/com/memcache/gateway/chaos/ChaosKillState.java

@@ -3,10 +3,8 @@
 public enum ChaosKillState {
     SUCCESS,
     CLUSTER_DEGRADED,
-    KILLING_IN_PROGRESS,
     FAILED,
     UNDER_COOLDOWN,
-    NOT_CORRECT_NODE_ID,
     KILLING_TIME_ANNOUNCED,
     ALREADY_KILLED
 }

beedb-gateway/src/main/java/com/memcache/gateway/chaos/ChaosService.java

@@ -1,9 +1,11 @@
 package com.memcache.gateway.chaos;
 
 import java.time.Clock;
+import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.Map;
 import java.util.concurrent.Executor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,7 +47,6 @@ public ChaosState getState(){
     public ChaosState tick(ClusterSnapshot clusterSnapshot){
 
         ChaosState currentState = chaosState.get();
-        LOGGER.info("tick chaoservice state is "+currentState+" current tiem is "+clock.instant());
         boolean isHealthyCluster = healthyCluster(clusterSnapshot);
         Instant currentInstant = clock.instant();
         if(isHealthyCluster && currentState.phase().equals(ChaosPhase.IDLE)){
@@ -76,13 +77,11 @@ else if(nodeStatus.raftRole() == null && !currentState.observedDown()){
                 }
             }
         }
-        LOGGER.info("chaoservice final state is "+chaosState.get());
         return chaosState.get();
 
     }
 
     public ChaosKillState requestKill(String nodeId, ClusterSnapshot clusterSnapshot){
-        LOGGER.info("requestKill called by "+Thread.currentThread().getName()+" time is "+clock.instant());
         ChaosState currentState = chaosState.get();
         Instant nextEligibleAt = currentState.nextEligibleAt();
         Instant currentInstant = clock.instant();
@@ -161,4 +160,27 @@ public boolean healthyCluster(ClusterSnapshot clusterSnapshot){
         }
         return true;
     }
+
+    public ChaosView getChaosView(){
+        ChaosState chaosState = getState();
+        Instant now = clock.instant();
+        Instant killAt = chaosState.killAt();
+        Long secondsUntilKill = null;
+        if(chaosState.phase().equals(ChaosPhase.ANNOUNCED) && killAt != null){
+            long msUntilKill = Duration.between(now, killAt).toMillis();
+            secondsUntilKill = (long) Math.max(0, Math.ceilDiv(msUntilKill, 1000L));
+
+        }
+        return new ChaosView(
+            chaosState.phase(),
+            chaosState.targetNodeId(),
+            chaosState.nextAutoAt(),
+            secondsUntilKill
+        );
+    }
+
+    public long secondsUntilEligible(){
+        long msUntilKill = Duration.between(clock.instant(), getState().nextEligibleAt()).toMillis();
+        return Math.max(0, Math.ceilDiv(msUntilKill, 1000L));
+    }
 }

beedb-gateway/src/main/java/com/memcache/gateway/chaos/ChaosView.java

@@ -0,0 +1,11 @@
+package com.memcache.gateway.chaos;
+
+import java.time.Instant;
+
+// this will be shared with the client(on frontend)
+public record ChaosView(
+    ChaosPhase phase,
+    String targetNodeId,
+    Instant nextAutoAt,
+    Long secondsUntilKill
+) {}

beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbConfig.java

@@ -1,10 +1,20 @@
 package com.memcache.gateway.client;
 
+import java.nio.file.Path;
 import java.time.Clock;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 
+import com.memcache.gateway.chaos.ChaosService;
 import com.memcache.gateway.kv.KeyRegistry;
 import com.memcache.gateway.kv.WriteRateLimiter;
+import com.memcache.gateway.supervisor.NodeSupervisor;
+import com.memcache.gateway.supervisor.ProcessNodeSupervisor;
+
+import ch.qos.logback.core.pattern.parser.Node;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -60,6 +70,13 @@ public static class BeedbProperties {
         /** if we dont receive kill command, automatically trigger leader kill after these many seconds */
         private int cooldownSeconds = 30;
 
+        private int chaosStartupGraceSeconds = 60;
+
+        private String supervisorServerDir = "../beedb-server";
+
+        private int restartAfterSeconds = 15;
+
+
         public int getKeyTtlSeconds() { return keyTtlSeconds; }
         public void setKeyTtlSeconds(int keyTtlSeconds) { this.keyTtlSeconds = keyTtlSeconds; }
         public int getMaxTrackedKeys() { return maxTrackedKeys; }
@@ -74,6 +91,12 @@ public static class BeedbProperties {
         public void setAutoKillSeconds(int autoKillSeconds) { this.autoKillSeconds = autoKillSeconds; }
         public int getCooldownSeconds() { return cooldownSeconds; }
         public void setCooldownSeconds(int cooldownSeconds) { this.cooldownSeconds = cooldownSeconds; }
+        public int getChaosStartupGraceSeconds() { return chaosStartupGraceSeconds; }
+        public void setChaosStartupGraceSeconds(int chaosStartupGraceSeconds) { this.chaosStartupGraceSeconds = chaosStartupGraceSeconds; }
+        public String getSupervisorServerDir() { return supervisorServerDir; }
+        public void setSupervisorServerDir(String supervisorServerDir) { this.supervisorServerDir = supervisorServerDir; }
+        public int getRestartAfterSeconds() { return restartAfterSeconds; }
+        public void setRestartAfterSeconds(int restartAfterSeconds) { this.restartAfterSeconds = restartAfterSeconds; }
 
         public Map<String, String> getNodes() {
             return nodes;
@@ -121,4 +144,31 @@ public BeedbClient beedbClient() {
     public Clock clock(){
         return Clock.systemUTC();
     }
+
+    @Bean 
+    public NodeSupervisor nodeSupervisor(){
+        Set<String> nodeIds = beedbProperties.getNodes().keySet();
+        String supervisorServerDir = beedbProperties.getSupervisorServerDir();
+        int restartAfterSeconds = beedbProperties.getRestartAfterSeconds();
+        return new ProcessNodeSupervisor(Set.copyOf(nodeIds), Path.of(supervisorServerDir), restartAfterSeconds);
+    }
+
+    @Bean 
+    public ChaosService chaosService(NodeSupervisor nodeSupervisor, Clock clock){
+        int announceLeadSeconds = beedbProperties.getAnnounceLeadSeconds();
+        int autoKillSeconds = beedbProperties.getAutoKillSeconds();
+        int cooldownSeconds = beedbProperties.getCooldownSeconds();
+        int startingSeconds = beedbProperties.getChaosStartupGraceSeconds();
+        Executor killExecutor = Executors.newFixedThreadPool(1 , killThreadFactory());
+        return new ChaosService(nodeSupervisor, announceLeadSeconds, autoKillSeconds, cooldownSeconds, clock, startingSeconds, killExecutor);
+    }
+
+    private ThreadFactory killThreadFactory(){
+        ThreadFactory threadFactory = runnable -> {
+            Thread thread = new Thread(runnable, "chaos-kill");
+            thread.setDaemon(true);
+            return thread;
+        };
+        return threadFactory;
+    }
 }

beedb-gateway/src/main/java/com/memcache/gateway/cluster/ClusterStatusService.java

@@ -12,7 +12,9 @@
 import org.springframework.stereotype.Service;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
+import com.memcache.gateway.chaos.ChaosService;
 import com.memcache.gateway.client.BeedbClient;
+import com.memcache.gateway.demo.DemoWriter;
 
 @Service 
 public class ClusterStatusService {
@@ -21,8 +23,14 @@ public class ClusterStatusService {
     private final BeedbClient beedbClient;
     private final List<SseEmitter> emitters  = new CopyOnWriteArrayList<>();
     private static final Logger LOGGER = LoggerFactory.getLogger(ClusterStatusService.class);
-    public ClusterStatusService(BeedbClient beedbClient){
+    private final ChaosService chaosService;
+    volatile ClusterView clusterView;
+    private final DemoWriter demoWriter;
+    public ClusterStatusService(BeedbClient beedbClient, ChaosService chaosService, DemoWriter demoWriter){
         this.beedbClient = beedbClient;
+        this.chaosService = chaosService;
+        this.demoWriter = demoWriter;
+        this.clusterView = new ClusterView(null, chaosService.getChaosView() , demoWriter.view());
     }
     public ClusterSnapshot getClusterSnapshot(){
         if(clusterSnapshot == null){
@@ -45,7 +53,7 @@ public SseEmitter registerEmitter(){
         });
         emitters.add(emitter);  
         try {
-            emitter.send(clusterSnapshot);
+            emitter.send(clusterView);
         } catch (Exception e) {
             logFailure(e, "failed to send cluster status");
         }
@@ -56,7 +64,7 @@ public void sendClusterStatus(){
         LOGGER.debug("sending cluster status");
         emitters.forEach(emitter -> {
             try {
-                emitter.send(clusterSnapshot);
+                emitter.send(clusterView);
                 LOGGER.debug("sent cluster status");
             } catch (Exception e) {
                 emitters.remove(emitter);
@@ -94,6 +102,12 @@ public void updateClusterStatus(){
     public void scheduledUpdateClusterStatus(){
         updateClusterStatus();
         // sending updated cluster status to all connected clients
+        try{
+            chaosService.tick(clusterSnapshot);
+        }catch(Exception e){
+            logFailure(e, "failed to tick chaos service");
+        }
+        clusterView = new ClusterView(clusterSnapshot, chaosService.getChaosView(), demoWriter.view());
         LOGGER.debug("cluster status sent to all connected clients");
         sendClusterStatus();
     }
@@ -107,4 +121,9 @@ private void logFailure(Exception e, String message){
             LOGGER.error("{} ", message , e);
         }
     }
+
+    public ClusterView getClusterView(){
+        return new ClusterView(clusterSnapshot, chaosService.getChaosView() ,demoWriter.view());
+    }
+
 }