mirror of
https://gitcode.com/JianFeeeee/homeagent-sdk.git
synced 2026-10-03 23:54:12 +00:00
feat(deepsearch): 联网检索插件 + SearXNG 生命周期托管
把原 websearch 示例改名为 deepsearch（目录/go.mod/plg.json/工具前缀/README 全量对齐， 工具名 websearch_* → deepsearch_*）。 新增 searxng.go：插件自己托管搜索后端 - 启动探 healthz：已在跑则直接接管（不重启），没跑就 docker compose up -d 并等就绪 - 关闭动作注册为 stop handler（幂等、限时 4s < 内核 5s 宽限期） - 配置 manage_searxng / searxng_dir / stop_searxng_on_exit - 崩溃/被 kill 时不关后端（下次启动接管）：安全失败方向 - 可注入 cmdRunner + 时间预算，8 项单测离线覆盖接管/拉起/失败/幂等/保留 契约依据（internal/plugin/proc）：停止插件 = plugin.stop → RunStopHandlers（LIFO、 幂等）→ Stop() → exit(0)，宽限期 5s；stdin 关闭同路径。 验证：19 项单测（18 通过 + 1 联调跳过）、-race 干净、vet/gofmt 干净；内核 E2E 真调用 返回 58 条/37 条、58/58 带摘要；线上 daemon 实测 stop handler 与冷启动（约 3s）。
This commit is contained in:
JianFeeeee
159
example/deepsearch/searxng.go
Normal file
159
example/deepsearch/searxng.go
Normal file
@ -0,0 +1,159 @@
|
||||
package main
|
||||
|
||||
// SearXNG 生命周期托管：插件启动时拉起搜索后端，插件停止时关闭它。
|
||||
//
|
||||
// 契约依据（内核侧 internal/plugin/proc/*，已逐行核对）：
|
||||
// - 内核停止插件：发 `plugin.stop` → 插件先跑 RunStopHandlers（LIFO、幂等）→ 再 Stop() → exit(0)
|
||||
// - 若插件未在 stopGracePeriod（**5 秒**）内退出，内核直接 SIGKILL
|
||||
// - stdin 关闭（内核消失）同样会跑 handlers + Stop()
|
||||
//
|
||||
// 因此这里的关闭动作必须**有界**：searxShutdownBudget 取 4s，留 1s 余量。
|
||||
// 若插件是被 kill -9 / OOM 带走的，关闭动作不会执行 —— SearXNG 会留在运行态；
|
||||
// 下次 Start 探测到它在跑就直接接管，这是更安全的失败方向。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"net/http"
|
||||
"os/exec"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
cfgManageSearx = "manage_searxng"
|
||||
cfgSearxDir = "searxng_dir"
|
||||
cfgStopOnExit = "stop_searxng_on_exit"
|
||||
|
||||
defaultSearxDir = "/root/searxng-agent"
|
||||
|
||||
searxProbeTimeout = 1500 * time.Millisecond // 单次 healthz 探测
|
||||
searxUpBudget = 20 * time.Second // docker compose up -d 的上限（正常 1s 内返回）
|
||||
searxReadyBudget = 6 * time.Second // up 之后等 healthz 就绪的上限
|
||||
searxShutdownBudget = 4 * time.Second // 必须 < 内核 5s 宽限期
|
||||
)
|
||||
|
||||
// searxBudget 把四个时间预算收拢，便于单测注入短值（否则测试要真等就绪窗口）。
|
||||
type searxBudget struct {
|
||||
probe time.Duration
|
||||
up time.Duration
|
||||
ready time.Duration
|
||||
shutdown time.Duration
|
||||
}
|
||||
|
||||
func (p *Plugin) budget() searxBudget {
|
||||
b := p.bud
|
||||
if b.probe == 0 {
|
||||
b.probe = searxProbeTimeout
|
||||
}
|
||||
if b.up == 0 {
|
||||
b.up = searxUpBudget
|
||||
}
|
||||
if b.ready == 0 {
|
||||
b.ready = searxReadyBudget
|
||||
}
|
||||
if b.shutdown == 0 {
|
||||
b.shutdown = searxShutdownBudget
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// cmdRunner 抽出来是为了让生命周期逻辑可单测：注入假执行器，不起真容器。
|
||||
type cmdRunner func(ctx context.Context, dir, name string, args ...string) (string, error)
|
||||
|
||||
func defaultRunner(ctx context.Context, dir, name string, args ...string) (string, error) {
|
||||
cmd := exec.CommandContext(ctx, name, args...)
|
||||
cmd.Dir = dir
|
||||
out, err := cmd.CombinedOutput()
|
||||
return string(out), err
|
||||
}
|
||||
|
||||
// searxReachable 探测搜索后端是否可用（只看 healthz，不发检索请求）。
|
||||
func (p *Plugin) searxReachable(timeout time.Duration) bool {
|
||||
if p.searxURL == "" {
|
||||
return false
|
||||
}
|
||||
base := p.http
|
||||
if base == nil {
|
||||
base = &http.Client{}
|
||||
}
|
||||
cl := *base // 复制一份，避免改到共享 client 的超时
|
||||
cl.Timeout = timeout
|
||||
req, err := http.NewRequest(http.MethodGet, p.searxURL+"/healthz", nil)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
req.Header.Set("User-Agent", p.userAgent)
|
||||
resp, err := cl.Do(req)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
return resp.StatusCode < 400
|
||||
}
|
||||
|
||||
// ensureSearxng 在插件启动时确保搜索后端在跑；已在跑则直接接管，不重启。
|
||||
func (p *Plugin) ensureSearxng() {
|
||||
b := p.budget()
|
||||
if !p.manageSearx {
|
||||
log.Printf("[%s] 未启用 SearXNG 托管（manage_searxng=false），假定 %s 由外部维护", p.name, p.searxURL)
|
||||
return
|
||||
}
|
||||
if p.searxReachable(b.probe) {
|
||||
log.Printf("[%s] SearXNG 已在运行（%s），直接接管", p.name, p.searxURL)
|
||||
p.markSearxOwned()
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), b.up)
|
||||
out, err := p.run(ctx, p.searxDir, "docker", "compose", "up", "-d")
|
||||
cancel()
|
||||
if err != nil {
|
||||
log.Printf("[%s] 拉起 SearXNG 失败（dir=%s，请检查 manage_searxng/searxng_dir 配置）: %v；输出: %s",
|
||||
p.name, p.searxDir, err, oneLine(out, 300))
|
||||
return
|
||||
}
|
||||
log.Printf("[%s] 已执行 docker compose up -d（%s）：%s", p.name, p.searxDir, oneLine(out, 200))
|
||||
|
||||
deadline := time.Now().Add(b.ready)
|
||||
for time.Now().Before(deadline) {
|
||||
if p.searxReachable(800 * time.Millisecond) {
|
||||
log.Printf("[%s] SearXNG 就绪", p.name)
|
||||
p.markSearxOwned()
|
||||
return
|
||||
}
|
||||
time.Sleep(600 * time.Millisecond)
|
||||
}
|
||||
log.Printf("[%s] SearXNG 已启动但 %s 内未就绪；首次检索会自动等待", p.name, b.ready)
|
||||
p.markSearxOwned()
|
||||
}
|
||||
|
||||
func (p *Plugin) markSearxOwned() {
|
||||
p.searxMu.Lock()
|
||||
p.searxOwned = true
|
||||
p.searxMu.Unlock()
|
||||
}
|
||||
|
||||
// shutdownSearxng 关闭搜索后端。幂等，且有界（内核宽限期 5s，这里最多 4s）。
|
||||
func (p *Plugin) shutdownSearxng() {
|
||||
b := p.budget()
|
||||
p.searxMu.Lock()
|
||||
owned := p.searxOwned
|
||||
p.searxOwned = false
|
||||
p.searxMu.Unlock()
|
||||
|
||||
if !owned {
|
||||
return // 不是我们拉起来的 / 已经关过
|
||||
}
|
||||
if !p.manageSearx || !p.stopOnExit {
|
||||
log.Printf("[%s] 保留 SearXNG 运行（stop_searxng_on_exit=false）", p.name)
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), b.shutdown)
|
||||
defer cancel()
|
||||
out, err := p.run(ctx, p.searxDir, "docker", "compose", "stop", "-t", "2")
|
||||
if err != nil {
|
||||
// 故意只记日志：这里再重试就会拖过内核宽限期，被 SIGKILL 更糟
|
||||
log.Printf("[%s] 关闭 SearXNG 失败（忽略）: %v；输出: %s", p.name, err, oneLine(out, 200))
|
||||
return
|
||||
}
|
||||
log.Printf("[%s] 已关闭 SearXNG", p.name)
|
||||
}
|
||||
Reference in New Issue
Block a user