Add circuitBreakers to UI and refactor

pbernet · pbernet · commit 269a0f7fae35 · 2025-08-05T12:45:06.000+02:00
diff --git a/src/main/resources/dashboard.html b/src/main/resources/dashboard.html
@@ -26,8 +26,7 @@
         .stats-grid {
             display: grid;
             grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
-            gap: 15px;
-            margin-bottom: 20px;
+            gap: 15px; 
         }
 
         .stat-card {
@@ -100,53 +99,118 @@
             border: 1px solid #f5c6cb;
         }
 
-        .auto-refresh {
-            margin: 10px 0;
+        .section {
+            background: white;
+            border-radius: 8px;
+            padding: 20px;
+            margin-bottom: 20px;
+            box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1);
+        }
+
+        .refresh-link {
+            color: white;
+            text-decoration: none;
+        }
+
+        .cb-grid {
+            display: grid;
+            grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
+            gap: 15px;
+        }
+
+        .cb-item {
+            border: 1px solid #ecf0f1;
+            border-radius: 6px;
+            padding: 15px;
+            background: #fafafa;
+        }
+
+        .cb-header {
+            display: flex;
+            justify-content: space-between;
+            align-items: center;
+            margin-bottom: 10px;
+        }
+
+        .cb-target {
+            font-weight: bold;
+        }
+
+        .cb-status {
+            font-size: 16px;
+            font-weight: bold;
+        }
+
+        .cb-status.closed {
+            color: #27ae60;
+        }
+
+        .cb-status.opened {
+            color: #e74c3c;
+        }
+
+        .cb-status.half-opened {
+            color: #f39c12;
+        }
+
+        .cb-metrics {
+            font-size: 12px;
+            color: #7f8c8d;
+        }
+
+        .cb-metrics span {
+            margin-right: 15px;
         }
     </style>
 </head>
 <body>
 <div class="container">
     <div class="header">
-        <h1>🔄 ReverseProxy Monitor</h1>
+        <h1><a href="#" onclick="location.reload(); return false;" class="refresh-link">🔄</a> ReverseProxy Monitor</h1>
         <p>Real-time traffic monitoring and statistics</p>
     </div>
 
     <div id="connection-status" class="connection-status disconnected">
         🔴 Disconnected from ReverseProxy
     </div>
 
-    <div class="auto-refresh">
-        <label><input type="checkbox" id="auto-refresh" checked> Auto-refresh every 2 seconds</label>
+    <div class="section">
+        <h3>🔧 Circuit Breakers</h3>
+        <div class="cb-grid" id="cb-grid">
+            <p>Loading circuit breaker status...</p>
+        </div>
     </div>
 
-    <div class="stats-grid">
-        <div class="stat-card">
-            <div class="stat-value" id="total-requests">0</div>
-            <div class="stat-label">Total Requests</div>
-        </div>
-        <div class="stat-card">
-            <div class="stat-value" id="total-responses">0</div>
-            <div class="stat-label">Total Responses</div>
-        </div>
-        <div class="stat-card">
-            <div class="stat-value" id="error-rate">0.0%</div>
-            <div class="stat-label">Error Rate</div>
-        </div>
-        <div class="stat-card">
-            <div class="stat-value" id="avg-response-time">0ms</div>
-            <div class="stat-label">Avg Response Time</div>
+    <div class="section">
+        <h3>📊 Statistics (max last 1000 requests)</h3>
+        <div class="stats-grid">
+            <div class="stat-card">
+                <div class="stat-value" id="total-requests">0</div>
+                <div class="stat-label">Total Requests</div>
+            </div>
+            <div class="stat-card">
+                <div class="stat-value" id="total-responses">0</div>
+                <div class="stat-label">Total Responses</div>
+            </div>
+            <div class="stat-card">
+                <div class="stat-value" id="error-rate">0.0%</div>
+                <div class="stat-label">Error Rate</div>
+            </div>
+            <div class="stat-card">
+                <div class="stat-value" id="avg-response-time">0ms</div>
+                <div class="stat-label">Avg Response Time</div>
+            </div>
         </div>
     </div>
 
     <div class="traffic-table">
