Compare commits

..

14 Commits

Author SHA1 Message Date
23c88b5f48 feat: 优化URL处理逻辑,清理多余空格和重复协议前缀
- 在 handleGet 和 handlePost 函数中添加URL清理逻辑,去除多余的空格和重复的 http:// 或 https:// 前缀。
- 确保在URL没有协议前缀时自动添加 http://。
- 更新响应结构,新增 statuscode 字段以更好地反映请求状态。
2025-12-27 21:58:35 +08:00
c9c4da01b6 feat: 添加时间同步配置功能至安装脚本
- 在 install.sh 中新增 sync_time 函数,配置系统时间同步,设置时区为 Asia/Shanghai,并安装 chrony。
- 配置 NTP 服务器为阿里云和腾讯云,确保时间同步的准确性。
- 更新主函数以调用时间同步配置,优化安装流程。
2025-12-24 03:31:35 +08:00
7a104bbe42 feat: 更新时间同步逻辑,使用K780时间API替代中国时间API
- 修改时间同步服务,使用K780时间API获取北京时间,增强了API请求的错误处理和日志记录。
- 更新响应结构,解析新的时间戳格式,确保时间同步的准确性和稳定性。
2025-12-24 02:57:37 +08:00
e0d97c4486 feat: 添加时间同步服务和北京时间支持
- 在主程序中集成时间同步服务,每30分钟同步一次时间。
- 在心跳报告中加载并使用北京时间,确保心跳在每分钟的第1秒发送。
- 增强了错误处理,确保在加载时区失败时使用默认时区。
2025-12-24 02:34:05 +08:00
bb73e0f384 feat: 更新打包和安装逻辑,支持新格式发布包
- 在 all-upload-release.sh 中添加临时打包目录,复制二进制文件及必要的脚本和配置文件。
- 修改 install.sh 以支持新格式发布包的提取,简化安装流程,无需从 Git 克隆。
- 更新 INSTALL.md 和 README.md,说明新格式发布包的优点和安装步骤。
- 确保安装脚本能够处理旧格式发布包,保持向后兼容性。
2025-12-24 01:31:30 +08:00
b5fc83065c feat: 更新文档和配置逻辑,增强心跳机制和持续测试功能
- 在 INSTALL.md 和 README.md 中添加配置优先级说明,确保环境变量优先级最高。
- 增强心跳机制,新增字段以传递节点信息。
- 持续测试功能优化,支持批量推送和自动清理。
- 更新版本号至 v1.1.4,完善文档以反映新功能和改进。
2025-12-24 01:21:45 +08:00
ef31a054c0 chore: 更新版本号至 v1.1.3
增加 version host_name 2个新字段传递
2025-12-23 23:09:55 +08:00
ff35510ef0 修复 2025-12-17 21:12:49 +08:00
21592ae8a0 fix: 修复 IPv6 地址解析中的端口处理逻辑
- 将 LastIndex 替换为 Index,以正确找到第一个闭合括号。
- 添加逻辑以在端口部分为空时使用默认端口 80,解决了潜在的连接问题。
2025-12-17 20:09:51 +08:00
f01547df35 refactor: 优化 HTTP 请求处理逻辑
- 改进了对重定向的处理,确保在 CheckRedirect 返回 ErrUseLastResponse 时能够正确处理响应。
- 移除了不必要的空行以提升代码可读性。
- 增强了错误处理逻辑,确保在没有响应的情况下返回适当的错误信息。
2025-12-17 20:08:31 +08:00
4a2532a83b 强壮安装脚本 2025-12-17 19:56:16 +08:00
b962265168 refactor: 优化 TCPing 任务的目标解析逻辑
- 改进了 TCPing 任务中对 host:port 格式的解析,支持 IPv6 地址格式并默认使用端口 80。
- 移除了不必要的空行以提升代码可读性。
- 更新了安装脚本,移除了不再使用的镜像源。
2025-12-17 19:16:39 +08:00
38acca6484 fix: 改进心跳报告中的错误处理和日志记录
- 增强了 RegisterNode 和 sendHeartbeat 函数中的错误消息,包含 URL 和响应体详情以便更好地调试。
- 移除了不必要的空行以使代码结构更清晰。
2025-12-07 18:37:17 +08:00
8d36ef495d 修复 2025-12-07 18:09:49 +08:00
17 changed files with 1107 additions and 231 deletions

1
.gitignore vendored
View File

@@ -4,3 +4,4 @@ agent
node.log
node.pid
config.yaml
.DS_Store

View File

