// Package libraryexecutor 回答一个问题:主机侧封存执行器现在是不是真的可用, // 而且是不是指向我们这个实例。 // // 为什么不能在页面里读一个配置字符串了事:正式栈曾经出现过"agent 顶着正式单元名 // 却在冻结候选实例"的事故——配置非空、页面显示可发起、真发起却封错了库。可用的 // 定义必须是**运行时观测**:socket 通、执行器自报的实例与期望一致、没有降级。 package libraryexecutor import ( "context" "encoding/json" "fmt" "time" "knowledge-governance/internal/libraryagent" ) // Identity 是执行器自报的身份:它要冻结哪个容器的写入者、归档写到哪。 // 只有容器名与路径,没有任何凭据。 type Identity struct { Project string `json:"project"` ArchiveRoot string `json:"archive_root"` JournalRoot string `json:"journal_root"` SourceApp string `json:"source_app"` SourcePG string `json:"source_pg"` SourceES string `json:"source_es"` Writers string `json:"writers"` SourceS3Endpoint string `json:"source_s3_endpoint,omitempty"` } // Observation 是一次真实探测的结果,会被落库供页面读取。 type Observation struct { Instance string `json:"instance"` Reachable bool `json:"reachable"` Busy bool `json:"busy"` Degraded string `json:"degraded,omitempty"` Identity Identity `json:"identity"` Error string `json:"error,omitempty"` ObservedAt time.Time `json:"observed_at"` } // Expectation 是治理侧对执行器的要求。ArchiveRoot 不参与判定:容器里的路径是 // 挂载后的路径,和宿主机上的真实路径本来就不是一个字符串;用它当门禁只会造成 // 永久误报,所以只作为观测值展示给运维比对。 type Expectation struct { Instance string Project string ArchiveRoot string } // Probe 用一次短超时的 health 往返取回执行器身份。它不跑子进程、不碰源库, // 所以可以当心跳反复调用。instance 是"这次观测属于哪个实例",由调用方的 // LIBRARY_TARGET_INSTANCE 决定。 func Probe(ctx context.Context, instance, socket string, timeout time.Duration, now func() time.Time) Observation { if now == nil { now = time.Now } observation := Observation{Instance: instance, ObservedAt: now().UTC()} if socket == "" { observation.Error = "未配置主机侧封存 agent socket(LIBRARY_AGENT_SOCKET)" return observation } if timeout <= 0 { timeout = 3 * time.Second } result, err := libraryagent.Health(ctx, libraryagent.ClientConfig{Socket: socket, Timeout: timeout}) if err != nil { observation.Error = err.Error() return observation } var payload struct { Identity Busy bool `json:"busy"` Degraded string `json:"degraded"` } if err = json.Unmarshal(result.Payload, &payload); err != nil { observation.Error = "执行器健康应答无法解析" return observation } observation.Reachable = true observation.Busy = payload.Busy observation.Degraded = payload.Degraded observation.Identity = payload.Identity return observation } // Evaluate 把"观测 + 期望 + 新鲜度"换算成页面上的可用性与原因。 // 返回的原因必须能直接告诉运维该去改什么,而不是"不可用"三个字。 func Evaluate(observation Observation, want Expectation, maxAge time.Duration, now time.Time) (bool, string) { if want.Instance == "" { return false, "未配置封存来源实例身份(LIBRARY_SOURCE_INSTANCE),无法确认执行器归属" } if observation.Instance != want.Instance { return false, fmt.Sprintf("执行器状态来自另一个实例(%q),不是 %q", observation.Instance, want.Instance) } age := now.Sub(observation.ObservedAt) if maxAge > 0 && age > maxAge { return false, fmt.Sprintf("执行器状态已过期:%d 秒前的观测,上限 %d 秒(worker 未上报?)", int(age.Seconds()), int(maxAge.Seconds())) } if !observation.Reachable { reason := "主机侧封存执行器不可达" if observation.Error != "" { reason += ":" + observation.Error } return false, reason } if want.Project == "" { return false, "未配置执行器源身份(LIBRARY_SOURCE_PROJECT),无法核对 agent 指向哪个实例" } if observation.Identity.Project != want.Project { return false, fmt.Sprintf("执行器指向的实例不匹配:agent=%q,期望=%q", observation.Identity.Project, want.Project) } if observation.Degraded != "" { return false, "执行器处于降级状态(上次恢复失败):" + observation.Degraded } if observation.Busy { return false, "执行器正在执行另一个封存/恢复,等它结束再发起" } return true, "" }