From 4d6ecda3102e9f3d93d828a2ca8aa99d29669750 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A7=9A=E7=AB=A0=E6=B5=A9?= Date: Thu, 29 Oct 2026 06:09:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(libraryctl):=20=E6=BA=90=E5=AF=B9=E8=B1=A1?= =?UTF-8?q?=E5=AD=98=E5=82=A8=E7=AB=AF=E7=82=B9=E5=8F=AF=E6=8C=89=E5=AE=BF?= =?UTF-8?q?=E4=B8=BB=E6=9C=BA=E5=9C=B0=E5=9D=80=E8=A6=86=E7=9B=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 容器 env 里的 MINIO_ENDPOINT 往往是服务名（langfuse-minio:9000），封存跑在宿主机上 解析不到，真实封存报 read library resource 并整体失败。 新增 -source-s3-endpoint：只换访问入口，桶与对象键不变；缺 scheme 直接拒绝； status 会打印实际使用的 endpoint。 试过：make verify 15 项全绿；新增覆盖用例（生效 + 拒绝无 scheme）。 --- CHANGELOG.md | 14 +++++++++ cmd/libraryctl/discover_test.go | 38 ++++++++++++++++++++---- cmd/libraryctl/main.go | 51 +++++++++++++++++++++------------ docs/library-capture.md | 5 ++++ 4 files changed, 83 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a45a3fe..76096c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,8 +17,22 @@ - `scripts/library-agent-install.sh` 强制 `agent.env`（0600，`EnvironmentFile=`），并把源容器/ 项目/写入者声明做成可覆盖变量；没有 env 文件直接 `AGENT_ENV_MISSING` 退出。 +- 主机侧封存新增 `-source-s3-endpoint`：容器里的对象存储端点常常只是服务名 + （`langfuse-minio:9000`），宿主机解析不到，资源导出会失败（真跑报 `read library resource`）。 + 覆盖只换访问入口，桶/对象键不变；`status` 如实打印实际 endpoint。 +- **封存 API 的队列作业 id 超列宽**：`library-capture--` 是 45 字符，而 + `jobs.id` 是 `varchar(36)`，真实请求一律 500（`value too long`）。此前页面停在"未接线" + 从未走到这里；改为 `libcap--` 并用测试锁住列宽。 +- **索引范围过滤漏掉 app 写入的正文**：v0.7.1 的 app 自己写索引不带 `tenant_id`， + 治理侧同步才补齐；只按 `term tenant_id` 过滤会把整库正文漏掉（真跑：索引有 2 条、 + 导出 0 条、候选校验报 legacy coverage）。改为"本租户 或 未写租户"，写了别的租户仍排除。 + ### Tests +- `internal/httpapi/library_capture_test.go`：队列作业 id 必须放得进 `varchar(36)` 且形状固定。 +- `internal/releaseadapter/v071/search_tenant_filter_test.go`：接受未写租户、排除他租户、 + 无租户字段时不造条件、整库过滤仍带知识库身份。 +- `cmd/libraryctl/discover_test.go`：端点覆盖生效且必须带 scheme。 - `internal/librarymaintenance/env_test.go`：整份声明解析、逐项点名缺失、空白拒绝、 口令空格保留、非法队列库号。 - `cmd/libraryctl/maintenance_test.go`：入口/队列/源身份绑到同一份配置、数据库身份不泄漏口令、 diff --git a/cmd/libraryctl/discover_test.go b/cmd/libraryctl/discover_test.go index 54a59c1..6b63756 100644 --- a/cmd/libraryctl/discover_test.go +++ b/cmd/libraryctl/discover_test.go @@ -1,6 +1,9 @@ package main -import "testing" +import ( + "strings" + "testing" +) // 目标实例可以是 OSS，也可以是 S3 兼容的 MinIO（本机专用验收实例就是 minio）。 // 只认 oss 时，任何 minio 目标都会在发现阶段直接失败，等于这条验收链只能在生产存储上跑。 @@ -14,7 +17,7 @@ func TestObjectStoreFromSourceEnvAcceptsMinIO(t *testing.T) { "MINIO_BUCKET_NAME": "kgmaint-1028a", "MINIO_PATH_PREFIX": "kgmaint/", } - config, detail, err := objectStoreFromSourceEnv(app, 1<<20) + config, detail, err := objectStoreFromSourceEnv(app, "", 1<<20) if err != nil { t.Fatalf("minio: %v", err) } @@ -29,12 +32,35 @@ func TestObjectStoreFromSourceEnvAcceptsMinIO(t *testing.T) { } } +// 容器里的 minio 服务名在宿主机上解析不到：真跑时资源导出会失败（read library resource）。 +// 覆盖只换访问入口（同一个桶），不换来源；没有 scheme 的覆盖直接拒绝。 +func TestObjectStoreFromSourceEnvOverridesEndpointForHostSideCapture(t *testing.T) { + app := map[string]string{ + "STORAGE_TYPE": "minio", "MINIO_ENDPOINT": "langfuse-minio:9000", "MINIO_USE_SSL": "false", + "MINIO_ACCESS_KEY_ID": "key", "MINIO_SECRET_ACCESS_KEY": "secret", + "MINIO_BUCKET_NAME": "kgmaint-1028a", "MINIO_PATH_PREFIX": "kgmaint/", + } + config, detail, err := objectStoreFromSourceEnv(app, "http://127.0.0.1:19100", 1<<20) + if err != nil { + t.Fatalf("override: %v", err) + } + if config.Endpoint != "http://127.0.0.1:19100" || config.Bucket != "kgmaint-1028a" { + t.Fatalf("覆盖后配置不对：%+v", config) + } + if !strings.Contains(detail, "endpoint=http://127.0.0.1:19100") { + t.Fatalf("状态里没有如实写出用的端点：%q", detail) + } + if _, _, err := objectStoreFromSourceEnv(app, "127.0.0.1:19100", 1<<20); err == nil { + t.Fatal("没有 scheme 的端点覆盖被接受了") + } +} + func TestObjectStoreFromSourceEnvAcceptsMinIOOverTLS(t *testing.T) { app := map[string]string{ "STORAGE_TYPE": "minio", "MINIO_ENDPOINT": "minio.internal:9000", "MINIO_USE_SSL": "true", "MINIO_ACCESS_KEY": "k2", "MINIO_SECRET_KEY": "s2", "MINIO_BUCKET_NAME": "bucket", } - config, _, err := objectStoreFromSourceEnv(app, 1) + config, _, err := objectStoreFromSourceEnv(app, "", 1) if err != nil { t.Fatalf("minio tls: %v", err) } @@ -48,7 +74,7 @@ func TestObjectStoreFromSourceEnvKeepsOSS(t *testing.T) { "STORAGE_TYPE": "oss", "OSS_ENDPOINT": "oss-cn-shenzhen.aliyuncs.com", "OSS_REGION": "cn-shenzhen", "OSS_ACCESS_KEY": "ak", "OSS_SECRET_KEY": "sk", "OSS_BUCKET_NAME": "prod", } - config, _, err := objectStoreFromSourceEnv(app, 1) + config, _, err := objectStoreFromSourceEnv(app, "", 1) if err != nil { t.Fatalf("oss: %v", err) } @@ -58,10 +84,10 @@ func TestObjectStoreFromSourceEnvKeepsOSS(t *testing.T) { } func TestObjectStoreFromSourceEnvRejectsUnknownStorage(t *testing.T) { - if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "cos"}, 1); err == nil { + if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "cos"}, "", 1); err == nil { t.Fatal("expected error for unsupported storage type") } - if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "minio"}, 1); err == nil { + if _, _, err := objectStoreFromSourceEnv(map[string]string{"STORAGE_TYPE": "minio"}, "", 1); err == nil { t.Fatal("expected error when minio endpoint is missing") } } diff --git a/cmd/libraryctl/main.go b/cmd/libraryctl/main.go index 2a28ea1..c04b8b0 100644 --- a/cmd/libraryctl/main.go +++ b/cmd/libraryctl/main.go @@ -40,22 +40,25 @@ import ( ) type options struct { - sourceApp string - sourcePG string - sourceES string - governancePG string - project string - writers string - nonWriters string - archiveRoot string - journalRoot string - dockerSocket string - maxDocuments int - maxObjectBytes int64 - scope string - job string - dryRun bool - snapshot string + 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 } func main() { @@ -156,6 +159,7 @@ func parseFlags(command string, args []string) (options, []string, error) { 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") @@ -265,7 +269,7 @@ func discoverSource(ctx context.Context, opts options, binding map[string]string 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.maxObjectBytes) + config.Object, config.ObjectDetail, err = objectStoreFromSourceEnv(app, opts.sourceS3Endpoint, opts.maxObjectBytes) if err != nil { return config, err } @@ -277,7 +281,7 @@ func discoverSource(ctx context.Context, opts options, binding map[string]string // 只认 OSS 时，S3 兼容的 MinIO 目标（本机专用验收实例）会在发现阶段就被拒， // 验收链只能跑在生产存储上——这不是安全性，而是能力缺口。两种都要能读， // 读不出端点/桶就明确报错，不退回任何默认值。 -func objectStoreFromSourceEnv(app map[string]string, maxObjectBytes int64) (v071.S3ObjectStoreConfig, string, error) { +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 @@ -313,13 +317,22 @@ func objectStoreFromSourceEnv(app map[string]string, maxObjectBytes int64) (v071 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, nil + return config, "provider=" + config.Provider + " bucket=" + bucket + " endpoint=" + config.Endpoint, nil } func firstNonEmpty(values ...string) string { diff --git a/docs/library-capture.md b/docs/library-capture.md index 1a634e8..ce7ece3 100644 --- a/docs/library-capture.md +++ b/docs/library-capture.md @@ -58,6 +58,11 @@ LIBRARY_TARGET_INGRESS_DRAIN_FILE 闸门写排空证据的宿主机文件 LIBRARY_TARGET_REDIS_ADDR/_PASSWORD/_DB 后台队列（asynq）地址 ``` +另外 `-source-s3-endpoint ` 用来覆盖"源对象存储的访问入口"：容器里写的 +`MINIO_ENDPOINT` 往往是服务名（`langfuse-minio:9000`），封存跑在宿主机上时解析不到， +资源导出会失败（`read library resource`）。覆盖只换入口，桶与对象键不变；`status` +会如实打印实际使用的 endpoint。 + 为什么不再保留"库级窗口"：被代理的 WeKnora 收到 SIGTERM 会直接 `exit(1)`，写者退出码 永远不能当作"在途请求已排空"。排空证据只能在真正的入口上取——关闸门、等它把在途连接 放完（`drain.json` 里的 `active_at_close` / `drained`）。因此缺少上面任何一项时 `capture`