@@ -37,13 +37,17 @@ GITHUB_BRANCH=develop curl -fsSL https://raw.githubusercontent.com/yourbask/link
1. **检测系统** - 自动识别 Linux 发行版和 CPU 架构
2. **安装依赖** - 自动安装 Git、Go、ping、traceroute、dnsutils 等工具
3. **克隆源码** - 从 GitHub 克隆 node 项目源码到 `/opt/linkmaster-node`
4. **编译安装** - 自动编译源码并安装二进制文件
5. **创建服务** - 自动创建 systemd 服务文件(使用 run.sh 启动)
6. **启动服务** - 自动启动并设置开机自启
7. **验证安装** - 检查服务状态和健康检查
3. **下载发布包** - 从 Releases 下载预编译发布包(包含二进制文件和所有脚本)
4. **提取文件** - 从发布包提取所有文件到 `/opt/linkmaster-node`(无需克隆 Git
5. **编译安装** - 如果下载失败,自动从源码编译安装
6. **创建服务** - 自动创建 systemd 服务文件(使用 run.sh 启动)
7. **启动服务** - 自动启动并设置开机自启
8. **验证安装** - 检查服务状态和健康检查
**注意** 每次服务启动时会自动拉取最新代码并重新编译,确保使用最新版本。
**重要说明**
-**新格式发布包**:包含二进制文件、安装脚本、运行脚本等所有必要文件,安装时直接从压缩包提取,无需克隆 Git 仓库
-**向后兼容**:如果发布包是旧格式(仅包含二进制文件),安装脚本会自动从 Git 克隆获取脚本文件
- ⚠️ **源码编译模式**:如果下载失败,会从 Git 克隆源码并编译(需要网络连接)
## 安装后管理
@@ -109,7 +113,37 @@ curl http://localhost:2200/api/health
如果无法使用一键安装脚本,可以手动安装:
### 1. 克隆源码并编译
### 方式一:从发布包安装(推荐)
**优点**:无需 Git 和 Go 环境,直接使用预编译文件
```bash
# 1. 下载发布包(替换为实际版本和平台)
wget https://gitee.nas.cpolar.cn/yoyo/linkmaster-node/releases/download/v1.1.4/agent-linux-amd64-v1.1.4.tar.gz
# 2. 解压
tar -xzf agent-linux-amd64-v1.1.4.tar.gz
cd agent-linux-amd64-v1.1.4
# 3. 复制文件到安装目录
sudo mkdir -p /opt/linkmaster-node
sudo cp -r * /opt/linkmaster-node/
sudo chmod +x /opt/linkmaster-node/agent
sudo chmod +x /opt/linkmaster-node/*.sh
# 4. 复制二进制文件到系统目录
sudo cp /opt/linkmaster-node/agent /usr/local/bin/linkmaster-node
sudo chmod +x /usr/local/bin/linkmaster-node
# 5. 创建配置文件(从示例复制)
sudo cp /opt/linkmaster-node/config.yaml.example /opt/linkmaster-node/config.yaml
# 编辑配置文件,设置后端地址
sudo nano /opt/linkmaster-node/config.yaml
# 6. 创建 systemd 服务(参考下面的服务配置)
```
### 方式二:克隆源码并编译
```bash
# 克隆仓库
@@ -169,6 +203,43 @@ EOF
**注意:** 使用 `run.sh` 启动的好处是每次启动会自动拉取最新代码并重新编译。
### 3.1. 配置说明
**配置优先级(从高到低):**
1. 环境变量 `BACKEND_URL`(最高优先级)
2. 配置文件 `config.yaml` 中的 `backend.url`
3. 默认值
**重要说明:**
- 环境变量 `BACKEND_URL` 会**覆盖**配置文件中的设置
- 即使配置文件存在,设置环境变量后也会优先使用环境变量的值
- 这确保了编译后的二进制文件不会硬编码后端地址
- 配置文件不会被编译进二进制文件,是运行时读取的
**使用环境变量(推荐):**
```bash
# 在 systemd 服务文件中设置
Environment="BACKEND_URL=http://your-backend-server:8080"
# 或在命令行中设置
BACKEND_URL=http://your-backend-server:8080 ./run.sh start
```
**使用配置文件:**
创建 `/opt/linkmaster-node/config.yaml`
```yaml
server:
port: 2200
backend:
url: http://your-backend-server:8080
heartbeat:
interval: 60
log:
file: node.log
level: info
debug: false
```
### 4. 启动服务
```bash
@@ -177,7 +248,11 @@ sudo systemctl enable linkmaster-node
sudo systemctl start linkmaster-node
```
**注意** 确保 `BACKEND_URL` 环境变量指向后端服务器的实际地址和端口(默认 8080不是前端地址。
**重要说明**
- 确保 `BACKEND_URL` 环境变量指向后端服务器的实际地址和端口(默认 8080不是前端地址
- `BACKEND_URL` 环境变量会**覆盖**配置文件中的 `backend.url` 设置(优先级最高)
- 即使配置文件存在,设置环境变量后也会优先使用环境变量的值
- 这确保了编译后的二进制文件不会硬编码后端地址
## 防火墙配置
@@ -238,12 +313,19 @@ sudo lsof -i :2200
**解决:**
- 检查后端地址是否正确(应该是 `http://backend-server:8080`,不是前端地址)
- 检查环境变量 `BACKEND_URL` 是否设置正确(优先级最高)
- 检查配置文件 `config.yaml` 中的 `backend.url` 是否正确
- 检查网络连通性:`ping your-backend-server`
- 检查端口是否开放:`telnet your-backend-server 8080``nc -zv your-backend-server 8080`
- 检查防火墙规则(确保后端服务器的 8080 端口开放)
- 检查后端服务是否运行:`curl http://your-backend-server:8080/api/public/nodes/online`
- 如果使用前端代理,节点端仍需要直接连接后端,不能使用前端地址
**配置优先级说明:**
- 环境变量 `BACKEND_URL` 优先级最高,会覆盖配置文件中的设置
- 如果同时设置了环境变量和配置文件,优先使用环境变量的值
- 这确保了编译后的二进制文件不会硬编码后端地址
## 卸载
```bash

101
README.md
View File

@@ -85,12 +85,25 @@ BACKEND_URL=http://your-backend-server:8080 ./run.sh start
## 配置
### 配置优先级
配置按以下优先级加载(高优先级会覆盖低优先级):
1. **环境变量**(最高优先级)
2. **配置文件** `config.yaml`
3. **默认值**
### 环境变量
- `BACKEND_URL`: 后端服务地址(必需,默认: http://localhost:8080
- `BACKEND_URL`: 后端服务地址(**优先级最高**,会覆盖配置文件中的设置
- `CONFIG_PATH`: 配置文件路径(可选,默认: config.yaml
- `LOG_FILE`: 日志文件路径(可选,默认: node.log
**重要说明:**
- `BACKEND_URL` 环境变量会**覆盖**配置文件中的 `backend.url` 设置
- 即使配置文件存在,设置环境变量后也会优先使用环境变量的值
- 这确保了编译后的二进制文件不会硬编码后端地址
### 配置文件(可选)
创建 `config.yaml` 文件:
@@ -99,20 +112,28 @@ BACKEND_URL=http://your-backend-server:8080 ./run.sh start
server:
port: 2200
backend:
url: http://your-backend-server:8080
url: http://your-backend-server:8080 # 会被 BACKEND_URL 环境变量覆盖
heartbeat:
interval: 60
log:
file: node.log # 日志文件路径(默认: node.log空则输出到标准错误
level: info # 日志级别: debug, info, warn, error默认: info
debug: false
node:
id: 0 # 节点ID通过心跳自动获取
ip: "" # 节点IP通过心跳自动获取
country: "" # 国家(通过心跳自动获取)
province: "" # 省份(通过心跳自动获取)
city: "" # 城市(通过心跳自动获取)
isp: "" # ISP通过心跳自动获取
```
**日志配置说明:**
**配置说明:**
- `backend.url`: 后端服务地址,会被 `BACKEND_URL` 环境变量覆盖
- `log.file`: 日志文件路径。如果为空日志将输出到标准错误stderr
- `log.level`: 日志级别,支持 `debug``info``warn``error`
- 也可以通过环境变量 `LOG_FILE` 设置日志文件路径
- 日志文件会自动创建,如果目录不存在会自动创建
- `node.*`: 节点信息通过心跳自动获取并保存,无需手动配置
- 配置文件不会被编译进二进制文件,是运行时读取的
## 运行脚本
@@ -175,6 +196,7 @@ GITHUB_BRANCH=develop curl -fsSL https://gitee.nas.cpolar.cn/yoyo/linkmaster-nod
- 自动安装系统依赖curl, wget, git, ping, traceroute 等)
- 自动安装 Go 环境(优先使用系统包管理器,失败则从官网下载)
- 优先从 Releases 下载预编译二进制文件,失败则从源码编译
- **发布包包含所有必要文件**:二进制文件、安装脚本、运行脚本等,无需从 Git 拉取
- 自动创建 systemd 服务并配置自启动
- 自动配置防火墙规则(开放 2200 端口)
- 自动登记节点到后端服务器
@@ -337,8 +359,8 @@ BACKEND_URL=http://192.168.1.100:8080 ./run.sh start
版本号统一从 `version.json` 文件读取:
```json
{
"version": "1.1.0",
"tag": "v1.1.0"
"version": "1.1.3",
"tag": "v1.1.3"
}
```
@@ -390,6 +412,7 @@ BACKEND_URL=http://192.168.1.100:8080 ./run.sh start
**功能特性:**
-**自动从 `version.json` 读取版本号和标签**(无需手动指定)
-**Token 已硬编码**(无需手动指定)
-**自动打包所有必要文件**:二进制文件、安装脚本、运行脚本、配置文件等
- ✅ 自动打包二进制文件tar.gz 或 zip
- ✅ 自动创建发布说明
- ✅ 支持指定平台上传
@@ -414,8 +437,8 @@ BACKEND_URL=http://192.168.1.100:8080 ./run.sh start
版本号和标签统一从 `version.json` 文件读取:
```json
{
"version": "1.1.0",
"tag": "v1.1.0"
"version": "1.1.3",
"tag": "v1.1.3"
}
```
@@ -603,9 +626,67 @@ grep -i "error" node.log
tail -n 100 node.log
```
## 心跳机制
节点会定期向后端发送心跳,上报节点状态和获取节点信息。
### 心跳请求字段
心跳请求包含以下字段:
- `type`: 固定值 `pingServer`
- `version`: 协议版本号,固定值 `2`
- `host_name`: 节点主机名(自动读取系统主机名)
### 心跳响应
心跳响应包含以下节点信息:
- `node_id`: 节点ID
- `node_ip`: 节点外网IP
- `country`: 国家
- `province`: 省份
- `city`: 城市
- `isp`: ISP
这些信息会自动保存到配置文件中,用于后续的数据推送。
## 持续测试功能
节点支持持续 Ping 和 TCPing 测试,测试结果会自动推送到后端服务器。
### 功能特性
- ✅ 实时推送测试结果到后端
- ✅ 批量推送优化减少HTTP请求频率
- ✅ 自动清理超时任务
- ✅ 资源自动清理(防止内存泄漏)
- ✅ 详细的调试日志debug模式
### 数据推送
- 测试结果会自动推送到后端 `/api/public/node/continuous/result` 接口
- 推送包含节点ID、IP、位置信息和测试结果
- 如果后端任务不存在,节点端会自动停止对应任务
## 更新日志
### v1.1.0 (最新)
### v1.1.4 (最新)
**新增功能:**
- ✨ 心跳请求新增 `version` 字段协议版本号默认值2
- ✨ 心跳请求新增 `host_name` 字段(自动读取系统主机名)
- ✨ 支持环境变量 `BACKEND_URL` 覆盖配置文件中的后端地址
- ✨ 持续测试功能增强,支持批量推送和自动清理
**改进:**
- 🔧 修复持续测试数据推送的锁管理问题
- 🔧 修复任务停止时未清理推送缓冲的内存泄漏问题
- 🔧 优化配置加载逻辑,环境变量优先级最高
- 🔧 增强日志记录,添加详细的调试信息
- 📝 完善文档,添加配置优先级和心跳机制说明
### v1.1.3
**新增功能:**
- ✨ 添加日志文件输出功能,支持配置日志文件路径和级别

View File

@@ -19,6 +19,8 @@ BUILD_DIR="bin"
RELEASE_DIR="release"
TEMP_DIR=$(mktemp -d)
# Gitea Token (硬编码)
GITEA_TOKEN="3becb08eee31b422481ce1b8986de1cd645b468e"
@@ -277,16 +279,52 @@ pack_files() {
local pack_name="${PROJECT_NAME}-${os}-${arch}-${VERSION}"
local pack_file
# 创建临时打包目录
local pack_dir="${TEMP_DIR}/${pack_name}"
mkdir -p "$pack_dir"
# 复制二进制文件并重命名为 agent
if [ "$os" = "windows" ]; then
cp "$binary" "$pack_dir/agent.exe"
else
cp "$binary" "$pack_dir/agent"
chmod +x "$pack_dir/agent"
fi
# 复制必要的脚本文件
local scripts=("install.sh" "run.sh" "start-systemd.sh" "uninstall.sh")
for script in "${scripts[@]}"; do
if [ -f "$script" ]; then
cp "$script" "$pack_dir/"
chmod +x "$pack_dir/$script"
echo -e "${BLUE} [包含]${NC} $script"
else
echo -e "${YELLOW} [警告]${NC} $script 不存在,跳过"
fi
done
# 复制示例配置文件
if [ -f "config.yaml.example" ]; then
cp "config.yaml.example" "$pack_dir/config.yaml.example"
echo -e "${BLUE} [包含]${NC} config.yaml.example"
else
echo -e "${YELLOW} [警告]${NC} config.yaml.example 不存在,跳过"
fi
# 打包
if [ "$os" = "windows" ]; then
pack_file="${TEMP_DIR}/${pack_name}.zip"
echo -e "${BLUE}[打包]${NC} ${platform} -> ${pack_name}.zip"
(cd "$BUILD_DIR" && zip -q "${pack_file}" "$(basename $binary)")
(cd "$TEMP_DIR" && zip -q -r "${pack_file}" "$(basename $pack_dir)")
else
pack_file="${TEMP_DIR}/${pack_name}.tar.gz"
echo -e "${BLUE}[打包]${NC} ${platform} -> ${pack_name}.tar.gz"
tar -czf "$pack_file" -C "$BUILD_DIR" "$(basename $binary)"
tar -czf "$pack_file" -C "$TEMP_DIR" "$(basename $pack_dir)"
fi
# 清理临时目录
rm -rf "$pack_dir"
echo "$pack_file"
return 0
}

18
config.yaml.example Normal file
View File

@@ -0,0 +1,18 @@
server:
port: 2200
backend:
url: http://your-backend-server:8080
heartbeat:
interval: 60
log:
file: node.log
level: info
debug: false
node:
id: 0
ip: ""
country: ""
province: ""
city: ""
isp: ""

View File

@@ -8,6 +8,30 @@
set -e
# 错误处理函数
error_handler() {
local line_number=$1
local command=$2
echo ""
echo -e "${RED}========================================${NC}"
echo -e "${RED} 脚本执行出错!${NC}"
echo -e "${RED}========================================${NC}"
echo -e "${YELLOW}错误位置: 第 ${line_number}${NC}"
echo -e "${YELLOW}失败命令: ${command}${NC}"
echo ""
echo -e "${YELLOW}故障排查建议:${NC}"
echo " 1. 检查网络连接是否正常"
echo " 2. 检查后端地址是否正确: ${BACKEND_URL:-未设置}"
echo " 3. 检查是否有足够的磁盘空间和权限"
echo " 4. 查看上面的详细错误信息"
echo ""
echo -e "${YELLOW}查看服务日志: sudo journalctl -u ${SERVICE_NAME:-linkmaster-node} -n 50${NC}"
exit 1
}
# 设置错误陷阱
trap 'error_handler ${LINENO} "${BASH_COMMAND}"' ERR
# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
@@ -96,7 +120,6 @@ detect_fastest_mirror() {
# Ubuntu/Debian 镜像源列表
UBUNTU_MIRRORS=(
"mirrors.tuna.tsinghua.edu.cn"
"mirrors.huaweicloud.com"
"mirrors.163.com"
"archive.ubuntu.com"
@@ -104,7 +127,6 @@ detect_fastest_mirror() {
# CentOS/RHEL 镜像源列表
CENTOS_MIRRORS=(
"mirrors.tuna.tsinghua.edu.cn"
"mirrors.huaweicloud.com"
)
@@ -388,6 +410,119 @@ install_dependencies() {
echo -e "${GREEN}✓ 系统依赖安装完成${NC}"
}
# 配置时间同步
sync_time() {
echo -e "${BLUE}配置时间同步...${NC}"
# 1. 设置时区为 Asia/Shanghai
echo -e "${BLUE}[1/6] 设置时区为 Asia/Shanghai${NC}"
if command -v timedatectl > /dev/null 2>&1; then
sudo timedatectl set-timezone Asia/Shanghai 2>/dev/null || {
echo -e "${YELLOW}⚠ timedatectl 设置时区失败,尝试其他方法${NC}"
# 尝试创建时区链接(适用于较老的系统)
if [ -f /usr/share/zoneinfo/Asia/Shanghai ]; then
sudo ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime 2>/dev/null || true
fi
}
elif [ -f /usr/share/zoneinfo/Asia/Shanghai ]; then
sudo ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime 2>/dev/null || true
fi
# 2. 安装 chrony
echo -e "${BLUE}[2/6] 安装 chrony${NC}"
if [ "$OS" = "ubuntu" ] || [ "$OS" = "debian" ]; then
if ! dpkg -l 2>/dev/null | grep -q "^ii.*chrony"; then
sudo apt-get install -y chrony
else
echo -e "${BLUE}chrony 已安装,跳过${NC}"
fi
elif [ "$OS" = "centos" ] || [ "$OS" = "rhel" ] || [ "$OS" = "rocky" ] || [ "$OS" = "almalinux" ]; then
if ! rpm -q chrony &>/dev/null; then
sudo yum install -y chrony
else
echo -e "${BLUE}chrony 已安装,跳过${NC}"
fi
else
echo -e "${YELLOW}⚠ 未知系统类型,跳过 chrony 安装${NC}"
return 0
fi
# 3. 配置 NTP 服务器
echo -e "${BLUE}[3/6] 配置 NTP 服务器${NC}"
CONF="/etc/chrony.conf"
if [ -f "$CONF" ]; then
# 备份配置文件
if [ ! -f "${CONF}.backup.$(date +%Y%m%d)" ]; then
sudo cp "$CONF" "${CONF}.backup.$(date +%Y%m%d)" 2>/dev/null || true
fi
# 注释掉原有的 server 行
sudo sed -i 's/^server /#server /g' "$CONF" 2>/dev/null || true
# 添加中国 NTP 服务器(如果还没有)
if ! grep -q "ntp.aliyun.com" "$CONF"; then
sudo tee -a "$CONF" > /dev/null <<EOF
# China NTP servers (added by LinkMaster Node installer)
server ntp.aliyun.com iburst
server ntp.tencent.com iburst
server ntp1.aliyun.com iburst
EOF
fi
else
echo -e "${YELLOW}⚠ chrony.conf 不存在,跳过配置${NC}"
fi
# 4. 启动并启用 chronyd
echo -e "${BLUE}[4/6] 启动 chronyd${NC}"
if command -v systemctl > /dev/null 2>&1; then
sudo systemctl enable chronyd --now 2>/dev/null || {
# 如果 systemctl 失败,尝试使用 service 命令
if command -v service > /dev/null 2>&1; then
sudo service chronyd start 2>/dev/null || true
sudo chkconfig chronyd on 2>/dev/null || true
fi
}
elif command -v service > /dev/null 2>&1; then
sudo service chronyd start 2>/dev/null || true
sudo chkconfig chronyd on 2>/dev/null || true
fi
# 等待服务启动
sleep 2
# 5. 立即强制同步
echo -e "${BLUE}[5/6] 强制同步系统时间${NC}"
if command -v chronyc > /dev/null 2>&1; then
sudo chronyc -a makestep 2>/dev/null || {
# 如果 makestep 失败,尝试使用 sources 和 sourcestats
sudo chronyc sources 2>/dev/null || true
sudo chronyc sourcestats 2>/dev/null || true
}
fi
# 6. 写入硬件时间
echo -e "${BLUE}[6/6] 写入硬件时钟${NC}"
if command -v hwclock > /dev/null 2>&1; then
sudo hwclock --systohc 2>/dev/null || true
fi
# 显示时间状态
echo -e "${BLUE}当前时间状态:${NC}"
if command -v timedatectl > /dev/null 2>&1; then
sudo timedatectl status 2>/dev/null || true
else
date
if command -v hwclock > /dev/null 2>&1; then
echo -e "${BLUE}硬件时钟:${NC}"
sudo hwclock 2>/dev/null || true
fi
fi
echo -e "${GREEN}✓ 时间同步配置完成${NC}"
}
# 从官网下载安装 Go
install_go_from_official() {
echo -e "${BLUE}从 Go 官网下载安装...${NC}"
@@ -811,21 +946,20 @@ download_binary_from_releases() {
fi
# 解压文件
echo -e "${BLUE}解压二进制文件...${NC}"
echo -e "${BLUE}解压发布包...${NC}"
cd "$temp_dir"
local extracted_dir=""
if [ "$file_ext" = "tar.gz" ]; then
if ! tar -xzf "$download_file" 2>/dev/null; then
echo -e "${YELLOW}⚠ 解压失败,将使用源码编译${NC}"
rm -rf "$temp_dir"
return 1
fi
# 查找解压后的二进制文件(优先查找 agent然后是 agent-*
local binary_file=""
if [ -f "./agent" ] && [ -x "./agent" ]; then
binary_file="./agent"
else
binary_file=$(find . -maxdepth 1 -type f \( -name "agent" -o -name "agent-*" -o -name "Agent" -o -name "Agent-*" \) ! -name "*.tar.gz" ! -name "*.zip" 2>/dev/null | head -1)
# 查找解压后的目录(可能是直接解压到当前目录,也可能是在子目录中
extracted_dir=$(find . -maxdepth 1 -type d ! -name "." ! -name ".." | head -1)
if [ -z "$extracted_dir" ]; then
extracted_dir="."
fi
else
# Windows zip 文件(虽然脚本主要在 Linux 上运行,但保留兼容性)
@@ -834,26 +968,82 @@ download_binary_from_releases() {
rm -rf "$temp_dir"
return 1
fi
local binary_file=$(find . -maxdepth 1 -type f \( -name "agent*.exe" -o -name "Agent*.exe" \) 2>/dev/null | head -1)
extracted_dir=$(find . -maxdepth 1 -type d ! -name "." ! -name ".." | head -1)
if [ -z "$extracted_dir" ]; then
extracted_dir="."
fi
fi
cd "$extracted_dir" || {
echo -e "${YELLOW}⚠ 无法进入解压目录,将使用源码编译${NC}"
rm -rf "$temp_dir"
return 1
}
# 显示解压后的目录内容(用于调试)
echo -e "${BLUE}解压目录内容:${NC}"
ls -la . 2>/dev/null || true
echo ""
# 查找二进制文件(先检查当前目录,再递归查找)
local binary_file=""
if [ "$OS_TYPE" = "windows" ]; then
if [ -f "./agent.exe" ]; then
binary_file="./agent.exe"
elif [ -f "agent.exe" ]; then
binary_file="agent.exe"
else
# 递归查找所有 .exe 文件
binary_file=$(find . -type f -name "*.exe" ! -name "*.tar.gz" ! -name "*.zip" 2>/dev/null | grep -i agent | head -1)
fi
else
# Linux/macOS: 先检查常见位置
if [ -f "./agent" ]; then
binary_file="./agent"
elif [ -f "agent" ]; then
binary_file="agent"
else
# 递归查找所有文件,排除压缩包和目录
# 查找名为 agent 的文件(不是目录)
binary_file=$(find . -type f \( -name "agent" -o -name "agent-*" \) ! -name "*.tar.gz" ! -name "*.zip" 2>/dev/null | head -1)
# 如果还是找不到,尝试查找所有可执行文件
if [ -z "$binary_file" ]; then
binary_file=$(find . -type f -perm +111 ! -name "*.tar.gz" ! -name "*.zip" ! -name "*.sh" 2>/dev/null | head -1)
fi
fi
fi
if [ -z "$binary_file" ] || [ ! -f "$binary_file" ]; then
echo -e "${YELLOW}⚠ 未找到解压后的二进制文件,将使用源码编译${NC}"
echo -e "${YELLOW} 解压目录内容:${NC}"
ls -la "$temp_dir" 2>/dev/null || true
echo -e "${YELLOW} 当前目录: $(pwd)${NC}"
echo -e "${YELLOW} 查找的文件: agent 或 agent-*${NC}"
echo -e "${YELLOW} 所有文件列表:${NC}"
find . -type f 2>/dev/null | head -20 || true
rm -rf "$temp_dir"
return 1
fi
# 确保使用绝对路径
local binary_path=""
if [[ "$binary_file" == /* ]]; then
binary_path="$binary_file"
else
# 转换为绝对路径
binary_path="$(cd "$(dirname "$binary_file")" && pwd)/$(basename "$binary_file")"
fi
echo -e "${GREEN}✓ 找到二进制文件: ${binary_path}${NC}"
# 验证二进制文件是否可执行
if [ ! -x "$binary_file" ]; then
chmod +x "$binary_file" 2>/dev/null || true
if [ ! -x "$binary_path" ]; then
chmod +x "$binary_path" 2>/dev/null || true
fi
# 验证二进制文件类型Linux 应该是 ELF 文件)
if [ "$OS_TYPE" = "linux" ]; then
if command -v file > /dev/null 2>&1; then
local file_type=$(file "$binary_file" 2>/dev/null || echo "")
local file_type=$(file "$binary_path" 2>/dev/null || echo "")
if [ -n "$file_type" ] && ! echo "$file_type" | grep -qi "ELF"; then
echo -e "${YELLOW}⚠ 二进制文件类型异常: ${file_type}${NC}"
echo -e "${YELLOW} 将使用源码编译${NC}"
@@ -863,26 +1053,52 @@ download_binary_from_releases() {
fi
fi
# 保存下载的二进制文件到临时位置
local downloaded_binary="${temp_dir}/downloaded_agent"
sudo cp "$binary_file" "$downloaded_binary"
sudo chmod +x "$downloaded_binary"
# 保存当前目录extracted_dir
local extracted_path="$(pwd)"
# 验证复制后的文件
if [ ! -f "$downloaded_binary" ] || [ ! -x "$downloaded_binary" ]; then
echo -e "${YELLOW}⚠ 二进制文件验证失败,将使用源码编译${NC}"
rm -rf "$temp_dir"
return 1
# 检查是否是新格式的发布包(包含脚本文件
local has_scripts=false
if [ -f "$extracted_path/install.sh" ] || [ -f "$extracted_path/run.sh" ] || [ -f "$extracted_path/start-systemd.sh" ]; then
has_scripts=true
echo -e "${GREEN}✓ 检测到新格式发布包(包含脚本文件)${NC}"
fi
# 清理临时下载文件
rm -f "$download_file"
# 克隆仓库(用于获取 run.sh 和 start-systemd.sh 等脚本)
echo -e "${BLUE}克隆仓库以获取启动脚本...${NC}"
# 创建源码目录
if [ -d "$SOURCE_DIR" ]; then
sudo rm -rf "$SOURCE_DIR"
fi
sudo mkdir -p "$SOURCE_DIR"
if [ "$has_scripts" = true ]; then
# 新格式:从压缩包提取所有文件
echo -e "${BLUE}从发布包提取所有文件...${NC}"
# 复制二进制文件
sudo cp "$binary_path" "$SOURCE_DIR/agent"
sudo chmod +x "$SOURCE_DIR/agent"
echo -e "${GREEN}✓ 已提取二进制文件${NC}"
# 复制脚本文件
local scripts=("install.sh" "run.sh" "start-systemd.sh" "uninstall.sh")
for script in "${scripts[@]}"; do
if [ -f "$extracted_path/$script" ]; then
sudo cp "$extracted_path/$script" "$SOURCE_DIR/"
sudo chmod +x "$SOURCE_DIR/$script"
echo -e "${GREEN}✓ 已提取 $script${NC}"
fi
done
# 复制示例配置文件
if [ -f "$extracted_path/config.yaml.example" ]; then
sudo cp "$extracted_path/config.yaml.example" "$SOURCE_DIR/config.yaml.example"
echo -e "${GREEN}✓ 已提取 config.yaml.example${NC}"
fi
echo -e "${GREEN}✓ 所有文件已从发布包提取,无需克隆 Git 仓库${NC}"
else
# 旧格式:只有二进制文件,需要克隆 Git 仓库获取脚本
echo -e "${BLUE}检测到旧格式发布包(仅包含二进制文件)${NC}"
echo -e "${BLUE}克隆仓库以获取启动脚本...${NC}"
if ! sudo git clone --branch "${GITHUB_BRANCH}" "https://gitee.nas.cpolar.cn/${GITHUB_REPO}.git" "$SOURCE_DIR" 2>&1; then
echo -e "${YELLOW}⚠ 克隆仓库失败,将使用源码编译${NC}"
@@ -890,65 +1106,44 @@ download_binary_from_releases() {
return 1
fi
# 对比 Git commit hash
# 对比 Git commit hash(仅旧格式需要)
if [ -n "$release_commit" ]; then
echo -e "${BLUE}验证 Git commit 版本...${NC}"
# 切换到源码目录(如果存在)
if [ -d "$SOURCE_DIR" ] && [ -d "$SOURCE_DIR/.git" ]; then
if [ -d "$SOURCE_DIR/.git" ]; then
cd "$SOURCE_DIR" || {
echo -e "${YELLOW}⚠ 无法切换到源码目录,跳过验证${NC}"
cd /tmp || true
}
# 获取当前分支的最新 commit hash
local current_commit=$(git rev-parse HEAD 2>/dev/null || echo "")
if [ -n "$current_commit" ]; then
# 截取短 commit hash 用于显示前7位
local release_commit_short=""
if [ -n "$current_commit" ] && [ "${#release_commit}" -eq 40 ] && [ "${#current_commit}" -eq 40 ]; then
local release_commit_short=$(echo "$release_commit" | cut -c1-7)
local current_commit_short=$(echo "$current_commit" | cut -c1-7)
if [ "${#release_commit}" -eq 40 ]; then
release_commit_short=$(echo "$release_commit" | cut -c1-7)
else
release_commit_short="$release_commit"
fi
echo -e "${BLUE} Release commit: ${release_commit_short}${NC}"
echo -e "${BLUE} 当前代码 commit: ${current_commit_short}${NC}"
# 对比 commit hash只有当 release_commit 是完整的 commit hash 时才对比)
if [ "${#release_commit}" -eq 40 ] && [ "${#current_commit}" -eq 40 ]; then
if [ "$release_commit" != "$current_commit" ]; then
echo -e "${YELLOW}⚠ Commit hash 不匹配,二进制文件可能不是最新代码编译的${NC}"
echo -e "${YELLOW} Release 基于较旧的代码,将使用源码编译最新版本${NC}"
# 保留已克隆的仓库目录,供 build_from_source 复用
cd /tmp || true
rm -rf "$temp_dir"
return 1
else
echo -e "${GREEN}✓ Commit hash 匹配,二进制文件是最新代码编译的${NC}"
echo -e "${GREEN}✓ Commit hash 匹配${NC}"
fi
else
echo -e "${YELLOW}⚠ 无法获取有效的 commit hash 进行对比,跳过验证${NC}"
fi
else
echo -e "${YELLOW}⚠ 无法获取当前代码的 commit hash跳过验证${NC}"
fi
# 返回临时目录
cd /tmp || true
else
echo -e "${YELLOW}⚠ 源码目录不存在,跳过验证${NC}"
fi
else
echo -e "${YELLOW}⚠ 无法获取 release 的 commit hash跳过验证${NC}"
fi
# 用下载的二进制文件覆盖克隆目录中的文件
sudo cp "$downloaded_binary" "$SOURCE_DIR/agent"
sudo cp "$binary_path" "$SOURCE_DIR/agent"
sudo chmod +x "$SOURCE_DIR/agent"
fi
# 复制到安装目录
sudo mkdir -p "$INSTALL_DIR"
@@ -960,9 +1155,9 @@ download_binary_from_releases() {
# 显示文件信息
local binary_size=$(du -h "$SOURCE_DIR/agent" | cut -f1)
echo -e "${GREEN}二进制文件下载完成 (文件大小: ${binary_size})${NC}"
echo -e "${GREEN}安装文件准备完成 (文件大小: ${binary_size})${NC}"
echo -e "${BLUE}版本: ${latest_tag}${NC}"
echo -e "${BLUE}二进制文件: ${SOURCE_DIR}/agent${NC}"
echo -e "${BLUE}安装目录: ${SOURCE_DIR}${NC}"
return 0
}
@@ -1264,11 +1459,18 @@ EOF
# 调用心跳API获取节点信息
echo -e "${BLUE}发送心跳请求获取节点信息...${NC}"
RESPONSE=$(curl -s -X POST "${BACKEND_URL}/api/node/heartbeat" \
echo -e "${BLUE}后端地址: ${BACKEND_URL}${NC}"
# 添加超时设置,避免长时间卡住
# 使用 set +e 临时禁用错误退出,因为心跳失败不应该阻止安装
set +e
RESPONSE=$(curl -s --connect-timeout 10 --max-time 30 -X POST "${BACKEND_URL}/api/node/heartbeat" \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "type=pingServer" 2>&1)
CURL_EXIT_CODE=$?
set -e # 重新启用错误退出
if [ $? -eq 0 ]; then
if [ $CURL_EXIT_CODE -eq 0 ]; then
# 尝试解析JSON响应
NODE_ID=$(echo "$RESPONSE" | grep -o '"node_id":[0-9]*' | grep -o '[0-9]*' | head -1)
NODE_IP=$(echo "$RESPONSE" | grep -o '"node_ip":"[^"]*"' | cut -d'"' -f4 | head -1)
@@ -1306,8 +1508,10 @@ EOF
echo -e "${YELLOW} 响应: ${RESPONSE}${NC}"
fi
else
echo -e "${YELLOW}⚠ 心跳请求失败,将在服务启动时重试${NC}"
echo -e "${YELLOW} 错误: ${RESPONSE}${NC}"
echo -e "${YELLOW}⚠ 心跳请求失败 (退出码: ${CURL_EXIT_CODE}),将在服务启动时重试${NC}"
echo -e "${YELLOW} 错误信息: ${RESPONSE}${NC}"
echo -e "${YELLOW} 提示: 请检查后端地址是否正确: ${BACKEND_URL}${NC}"
echo -e "${YELLOW} 测试连接: curl -v ${BACKEND_URL}/api/public/nodes/online${NC}"
fi
# 设置配置文件权限
@@ -1318,18 +1522,52 @@ EOF
start_service() {
echo -e "${BLUE}启动服务...${NC}"
sudo systemctl enable ${SERVICE_NAME} > /dev/null 2>&1
sudo systemctl restart ${SERVICE_NAME}
# 先检查服务文件是否存在
if [ ! -f "/etc/systemd/system/${SERVICE_NAME}.service" ]; then
echo -e "${RED}✗ 错误: 服务文件不存在${NC}"
echo -e "${YELLOW} 路径: /etc/systemd/system/${SERVICE_NAME}.service${NC}"
exit 1
fi
# 检查二进制文件是否存在
if [ ! -f "$SOURCE_DIR/agent" ] && [ ! -f "$INSTALL_DIR/$BINARY_NAME" ]; then
echo -e "${RED}✗ 错误: 二进制文件不存在${NC}"
echo -e "${YELLOW} 检查路径: $SOURCE_DIR/agent 或 $INSTALL_DIR/$BINARY_NAME${NC}"
exit 1
fi
# 启用服务(显示输出以便调试)
echo -e "${BLUE}启用服务...${NC}"
if ! sudo systemctl enable ${SERVICE_NAME} 2>&1; then
echo -e "${RED}✗ 启用服务失败${NC}"
exit 1
fi
# 重新加载 systemd
echo -e "${BLUE}重新加载 systemd...${NC}"
sudo systemctl daemon-reload
# 启动服务(显示输出以便调试)
echo -e "${BLUE}启动服务...${NC}"
if ! sudo systemctl restart ${SERVICE_NAME} 2>&1; then
echo -e "${RED}✗ 启动服务失败${NC}"
echo -e "${YELLOW}查看详细日志: sudo journalctl -u ${SERVICE_NAME} -n 100 --no-pager${NC}"
echo -e "${YELLOW}查看服务状态: sudo systemctl status ${SERVICE_NAME}${NC}"
exit 1
fi
# 等待服务启动
echo -e "${BLUE}等待服务启动...${NC}"
sleep 3
# 检查服务状态
if sudo systemctl is-active --quiet ${SERVICE_NAME}; then
if sudo systemctl is-active --quiet ${SERVICE_NAME} 2>/dev/null; then
echo -e "${GREEN}✓ 服务启动成功${NC}"
else
echo -e "${RED}✗ 服务启动失败${NC}"
echo -e "${YELLOW}查看日志: sudo journalctl -u ${SERVICE_NAME} -n 50${NC}"
echo -e "${YELLOW}服务状态:${NC}"
sudo systemctl status ${SERVICE_NAME} --no-pager -l || true
echo -e "${YELLOW}查看详细日志: sudo journalctl -u ${SERVICE_NAME} -n 100 --no-pager${NC}"
exit 1
fi
}
@@ -1387,20 +1625,32 @@ main() {
echo -e "${GREEN} LinkMaster 节点端安装程序${NC}"
echo -e "${GREEN}========================================${NC}"
echo ""
echo -e "${BLUE}后端地址: ${BACKEND_URL}${NC}"
echo ""
echo -e "${BLUE}[1/11] 检测系统类型...${NC}"
detect_system
# 检查是否已安装,如果已安装则先卸载
if check_installed; then
echo -e "${BLUE}[2/11] 卸载已存在的服务...${NC}"
uninstall_service
else
echo -e "${BLUE}[2/11] 检查已安装服务...${NC}"
echo -e "${GREEN}✓ 未检测到已安装的服务${NC}"
fi
# 检测并配置最快的镜像源(在安装依赖之前)
echo -e "${BLUE}[3/11] 检测并配置镜像源...${NC}"
detect_fastest_mirror
echo -e "${BLUE}[4/11] 安装系统依赖...${NC}"
install_dependencies
echo -e "${BLUE}[5/11] 配置时间同步...${NC}"
sync_time
# 优先尝试从 Releases 下载二进制文件
echo -e "${BLUE}[6/11] 下载或编译二进制文件...${NC}"
if ! download_binary_from_releases; then
echo -e "${BLUE}从 Releases 下载失败,开始从源码编译...${NC}"
build_from_source
@@ -1408,10 +1658,19 @@ main() {
echo -e "${GREEN}✓ 使用预编译二进制文件,跳过编译步骤${NC}"
fi
echo -e "${BLUE}[7/11] 创建 systemd 服务...${NC}"
create_service
echo -e "${BLUE}[8/11] 配置防火墙规则...${NC}"
configure_firewall
echo -e "${BLUE}[9/11] 登记节点到后端服务器...${NC}"
register_node
echo -e "${BLUE}[10/11] 启动服务...${NC}"
start_service
echo -e "${BLUE}[11/11] 验证安装...${NC}"
verify_installation
echo ""

View File

@@ -72,6 +72,12 @@ func Load() (*Config, error) {
}
}
// 环境变量优先级最高,覆盖配置文件中的设置
// 支持 BACKEND_URL 环境变量覆盖后端地址
if backendURL := os.Getenv("BACKEND_URL"); backendURL != "" {
cfg.Backend.URL = backendURL
}
// 如果配置文件中没有设置日志文件,使用环境变量或默认值
if cfg.Log.File == "" {
logFile := os.Getenv("LOG_FILE")

View File

@@ -28,14 +28,47 @@ type TCPingTask struct {
}
func NewTCPingTask(taskID, target string, interval, maxDuration time.Duration) (*TCPingTask, error) {
// 解析host:port
parts := strings.Split(target, ":")
if len(parts) != 2 {
return nil, fmt.Errorf("无效的target格式需要 host:port")
// 解析host:port如果没有端口则默认80
var host string
var portStr string
var port int
// 检查是否是IPv6格式如 [::1]:8080
if strings.HasPrefix(target, "[") {
// IPv6格式 - 使用 Index 而不是 LastIndex 来找到第一个闭合括号
closeBracket := strings.Index(target, "]")
if closeBracket == -1 {
return nil, fmt.Errorf("无效的target格式IPv6地址格式应为 [host]:port")
}
host = target[1:closeBracket]
if closeBracket+1 < len(target) && target[closeBracket+1] == ':' {
portStr = target[closeBracket+2:]
// 如果端口部分为空使用默认端口80修复 Bug 1
if portStr == "" {
portStr = "80"
}
} else {
portStr = "80" // 默认端口
}
} else {
// 普通格式 host:port 或 host
lastColonIndex := strings.LastIndex(target, ":")
if lastColonIndex == -1 {
// 没有冒号使用默认端口80
host = target
portStr = "80"
} else {
host = target[:lastColonIndex]
portStr = target[lastColonIndex+1:]
// 如果端口部分为空使用默认端口80
if portStr == "" {
portStr = "80"
}
}
}
host := parts[0]
port, err := strconv.Atoi(parts[1])
var err error
port, err = strconv.Atoi(portStr)
if err != nil {
return nil, fmt.Errorf("无效的端口: %v", err)
}
@@ -185,4 +218,3 @@ func (t *TCPingTask) executeTCPing() map[string]interface{} {
"ip": targetIP,
}
}

View File

@@ -8,6 +8,7 @@ import (
"io"
"net"
"net/http"
"os"
"strings"
"sync"
"time"
@@ -18,6 +19,7 @@ import (
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)
var continuousTasks = make(map[string]*ContinuousTask)
@@ -46,7 +48,52 @@ const (
func InitContinuousHandler(cfg *config.Config) {
backendURL = cfg.Backend.URL
logger, _ = zap.NewProduction()
// 根据配置创建logger
var level zapcore.Level
logLevel := cfg.Log.Level
if logLevel == "" {
if cfg.Debug {
logLevel = "debug"
} else {
logLevel = "info"
}
}
switch logLevel {
case "debug":
level = zapcore.DebugLevel
case "info":
level = zapcore.InfoLevel
case "warn":
level = zapcore.WarnLevel
case "error":
level = zapcore.ErrorLevel
default:
level = zapcore.InfoLevel
}
// 创建编码器配置
encoderConfig := zap.NewProductionEncoderConfig()
if cfg.Debug {
encoderConfig = zap.NewDevelopmentEncoderConfig()
}
encoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder
// 创建核心 - 输出到标准错误日志文件由main.go统一管理这里输出到stderr便于调试
core := zapcore.NewCore(
zapcore.NewJSONEncoder(encoderConfig),
zapcore.AddSync(os.Stderr),
level,
)
// 创建logger
logger = zap.New(core, zap.AddCaller(), zap.AddStacktrace(zapcore.ErrorLevel))
logger.Info("持续测试处理器已初始化",
zap.String("backend_url", backendURL),
zap.String("log_level", logLevel))
}
type ContinuousTask struct {
@@ -160,7 +207,15 @@ func HandleContinuousStop(c *gin.Context) {
if task.tcpingTask != nil {
task.tcpingTask.Stop()
}
// 关闭停止通道
select {
case <-task.StopCh:
// 已经关闭
default:
close(task.StopCh)
}
delete(continuousTasks, req.TaskID)
}
taskMutex.Unlock()
@@ -170,6 +225,17 @@ func HandleContinuousStop(c *gin.Context) {
return
}
// 清理推送缓冲
bufferMutex.Lock()
if buffer, exists := pushBuffers[req.TaskID]; exists {
if buffer.pushTimer != nil {
buffer.pushTimer.Stop()
}
delete(pushBuffers, req.TaskID)
logger.Debug("已清理任务推送缓冲", zap.String("task_id", req.TaskID))
}
bufferMutex.Unlock()
c.JSON(http.StatusOK, gin.H{"message": "任务已停止"})
}
@@ -237,7 +303,8 @@ func pushResultToBackend(taskID string, result map[string]interface{}) {
logger.Warn("节点ID未获取跳过推送结果",
zap.String("task_id", taskID),
zap.String("node_ip", nodeIP),
zap.String("hint", "等待心跳返回node_id后再推送"))
zap.String("hint", "等待心跳返回node_id后再推送"),
zap.Any("result", result))
return
}
@@ -246,10 +313,18 @@ func pushResultToBackend(taskID string, result map[string]interface{}) {
logger.Warn("节点IP未获取跳过推送结果",
zap.String("task_id", taskID),
zap.Uint("node_id", nodeID),
zap.String("hint", "等待心跳返回node_ip后再推送"))
zap.String("hint", "等待心跳返回node_ip后再推送"),
zap.Any("result", result))
return
}
// 记录调试信息
logger.Debug("准备推送结果到后端",
zap.String("task_id", taskID),
zap.Uint("node_id", nodeID),
zap.String("node_ip", nodeIP),
zap.Any("result", result))
// 添加到批量推送缓冲
addToPushBuffer(taskID, nodeID, nodeIP, result)
}
@@ -269,21 +344,34 @@ func addToPushBuffer(taskID string, nodeID uint, nodeIP string, result map[strin
bufferMutex.Unlock()
buffer.mu.Lock()
defer buffer.mu.Unlock()
// 添加结果到缓冲
buffer.results = append(buffer.results, result)
// 如果缓冲已满,立即推送
shouldFlush := len(buffer.results) >= batchPushMaxSize
buffer.mu.Unlock()
if shouldFlush {
flushPushBuffer(taskID, nodeID, nodeIP)
return
// 复制结果列表
results := make([]map[string]interface{}, len(buffer.results))
copy(results, buffer.results)
buffer.results = buffer.results[:0] // 清空缓冲
// 停止定时器
if buffer.pushTimer != nil {
buffer.pushTimer.Stop()
buffer.pushTimer = nil
}
buffer.mu.Lock()
buffer.lastPush = time.Now()
buffer.mu.Unlock()
// 批量推送结果
for _, r := range results {
pushSingleResult(taskID, nodeID, nodeIP, r)
}
return
}
// 如果距离上次推送超过间隔时间,启动定时器推送
if buffer.pushTimer == nil {
@@ -291,6 +379,8 @@ func addToPushBuffer(taskID string, nodeID uint, nodeIP string, result map[strin
flushPushBuffer(taskID, nodeID, nodeIP)
})
}
buffer.mu.Unlock()
}
// flushPushBuffer 刷新并推送缓冲中的结果
@@ -362,13 +452,21 @@ func pushSingleResult(taskID string, nodeID uint, nodeIP string, result map[stri
jsonData, err := json.Marshal(data)
if err != nil {
logger.Error("序列化结果失败", zap.Error(err), zap.String("task_id", taskID))
logger.Error("序列化结果失败",
zap.Error(err),
zap.String("task_id", taskID),
zap.Uint("node_id", nodeID),
zap.String("node_ip", nodeIP),
zap.Any("data", data))
return
}
req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
if err != nil {
logger.Error("创建请求失败", zap.Error(err), zap.String("task_id", taskID))
logger.Error("创建请求失败",
zap.Error(err),
zap.String("task_id", taskID),
zap.String("url", url))
return
}
@@ -380,7 +478,9 @@ func pushSingleResult(taskID string, nodeID uint, nodeIP string, result map[stri
logger.Warn("推送结果失败,继续运行",
zap.Error(err),
zap.String("task_id", taskID),
zap.String("url", url))
zap.String("url", url),
zap.Uint("node_id", nodeID),
zap.String("node_ip", nodeIP))
// 推送失败不停止任务,继续运行
return
}
@@ -394,7 +494,9 @@ func pushSingleResult(taskID string, nodeID uint, nodeIP string, result map[stri
if containsTaskNotFoundError(bodyStr) {
logger.Warn("后端任务不存在,停止节点端任务",
zap.String("task_id", taskID),
zap.String("response", bodyStr))
zap.String("response", bodyStr),
zap.Uint("node_id", nodeID),
zap.String("node_ip", nodeIP))
// 停止对应的持续测试任务
stopTaskByTaskID(taskID)
return
@@ -404,12 +506,18 @@ func pushSingleResult(taskID string, nodeID uint, nodeIP string, result map[stri
zap.Int("status", resp.StatusCode),
zap.String("task_id", taskID),
zap.String("url", url),
zap.String("response", bodyStr))
zap.String("response", bodyStr),
zap.Uint("node_id", nodeID),
zap.String("node_ip", nodeIP))
// 其他错误不停止任务,继续运行
return
}
logger.Debug("推送结果成功", zap.String("task_id", taskID))
logger.Debug("推送结果成功",
zap.String("task_id", taskID),
zap.Uint("node_id", nodeID),
zap.String("node_ip", nodeIP),
zap.Any("result", result))
}
// containsTaskNotFoundError 检查响应中是否包含任务不存在的错误
@@ -522,23 +630,20 @@ func StartTaskCleanup() {
for range ticker.C {
now := time.Now()
taskMutex.Lock()
var tasksToDelete []string
for taskID, task := range continuousTasks {
shouldDelete := false
// 检查最大运行时长
if now.Sub(task.StartTime) > task.MaxDuration {
logger.Info("任务达到最大运行时长,自动停止", zap.String("task_id", taskID))
task.IsRunning = false
if task.pingTask != nil {
task.pingTask.Stop()
}
if task.tcpingTask != nil {
task.tcpingTask.Stop()
}
delete(continuousTasks, taskID)
continue
}
shouldDelete = true
} else if now.Sub(task.LastRequest) > 30*time.Minute {
// 检查无客户端连接30分钟无请求
if now.Sub(task.LastRequest) > 30*time.Minute {
logger.Info("任务无客户端连接,自动停止", zap.String("task_id", taskID))
shouldDelete = true
}
if shouldDelete {
task.IsRunning = false
if task.pingTask != nil {
task.pingTask.Stop()
@@ -546,10 +651,41 @@ func StartTaskCleanup() {
if task.tcpingTask != nil {
task.tcpingTask.Stop()
}
delete(continuousTasks, taskID)
// 关闭停止通道
select {
case <-task.StopCh:
// 已经关闭
default:
close(task.StopCh)
}
tasksToDelete = append(tasksToDelete, taskID)
}
}
taskMutex.Unlock()
// 清理任务和推送缓冲
if len(tasksToDelete) > 0 {
taskMutex.Lock()
for _, taskID := range tasksToDelete {
delete(continuousTasks, taskID)
}
taskMutex.Unlock()
// 清理推送缓冲
bufferMutex.Lock()
for _, taskID := range tasksToDelete {
if buffer, exists := pushBuffers[taskID]; exists {
if buffer.pushTimer != nil {
buffer.pushTimer.Stop()
}
delete(pushBuffers, taskID)
logger.Debug("已清理任务推送缓冲", zap.String("task_id", taskID))
}
}
bufferMutex.Unlock()
}
}
}()
}

View File

@@ -114,7 +114,30 @@ func handleGet(c *gin.Context, urlStr string, params map[string]interface{}) {
seq = seqVal
}
// 解析URL
// 清理URL:去除多余的空格和重复的协议前缀
urlStr = strings.TrimSpace(urlStr)
// 如果URL中包含多个 http:// 或 https://,只保留第一个
if strings.Contains(urlStr, "http://") {
// 找到第一个 http:// 的位置
firstHttp := strings.Index(urlStr, "http://")
if firstHttp > 0 {
// 如果 http:// 不在开头,说明前面有内容,需要清理
urlStr = urlStr[firstHttp:]
}
// 移除后续重复的 http://
urlStr = strings.Replace(urlStr, "http://http://", "http://", -1)
urlStr = strings.Replace(urlStr, "http://https://", "https://", -1)
}
if strings.Contains(urlStr, "https://") {
firstHttps := strings.Index(urlStr, "https://")
if firstHttps > 0 {
urlStr = urlStr[firstHttps:]
}
urlStr = strings.Replace(urlStr, "https://https://", "https://", -1)
urlStr = strings.Replace(urlStr, "https://http://", "http://", -1)
}
// 如果URL没有协议前缀添加 http://
if !strings.HasPrefix(urlStr, "http://") && !strings.HasPrefix(urlStr, "https://") {
urlStr = "http://" + urlStr
}
@@ -145,11 +168,8 @@ func handleGet(c *gin.Context, urlStr string, params map[string]interface{}) {
Transport: timingTransport,
Timeout: 15 * time.Second,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
// 跟随重定向,最多20次
if len(via) >= 20 {
return fmt.Errorf("重定向次数过多")
}
return nil
// 跟随重定向,返回第一个状态码和 header
return http.ErrUseLastResponse
},
}
@@ -181,8 +201,11 @@ func handleGet(c *gin.Context, urlStr string, params map[string]interface{}) {
// 执行请求
startTime := time.Now()
resp, err := client.Do(req)
if err != nil {
// 错误处理
// 处理重定向错误:当 CheckRedirect 返回 ErrUseLastResponse 时,
// client.Do 会返回响应和错误,但响应仍然有效(包含重定向状态码和 header
if err != nil && resp == nil {
// 真正的错误,没有响应
errMsg := err.Error()
if strings.Contains(errMsg, "no such host") {
result["ip"] = "域名无法解析"
@@ -194,6 +217,7 @@ func handleGet(c *gin.Context, urlStr string, params map[string]interface{}) {
result["ip"] = "访问失败"
}
result["error"] = errMsg
result["statuscode"] = 0
result["totaltime"] = "*"
result["downtime"] = "*"
result["downsize"] = "*"
@@ -204,7 +228,25 @@ func handleGet(c *gin.Context, urlStr string, params map[string]interface{}) {
c.JSON(200, result)
return
}
// 如果有响应(包括重定向响应),继续处理
if resp != nil {
defer resp.Body.Close()
} else {
// 没有响应也没有错误,不应该发生
result["error"] = "未知错误"
result["ip"] = "访问失败"
result["statuscode"] = 0
result["totaltime"] = "*"
result["downtime"] = "*"
result["downsize"] = "*"
result["downspeed"] = "*"
result["firstbytetime"] = "*"
result["conntime"] = "*"
result["size"] = "*"
c.JSON(200, result)
return
}
// 获取时间信息
timingTransport.mu.Lock()
@@ -237,7 +279,7 @@ func handleGet(c *gin.Context, urlStr string, params map[string]interface{}) {
bodyReader := io.LimitReader(resp.Body, 1024*1024) // 限制1MB
bodyStartTime := time.Now()
body, err := io.ReadAll(bodyReader)
bodyReadTime := time.Now().Sub(bodyStartTime)
bodyReadTime := time.Since(bodyStartTime)
if err != nil && err != io.EOF {
result["error"] = err.Error()
}
@@ -296,7 +338,30 @@ func handlePost(c *gin.Context, urlStr string, params map[string]interface{}) {
seq = seqVal
}
// 解析URL
// 清理URL:去除多余的空格和重复的协议前缀
urlStr = strings.TrimSpace(urlStr)
// 如果URL中包含多个 http:// 或 https://,只保留第一个
if strings.Contains(urlStr, "http://") {
// 找到第一个 http:// 的位置
firstHttp := strings.Index(urlStr, "http://")
if firstHttp > 0 {
// 如果 http:// 不在开头,说明前面有内容,需要清理
urlStr = urlStr[firstHttp:]
}
// 移除后续重复的 http://
urlStr = strings.Replace(urlStr, "http://http://", "http://", -1)
urlStr = strings.Replace(urlStr, "http://https://", "https://", -1)
}
if strings.Contains(urlStr, "https://") {
firstHttps := strings.Index(urlStr, "https://")
if firstHttps > 0 {
urlStr = urlStr[firstHttps:]
}
urlStr = strings.Replace(urlStr, "https://https://", "https://", -1)
urlStr = strings.Replace(urlStr, "https://http://", "http://", -1)
}
// 如果URL没有协议前缀添加 http://
if !strings.HasPrefix(urlStr, "http://") && !strings.HasPrefix(urlStr, "https://") {
urlStr = "http://" + urlStr
}
@@ -333,10 +398,8 @@ func handlePost(c *gin.Context, urlStr string, params map[string]interface{}) {
Transport: timingTransport,
Timeout: 15 * time.Second,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
if len(via) >= 20 {
return fmt.Errorf("重定向次数过多")
}
return nil
// 不跟随重定向,返回第一个状态码和 header
return http.ErrUseLastResponse
},
}
@@ -363,7 +426,11 @@ func handlePost(c *gin.Context, urlStr string, params map[string]interface{}) {
// 执行请求
startTime := time.Now()
resp, err := client.Do(req)
if err != nil {
// 处理重定向错误:当 CheckRedirect 返回 ErrUseLastResponse 时,
// client.Do 会返回响应和错误,但响应仍然有效(包含重定向状态码和 header
if err != nil && resp == nil {
// 真正的错误,没有响应
errMsg := err.Error()
if strings.Contains(errMsg, "no such host") {
result["ip"] = "域名无法解析"
@@ -375,6 +442,7 @@ func handlePost(c *gin.Context, urlStr string, params map[string]interface{}) {
result["ip"] = "访问失败"
}
result["error"] = errMsg
result["statuscode"] = 0
result["totaltime"] = "*"
result["downtime"] = "*"
result["downsize"] = "*"
@@ -385,7 +453,25 @@ func handlePost(c *gin.Context, urlStr string, params map[string]interface{}) {
c.JSON(200, result)
return
}
// 如果有响应(包括重定向响应),继续处理
if resp != nil {
defer resp.Body.Close()
} else {
// 没有响应也没有错误,不应该发生
result["error"] = "未知错误"
result["ip"] = "访问失败"
result["statuscode"] = 0
result["totaltime"] = "*"
result["downtime"] = "*"
result["downsize"] = "*"
result["downspeed"] = "*"
result["firstbytetime"] = "*"
result["conntime"] = "*"
result["size"] = "*"
c.JSON(200, result)
return
}
// 获取时间信息
timingTransport.mu.Lock()

View File

@@ -16,21 +16,53 @@ func handleTCPing(c *gin.Context, url string, params map[string]interface{}) {
seq = seqVal
}
// 解析host:port格式
parts := strings.Split(url, ":")
if len(parts) != 2 {
// 解析host:port格式如果没有端口则默认80
var host string
var portStr string
var port int
// 检查是否是IPv6格式如 [::1]:8080
if strings.HasPrefix(url, "[") {
// IPv6格式 - 使用 Index 而不是 LastIndex 来找到第一个闭合括号
closeBracket := strings.Index(url, "]")
if closeBracket == -1 {
c.JSON(200, gin.H{
"seq": seq,
"type": "ceTCPing",
"url": url,
"error": "格式错误,需要 host:port",
"error": "格式错误,IPv6地址格式应为 [host]:port",
})
return
}
host = url[1:closeBracket]
if closeBracket+1 < len(url) && url[closeBracket+1] == ':' {
portStr = url[closeBracket+2:]
// 如果端口部分为空使用默认端口80修复 Bug 1
if portStr == "" {
portStr = "80"
}
} else {
portStr = "80" // 默认端口
}
} else {
// 普通格式 host:port 或 host
lastColonIndex := strings.LastIndex(url, ":")
if lastColonIndex == -1 {
// 没有冒号使用默认端口80
host = url
portStr = "80"
} else {
host = url[:lastColonIndex]
portStr = url[lastColonIndex+1:]
// 如果端口部分为空使用默认端口80
if portStr == "" {
portStr = "80"
}
}
}
host := parts[0]
portStr := parts[1]
port, err := strconv.Atoi(portStr)
var err error
port, err = strconv.Atoi(portStr)
if err != nil {
c.JSON(200, gin.H{
"seq": seq,
@@ -160,4 +192,3 @@ func handleTCPing(c *gin.Context, url string, params map[string]interface{}) {
c.JSON(200, result)
}

View File

@@ -7,6 +7,8 @@ import (
"fmt"
"io"
"net/http"
"net/url"
"os"
"sync"
"time"
@@ -88,21 +90,31 @@ func NewReporter(cfg *config.Config) *Reporter {
}
func (r *Reporter) Start(ctx context.Context) {
ticker := time.NewTicker(time.Duration(r.cfg.Heartbeat.Interval) * time.Second)
defer ticker.Stop()
// 立即发送一次心跳
r.sendHeartbeat()
for {
// 计算到下一分钟第1秒的时间
now := time.Now()
nextMinute := now.Truncate(time.Minute).Add(time.Minute)
nextHeartbeatTime := nextMinute.Add(1 * time.Second)
durationUntilNext := nextHeartbeatTime.Sub(now)
// 等待到下一分钟的第1秒
timer := time.NewTimer(durationUntilNext)
select {
case <-ctx.Done():
timer.Stop()
return
case <-r.stopCh:
timer.Stop()
return
case <-ticker.C:
case <-timer.C:
// 在每分钟的第1秒发送心跳
r.sendHeartbeat()
}
timer.Stop()
}
}
@@ -110,10 +122,25 @@ func (r *Reporter) Stop() {
close(r.stopCh)
}
// buildHeartbeatBody 构建心跳请求体
func buildHeartbeatBody() string {
hostname, err := os.Hostname()
if err != nil {
hostname = "unknown"
}
values := url.Values{}
values.Set("type", "pingServer")
values.Set("version", "2")
values.Set("host_name", hostname)
return values.Encode()
}
// RegisterNode 注册节点(安装时或首次启动时调用)
func RegisterNode(cfg *config.Config) error {
url := fmt.Sprintf("%s/api/node/heartbeat", cfg.Backend.URL)
req, err := http.NewRequest("POST", url, bytes.NewBufferString("type=pingServer"))
req, err := http.NewRequest("POST", url, bytes.NewBufferString(buildHeartbeatBody()))
if err != nil {
return fmt.Errorf("创建心跳请求失败: %w", err)
}
@@ -123,7 +150,7 @@ func RegisterNode(cfg *config.Config) error {
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("发送心跳失败: %w", err)
return fmt.Errorf("发送心跳失败 (URL: %s): %w", url, err)
}
defer resp.Body.Close()
@@ -173,16 +200,27 @@ func RegisterNode(cfg *config.Config) error {
return nil
}
}
return fmt.Errorf("心跳响应格式无效或节点信息不完整")
return fmt.Errorf("心跳响应格式无效或节点信息不完整 (响应体: %s)", string(body))
}
return fmt.Errorf("心跳请求失败,状态码: %d", resp.StatusCode)
// 读取响应体以便记录错误详情
body, err := io.ReadAll(resp.Body)
bodyStr := ""
if err == nil && len(body) > 0 {
// 限制响应体长度,避免错误信息过长
if len(body) > 500 {
bodyStr = string(body[:500]) + "..."
} else {
bodyStr = string(body)
}
}
return fmt.Errorf("心跳请求失败,状态码: %d, URL: %s, 响应体: %s", resp.StatusCode, url, bodyStr)
}
func (r *Reporter) sendHeartbeat() {
// 发送心跳使用Form格式兼容旧接口
url := fmt.Sprintf("%s/api/node/heartbeat", r.cfg.Backend.URL)
req, err := http.NewRequest("POST", url, bytes.NewBufferString("type=pingServer"))
req, err := http.NewRequest("POST", url, bytes.NewBufferString(buildHeartbeatBody()))
if err != nil {
r.logger.Error("创建心跳请求失败", zap.Error(err))
return
@@ -192,7 +230,9 @@ func (r *Reporter) sendHeartbeat() {
resp, err := r.client.Do(req)
if err != nil {
r.logger.Warn("发送心跳失败", zap.Error(err))
r.logger.Warn("发送心跳失败",
zap.String("url", url),
zap.Error(err))
return
}
defer resp.Body.Close()
@@ -260,7 +300,21 @@ func (r *Reporter) sendHeartbeat() {
}
r.logger.Debug("心跳发送成功")
} else {
r.logger.Warn("心跳发送失败", zap.Int("status", resp.StatusCode))
// 读取响应体以便记录错误详情
body, err := io.ReadAll(resp.Body)
bodyStr := ""
if err == nil && len(body) > 0 {
// 限制响应体长度,避免日志过长
if len(body) > 500 {
bodyStr = string(body[:500]) + "..."
} else {
bodyStr = string(body)
}
}
r.logger.Warn("心跳发送失败",
zap.Int("status", resp.StatusCode),
zap.String("url", url),
zap.String("response_body", bodyStr))
}
}

View File

@@ -1 +0,0 @@
48748

53
time.sh Normal file
View File

@@ -0,0 +1,53 @@
#!/bin/bash
set -e
echo "=== CentOS 7 时间同步脚本开始 ==="
# 1. 检查是否 root
if [ "$(id -u)" -ne 0 ]; then
echo "请使用 root 用户执行"
exit 1
fi
# 2. 设置时区
echo "[1/6] 设置时区为 Asia/Shanghai"
timedatectl set-timezone Asia/Shanghai
# 3. 安装 chrony
echo "[2/6] 安装 chrony"
if ! rpm -q chrony &>/dev/null; then
yum install -y chrony
else
echo "chrony 已安装,跳过"
fi
# 4. 配置 NTP 服务器
echo "[3/6] 配置 NTP 服务器"
CONF="/etc/chrony.conf"
sed -i 's/^server /#server /g' "$CONF"
grep -q "ntp.aliyun.com" "$CONF" || cat >> "$CONF" <<EOF
# China NTP servers
server ntp.aliyun.com iburst
server ntp.tencent.com iburst
server ntp1.aliyun.com iburst
EOF
# 5. 启动并启用 chronyd
echo "[4/6] 启动 chronyd"
systemctl enable chronyd --now
# 6. 立即强制同步
echo "[5/6] 强制同步系统时间"
chronyc -a makestep
# 7. 写入硬件时间
echo "[6/6] 写入硬件时钟"
hwclock --systohc
echo "=== 时间同步完成 ==="
echo
timedatectl status

View File

@@ -61,7 +61,7 @@ v1.2.0 (2019-09-26)
and `errors.Is`.
v1.1.0 (2017-06-30)
v1.1.2 (2017-06-30)
===================
- Added an `Errors(error) []error` function to extract the underlying list of

View File

@@ -489,7 +489,7 @@ Enhancements:
[#402]: https://github.com/uber-go/zap/pull/402
## v1.1.0 (31 Mar 2017)
## v1.1.2 (31 Mar 2017)
This release fixes two bugs and adds some enhancements to zap's testing helpers.
It is fully backward-compatible.

View File

@@ -1,4 +1,4 @@
{
"version": "1.1.0",
"tag": "v1.1.0"
"version": "1.1.4",
"tag": "v1.1.4"
}