feat(configs+queue): managed-config server with token auth; queue_output bridge

app.go

@@ -35,6 +35,7 @@ import (
 	"os/signal"
 	"runtime"
 	"runtime/debug"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -184,6 +185,14 @@ func (app *App) initWithFlags() {
 	flag.StringVar(&app.svcFlag, "service", "", "service management, options: install,uninstall,start,stop")
 	flag.StringVar(&app.svcUser, "service-user", "", "OS user account used to run the service")
 
+	// -e KEY=VALUE (repeatable): environment overrides applied BEFORE the
+	// config loads. Picked up by $[[env.KEY]] template expansion in the
+	// YAML (OS env wins over the YAML env: section), so deployments can
+	// parameterize any templated setting from the command line instead of
+	// editing config files. Example:
+	//   agent -e MANAGED=true -e REMOTE_CONFIG_SERVERS=http://lp:29000
+	flag.Var(&cliEnvOverrides{}, "e", "environment override KEY=VALUE (repeatable, feeds $[[env.KEY]])")
+
 	if debugFlagInitFunc != nil {
 		debugFlagInitFunc()
 	}
@@ -622,3 +631,20 @@ func (app *App) Run() {
 		log.Error(err)
 	}
 }
+
+// cliEnvOverrides collects repeatable -e KEY=VALUE flags; Apply sets them
+// into the process environment before configuration loading begins.
+type cliEnvOverrides struct {
+	pairs []string
+}
+
+func (c *cliEnvOverrides) String() string { return strings.Join(c.pairs, " ") }
+
+func (c *cliEnvOverrides) Set(v string) error {
+	idx := strings.Index(v, "=")
+	if idx <= 0 {
+		return fmt.Errorf("-e expects KEY=VALUE, got %q", v)
+	}
+	c.pairs = append(c.pairs, v)
+	return os.Setenv(v[:idx], v[idx+1:])
+}

core/api/web.go

@@ -112,6 +112,7 @@ func StartWeb(cfg config.WebAppConfig) {
 
 	registerMCPAutoUIHandler(cfg)
 
+	mountedFuncPatterns := map[string]bool{} // embedding_api 已挂载的 ServeMux 模式
 	if cfg.EmbeddingAPI {
 		if registeredAPIMethodHandler != nil {
 			for k, v := range registeredAPIMethodHandler {
@@ -125,13 +126,19 @@ func StartWeb(cfg config.WebAppConfig) {
 			for k, v := range registeredAPIFuncHandler {
 				log.Debug("register http handler: ", k)
 				uiServeMux.HandleFunc(k, v)
+				mountedFuncPatterns[k] = true
 			}
 		}
 	}
 
 	if cfg.WebsocketConfig.Enabled {
 		websocket.InitWebSocket(cfg.WebsocketConfig)
-		uiServeMux.HandleFunc("/ws", websocket.ServeWs)
+		// embedding_api 可能已把 API 域注册的 /ws (StartAPI 的
+		// HandleAPIFunc) 挂到本 mux — 重复 HandleFunc 在 Go 1.22+
+		// 的 ServeMux 语义下 panic, 跳过。
+		if !mountedFuncPatterns["/ws"] && !mountedFuncPatterns[cfg.WebsocketConfig.BasePath] {
+			uiServeMux.HandleFunc("/ws", websocket.ServeWs)
+		}
 		if registeredWebSocketCommandHandler != nil {
 			for k, v := range registeredWebSocketCommandHandler {
 				log.Debug("register websocket handler: ", k, " ", v)

core/api/websocket/conn.go

@@ -30,7 +30,6 @@ package websocket
 import (
 	log "github.com/cihub/seelog"
 	"github.com/gorilla/websocket"
-	"infini.sh/framework/core/global"
 	"infini.sh/framework/core/util"
 	"net/http"
 	"strings"
@@ -48,8 +47,10 @@ const (
 	// Send pings to peer with this period. Must be less than pongWait.
 	pingPeriod = (pongWait * 9) / 10
 
-	// Maximum message size allowed from peer.
-	maxMessageSize = 512
+	// Maximum message size allowed from peer. The reverse channel streams
+	// chunked proxied HTTP bodies through this hub (base64 chunks of up to
+	// 32KB), so this must comfortably exceed a chunk frame.
+	maxMessageSize = 8 * 1024 * 1024
 )
 
 var upgrader = websocket.Upgrader{
@@ -212,9 +213,7 @@ func ServeWs(w http.ResponseWriter, r *http.Request) {
 		for _, v := range callbacksOnConnect {
 			err := v(c.id, w, r)
 			if err != nil {
-				if global.Env().IsDebug {
-					log.Error(err)
-				}
+				log.Warnf("websocket connection rejected: %v", err)
 				closeMessage := websocket.FormatCloseMessage(websocket.ClosePolicyViolation, err.Error())
 				if closeErr := ws.WriteMessage(websocket.CloseMessage, closeMessage); closeErr != nil {
 					log.Error("Failed to send close message:", closeErr)

core/config/system.go

@@ -302,12 +302,14 @@ type ConfigsConfig struct {
 	ValidConfigsExtensions     []string  `config:"valid_config_extensions"`
 	TLSConfig                  TLSConfig `config:"tls"` //server or client's certs
 	ManagerConfig              struct {
-		LocalConfigsRepoPath string    `config:"local_configs_repo_path"`
-		BasicAuth            BasicAuth `config:"basic_auth"`
+		LocalConfigsRepoPath string            `config:"local_configs_repo_path"`
+		BasicAuth            BasicAuth         `config:"basic_auth"`
+		AccessToken          ucfg.SecretString `config:"access_token"`
 	} `config:"manager"`
-	AlwaysRegisterAfterRestart bool     `config:"always_register_after_restart"`
-	AllowGeneratedMetricsTasks bool     `config:"allow_generated_metrics_tasks"`
-	IgnoredPath                []string `config:"ignored_path"`
+	EnrollmentToken            ucfg.SecretString `config:"enrollment_token"` // one-time registration pass (configs.server.enrollment.required)
+	AlwaysRegisterAfterRestart bool              `config:"always_register_after_restart"`
+	AllowGeneratedMetricsTasks bool              `config:"allow_generated_metrics_tasks"`
+	IgnoredPath                []string          `config:"ignored_path"`
 }
 
 type BasicAuth struct {

core/model/const.go

@@ -0,0 +1,29 @@
+// Copyright (C) INFINI Labs & INFINI LIMITED.
+//
+// The INFINI Framework is offered under the GNU Affero General Public License v3.0
+// and as commercial software.
+//
+// For commercial licensing, contact us at:
+//   - Website: infinilabs.com
+//   - Email: hello@infini.ltd
+//
+// Open Source licensed under AGPL V3:
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package model
+
+const (
+	CredentialIDSystemKey = "credential_id"
+	API_TOKEN             = "X-API-TOKEN"
+)

core/model/instance.go

@@ -59,13 +59,24 @@ type Instance struct {
 	Labels map[string]string `json:"labels,omitempty" elastic_mapping:"labels:{type:object}"`
 	Tags   []string          `json:"tags,omitempty"`
 
+	// Groups is the SERVER-OWNED grouping of instances (e.g. "es",
+	// "gateway-edge"): managed from the management UI, used to target
+	// config delivery (ManagedConfig.Groups). Instances do not report it —
+	// registration/heartbeat upserts preserve the stored value.
+	Groups []string `json:"groups,omitempty" elastic_mapping:"groups:{type:keyword}}"`
+
 	//user can pass
 	Description string `json:"description,omitempty" config:"description" elastic_mapping:"description:{type:keyword}"`
 
 	Endpoint string `json:"endpoint,omitempty" elastic_mapping:"endpoint: { type: keyword }"` //API endpoint
 
 	Host *HostInfo `json:"host,omitempty" elastic_mapping:"host: { type: object }"`
 
+	// AccessToken is the agent's self-generated API token (the console
+	// token-exchange convention): managers store it at registration and
+	// use it for reverse calls (stats, pipeline tasks, proxying).
+	AccessToken *Token `config:"access_token" json:"access_token,omitempty" elastic_mapping:"access_token:{type:object}"`
+
 	Network  NetworkInfo   `json:"network,omitempty" elastic_mapping:"network: { type: object }"`
 	Services []ServiceInfo `json:"services,omitempty" elastic_mapping:"services: { type: object }"`
 	Status   string        `json:"status,omitempty" elastic_mapping:"status: { type: keyword, copy_to:search_text }"`
@@ -137,7 +148,20 @@ func GetInstanceInfo() Instance {
 
 	_, publicIP, _, _ := util.GetPublishNetworkDeviceInfo(global.Env().SystemConfig.NodeConfig.MajorIpPattern)
 
-	instance.Endpoint = global.Env().SystemConfig.APIConfig.GetEndpoint()
+	// The advertised endpoint must point at a server that actually serves
+	// requests. Deployments commonly disable the dedicated API port
+	// (api.enabled: false) and serve everything on the web port — in that
+	// case advertise the web address, not the (unserving) API default.
+	sysCfg := global.Env().SystemConfig
+	if sysCfg.APIConfig.Enabled {
+		instance.Endpoint = sysCfg.APIConfig.GetEndpoint()
+	} else {
+		schema := "http"
+		if sysCfg.WebAppConfig.TLSConfig.TLSEnabled {
+			schema = "https"
+		}
+		instance.Endpoint = fmt.Sprintf("%s://%s", schema, sysCfg.WebAppConfig.NetworkConfig.GetPublishAddr())
+	}
 
 	ips := util.GetLocalIPs()
 	if len(ips) > 0 {

core/model/token.go

@@ -0,0 +1,9 @@
+/* Copyright © INFINI Ltd. All rights reserved. */
+
+package model
+
+// Token is a bearer credential carried on registered instances (the
+// agent's self-generated API token, stored at registration).
+type Token struct {
+	Value string `json:"value,omitempty" config:"value"`
+}

core/pipeline/record.go

@@ -126,3 +126,33 @@ func AppendFailureTag(ctx *Context, tag string) {
 		*tags = append(*tags, tag)
 	}
 }
+
+// CloneContextKey is the Context key under which the clone processor
+// deposits cloned records; the batch-splitting host materializes them as
+// additional batch members after the sub-chain finishes the original.
+const CloneContextKey = "record_clones"
+
+// AppendClone registers a clone of the current record for the host to
+// materialize (see clone processor). No-op outside a record scope.
+func AppendClone(ctx *Context, rec *event.Event) {
+	if ctx == nil || rec == nil {
+		return
+	}
+	if list, ok := ctx.Get(CloneContextKey).(*[]*event.Event); ok {
+		*list = append(*list, rec)
+	}
+}
+
+// TakeClones returns and clears the pending clones registered in this
+// context (host-side; call after the per-record sub-chain).
+func TakeClones(ctx *Context) []*event.Event {
+	if ctx == nil {
+		return nil
+	}
+	if list, ok := ctx.Get(CloneContextKey).(*[]*event.Event); ok && len(*list) > 0 {
+		out := *list
+		*list = nil
+		return out
+	}
+	return nil
+}

core/pipeline/register.go

@@ -251,6 +251,16 @@ type Constructor func(config *config.Config) (ProcessorBase, error)
 
 var registry = NewNamespace()
 
+// LookupProcessorConstructor resolves a processor constructor by its
+// registered (bare) name — nil when unknown. Hosts use this to build
+// ad-hoc chains (e.g. dry-run replay of a spec's processor list).
+func LookupProcessorConstructor(name string) ProcessorConstructor {
+	if ctor, ok := registry.ProcessorConstructors()[name]; ok {
+		return ctor
+	}
+	return nil
+}
+
 func RegisterProcessorPlugin(name string, constructor ProcessorConstructor) {
 	err := registry.RegisterProcessor(name, constructor)
 	if err != nil {