姚章浩
47eb23cf68
页面上的"可发起封存"原来只判 LIBRARY_AGENT_SOCKET 非空：正式栈的 API 根本没挂
那个 socket，而 agent 曾顶着正式单元名冻结候选实例——配置非空、页面可点、真发起
却封错库。
现在：agent 新增 health 动作（不占封存操作锁、不跑子进程，只回答身份/降级/忙），
worker 每 30 秒探测并写入迁移 074 的 library_executor_status；overview、快照列表与
POST /library/captures 共用同一份判定（过期、不可达、实例不符、降级、正忙都拒绝并
给原因）。封存结果由 agent 附上自报身份，worker 在停写入者之前再探测一次、执行后
用结果复核，不一致就不登记测试封存——心跳只能证明"刚才"。
期望身份用 LIBRARY_SOURCE_*（封存源），不是 LIBRARY_TARGET_*（同步目标），否则
正确的执行器会被判成不匹配。另加 GET /library/restore-availability：恢复能否执行由
双闸、该快照的恢复授权与实例维护状态决定，页面不再写死 disabled。
真跑（候选隔离栈，两处 socket 隔离不变）：overview 返回 executor
{kgmaint-weknora/kgmaint-app}；停 agent 后转不可达并 503 带原因；重启后恢复。
252 lines
11 KiB
Go
252 lines
11 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"knowledge-governance/internal/appconfig"
|
|
"knowledge-governance/internal/backuphost"
|
|
"knowledge-governance/internal/dingtalkdiscovery"
|
|
"knowledge-governance/internal/httpapi"
|
|
"knowledge-governance/internal/monitoring"
|
|
postgresstore "knowledge-governance/internal/platform/postgres"
|
|
"knowledge-governance/internal/platform/weknora"
|
|
"knowledge-governance/internal/qaopsai"
|
|
"knowledge-governance/internal/qaopsorg"
|
|
"knowledge-governance/internal/qaopsquality"
|
|
"knowledge-governance/internal/qawemail"
|
|
"knowledge-governance/internal/serviceverify"
|
|
"knowledge-governance/internal/version"
|
|
)
|
|
|
|
// dingtalkSourceLister 是「系统设置 → 数据源配置」的数据源发现实现：只读列举钉钉知识库与
|
|
// 钉盘空间。它跑在 api 进程里，但用的是与 worker 共享的同一份 dws 状态目录
|
|
// （见 compose.yaml 的 /data/dws 与文件型 keychain 挂载）——凭据只有一份。
|
|
type dingtalkSourceLister struct{}
|
|
|
|
func (dingtalkSourceLister) ReadDWSMetadata(ctx context.Context) (dingtalkdiscovery.CredentialStatus, error) {
|
|
return dingtalkdiscovery.ReadCredentialStatus(ctx, dingtalkdiscovery.ExecRunner{})
|
|
}
|
|
|
|
func (dingtalkSourceLister) ListAvailableDingtalkSources(ctx context.Context) (dingtalkdiscovery.Result, error) {
|
|
return dingtalkdiscovery.Discover(ctx, dingtalkdiscovery.ExecRunner{})
|
|
}
|
|
|
|
// 同一份 dws 凭据也用来列在线文档/表格并校验可读。
|
|
func (dingtalkSourceLister) ListDingtalkDocuments(ctx context.Context, query string, limit int) (dingtalkdiscovery.DocResult, error) {
|
|
return dingtalkdiscovery.DiscoverDocs(ctx, dingtalkdiscovery.ExecRunner{}, query, limit)
|
|
}
|
|
|
|
func (dingtalkSourceLister) ValidateDingtalkDocument(ctx context.Context, node string) (dingtalkdiscovery.DocumentNode, error) {
|
|
return dingtalkdiscovery.ValidateDocument(ctx, dingtalkdiscovery.ExecRunner{}, node)
|
|
}
|
|
|
|
func main() {
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
if len(os.Args) == 2 && os.Args[1] == "--version-record-status" {
|
|
os.Exit(versionRecordStatus(ctx, os.Getenv))
|
|
}
|
|
if err := run(ctx, os.Getenv); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func run(ctx context.Context, getenv func(string) string) error {
|
|
config, err := appconfig.Load(getenv)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
pool, err := pgxpool.New(ctx, config.DatabaseURL)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer pool.Close()
|
|
if err := pool.Ping(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := postgresstore.ApplyMigrations(ctx, pool); err != nil {
|
|
return err
|
|
}
|
|
authStore := postgresstore.NewStore(pool)
|
|
if err := bootstrapInitialAdmin(ctx, authStore, time.Now); err != nil {
|
|
return err
|
|
}
|
|
releaseAuthorizationToken := ""
|
|
if config.ProductionReleaseRequestsEnabled {
|
|
releaseAuthorizationToken, err = weknora.ReadAPIKeyFile(config.ProductionReleaseAuthTokenFile)
|
|
if err != nil {
|
|
return errors.New("read production release authorization token")
|
|
}
|
|
}
|
|
// P10: with a mounted API key the probe verifies a real knowledge-base read
|
|
// (readiness); without one it degrades to the liveness-only HTTP check.
|
|
var weknoraReadiness func(context.Context) error
|
|
var knowledgeBaseLister httpapi.KnowledgeBaseLister
|
|
if strings.TrimSpace(config.TestWeKnoraBaseURL) != "" {
|
|
if apiKey, keyErr := weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile); keyErr == nil {
|
|
probeTimeout := config.TestWeKnoraHTTPTimeout
|
|
if probeTimeout > 10*time.Second || probeTimeout <= 0 {
|
|
probeTimeout = 10 * time.Second
|
|
}
|
|
if probeClient, clientErr := weknora.NewClient(weknora.ClientConfig{BaseURL: config.TestWeKnoraBaseURL, APIKey: apiKey, Timeout: probeTimeout, AllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP}); clientErr == nil {
|
|
knowledgeBaseLister = probeClient
|
|
routingKB := config.InternalRoutingKnowledgeBaseID
|
|
weknoraReadiness = func(probeCtx context.Context) error {
|
|
_, _, err := probeClient.ListKnowledge(probeCtx, routingKB, 1, 1)
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// Production WeKnora is probed for liveness only: the console must show the
|
|
// real state of that endpoint, while credentials and the publish target stay
|
|
// in the release runtime config (RELEASE_CONFIG_FILE).
|
|
// 生产 WeKnora 端点：配置入口在网页「系统设置 → 服务配置」，管理员保存的值是
|
|
// 这里的真源；容器 env 只作为未在网页配置时的兜底。凭据与发布目标仍留在
|
|
// release 运行时配置（RELEASE_CONFIG_FILE）里，不经网页也不落明文。
|
|
actionStore := postgresstore.NewStore(pool)
|
|
productionEndpointResolver := func(resolveCtx context.Context) string {
|
|
settings, err := actionStore.GetActiveUISettings(resolveCtx)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
return settings.ProductionWeKnoraBaseURL
|
|
}
|
|
testEndpointResolver := func(resolveCtx context.Context) string {
|
|
settings, err := actionStore.GetActiveUISettings(resolveCtx)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
return settings.TestWeKnoraBaseURL
|
|
}
|
|
monitorProvider, err := monitoring.NewRealProvider(monitoring.RealProviderConfig{DB: pool, WeKnoraBaseURL: config.TestWeKnoraBaseURL, TestWeKnoraResolver: testEndpointResolver, ProductionWeKnoraBaseURL: config.ProductionWeKnoraBaseURL, ProductionWeKnoraResolver: productionEndpointResolver, MinerUBaseURL: config.MinerUBaseURL, WeKnoraReadiness: weknoraReadiness, WorkerStaleAfter: config.WorkerLivenessStaleAfter})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
serviceTargetVerifier := serviceverify.New(serviceverify.Config{
|
|
TestAPIKeyFile: config.TestWeKnoraAPIKeyFile,
|
|
ProductionAPIKeyFile: config.ProductionWeKnoraAPIKeyFile,
|
|
TestBaseURLFallback: config.TestWeKnoraBaseURL,
|
|
ProductionBaseURLFallback: config.ProductionWeKnoraBaseURL,
|
|
TestAllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP,
|
|
ProductionAllowDockerHTTP: config.ProductionWeKnoraAllowDockerHTTP,
|
|
})
|
|
// 「版本记录」：只读查询面在 API 侧接线；真正的抓取在 worker 侧（未配置来源即未接线）。
|
|
versionHistoryStore, versionHistoryRepo := httpapi.VersionHistoryStore(nil), ""
|
|
if config.VersionHistoryGiteaBaseURL != "" {
|
|
versionHistoryStore, versionHistoryRepo = postgresstore.NewStore(pool), config.VersionHistoryRepo
|
|
info := version.Snapshot()
|
|
if info.IdentityAvailable && info.CommitSHA != nil {
|
|
if _, e := actionStore.EnqueueVersionHistorySync(ctx, config.VersionHistoryRepo, "system:deployment", *info.CommitSHA, time.Now()); e != nil {
|
|
log.Print("version history deployment synchronization could not be queued")
|
|
}
|
|
}
|
|
}
|
|
qaReader, qaSettings := openQAOperations(ctx, getenv, config.DatabaseURL, pool)
|
|
if closer, ok := qaReader.(interface{ Close() }); ok {
|
|
defer closer.Close()
|
|
}
|
|
if qaSettings.Problem != "" {
|
|
log.Print("QA operations source is unavailable; governance remains online")
|
|
}
|
|
qaOrganizationStore := qaopsorg.NewStore(pool, getenv("DOBOTAI_OPS_TENANT_KEY"))
|
|
qaSyncSettings, qaSyncErr := qaopsorg.LoadSyncSettings(getenv)
|
|
if qaSyncErr != nil {
|
|
log.Print("QA directory synchronization is not configured; governance remains online")
|
|
}
|
|
var qaSyncStore httpapi.QAOrgSyncStore
|
|
if getenv("DOBOTAI_OPS_TENANT_KEY") != "" {
|
|
qaSyncStore = qaOrganizationStore
|
|
}
|
|
handler, err := httpapi.NewHandler(httpapi.Config{
|
|
QAAI: qaopsai.NewFromEnvironment(pool, getenv, config.DatabaseURL),
|
|
QAWeeklyMail: qawemail.NewStore(pool, qaSettings.TenantKey, getenv("QA_AI_MASTER_KEY_FILE"), nil),
|
|
QAAITrustTLSProxy: getenv("QA_AI_TRUST_TLS_PROXY") == "true",
|
|
QAQuality: qaopsquality.NewStore(pool, getenv("DOBOTAI_OPS_TENANT_KEY")),
|
|
QAOrgSync: qaSyncStore,
|
|
QAOrgSyncEnabled: qaSyncErr == nil && qaSyncSettings.Enabled,
|
|
QAOrg: qaOrganizationStore,
|
|
QAOps: qaReader,
|
|
QAOpsTenantKey: qaSettings.TenantKey,
|
|
QAOpsEnabled: qaSettings.Enabled,
|
|
QAOpsProblem: qaSettings.Problem,
|
|
Store: postgresstore.NewStore(pool),
|
|
Monitor: monitorProvider,
|
|
InternalRoutingKnowledgeBaseID: config.InternalRoutingKnowledgeBaseID,
|
|
ExternalRoutingKnowledgeBaseID: config.ExternalRoutingKnowledgeBaseID,
|
|
RoutingBindings: postgresstore.NewStore(pool),
|
|
KnowledgeBases: knowledgeBaseLister,
|
|
ProductionBindings: postgresstore.NewStore(pool),
|
|
ProductionKnowledgeBases: serviceTargetVerifier,
|
|
ProductionWeKnoraBaseURL: config.ProductionWeKnoraBaseURL,
|
|
ServiceTargetVerifier: serviceTargetVerifier,
|
|
ProductionReleaseRequestsEnabled: config.ProductionReleaseRequestsEnabled,
|
|
TestAutoReviewPolicy: config.TestAutoReviewPolicy,
|
|
ProductionBackupRestoreVerified: config.ProductionBackupRestoreVerified,
|
|
ProductionReleaseAuthorizerRef: config.ProductionReleaseAuthorizerRef,
|
|
ProductionReleaseAuthorizationToken: releaseAuthorizationToken,
|
|
// 数据源发现（系统设置 → 数据源配置）：用与 worker 同一份 dws 状态做只读列举。
|
|
DingtalkSources: dingtalkSourceLister{},
|
|
DingtalkDocuments: dingtalkSourceLister{},
|
|
BackupSchedule: postgresstore.NewStore(pool),
|
|
VersionHistory: versionHistoryStore,
|
|
VersionHistoryRepo: versionHistoryRepo,
|
|
LibraryArchiveRoot: config.LibraryArchiveRoot,
|
|
LibraryTargetInstance: config.LibraryTargetInstance,
|
|
LibrarySourceInstance: config.SourceInstance(),
|
|
LibrarySourceProject: config.SourceProject(),
|
|
LibraryAgentSocket: config.LibraryAgentSocket,
|
|
LibraryExecutorStatusMaxAge: config.LibraryExecutorStatusMaxAge,
|
|
LibraryCapacityBudgetBytes: config.LibraryCapacityBudgetBytes,
|
|
LibraryProductionDatabaseURL: config.LibraryProductionDatabaseURL,
|
|
LibraryProductionInternalKB: config.LibraryProductionInternalKB,
|
|
LibraryProductionExternalKB: config.LibraryProductionExternalKB,
|
|
LibraryProductionTenantID: config.LibraryProductionTenantID,
|
|
LibraryRetentionVersions: config.LibraryRetentionVersions,
|
|
LibraryRetentionDays: config.LibraryRetentionDays,
|
|
BackupHost: backuphost.New(backuphost.Config{
|
|
SSHConfig: config.BackupHostSSHConfig,
|
|
Target: config.BackupHostTarget,
|
|
ScriptDir: config.BackupHostScriptDir,
|
|
Root: config.BackupHostRoot,
|
|
}),
|
|
Auth: &httpapi.AuthConfig{Store: authStore, SessionTTL: 12 * time.Hour},
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
server := &http.Server{
|
|
Addr: config.AppAddr,
|
|
Handler: handler,
|
|
ReadHeaderTimeout: 5 * time.Second,
|
|
ReadTimeout: 15 * time.Second,
|
|
WriteTimeout: 30 * time.Second,
|
|
IdleTimeout: 60 * time.Second,
|
|
}
|
|
|
|
errCh := make(chan error, 1)
|
|
go func() { errCh <- server.ListenAndServe() }()
|
|
log.Printf("governance API listening on %s", config.AppAddr)
|
|
select {
|
|
case err := <-errCh:
|
|
if errors.Is(err, http.ErrServerClosed) {
|
|
return nil
|
|
}
|
|
return err
|
|
case <-ctx.Done():
|
|
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
return server.Shutdown(shutdownCtx)
|
|
}
|
|
}
|