姚章浩
cfad003499
正式 release-worker 崩溃重启时日志只有一句 connect release target database， 主机名、库名和底层原因都不在里面。现在错误里带 host:port/db、路由标识与原因。 端点去口令不等于整条日志安全：pgx 的解析错误可能原样回显连接串。新增 internal/pgdsn.Redact（按已知口令精确替换 + 按 URL/keyword 形状兜底）， 错误保留 Unwrap 以便 errors.Is/As，技术原因不丢。cmd/worker 里只留库名的旧实现 换成同一个包，避免两份脱敏逻辑。libraryctl 用法补上 dead-letters。
1154 lines
43 KiB
Go
1154 lines
43 KiB
Go
// Command libraryctl is the operator entry point for the whole-library capture
|
||
// chain: it reads one test knowledge base (internal or external) from the real
|
||
// WeKnora test stack, freezes its writers for the duration of the export and
|
||
// seals a content-addressed candidate artifact.
|
||
//
|
||
// Boundaries this command keeps:
|
||
// - the source is read through a read-only PostgreSQL session; no source
|
||
// schema, data or program is modified;
|
||
// - only SIGTERM is sent to the declared writer group, and only when every
|
||
// service of the Compose project is declared (a new sibling container is
|
||
// refused, not ignored);
|
||
// - credentials are read from the running containers at the moment of the
|
||
// call and are never printed, logged or written to disk;
|
||
// - the sealed artifact is a *candidate*: it is not a release, and it does
|
||
// not replace snapshot-bound content, Q&A and performance acceptance.
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"flag"
|
||
"fmt"
|
||
"os"
|
||
"os/exec"
|
||
"os/signal"
|
||
"path/filepath"
|
||
"sort"
|
||
"strings"
|
||
"syscall"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
||
"knowledge-governance/internal/librarydeadletter"
|
||
"knowledge-governance/internal/libraryfreeze"
|
||
"knowledge-governance/internal/librarymaintenance"
|
||
"knowledge-governance/internal/librarysnapshot"
|
||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||
releaseworkflow "knowledge-governance/internal/workflows/release"
|
||
)
|
||
|
||
type options struct {
|
||
sourceApp string
|
||
sourcePG string
|
||
sourceES string
|
||
governancePG string
|
||
project string
|
||
writers string
|
||
nonWriters string
|
||
archiveRoot string
|
||
journalRoot string
|
||
dockerSocket string
|
||
// sourceS3Endpoint：容器里写的对象存储端点往往只在容器网络里能解析
|
||
// （例如 minio 服务名），而封存跑在宿主机上。这里允许用宿主机可达地址覆盖。
|
||
sourceS3Endpoint string
|
||
maxDocuments int
|
||
maxObjectBytes int64
|
||
scope string
|
||
job string
|
||
dryRun bool
|
||
snapshot string
|
||
// releasedDeadLetters 是治理平台审计过的精确解除指纹（逗号分隔）。它只让这些
|
||
// 确切的行不再阻断；实时行必须逐字节命中，否则照旧拒绝。
|
||
releasedDeadLetters string
|
||
}
|
||
|
||
func main() {
|
||
if len(os.Args) < 2 {
|
||
usage()
|
||
os.Exit(2)
|
||
}
|
||
command := os.Args[1]
|
||
opts, rest, err := parseFlags(command, os.Args[2:])
|
||
if err != nil {
|
||
fmt.Fprintln(os.Stderr, "LIBRARYCTL_FLAGS:", err)
|
||
os.Exit(2)
|
||
}
|
||
_ = rest
|
||
// SIGTERM must not kill this process mid-window: the capture path restores the
|
||
// source writers through a cancelled-context recovery, so the signal is turned
|
||
// into a context cancellation and the process exits only after that cleanup.
|
||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||
defer stop()
|
||
ctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
|
||
defer cancel()
|
||
var runErr error
|
||
switch command {
|
||
case "status":
|
||
runErr = runStatus(ctx, opts)
|
||
case "preflight":
|
||
runErr = runPreflight(ctx, opts)
|
||
case "capture":
|
||
runErr = runCapture(ctx, opts)
|
||
case "list":
|
||
runErr = runList(opts)
|
||
case "verify":
|
||
runErr = runVerify(opts)
|
||
case "recover":
|
||
runErr = runRecover(ctx, opts)
|
||
case "dead-letters":
|
||
runErr = runDeadLetters(ctx, opts)
|
||
default:
|
||
usage()
|
||
os.Exit(2)
|
||
}
|
||
if runErr != nil {
|
||
fmt.Fprintln(os.Stderr, "LIBRARYCTL_FAILED:", runErr)
|
||
os.Exit(1)
|
||
}
|
||
}
|
||
|
||
func usage() {
|
||
fmt.Fprint(os.Stderr, `libraryctl <status|preflight|capture|list|verify|recover|dead-letters> [flags]
|
||
|
||
status probe the real source and print what is configured / blocked
|
||
preflight read the source, run settlement + candidate validation (no writes, no stop)
|
||
capture freeze the declared writer group, archive a candidate, restore writers
|
||
list list sealed candidates and stored bytes
|
||
verify re-verify one sealed candidate (digests, coverage, identity)
|
||
recover restore writers from an interrupted freeze journal
|
||
dead-letters print the source's live dead letters (read-only) with a digest the
|
||
governance platform binds a precise release to
|
||
|
||
Flags (defaults target the local WKNFull test stack):
|
||
-source-app WKNFull-app container whose env describes the source
|
||
-source-pg WKNFull-postgres source PostgreSQL container (address resolution)
|
||
-source-es WKNFull-elasticsearch
|
||
-governance-pg knowledge-governance-p1-postgres-1 routing bindings read here
|
||
-project weknora071full Compose project owning the writer group
|
||
-writers WKNFull-app declared writer services (comma separated)
|
||
-non-writers WKNFull-frontend=reason,... every other project service
|
||
-archive ~/.local/share/knowledge-governance/library
|
||
-journal ~/.local/share/knowledge-governance/freeze
|
||
-docker-socket /var/run/docker.sock
|
||
-scope internal|external (preflight/capture)
|
||
-job <id> (capture; default: lib-<scope>-<UTC timestamp>)
|
||
-dry-run (capture: export + validate only, no archive, no stop)
|
||
-snapshot <id> (verify)
|
||
-released-dead-letters a,b (capture/preflight; 治理平台精确解除的死信指纹)
|
||
dead-letters (read-only) 打印整库死信原始行与观测摘要，供治理平台审计
|
||
`)
|
||
}
|
||
|
||
func parseFlags(command string, args []string) (options, []string, error) {
|
||
home, _ := os.UserHomeDir()
|
||
opts := options{
|
||
sourceApp: "WKNFull-app",
|
||
sourcePG: "WKNFull-postgres",
|
||
sourceES: "WKNFull-elasticsearch",
|
||
governancePG: "knowledge-governance-p1-postgres-1",
|
||
project: "weknora071full",
|
||
writers: "WKNFull-app",
|
||
archiveRoot: filepath.Join(home, ".local/share/knowledge-governance/library"),
|
||
journalRoot: filepath.Join(home, ".local/share/knowledge-governance/freeze"),
|
||
dockerSocket: "/var/run/docker.sock",
|
||
maxDocuments: 100000,
|
||
maxObjectBytes: 134217728,
|
||
}
|
||
set := flag.NewFlagSet(command, flag.ContinueOnError)
|
||
set.SetOutput(os.Stderr)
|
||
set.StringVar(&opts.sourceApp, "source-app", opts.sourceApp, "source container describing the stack")
|
||
set.StringVar(&opts.sourcePG, "source-pg", opts.sourcePG, "source PostgreSQL container")
|
||
set.StringVar(&opts.sourceES, "source-es", opts.sourceES, "source Elasticsearch container")
|
||
set.StringVar(&opts.governancePG, "governance-pg", opts.governancePG, "governance PostgreSQL container")
|
||
set.StringVar(&opts.project, "project", opts.project, "Compose project owning the writers")
|
||
set.StringVar(&opts.writers, "writers", opts.writers, "declared writer services")
|
||
set.StringVar(&opts.nonWriters, "non-writers", "", "service=reason for every other project service")
|
||
set.StringVar(&opts.archiveRoot, "archive", opts.archiveRoot, "sealed artifact root")
|
||
set.StringVar(&opts.journalRoot, "journal", opts.journalRoot, "freeze journal root")
|
||
set.StringVar(&opts.dockerSocket, "docker-socket", opts.dockerSocket, "Docker engine socket")
|
||
set.StringVar(&opts.sourceS3Endpoint, "source-s3-endpoint", "", "host-reachable source object-store endpoint (overrides the container-internal one)")
|
||
set.IntVar(&opts.maxDocuments, "max-documents", opts.maxDocuments, "maximum indexed documents per scope")
|
||
set.Int64Var(&opts.maxObjectBytes, "max-object-bytes", opts.maxObjectBytes, "maximum bytes per stored object")
|
||
set.StringVar(&opts.scope, "scope", "", "internal|external")
|
||
set.StringVar(&opts.job, "job", "", "capture job id")
|
||
set.BoolVar(&opts.dryRun, "dry-run", false, "export and validate without archiving or stopping")
|
||
set.StringVar(&opts.snapshot, "snapshot", "", "snapshot id for verify")
|
||
set.StringVar(&opts.releasedDeadLetters, "released-dead-letters", "", "audited dead-letter fingerprints (comma separated)")
|
||
if err := set.Parse(args); err != nil {
|
||
return opts, nil, err
|
||
}
|
||
return opts, set.Args(), nil
|
||
}
|
||
|
||
// ---------------------------------------------------------------- discovery
|
||
|
||
func dockerOutput(ctx context.Context, args ...string) (string, error) {
|
||
bounded, cancel := context.WithTimeout(ctx, 60*time.Second)
|
||
defer cancel()
|
||
out, err := exec.CommandContext(bounded, "docker", args...).Output()
|
||
if err != nil {
|
||
var exit *exec.ExitError
|
||
if errors.As(err, &exit) && len(exit.Stderr) > 0 {
|
||
return "", fmt.Errorf("docker %s failed", args[0])
|
||
}
|
||
return "", fmt.Errorf("docker %s failed", args[0])
|
||
}
|
||
return strings.TrimSpace(string(out)), nil
|
||
}
|
||
|
||
func containerEnv(ctx context.Context, name string) (map[string]string, error) {
|
||
raw, err := dockerOutput(ctx, "inspect", "--format", "{{json .Config.Env}}", name)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var entries []string
|
||
if err := json.Unmarshal([]byte(raw), &entries); err != nil {
|
||
return nil, errors.New("invalid container environment")
|
||
}
|
||
env := map[string]string{}
|
||
for _, entry := range entries {
|
||
if key, value, ok := strings.Cut(entry, "="); ok {
|
||
env[key] = value
|
||
}
|
||
}
|
||
return env, nil
|
||
}
|
||
|
||
func containerAddress(ctx context.Context, name string) (string, error) {
|
||
raw, err := dockerOutput(ctx, "inspect", "--format",
|
||
"{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}", name)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
for _, field := range strings.Fields(raw) {
|
||
if field != "" && field != "0.0.0.0" {
|
||
return field, nil
|
||
}
|
||
}
|
||
return "", fmt.Errorf("container %s has no reachable address", name)
|
||
}
|
||
|
||
func containerImageID(ctx context.Context, name string) (string, error) {
|
||
return dockerOutput(ctx, "inspect", "--format", "{{.Image}}", name)
|
||
}
|
||
|
||
// sourceConfig is everything needed to read the test library. Credential fields
|
||
// are never printed: String() masks them and no log line formats the struct.
|
||
type sourceConfig struct {
|
||
Scopes map[string]string
|
||
DatabaseURL string
|
||
Search v071.SearchExporterConfig
|
||
Object v071.S3ObjectStoreConfig
|
||
AppImageID string
|
||
SearchIndex string
|
||
ObjectDetail string
|
||
}
|
||
|
||
func discoverSource(ctx context.Context, opts options, binding map[string]string) (sourceConfig, error) {
|
||
var config sourceConfig
|
||
if binding == nil {
|
||
return config, errors.New("routing bindings are required")
|
||
}
|
||
config.Scopes = binding
|
||
app, err := containerEnv(ctx, opts.sourceApp)
|
||
if err != nil {
|
||
return config, err
|
||
}
|
||
if app["DB_DRIVER"] != "postgres" {
|
||
return config, errors.New("source database driver is not postgres")
|
||
}
|
||
address, err := containerAddress(ctx, opts.sourcePG)
|
||
if err != nil {
|
||
return config, err
|
||
}
|
||
dsn := strings.Join([]string{
|
||
"postgres://" + urlEscape(app["DB_USER"]) + ":" + urlEscape(app["DB_PASSWORD"]) +
|
||
"@" + address + ":" + app["DB_PORT"] + "/" + urlEscape(app["DB_NAME"]) + "?sslmode=disable",
|
||
}, "")
|
||
config.DatabaseURL = dsn
|
||
config.AppImageID, err = containerImageID(ctx, opts.sourceApp)
|
||
if err != nil {
|
||
return config, err
|
||
}
|
||
esAddress, err := containerAddress(ctx, opts.sourceES)
|
||
if err != nil {
|
||
return config, err
|
||
}
|
||
config.SearchIndex = app["ELASTICSEARCH_INDEX"]
|
||
config.Search = v071.SearchExporterConfig{BaseURL: "http://" + esAddress + ":9200", Index: app["ELASTICSEARCH_INDEX"],
|
||
AuthorizationToken: app["ELASTICSEARCH_PASSWORD"], PageSize: 500}
|
||
config.Object, config.ObjectDetail, err = objectStoreFromSourceEnv(app, opts.sourceS3Endpoint, opts.maxObjectBytes)
|
||
if err != nil {
|
||
return config, err
|
||
}
|
||
return config, nil
|
||
}
|
||
|
||
// objectStoreFromSourceEnv 把源应用的存储环境映射成对象存储配置。
|
||
//
|
||
// 只认 OSS 时，S3 兼容的 MinIO 目标（本机专用验收实例）会在发现阶段就被拒，
|
||
// 验收链只能跑在生产存储上——这不是安全性，而是能力缺口。两种都要能读，
|
||
// 读不出端点/桶就明确报错，不退回任何默认值。
|
||
func objectStoreFromSourceEnv(app map[string]string, endpointOverride string, maxObjectBytes int64) (v071.S3ObjectStoreConfig, string, error) {
|
||
storage := strings.ToLower(strings.TrimSpace(app["STORAGE_TYPE"]))
|
||
var config v071.S3ObjectStoreConfig
|
||
var endpoint, bucket string
|
||
switch storage {
|
||
case "oss":
|
||
config = v071.S3ObjectStoreConfig{Provider: "oss", Region: app["OSS_REGION"],
|
||
AccessKeyID: app["OSS_ACCESS_KEY"], SecretAccessKey: app["OSS_SECRET_KEY"],
|
||
Bucket: app["OSS_BUCKET_NAME"], MaxObjectBytes: maxObjectBytes}
|
||
endpoint, bucket = app["OSS_ENDPOINT"], app["OSS_BUCKET_NAME"]
|
||
if !strings.Contains(endpoint, "://") {
|
||
endpoint = "https://" + endpoint
|
||
}
|
||
case "minio":
|
||
accessKey := firstNonEmpty(app["MINIO_ACCESS_KEY_ID"], app["MINIO_ACCESS_KEY"])
|
||
secretKey := firstNonEmpty(app["MINIO_SECRET_ACCESS_KEY"], app["MINIO_SECRET_KEY"])
|
||
config = v071.S3ObjectStoreConfig{Provider: "minio", Region: app["MINIO_REGION"],
|
||
AccessKeyID: accessKey, SecretAccessKey: secretKey, Bucket: app["MINIO_BUCKET_NAME"],
|
||
MaxObjectBytes: maxObjectBytes}
|
||
endpoint, bucket = app["MINIO_ENDPOINT"], app["MINIO_BUCKET_NAME"]
|
||
if strings.TrimSpace(endpoint) == "" {
|
||
return v071.S3ObjectStoreConfig{}, "", errors.New("source minio endpoint is not configured")
|
||
}
|
||
scheme := "http"
|
||
if parseBool(app["MINIO_USE_SSL"]) {
|
||
scheme = "https"
|
||
}
|
||
if !strings.Contains(endpoint, "://") {
|
||
endpoint = scheme + "://" + endpoint
|
||
}
|
||
default:
|
||
return v071.S3ObjectStoreConfig{}, "", fmt.Errorf("unsupported source storage type %q", app["STORAGE_TYPE"])
|
||
}
|
||
if strings.TrimSpace(endpoint) == "" {
|
||
return v071.S3ObjectStoreConfig{}, "", errors.New("source object endpoint is not configured")
|
||
}
|
||
// 宿主机可达地址优先：容器里的 minio 服务名在宿主机上根本解析不到，
|
||
// 真跑时资源导出会直接失败（read library resource）。覆盖仍是"同一个桶"，
|
||
// 只换访问入口，不换来源。
|
||
if override := strings.TrimSpace(endpointOverride); override != "" {
|
||
if !strings.Contains(override, "://") {
|
||
return v071.S3ObjectStoreConfig{}, "", errors.New("source object endpoint override needs a scheme")
|
||
}
|
||
endpoint = override
|
||
}
|
||
config.Endpoint = endpoint
|
||
config.Secure = strings.HasPrefix(endpoint, "https://")
|
||
config.ForcePathStyle = storage == "minio"
|
||
if config.Bucket == "" {
|
||
return v071.S3ObjectStoreConfig{}, "", errors.New("source object bucket is not configured")
|
||
}
|
||
return config, "provider=" + config.Provider + " bucket=" + bucket + " endpoint=" + config.Endpoint, nil
|
||
}
|
||
|
||
func firstNonEmpty(values ...string) string {
|
||
for _, value := range values {
|
||
if strings.TrimSpace(value) != "" {
|
||
return value
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func parseBool(value string) bool {
|
||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||
case "1", "true", "yes", "on":
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
func urlEscape(value string) string {
|
||
replacer := strings.NewReplacer("%", "%25", ":", "%3A", "@", "%40", "/", "%2F", "?", "%3F", "#", "%23", "&", "%26", " ", "%20")
|
||
return replacer.Replace(value)
|
||
}
|
||
|
||
// scopeBindings 返回 scope→知识库 映射。治理 API 用的是 LIBRARY_PRODUCTION_KB_*，
|
||
// CLI 读的是治理库里的 routing bindings；同一台机器上两份不一致时，
|
||
// CLI 抓到的会是"配置里那个库"而不是 API 认为的目标——所以环境变量优先，
|
||
// 读不到才退回 bindings。两边讲同一份身份，验收才可比。
|
||
func scopeBindings(ctx context.Context, container string) (map[string]string, error) {
|
||
binding, err := readRoutingBindings(ctx, container)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
for scope, key := range map[string]string{"internal": "LIBRARY_PRODUCTION_KB_INTERNAL", "external": "LIBRARY_PRODUCTION_KB_EXTERNAL"} {
|
||
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
|
||
binding[scope] = value
|
||
}
|
||
}
|
||
return binding, nil
|
||
}
|
||
|
||
func readRoutingBindings(ctx context.Context, container string) (map[string]string, error) {
|
||
raw, err := dockerOutput(ctx, "exec", container, "psql", "-U", "governance", "-d", "governance", "-tAc",
|
||
"SELECT internal_kb_id||'|'||external_kb_id FROM routing_kb_bindings WHERE id='active'")
|
||
if err != nil {
|
||
return nil, errors.New("read routing bindings failed")
|
||
}
|
||
internal, external, ok := strings.Cut(strings.TrimSpace(raw), "|")
|
||
if !ok || internal == "" || external == "" {
|
||
return nil, errors.New("routing bindings are incomplete")
|
||
}
|
||
return map[string]string{"internal": internal, "external": external}, nil
|
||
}
|
||
|
||
func openSourceDatabase(ctx context.Context, config sourceConfig) (*pgxpool.Pool, error) {
|
||
parsed, err := pgxpool.ParseConfig(config.DatabaseURL)
|
||
if err != nil {
|
||
return nil, errors.New("invalid source database url")
|
||
}
|
||
// The capture chain never writes to the source. The read-only session makes
|
||
// that a database-enforced property rather than a convention.
|
||
parsed.ConnConfig.RuntimeParams["default_transaction_read_only"] = "on"
|
||
parsed.ConnConfig.RuntimeParams["statement_timeout"] = "120000"
|
||
pool, err := pgxpool.NewWithConfig(ctx, parsed)
|
||
if err != nil {
|
||
return nil, errors.New("open source database failed")
|
||
}
|
||
var readOnly string
|
||
if err = pool.QueryRow(ctx, "SHOW default_transaction_read_only").Scan(&readOnly); err != nil || readOnly != "on" {
|
||
pool.Close()
|
||
return nil, errors.New("source session is not read-only")
|
||
}
|
||
return pool, nil
|
||
}
|
||
|
||
func tenantAndKnowledgeBase(ctx context.Context, pool *pgxpool.Pool, knowledgeBaseID string) (uint64, error) {
|
||
var tenant uint64
|
||
err := pool.QueryRow(ctx, "SELECT tenant_id FROM knowledge_bases WHERE id=$1 AND deleted_at IS NULL", knowledgeBaseID).Scan(&tenant)
|
||
if err != nil || tenant == 0 {
|
||
return 0, errors.New("knowledge base not found in source")
|
||
}
|
||
return tenant, nil
|
||
}
|
||
|
||
func writerDeclaration(opts options) ([]string, map[string]string, error) {
|
||
writers := []string{}
|
||
for _, name := range strings.Split(opts.writers, ",") {
|
||
if trimmed := strings.TrimSpace(name); trimmed != "" {
|
||
writers = append(writers, trimmed)
|
||
}
|
||
}
|
||
if len(writers) == 0 {
|
||
return nil, nil, errors.New("no writer service declared")
|
||
}
|
||
nonWriters := map[string]string{}
|
||
for _, entry := range strings.Split(opts.nonWriters, ",") {
|
||
entry = strings.TrimSpace(entry)
|
||
if entry == "" {
|
||
continue
|
||
}
|
||
name, reason, ok := strings.Cut(entry, "=")
|
||
if !ok || strings.TrimSpace(reason) == "" {
|
||
return nil, nil, fmt.Errorf("non-writer %q needs service=reason", entry)
|
||
}
|
||
nonWriters[strings.TrimSpace(name)] = strings.TrimSpace(reason)
|
||
}
|
||
return writers, nonWriters, nil
|
||
}
|
||
|
||
func openFreeze(ctx context.Context, opts options) (*libraryfreeze.Coordinator, *libraryfreeze.ProjectController, error) {
|
||
writers, nonWriters, err := writerDeclaration(opts)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
if err = os.MkdirAll(opts.journalRoot, 0700); err != nil {
|
||
return nil, nil, err
|
||
}
|
||
if err = os.Chmod(opts.journalRoot, 0700); err != nil {
|
||
return nil, nil, err
|
||
}
|
||
control, err := libraryfreeze.NewProjectController(opts.dockerSocket, opts.project, writers, nonWriters, 2*time.Minute)
|
||
if err != nil {
|
||
return nil, nil, err
|
||
}
|
||
coordinator, err := libraryfreeze.Open(opts.journalRoot, opts.project, control)
|
||
if err != nil {
|
||
control.Close()
|
||
return nil, nil, err
|
||
}
|
||
return coordinator, control, nil
|
||
}
|
||
|
||
// captureMaintenanceConfig 把"这次封存动的是哪个实例"落成实例级维护目标。
|
||
//
|
||
// 为什么必须实例级：被代理的 WeKnora 收到 SIGTERM 会直接 exit(1)，写者的退出码
|
||
// 因此永远不能当作"在途请求已排空"。排空证据只能在真正的入口上取——关掉闸门、
|
||
// 等它把在途连接放完。没有闸门就没有证据，那种封存不允许以"库级窗口"的名义跑
|
||
// （真跑过：被 exit=1 拒掉，这是校验在起作用，不是要放松的地方）。
|
||
//
|
||
// 入口/队列的声明与容器内 worker 共用同一组 LIBRARY_TARGET_* 名称，
|
||
// 数据库/索引/桶则取自本次真实发现的源，避免"关的是 A 的门、读的是 B 的队列"。
|
||
func captureMaintenanceConfig(opts options, source sourceConfig, get func(string) string) (librarymaintenance.Config, error) {
|
||
target, err := librarymaintenance.TargetFromEnv(get)
|
||
if err != nil {
|
||
return librarymaintenance.Config{}, err
|
||
}
|
||
writers, nonWriters, err := writerDeclaration(opts)
|
||
if err != nil {
|
||
return librarymaintenance.Config{}, err
|
||
}
|
||
identity, err := librarymaintenance.DatabaseIdentity(source.DatabaseURL)
|
||
if err != nil {
|
||
return librarymaintenance.Config{}, err
|
||
}
|
||
return librarymaintenance.Config{
|
||
Instance: target.Instance,
|
||
Project: opts.project,
|
||
Writers: writers,
|
||
NonWriters: nonWriters,
|
||
DockerSocket: opts.dockerSocket,
|
||
IngressURL: target.IngressURL,
|
||
IngressService: target.IngressService,
|
||
IngressDrainFile: target.IngressDrainFile,
|
||
Database: identity,
|
||
Index: source.SearchIndex,
|
||
Bucket: source.Object.Bucket,
|
||
// 后台静默检查连的是"正在被封的这个源"：未结清的后台账本与队列都在它身上。
|
||
DatabaseURL: source.DatabaseURL,
|
||
RedisAddr: target.RedisAddr,
|
||
RedisPassword: target.RedisPassword,
|
||
RedisDB: target.RedisDB,
|
||
}, nil
|
||
}
|
||
|
||
// openMaintenance 用实例级目标装配协调器：入口闸门、后台静默检查、目标指纹、
|
||
// 入口探针全部来自上面那份声明，与发布链路用的是同一套实现（不是第二份协议）。
|
||
func openMaintenance(ctx context.Context, opts options, source sourceConfig) (*libraryfreeze.Coordinator, *libraryfreeze.ProjectController, *librarymaintenance.Manager, error) {
|
||
config, err := captureMaintenanceConfig(opts, source, os.Getenv)
|
||
if err != nil {
|
||
return nil, nil, nil, err
|
||
}
|
||
manager, err := librarymaintenance.New(config)
|
||
if err != nil {
|
||
return nil, nil, nil, err
|
||
}
|
||
// 拓扑先过关：写入者发布了端口 = 存在绕闸直连，这样的"维护"只关了半扇门。
|
||
if err = manager.VerifyTopology(ctx); err != nil {
|
||
manager.Close()
|
||
return nil, nil, nil, fmt.Errorf("目标拓扑不满足唯一入口约束，拒绝开维护窗口：%w", err)
|
||
}
|
||
coordinator, control, err := manager.Open(opts.journalRoot)
|
||
if err != nil {
|
||
manager.Close()
|
||
return nil, nil, nil, err
|
||
}
|
||
return coordinator, control, manager, nil
|
||
}
|
||
|
||
// discoverMaintenanceSource 只读"这个源是什么"，不碰治理库的 scope 绑定：
|
||
// 恢复/收尾不该因为治理库读不到就被卡住。
|
||
func discoverMaintenanceSource(ctx context.Context, opts options) (sourceConfig, error) {
|
||
return discoverSource(ctx, opts, map[string]string{})
|
||
}
|
||
|
||
func openArchive(opts options) (*librarysnapshot.Archive, error) {
|
||
if err := os.MkdirAll(opts.archiveRoot, 0700); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := os.Chmod(opts.archiveRoot, 0700); err != nil {
|
||
return nil, err
|
||
}
|
||
return librarysnapshot.Open(opts.archiveRoot, 256*1024*1024)
|
||
}
|
||
|
||
// ---------------------------------------------------------------- commands
|
||
|
||
func runStatus(ctx context.Context, opts options) error {
|
||
report := map[string]any{}
|
||
binding, bindingErr := scopeBindings(ctx, opts.governancePG)
|
||
var config sourceConfig
|
||
var err error
|
||
if bindingErr != nil {
|
||
err = bindingErr
|
||
} else {
|
||
config, err = discoverSource(ctx, opts, binding)
|
||
}
|
||
if err != nil {
|
||
report["source_discovery"] = err.Error()
|
||
} else {
|
||
report["source"] = map[string]any{
|
||
"app_image": config.AppImageID,
|
||
"scopes": config.Scopes,
|
||
"search_index": config.SearchIndex,
|
||
"elasticsearch": config.Search.BaseURL,
|
||
"object_store": config.ObjectDetail,
|
||
"database": "postgres (read-only session, credentials withheld)",
|
||
"credentials_logged": false,
|
||
}
|
||
pool, poolErr := openSourceDatabase(ctx, config)
|
||
if poolErr != nil {
|
||
report["database"] = poolErr.Error()
|
||
} else {
|
||
defer pool.Close()
|
||
settled := "settled"
|
||
if settleErr := v071.LibrarySourceSettled(ctx, pool); settleErr != nil {
|
||
settled = settleErr.Error()
|
||
}
|
||
report["settlement"] = settled
|
||
counts := map[string]int{}
|
||
for _, scope := range []string{"internal", "external"} {
|
||
kb := config.Scopes[scope]
|
||
tenant, tenantErr := tenantAndKnowledgeBase(ctx, pool, kb)
|
||
if tenantErr != nil {
|
||
counts[scope+"_error"] = -1
|
||
continue
|
||
}
|
||
pg, exportErr := v071.ExportLibrary(ctx, pool, v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb})
|
||
if exportErr != nil {
|
||
counts[scope+"_error"] = -1
|
||
continue
|
||
}
|
||
for _, table := range pg.Tables {
|
||
if table.Name == "chunks" || table.Name == "knowledges" {
|
||
counts[scope+"_"+table.Name] = len(table.Rows)
|
||
}
|
||
}
|
||
_ = kb
|
||
}
|
||
report["postgres_rows"] = counts
|
||
}
|
||
}
|
||
archive, err := openArchive(opts)
|
||
if err != nil {
|
||
report["archive"] = err.Error()
|
||
} else {
|
||
defer archive.Close()
|
||
usage, usageErr := archive.Usage()
|
||
if usageErr != nil {
|
||
report["archive"] = usageErr.Error()
|
||
} else {
|
||
total := int64(0)
|
||
for _, item := range usage {
|
||
total += item.UniqueBytes
|
||
}
|
||
report["archive"] = map[string]any{"root": opts.archiveRoot, "snapshots": len(usage), "stored_bytes": total}
|
||
}
|
||
}
|
||
if coordinator, control, freezeErr := openFreeze(ctx, opts); freezeErr != nil {
|
||
report["freeze"] = freezeErr.Error()
|
||
} else {
|
||
defer coordinator.Close()
|
||
defer control.Close()
|
||
items, inventoryErr := control.Inventory(ctx)
|
||
if inventoryErr != nil {
|
||
report["freeze"] = inventoryErr.Error()
|
||
} else {
|
||
group := []map[string]any{}
|
||
for _, item := range items {
|
||
group = append(group, map[string]any{"id": item.ID, "running": item.Running, "healthy": item.Healthy})
|
||
}
|
||
report["freeze"] = map[string]any{"project": opts.project, "writers": group, "journal": opts.journalRoot}
|
||
}
|
||
}
|
||
printJSON(report)
|
||
return nil
|
||
}
|
||
|
||
func runPreflight(ctx context.Context, opts options) error {
|
||
if opts.scope != "internal" && opts.scope != "external" {
|
||
return errors.New("-scope must be internal or external")
|
||
}
|
||
binding, err := scopeBindings(ctx, opts.governancePG)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
config, err := discoverSource(ctx, opts, binding)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
pool, err := openSourceDatabase(ctx, config)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer pool.Close()
|
||
kb := config.Scopes[opts.scope]
|
||
tenant, err := tenantAndKnowledgeBase(ctx, pool, kb)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
released, err := parseReleasedDeadLetters(opts.releasedDeadLetters)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err = v071.LibrarySourceSettledWithReleases(ctx, pool, released); err != nil {
|
||
return err
|
||
}
|
||
search, err := v071.NewSearchExporter(config.Search)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
contract, err := compatibilityContract(ctx, pool, search, config.SearchIndex, tenant, kb)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
pg, err := v071.ExportLibrary(ctx, pool, v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
es, err := search.ExportLibrary(ctx, v071.SearchIdentity{TenantID: tenant, KnowledgeBaseID: kb}, opts.maxDocuments)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
validation := "candidate-valid"
|
||
if err = v071.ValidateLibraryCandidateForPreflight(v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb}, pg, es); err != nil {
|
||
validation = err.Error()
|
||
}
|
||
tables := map[string]int{}
|
||
for _, table := range pg.Tables {
|
||
tables[table.Name] = len(table.Rows)
|
||
}
|
||
printJSON(map[string]any{"scope": opts.scope, "knowledge_base_id": kb, "tenant_id": tenant,
|
||
"compatibility": contract, "postgres_rows": tables, "index_documents": len(es.Documents),
|
||
"mapping_sha256": es.MappingSHA256, "validation": validation, "writes": 0, "writers_stopped": 0})
|
||
if validation != "candidate-valid" {
|
||
return errors.New("candidate validation failed: " + validation)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func compatibilityContract(ctx context.Context, pool *pgxpool.Pool, search *v071.SearchExporter, index string, tenant uint64, kb string) (v071.LibrarySnapshotCompatibility, error) {
|
||
inspector, err := v071.NewTargetContractInspector(v071.TargetContractInspectorConfig{DB: pool, Mapping: search,
|
||
KnowledgeBaseID: kb, TenantID: tenant, IndexName: index, ReadTestedBase: true})
|
||
if err != nil {
|
||
return v071.LibrarySnapshotCompatibility{}, err
|
||
}
|
||
contract, err := inspector.Contract(ctx, releaseworkflow.PlanItem{TestKnowledgeBaseID: kb})
|
||
if err != nil {
|
||
return v071.LibrarySnapshotCompatibility{}, err
|
||
}
|
||
return v071.LibrarySnapshotCompatibility{WeKnoraVersion: contract.WeKnoraVersion, SchemaVersion: contract.MigrationVersion,
|
||
RetrieverEngine: contract.RetrieverEngine, EmbeddingModel: contract.EmbeddingModel,
|
||
EmbeddingDimension: contract.EmbeddingDimension, IndexMappingSHA256: contract.ElasticsearchMappingSHA256}, nil
|
||
}
|
||
|
||
func runCapture(ctx context.Context, opts options) error {
|
||
if opts.scope != "internal" && opts.scope != "external" {
|
||
return errors.New("-scope must be internal or external")
|
||
}
|
||
released, err := parseReleasedDeadLetters(opts.releasedDeadLetters)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if opts.dryRun {
|
||
return runPreflight(ctx, opts)
|
||
}
|
||
job := opts.job
|
||
if job == "" {
|
||
job = "lib-" + opts.scope + "-" + time.Now().UTC().Format("20060102T150405")
|
||
}
|
||
archive, err := openArchive(opts)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer archive.Close()
|
||
binding, err := scopeBindings(ctx, opts.governancePG)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
config, err := discoverSource(ctx, opts, binding)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// 维护窗口必须先按"实例级"接好：入口闸门给出在途排空证据，后台静默检查
|
||
// 给出停写者那一刻队列已空。缺任何一项都拒绝封存——没有证据的封存不是候选。
|
||
coordinator, control, manager, err := openMaintenance(ctx, opts, config)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer manager.Close()
|
||
defer coordinator.Close()
|
||
defer control.Close()
|
||
if err = coordinator.Recover(ctx); err != nil {
|
||
return fmt.Errorf("pending freeze requires recovery: %w", err)
|
||
}
|
||
// Replay is answered from the persisted journal + sealed archive: the same job
|
||
// re-used with the same scope/binding returns the original candidate without
|
||
// reading the source or stopping a writer, and a job re-used with different
|
||
// parameters is refused instead of silently sealing under a stale identity.
|
||
existing, found, err := coordinator.Inspect(job)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
scopedKB := binding[opts.scope]
|
||
if found {
|
||
manifest, err := replayCapture(job, opts.scope, scopedKB, existing, archive)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return printCapture(job, existing.SnapshotID, manifest, true)
|
||
}
|
||
pool, err := openSourceDatabase(ctx, config)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer pool.Close()
|
||
kb := config.Scopes[opts.scope]
|
||
tenant, err := tenantAndKnowledgeBase(ctx, pool, kb)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
search, err := v071.NewSearchExporter(config.Search)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
objects, err := v071.NewS3ObjectStore(config.Object)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
contract, err := compatibilityContract(ctx, pool, search, config.SearchIndex, tenant, kb)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// The contract above is read before the maintenance window. Re-read it inside
|
||
// the stopped window and refuse if a model/index change happened in between:
|
||
// sealing a candidate whose contract no longer matches the source would hide
|
||
// exactly the change a later production sync cannot reproduce.
|
||
verifyContract := func(ctx context.Context) error {
|
||
live, err := compatibilityContract(ctx, pool, search, config.SearchIndex, tenant, kb)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if live != contract {
|
||
return errors.New("source contract changed between preflight and the frozen window; capture aborted")
|
||
}
|
||
return nil
|
||
}
|
||
// sourceReady proves the declared writer group is complete and no other
|
||
// running container uses a declared writer image. Recorded work settlement is
|
||
// enforced separately by CaptureLibrary; neither check can see unrecorded or
|
||
// remote work, so this is not a cross-system consistency proof.
|
||
sourceReady := func(ctx context.Context) error {
|
||
items, err := control.Inventory(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(items) == 0 {
|
||
return errors.New("writer group is empty")
|
||
}
|
||
declared := make(map[string]bool, len(items))
|
||
for _, item := range items {
|
||
declared[item.ID] = true
|
||
}
|
||
images, err := writerImageIDs(ctx, items)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
app, err := containerEnv(ctx, opts.sourceApp)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return checkNoUndeclaredWriter(ctx, images, declared, opts.project, sourceTargetOf(app))
|
||
}
|
||
result, err := v071.CaptureLibrary(ctx, coordinator, archive, pool, search, objects, sourceReady,
|
||
v071.LibraryCaptureRequest{JobID: job, Scope: opts.scope,
|
||
Identity: v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb},
|
||
Compatibility: contract, CreatedAt: time.Now().UTC(), MaxDocuments: opts.maxDocuments,
|
||
VerifyContract: verifyContract, ReleasedDeadLetters: released})
|
||
if err != nil {
|
||
printJSON(map[string]any{"job": job, "state": result.State, "snapshot_id": result.SnapshotID, "error": err.Error()})
|
||
return err
|
||
}
|
||
manifest, err := archive.Load(result.SnapshotID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return printCapture(job, result.SnapshotID, manifest, false)
|
||
}
|
||
|
||
// replayCapture answers a repeated job from persisted state. It never reads the
|
||
// source and never stops a writer; conflicting parameters are refused.
|
||
func replayCapture(job, scope, expectedKB string, result libraryfreeze.Result, archive *librarysnapshot.Archive) (librarysnapshot.Manifest, error) {
|
||
if result.State != "succeeded" || result.SnapshotID == "" {
|
||
return librarysnapshot.Manifest{}, fmt.Errorf("job %q already exists with state %q: recover it or use a new job id", job, result.State)
|
||
}
|
||
manifest, err := archive.Load(result.SnapshotID)
|
||
if err != nil {
|
||
return librarysnapshot.Manifest{}, errors.New("captured library archive no longer verifies")
|
||
}
|
||
if manifest.ValidationPolicy != librarysnapshot.ArtifactAcceptancePolicy {
|
||
return librarysnapshot.Manifest{}, errors.New("existing job was sealed under a different validation policy")
|
||
}
|
||
if manifest.Scope != scope {
|
||
return librarysnapshot.Manifest{}, fmt.Errorf("job %q was captured for scope %q, not %q", job, manifest.Scope, scope)
|
||
}
|
||
if manifest.KnowledgeBaseID != expectedKB {
|
||
return librarysnapshot.Manifest{}, fmt.Errorf("job %q was captured for knowledge base %q, which is no longer the %q binding", job, manifest.KnowledgeBaseID, scope)
|
||
}
|
||
return manifest, nil
|
||
}
|
||
|
||
func printCapture(job, snapshotID string, manifest librarysnapshot.Manifest, replay bool) error {
|
||
parts := make([]map[string]any, 0, len(manifest.Parts))
|
||
var total int64
|
||
for _, part := range manifest.Parts {
|
||
parts = append(parts, map[string]any{"kind": part.Kind, "name": part.Name, "count": part.Count, "bytes": part.Bytes, "sha256": part.SHA256})
|
||
total += part.Bytes
|
||
}
|
||
sort.Slice(parts, func(i, j int) bool {
|
||
return fmt.Sprint(parts[i]["kind"])+fmt.Sprint(parts[i]["name"]) < fmt.Sprint(parts[j]["kind"])+fmt.Sprint(parts[j]["name"])
|
||
})
|
||
printJSON(map[string]any{"job": job, "state": "succeeded", "snapshot_id": snapshotID,
|
||
"scope": manifest.Scope, "knowledge_base_id": manifest.KnowledgeBaseID, "tenant_id": manifest.TenantID,
|
||
"compatibility": manifest.Compatibility, "parts": parts, "logical_bytes": total,
|
||
"validation_policy": manifest.ValidationPolicy, "writers_restored": true,
|
||
"replay": replay, "source_read": !replay, "writers_stopped": !replay})
|
||
return nil
|
||
}
|
||
|
||
func writerImageIDs(ctx context.Context, items []libraryfreeze.Service) ([]string, error) {
|
||
images := make([]string, 0, len(items))
|
||
seen := map[string]bool{}
|
||
for _, item := range items {
|
||
image, err := containerImageOf(ctx, item.ID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !seen[image] {
|
||
seen[image] = true
|
||
images = append(images, image)
|
||
}
|
||
}
|
||
return images, nil
|
||
}
|
||
|
||
// checkNoUndeclaredWriter refuses to run when a running container uses a declared
|
||
// writer image but is not part of the declared group: such a container could keep
|
||
// writing during the maintenance window.
|
||
func checkNoUndeclaredWriter(ctx context.Context, images []string, declared map[string]bool,
|
||
sourceProject string, source map[string]string) error {
|
||
for _, image := range images {
|
||
// 必须 --no-trunc：`docker ps` 默认只给 12 位短 ID，和 declared 里的
|
||
// 64 位容器 ID 永远不相等——那样连"已声明的写入者"都会被判成未声明。
|
||
out, err := dockerOutput(ctx, "ps", "-aq", "--no-trunc", "--filter", "ancestor="+image)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, id := range strings.Fields(out) {
|
||
if declaredWriter(declared, id) {
|
||
continue
|
||
}
|
||
project, err := dockerOutput(ctx, "inspect", "--format", `{{index .Config.Labels "com.docker.compose.project"}}`, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
env, err := containerEnv(ctx, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if conflict, reason := undeclaredWriterConflict(false, project, sourceProject, env, source); conflict {
|
||
return fmt.Errorf("container %s uses a declared writer image but is not in the declared writer group: %s", id[:12], reason)
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// declaredWriter 同时接受完整 ID 与短 ID：证据里两种写法都出现过，
|
||
// 只比一种会把同一容器判成两个身份。
|
||
func declaredWriter(declared map[string]bool, id string) bool {
|
||
if declared[id] {
|
||
return true
|
||
}
|
||
for key := range declared {
|
||
if strings.HasPrefix(key, id) || strings.HasPrefix(id, key) {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// sourceTargetOf 只取"写到哪儿"的身份：同一个数据库或同一个队列才是真冲突。
|
||
func sourceTargetOf(env map[string]string) map[string]string {
|
||
return map[string]string{
|
||
"DB_HOST": env["DB_HOST"], "DB_NAME": env["DB_NAME"], "DB_PORT": env["DB_PORT"],
|
||
"REDIS_ADDR": env["REDIS_ADDR"], "REDIS_NAMESPACE": env["REDIS_NAMESPACE"],
|
||
}
|
||
}
|
||
|
||
// undeclaredWriterConflict 判断一个同镜像容器是否构成"未声明的并发写入者"。
|
||
//
|
||
// 同一 compose 项目或没有项目标签 ⇒ 无法证明隔离，直接冲突。
|
||
// 另一个项目 ⇒ 只有数据目标（数据库/队列）与源相同才算冲突：
|
||
// 镜像相同不等于写同一个库，本机共享业务栈与专用验收实例就是这种关系。
|
||
func undeclaredWriterConflict(declared bool, project, sourceProject string, env, source map[string]string) (bool, string) {
|
||
if declared {
|
||
return false, ""
|
||
}
|
||
if project == "" {
|
||
return true, "没有 compose 项目标签，无法证明它不写同一个目标"
|
||
}
|
||
if project == sourceProject {
|
||
return true, "属于同一 compose 项目，却不是声明的写入服务"
|
||
}
|
||
if source["DB_HOST"] != "" && env["DB_HOST"] == source["DB_HOST"] && env["DB_NAME"] == source["DB_NAME"] {
|
||
return true, "另一个项目的同镜像容器指向同一个源数据库"
|
||
}
|
||
if source["REDIS_ADDR"] != "" && env["REDIS_ADDR"] == source["REDIS_ADDR"] &&
|
||
env["REDIS_NAMESPACE"] == source["REDIS_NAMESPACE"] {
|
||
return true, "另一个项目的同镜像容器共用同一个源队列"
|
||
}
|
||
return false, "另一套隔离栈（数据目标不同）"
|
||
}
|
||
|
||
func containerImageOf(ctx context.Context, id string) (string, error) {
|
||
image, err := dockerOutput(ctx, "inspect", "--format", "{{.Image}}", id)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if image == "" {
|
||
return "", errors.New("declared writer has no image identity")
|
||
}
|
||
return image, nil
|
||
}
|
||
func runList(opts options) error {
|
||
archive, err := openArchive(opts)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer archive.Close()
|
||
usage, err := archive.Usage()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
total, err := archive.ContentUsage()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
items := make([]map[string]any, 0, len(usage))
|
||
for _, item := range usage {
|
||
manifest, err := archive.Load(item.VersionID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
items = append(items, map[string]any{"snapshot_id": item.VersionID, "scope": manifest.Scope,
|
||
"knowledge_base_id": manifest.KnowledgeBaseID, "created_at": manifest.CreatedAt,
|
||
"logical_bytes": item.LogicalBytes, "unique_bytes": item.UniqueBytes, "shared_bytes": item.SharedBytes,
|
||
"parts": len(manifest.Parts), "policy": manifest.ValidationPolicy})
|
||
}
|
||
printJSON(map[string]any{"archive": opts.archiveRoot, "snapshots": items,
|
||
// 跨版本去重后的内容字节；文件系统事实单列，任何一项都不是可回收空间。
|
||
"versions": total.Versions, "content_bytes": total.ContentBytes,
|
||
"blob_file_bytes": total.BlobFileBytes, "blob_allocated_bytes": total.BlobAllocatedBytes,
|
||
"unreferenced_blobs": total.UnreferencedBlobs, "unreferenced_bytes": total.UnreferencedBytes,
|
||
"manifest_file_bytes": total.ManifestFileBytes, "manifest_allocated_bytes": total.ManifestAllocatedBytes})
|
||
return nil
|
||
}
|
||
|
||
func runVerify(opts options) error {
|
||
if opts.snapshot == "" {
|
||
return errors.New("-snapshot is required")
|
||
}
|
||
archive, err := openArchive(opts)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer archive.Close()
|
||
manifest, err := archive.Load(opts.snapshot)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
parts := make([]map[string]any, 0, len(manifest.Parts))
|
||
for _, part := range manifest.Parts {
|
||
parts = append(parts, map[string]any{"kind": part.Kind, "name": part.Name, "count": part.Count, "bytes": part.Bytes, "sha256": part.SHA256})
|
||
}
|
||
printJSON(map[string]any{"snapshot_id": opts.snapshot, "verified": true, "format": manifest.Format,
|
||
"policy": manifest.ValidationPolicy, "scope": manifest.Scope, "knowledge_base_id": manifest.KnowledgeBaseID,
|
||
"tenant_id": manifest.TenantID, "created_at": manifest.CreatedAt, "compatibility": manifest.Compatibility, "parts": parts})
|
||
return nil
|
||
}
|
||
|
||
func runRecover(ctx context.Context, opts options) error {
|
||
// 收尾也必须走同一份实例级目标：崩溃留下的窗口是"入口已关、写入者已停"，
|
||
// 恢复要用闸门把入口真的重新打开，没有闸门就没有"目标已恢复"的证据。
|
||
source, err := discoverMaintenanceSource(ctx, opts)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
coordinator, control, manager, err := openMaintenance(ctx, opts, source)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer manager.Close()
|
||
defer coordinator.Close()
|
||
defer control.Close()
|
||
if err = coordinator.Recover(ctx); err != nil {
|
||
return err
|
||
}
|
||
items, err := control.Inventory(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
group := []map[string]any{}
|
||
for _, item := range items {
|
||
group = append(group, map[string]any{"id": item.ID, "running": item.Running, "healthy": item.Healthy})
|
||
}
|
||
printJSON(map[string]any{"recovered": true, "writers": group})
|
||
return nil
|
||
}
|
||
|
||
func printJSON(value any) {
|
||
encoder := json.NewEncoder(os.Stdout)
|
||
encoder.SetIndent("", " ")
|
||
_ = encoder.Encode(value)
|
||
}
|
||
|
||
// parseReleasedDeadLetters 把逗号分隔的指纹转成列表，并在本地先校验形状：一个
|
||
// 打错的指纹不应该走到源库才失败，更不应该被当成通配符。
|
||
func parseReleasedDeadLetters(raw string) ([]string, error) {
|
||
out := []string{}
|
||
for _, item := range strings.Split(raw, ",") {
|
||
item = strings.TrimSpace(item)
|
||
if item == "" {
|
||
continue
|
||
}
|
||
if !librarydeadletter.ValidFingerprint(item) {
|
||
return nil, errors.New("released dead letter entries must be sha256:<64 hex> fingerprints")
|
||
}
|
||
out = append(out, item)
|
||
}
|
||
if len(out) > 64 {
|
||
return nil, errors.New("at most 64 dead letter fingerprints per capture")
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// runDeadLetters 只读：打印整库死信原始行与观测摘要。治理平台的解除记录绑定这份
|
||
// 摘要；平台自己不连源库，因此观测必须由能读源库的一方生成并原样提交。
|
||
func runDeadLetters(ctx context.Context, opts options) error {
|
||
if opts.scope != "internal" && opts.scope != "external" {
|
||
return errors.New("-scope must be internal or external")
|
||
}
|
||
binding, err := scopeBindings(ctx, opts.governancePG)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
config, err := discoverSource(ctx, opts, binding)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
pool, err := openSourceDatabase(ctx, config)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer pool.Close()
|
||
kb := config.Scopes[opts.scope]
|
||
tenant, err := tenantAndKnowledgeBase(ctx, pool, kb)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
payload, err := v071.ObserveDeadLetters(ctx, pool, opts.project, tenant, kb)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
digest := librarydeadletter.Fingerprint(payload)
|
||
printJSON(map[string]any{"observation": json.RawMessage(payload), "observation_digest": digest})
|
||
return nil
|
||
}
|