diff --git a/.env.example b/.env.example index 13e9d634f..ea6cba200 100644 --- a/.env.example +++ b/.env.example @@ -811,3 +811,18 @@ CONCURRENCY_POOL_SIZE=5 # BROWSERSKILL_INTERNAL_URL=http://10.0.0.12:8080 # 所有副本使用相同的随机密钥（至少 32 字符），从 Secret 注入，不提交到仓库。 # BROWSERSKILL_CLUSTER_SECRET= + +# ========== J4. 插件运行（可选） ========== +# 代码插件（runtime.type: host）默认在每个 app 进程内运行。多机或想把 Python 插件隔开时， +# 可以另起 `WeKnora plugin-host`（compose：--profile plugin-host；helm：pluginHost.enabled）。 +# plugin-host 与 app 共用数据库、Redis、存储和 SYSTEM_AES_KEY（或 JWT_SECRET），经 Redis 通告自己。 +# app 自己运行哪些 kind（binary,python）；none 表示全部交给 plugin-host。默认本机能跑的都跑。 +# WEKNORA_PLUGIN_EMBEDDED_KINDS=none +# 不在 app 本机运行的插件（remote、plugin-host 上的）回调 Host API 的地址。 +# WEKNORA_PLUGIN_HOST_API_URL=http://app:8080 +# 仅 plugin-host：运行哪些 kind、监听地址、app 访问它的地址（默认 http://:<端口>）。 +# WEKNORA_PLUGIN_HOST_KINDS=python +# WEKNORA_PLUGIN_HOST_ADDR=:8081 +# WEKNORA_PLUGIN_HOST_URL=http://plugin-host:8081 +# Python 插件使用的解释器（默认 PATH 中的 python3）。 +# WEKNORA_PLUGIN_PYTHON=python3 diff --git a/cmd/server/main.go b/cmd/server/main.go index 021a8bd25..3525fddfa 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -41,6 +41,9 @@ import ( ) func main() { + if code, ran := subcommand(); ran { + os.Exit(code) + } // Set Gin mode if os.Getenv("GIN_MODE") == "release" { gin.SetMode(gin.ReleaseMode) diff --git a/cmd/server/plugin_host.go b/cmd/server/plugin_host.go new file mode 100644 index 000000000..3832015b8 --- /dev/null +++ b/cmd/server/plugin_host.go @@ -0,0 +1,30 @@ +package main + +import ( + "context" + "os" + "os/signal" + + "github.com/Tencent/WeKnora/internal/container" + "github.com/Tencent/WeKnora/internal/logger" +) + +// runPluginHost is `WeKnora plugin-host`: a standalone plugin host that +// runs host plugins for the app nodes (see container.RunPluginHost). +func runPluginHost() int { + ctx, stop := signal.NotifyContext(context.Background(), shutdownSignals...) + defer stop() + if err := container.RunPluginHost(ctx); err != nil { + logger.Errorf(context.Background(), "[plugin-host] %v", err) + return 1 + } + return 0 +} + +// subcommand runs a subcommand named on the command line, if any. +func subcommand() (code int, ran bool) { + if len(os.Args) > 1 && os.Args[1] == "plugin-host" { + return runPluginHost(), true + } + return 0, false +} diff --git a/docker-compose.yml b/docker-compose.yml index 0aca52475..439e02e51 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -381,6 +381,11 @@ services: - OIDC_USER_INFO_MAPPING_EMAIL=${OIDC_USER_INFO_MAPPING_EMAIL:-} # 飞书云文档解析模式 export：导出为docx，blocks：根据飞书云文档块解析为markdown。默认为:export - FEISHU_DOCX_PARSE_MODE=${FEISHU_DOCX_PARSE_MODE:-export} + # ========== 插件 ========== + # app 自己运行哪些 kind 的宿主插件（binary,python；none 表示全部交给 plugin-host），默认本机能跑的都跑 + - WEKNORA_PLUGIN_EMBEDDED_KINDS=${WEKNORA_PLUGIN_EMBEDDED_KINDS:-} + # 不在本机运行的插件（remote、plugin-host 上的）回调 Host API 的地址；启用 plugin-host 时设为 http://app:8080 + - WEKNORA_PLUGIN_HOST_API_URL=${WEKNORA_PLUGIN_HOST_API_URL:-} depends_on: redis: condition: service_started @@ -394,6 +399,66 @@ services: extra_hosts: - "host.docker.internal:host-gateway" + # 独立插件宿主（可选，profile: plugin-host）：在单独的容器里运行宿主插件， + # 例如把 Python 插件与 app 隔开。与 app 共用镜像、数据库、Redis 和存储， + # 通过 Redis 向 app 通告自己，app 用 SYSTEM_AES_KEY 派生的密钥签名调用它。 + # 启用：docker compose --profile plugin-host up -d，并在 .env 里设 + # WEKNORA_PLUGIN_EMBEDDED_KINDS=none（或 binary） + # WEKNORA_PLUGIN_HOST_API_URL=http://app:8080 + plugin-host: + image: wechatopenai/weknora-app:${WEKNORA_VERSION:-latest} + container_name: WeKnora-plugin-host + profiles: + - plugin-host + command: ["./WeKnora", "plugin-host"] + volumes: + # STORAGE_TYPE=local 时插件包存在这里 + - data-files:/data/files + env_file: + - .env + environment: + - GIN_MODE=${GIN_MODE:-release} + - TZ=${TZ:-Asia/Shanghai} + - LOG_LEVEL=${LOG_LEVEL:-info} + - WEKNORA_PLUGIN_HOST_ADDR=:8081 + - WEKNORA_PLUGIN_HOST_URL=http://plugin-host:8081 + - WEKNORA_PLUGIN_HOST_KINDS=${WEKNORA_PLUGIN_HOST_KINDS:-} + - WEKNORA_PLUGIN_HOST_API_URL=${WEKNORA_PLUGIN_HOST_API_URL:-http://app:8080} + - DB_DRIVER=${DB_DRIVER:-postgres} + - DB_HOST=${DB_HOST:-postgres} + - DB_PORT=${DB_PORT:-5432} + - DB_USER=${DB_USER:-} + - DB_PASSWORD=${DB_PASSWORD:-} + - DB_NAME=${DB_NAME:-} + - STORAGE_TYPE=${STORAGE_TYPE:-local} + - LOCAL_STORAGE_BASE_DIR=${LOCAL_STORAGE_BASE_DIR:-/data/files} + - MINIO_ENDPOINT=${MINIO_ENDPOINT:-minio:9000} + - MINIO_ACCESS_KEY_ID=${MINIO_ACCESS_KEY_ID:-minioadmin} + - MINIO_SECRET_ACCESS_KEY=${MINIO_SECRET_ACCESS_KEY:-minioadmin} + - REDIS_ADDR=${REDIS_ADDR:-redis:6379} + - REDIS_USERNAME=${REDIS_USERNAME:-} + - REDIS_PASSWORD=${REDIS_PASSWORD:-} + - REDIS_DB=${REDIS_DB:-} + - WEKNORA_REDIS_NAMESPACE=${WEKNORA_REDIS_NAMESPACE:-} + - SYSTEM_AES_KEY=${SYSTEM_AES_KEY:-} + - JWT_SECRET=${JWT_SECRET:-} + - SSRF_WHITELIST=${SSRF_WHITELIST:-} + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8081/healthz"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 30s + depends_on: + redis: + condition: service_started + postgres: + condition: service_healthy + networks: + - WeKnora-network + restart: unless-stopped + stop_grace_period: 75s + # Sandbox 镜像：仅用于 build/pull，非常驻。Docker 沙箱默认关闭，需 # WEKNORA_SANDBOX_DOCKER_ENABLED=true 并挂载 docker.sock 后，app 才会按会话创建容器。 sandbox: diff --git a/helm/templates/app.yaml b/helm/templates/app.yaml index 9b79ca83d..40af2e66a 100644 --- a/helm/templates/app.yaml +++ b/helm/templates/app.yaml @@ -126,6 +126,15 @@ spec: # whatever Engine API this process can reach (often host root). - name: WEKNORA_SANDBOX_DOCKER_ENABLED value: {{ .Values.app.env.WEKNORA_SANDBOX_DOCKER_ENABLED | default "false" | quote }} + {{- if .Values.pluginHost.enabled }} + # Host plugins of these kinds still run in the app pods; the rest + # go to the plugin host pods. Plugins there reach the Host API + # through the app Service. + - name: WEKNORA_PLUGIN_EMBEDDED_KINDS + value: {{ .Values.pluginHost.appKinds | quote }} + - name: WEKNORA_PLUGIN_HOST_API_URL + value: "http://app:{{ .Values.app.service.port }}" + {{- end }} {{- if .Values.neo4j.enabled }} # Neo4j configuration (for GraphRAG) # NEO4J_ENABLE 是知识图谱的唯一开关（Go 代码认它，不认 ENABLE_GRAPH_RAG）。 diff --git a/helm/templates/plugin-host.yaml b/helm/templates/plugin-host.yaml new file mode 100644 index 000000000..279795d35 --- /dev/null +++ b/helm/templates/plugin-host.yaml @@ -0,0 +1,157 @@ +{{/* +Copyright 2025 Tencent +SPDX-License-Identifier: MIT + +WeKnora standalone plugin host (WeKnora plugin-host): runs host plugins for the +app pods. Hosts announce themselves in Redis and app pods call them directly +by pod IP through a gateway signed with a key derived from SYSTEM_AES_KEY, so +no Service is needed. +*/}} +{{- if .Values.pluginHost.enabled }} +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "weknora.fullname" . }}-plugin-host + namespace: {{ .Release.Namespace }} + labels: + {{- include "weknora.componentLabels" (dict "component" "plugin-host" "context" .) | nindent 4 }} +spec: + replicas: {{ .Values.pluginHost.replicaCount }} + selector: + matchLabels: + {{- include "weknora.componentSelectorLabels" (dict "component" "plugin-host" "context" .) | nindent 6 }} + template: + metadata: + labels: + {{- include "weknora.componentSelectorLabels" (dict "component" "plugin-host" "context" .) | nindent 8 }} + spec: + {{- include "weknora.imagePullSecrets" . | nindent 6 }} + serviceAccountName: {{ include "weknora.serviceAccountName" . }} + # The host withdraws from Redis, then lets calls in flight finish. + terminationGracePeriodSeconds: 75 + {{- with .Values.pluginHost.podSecurityContext | default .Values.global.podSecurityContext }} + securityContext: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: plugin-host + image: {{ include "weknora.app.image" . }} + imagePullPolicy: {{ .Values.app.image.pullPolicy }} + args: ["./WeKnora", "plugin-host"] + {{- with .Values.pluginHost.securityContext }} + securityContext: + {{- toYaml . | nindent 12 }} + {{- end }} + ports: + - containerPort: 8081 + name: gateway + protocol: TCP + env: + - name: POD_IP + valueFrom: + fieldRef: + fieldPath: status.podIP + - name: WEKNORA_PLUGIN_HOST_ADDR + value: ":8081" + - name: WEKNORA_PLUGIN_HOST_URL + value: "http://$(POD_IP):8081" + - name: WEKNORA_PLUGIN_HOST_KINDS + value: {{ .Values.pluginHost.kinds | quote }} + # Plugins granted Host API scopes call the app through its Service. + - name: WEKNORA_PLUGIN_HOST_API_URL + value: "http://app:{{ .Values.app.service.port }}" + - name: TZ + value: {{ .Values.app.env.TZ | quote }} + # The same database, Redis, storage and secrets as the app. + - name: DB_DRIVER + value: "postgres" + - name: DB_HOST + value: "postgres" + - name: DB_PORT + value: "5432" + - name: DB_USER + valueFrom: + secretKeyRef: + name: {{ include "weknora.secretName" . }} + key: DB_USER + - name: DB_PASSWORD + valueFrom: + secretKeyRef: + name: {{ include "weknora.secretName" . }} + key: DB_PASSWORD + - name: DB_NAME + valueFrom: + secretKeyRef: + name: {{ include "weknora.secretName" . }} + key: DB_NAME + - name: REDIS_ADDR + value: "redis:6379" + - name: REDIS_USERNAME + valueFrom: + secretKeyRef: + name: {{ include "weknora.secretName" . }} + key: REDIS_USERNAME + optional: true + - name: REDIS_PASSWORD + valueFrom: + secretKeyRef: + name: {{ include "weknora.secretName" . }} + key: REDIS_PASSWORD + - name: REDIS_DB + value: "0" + - name: JWT_SECRET + valueFrom: + secretKeyRef: + name: {{ include "weknora.secretName" . }} + key: JWT_SECRET + - name: SYSTEM_AES_KEY + valueFrom: + secretKeyRef: + name: {{ include "weknora.secretName" . }} + key: SYSTEM_AES_KEY + - name: STORAGE_TYPE + value: {{ .Values.app.env.STORAGE_TYPE | quote }} + - name: LOCAL_STORAGE_BASE_DIR + value: {{ .Values.app.env.LOCAL_STORAGE_BASE_DIR | quote }} + {{- with .Values.pluginHost.extraEnv }} + {{- toYaml . | nindent 12 }} + {{- end }} + volumeMounts: + # Plugin packages live in the app's storage; with STORAGE_TYPE=local + # that is this volume, so it must be ReadWriteMany. + - name: data-files + mountPath: /data/files + resources: + {{- toYaml .Values.pluginHost.resources | nindent 12 }} + livenessProbe: + httpGet: + path: /healthz + port: gateway + initialDelaySeconds: 20 + periodSeconds: 10 + readinessProbe: + httpGet: + path: /healthz + port: gateway + periodSeconds: 5 + volumes: + - name: data-files + {{- if .Values.dataFiles.persistence.enabled }} + persistentVolumeClaim: + claimName: {{ .Values.dataFiles.persistence.existingClaim | default (printf "%s-data-files" (include "weknora.fullname" .)) }} + {{- else }} + emptyDir: {} + {{- end }} + {{- with .Values.pluginHost.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.pluginHost.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.pluginHost.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} +{{- end }} diff --git a/helm/values.yaml b/helm/values.yaml index 19c1b8534..f14873505 100644 --- a/helm/values.yaml +++ b/helm/values.yaml @@ -146,6 +146,45 @@ app: # -- Affinity rules affinity: {} +# ----------------------------------------------------------------------------- +# Plugin host (WeKnora plugin-host) +# ----------------------------------------------------------------------------- +# Runs installed host plugins in their own pods instead of inside every app +# pod. Uses the app image, database, Redis and storage. With +# dataFiles.persistence and STORAGE_TYPE=local the data-files PVC must be +# ReadWriteMany, since plugin packages live there. +pluginHost: + # -- Enable standalone plugin hosts + enabled: false + + # -- Number of plugin host pods; each runs every plugin of its kinds + replicaCount: 1 + + # -- Kinds of host plugin the plugin hosts run (binary, python) + kinds: "binary,python" + + # -- Kinds the app pods keep running themselves ("none" = all to plugin hosts) + appKinds: "none" + + resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: "2" + memory: 2Gi + + podSecurityContext: {} + securityContext: + allowPrivilegeEscalation: false + + # -- Additional environment variables (e.g. storage credentials) + extraEnv: [] + + nodeSelector: {} + tolerations: [] + affinity: {} + # ----------------------------------------------------------------------------- # Frontend (Web UI) # ----------------------------------------------------------------------------- diff --git a/internal/container/container.go b/internal/container/container.go index b38b3848d..5251ed081 100644 --- a/internal/container/container.go +++ b/internal/container/container.go @@ -74,7 +74,6 @@ import ( "github.com/Tencent/WeKnora/internal/plugin/activate" pluginbuiltin "github.com/Tencent/WeKnora/internal/plugin/builtin" plugindriver "github.com/Tencent/WeKnora/internal/plugin/driver" - pluginhost "github.com/Tencent/WeKnora/internal/plugin/host" pluginmanifest "github.com/Tencent/WeKnora/internal/plugin/manifest" "github.com/Tencent/WeKnora/internal/plugin/reconcile" pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry" @@ -168,8 +167,10 @@ func BuildContainer(container *dig.Container) *dig.Container { must(container.Provide(activate.NewMCPServers)) must(container.Provide(activate.NewSkills)) must(container.Provide(activate.NewModelVendors)) - must(container.Provide(pluginhost.NewManager)) + must(container.Provide(newPluginHostManager)) must(container.Provide(pluginremote.NewManager)) + must(container.Provide(newPluginHostPool)) + must(container.Provide(newPluginDelegation)) must(container.Provide(newPluginInvoker)) must(container.Provide(repository.NewPluginKVRepository)) must(container.Provide(newPluginHostAPI)) diff --git a/internal/container/plugin_host.go b/internal/container/plugin_host.go new file mode 100644 index 000000000..e9bb1e65b --- /dev/null +++ b/internal/container/plugin_host.go @@ -0,0 +1,152 @@ +package container + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "net/url" + "os" + "strconv" + "strings" + "time" + + "github.com/Tencent/WeKnora/internal/application/repository" + "github.com/Tencent/WeKnora/internal/config" + "github.com/Tencent/WeKnora/internal/logger" + "github.com/Tencent/WeKnora/internal/plugin/host" + "github.com/Tencent/WeKnora/internal/plugin/hostpool" + "github.com/Tencent/WeKnora/internal/plugin/manifest" + "github.com/Tencent/WeKnora/internal/plugin/reconcile" + pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry" +) + +// Standalone plugin host settings. +const ( + envPluginHostAddr = "WEKNORA_PLUGIN_HOST_ADDR" // listen address, default :8081 + envPluginHostURL = "WEKNORA_PLUGIN_HOST_URL" // gateway URL app nodes use + envPluginHostKinds = "WEKNORA_PLUGIN_HOST_KINDS" // kinds to run, default all available +) + +// RunPluginHost runs this binary as a standalone plugin host +// (weknora plugin-host) until ctx ends. It shares the app's database, +// object storage and Redis, runs the installed host plugins of the kinds it +// supports, announces them in Redis and serves them to app nodes through a +// gateway signed with the cluster key. It never migrates the database; +// app nodes do. +func RunPluginHost(ctx context.Context) error { + if _, set := os.LookupEnv("AUTO_MIGRATE"); !set { + _ = os.Setenv("AUTO_MIGRATE", "false") + } + key, err := hostpool.ClusterKey() + if err != nil { + return err + } + rdb, err := initRedisClient() + if err != nil { + return err + } + if rdb == nil { + return errors.New("a plugin host needs Redis (REDIS_ADDR) to announce itself to app nodes") + } + cfg, err := config.LoadConfig() + if err != nil { + return fmt.Errorf("load config: %w", err) + } + db, err := initDatabase(cfg) + if err != nil { + return fmt.Errorf("database: %w", err) + } + store, err := newPluginPackageStore(cfg) + if err != nil { + return fmt.Errorf("package storage: %w", err) + } + + kinds := host.KindsFromEnv(envPluginHostKinds) + if len(kinds) == 0 { + return fmt.Errorf("%s leaves this plugin host nothing to run", envPluginHostKinds) + } + mgr := host.NewStandaloneManager(kinds) + // Plugins call the Host API on an app node, not through their egress + // proxy. + hostAPI := strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_HOST_API_URL")) + if u, err := url.Parse(hostAPI); err == nil && u.Hostname() != "" { + mgr.SetDirectHosts(u.Hostname()) + } + r := reconcile.New(reconcile.Options{ + Repo: repository.NewPluginRepository(db), Store: store, Registry: pluginregistry.New(), Redis: rdb, + Activators: []reconcile.Activator{mgr}, Runtimes: []manifest.RuntimeType{manifest.RuntimeHost}, + Accept: func(m *manifest.Manifest) bool { return mgr.Runs(m.Runtime.Kind) }, Role: "plugin-host", + }) + mgr.SetReporter(r) + + addr := os.Getenv(envPluginHostAddr) + if addr == "" { + addr = ":8081" + } + ln, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("listen %s: %w", addr, err) + } + advertise, err := advertiseURL(ln.Addr()) + if err != nil { + _ = ln.Close() + return err + } + srv := &http.Server{Handler: mgr.Gateway(key), ReadHeaderTimeout: 10 * time.Second} + serveErr := make(chan error, 1) + go func() { serveErr <- srv.Serve(ln) }() + + runCtx, stop := context.WithCancel(ctx) + defer stop() + r.Start(runCtx) + hostname, _ := os.Hostname() + announced := make(chan struct{}) + go func() { + hostpool.NewAnnouncer(rdb, mgr, r.NodeName(), advertise).Run(runCtx) + close(announced) + }() + logger.Infof(ctx, "[plugin-host] %s runs %s plugins, serving app nodes at %s", + hostname, strings.Join(kinds, ", "), advertise) + + select { + case <-ctx.Done(): + case err := <-serveErr: + stop() + <-announced + mgr.Close() + return fmt.Errorf("gateway: %w", err) + } + // Withdraw first so app nodes stop picking this host, then let calls in + // flight finish before the plugins stop. + stop() + <-announced + shutdown, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + if err := srv.Shutdown(shutdown); err != nil { + logger.Warnf(ctx, "[plugin-host] shutdown: %v", err) + } + mgr.Close() + return nil +} + +// advertiseURL is the gateway URL app nodes use: WEKNORA_PLUGIN_HOST_URL, +// or http://:, which suits compose and Kubernetes pod DNS. +func advertiseURL(addr net.Addr) (string, error) { + if u := strings.TrimSpace(os.Getenv(envPluginHostURL)); u != "" { + if _, err := url.Parse(u); err != nil { + return "", fmt.Errorf("%s: %w", envPluginHostURL, err) + } + return strings.TrimSuffix(u, "/"), nil + } + tcp, ok := addr.(*net.TCPAddr) + if !ok { + return "", fmt.Errorf("set %s", envPluginHostURL) + } + name, err := os.Hostname() + if err != nil || name == "" { + return "", fmt.Errorf("no hostname to advertise; set %s", envPluginHostURL) + } + return "http://" + net.JoinHostPort(name, strconv.Itoa(tcp.Port)), nil +} diff --git a/internal/container/plugin_runtime.go b/internal/container/plugin_runtime.go index c1c3917f4..173f4a948 100644 --- a/internal/container/plugin_runtime.go +++ b/internal/container/plugin_runtime.go @@ -15,10 +15,13 @@ import ( "github.com/Tencent/WeKnora/internal/application/service" "github.com/Tencent/WeKnora/internal/config" "github.com/Tencent/WeKnora/internal/handler" + "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/plugin/activate" "github.com/Tencent/WeKnora/internal/plugin/host" "github.com/Tencent/WeKnora/internal/plugin/hostapi" + "github.com/Tencent/WeKnora/internal/plugin/hostpool" "github.com/Tencent/WeKnora/internal/plugin/install" + "github.com/Tencent/WeKnora/internal/plugin/manifest" "github.com/Tencent/WeKnora/internal/plugin/reconcile" pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry" "github.com/Tencent/WeKnora/internal/plugin/remote" @@ -44,6 +47,7 @@ type pluginActivators struct { Host *host.Manager Remote *remote.Manager + Delegation *pluginDelegation WebSearch *activate.WebSearch Connectors *activate.Connectors Parsers *activate.Parsers @@ -57,27 +61,26 @@ type pluginActivators struct { // reachable before anything routes calls to it. func (a pluginActivators) list() []reconcile.Activator { return []reconcile.Activator{ - a.Host, a.Remote, a.WebSearch, a.Connectors, a.Parsers, a.Vendors, a.MCP, a.Skills, + a.Host, a.Remote, a.Delegation, a.WebSearch, a.Connectors, a.Parsers, a.Vendors, a.MCP, a.Skills, } } // newPluginHostAPI serves the Host API and gives calls a way back to it: the -// embedded host's plugins reach this node on loopback, remote plugins only -// through WEKNORA_PLUGIN_HOST_API_URL. +// embedded host's plugins reach this node on loopback, plugins elsewhere +// (remote, on a plugin host) only through WEKNORA_PLUGIN_HOST_API_URL. func newPluginHostAPI( cfg *config.Config, iv *activate.Invoker, repo interfaces.PluginKVRepository, cleaner interfaces.ResourceCleaner, ) *hostapi.Handler { issuer := hostapi.NewIssuerFromEnv() - url := strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_HOST_API_URL")) - iv.SetPublicHostAPI(url) - if url == "" { - port := 8080 - if cfg != nil && cfg.Server != nil && cfg.Server.Port > 0 { - port = cfg.Server.Port - } - url = fmt.Sprintf("http://127.0.0.1:%d", port) + // Plugins on this node always use loopback; the public address is for + // plugins elsewhere (remote, on a plugin host), which reach this node + // the way the deployment routes to it. + port := 8080 + if cfg != nil && cfg.Server != nil && cfg.Server.Port > 0 { + port = cfg.Server.Port } - iv.SetHostAPI(issuer, url) + iv.SetHostAPI(issuer, fmt.Sprintf("http://127.0.0.1:%d", port)) + iv.SetPublicHostAPI(strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_HOST_API_URL"))) kv := hostapi.NewKV(repo) ctx, cancel := context.WithCancel(context.Background()) kv.StartSweeper(ctx, 10*time.Minute) @@ -97,25 +100,88 @@ func bindPluginActivators( skills.SetPluginSkills(a.Skills) } -// newPluginInvoker reaches code plugins through this node's plugin host or, -// for remote ones, at their registered URLs. -func newPluginInvoker(h *host.Manager, r *remote.Manager) *activate.Invoker { - return activate.NewInvoker(pluginClients{host: h, remote: r}) +// newPluginHostManager is this node's embedded plugin host. It runs the +// kinds WEKNORA_PLUGIN_EMBEDDED_KINDS names (default: every kind this +// machine can run; "none" leaves all host plugins to standalone hosts). +func newPluginHostManager() *host.Manager { + m := host.NewManager() + m.SetKinds(host.KindsFromEnv("WEKNORA_PLUGIN_EMBEDDED_KINDS")) + return m } -// pluginClients finds a code plugin in whichever runtime loaded it. +// newPluginHostPool finds standalone plugin hosts (weknora plugin-host) in +// Redis. It is nil without Redis or a cluster key: host plugins then run on +// this node or nowhere. +func newPluginHostPool(rdb *redis.Client) *hostpool.Pool { + if rdb == nil { + return nil + } + key, err := hostpool.ClusterKey() + if err != nil { + logger.Warnf(context.Background(), "[plugin] standalone plugin hosts are off: %v", err) + return nil + } + return hostpool.NewPool(rdb, key) +} + +// newPluginInvoker reaches code plugins through this node's plugin host, a +// standalone plugin host, or a remote plugin's registered URL. +func newPluginInvoker(h *host.Manager, r *remote.Manager, pool *hostpool.Pool) *activate.Invoker { + return activate.NewInvoker(pluginClients{host: h, remote: r, pool: pool}) +} + +// pluginClients finds a code plugin in whichever runtime serves it. type pluginClients struct { host *host.Manager remote *remote.Manager + pool *hostpool.Pool } -func (c pluginClients) Client(pluginID string) (*client.Client, error) { - if c.remote.Owns(pluginID) { - return c.remote.Client(pluginID) +func (c pluginClients) Client(ctx context.Context, m *manifest.Manifest) (*client.Client, error) { + switch { + case c.remote.Owns(m.ID): + return c.remote.Client(m.ID) + case c.host.Local(m.ID) || c.pool == nil || m.Runtime.Type != manifest.RuntimeHost: + return c.host.Client(m.ID) + default: + return c.pool.Client(ctx, m.ID, m.Version) } - return c.host.Client(pluginID) } +func (c pluginClients) OnThisNode(pluginID string) bool { return c.host.Local(pluginID) } + +// pluginDelegation fails a host plugin this node leaves to standalone +// plugin hosts when there are none to leave it to. With hosts configured it +// only logs: they may start later, and calls say which plugin is missing. +type pluginDelegation struct { + host *host.Manager + pool *hostpool.Pool +} + +func newPluginDelegation(h *host.Manager, pool *hostpool.Pool) *pluginDelegation { + return &pluginDelegation{host: h, pool: pool} +} + +func (d *pluginDelegation) Name() string { return "plugin-hosts" } + +func (d *pluginDelegation) Activate(ctx context.Context, l *reconcile.Loaded) error { + m := l.Manifest + if m.Runtime.Type != manifest.RuntimeHost || d.host.Runs(m.Runtime.Kind) { + return nil + } + if d.pool == nil { + return fmt.Errorf("this node does not run %s plugins (WEKNORA_PLUGIN_EMBEDDED_KINDS) and no plugin host "+ + "is configured; run weknora plugin-host with Redis and the same SYSTEM_AES_KEY", m.Runtime.Kind) + } + if !d.pool.Runs(ctx, m.ID, m.Version) { + logger.Infof(ctx, "[plugin] %s@%s waits for a plugin host that runs %s plugins", + m.ID, m.Version, m.Runtime.Kind) + } + return nil +} + +func (d *pluginDelegation) Deactivate(context.Context, string) error { return nil } + func newMCPServiceRepository(db *gorm.DB, plugins *activate.MCPServers) interfaces.MCPServiceRepository { return plugins.Repository(repository.NewMCPServiceRepository(db)) } @@ -146,6 +212,9 @@ func startPluginReconciler( r *reconcile.Reconciler, hostManager *host.Manager, remoteManager *remote.Manager, cleaner interfaces.ResourceCleaner, ) { + if kinds := hostManager.Kinds(); len(kinds) > 0 { + logger.Infof(context.Background(), "[plugin] this node runs %s host plugins", strings.Join(kinds, ", ")) + } hostManager.SetReporter(r) remoteManager.SetReporter(r) ctx, cancel := context.WithCancel(context.Background()) diff --git a/internal/container/plugin_runtime_test.go b/internal/container/plugin_runtime_test.go new file mode 100644 index 000000000..78c269a82 --- /dev/null +++ b/internal/container/plugin_runtime_test.go @@ -0,0 +1,72 @@ +package container + +import ( + "context" + "strings" + "testing" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + + "github.com/Tencent/WeKnora/internal/plugin/host" + "github.com/Tencent/WeKnora/internal/plugin/hostpool" + "github.com/Tencent/WeKnora/internal/plugin/manifest" + "github.com/Tencent/WeKnora/internal/plugin/reconcile" + "github.com/Tencent/WeKnora/internal/plugin/remote" + "github.com/Tencent/WeKnora/pluginsdk/pluginapi" +) + +func pythonPlugin() *reconcile.Loaded { + return &reconcile.Loaded{Manifest: &manifest.Manifest{ + ID: "acme.py", Version: "1.0.0", + Runtime: manifest.Runtime{Type: manifest.RuntimeHost, Kind: host.KindPython, Entry: "main.py"}, + }} +} + +// A node that does not run a kind hands it to plugin hosts; without any it +// says so instead of pretending the plugin loaded. +func TestDelegationNeedsPluginHosts(t *testing.T) { + ctx := context.Background() + h := host.NewManager() + h.SetKinds([]string{host.KindBinary}) + + err := newPluginDelegation(h, nil).Activate(ctx, pythonPlugin()) + if err == nil || !strings.Contains(err.Error(), "no plugin host is configured") { + t.Fatalf("without plugin hosts = %v", err) + } + + mr := miniredis.RunT(t) + pool := hostpool.NewPool(redis.NewClient(&redis.Options{Addr: mr.Addr()}), []byte("k")) + if err := newPluginDelegation(h, pool).Activate(ctx, pythonPlugin()); err != nil { + t.Fatalf("with plugin hosts configured, a missing one is not a load failure: %v", err) + } + + h.SetKinds([]string{host.KindPython}) + if err := newPluginDelegation(h, nil).Activate(ctx, pythonPlugin()); err != nil { + t.Fatalf("a kind the node runs needs no plugin host: %v", err) + } +} + +// Calls go to the pool only for host plugins this node does not run. +func TestPluginClientsRouteToThePool(t *testing.T) { + ctx := context.Background() + h := host.NewManager() + h.SetKinds([]string{host.KindBinary}) + mr := miniredis.RunT(t) + pool := hostpool.NewPool(redis.NewClient(&redis.Options{Addr: mr.Addr()}), []byte("k")) + clients := pluginClients{host: h, remote: remote.NewManager(), pool: pool} + + _, err := clients.Client(ctx, pythonPlugin().Manifest) + if pe, ok := pluginapi.AsError(err); !ok || !strings.Contains(pe.Message, "no plugin host runs acme.py@1.0.0") { + t.Fatalf("a delegated plugin is looked up in the pool, got %v", err) + } + if clients.OnThisNode("acme.py") { + t.Fatal("a delegated plugin is not on this node") + } + + clients.pool = nil + _, err = clients.Client(ctx, pythonPlugin().Manifest) + if pe, ok := pluginapi.AsError(err); !ok || !strings.Contains(pe.Message, "not running on this node") { + t.Fatalf("without a pool the node's own host answers, got %v", err) + } +} diff --git a/internal/plugin/activate/invoke.go b/internal/plugin/activate/invoke.go index 593a68f55..8780c3c24 100644 --- a/internal/plugin/activate/invoke.go +++ b/internal/plugin/activate/invoke.go @@ -14,9 +14,13 @@ import ( "github.com/Tencent/WeKnora/pluginsdk/pluginapi" ) -// ClientSource reaches running code plugins (the node's plugin host). +// ClientSource reaches running code plugins wherever they run: this node's +// plugin host, a standalone plugin host, a remote service. type ClientSource interface { - Client(pluginID string) (*client.Client, error) + Client(ctx context.Context, m *manifest.Manifest) (*client.Client, error) + // OnThisNode reports whether a plugin runs on this node, where the + // loopback Host API address reaches WeKnora. + OnThisNode(pluginID string) bool } // Invoker is how the remote adapters call a code plugin: it finds the @@ -29,7 +33,7 @@ type Invoker struct { tenancy *tenancy.Service plugins interfaces.PluginRepository // tokens and hostURL give plugins granted Host API scopes a way back; - // remote plugins, off this node, use publicHostURL. + // plugins off this node (remote, on a plugin host) use publicHostURL. tokens TokenIssuer hostURL string publicHostURL string @@ -104,7 +108,7 @@ func (iv *Invoker) Envelope( } iv.mu.RLock() t, plugins, tokens, hostURL := iv.tenancy, iv.plugins, iv.tokens, iv.hostURL - if m.Runtime.Type == manifest.RuntimeRemote { + if !iv.clients.OnThisNode(m.ID) { hostURL = iv.publicHostURL } iv.mu.RUnlock() @@ -160,7 +164,7 @@ func present(pluginID string, err error) error { func (iv *Invoker) Call( ctx context.Context, m *manifest.Manifest, path string, instance map[string]any, input, out any, ) error { - c, err := iv.clients.Client(m.ID) + c, err := iv.clients.Client(ctx, m) if err != nil { return present(m.ID, err) } @@ -176,7 +180,7 @@ func (iv *Invoker) Stream( ctx context.Context, m *manifest.Manifest, path string, instance map[string]any, input any, fn func(pluginapi.Event) error, ) ([]byte, error) { - c, err := iv.clients.Client(m.ID) + c, err := iv.clients.Client(ctx, m) if err != nil { return nil, present(m.ID, err) } diff --git a/internal/plugin/activate/invoke_test.go b/internal/plugin/activate/invoke_test.go index 9fca7df9d..fe3e31cf8 100644 --- a/internal/plugin/activate/invoke_test.go +++ b/internal/plugin/activate/invoke_test.go @@ -36,7 +36,7 @@ func TestEnvelopeCarriesHostAccessOnlyForGrantedPlugins(t *testing.T) { } remote := *granted - remote.Runtime.Type = manifest.RuntimeRemote + remote.ID = "acme.remote" if env, _ := iv.Envelope(ctx, &remote, nil); env.Context.Host != nil { t.Fatal("a remote plugin must not be sent to this node's loopback address") } diff --git a/internal/plugin/activate/parsers.go b/internal/plugin/activate/parsers.go index 59cd4d0cd..b2e8430f2 100644 --- a/internal/plugin/activate/parsers.go +++ b/internal/plugin/activate/parsers.go @@ -103,7 +103,7 @@ func (e *pluginEngine) FileTypes(bool) []string { return e.fileTypes } func (e *pluginEngine) PluginID() string { return e.m.ID } func (e *pluginEngine) DisplayNames() map[string]string { return e.names } func (e *pluginEngine) CheckAvailable(bool, map[string]string) (bool, string) { - if _, err := e.iv.clients.Client(e.m.ID); err != nil { + if _, err := e.iv.clients.Client(context.Background(), e.m); err != nil { return false, err.Error() } return true, "" diff --git a/internal/plugin/activate/websearch_test.go b/internal/plugin/activate/websearch_test.go index d36ac8572..5cb187d80 100644 --- a/internal/plugin/activate/websearch_test.go +++ b/internal/plugin/activate/websearch_test.go @@ -8,6 +8,7 @@ import ( "testing" infra_web_search "github.com/Tencent/WeKnora/internal/infrastructure/web_search" + "github.com/Tencent/WeKnora/internal/plugin/manifest" "github.com/Tencent/WeKnora/internal/plugin/pkg" "github.com/Tencent/WeKnora/internal/plugin/plugintest" "github.com/Tencent/WeKnora/internal/plugin/reconcile" @@ -36,7 +37,13 @@ contributes: // fakeClients serves one plugin from an in-process SDK handler. type fakeClients struct{ c *client.Client } -func (f fakeClients) Client(string) (*client.Client, error) { return f.c, nil } +func (f fakeClients) Client(context.Context, *manifest.Manifest) (*client.Client, error) { + return f.c, nil +} + +// OnThisNode is true unless the manifest's runtime is remote, like the +// embedded host. +func (f fakeClients) OnThisNode(id string) bool { return id != "acme.remote" } func searchLoaded(t *testing.T, schema string) *reconcile.Loaded { t.Helper() diff --git a/internal/plugin/host/gateway.go b/internal/plugin/host/gateway.go new file mode 100644 index 000000000..729596197 --- /dev/null +++ b/internal/plugin/host/gateway.go @@ -0,0 +1,119 @@ +package host + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "strings" + "time" + + "github.com/Tencent/WeKnora/pluginsdk/pluginapi" +) + +// GatewayPrefix starts the paths a plugin host serves its plugins under: +// /p/{pluginId}/{version}/{protocol path}. +const GatewayPrefix = "/p/" + +// maxGatewayBody bounds one relayed request: a parse call carries the whole +// document. +const maxGatewayBody = 512 << 20 + +// Gateway serves the plugins this host runs to the other WeKnora nodes, at +// /p/{pluginId}/{version}/... Requests must be signed with the cluster key +// (pluginapi.Sign over the body); answers, streams included, are relayed as +// the plugin gives them. +func (m *Manager) Gateway(key []byte) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc(GatewayPrefix, func(w http.ResponseWriter, r *http.Request) { + m.inFlight.Add(1) + defer m.inFlight.Add(-1) + m.relay(w, r, key) + }) + return mux +} + +func gatewayError(w http.ResponseWriter, e *pluginapi.Error) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(e.Code.HTTPStatus()) + _ = json.NewEncoder(w).Encode(pluginapi.ErrorBody{Error: *e}) +} + +func (m *Manager) relay(w http.ResponseWriter, r *http.Request, key []byte) { + body, err := io.ReadAll(io.LimitReader(r.Body, maxGatewayBody+1)) + if err != nil { + gatewayError(w, pluginapi.Errorf(pluginapi.CodeBadRequest, "read request: %v", err)) + return + } + if len(body) > maxGatewayBody { + gatewayError(w, pluginapi.Errorf(pluginapi.CodeBadRequest, "request is over %d bytes", maxGatewayBody)) + return + } + err = pluginapi.VerifySignature(key, r.Header.Get(pluginapi.TimestampHeader), + r.Header.Get(pluginapi.SignatureHeader), body, time.Now()) + if err != nil { + gatewayError(w, pluginapi.Errorf(pluginapi.CodeUnauthorized, "gateway: %v", err)) + return + } + parts := strings.SplitN(strings.TrimPrefix(r.URL.Path, GatewayPrefix), "/", 3) + if len(parts) < 3 || parts[0] == "" || parts[1] == "" { + gatewayError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no endpoint %s", r.URL.Path)) + return + } + id, version, path := parts[0], parts[1], "/"+parts[2] + m.mu.Lock() + p := m.procs[id] + m.mu.Unlock() + if p == nil || p.spec.m.Version != version { + // The caller's view of this host is a heartbeat old; it retries + // elsewhere. + gatewayError(w, pluginapi.Errorf(pluginapi.CodeUnavailable, "this host does not run %s@%s", id, version)) + return + } + c, err := p.Client() + if err != nil { + if pe, ok := pluginapi.AsError(err); ok { + gatewayError(w, pe) + } else { + gatewayError(w, pluginapi.Errorf(pluginapi.CodeUnavailable, "%v", err)) + } + return + } + var sent []byte + if len(body) > 0 { + sent = body + } + resp, err := c.Raw(r.Context(), r.Method, path, sent, r.Header.Get(pluginapi.RequestIDHeader)) + if err != nil { + gatewayError(w, pluginapi.Errorf(pluginapi.CodeUnavailable, "%v", err)) + return + } + defer func() { _ = resp.Body.Close() }() + if ct := resp.Header.Get("Content-Type"); ct != "" { + w.Header().Set("Content-Type", ct) + } + w.WriteHeader(resp.StatusCode) + flusher, _ := w.(http.Flusher) + buf := make([]byte, 32<<10) + for { + n, err := resp.Body.Read(buf) + if n > 0 { + if _, werr := w.Write(buf[:n]); werr != nil { + return + } + // Stream events go out as the plugin writes them. + if flusher != nil && bytes.IndexByte(buf[:n], '\n') >= 0 { + flusher.Flush() + } + } + if err != nil { + if flusher != nil { + flusher.Flush() + } + return + } + } +} diff --git a/internal/plugin/host/host.go b/internal/plugin/host/host.go index 005d45504..270de4327 100644 --- a/internal/plugin/host/host.go +++ b/internal/plugin/host/host.go @@ -12,7 +12,12 @@ package host import ( "context" "fmt" + "os" + "os/exec" + "sort" + "strings" "sync" + "sync/atomic" "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/plugin/manifest" @@ -31,13 +36,101 @@ type StateReporter interface { // before the activators that route calls to the plugins. type Manager struct { reporter StateReporter + inFlight atomic.Int64 mu sync.Mutex procs map[string]*process + kinds map[string]bool + // standalone hosts have no one to hand other kinds to. + standalone bool + // direct are hosts plugins reach without the egress proxy: the Host API + // when it is not on this machine. + direct []string } -// NewManager creates an empty host. -func NewManager() *Manager { return &Manager{procs: map[string]*process{}} } +// NewManager creates an empty host that runs every kind this machine can. +func NewManager() *Manager { + m := &Manager{procs: map[string]*process{}} + m.SetKinds(AvailableKinds()) + return m +} + +// NewStandaloneManager creates the host of a standalone plugin host +// (weknora plugin-host): it runs the given kinds and refuses the others. +func NewStandaloneManager(kinds []string) *Manager { + m := &Manager{procs: map[string]*process{}, standalone: true} + m.SetKinds(kinds) + return m +} + +// AvailableKinds are the kinds this machine can run: binaries, and python +// when an interpreter is installed. +func AvailableKinds() []string { + kinds := []string{KindBinary} + if _, err := exec.LookPath(PythonCommand()); err == nil { + kinds = append(kinds, KindPython) + } + return kinds +} + +// KindsFromEnv reads a comma-separated list of kinds from an environment +// variable: unset means every available kind, "none" means none. +func KindsFromEnv(name string) []string { + raw, ok := os.LookupEnv(name) + raw = strings.TrimSpace(raw) + if !ok || raw == "" { + return AvailableKinds() + } + if raw == "none" { + return nil + } + var out []string + for _, k := range strings.Split(raw, ",") { + if k = strings.TrimSpace(k); Supported(k) { + out = append(out, k) + } + } + return out +} + +// SetKinds limits the kinds this host runs; plugins of other kinds are left +// to plugin hosts elsewhere. +func (m *Manager) SetKinds(kinds []string) { + m.mu.Lock() + defer m.mu.Unlock() + m.kinds = map[string]bool{} + for _, k := range kinds { + m.kinds[k] = true + } +} + +// Kinds lists the kinds this host runs. +func (m *Manager) Kinds() []string { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]string, 0, len(m.kinds)) + for k := range m.kinds { + out = append(out, k) + } + sort.Strings(out) + return out +} + +// Runs reports whether this host runs plugins of a kind. +func (m *Manager) Runs(kind string) bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.kinds[kind] +} + +// SetDirectHosts names hosts plugins reach without the egress proxy, such +// as the Host API's when it is on another machine. It applies to processes +// started afterwards. +func (m *Manager) SetDirectHosts(hosts ...string) { + m.mu.Lock() + m.direct = append([]string(nil), hosts...) + m.mu.Unlock() +} // SetReporter wires health changes to the reconciler. func (m *Manager) SetReporter(r StateReporter) { @@ -63,8 +156,18 @@ func (m *Manager) Activate(ctx context.Context, l *reconcile.Loaded) error { return fmt.Errorf("runtime.kind %q is not supported by this host; binary and python are", l.Manifest.Runtime.Kind) } + if !m.Runs(l.Manifest.Runtime.Kind) { + if m.standalone { + return fmt.Errorf("this plugin host does not run %s plugins (it runs %s)", + l.Manifest.Runtime.Kind, strings.Join(m.Kinds(), ", ")) + } + return nil // a plugin host elsewhere runs it + } id := l.Manifest.ID - p, err := startProcess(spec{m: l.Manifest, dir: l.Dir}, func(s State, err error) { + m.mu.Lock() + direct := m.direct + m.mu.Unlock() + p, err := startProcess(spec{m: l.Manifest, dir: l.Dir, direct: direct}, func(s State, err error) { m.report(id, s, err) }) if err != nil { @@ -123,6 +226,46 @@ func (m *Manager) Client(pluginID string) (*client.Client, error) { return p.Client() } +// Local reports whether a plugin runs on this host. +func (m *Manager) Local(pluginID string) bool { + m.mu.Lock() + defer m.mu.Unlock() + _, ok := m.procs[pluginID] + return ok +} + +// Running is one plugin this host runs. +type Running struct { + ID string `json:"id"` + Version string `json:"version"` + Kind string `json:"kind"` + State State `json:"state"` +} + +// Running lists the plugins this host runs and their state. +func (m *Manager) Running() []Running { + m.mu.Lock() + procs := make([]*process, 0, len(m.procs)) + for _, p := range m.procs { + procs = append(procs, p) + } + m.mu.Unlock() + out := make([]Running, 0, len(procs)) + for _, p := range procs { + p.mu.RLock() + st := p.state + p.mu.RUnlock() + out = append(out, Running{ + ID: p.spec.m.ID, Version: p.spec.m.Version, Kind: p.spec.m.Runtime.Kind, State: st, + }) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out +} + +// InFlight is how many gateway calls this host is answering. +func (m *Manager) InFlight() int64 { return m.inFlight.Load() } + // Close stops every plugin, for shutdown. func (m *Manager) Close() { m.mu.Lock() diff --git a/internal/plugin/host/process.go b/internal/plugin/host/process.go index e051864e9..10741a3fc 100644 --- a/internal/plugin/host/process.go +++ b/internal/plugin/host/process.go @@ -51,6 +51,8 @@ const ( type spec struct { m *manifest.Manifest dir string // extracted package + // direct are hosts reached without the egress proxy. + direct []string } // Kinds of host plugin this host runs. @@ -200,11 +202,11 @@ func (p *process) childEnv(network, socket, token string) []string { "HTTPS_PROXY=" + p.proxy.URL(), "http_proxy=" + p.proxy.URL(), "https_proxy=" + p.proxy.URL(), - // The Host API is on this node's loopback; the proxy is for the - // outside world. - "NO_PROXY=127.0.0.1,localhost,::1", - "no_proxy=127.0.0.1,localhost,::1", } + // The Host API is on this node's loopback, or on a host named direct; + // the proxy is for the outside world. + noProxy := strings.Join(append([]string{"127.0.0.1", "localhost", "::1"}, p.spec.direct...), ",") + env = append(env, "NO_PROXY="+noProxy, "no_proxy="+noProxy) for _, k := range []string{"PATH", "LANG", "LC_ALL", "TZ", "SYSTEMROOT", "WINDIR"} { if v, ok := os.LookupEnv(k); ok { env = append(env, k+"="+v) diff --git a/internal/plugin/host/python_test.go b/internal/plugin/host/python_test.go index 84be53d51..074f6066c 100644 --- a/internal/plugin/host/python_test.go +++ b/internal/plugin/host/python_test.go @@ -128,3 +128,30 @@ func TestEntryName(t *testing.T) { t.Fatal("this host runs binaries and python") } } + +func TestKindsDecideWhereAPluginRuns(t *testing.T) { + ctx := context.Background() + l := install(t, "1.0.0", "") + + embedded := NewManager() + embedded.SetKinds([]string{KindPython}) + defer embedded.Close() + if err := embedded.Activate(ctx, l); err != nil || embedded.Local("acme.echo") { + t.Fatalf("an app node hands other kinds on: %v", err) + } + + standalone := NewStandaloneManager([]string{KindPython}) + defer standalone.Close() + if err := standalone.Activate(ctx, l); err == nil || !strings.Contains(err.Error(), "does not run binary") { + t.Fatalf("a standalone host refuses other kinds, got %v", err) + } + + t.Setenv("WEKNORA_PLUGIN_EMBEDDED_KINDS", "none") + if got := KindsFromEnv("WEKNORA_PLUGIN_EMBEDDED_KINDS"); len(got) != 0 { + t.Fatalf("none = %v", got) + } + t.Setenv("WEKNORA_PLUGIN_EMBEDDED_KINDS", "binary, node") + if got := KindsFromEnv("WEKNORA_PLUGIN_EMBEDDED_KINDS"); len(got) != 1 || got[0] != KindBinary { + t.Fatalf("binary, node = %v", got) + } +} diff --git a/internal/plugin/hostpool/hostpool.go b/internal/plugin/hostpool/hostpool.go new file mode 100644 index 000000000..b69d69509 --- /dev/null +++ b/internal/plugin/hostpool/hostpool.go @@ -0,0 +1,261 @@ +// Package hostpool connects app nodes to standalone plugin hosts +// (weknora plugin-host) without Kubernetes: each host announces itself and +// the plugins it runs in Redis, and app nodes call the least busy host that +// runs the version they need, through the host's signed gateway. +package hostpool + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "math/rand/v2" + "net/http" + "os" + "runtime" + "strings" + "sync" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/Tencent/WeKnora/internal/logger" + "github.com/Tencent/WeKnora/internal/plugin/host" + "github.com/Tencent/WeKnora/internal/utils" + "github.com/Tencent/WeKnora/pluginsdk/client" + "github.com/Tencent/WeKnora/pluginsdk/pluginapi" +) + +// Heartbeat timing: a host refreshes its record every HeartbeatInterval and +// drops out HeartbeatTTL after its last one. +const ( + HeartbeatInterval = 5 * time.Second + HeartbeatTTL = 15 * time.Second + // refreshAfter is how long an app node trusts its view of the hosts. + refreshAfter = 2 * time.Second +) + +const keyBase = "weknora:plugin-host:" + +func keyPrefix() string { + if ns := strings.TrimSpace(os.Getenv("WEKNORA_REDIS_NAMESPACE")); ns != "" { + return keyBase + ns + ":" + } + return keyBase +} + +// ErrNoClusterKey means neither SYSTEM_AES_KEY nor JWT_SECRET is set, so +// app nodes and plugin hosts share no secret to sign calls with. +var ErrNoClusterKey = errors.New("plugin hosts need SYSTEM_AES_KEY or JWT_SECRET, the same on every node") + +// ClusterKey is the key app nodes sign gateway calls with, derived from the +// secrets every node already shares. +func ClusterKey() ([]byte, error) { + var secret []byte + switch { + case utils.GetAESKey() != nil: + secret = utils.GetAESKey() + case strings.TrimSpace(os.Getenv("JWT_SECRET")) != "": + secret = []byte(strings.TrimSpace(os.Getenv("JWT_SECRET"))) + default: + return nil, ErrNoClusterKey + } + mac := hmac.New(sha256.New, secret) + _, _ = mac.Write([]byte("weknora plugin host gateway")) + return mac.Sum(nil), nil +} + +// Info is one plugin host's announcement. +type Info struct { + ID string `json:"id"` + URL string `json:"url"` // gateway base URL, reachable from app nodes + OS string `json:"os"` + // Arch and Kinds say what the host can run. + Arch string `json:"arch"` + Kinds []string `json:"kinds"` + Plugins []host.Running `json:"plugins"` + InFlight int64 `json:"inFlight"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// runs reports whether the host serves a plugin version right now. +func (i Info) runs(pluginID, version string) bool { + for _, p := range i.Plugins { + if p.ID == pluginID && p.Version == version && p.State == host.StateReady { + return true + } + } + return false +} + +// Announcer keeps a plugin host's record in Redis. +type Announcer struct { + rdb *redis.Client + mgr *host.Manager + id string + url string +} + +// NewAnnouncer announces the host at url (its gateway, as app nodes reach +// it) under id. +func NewAnnouncer(rdb *redis.Client, mgr *host.Manager, id, url string) *Announcer { + return &Announcer{rdb: rdb, mgr: mgr, id: id, url: strings.TrimSuffix(url, "/")} +} + +func (a *Announcer) info() Info { + return Info{ + ID: a.id, URL: a.url, OS: runtime.GOOS, Arch: runtime.GOARCH, Kinds: a.mgr.Kinds(), + Plugins: a.mgr.Running(), InFlight: a.mgr.InFlight(), UpdatedAt: time.Now(), + } +} + +// Announce writes the record once. +func (a *Announcer) Announce(ctx context.Context) error { + b, err := json.Marshal(a.info()) + if err != nil { + return err + } + return a.rdb.Set(ctx, keyPrefix()+a.id, b, HeartbeatTTL).Err() +} + +// Run announces every HeartbeatInterval until ctx ends, then withdraws the +// record so app nodes stop calling at once. +func (a *Announcer) Run(ctx context.Context) { + t := time.NewTicker(HeartbeatInterval) + defer t.Stop() + for { + if err := a.Announce(ctx); err != nil && ctx.Err() == nil { + logger.Warnf(ctx, "[plugin-host] heartbeat: %v", err) + } + select { + case <-ctx.Done(): + wctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + _ = a.rdb.Del(wctx, keyPrefix()+a.id).Err() + cancel() + return + case <-t.C: + } + } +} + +// Pool is an app node's view of the plugin hosts. +type Pool struct { + rdb *redis.Client + key []byte + http *http.Client + now func() time.Time + + mu sync.Mutex + hosts []Info + fetched time.Time + clients map[string]*client.Client +} + +// NewPool reads plugin hosts from Redis and signs calls with key. +func NewPool(rdb *redis.Client, key []byte) *Pool { + return &Pool{ + rdb: rdb, key: key, now: time.Now, clients: map[string]*client.Client{}, + // Calls are bounded by their context; syncs stream for long. + http: &http.Client{Transport: &http.Transport{MaxIdleConnsPerHost: 32, IdleConnTimeout: 90 * time.Second}}, + } +} + +// Hosts lists the live plugin hosts. +func (p *Pool) Hosts(ctx context.Context) ([]Info, error) { + p.mu.Lock() + defer p.mu.Unlock() + if p.hosts != nil && p.now().Sub(p.fetched) < refreshAfter { + return p.hosts, nil + } + hosts, err := p.fetch(ctx) + if err != nil { + if p.hosts != nil { + logger.Warnf(ctx, "[plugin] read plugin hosts, keeping the last view: %v", err) + return p.hosts, nil + } + return nil, err + } + p.hosts, p.fetched = hosts, p.now() + return hosts, nil +} + +func (p *Pool) fetch(ctx context.Context) ([]Info, error) { + var keys []string + iter := p.rdb.Scan(ctx, 0, keyPrefix()+"*", 100).Iterator() + for iter.Next(ctx) { + keys = append(keys, iter.Val()) + } + if err := iter.Err(); err != nil { + return nil, err + } + hosts := []Info{} + if len(keys) == 0 { + return hosts, nil + } + vals, err := p.rdb.MGet(ctx, keys...).Result() + if err != nil { + return nil, err + } + for _, v := range vals { + s, ok := v.(string) + if !ok { + continue + } + var info Info + if json.Unmarshal([]byte(s), &info) == nil && info.URL != "" && + p.now().Sub(info.UpdatedAt) < HeartbeatTTL { + hosts = append(hosts, info) + } + } + return hosts, nil +} + +// Client reaches a plugin version on the least busy host that runs it. +func (p *Pool) Client(ctx context.Context, pluginID, version string) (*client.Client, error) { + hosts, err := p.Hosts(ctx) + if err != nil { + return nil, pluginapi.Errorf(pluginapi.CodeUnavailable, "read plugin hosts: %v", err) + } + var best []Info + for _, h := range hosts { + if !h.runs(pluginID, version) { + continue + } + switch { + case len(best) == 0 || h.InFlight < best[0].InFlight: + best = []Info{h} + case h.InFlight == best[0].InFlight: + best = append(best, h) + } + } + if len(best) == 0 { + return nil, pluginapi.Errorf(pluginapi.CodeUnavailable, + "no plugin host runs %s@%s; is weknora plugin-host up?", pluginID, version) + } + h := best[rand.IntN(len(best))] //nolint:gosec // load spreading, not security + base := fmt.Sprintf("%s%s%s/%s", h.URL, host.GatewayPrefix, pluginID, version) + p.mu.Lock() + defer p.mu.Unlock() + c, ok := p.clients[base] + if !ok { + c = client.New(base, p.http, client.Signed(p.key)) + p.clients[base] = c + } + return c, nil +} + +// Runs reports whether some live host runs a plugin version. +func (p *Pool) Runs(ctx context.Context, pluginID, version string) bool { + hosts, err := p.Hosts(ctx) + if err != nil { + return false + } + for _, h := range hosts { + if h.runs(pluginID, version) { + return true + } + } + return false +} diff --git a/internal/plugin/hostpool/hostpool_test.go b/internal/plugin/hostpool/hostpool_test.go new file mode 100644 index 000000000..c7c947f8a --- /dev/null +++ b/internal/plugin/hostpool/hostpool_test.go @@ -0,0 +1,179 @@ +package hostpool + +import ( + "context" + "net/http" + "net/http/httptest" + "os/exec" + "path/filepath" + "runtime" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + + "github.com/Tencent/WeKnora/internal/plugin/host" + "github.com/Tencent/WeKnora/internal/plugin/manifest" + "github.com/Tencent/WeKnora/internal/plugin/reconcile" + "github.com/Tencent/WeKnora/pluginsdk/client" + "github.com/Tencent/WeKnora/pluginsdk/pluginapi" +) + +// echoPackage builds the host package's echo test plugin into an extracted +// package directory. +func echoPackage(t *testing.T) *reconcile.Loaded { + t.Helper() + dir := t.TempDir() + name := "echo" + if runtime.GOOS == "windows" { + name += ".exe" + } + bin := filepath.Join(dir, "bin", runtime.GOOS+"-"+runtime.GOARCH, name) + out, err := exec.Command("go", "build", "-o", bin, "../host/testdata/echoplugin").CombinedOutput() + if err != nil { + t.Fatalf("build echo plugin: %v\n%s", err, out) + } + m := &manifest.Manifest{ + SchemaVersion: manifest.SchemaVersion, ID: "acme.echo", Version: "1.0.0", APIVersion: pluginapi.APIVersion, + Name: manifest.Text("Echo", nil), Publisher: manifest.Publisher{ID: "acme"}, + Runtime: manifest.Runtime{Type: manifest.RuntimeHost, Kind: host.KindBinary, Entry: "bin/{os}-{arch}/echo"}, + Contributes: manifest.Contributions{ + manifest.PointWebSearch: {{ID: "echo", Name: manifest.Text("Echo", nil)}}, + }, + } + return &reconcile.Loaded{Manifest: m, Dir: dir} +} + +func search(ctx context.Context, c *client.Client, q string) (string, error) { + var out pluginapi.SearchOutput + err := c.Call(ctx, pluginapi.SearchPath("echo"), pluginapi.Envelope{}, pluginapi.SearchInput{Query: q}, &out) + if err != nil { + return "", err + } + return out.Results[0].Title, nil +} + +func TestPoolReachesAPluginThroughAHost(t *testing.T) { + ctx := context.Background() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Setenv("SYSTEM_AES_KEY", "0123456789abcdef0123456789abcdef") + key, err := ClusterKey() + if err != nil { + t.Fatal(err) + } + + // The plugin host: a manager running the plugin behind its gateway. + mgr := host.NewManager() + defer mgr.Close() + if err := mgr.Activate(ctx, echoPackage(t)); err != nil { + t.Fatal(err) + } + gw := httptest.NewServer(mgr.Gateway(key)) + defer gw.Close() + a := NewAnnouncer(rdb, mgr, "h1", gw.URL+"/") + if err := a.Announce(ctx); err != nil { + t.Fatal(err) + } + + pool := NewPool(rdb, key) + c, err := pool.Client(ctx, "acme.echo", "1.0.0") + if err != nil { + t.Fatal(err) + } + if got, err := search(ctx, c, "hello"); err != nil || got != "hello" { + t.Fatalf("search via host = %q, %v", got, err) + } + if !pool.Runs(ctx, "acme.echo", "1.0.0") || pool.Runs(ctx, "acme.echo", "2.0.0") { + t.Fatal("Runs should match the running version only") + } + if _, err := pool.Client(ctx, "acme.echo", "2.0.0"); !isCode(err, pluginapi.CodeUnavailable) { + t.Fatalf("other version = %v", err) + } + + // The gateway refuses calls without the cluster key, and relays a + // version it does not run as unavailable. + unsigned := client.New(gw.URL+"/p/acme.echo/1.0.0", nil, nil) + if _, err := search(ctx, unsigned, "x"); !isCode(err, pluginapi.CodeUnauthorized) { + t.Fatalf("unsigned = %v", err) + } + stale := client.New(gw.URL+"/p/acme.echo/0.9.0", nil, client.Signed(key)) + if _, err := search(ctx, stale, "x"); !isCode(err, pluginapi.CodeUnavailable) { + t.Fatalf("stale version = %v", err) + } + resp, err := http.Get(gw.URL + "/healthz") + if err != nil || resp.StatusCode != http.StatusOK { + t.Fatalf("healthz = %v, %v", resp, err) + } + _ = resp.Body.Close() +} + +func TestHostsAgeOutAndWithdraw(t *testing.T) { + ctx := context.Background() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + mgr := host.NewManager() + pool := NewPool(rdb, []byte("k")) + now := time.Now() + pool.now = func() time.Time { return now } + + runCtx, stop := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { NewAnnouncer(rdb, mgr, "h1", "http://h1:8081").Run(runCtx); close(done) }() + waitFor(t, func() bool { return mr.Exists(keyPrefix() + "h1") }) + hosts, err := pool.Hosts(ctx) + if err != nil || len(hosts) != 1 || hosts[0].URL != "http://h1:8081" || len(hosts[0].Kinds) == 0 { + t.Fatalf("hosts = %+v, %v", hosts, err) + } + + // A host that stops announcing drops out when its record expires. + stop() + <-done + if mr.Exists(keyPrefix() + "h1") { + t.Fatal("a stopped host must withdraw its record") + } + now = now.Add(refreshAfter) + if hosts, _ := pool.Hosts(ctx); len(hosts) != 0 { + t.Fatalf("hosts after withdrawal = %+v", hosts) + } + + // A record that outlived its heartbeat (clock skew, a stuck host) is + // ignored even if Redis still has it. + _ = NewAnnouncer(rdb, mgr, "h2", "http://h2:8081").Announce(ctx) + now = now.Add(refreshAfter + HeartbeatTTL) + if hosts, _ := pool.Hosts(ctx); len(hosts) != 0 { + t.Fatalf("stale hosts = %+v", hosts) + } +} + +func TestClusterKey(t *testing.T) { + t.Setenv("SYSTEM_AES_KEY", "") + t.Setenv("JWT_SECRET", "") + if _, err := ClusterKey(); err == nil { + t.Fatal("no shared secret must be an error") + } + t.Setenv("JWT_SECRET", "s") + a, _ := ClusterKey() + t.Setenv("JWT_SECRET", "t") + b, _ := ClusterKey() + if len(a) != 32 || string(a) == string(b) { + t.Fatal("the key derives from the secret") + } +} + +func isCode(err error, code pluginapi.ErrorCode) bool { + pe, ok := pluginapi.AsError(err) + return ok && pe.Code == code +} + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatal("timed out") + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/internal/plugin/plugintest/plugintest.go b/internal/plugin/plugintest/plugintest.go index c07d9b885..3a362ca2d 100644 --- a/internal/plugin/plugintest/plugintest.go +++ b/internal/plugin/plugintest/plugintest.go @@ -5,6 +5,7 @@ import ( "archive/zip" "bytes" "context" + "encoding/json" "fmt" "sort" "sync" @@ -195,8 +196,10 @@ func Install(t testing.TB, repo *MemRepo, store *MemStore, data []byte, state st t.Fatal(err) } uri, _ := store.Put(ctx, p.Digest, data) + manifestJSON, _ := json.Marshal(p.Manifest) _ = repo.SaveVersion(ctx, &types.PluginVersion{ PluginID: p.Manifest.ID, Version: p.Manifest.Version, Digest: p.Digest, PackageURI: uri, + Manifest: types.JSON(manifestJSON), }) _ = repo.SavePlugin(ctx, &types.InstalledPlugin{ ID: p.Manifest.ID, ActiveVersion: p.Manifest.Version, DesiredState: state, Runtime: "declarative", diff --git a/internal/plugin/reconcile/nodes.go b/internal/plugin/reconcile/nodes.go index 28486dcec..f86e89ec0 100644 --- a/internal/plugin/reconcile/nodes.go +++ b/internal/plugin/reconcile/nodes.go @@ -37,6 +37,9 @@ func (r *Reconciler) NodeName() string { if host == "" { host = "node" } + if r.role != "" { + host = r.role + ":" + host + } return host + "/" + r.instanceID[:8] } diff --git a/internal/plugin/reconcile/reconcile.go b/internal/plugin/reconcile/reconcile.go index ace3a2924..98ecbd384 100644 --- a/internal/plugin/reconcile/reconcile.go +++ b/internal/plugin/reconcile/reconcile.go @@ -105,6 +105,9 @@ type Reconciler struct { activators []Activator instanceID string interval time.Duration + runtimes map[string]bool + accept func(*manifest.Manifest) bool + role string mu sync.Mutex // serializes passes loaded map[string]*Loaded @@ -149,6 +152,14 @@ type Options struct { Redis *redis.Client // nil on single-node deployments Activators []Activator Interval time.Duration + // Runtimes limits the node to plugins of these runtimes (a standalone + // plugin host loads host plugins only); empty means all. + Runtimes []manifest.RuntimeType + // Accept further limits the node by the active version's manifest (a + // plugin host that runs python plugins only); nil accepts all. + Accept func(*manifest.Manifest) bool + // Role names what the node is in status reports, e.g. "plugin-host". + Role string } // New creates a Reconciler. @@ -159,9 +170,17 @@ func New(o Options) *Reconciler { if o.CacheDir == "" { o.CacheDir = DefaultCacheDir() } + var runtimes map[string]bool + if len(o.Runtimes) > 0 { + runtimes = map[string]bool{} + for _, rt := range o.Runtimes { + runtimes[string(rt)] = true + } + } return &Reconciler{ repo: o.Repo, store: o.Store, registry: o.Registry, cacheDir: o.CacheDir, rdb: o.Redis, activators: o.Activators, instanceID: uuid.NewString(), interval: o.Interval, + runtimes: runtimes, accept: o.Accept, role: o.Role, loaded: map[string]*Loaded{}, digests: map[string]string{}, retries: map[string]retry{}, status: map[string]Status{}, now: time.Now, } @@ -188,7 +207,8 @@ func (r *Reconciler) Reconcile(ctx context.Context) error { var errs []error desired := map[string]bool{} for _, row := range rows { - if row.DesiredState != types.PluginStateEnabled { + if row.DesiredState != types.PluginStateEnabled || (r.runtimes != nil && !r.runtimes[row.Runtime]) || + !r.accepts(ctx, row) { continue } desired[row.ID] = true @@ -215,6 +235,23 @@ func (r *Reconciler) Reconcile(ctx context.Context) error { return errors.Join(errs...) } +// accepts applies Options.Accept to a plugin's active version. A version +// that cannot be read is accepted, so ensure reports why. +func (r *Reconciler) accepts(ctx context.Context, row types.InstalledPlugin) bool { + if r.accept == nil { + return true + } + v, err := r.repo.GetVersion(ctx, row.ID, row.ActiveVersion) + if err != nil || v == nil { + return true + } + var m manifest.Manifest + if json.Unmarshal(v.Manifest, &m) != nil { + return true + } + return r.accept(&m) +} + // runtimeTarget identifies where a plugin runs beyond its package: a new // remote URL or secret reloads the plugin like a new version would. func runtimeTarget(row types.InstalledPlugin) string { diff --git a/internal/plugin/reconcile/reconcile_test.go b/internal/plugin/reconcile/reconcile_test.go index a6d35dda2..b21f400d7 100644 --- a/internal/plugin/reconcile/reconcile_test.go +++ b/internal/plugin/reconcile/reconcile_test.go @@ -226,3 +226,35 @@ func TestReconcileReloadsOnRuntimeTargetChange(t *testing.T) { t.Fatalf("Loaded carries secret %q", got) } } + +// A standalone plugin host loads host plugins only. +func TestRuntimesLimitWhatANodeLoads(t *testing.T) { + ctx := context.Background() + repo, store, reg, act := plugintest.NewMemRepo(), &plugintest.MemStore{}, registry.New(), &recorder{} + r := New(Options{ + Repo: repo, Store: store, Registry: reg, CacheDir: t.TempDir(), Activators: []Activator{act}, + Runtimes: []manifest.RuntimeType{manifest.RuntimeHost}, Role: "plugin-host", + }) + plugintest.Install(t, repo, store, plugintest.KitPackage(t, "1.0.0"), types.PluginStateEnabled) + if err := r.Reconcile(ctx); err != nil { + t.Fatal(err) + } + if len(act.calls) != 0 || len(r.Loaded()) != 0 { + t.Fatalf("a declarative plugin was loaded: %v", act.calls) + } + if !strings.HasPrefix(r.NodeName(), "plugin-host:") { + t.Fatalf("node name = %s", r.NodeName()) + } + + // Accept filters on the manifest: nothing is loaded, nothing reported. + r = New(Options{ + Repo: repo, Store: store, Registry: registry.New(), CacheDir: t.TempDir(), Activators: []Activator{act}, + Accept: func(m *manifest.Manifest) bool { return m.Version != "1.0.0" }, + }) + if err := r.Reconcile(ctx); err != nil || len(r.Loaded()) != 0 { + t.Fatalf("an unaccepted plugin was loaded: %v", err) + } + if _, ok := r.Status("acme.kit"); ok { + t.Fatal("an unaccepted plugin has no status here") + } +} diff --git a/pluginsdk/client/client.go b/pluginsdk/client/client.go index e0c2de4ba..060899dd0 100644 --- a/pluginsdk/client/client.go +++ b/pluginsdk/client/client.go @@ -126,6 +126,13 @@ func (c *Client) do(ctx context.Context, method, path string, body []byte, reque return resp, nil } +// Raw sends one request as the client would and returns the answer +// unread, whatever its status: for gateways that relay a plugin's answers, +// streams included. The caller closes the body. +func (c *Client) Raw(ctx context.Context, method, path string, body []byte, requestID string) (*http.Response, error) { + return c.do(ctx, method, path, body, requestID) +} + // decodeError turns a non-2xx answer into a protocol error. func decodeError(resp *http.Response) error { b, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10)) diff --git a/website-docs/03-features/25-plugins.md b/website-docs/03-features/25-plugins.md new file mode 100644 index 000000000..f83224441 --- /dev/null +++ b/website-docs/03-features/25-plugins.md @@ -0,0 +1,100 @@ +# 插件 + +插件是 WeKnora 扩展能力的打包与分发单元：模型厂商、数据源、联网搜索、文档解析、技能、MCP 服务等都可以由插件提供。内置能力本身也以内置插件的形式登记，和第三方插件共用一套目录与开关。 + +两类角色分工如下： + +| | 系统管理员 | 空间管理员 | +| --- | --- | --- | +| 做什么 | 安装、升级、回滚、卸载插件，填写平台配置 | 在本空间启用或停用插件，填写空间配置 | +| 入口 | 「设置 → 插件管理」 | 「设置 → 插件」 | + +安装后的插件对所有空间可见，但**默认停用**，由各空间管理员自行启用。停用插件后，它的集成不再出现在类型列表中、不能新建，已有实例照常工作。 + +## 插件的运行方式 + +插件包是一个 `.wkp` 文件（根目录带 `plugin.yaml` 的 zip）。`runtime.type` 决定插件代码在哪里运行： + +| 运行方式 | 说明 | 适用 | +| --- | --- | --- | +| `declarative` | 没有代码，由 WeKnora 解释清单 | 技能、模型厂商定义、远程 MCP | +| `host` | WeKnora 的插件宿主拉起的子进程，`kind` 为 `binary` 或 `python` | 自托管的代码插件 | +| `remote` | 插件作者自行部署的 HTTP 服务，安装时登记地址 | SaaS 类插件、独立团队维护的服务 | + +代码插件与 WeKnora 之间走统一的扩展协议 v1（HTTP + JSON，流式同步用 NDJSON）。 + +## 安装插件 + +1. 以系统管理员身份打开「设置 → 插件管理」，点击「安装插件」。 +2. 上传 `.wkp` 或填写下载地址。WeKnora 先解析插件包，展示它提供的能力、申请的权限（可访问的外部域名、Host API 权限等）和包摘要。 +3. 确认后安装。安装请求带着审阅时的摘要，下载地址在此期间换了内容会被拒绝。 + +同一插件再次安装更高版本即升级，旧版本保留，可在详情中回滚。 + +### 远程插件 + +`runtime.type: remote` 的插件在安装时还需填写服务地址： + +- 服务在内网时，需把地址加入 `SSRF_WHITELIST`。 +- 安装后 WeKnora 生成一个签名密钥，**只显示这一次**。把它配置为服务的 `WEKNORA_PLUGIN_SECRET` 环境变量，服务据此确认请求来自 WeKnora。 +- 插件详情中可以修改服务地址或轮换密钥。轮换后，服务换上新密钥之前的调用都会失败。 +- 服务版本必须与安装的插件包一致，否则插件显示为异常、调用被拒绝。升级时同时升级服务与插件包。 + +## 运行代码插件 + +### 内嵌宿主（默认） + +默认情况下，每个 app 进程自带插件宿主，在本机运行已安装的 `host` 插件： + +- 插件进程只拿到 `WEKNORA_PLUGIN_*` 环境变量，看不到数据库密码等 WeKnora 密钥。 +- 出网流量经宿主的出口代理，只放行插件在 `permissions.egress` 中声明的域名，内网地址一律拒绝。 +- 宿主检查运行的进程与安装的包一致，健康检查失败时按退避重启，并在插件详情中显示为「异常」。 +- Python 插件使用本机的 `python3`（可用 `WEKNORA_PLUGIN_PYTHON` 指定解释器）。Docker app 镜像已带 Python。 + +### 独立插件宿主 + +插件较多、希望把插件与 app 隔开，或 app 节点不便运行 Python 时，可以单独运行 `WeKnora plugin-host`： + +- 它与 app 使用同一镜像，共用数据库、Redis、对象存储和 `SYSTEM_AES_KEY`（或 `JWT_SECRET`）。 +- 启动后每 5 秒在 Redis 中通告自己运行的插件。app 按版本挑选最空闲的宿主，用由 `SYSTEM_AES_KEY` 派生的密钥签名调用它。 +- 宿主停止时先撤回通告，再等进行中的调用结束。 +- 插件详情的「节点状态」中，`plugin-host:` 开头的就是独立宿主。 + +docker compose 启用方式：在 `.env` 中设置 + +```bash +WEKNORA_PLUGIN_EMBEDDED_KINDS=none +WEKNORA_PLUGIN_HOST_API_URL=http://app:8080 +``` + +然后执行： + +```bash +docker compose --profile plugin-host up -d +``` + +Helm 设置 `pluginHost.enabled=true` 即可。使用本地存储（`STORAGE_TYPE=local`）时，插件包存放在 data-files 卷中，该卷需支持多 Pod 读写（ReadWriteMany）。 + +相关环境变量： + +| 变量 | 作用于 | 说明 | +| --- | --- | --- | +| `WEKNORA_PLUGIN_EMBEDDED_KINDS` | app | app 自己运行的 kind（`binary`、`python`，逗号分隔）；`none` 表示全部交给独立宿主。默认本机能跑的都跑 | +| `WEKNORA_PLUGIN_HOST_API_URL` | app、plugin-host | 不在 app 本机运行的插件（远程插件、独立宿主上的插件）回调 Host API 的地址 | +| `WEKNORA_PLUGIN_HOST_KINDS` | plugin-host | 宿主运行的 kind，默认本机能跑的都跑 | +| `WEKNORA_PLUGIN_HOST_ADDR` | plugin-host | 监听地址，默认 `:8081` | +| `WEKNORA_PLUGIN_HOST_URL` | plugin-host | app 访问该宿主的地址，默认 `http://<主机名>:<端口>`；Helm 中为 Pod IP | +| `WEKNORA_PLUGIN_PYTHON` | 两者 | Python 插件的解释器，默认 `python3` | + +独立宿主需要 Redis，且所有节点的 `SYSTEM_AES_KEY`（或 `JWT_SECRET`）必须一致。app 不运行某个 kind、又没有配置独立宿主时，该 kind 的插件在插件详情中显示为加载失败，并说明原因。 + +## 开发插件 + +- **Go**：[pluginsdk](https://github.com/Tencent/WeKnora/tree/main/pluginsdk)，含协议定义、SDK、客户端和一致性测试工具 `weknora-plugin-conformance`。 +- **Python**：[pluginsdk/python](https://github.com/Tencent/WeKnora/tree/main/pluginsdk/python)，Python 3.9+，只依赖标准库。 +- **示例**：[examples/plugins](https://github.com/Tencent/WeKnora/tree/main/examples/plugins)： + - `rss`：数据源连接器，Go； + - `subtitles`：文档解析器，Go，使用 Host API； + - `notebooks`：Jupyter 笔记本解析器，Python。 + +同一个插件既可以打包成 `host` 插件由 WeKnora 运行，也可以作为 `remote` 服务独立部署，代码不用改。