+        <h3>🚦 Traffic data (last 100 requests)</h3>
         <table>
             <thead>
             <tr>
                 <th>Time</th>
                 <th>Method</th>
-                <th>Path</th>
-                <th>Target</th>
+                <th>Target URL</th>
                 <th>Status</th>
                 <th>Duration</th>
                 <th>Correlation ID</th>
@@ -163,14 +227,13 @@ <h1>🔄 ReverseProxy Monitor</h1>
 
 <script>
     let ws = null;
-    let autoRefreshEnabled = true;
 
     function formatTimestamp(timestamp) {
         return new Date(timestamp).toLocaleTimeString();
     }
 
     function formatDuration(ms) {
-        return ms ? ms + 'ms' : '-';
+        return ms ? ms + ' ms' : '-';
     }
 
     function formatStatus(status) {
@@ -184,6 +247,36 @@ <h1>🔄 ReverseProxy Monitor</h1>
         document.getElementById('total-responses').textContent = stats.totalResponses;
         document.getElementById('error-rate').textContent = (stats.errorRate * 100).toFixed(1) + '%';
         document.getElementById('avg-response-time').textContent = Math.round(stats.avgResponseTime) + 'ms';
+        updateCircuitBreakers(stats.circuitBreakers);
+    }
+
+    function updateCircuitBreakers(circuitBreakers) {
+        const container = document.getElementById('cb-grid');
+
+        if (circuitBreakers.length === 0) {
+            container.innerHTML = '<p>No circuit breakers configured</p>';
+            return;
+        }
+
+        const items = circuitBreakers.map(cb => {
+            const lastFailure = cb.lastFailure
+                ? new Date(cb.lastFailure).toLocaleString()
+                : '-';
+
+            return `
+                    <div class="cb-item">
+                        <div class="cb-header">
+                            <span class="cb-target">${cb.target}</span>
+                            <span class="cb-status ${cb.state.toLowerCase()}">${cb.state}</span>
+                        </div>
+                        <div class="cb-metrics">
+                            <span>Last Failure: ${lastFailure}</span>
+                        </div>
+                    </div>
+                `;
+        }).join('');
+
+        container.innerHTML = items;
     }
 
     function updateTrafficTable(traffic) {
@@ -193,15 +286,14 @@ <h1>🔄 ReverseProxy Monitor</h1>
             return;
         }
 
