容器 env 里的 MINIO_ENDPOINT 往往是服务名(langfuse-minio:9000),封存跑在宿主机上 解析不到,真实封存报 read library resource 并整体失败。 新增 -source-s3-endpoint:只换访问入口,桶与对象键不变;缺 scheme 直接拒绝; status 会打印实际使用的 endpoint。 试过:make verify 15 项全绿;新增覆盖用例(生效 + 拒绝无 scheme)。
This commit is contained in:
@@ -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-<scope>-<nano>` 是 45 字符,而
|
||||
`jobs.id` 是 `varchar(36)`,真实请求一律 500(`value too long`)。此前页面停在"未接线"
|
||||
从未走到这里;改为 `libcap-<scope>-<nano>` 并用测试锁住列宽。
|
||||
- **索引范围过滤漏掉 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`:入口/队列/源身份绑到同一份配置、数据库身份不泄漏口令、
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
+32
-19
@@ -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 {
|
||||
|
||||
@@ -58,6 +58,11 @@ LIBRARY_TARGET_INGRESS_DRAIN_FILE 闸门写排空证据的宿主机文件
|
||||
LIBRARY_TARGET_REDIS_ADDR/_PASSWORD/_DB 后台队列(asynq)地址
|
||||
```
|
||||
|
||||
另外 `-source-s3-endpoint <URL>` 用来覆盖"源对象存储的访问入口":容器里写的
|
||||
`MINIO_ENDPOINT` 往往是服务名(`langfuse-minio:9000`),封存跑在宿主机上时解析不到,
|
||||
资源导出会失败(`read library resource`)。覆盖只换入口,桶与对象键不变;`status`
|
||||
会如实打印实际使用的 endpoint。
|
||||
|
||||
为什么不再保留"库级窗口":被代理的 WeKnora 收到 SIGTERM 会直接 `exit(1)`,写者退出码
|
||||
永远不能当作"在途请求已排空"。排空证据只能在真正的入口上取——关闸门、等它把在途连接
|
||||
放完(`drain.json` 里的 `active_at_close` / `drained`)。因此缺少上面任何一项时 `capture`
|
||||
|
||||
Reference in New Issue
Block a user