fix: 增加上线失败，延长重试时间

Author: tianxiu2b2t

assets/js/index.js

@@ -0,0 +1,86 @@
+class Utils {
+    static uuid(len, radix) {
+        var chars = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'.split('');
+        var uuid = [], i;
+        radix = radix || chars.length;
+     
+        if (len) {
+            for (i = 0; i < len; i++) uuid[i] = chars[0 | Math.random()*radix];
+        } else {
+            var r;
+            uuid[8] = uuid[13] = uuid[18] = uuid[23] = '-';
+            uuid[14] = '4';
+            for (i = 0; i < 36; i++) {
+                if (!uuid[i]) {
+                    r = 0 | Math.random()*16;
+                    uuid[i] = chars[(i == 19) ? (r & 0x3) | 0x8 : r];
+                }
+            }
+        }
+        return uuid.join('');
+    }
+}
+
+class Socket {
+    constructor(baseurl) {
+        this._baseurl = baseurl;
+        this._ws = null;
+        this._wsReconnectTask = null;
+        this._wsConnect()
+        window.addEventListener("beforeinput", () => {
+            this._ws?.close()
+        })
+    }
+    async _sendByFetch(event, data) {
+        var resp = await fetch({
+            "url": this._baseurl,
+            "body": JSON.stringify(
+                {
+                    "event": event,
+                    "data": data,
+                    "echo_id": Utils.uuid()
+                }
+            )
+        })
+        return await resp.json()
+    }
+    _wsConnect() {
+        clearTimeout(this._wsReconnectTask)
+        this._ws?.close()
+        this._ws = new WebSocket(
+            this._baseurl
+        )
+        this._ws.addEventListener("close", () => {
+            console.warn("The websocket has disconnected. After 5s to reconnect.")
+            setTimeout(() => {
+                this._wsConnect()
+            }, 5000)
+        })
+        this._ws.addEventListener("message", (event) => {
+            var raw_data = JSON.parse(event.data);
+        })
+    }
+    async _sendByWs(event, data) {
+        if (this._ws?.state != WebSocket.OPEN) return;
+        var echo_id = Utils.uuid();
+        this._ws.send({
+            "event": event,
+            "data": data,
+            echo_id
+        })
+        var promise = new Promise((resolve, reject) => {
+            
+        });
+        return promise;
+    }
+    async send(event, data) {
+        if (this._ws != null && this._ws.state == WebSocket.OPEN) {
+            return this._sendByWs(event, data)
+        } else {
+            return this._sendByFetch(event, data)
+        }
+    }
+    
+}
+
+const $socket = new Socket(window.location.origin + "/api")

core/cluster.py

@@ -504,6 +504,8 @@ def __init__(self, id: str, secret: str):
         self.socket_io = ClusterSocketIO(self)
         self.want_enable: bool = False
         self.enabled = False
+        self.banned = False
+        self.enable_count: int = 0
         self.keepalive_task: Optional[int] = None
         self.delay_enable_task: Optional[int] = None
         self.counter = ClusterCounter()
@@ -572,29 +574,37 @@ async def request_cert(self):
     async def enable(self):
         cert = await clusters.get_certificate()
         scheduler.cancel(self.delay_enable_task)
-        if self.want_enable:
+        if self.want_enable or self.banned:
             return
         self.want_enable = True
+        self.enable_count += 1
         logger.tinfo("cluster.info.want_enable", cluster=self.id)
-        result = await self.socket_io.emit(
-            "enable", {
-                "host": cert.host,
-                "port": config.const.public_port,
-                "byoc": True,
-                "version": API_VERSION,
-                "noFastEnable": True,
-                "flavor": {
-                    "storage": "local",
-                    "runtime": "python"
+        try:
+            result = await self.socket_io.emit(
+                "enable", {
+                    "host": cert.host,
+                    "port": config.const.public_port,
+                    "byoc": True,
+                    "version": API_VERSION,
+                    "noFastEnable": True,
+                    "flavor": {
+                        "storage": "local",
+                        "runtime": "python"
+                    }
                 }
-            }
-        )
-        self.want_enable = False
+            , 120)
+        except:
+            logger.terror("cluster.error.cluster.banned", cluster=self.id)
+            self.banned = True
+            return
+        finally:
+            self.want_enable = False
         if result.err:
             self.socketio_error("enable", result)
             self.retry()
             return
         self.enabled = True
+        self.enable_count = 0
         scheduler.cancel(self.keepalive_task)
         self.keepalive_task = scheduler.run_repeat_later(self.keepalive, 1, interval=60)
         logger.tsuccess("cluster.success.enabled", cluster=self.id)
@@ -640,8 +650,9 @@ async def disable(self, exit: bool = False):
             self.retry()
 
     def retry(self):
-        self.delay_enable_task = scheduler.run_later(self.enable, 60)
-        logger.tinfo("cluster.info.cluster.retry_enable", delay=units.format_count_datetime(60))
+        delay = ((self.enable_count  + 1) ** 2) * 60
+        self.delay_enable_task = scheduler.run_later(self.enable, delay)
+        logger.tinfo("cluster.info.cluster.retry_enable", delay=units.format_count_datetime(delay))
 
 
     @property
@@ -695,6 +706,12 @@ async def _(message: Any):
                 message = message["message"]
             logger.tinfo("cluster.info.socketio.message", cluster=self.cluster.id, message=message)
 
+        @self.sio.on("exception") # type: ignore
+        async def _(message: Any):
+            if isinstance(message, dict) and "message" in message:
+                message = message["message"]
+            logger.terror("cluster.error.socketio.message", cluster=self.cluster.id, message=message)
+
     async def disconnect(self):
         await self.sio.disconnect()
 

core/dashboard.py

@@ -1,5 +1,6 @@
 from collections import deque
 from dataclasses import dataclass
+import json
 import os
 import socket
 
@@ -105,6 +106,32 @@ async def _(request: web.Request):
             return web.json_response(
                 await resp.json(),
             )
+        
+@route.get("/api")
+async def _(request: web.Request):
+    if request.headers.get("Connection", "").lower() == "upgrade" and request.headers.get("Upgrade", "").lower() == "websocket":
+        ws = web.WebSocketResponse()
+        ws.can_prepare(request)
+        await ws.prepare(request)
+        await ws.send_json({
+            "event": "echo",
+            "data": "hello world",
+            "echo_id": None
+        })
+        while not ws.closed:
+            try:
+                raw_data = await ws.receive()
+                try:
+                    print(raw_data)
+                except:
+                    ...
+            except:
+                break
+        return ws
+    else:
+        return web.json_response({
+        })
+    
 
 route.static("/assets", "./assets")
 

i18n/zh_cn.json

@@ -31,5 +31,6 @@
     "cluster.success.cluster.disable": "节点 [${cluster}] 已正常下线",
     "cluster.info.want_enable": "节点 [${cluster}] 正在启用中……",
     "cluster.info.cluster.retry_enable": "节点 [${cluster}] 将在 [${delay}] 后重新启用",
-    "cluster.success.enabled": "节点 [${cluster}] 已正常上线"
+    "cluster.success.enabled": "节点 [${cluster}] 已正常上线",
+    "cluster.error.socketio.message": "节点 [${cluster}] 收到错误消息，原因 [${message}]"
 }