diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bba9eb..a036c28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,17 @@ - `LibraryJobRecord.kind` 之前只有 `seal | sync`,恢复作业(restore)在类型上不存在; 补上后页面不再把恢复作业当成未知类型。 +### Fixed + +- **`libraryctl` 只认 OSS 源**:`STORAGE_TYPE=minio` 的目标在发现阶段就被拒 + (`unsupported source storage type "minio"`),S3 兼容的专用验收实例无法封存。 + 现在按 `MINIO_ENDPOINT / MINIO_USE_SSL / MINIO_ACCESS_KEY_ID / MINIO_SECRET_ACCESS_KEY / + MINIO_BUCKET_NAME` 映射成 `provider=minio` + path-style;读不到端点/桶就明确报错, + 不退回默认值。 +- **CLI 与 API 的 scope→知识库 映射不一致**:API 用 `LIBRARY_PRODUCTION_KB_*`,CLI 只读 + routing bindings,同一台机器上会封存"配置里那个库"而不是 API 的验收目标。现在环境变量 + 优先、读不到才退回 bindings。 + ### Notes - 本次真实验收(kgmaint 专用实例 + 原版 WeKnora API)与本快照之外的部署计划见 diff --git a/cmd/libraryctl/discover_test.go b/cmd/libraryctl/discover_test.go new file mode 100644 index 0000000..54a59c1 --- /dev/null +++ b/cmd/libraryctl/discover_test.go @@ -0,0 +1,67 @@ +package main + +import "testing" + +// 目标实例可以是 OSS,也可以是 S3 兼容的 MinIO(本机专用验收实例就是 minio)。 +// 只认 oss 时,任何 minio 目标都会在发现阶段直接失败,等于这条验收链只能在生产存储上跑。 +func TestObjectStoreFromSourceEnvAcceptsMinIO(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, 1<<20) + if err != nil { + t.Fatalf("minio: %v", err) + } + if config.Provider != "minio" || config.Endpoint != "http://langfuse-minio:9000" || config.Secure { + t.Fatalf("config: %+v", config) + } + if config.Bucket != "kgmaint-1028a" || config.AccessKeyID != "key" || config.SecretAccessKey != "secret" { + t.Fatalf("credentials mapping: %+v", config) + } + if detail == "" || config.MaxObjectBytes != 1<<20 { + t.Fatalf("detail=%q max=%d", detail, config.MaxObjectBytes) + } +} + +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) + if err != nil { + t.Fatalf("minio tls: %v", err) + } + if !config.Secure || config.Endpoint != "https://minio.internal:9000" || config.AccessKeyID != "k2" { + t.Fatalf("config: %+v", config) + } +} + +func TestObjectStoreFromSourceEnvKeepsOSS(t *testing.T) { + app := map[string]string{ + "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) + if err != nil { + t.Fatalf("oss: %v", err) + } + if config.Provider != "oss" || config.Endpoint != "https://oss-cn-shenzhen.aliyuncs.com" || !config.Secure { + t.Fatalf("config: %+v", config) + } +} + +func TestObjectStoreFromSourceEnvRejectsUnknownStorage(t *testing.T) { + 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 { + t.Fatal("expected error when minio endpoint is missing") + } +} diff --git a/cmd/libraryctl/main.go b/cmd/libraryctl/main.go index b8a82f2..004b96f 100644 --- a/cmd/libraryctl/main.go +++ b/cmd/libraryctl/main.go @@ -244,9 +244,6 @@ func discoverSource(ctx context.Context, opts options, binding map[string]string if app["DB_DRIVER"] != "postgres" { return config, errors.New("source database driver is not postgres") } - if app["STORAGE_TYPE"] != "oss" { - return config, fmt.Errorf("unsupported source storage type %q", app["STORAGE_TYPE"]) - } address, err := containerAddress(ctx, opts.sourcePG) if err != nil { return config, err @@ -267,25 +264,102 @@ 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} - endpoint := app["OSS_ENDPOINT"] - if endpoint == "" { - return config, errors.New("source object endpoint is not configured") + config.Object, config.ObjectDetail, err = objectStoreFromSourceEnv(app, opts.maxObjectBytes) + if err != nil { + return config, err } - if !strings.Contains(endpoint, "://") { - endpoint = "https://" + endpoint - } - config.Object = v071.S3ObjectStoreConfig{Provider: "oss", Endpoint: endpoint, Region: app["OSS_REGION"], - AccessKeyID: app["OSS_ACCESS_KEY"], SecretAccessKey: app["OSS_SECRET_KEY"], Bucket: app["OSS_BUCKET_NAME"], - Secure: strings.HasPrefix(endpoint, "https://"), MaxObjectBytes: opts.maxObjectBytes} - config.ObjectDetail = "bucket=" + app["OSS_BUCKET_NAME"] + " region=" + app["OSS_REGION"] return config, nil } +// objectStoreFromSourceEnv 把源应用的存储环境映射成对象存储配置。 +// +// 只认 OSS 时,S3 兼容的 MinIO 目标(本机专用验收实例)会在发现阶段就被拒, +// 验收链只能跑在生产存储上——这不是安全性,而是能力缺口。两种都要能读, +// 读不出端点/桶就明确报错,不退回任何默认值。 +func objectStoreFromSourceEnv(app map[string]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") + } + 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 +} + +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'") @@ -391,7 +465,7 @@ func openArchive(opts options) (*librarysnapshot.Archive, error) { func runStatus(ctx context.Context, opts options) error { report := map[string]any{} - binding, bindingErr := readRoutingBindings(ctx, opts.governancePG) + binding, bindingErr := scopeBindings(ctx, opts.governancePG) var config sourceConfig var err error if bindingErr != nil {