fix(cron): 兜底任务 panic，并支持自定义渠道探测间隔
panic 兜底：cron.New() 默认不带 Recover 链，robfig/cron v3 的 startJob 是裸的
`go func(){ j.Run() }()`——任意一个定时任务 panic 会直接崩掉整个进程，而不只是让该
任务失败。在 runJobInternal 内兜住并转成普通 error，进程存活且失败经 recordExecution
落进 sys_cron_jobs，管理后台可见，不再表现为一次无声重启。堆栈只进日志，避免撑爆
last_error 的展示。渠道自动探测再加一层 per-channel recover：单个渠道 panic 只跳过
自己，本轮剩余渠道继续（否则整轮被中断）。
探测间隔可配置：新增 channel_auto_test_interval_minutes（默认 5，范围 1-1440），
原先固定 */5 * * * *。用 @every 而非 */N——*/N 只在 N<60 时有意义，*/90 会被静默
解析成 */30，给出与配置不符的频率；@every 还让各实例按启动时刻错开，不会所有实例
卡在同一分钟同时探测。
配套两处基础能力：CronScheduler 记录 cron.EntryID 并新增 Reschedule（先加新条目、
成功后再摘旧条目，非法表达式时保持原调度，避免一次错配把任务彻底摘出调度器）；
ConfigService 新增 OnSettingsChanged 回调钩子，挂在既有的 settings:changed 订阅上，
使配置变更能触发动作而不只是让缓存失效，多实例各自重排、无需重启。
This commit is contained in:
chengqian authored and qian committed
1 parent
dd02bf4ac3
commit
ce59b12622
7 files changed
+315
-17
No files matched your search
+30
-1
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/net/ghttp"
|
||||
"github.com/gogf/gf/v2/os/gcmd"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
|
||||
"github.com/qianfree/team-api/internal/consts"
|
||||
adminController "github.com/qianfree/team-api/internal/controller/admin"
|
||||
@@ -335,9 +336,37 @@ func printBanner() {
|
||||
fmt.Println()
|
||||
}
|
||||
|
||||
// channelAutoTestSchedule 按配置的探测间隔生成 cron 表达式。
|
||||
//
|
||||
// 用 @every 而非 */N：*/N 只在 N < 60 时有意义（*/90 在分钟位上等价于 */30，
|
||||
// 静默给出与配置不符的频率），而 @every 支持任意时长。副作用是触发时刻相对进程
|
||||
// 启动时间而非整点对齐——多实例部署时反而更好，不会所有实例卡在同一分钟同时探测。
|
||||
func channelAutoTestSchedule(ctx context.Context) string {
|
||||
minutes := common.Config().GetInt(ctx, "channel_auto_test_interval_minutes")
|
||||
// 配置缺失（GetInt 返回 0）或越界时回落默认值，避免非法表达式让任务注册失败
|
||||
if minutes < 1 || minutes > 1440 {
|
||||
minutes = 5
|
||||
}
|
||||
return fmt.Sprintf("@every %dm", minutes)
|
||||
}
|
||||
|
||||
// watchChannelAutoTestInterval 探测间隔配置变更时重排 cron，无需重启。
|
||||
// 回调经 Redis pub/sub 触发，多实例会各自重排自己的调度器。
|
||||
func watchChannelAutoTestInterval(cs *common.CronScheduler) {
|
||||
common.OnSettingsChanged(func(ctx context.Context, key string) {
|
||||
if key != "channel_auto_test_interval_minutes" {
|
||||
return
|
||||
}
|
||||
if err := cs.Reschedule(ctx, "channel_auto_test", channelAutoTestSchedule(ctx)); err != nil {
|
||||
g.Log().Errorf(ctx, "渠道自动探测重排失败，沿用原间隔: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// registerCronJobs 集中注册所有定时任务，避免散落在主启动流程中。
|
||||
// 新增 cron 任务时在此追加一项即可；任务名（用于分布式锁 key）需保持唯一且稳定。
|
||||
func registerCronJobs(cs *common.CronScheduler) {
|
||||
watchChannelAutoTestInterval(cs)
|
||||
cs.Register("ops_system_collector", "系统指标采集", "* * * * *", func(ctx context.Context) error {
|
||||
return monitor.CollectSystemMetrics(ctx)
|
||||
})
|
||||
@@ -353,7 +382,7 @@ func registerCronJobs(cs *common.CronScheduler) {
|
||||
cs.Register("health_snapshot", "渠道健康快照", "*/5 * * * *", func(ctx context.Context) error {
|
||||
return task.SnapshotHealthScores(ctx)
|
||||
})
|
||||
cs.Register("channel_auto_test", "渠道自动测试", "*/5 * * * *", func(ctx context.Context) error {
|
||||
cs.Register("channel_auto_test", "渠道自动测试", channelAutoTestSchedule(gctx.New()), func(ctx context.Context) error {
|
||||
if common.Config().GetBool(ctx, "channel_auto_test_enabled") {
|
||||
task.AutoTestChannels(ctx)
|
||||
}
|
||||
|
||||
@@ -387,6 +387,8 @@ func (s *ConfigService) StartSubscriber(ctx context.Context) {
|
||||
s.cache.Delete(ctx, "public_options")
|
||||
s.cache.Delete(ctx, "category:"+getCategoryForKey(key))
|
||||
g.Log().Debugf(ctx, "[PubSub:settings] 缓存已失效: %s", key)
|
||||
// 缓存失效后再触发回调：回调内部大概率要读该配置的新值
|
||||
fireSettingsChanged(ctx, key)
|
||||
}
|
||||
|
||||
conn.Close(ctx)
|
||||
@@ -394,6 +396,46 @@ func (s *ConfigService) StartSubscriber(ctx context.Context) {
|
||||
}()
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────
|
||||
// 配置变更回调
|
||||
// ──────────────────────────────────────────
|
||||
|
||||
var (
|
||||
settingsHookMu sync.RWMutex
|
||||
settingsHooks []func(ctx context.Context, key string)
|
||||
)
|
||||
|
||||
// OnSettingsChanged 注册配置变更回调，用于"改了配置要立刻做点什么"的场景
|
||||
// （如按新的探测间隔重排 cron），而不只是让缓存失效。
|
||||
//
|
||||
// 回调经 Redis pub/sub 触发，因此本实例自己的写入同样会走一遍（发布者也订阅了该频道），
|
||||
// 无需在写入路径上重复调用。Redis 不可用时回调不触发——与缓存失效同等降级。
|
||||
// 回调在订阅 goroutine 内串行执行，实现必须快速返回、自行兜底 panic。
|
||||
func OnSettingsChanged(fn func(ctx context.Context, key string)) {
|
||||
settingsHookMu.Lock()
|
||||
defer settingsHookMu.Unlock()
|
||||
settingsHooks = append(settingsHooks, fn)
|
||||
}
|
||||
|
||||
// fireSettingsChanged 依次触发已注册回调，单个回调 panic 不影响其余回调与订阅循环。
|
||||
func fireSettingsChanged(ctx context.Context, key string) {
|
||||
settingsHookMu.RLock()
|
||||
hooks := make([]func(ctx context.Context, key string), len(settingsHooks))
|
||||
copy(hooks, settingsHooks)
|
||||
settingsHookMu.RUnlock()
|
||||
|
||||
for _, fn := range hooks {
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
g.Log().Errorf(ctx, "[PubSub:settings] 变更回调 panic (key=%s): %v", key, r)
|
||||
}
|
||||
}()
|
||||
fn(ctx, key)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────
|
||||
// Validation
|
||||
// ──────────────────────────────────────────
|
||||
|
||||
@@ -2,6 +2,7 @@ package common
|
||||
|
||||
import (
|
||||
"context"
|
||||
"runtime/debug"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -35,6 +36,10 @@ type CronScheduler struct {
|
||||
running map[string]bool
|
||||
runMu sync.Mutex
|
||||
cron *cron.Cron
|
||||
// entryIDs job 名 → robfig/cron 条目 ID，供 Reschedule 摘除旧条目。
|
||||
// 仅在 StartBackground 之后有值。
|
||||
entryIDs map[string]cron.EntryID
|
||||
entryMu sync.Mutex
|
||||
}
|
||||
|
||||
var (
|
||||
@@ -60,8 +65,9 @@ func GetCronScheduler() *CronScheduler {
|
||||
// NewCronScheduler creates a new CronScheduler.
|
||||
func NewCronScheduler() *CronScheduler {
|
||||
return &CronScheduler{
|
||||
running: make(map[string]bool),
|
||||
cron: cron.New(),
|
||||
running: make(map[string]bool),
|
||||
entryIDs: make(map[string]cron.EntryID),
|
||||
cron: cron.New(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -180,7 +186,7 @@ func (cs *CronScheduler) runJobInternal(ctx context.Context, name, triggeredBy s
|
||||
|
||||
// Execute job
|
||||
startTime := time.Now()
|
||||
handlerErr := job.Handler(ctx)
|
||||
handlerErr := runHandlerSafely(ctx, job)
|
||||
duration := time.Since(startTime)
|
||||
|
||||
status := "succeeded"
|
||||
@@ -197,6 +203,23 @@ func (cs *CronScheduler) runJobInternal(ctx context.Context, name, triggeredBy s
|
||||
return handlerErr
|
||||
}
|
||||
|
||||
// runHandlerSafely 执行 job handler 并兜底 panic。
|
||||
//
|
||||
// robfig/cron v3 的 cron.New() 默认不带 Recover 链，其 startJob 是裸的
|
||||
// `go func() { j.Run() }()`——任意一个 job panic 都会直接崩掉整个进程，而不只是
|
||||
// 让该任务失败。此处把 panic 转成普通 error：进程存活，且失败会经 recordExecution
|
||||
// 落进 sys_cron_jobs，管理后台的任务列表能看到，不会变成一次无声的重启。
|
||||
func runHandlerSafely(ctx context.Context, job *CronJob) (err error) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
// 堆栈只进日志：last_error 会入库并在管理后台展示，塞完整堆栈会撑爆展示
|
||||
g.Log().Errorf(ctx, "cron job panicked: %s: %v\n%s", job.Name, r, debug.Stack())
|
||||
err = gerror.Newf("任务执行 panic: %v", r)
|
||||
}
|
||||
}()
|
||||
return job.Handler(ctx)
|
||||
}
|
||||
|
||||
// recordExecution persists an execution record to the database.
|
||||
// Always UPSERT sys_cron_jobs with latest status; no separate execution log table.
|
||||
func (cs *CronScheduler) recordExecution(ctx context.Context, jobName, status string, startedAt time.Time, duration time.Duration, errMsg, triggeredBy string) {
|
||||
@@ -268,7 +291,7 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
|
||||
|
||||
for _, job := range jobs {
|
||||
j := job
|
||||
_, err := cs.cron.AddFunc(j.Schedule, func() {
|
||||
entryID, err := cs.cron.AddFunc(j.Schedule, func() {
|
||||
bgCtx := gctx.New()
|
||||
if err := cs.RunJob(bgCtx, j.Name); err != nil {
|
||||
g.Log().Errorf(bgCtx, "background cron error: %s: %v", j.Name, err)
|
||||
@@ -276,7 +299,11 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
|
||||
})
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "failed to register cron job %s with schedule %s: %v", j.Name, j.Schedule, err)
|
||||
continue
|
||||
}
|
||||
cs.entryMu.Lock()
|
||||
cs.entryIDs[j.Name] = entryID
|
||||
cs.entryMu.Unlock()
|
||||
}
|
||||
|
||||
cs.cron.Start()
|
||||
@@ -291,6 +318,58 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
|
||||
g.Log().Info(ctx, "cron scheduler started in background")
|
||||
}
|
||||
|
||||
// Reschedule 更新已注册 job 的调度表达式并即时生效（配置热更新用）。
|
||||
//
|
||||
// 先添加新条目、成功后再摘除旧条目：表达式非法时保持原调度不变，
|
||||
// 避免一次错误配置把任务从调度器里彻底摘掉。
|
||||
// StartBackground 之前调用无意义（尚无条目），此时只更新 job.Schedule 记录。
|
||||
func (cs *CronScheduler) Reschedule(ctx context.Context, name, schedule string) error {
|
||||
cs.mu.RLock()
|
||||
var job *CronJob
|
||||
for _, j := range cs.jobs {
|
||||
if j.Name == name {
|
||||
job = j
|
||||
break
|
||||
}
|
||||
}
|
||||
cs.mu.RUnlock()
|
||||
if job == nil {
|
||||
return gerror.Newf("job %s not registered", name)
|
||||
}
|
||||
|
||||
cs.entryMu.Lock()
|
||||
defer cs.entryMu.Unlock()
|
||||
|
||||
oldID, started := cs.entryIDs[name]
|
||||
if !started {
|
||||
cs.mu.Lock()
|
||||
job.Schedule = schedule
|
||||
cs.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
newID, err := cs.cron.AddFunc(schedule, func() {
|
||||
bgCtx := gctx.New()
|
||||
if err := cs.RunJob(bgCtx, name); err != nil {
|
||||
g.Log().Errorf(bgCtx, "background cron error: %s: %v", name, err)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return gerror.Wrapf(err, "invalid schedule %q for job %s", schedule, name)
|
||||
}
|
||||
|
||||
cs.cron.Remove(oldID)
|
||||
cs.entryIDs[name] = newID
|
||||
|
||||
cs.mu.Lock()
|
||||
oldSchedule := job.Schedule
|
||||
job.Schedule = schedule
|
||||
cs.mu.Unlock()
|
||||
|
||||
g.Log().Infof(ctx, "cron job %s rescheduled: %s → %s", name, oldSchedule, schedule)
|
||||
return nil
|
||||
}
|
||||
|
||||
// cronLockTTL 分布式锁过期时间：仅用于实例崩溃时的兜底自动释放（正常完成会主动释放）。
|
||||
// 取值需大于绝大多数 job 的执行时长；过长则崩溃后会阻塞该 job 至多这么久。
|
||||
const cronLockTTL = 10 * time.Minute
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
package common
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestRunHandlerSafely_Panic转为错误 robfig/cron v3 的 cron.New() 不带 Recover 链，
|
||||
// startJob 是裸的 `go j.Run()`——任务 panic 会崩掉整个进程。此处必须兜住并转成
|
||||
// 普通 error，才能既保住进程、又让失败落进 sys_cron_jobs 供管理后台查看。
|
||||
func TestRunHandlerSafely_Panic转为错误(t *testing.T) {
|
||||
job := &CronJob{
|
||||
Name: "panicking_job",
|
||||
Handler: func(_ context.Context) error {
|
||||
panic("boom")
|
||||
},
|
||||
}
|
||||
|
||||
var err error
|
||||
assert.NotPanics(t, func() {
|
||||
err = runHandlerSafely(context.Background(), job)
|
||||
}, "panic 必须被兜住，否则整个进程会崩")
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "boom", "错误信息应保留 panic 内容供排查")
|
||||
}
|
||||
|
||||
func TestRunHandlerSafely_正常返回值透传(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
okJob := &CronJob{Name: "ok", Handler: func(_ context.Context) error { return nil }}
|
||||
assert.NoError(t, runHandlerSafely(ctx, okJob))
|
||||
|
||||
failJob := &CronJob{Name: "fail", Handler: func(_ context.Context) error {
|
||||
return assert.AnError
|
||||
}}
|
||||
assert.ErrorIs(t, runHandlerSafely(ctx, failJob), assert.AnError, "普通错误必须原样返回")
|
||||
}
|
||||
|
||||
// TestReschedule_未启动时只更新记录 StartBackground 之前没有 cron 条目，
|
||||
// 此时重排只改 job.Schedule，供随后的 StartBackground 按新表达式注册。
|
||||
func TestReschedule_未启动时只更新记录(t *testing.T) {
|
||||
cs := NewCronScheduler()
|
||||
cs.Register("job_a", "任务A", "*/5 * * * *", func(_ context.Context) error { return nil })
|
||||
|
||||
require.NoError(t, cs.Reschedule(context.Background(), "job_a", "@every 30m"))
|
||||
|
||||
jobs := cs.ListJobs()
|
||||
require.Len(t, jobs, 1)
|
||||
assert.Equal(t, "@every 30m", jobs[0].Schedule)
|
||||
}
|
||||
|
||||
func TestReschedule_未注册任务报错(t *testing.T) {
|
||||
cs := NewCronScheduler()
|
||||
assert.Error(t, cs.Reschedule(context.Background(), "nonexistent", "@every 5m"))
|
||||
}
|
||||
|
||||
// TestReschedule_非法表达式保持原调度 一次错误配置不能把任务从调度器里彻底摘掉，
|
||||
// 因此实现是「先加新条目、成功后再摘旧条目」。
|
||||
func TestReschedule_非法表达式保持原调度(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cs := NewCronScheduler()
|
||||
cs.Register("job_b", "任务B", "@every 5m", func(_ context.Context) error { return nil })
|
||||
cs.StartBackground(ctx)
|
||||
defer cs.cron.Stop()
|
||||
|
||||
before := cs.ListJobs()[0].Schedule
|
||||
require.Error(t, cs.Reschedule(ctx, "job_b", "不是合法表达式"))
|
||||
assert.Equal(t, before, cs.ListJobs()[0].Schedule, "非法表达式不得改动原调度")
|
||||
|
||||
// 条目仍在调度器中，未被误摘
|
||||
assert.Len(t, cs.cron.Entries(), 1)
|
||||
}
|
||||
|
||||
// TestReschedule_有效表达式替换条目 同时验证 @every 语法确实被 cron.New() 的
|
||||
// 默认解析器接受——探测间隔配置正是靠它支持任意分钟数（*/N 在 N>=60 时会静默失真）。
|
||||
func TestReschedule_有效表达式替换条目(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cs := NewCronScheduler()
|
||||
cs.Register("job_c", "任务C", "@every 5m", func(_ context.Context) error { return nil })
|
||||
cs.StartBackground(ctx)
|
||||
defer cs.cron.Stop()
|
||||
|
||||
require.Len(t, cs.cron.Entries(), 1)
|
||||
oldID := cs.entryIDs["job_c"]
|
||||
|
||||
require.NoError(t, cs.Reschedule(ctx, "job_c", "@every 90m"))
|
||||
|
||||
assert.Equal(t, "@every 90m", cs.ListJobs()[0].Schedule)
|
||||
assert.Len(t, cs.cron.Entries(), 1, "旧条目必须被摘除，否则会双倍触发")
|
||||
assert.NotEqual(t, oldID, cs.entryIDs["job_c"], "条目 ID 应已更新")
|
||||
}
|
||||
|
||||
// TestOnSettingsChanged_回调panic不影响其余回调 回调在订阅 goroutine 内串行执行，
|
||||
// 一个回调 panic 不能带崩订阅循环或阻断后续回调。
|
||||
func TestOnSettingsChanged_回调panic不影响其余回调(t *testing.T) {
|
||||
settingsHookMu.Lock()
|
||||
saved := settingsHooks
|
||||
settingsHooks = nil
|
||||
settingsHookMu.Unlock()
|
||||
defer func() {
|
||||
settingsHookMu.Lock()
|
||||
settingsHooks = saved
|
||||
settingsHookMu.Unlock()
|
||||
}()
|
||||
|
||||
var firstCalled, thirdCalled bool
|
||||
OnSettingsChanged(func(_ context.Context, _ string) { firstCalled = true })
|
||||
OnSettingsChanged(func(_ context.Context, _ string) { panic("hook boom") })
|
||||
OnSettingsChanged(func(_ context.Context, _ string) { thirdCalled = true })
|
||||
|
||||
assert.NotPanics(t, func() {
|
||||
fireSettingsChanged(context.Background(), "some_key")
|
||||
})
|
||||
assert.True(t, firstCalled)
|
||||
assert.True(t, thirdCalled, "前一个回调 panic 不得阻断后续回调")
|
||||
}
|
||||
@@ -213,6 +213,9 @@ var Registry = []SettingDef{
|
||||
// ── Channel ──
|
||||
{Key: "channel_auto_test_enabled", Type: SettingTypeBool, Default: "true", Category: "channel",
|
||||
Label: "渠道自动探测", Description: "定期向活跃渠道发送测试请求，检测连通性并更新健康度（会消耗少量 Token）。禁用渠道不自动探测，由管理员手动测试确认后再启用"},
|
||||
{Key: "channel_auto_test_interval_minutes", Type: SettingTypeInt, Default: "5", Category: "channel",
|
||||
Label: "自动探测间隔（分钟）", Validation: "min:1,max:1440",
|
||||
Description: "两轮自动探测之间的间隔，保存后即时生效。探测是串行的（每渠道最多 30 秒），渠道多或上游超时多时单轮耗时会变长；上一轮未结束时下一轮自动跳过，因此间隔不宜小于「渠道数 × 30 秒」"},
|
||||
{Key: "channel_auto_disable_enabled", Type: SettingTypeBool, Default: "false", Category: "channel",
|
||||
Label: "渠道自动禁用", Description: "渠道熔断持续超过路由策略 breaker.autoDisableAfterSeconds（默认 10 分钟）未恢复时自动禁用"},
|
||||
{Key: "channel_routing_policy", Type: SettingTypeJSON, Default: "", Category: "channel",
|
||||
|
||||
@@ -43,18 +43,7 @@ func AutoTestChannels(ctx context.Context) {
|
||||
failCount := 0
|
||||
|
||||
for _, ch := range channels {
|
||||
testCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
result, err := admin.New().TestChannel(testCtx, &v1.ChannelTestReq{ID: ch.ID})
|
||||
cancel()
|
||||
|
||||
if err != nil {
|
||||
failCount++
|
||||
// 探测请求尚未发出（无可用 Key/渠道不存在等），TestChannel 内部无结果日志，此处记录
|
||||
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测执行出错: %v", ch.Name, ch.ID, err)
|
||||
continue
|
||||
}
|
||||
|
||||
if result.Success {
|
||||
if probeOneChannel(ctx, ch.ID, ch.Name) {
|
||||
successCount++
|
||||
} else {
|
||||
failCount++
|
||||
@@ -65,3 +54,27 @@ func AutoTestChannels(ctx context.Context) {
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 自动探测完成: 共 %d 个渠道 | 成功 %d | 失败 %d",
|
||||
len(channels), successCount, failCount)
|
||||
}
|
||||
|
||||
// probeOneChannel 探测单个渠道，返回是否成功。
|
||||
// 独立超时 + panic 兜底：单个渠道无论超时、报错还是 panic，都不影响本轮后续渠道
|
||||
// （panic 若逃逸到 cron 层虽有 runHandlerSafely 兜底进程，但本轮剩余渠道会被整体跳过）。
|
||||
func probeOneChannel(ctx context.Context, channelID int64, channelName string) (ok bool) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
ok = false
|
||||
g.Log().Errorf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测 panic，跳过该渠道继续下一个: %v",
|
||||
channelName, channelID, r)
|
||||
}
|
||||
}()
|
||||
|
||||
testCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
result, err := admin.New().TestChannel(testCtx, &v1.ChannelTestReq{ID: channelID})
|
||||
if err != nil {
|
||||
// 探测请求尚未发出（无可用 Key/渠道不存在等），TestChannel 内部无结果日志，此处记录
|
||||
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测执行出错: %v", channelName, channelID, err)
|
||||
return false
|
||||
}
|
||||
return result.Success
|
||||
}
|
||||
@@ -16,6 +16,19 @@ const values = useFormValues()
|
||||
@change="(v: string | number | boolean) => values['channel_auto_test_enabled'] = v"
|
||||
/>
|
||||
</div>
|
||||
<div class="section-grid" style="margin-top: 12px">
|
||||
<AFormItem label="探测间隔（分钟）">
|
||||
<AInputNumber
|
||||
:model-value="values['channel_auto_test_interval_minutes'] as number"
|
||||
@change="(v: number | undefined) => values['channel_auto_test_interval_minutes'] = v ?? 5"
|
||||
:min="1" :max="1440" style="width: 100%"
|
||||
/>
|
||||
</AFormItem>
|
||||
</div>
|
||||
<div class="section-desc">
|
||||
保存后即时生效，无需重启。探测是串行的（每渠道最多 30 秒），渠道多或上游超时多时单轮耗时会变长；
|
||||
上一轮未结束时下一轮自动跳过，因此间隔不宜小于「渠道数 × 30 秒」。
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- 自动禁用 -->
|
||||
|
||||
Reference in new issue
Block a user