-        const rows = traffic.slice(-100).reverse().map(entry => {
+        const rows = traffic.reverse().map(entry => {
             const req = entry.request;
             const resp = entry.response;
             return `
                     <tr>
                         <td>${formatTimestamp(req.timestamp)}</td>
                         <td>${req.method}</td>
-                        <td>${req.uri}</td>
-                        <td>${req.targetHost}</td>
+                        <td>${req.targetUrl}</td>
                         <td>${formatStatus(resp ? resp.status : null)}</td>
                         <td>${formatDuration(entry.duration)}</td>
                         <td>${req.correlationId}</td>
@@ -261,19 +353,13 @@ <h1>🔄 ReverseProxy Monitor</h1>
             .catch(error => console.error('Error loading stats:', error));
     }
 
-    // Auto-refresh toggle
-    document.getElementById('auto-refresh').addEventListener('change', function () {
-        autoRefreshEnabled = this.checked;
-    });
 
     // Auto-refresh interval
     setInterval(() => {
-        if (autoRefreshEnabled) {
             loadTrafficData();
             if (!ws || ws.readyState !== WebSocket.OPEN) {
                 loadStats();
             }
-        }
     }, 2000);
 
     // Initial load
diff --git a/src/main/scala/akkahttp/ReverseProxy.scala b/src/main/scala/akkahttp/ReverseProxy.scala
@@ -20,8 +20,8 @@ import org.bouncycastle.util.encoders.Hex
 import org.slf4j.{Logger, LoggerFactory}
 
 import java.security.MessageDigest
-import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.{ConcurrentHashMap, ThreadLocalRandom}
 import scala.collection.parallel.CollectionConverters.ImmutableIterableIsParallelizable
 import scala.concurrent.*
 import scala.concurrent.duration.DurationInt
@@ -47,12 +47,11 @@ import scala.util.{Failure, Success}
   *
   * Remarks:
   *  - The target server selection is via the "Host" HTTP header
-  *  - Local/Remote target servers are designed to be flaky to show Retry/CircuitBreaker behavior
-  *    e.g. for Local adjust [[responseCodes]]
+  *  - Local/Remote target servers are designed to be faulty to show Retry/CircuitBreaker behavior
+  *    e.g. for mode Local adjust [[responseCodes]]
   *  - On top of the built-in client, you may also try other clients, see below
-  *  - This PoC may not scale well, possible bottlenecks are:
-  *     - Combination of CircuitBreaker which wraps Retry
-  *     - Round robin impl. with `requestCounter` means shared state
+  *  - This PoC may not scale well, because the 'round robin' implementation
+  *    with `requestCounter` means shared state
   *
   * Gatling client: [[ReverseProxySimulation]]
   *
@@ -77,8 +76,6 @@ object ReverseProxy extends App {
 
   val http: HttpExt = Http(system)
 
-  ReverseProxyMonitor.initializeWebUI(system)
-
   val circuitBreakers = new ConcurrentHashMap[String, CircuitBreaker]()
   val requestCounter = new AtomicInteger(0)
 
@@ -98,14 +95,19 @@ object ReverseProxy extends App {
     )
   )
 
-  // For Mode.local: Adjust to provoke more retries on ReverseProxy
-  val responseCodes = List(200, 200, 200, 200, 200, 200, 200, 200, 500, 503)
+  // For Mode.local: Add more failure response codes to provoke more retries on ReverseProxy
+  // and thus provoke the CircuitBreaker to open
+  //val responseCodes = List(200, 200, 200, 200, 200, 200, 200, 200, 500, 503)
+  val responseCodes = List(200, 200, 500, 500, 500, 500, 503, 503, 503, 503)
 
   localTargetServers(maxConnections = 100) // 1-1024
   reverseProxy()
-  // Switch Mode to let ReverseProxy forward client requests to local/remote target server(s)
+
+  // Switch mode to let ReverseProxy forward client requests to local/remote target server(s)
   // Note that the remote servers can not interpret the X-Correlation-ID header
-  clients(nbrOfClients = 10, requestsPerClient = 10, Mode.local)
+  val mode = Mode.remote
+  clients(nbrOfClients = 10, requestsPerClient = 10, mode)
+  ReverseProxyMonitor.initializeWebUI(system, services(mode))
 
   sys.addShutdownHook {
     ReverseProxyMonitor.shutdown()
@@ -131,7 +133,7 @@ object ReverseProxy extends App {
       }
 
       Source(1 to nbrOfRequests)
-        .throttle(1, 1.second, 10, ThrottleMode.shaping)
+        .throttle(1, 2.seconds, 10, ThrottleMode.shaping)
         .wireTap(each => logger.info(s"Client: $clientId about to send request with id: $clientId-$each..."))
         .mapAsync(1)(each => http.singleRequest(HttpRequest(uri = s"http://$proxyHost:$proxyPort/$fixedPath")
           .withHeaders(Seq(RawHeader("Host", targetHost.toString), RawHeader("X-Correlation-ID", s"$clientId-$each")))))
@@ -162,7 +164,6 @@ object ReverseProxy extends App {
       val mode = Mode.values.find(_.toString == host).getOrElse(Mode.local)
       val id = request.getHeader("X-Correlation-ID").orElse(RawHeader("X-Correlation-ID", "N/A")).value()
 
-      val requestId = ReverseProxyMonitor.logRequest(request, host, id)
       val startTime = System.currentTimeMillis()
 
       def headers(target: Target) = {
@@ -205,20 +206,19 @@ object ReverseProxy extends App {
           val target = seq(index)
           logger.info(s"Forwarding request with id: $id to $mode target server: ${target.url}")
 
+          val requestId = ReverseProxyMonitor.logRequest(request, target.url, id)
+
           val circuitBreaker = circuitBreakers.computeIfAbsent(target.url, _ => {
             val cb = new CircuitBreaker(
               system.scheduler,
-              // Account for retry attempts
-              // A lower value opens the circuit breaker for subsequent requests (until resetTimeout)
               maxFailures = 5,
               // Needs to be shorter than pekko-http 'request-timeout' (20s)
               // If not, clients get 503 from pekko-http
-              callTimeout = 15.seconds,
-              resetTimeout = 10.seconds)
-            // For yet unknown reasons these callbacks are not executed
-            cb.onOpen { () => logger.info(s"CircuitBreaker OPENED for ${target.url}") }
-            cb.onClose { () => logger.info(s"CircuitBreaker CLOSED for ${target.url}") }
-            cb.onHalfOpen { () => logger.info(s"CircuitBreaker HALF-OPEN for ${target.url}") }
+              callTimeout = 5.seconds,
+              resetTimeout = 5.seconds)
+            cb.onOpen(ReverseProxyMonitor.logCircuitBreakerEvent(target.url, "OPENED"))
+            cb.onClose(ReverseProxyMonitor.logCircuitBreakerEvent(target.url, "CLOSED"))
+            cb.onHalfOpen(ReverseProxyMonitor.logCircuitBreakerEvent(target.url, "HALF-OPENED"))
             cb
           })
 
@@ -248,17 +248,18 @@ object ReverseProxy extends App {
             case Failure(exception) =>
               val responseTime = System.currentTimeMillis() - startTime
               val errorResponse = exception match {
-                case _: CircuitBreakerOpenException => BadGateway(id, "Circuit breaker opened")
+                case e: CircuitBreakerOpenException => BadGateway(id, e.getMessage)
                 case _: TimeoutException => GatewayTimeout(id)
                 case e => BadGateway(id, e.getMessage)
               }
               ReverseProxyMonitor.logResponse(requestId, errorResponse, responseTime, id, Some(exception.getMessage))
           }.recover {
-            case _: CircuitBreakerOpenException => BadGateway(id, "Circuit breaker opened")
+            case e: CircuitBreakerOpenException => BadGateway(id, e.getMessage)
             case _: TimeoutException => GatewayTimeout(id)
             case e => BadGateway(id, e.getMessage)
           }
         case None =>
+          val requestId = ReverseProxyMonitor.logRequest(request, host, id)
           val notFoundResponse = NotFound(id, host)
           val responseTime = System.currentTimeMillis() - startTime
           ReverseProxyMonitor.logResponse(requestId, notFoundResponse, responseTime, id, Some("Host not found"))
@@ -281,7 +282,7 @@ object ReverseProxy extends App {
     val echoRoute: Route =
       extractRequest { request =>
         complete {
-          Thread.sleep(500)
+          Thread.sleep(ThreadLocalRandom.current.nextInt(1, 20) * 100)
           val id = request.getHeader("X-Correlation-ID").orElse(RawHeader("X-Correlation-ID", "N/A")).value()
 
           val randomResponseCode = responseCodes(new scala.util.Random().nextInt(responseCodes.length))
diff --git a/src/main/scala/akkahttp/ReverseProxyMonitor.scala b/src/main/scala/akkahttp/ReverseProxyMonitor.scala
