knowledge-governance/internal/libraryexecutor/executor.go

// 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, ""
}