feat: 统计节点带宽

Author: tianxiu2b2t

assets/js/index.js

@@ -1203,6 +1203,9 @@ class Tools {
     static formatDate(d) {
         return `${d.getFullYear().toString().padStart(4, "0")}-${(d.getMonth() + 1).toString().padStart(2, "0")}-${d.getDate().toString().padStart(2, "0")}`
     }
+    static formatDateHourMinute(d) {
+        return `${d.getFullYear().toString().padStart(4, "0")}-${(d.getMonth() + 1).toString().padStart(2, "0")}-${d.getDate().toString().padStart(2, "0")} ${d.getHours().toString().padStart(2, "0")}:${d.getMinutes().toString().padStart(2, "0")}`
+    }
 }
 class UserAuth {
     constructor() {
@@ -1812,12 +1815,29 @@ async function load() {
                     var type = $dashboard_locals.advanced_time_switch.current.button
                     var day = +type.slice(type.lastIndexOf(".") + 1)
                     var ip_access = await $channel.send("response_ip_access", day)
+                    var resp = {}
+                    var server_time = $dashboard_locals.info_runtime
+                    var datetime = (server_time.current_time - server_time.diff / 1000.0 + (+new Date() - server_time.resp_timestamp) / 1000.0);
+                    var previous;
+                    if (day == 30) {
+                        previous = (datetime - ((datetime + UTC_OFFSET) % 86400) - 86400 * day);
+                        for (var i = 0; i <= day; i++) {
+                            let key = Tools.formatDate(new Date((previous + 86400 * i) * 1000.0))
+                            resp[key] = ip_access[key] || 0
+                        }
+                    } else {
+                        previous = (datetime - ((datetime + UTC_OFFSET) % 3600) - 3600 * (day * 24));
+                        for (var i = 0; i <= day * 24; i++) {
+                            let key = Tools.formatDateHourMinute(new Date((previous + i * 3600) * 1000.0))
+                            resp[Tools.formatDateHourMinute(new Date((previous + i * 3600) * 1000.0))] = ip_access[key] || 0
+                        }
+                    }
                     $dashboard_locals.advanced_ip_access_echarts.base.setOption({
                         xAxis: {
-                            data: Object.keys(ip_access),
+                            data: Object.keys(resp),
                         },
                         series: [{
-                            data: Object.values(ip_access),
+                            data: Object.values(resp),
                         }]
                     })
                     waitTaskResponse.ip_access = false;

core/cluster.py

@@ -1,7 +1,7 @@
 import abc
 import asyncio
 from collections import defaultdict, deque
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, field
 import datetime
 import enum
 import hashlib
@@ -943,6 +943,7 @@ def __init__(self, id: str, secret: str):
         self.want_enable: bool = False
         self.enabled = False
         self.enable_count: int = 0
+        self.keepalive_error: int = 0
         self.keepalive_task: Optional[int] = None
         self.delay_enable_task: Optional[int] = None
         self.counter: defaultdict[storages.iStorage, ClusterCounter] = defaultdict(ClusterCounter)
@@ -1052,6 +1053,7 @@ async def enable(self):
         self.enabled = True
         clusters.status.enter()
         self.enable_count = 0
+        self.keepalive_error = 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)
@@ -1083,9 +1085,14 @@ async def keepalive(self):
             }
         )
         if result.err or not result.ack:
-            logger.twarning("cluster.warning.kicked_by_remote")
-            await self.disable()
+            if self.keepalive_error >= 3:
+                logger.twarning("cluster.warning.kicked_by_remote", cluster=self.id)
+                await self.disable()
+            else:
+                self.keepalive_error += 1
+                logger.twarning("cluster.warning.keepalive", cluster=self.id, count=self.keepalive_error)
             return
+        self.keepalive_error = 0
         self.no_storage_counter -= commit_no_storage_counter
         for storage, counter in commit_counter.items():
             self.counter[storage] -= counter
@@ -1290,6 +1297,53 @@ def __init__(self, hash: str, size: int, mtime: float, data: bytes, storage: Opt
         super().__init__(hash=hash, size=size, mtime=mtime, storage=storage)
         self.data = data
 
+@dataclass
+class Bandwidth:
+    cluster_id: str
+    bytes: int
+
+@dataclass
+class BandwidthList:
+    cluster_id: str
+    bytes: deque[int] = field(default_factory=deque)
+
+class BandwidthCounter:
+    def __init__(
+        self
+    ):
+        self.bandwidths: defaultdict[str, defaultdict[int, int]] = defaultdict(lambda: defaultdict(int))
+        scheduler.run_repeat_later(self.gc, 600, 5)
+
+    def hit(self, cluster_id: str, bytes: int):
+        current = int(time.monotonic())
+        self.bandwidths[cluster_id][current] += bytes
+
+    
+    def gc(self):
+        current = int(time.monotonic())
+        for cluster_id in self.bandwidths:
+            for timestamp in list(self.bandwidths[cluster_id]):
+                if current - timestamp > 600:
+                    del self.bandwidths[cluster_id][timestamp]
+
+    def total(self, interval: int = 1) -> list[Bandwidth]:
+        res: list[Bandwidth] = []
+        for b in self.get(interval):
+            res.append(Bandwidth(b.cluster_id, sum(b.bytes)))
+        return res
+
+    def get(self, interval: int = 1) -> list[BandwidthList]:
+        current = int(time.monotonic()) - 1
+        bandwidths: list[BandwidthList] = []
+        for cluster_id, b in self.bandwidths.items():
+            obj = BandwidthList(cluster_id)
+            for timestamp in range(current - interval, current + 1):
+                if current - timestamp > interval:
+                    continue
+                obj.bytes.append(b[timestamp])
+            bandwidths.append(obj)
+        return bandwidths
+
 ROOT = Path(__file__).parent.parent
 
 CHECK_FILE_CONTENT = "Python OpenBMCLAPI"
@@ -1307,7 +1361,7 @@ def __init__(self, hash: str, size: int, mtime: float, data: bytes, storage: Opt
 DEFAULT_MEASURES = [
     10
 ]
-
+BANDWIDTH_COUNTER = BandwidthCounter()
 routes = web.routes
 aweb = web.web
 clusters = ClusterManager()
@@ -1525,6 +1579,7 @@ async def _(request: aweb.Request):
         size = end - start + 1
 
         cluster.hit(file.storage, size)
+        BANDWIDTH_COUNTER.hit(cluster.id, size)
         # add database
         # stats
         name = query.get("name")

core/config.py

@@ -190,7 +190,7 @@ def disallow_public_dashboard(self):
 
 const = Const()
 
-VERSION = "3.4.2"
+VERSION = "3.5.0"
 API_VERSION = "1.13.1"
 USER_AGENT = f"openbmclapi/{API_VERSION} python-openbmclapi/{VERSION}"
 PYTHON_VERSION = ".".join(map(str, (sys.version_info.major, sys.version_info.minor, sys.version_info.micro)))

core/dashboard.py

@@ -689,13 +689,13 @@ def _(req_data: Any) -> Any:
     if day > 7:
         for hour, data in query_data.items():
             date = datetime.datetime.fromtimestamp(hour * 3600)
-            key = f"{date.year}-{date.month}-{date.day}"
+            key = f"{date.year:04d}-{date.month:02d}-{date.day:02d}"
             temp_data[key] |= data.keys()
     else:
         for hour, data in query_data.items():
             date = datetime.datetime.fromtimestamp(hour * 3600)
             hour = date.hour
-            key = f"{date.year}-{date.month}-{date.day} {date.hour:02d}:{date.minute:02d}"
+            key = f"{date.year:04d}-{date.month:02d}-{date.day:02d} {date.hour:02d}:{date.minute:02d}"
             temp_data[key] |= data.keys()
     resp_data: defaultdict[str, int] = defaultdict(int)
     for key, data in temp_data.items():
@@ -797,6 +797,10 @@ def _(req_data: Any) -> Any:
             days_data[storage_id].append(APIStatistics(units.format_date(day * 86400), item.bytes, item.hits))
     return days_data
 
+@API.on("clusters_bandwidth")
+def _(req_data: Any) -> Any:
+    return cluster.BANDWIDTH_COUNTER.get(max(1, req_data) if isinstance(req_data, int) else 1)
+
 @DeprecationWarning
 async def handle_api(
     event: str,

core/storages/alist.py

@@ -12,7 +12,7 @@
 from core.logger import logger
 from core.utils import WrapperTQDM
 
-from .base import MeasureFile, iNetworkStorage, Range, DOWNLOAD_DIR, File, CollectionFile
+from .base import FileInfo, MeasureFile, iNetworkStorage, Range, DOWNLOAD_DIR, File, CollectionFile
 
 
 ALIST_TOKEN_DEAULT_EXPIRY = 86400 * 2
@@ -58,8 +58,7 @@ def __init__(
         cache_timeout: float = 60,
         retries: int = 3, 
         public_webdav_endpoint: str = "",
-        public_webdav_username: Optional[str] = None,
-        public_webdav_password: Optional[str] = None
+        s3_custom_host: str = ""
     ):
         super().__init__(path, username, password, endpoint, weight, list_concurrent, name, cache_timeout)
         self.retries = retries
@@ -70,6 +69,10 @@ def __init__(
             }
         )
         self._public_webdav_endpoint = ""
+        self._s3_custom_host = s3_custom_host
+        if s3_custom_host:
+            urlobject = urlparse.urlparse(s3_custom_host)
+            self._s3_custom_host = f"{urlobject.scheme}://{urlobject.netloc}{urlobject.path}"
         if public_webdav_endpoint:
             urlobject = urlparse.urlparse(public_webdav_endpoint)
             self._public_webdav_endpoint = f"{urlobject.scheme}://{urlparse.quote(self.username)}:{urlparse.quote(self.password)}@{urlobject.netloc}{urlobject.path}"
@@ -165,11 +168,18 @@ async def get_files(root_id: int):
 
         sem = asyncio.Semaphore(self.list_concurrent)
         results = set()
-        for result in await asyncio.gather(*(
-            get_files(root_id)
-            for root_id in Range()
+        for root_id, result in zip(
+            Range(),
+            await asyncio.gather(*(
+                get_files(root_id)
+                for root_id in Range()
+            )
         )):
             for file in result:
+                self.filelist[str(self.path / DOWNLOAD_DIR / f"{root_id:02x}" / file["name"])] = FileInfo(
+                    file["size"],
+                    utils.parse_isotime_to_timestamp(file["modified"]),
+                )
                 results.add(File(
                     file["name"],
                     file["size"],
@@ -219,20 +229,28 @@ async def read_file(self, file: File) -> io.BytesIO:
             return io.BytesIO(await resp.read())
 
     async def get_url(self, file: CollectionFile) -> str:
+        if self._s3_custom_host:
+            return f"{self._s3_custom_host}{str(self.get_path(file))}"
         if self._public_webdav_endpoint:
             return f"{self._public_webdav_endpoint}{str(self.get_path(file))}"
         info = await self.__info_file(file)
         return '' if info.size == -1 else info.raw_url
 
     async def write_file(self, file: MeasureFile | File, content: io.BytesIO):
+        path = str(self.get_path(file))
         result = await self.__action_data(
             "put",
             "/api/fs/put",
             content.getvalue(),
             {
-                "File-Path": urlparse.quote(str(self.get_path(file))),
+                "File-Path": urlparse.quote(path),
             }
         )
+        if result.code == 200:
+            self.filelist[path] = FileInfo(
+                result.data["size"],
+                utils.parse_isotime_to_timestamp(result.data["modified"])
+            )
         return result.code == 200
 
     async def close(self):
@@ -256,6 +274,13 @@ async def get_mtime(self, file: MeasureFile | File) -> float:
         return (await self.__info_file(file)).modified
 
     async def get_size(self, file: MeasureFile | File) -> int:
+        path = str(self.get_path(file))
+        if path in self.filelist:
+            return self.filelist[path].size
         info = await self.__info_file(file)
+        self.filelist[path] = FileInfo(
+            info.size,
+            info.modified
+        )
         return max(0, info.size)
 

core/storages/base.py

@@ -1,4 +1,5 @@
 import abc
+import collections
 from dataclasses import dataclass
 import hashlib
 import io
@@ -28,6 +29,11 @@ class MeasureFile:
     def __hash__(self) -> int:
         return hash(self.size)
 
+@dataclass
+class FileInfo:
+    size: int
+    mtime: float
+
 
 class FilePath(object):
     def __init__(self, path: str) -> None:
@@ -90,6 +96,36 @@ def __rtruediv__(self, __o: Any) -> 'FilePath':
 CollectionFile = MeasureFile | File
 Range = lambda: range(0, 256)
 
+class FileList:
+    def __init__(
+        self,
+    ):
+        self._data: collections.OrderedDict[str, FileInfo] = collections.OrderedDict()
+
+    def __contains__(self, key: str):
+        return key in self._data
+    
+    def __getitem__(self, key: str):
+        return self._data[key]
+    
+    def __setitem__(self, key: str, value: FileInfo):
+        self._data[key] = value
+
+    def __delitem__(self, key: str):
+        del self._data[key]
+
+    def __iter__(self):
+        return iter(self._data)
+    
+    def __len__(self):
+        return len(self._data)
+    
+    def __repr__(self):
+        return f"FileList({self._data})"
+    
+    def __str__(self):
+        return f"FileList({self._data})"
+    
 
 class iStorage(metaclass=abc.ABCMeta):
     type: str = "_interface"
@@ -105,6 +141,7 @@ def __init__(
         self.list_concurrent = list_concurrent
         self.current_weight = 0
         self._name = name
+        self.filelist = FileList()
 
     @property
     @abc.abstractmethod