diff --git a/config/generated_framework-info.go b/config/generated_framework-info.go
new file mode 100644
index 000000000..d7d4a8c96
--- /dev/null
+++ b/config/generated_framework-info.go
@@ -0,0 +1,4 @@
+package config
+
+const LastFrameworkCommitLog = "N/A"
+const LastFrameworkVendorCommitLog = "N/A"
diff --git a/core/api/websocket/reverse/manager.go b/core/api/websocket/reverse/manager.go
new file mode 100644
index 000000000..2a040e1e4
--- /dev/null
+++ b/core/api/websocket/reverse/manager.go
@@ -0,0 +1,312 @@
+package reverse
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "net/http"
+ "strings"
+ "sync"
+ "time"
+
+ "infini.sh/framework/core/util"
+)
+
+const (
+ DefaultTimeout = 30 * time.Second
+ DefaultMaxResponseBytes = 8 * 1024 * 1024
+ DefaultReconnectWait = 6 * time.Second
+ DefaultReconnectPoll = 200 * time.Millisecond
+)
+
+var (
+ ErrDisconnected = errors.New("reverse channel disconnected")
+ ErrNotConnected = errors.New("reverse channel is not connected")
+)
+
+type ManagerOptions struct {
+ DefaultTimeout time.Duration
+ MaxResponseBytes int
+ ReconnectWait time.Duration
+ ReconnectPoll time.Duration
+}
+
+type pendingResponse struct {
+ peerID string
+ body bytes.Buffer
+ status int
+ err error
+ done chan struct{}
+ completed bool
+}
+
+type SessionManager struct {
+ options ManagerOptions
+ mu sync.Mutex
+ pendingSessions map[string]string
+ activeSessions map[string]string
+ activeSessionsByID map[string]string
+ pendingResponses map[string]*pendingResponse
+}
+
+func NewSessionManager(options ManagerOptions) *SessionManager {
+ if options.DefaultTimeout <= 0 {
+ options.DefaultTimeout = DefaultTimeout
+ }
+ if options.MaxResponseBytes <= 0 {
+ options.MaxResponseBytes = DefaultMaxResponseBytes
+ }
+ if options.ReconnectWait <= 0 {
+ options.ReconnectWait = DefaultReconnectWait
+ }
+ if options.ReconnectPoll <= 0 {
+ options.ReconnectPoll = DefaultReconnectPoll
+ }
+ return &SessionManager{
+ options: options,
+ pendingSessions: map[string]string{},
+ activeSessions: map[string]string{},
+ activeSessionsByID: map[string]string{},
+ pendingResponses: map[string]*pendingResponse{},
+ }
+}
+
+func (m *SessionManager) RegisterPendingSession(sessionID, peerID string) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.pendingSessions[sessionID] = strings.TrimSpace(peerID)
+}
+
+func (m *SessionManager) ActivateSession(sessionID, peerID string) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ peerID = strings.TrimSpace(peerID)
+ if expectedPeerID, ok := m.pendingSessions[sessionID]; !ok || expectedPeerID != peerID {
+ return fmt.Errorf("session handshake mismatch")
+ }
+ delete(m.pendingSessions, sessionID)
+
+ if previousSession, ok := m.activeSessions[peerID]; ok && previousSession != sessionID {
+ delete(m.activeSessionsByID, previousSession)
+ }
+
+ m.activeSessions[peerID] = sessionID
+ m.activeSessionsByID[sessionID] = peerID
+ return nil
+}
+
+func (m *SessionManager) HandleHelloPayload(payload string) error {
+ msg, err := ParseHelloPayload(payload)
+ if err != nil {
+ return err
+ }
+ return m.ActivateSession(msg.SessionID, msg.PeerID)
+}
+
+func (m *SessionManager) HandleResponsePayload(payload string) error {
+ msg, err := ParseResponsePayload(payload)
+ if err != nil {
+ return err
+ }
+ m.acceptResponse(msg)
+ return nil
+}
+
+func (m *SessionManager) OnDisconnect(sessionID string) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ delete(m.pendingSessions, sessionID)
+ peerID, ok := m.activeSessionsByID[sessionID]
+ if !ok {
+ return
+ }
+
+ delete(m.activeSessionsByID, sessionID)
+ if currentSession, exists := m.activeSessions[peerID]; exists && currentSession == sessionID {
+ delete(m.activeSessions, peerID)
+ }
+ m.failPendingLocked(peerID, ErrDisconnected)
+}
+
+func (m *SessionManager) IsConnected(peerID string) bool {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ sessionID, ok := m.activeSessions[peerID]
+ return ok && sessionID != ""
+}
+
+func (m *SessionManager) WaitForReconnect(ctx context.Context, peerID string) bool {
+ waitCtx, cancel := context.WithTimeout(ctx, m.options.ReconnectWait)
+ defer cancel()
+
+ if m.IsConnected(peerID) {
+ return true
+ }
+
+ ticker := time.NewTicker(m.options.ReconnectPoll)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-waitCtx.Done():
+ return false
+ case <-ticker.C:
+ if m.IsConnected(peerID) {
+ return true
+ }
+ }
+ }
+}
+
+func IsRecoverableError(err error) bool {
+ return errors.Is(err, ErrDisconnected) || errors.Is(err, ErrNotConnected)
+}
+
+func (m *SessionManager) ProxyRequest(peerID string, req *util.Request, headers http.Header, send func(sessionID, payload string) error, responseObjectToUnmarshal interface{}) (*util.Result, error) {
+ if req == nil {
+ return nil, fmt.Errorf("request is nil")
+ }
+
+ ctx := req.Context
+ if ctx == nil {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(context.Background(), m.options.DefaultTimeout)
+ defer cancel()
+ } else if _, hasDeadline := ctx.Deadline(); !hasDeadline {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(ctx, m.options.DefaultTimeout)
+ defer cancel()
+ }
+
+ var lastErr error
+ for attempt := 0; attempt < 2; attempt++ {
+ res, err := m.proxyRequestOnce(ctx, strings.TrimSpace(peerID), req, headers, send, responseObjectToUnmarshal)
+ if err == nil {
+ return res, nil
+ }
+ lastErr = err
+ if attempt == 0 && IsRecoverableError(err) && m.WaitForReconnect(ctx, peerID) {
+ continue
+ }
+ return res, err
+ }
+ return nil, lastErr
+}
+
+func (m *SessionManager) proxyRequestOnce(ctx context.Context, peerID string, req *util.Request, headers http.Header, send func(sessionID, payload string) error, responseObjectToUnmarshal interface{}) (*util.Result, error) {
+ requestID := util.GetUUID()
+ msg := RequestMessage{
+ RequestID: requestID,
+ PeerID: peerID,
+ Method: req.Method,
+ Path: req.Path,
+ Headers: headers,
+ }
+ msg.SetBody(req.Body)
+ if authorization := strings.TrimSpace(msg.Headers.Get("Authorization")); strings.HasPrefix(strings.ToLower(authorization), "bearer ") {
+ msg.AccessToken = strings.TrimSpace(authorization[7:])
+ }
+
+ pending := &pendingResponse{
+ peerID: peerID,
+ done: make(chan struct{}),
+ }
+
+ m.mu.Lock()
+ sessionID, ok := m.activeSessions[peerID]
+ if !ok || sessionID == "" {
+ m.mu.Unlock()
+ return nil, fmt.Errorf("%w for peer [%s]", ErrNotConnected, peerID)
+ }
+ m.pendingResponses[requestID] = pending
+ m.mu.Unlock()
+
+ if err := send(sessionID, FormatRequestCommand(msg)); err != nil {
+ m.mu.Lock()
+ delete(m.pendingResponses, requestID)
+ m.mu.Unlock()
+ return nil, err
+ }
+
+ select {
+ case <-pending.done:
+ case <-ctx.Done():
+ m.mu.Lock()
+ delete(m.pendingResponses, requestID)
+ m.mu.Unlock()
+ return nil, ctx.Err()
+ }
+
+ if pending.err != nil {
+ return nil, pending.err
+ }
+
+ res := &util.Result{
+ Body: pending.body.Bytes(),
+ StatusCode: pending.status,
+ }
+ if res.StatusCode != http.StatusOK {
+ return res, fmt.Errorf("request error: %s", string(res.Body))
+ }
+ if responseObjectToUnmarshal != nil && len(res.Body) > 0 {
+ return res, util.FromJSONBytes(res.Body, responseObjectToUnmarshal)
+ }
+ return res, nil
+}
+
+func (m *SessionManager) acceptResponse(msg ResponseMessage) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ pending, ok := m.pendingResponses[msg.RequestID]
+ if !ok || pending.completed {
+ return
+ }
+ if msg.PeerID != "" && pending.peerID != "" && msg.PeerID != pending.peerID {
+ return
+ }
+
+ if msg.Chunk != "" {
+ chunk, err := msg.ChunkBytes()
+ if err != nil {
+ m.completePendingLocked(msg.RequestID, pending, 0, fmt.Errorf("decode reverse response chunk: %w", err))
+ return
+ }
+ if pending.body.Len()+len(chunk) > m.options.MaxResponseBytes {
+ m.completePendingLocked(msg.RequestID, pending, 0, fmt.Errorf("reverse response exceeds %d bytes", m.options.MaxResponseBytes))
+ return
+ }
+ _, _ = pending.body.Write(chunk)
+ }
+
+ if msg.Done {
+ status := msg.Status
+ if status == 0 {
+ status = http.StatusOK
+ }
+ m.completePendingLocked(msg.RequestID, pending, status, nil)
+ }
+}
+
+func (m *SessionManager) completePendingLocked(requestID string, pending *pendingResponse, status int, err error) {
+ if pending.completed {
+ return
+ }
+ pending.completed = true
+ pending.status = status
+ pending.err = err
+ close(pending.done)
+ delete(m.pendingResponses, requestID)
+}
+
+func (m *SessionManager) failPendingLocked(peerID string, err error) {
+ for requestID, pending := range m.pendingResponses {
+ if pending.peerID != peerID {
+ continue
+ }
+ m.completePendingLocked(requestID, pending, 0, err)
+ }
+}
diff --git a/core/api/websocket/reverse/manager_test.go b/core/api/websocket/reverse/manager_test.go
new file mode 100644
index 000000000..bc94cb97b
--- /dev/null
+++ b/core/api/websocket/reverse/manager_test.go
@@ -0,0 +1,73 @@
+package reverse
+
+import (
+ "net/http"
+ "strings"
+ "testing"
+
+ "infini.sh/framework/core/util"
+)
+
+func TestSessionManagerProxyRequestRoundTrip(t *testing.T) {
+ manager := NewSessionManager(ManagerOptions{})
+ manager.RegisterPendingSession("session-1", "peer-1")
+ if err := manager.ActivateSession("session-1", "peer-1"); err != nil {
+ t.Fatalf("activate session: %v", err)
+ }
+
+ headers := http.Header{}
+ headers.Set("Authorization", "Bearer token-1")
+
+ send := func(sessionID, payload string) error {
+ if sessionID != "session-1" {
+ t.Fatalf("unexpected session id: %s", sessionID)
+ }
+ if !strings.HasPrefix(payload, RequestCommand+" ") {
+ t.Fatalf("unexpected payload: %s", payload)
+ }
+ msg, err := ParseRequestPayload(strings.TrimPrefix(payload, RequestCommand+" "))
+ if err != nil {
+ t.Fatalf("parse request payload: %v", err)
+ }
+ if msg.BearerToken() != "token-1" {
+ t.Fatalf("unexpected bearer token: %s", msg.BearerToken())
+ }
+ return WriteChunkedResponse(func(responsePayload string) error {
+ if !strings.HasPrefix(responsePayload, ResponseCommand+" ") {
+ t.Fatalf("unexpected response payload: %s", responsePayload)
+ }
+ return manager.HandleResponsePayload(strings.TrimPrefix(responsePayload, ResponseCommand+" "))
+ }, msg.RequestID, msg.PeerID, http.StatusOK, []byte(`{"ack":true}`), DefaultResponseChunkBytes)
+ }
+
+ var response map[string]bool
+ req := &util.Request{Method: http.MethodGet, Path: "/stats"}
+ res, err := manager.ProxyRequest("peer-1", req, headers, send, &response)
+ if err != nil {
+ t.Fatalf("proxy request: %v", err)
+ }
+ if res.StatusCode != http.StatusOK {
+ t.Fatalf("unexpected status: %d", res.StatusCode)
+ }
+ if !response["ack"] {
+ t.Fatal("expected response to unmarshal")
+ }
+}
+
+func TestSessionManagerDisconnectFailsPendingRequest(t *testing.T) {
+ manager := NewSessionManager(ManagerOptions{})
+ manager.RegisterPendingSession("session-1", "peer-1")
+ if err := manager.ActivateSession("session-1", "peer-1"); err != nil {
+ t.Fatalf("activate session: %v", err)
+ }
+
+ send := func(sessionID, payload string) error {
+ manager.OnDisconnect(sessionID)
+ return nil
+ }
+
+ _, err := manager.ProxyRequest("peer-1", &util.Request{Method: http.MethodGet, Path: "/stats"}, nil, send, nil)
+ if !IsRecoverableError(err) {
+ t.Fatalf("expected recoverable disconnect error, got %v", err)
+ }
+}
diff --git a/core/api/websocket/reverse/protocol.go b/core/api/websocket/reverse/protocol.go
new file mode 100644
index 000000000..173337c88
--- /dev/null
+++ b/core/api/websocket/reverse/protocol.go
@@ -0,0 +1,162 @@
+package reverse
+
+import (
+ "encoding/base64"
+ "net/http"
+ "strings"
+
+ "infini.sh/framework/core/util"
+)
+
+const (
+ HeaderPeerID = "X-INFINI-INSTANCE-ID"
+ HelloCommand = "reverse_hello"
+ RequestCommand = "reverse_request"
+ ResponseCommand = "reverse_response"
+ DefaultResponseChunkBytes = 32 * 1024
+)
+
+type HelloMessage struct {
+ SessionID string `json:"session_id"`
+ PeerID string `json:"instance_id"`
+}
+
+type RequestMessage struct {
+ RequestID string `json:"request_id"`
+ PeerID string `json:"instance_id"`
+ Method string `json:"method"`
+ Path string `json:"path"`
+ Body string `json:"body,omitempty"`
+ Headers http.Header `json:"headers,omitempty"`
+ AccessToken string `json:"access_token,omitempty"`
+}
+
+type ResponseMessage struct {
+ RequestID string `json:"request_id"`
+ PeerID string `json:"instance_id"`
+ Chunk string `json:"chunk,omitempty"`
+ Status int `json:"status,omitempty"`
+ Done bool `json:"done,omitempty"`
+}
+
+func ParseHelloPayload(payload string) (HelloMessage, error) {
+ msg := HelloMessage{}
+ return msg, util.FromJSONBytes([]byte(payload), &msg)
+}
+
+func ParseRequestPayload(payload string) (RequestMessage, error) {
+ msg := RequestMessage{}
+ return msg, util.FromJSONBytes([]byte(payload), &msg)
+}
+
+func ParseResponsePayload(payload string) (ResponseMessage, error) {
+ msg := ResponseMessage{}
+ return msg, util.FromJSONBytes([]byte(payload), &msg)
+}
+
+func FormatHelloCommand(msg HelloMessage) string {
+ return HelloCommand + " " + string(util.MustToJSONBytes(msg))
+}
+
+func FormatRequestCommand(msg RequestMessage) string {
+ return RequestCommand + " " + string(util.MustToJSONBytes(msg))
+}
+
+func FormatResponseCommand(msg ResponseMessage) string {
+ return ResponseCommand + " " + string(util.MustToJSONBytes(msg))
+}
+
+func (m *RequestMessage) SetBody(body []byte) {
+ if len(body) == 0 {
+ m.Body = ""
+ return
+ }
+ m.Body = base64.StdEncoding.EncodeToString(body)
+}
+
+func (m RequestMessage) BodyBytes() ([]byte, error) {
+ if m.Body == "" {
+ return nil, nil
+ }
+ return base64.StdEncoding.DecodeString(m.Body)
+}
+
+func (m RequestMessage) NormalizedHeaders() http.Header {
+ headers := http.Header{}
+ for key, values := range m.Headers {
+ copied := append([]string(nil), values...)
+ headers[key] = copied
+ }
+ if headers.Get("Authorization") == "" && strings.TrimSpace(m.AccessToken) != "" {
+ headers.Set("Authorization", "Bearer "+strings.TrimSpace(m.AccessToken))
+ }
+ return headers
+}
+
+func (m RequestMessage) ApplyHeaders(req *http.Request) {
+ if req == nil {
+ return
+ }
+ if req.Header == nil {
+ req.Header = http.Header{}
+ }
+ for key := range req.Header {
+ req.Header.Del(key)
+ }
+ for key, values := range m.NormalizedHeaders() {
+ for _, value := range values {
+ req.Header.Add(key, value)
+ }
+ }
+}
+
+func (m RequestMessage) BearerToken() string {
+ value := strings.TrimSpace(m.NormalizedHeaders().Get("Authorization"))
+ if !strings.HasPrefix(strings.ToLower(value), "bearer ") {
+ return ""
+ }
+ return strings.TrimSpace(value[7:])
+}
+
+func (m *ResponseMessage) SetChunk(body []byte) {
+ if len(body) == 0 {
+ m.Chunk = ""
+ return
+ }
+ m.Chunk = base64.StdEncoding.EncodeToString(body)
+}
+
+func (m ResponseMessage) ChunkBytes() ([]byte, error) {
+ if m.Chunk == "" {
+ return nil, nil
+ }
+ return base64.StdEncoding.DecodeString(m.Chunk)
+}
+
+func WriteChunkedResponse(write func(payload string) error, requestID, peerID string, status int, body []byte, chunkBytes int) error {
+ if chunkBytes <= 0 {
+ chunkBytes = DefaultResponseChunkBytes
+ }
+ for start := 0; start < len(body); start += chunkBytes {
+ end := start + chunkBytes
+ if end > len(body) {
+ end = len(body)
+ }
+ msg := ResponseMessage{
+ RequestID: requestID,
+ PeerID: peerID,
+ }
+ msg.SetChunk(body[start:end])
+ if err := write(FormatResponseCommand(msg)); err != nil {
+ return err
+ }
+ }
+
+ done := ResponseMessage{
+ RequestID: requestID,
+ PeerID: peerID,
+ Status: status,
+ Done: true,
+ }
+ return write(FormatResponseCommand(done))
+}
diff --git a/core/api/websocket/reverse/protocol_test.go b/core/api/websocket/reverse/protocol_test.go
new file mode 100644
index 000000000..a2085d9da
--- /dev/null
+++ b/core/api/websocket/reverse/protocol_test.go
@@ -0,0 +1,43 @@
+package reverse
+
+import (
+ "net/http"
+ "testing"
+)
+
+func TestRequestMessageNormalizedHeadersFallsBackToLegacyAccessToken(t *testing.T) {
+ msg := RequestMessage{
+ AccessToken: "token-1",
+ }
+
+ headers := msg.NormalizedHeaders()
+ if got := headers.Get("Authorization"); got != "Bearer token-1" {
+ t.Fatalf("unexpected authorization header: %s", got)
+ }
+ if got := msg.BearerToken(); got != "token-1" {
+ t.Fatalf("unexpected bearer token: %s", got)
+ }
+}
+
+func TestRequestMessageApplyHeaders(t *testing.T) {
+ msg := RequestMessage{
+ Headers: http.Header{
+ "Authorization": []string{"Bearer token-2"},
+ "X-Test": []string{"value"},
+ },
+ }
+ req, _ := http.NewRequest(http.MethodGet, "http://example.com", nil)
+ req.Header.Set("Existing", "old")
+
+ msg.ApplyHeaders(req)
+
+ if req.Header.Get("Existing") != "" {
+ t.Fatal("expected old header to be removed")
+ }
+ if req.Header.Get("Authorization") != "Bearer token-2" {
+ t.Fatalf("unexpected authorization header: %s", req.Header.Get("Authorization"))
+ }
+ if req.Header.Get("X-Test") != "value" {
+ t.Fatalf("unexpected x-test header: %s", req.Header.Get("X-Test"))
+ }
+}
diff --git a/core/config/system.go b/core/config/system.go
index 8824e028a..a867b3e74 100755
--- a/core/config/system.go
+++ b/core/config/system.go
@@ -304,6 +304,9 @@ type ConfigsConfig struct {
ManagerConfig struct {
LocalConfigsRepoPath string `config:"local_configs_repo_path"`
BasicAuth BasicAuth `config:"basic_auth"`
+ // Token is the static manager token (Authorization: Bearer) used
+ // before the manager mints a per-instance token at registration.
+ Token ucfg.SecretString `config:"token"`
} `config:"manager"`
AlwaysRegisterAfterRestart bool `config:"always_register_after_restart"`
AllowGeneratedMetricsTasks bool `config:"allow_generated_metrics_tasks"`
diff --git a/core/otel/bench_test.go b/core/otel/bench_test.go
new file mode 100644
index 000000000..728e67fb7
--- /dev/null
+++ b/core/otel/bench_test.go
@@ -0,0 +1,25 @@
+package otel
+
+import (
+ "testing"
+ "time"
+ "infini.sh/framework/core/event"
+ "infini.sh/framework/core/util"
+)
+
+func benchEvent() *event.Event {
+ return &event.Event{
+ Timestamp: time.Date(2026, 8, 15, 10, 0, 0, 0, time.UTC),
+ Fields: util.MapStr{"message": "ERROR payment-svc: connection refused to db-1:5432, retry 3", "log_level": "ERROR", "service_name": "payment-svc"},
+ }
+}
+
+func BenchmarkEnvelopeRoundTrip(b *testing.B) {
+ data, _ := EncodeEnvelope(benchEvent())
+ b.SetBytes(int64(len(data)))
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ rec, _ := DecodeEnvelope(data)
+ _, _ = EncodeEnvelope(rec)
+ }
+}
diff --git a/core/otel/convert.go b/core/otel/convert.go
new file mode 100644
index 000000000..c3e8c96ec
--- /dev/null
+++ b/core/otel/convert.go
@@ -0,0 +1,174 @@
+// 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 .
+
+/* ©INFINI, All Rights Reserved.
+ * mail: contact#infini.ltd */
+
+package otel
+
+import (
+ "time"
+
+ "infini.sh/framework/core/event"
+ "infini.sh/framework/core/util"
+)
+
+// FromEvent projects a framework event.Event onto the canonical OTel
+// LogRecord model.
+//
+// Mapping (see package doc):
+//
+// event.Timestamp -> Timestamp
+// event.Fields -> Attributes (plus the typed fields below)
+// event.Meta["resource"] -> Resource
+// Fields["log_level"] -> SeverityText
+// Fields["severity_number"]-> SeverityNumber
+// Fields["message"] -> Body
+// Fields["event_name"] -> EventName
+// Fields["trace_id"] etc. -> TraceID/SpanID/TraceFlags
+// Fields["observed_timestamp"] -> ObservedTimestamp
+func FromEvent(e *event.Event) *LogRecord {
+ if e == nil {
+ return nil
+ }
+ r := &LogRecord{
+ Timestamp: e.Timestamp,
+ Attributes: util.MapStr{},
+ }
+ if e.Fields != nil {
+ r.Attributes = e.Fields.Clone()
+ }
+ if v, ok := resourceMap(e.Meta); ok {
+ r.Resource = v
+ }
+
+ r.SeverityText, _ = r.Attributes[FieldLogLevel].(string)
+ r.Body = r.Attributes[FieldMessage]
+ r.EventName, _ = r.Attributes[FieldEventName].(string)
+ r.TraceID, _ = r.Attributes[FieldTraceID].(string)
+ r.SpanID, _ = r.Attributes[FieldSpanID].(string)
+ r.TraceFlags, _ = r.Attributes[FieldTraceFlags].(string)
+ if n, ok := toInt64(r.Attributes[FieldSeverityNumber]); ok {
+ r.SeverityNumber = int32(n)
+ }
+ r.ObservedTimestamp = parseTimeField(r.Attributes[FieldObservedTimestamp])
+ return r
+}
+
+// ToEvent renders the LogRecord back onto a framework event.Event. The
+// typed fields are written into Fields under their canonical snake_case
+// keys so the event round-trips losslessly.
+func (r *LogRecord) ToEvent() *event.Event {
+ e := &event.Event{
+ Timestamp: r.Timestamp,
+ Fields: util.MapStr{},
+ }
+ if r.Attributes != nil {
+ e.Fields = r.Attributes.Clone()
+ } else {
+ e.Fields = util.MapStr{}
+ }
+
+ if r.ObservedTimestamp.IsZero() {
+ r.ObservedTimestamp = time.Now().UTC()
+ }
+ e.Fields[FieldObservedTimestamp] = r.ObservedTimestamp.UTC().Format(time.RFC3339Nano)
+ if r.SeverityText != "" {
+ e.Fields[FieldLogLevel] = r.SeverityText
+ }
+ if r.SeverityNumber != 0 {
+ e.Fields[FieldSeverityNumber] = r.SeverityNumber
+ }
+ if r.Body != nil {
+ e.Fields[FieldMessage] = r.Body
+ }
+ if r.EventName != "" {
+ e.Fields[FieldEventName] = r.EventName
+ }
+ if r.TraceID != "" {
+ e.Fields[FieldTraceID] = r.TraceID
+ }
+ if r.SpanID != "" {
+ e.Fields[FieldSpanID] = r.SpanID
+ }
+ if r.TraceFlags != "" {
+ e.Fields[FieldTraceFlags] = r.TraceFlags
+ }
+ if len(r.Resource) > 0 {
+ if e.Meta == nil {
+ e.Meta = util.MapStr{}
+ }
+ e.Meta[MetaResourceKey] = r.Resource.Clone()
+ }
+ return e
+}
+
+func resourceMap(meta util.MapStr) (util.MapStr, bool) {
+ if meta == nil {
+ return nil, false
+ }
+ v, ok := meta[MetaResourceKey]
+ if !ok {
+ return nil, false
+ }
+ switch m := v.(type) {
+ case util.MapStr:
+ return m, true
+ case map[string]interface{}:
+ return util.MapStr(m), true
+ }
+ return nil, false
+}
+
+func toInt64(v interface{}) (int64, bool) {
+ switch n := v.(type) {
+ case int:
+ return int64(n), true
+ case int32:
+ return int64(n), true
+ case int64:
+ return n, true
+ case float64:
+ return int64(n), true
+ case float32:
+ return int64(n), true
+ }
+ return 0, false
+}
+
+func parseTimeField(v interface{}) time.Time {
+ switch t := v.(type) {
+ case time.Time:
+ return t
+ case util.Time:
+ return time.Time(t)
+ case string:
+ if ts, err := time.Parse(time.RFC3339Nano, t); err == nil {
+ return ts
+ }
+ if ts, err := time.Parse(time.RFC3339, t); err == nil {
+ return ts
+ }
+ }
+ return time.Time{}
+}
diff --git a/core/otel/envelope.go b/core/otel/envelope.go
new file mode 100644
index 000000000..f7b51f9ac
--- /dev/null
+++ b/core/otel/envelope.go
@@ -0,0 +1,109 @@
+// 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 .
+
+/* ©INFINI, All Rights Reserved.
+ * mail: contact#infini.ltd */
+
+package otel
+
+import (
+ "encoding/json"
+ "time"
+
+ "infini.sh/framework/core/event"
+ "infini.sh/framework/core/util"
+)
+
+// Envelope is the on-the-wire (queue) representation of one log record,
+// byte-compatible with the agent's LogEvent JSON:
+//
+// {"agent":{...},"metadata":{...},"payload":{...},"timestamp":"..."}
+//
+// payload carries the record Attributes (canonical snake_case keys plus
+// the typed otel fields, see log_record.go); metadata carries the stable
+// Resource attributes; agent identifies the collecting agent.
+type Envelope struct {
+ Agent *event.AgentMeta `json:"agent,omitempty"`
+ Meta util.MapStr `json:"metadata,omitempty"`
+ Fields util.MapStr `json:"payload,omitempty"`
+ Timestamp string `json:"timestamp,omitempty"`
+}
+
+// DecodeEnvelope decodes one queue message payload into an event.Event.
+//
+// It tolerates both the agent LogEvent envelope and a bare JSON map
+// (which is then treated as the record's Attributes wholesale).
+func DecodeEnvelope(data []byte) (*event.Event, error) {
+ var env Envelope
+ if err := json.Unmarshal(data, &env); err != nil {
+ return nil, err
+ }
+
+ e := &event.Event{Agent: env.Agent}
+ if e.Fields == nil {
+ e.Fields = util.MapStr{}
+ }
+ e.Fields = env.Fields
+ if e.Fields == nil {
+ e.Fields = util.MapStr{}
+ }
+
+ // metadata holds the free-form/stable attributes; expose the otel
+ // Resource collection under Meta["resource"] for model symmetry.
+ e.Meta = util.MapStr{}
+ if len(env.Meta) > 0 {
+ e.Meta = env.Meta.Clone()
+ if _, ok := e.Meta[MetaResourceKey]; !ok {
+ e.Meta[MetaResourceKey] = env.Meta.Clone()
+ }
+ }
+
+ if env.Timestamp != "" {
+ if ts, err := time.Parse(time.RFC3339Nano, env.Timestamp); err == nil {
+ e.Timestamp = ts
+ }
+ }
+ if e.Timestamp.IsZero() {
+ e.Timestamp = time.Now().UTC()
+ }
+ return e, nil
+}
+
+// EncodeEnvelope encodes an event.Event back into the queue envelope.
+func EncodeEnvelope(e *event.Event) ([]byte, error) {
+ env := Envelope{Agent: e.Agent}
+ if len(e.Fields) > 0 {
+ env.Fields = e.Fields
+ }
+ if len(e.Meta) > 0 {
+ if res, ok := e.Meta[MetaResourceKey].(util.MapStr); ok && len(e.Meta) == 1 {
+ env.Meta = res
+ } else {
+ env.Meta = e.Meta.Clone()
+ }
+ }
+ if !e.Timestamp.IsZero() {
+ env.Timestamp = e.Timestamp.UTC().Format(time.RFC3339Nano)
+ }
+ return json.Marshal(env)
+}
diff --git a/core/otel/log_record.go b/core/otel/log_record.go
new file mode 100644
index 000000000..f1928c35c
--- /dev/null
+++ b/core/otel/log_record.go
@@ -0,0 +1,151 @@
+// 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 .
+
+/* ©INFINI, All Rights Reserved.
+ * mail: contact#infini.ltd */
+
+// Package otel implements the canonical log data model of INFINI's data
+// pipeline, aligned with the OpenTelemetry Logs Data Model:
+//
+// https://opentelemetry.io/docs/specs/otel/logs/data-model/
+//
+// Structural standard : OTel LogRecord (top-level typed fields + Resource
+// /Attributes attribute collections).
+// Field naming standard: lowercase snake_case keys ("log_level",
+// "host_name", ...), matching the house convention. The
+// field_standardize processor (enterprise) can switch attribute naming to
+// dotted OTel semconv ("log.level") or keep nested maps when needed.
+//
+// The in-memory working unit of the pipeline stays framework's
+// event.Event: Attributes live in event.Fields, stable source attributes
+// (Resource) live in event.Meta["resource"] so they are not repeated per
+// record on the wire. LogRecord is the boundary type used when decoding
+// (OTLP intake) or encoding (OTLP export) and as the model reference.
+package otel
+
+import (
+ "time"
+
+ "infini.sh/framework/core/util"
+)
+
+// Canonical snake_case keys of the OTel top-level fields, as they appear
+// inside event.Fields (i.e. the LogRecord's Attributes collection).
+const (
+ FieldTimestamp = "timestamp"
+ FieldObservedTimestamp = "observed_timestamp"
+ FieldLogLevel = "log_level" // OTel SeverityText, original string
+ FieldSeverityNumber = "severity_number" // OTel SeverityNumber, 1-24
+ FieldMessage = "message" // OTel Body
+ FieldEventName = "event_name"
+ FieldTraceID = "trace_id"
+ FieldSpanID = "span_id"
+ FieldTraceFlags = "trace_flags"
+
+ // MetaResourceKey is the event.Meta key under which the stable
+ // Resource attribute collection (host_name, service_name, ...) is
+ // carried, separated from per-record Attributes.
+ MetaResourceKey = "resource"
+)
+
+// Common Resource attribute keys (stable per source, snake_case).
+const (
+ ResourceHostName = "host_name"
+ ResourceHostIP = "host_ip"
+ ResourceServiceName = "service_name"
+ ResourceServiceVersion = "service_version"
+ ResourceProcessPID = "process_pid"
+ ResourceProcessName = "process_name"
+ ResourceCloudProvider = "cloud_provider"
+ ResourceCloudRegion = "cloud_region"
+ ResourceCloudAccountID = "cloud_account_id"
+)
+
+// OTel SeverityNumber scale (see the Logs Data Model): 1-24, the 9 "well
+// known" levels land on 1,5,9,13,17,21.
+const (
+ SeverityTrace1 int32 = 1
+ SeverityDebug int32 = 5
+ SeverityInfo int32 = 9
+ SeverityWarn int32 = 13
+ SeverityError int32 = 17
+ SeverityFatal int32 = 21
+)
+
+// LogRecord mirrors the OpenTelemetry LogRecord.
+type LogRecord struct {
+ // Time when the event occurred, measured by the origin clock.
+ Timestamp time.Time
+ // Time when the collection system observed the event.
+ ObservedTimestamp time.Time
+
+ TraceID string // W3C trace id, hex
+ SpanID string // W3C span id, hex
+ TraceFlags string
+
+ SeverityText string // original severity string, e.g. "ERROR"
+ SeverityNumber int32 // normalized severity, 1-24
+ Body interface{} // log payload, usually the message string
+ EventName string
+
+ // Resource describes the source of the log; it is identical across
+ // records coming from the same source (host/service/process).
+ Resource util.MapStr
+ // Attributes carries additional per-event information.
+ Attributes util.MapStr
+}
+
+// SeverityNumberFromText maps a severity text to the OTel SeverityNumber
+// scale, tolerating common variants ("WARN", "warning", "err", "fatal",
+// "critical", numeric strings). Unknown text maps to SeverityInfo.
+func SeverityNumberFromText(text string) int32 {
+ switch normalizeSeverity(text) {
+ case "":
+ return 0
+ case "trace", "verbose":
+ return SeverityTrace1
+ case "debug":
+ return SeverityDebug
+ case "info", "information", "notice":
+ return SeverityInfo
+ case "warn", "warning":
+ return SeverityWarn
+ case "err", "error":
+ return SeverityError
+ case "fatal", "critical", "crit", "emerg", "alert", "panic":
+ return SeverityFatal
+ default:
+ return SeverityInfo
+ }
+}
+
+func normalizeSeverity(text string) string {
+ t := text
+ for i := 0; i < len(t); i++ {
+ c := t[i]
+ if c >= 'A' && c <= 'Z' {
+ t = t[:i] + string(c-'A'+'a') + t[i+1:]
+ }
+ }
+ return t
+}
diff --git a/core/otel/otel_test.go b/core/otel/otel_test.go
new file mode 100644
index 000000000..a94b92f2e
--- /dev/null
+++ b/core/otel/otel_test.go
@@ -0,0 +1,120 @@
+// 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 .
+
+package otel
+
+import (
+ "testing"
+ "time"
+
+ "infini.sh/framework/core/event"
+ "infini.sh/framework/core/util"
+)
+
+func TestEventRoundTrip(t *testing.T) {
+ ts := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC)
+ e := &event.Event{
+ Timestamp: ts,
+ Fields: util.MapStr{
+ FieldLogLevel: "WARN",
+ FieldMessage: "disk usage 91%",
+ "device": "sda1",
+ },
+ Meta: util.MapStr{
+ MetaResourceKey: util.MapStr{ResourceHostName: "db-1"},
+ },
+ }
+
+ rec := FromEvent(e)
+ if rec.SeverityText != "WARN" {
+ t.Fatalf("severity = %v", rec.SeverityText)
+ }
+ if rec.Body != "disk usage 91%" {
+ t.Fatalf("body = %v", rec.Body)
+ }
+ if rec.Resource[ResourceHostName] != "db-1" {
+ t.Fatalf("resource = %v", rec.Resource)
+ }
+
+ back := rec.ToEvent()
+ if !back.Timestamp.Equal(ts) {
+ t.Fatalf("timestamp = %v", back.Timestamp)
+ }
+ if back.Fields[FieldMessage] != "disk usage 91%" || back.Fields["device"] != "sda1" {
+ t.Fatalf("fields = %v", back.Fields)
+ }
+ res, ok := back.Meta[MetaResourceKey].(util.MapStr)
+ if !ok || res[ResourceHostName] != "db-1" {
+ t.Fatalf("resource = %v", back.Meta[MetaResourceKey])
+ }
+}
+
+func TestEnvelopeRoundTrip(t *testing.T) {
+ ts := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC)
+ e := &event.Event{
+ Timestamp: ts,
+ Agent: &event.AgentMeta{AgentID: "agent-1"},
+ Fields: util.MapStr{FieldMessage: "hello", FieldLogLevel: "INFO"},
+ Meta: util.MapStr{MetaResourceKey: util.MapStr{ResourceServiceName: "svc"}},
+ }
+
+ data, err := EncodeEnvelope(e)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec, err := DecodeEnvelope(data)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec.Fields[FieldMessage] != "hello" {
+ t.Fatalf("message = %v", rec.Fields[FieldMessage])
+ }
+ if rec.Agent == nil || rec.Agent.AgentID != "agent-1" {
+ t.Fatalf("agent = %v", rec.Agent)
+ }
+ if !rec.Timestamp.Equal(ts) {
+ t.Fatalf("timestamp = %v, want %v", rec.Timestamp, ts)
+ }
+ res, ok := rec.Meta[MetaResourceKey].(util.MapStr)
+ if !ok || res[ResourceServiceName] != "svc" {
+ t.Fatalf("resource = %v", rec.Meta[MetaResourceKey])
+ }
+}
+
+func TestSeverityNumberFromText(t *testing.T) {
+ cases := map[string]int32{
+ "TRACE": SeverityTrace1,
+ "debug": SeverityDebug,
+ "INFO": SeverityInfo,
+ "WARN": SeverityWarn,
+ "warn": SeverityWarn,
+ "error": SeverityError,
+ "FATAL": SeverityFatal,
+ "": 0,
+ }
+ for text, want := range cases {
+ if got := SeverityNumberFromText(text); got != want {
+ t.Fatalf("SeverityNumberFromText(%q) = %d, want %d", text, got, want)
+ }
+ }
+}
diff --git a/core/pipeline/record.go b/core/pipeline/record.go
new file mode 100644
index 000000000..f352e1895
--- /dev/null
+++ b/core/pipeline/record.go
@@ -0,0 +1,72 @@
+// 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 .
+
+/* ©INFINI, All Rights Reserved.
+ * mail: contact#infini.ltd */
+
+package pipeline
+
+import "infini.sh/framework/core/event"
+
+// RecordContextKey is the Context key under which per-record processors
+// find the *event.Event of the record currently being processed.
+//
+// The for_each processor (see modules/pipeline) splits a batch of
+// queue.Message into records, decodes each one, stores it under this key
+// and runs its sub-chain; transform processors (e.g. the enterprise
+// dissect/field_standardize processors) read and mutate the record
+// in place through this convention.
+const RecordContextKey = "record"
+
+// CurrentRecord returns the record being processed in this context, if
+// the processor is running inside a per-record (for_each) sub-chain.
+func CurrentRecord(ctx *Context) (*event.Event, bool) {
+ if ctx == nil {
+ return nil, false
+ }
+ rec, ok := ctx.Get(RecordContextKey).(*event.Event)
+ if !ok || rec == nil {
+ return nil, false
+ }
+ return rec, true
+}
+
+// droppedMarker is stored in event.Private by MarkDropped; for_each
+// removes dropped records from the batch before it is forwarded.
+type droppedMarker struct{}
+
+// MarkDropped marks the current record to be dropped from the batch.
+func MarkDropped(rec *event.Event) {
+ if rec != nil {
+ rec.Private = droppedMarker{}
+ }
+}
+
+// IsDropped reports whether the record was marked for dropping.
+func IsDropped(rec *event.Event) bool {
+ if rec == nil {
+ return false
+ }
+ _, ok := rec.Private.(droppedMarker)
+ return ok
+}
diff --git a/core/pipeline/register.go b/core/pipeline/register.go
index d2b03c885..e1a28f949 100644
--- a/core/pipeline/register.go
+++ b/core/pipeline/register.go
@@ -258,6 +258,31 @@ func RegisterProcessorPlugin(name string, constructor ProcessorConstructor) {
}
}
+// processorMetadata carries the extracted config schema of processors
+// registered via RegisterProcessorPluginWithConfigMetadata, so that the
+// discovery API (see modules/pipeline) can render configuration forms.
+var processorMetadata = map[string]map[string]FilterProperty{}
+
+// RegisterProcessorPluginWithConfigMetadata registers a processor and
+// records the schema of its config struct for discovery.
+func RegisterProcessorPluginWithConfigMetadata(name string, constructor ProcessorConstructor, configStruct interface{}) {
+ RegisterProcessorPlugin(name, constructor)
+ processorMetadata[name] = ExtractFilterMetadata(configStruct)
+}
+
+// GetProcessorMetadata returns {name: {properties}} for every registered
+// processor.
+func GetProcessorMetadata() util.MapStr {
+ result := util.MapStr{}
+ for name := range registry.ProcessorConstructors() {
+ x, _ := processorMetadata[name]
+ result[name] = util.MapStr{
+ "properties": x,
+ }
+ }
+ return result
+}
+
func RegisterFilterPlugin(name string, constructor FilterConstructor) {
err := registry.RegisterFilter(name, constructor)
if err != nil {
diff --git a/core/shipper/shipper.go b/core/shipper/shipper.go
new file mode 100644
index 000000000..e7b956616
--- /dev/null
+++ b/core/shipper/shipper.go
@@ -0,0 +1,78 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+// Package shipper defines the direct-ship registry: producers that have
+// their own durable source (e.g. the agent's file collector, where the
+// file itself plus offset checkpoints provide durability) can bypass the
+// local queue and hand envelopes straight to a registered shipper.
+//
+// Implementations live outside this package and register a Factory via
+// Register, typically from an init(); products activate them with a
+// blank-import (the same pattern as pipeline processor registration).
+package shipper
+
+import (
+ "fmt"
+ "sort"
+ "sync"
+)
+
+// Shipper delivers batches of serialized log-event envelopes (the otel
+// envelope JSON also used at the queue boundary).
+//
+// Ship returns an error when the batch was NOT delivered; the caller
+// must then treat the events as undelivered -- for file sources that
+// means keeping the offset at the last committed position so the source
+// is re-read.
+type Shipper interface {
+ // Ship delivers one batch. The batch is only mutated by the caller
+ // after Ship returns.
+ Ship(batch [][]byte) error
+
+ // Close releases the shipper's resources.
+ Close() error
+}
+
+// Factory builds a Shipper from a configuration map (structure defined
+// by the implementation).
+type Factory func(cfg map[string]interface{}) (Shipper, error)
+
+var (
+ mu sync.RWMutex
+ factories = map[string]Factory{}
+)
+
+// Register adds a shipper factory under the given name. Intended for
+// package init(); a duplicate name panics to surface wiring mistakes.
+func Register(name string, f Factory) {
+ mu.Lock()
+ defer mu.Unlock()
+ if _, exists := factories[name]; exists {
+ panic(fmt.Sprintf("shipper factory with same name already exists: %v", name))
+ }
+ factories[name] = f
+}
+
+// Get builds the named shipper.
+func Get(name string, cfg map[string]interface{}) (Shipper, error) {
+ mu.RLock()
+ f, ok := factories[name]
+ mu.RUnlock()
+ if !ok {
+ return nil, fmt.Errorf("no shipper registered under name %q (available: %v)", name, Names())
+ }
+ return f(cfg)
+}
+
+// Names lists the registered shipper names (sorted, for errors and UI).
+func Names() []string {
+ mu.RLock()
+ defer mu.RUnlock()
+ names := make([]string, 0, len(factories))
+ for n := range factories {
+ names = append(names, n)
+ }
+ sort.Strings(names)
+ return names
+}
diff --git a/core/util/json_secrets.go b/core/util/json_secrets.go
new file mode 100644
index 000000000..d7cc85fb1
--- /dev/null
+++ b/core/util/json_secrets.go
@@ -0,0 +1,161 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package util
+
+import (
+ "encoding/json"
+ "reflect"
+ "strings"
+
+ "infini.sh/framework/lib/go-ucfg"
+)
+
+var secretStringType = reflect.TypeOf(ucfg.SecretString(""))
+
+// MustToJSONBytesWithSecrets marshals v to JSON with every embedded
+// ucfg.SecretString replaced by its resolved plaintext value.
+//
+// It exists for STORAGE round-trips (e.g. the sqlite ORM persists whole
+// objects as JSON blobs): SecretString.MarshalJSON intentionally emits
+// the "******" mask for display, which would destroy plain secrets on
+// persist and break every later read (test-connection uses the in-memory
+// value and passes, but anything re-loaded from storage authenticates
+// with the mask and gets a 401).
+//
+// Implementation note: the masked form is produced first (so all other
+// json.Marshaler semantics, tags and omitempty behavior are preserved
+// exactly), then the resolved secret values are patched into the decoded
+// generic tree along the reflecting walk of v. The input is never
+// mutated and no SecretString copy is re-marshaled (which would mask
+// again).
+func MustToJSONBytesWithSecrets(v interface{}) []byte {
+ masked, err := json.Marshal(v)
+ if err != nil {
+ panic(err)
+ }
+
+ var root interface{}
+ if err := json.Unmarshal(masked, &root); err != nil {
+ return masked // not an object/array graph (scalar, raw literal): as-is
+ }
+ patchSecrets(root, reflect.ValueOf(v))
+ out, err := json.Marshal(root)
+ if err != nil {
+ panic(err)
+ }
+ return out
+}
+
+// patchSecrets walks the marshaled node tree in lockstep with the source
+// value and replaces masked secret entries with their resolved values.
+func patchSecrets(node interface{}, v reflect.Value) {
+ m, ok := node.(map[string]interface{})
+ if !ok {
+ return
+ }
+ for v.Kind() == reflect.Ptr || v.Kind() == reflect.Interface {
+ if v.IsNil() {
+ return
+ }
+ v = v.Elem()
+ }
+ if v.Kind() != reflect.Struct {
+ return
+ }
+
+ t := v.Type()
+ for i := 0; i < t.NumField(); i++ {
+ f := t.Field(i)
+ if f.PkgPath != "" {
+ continue // unexported: not in JSON
+ }
+ name := jsonFieldName(f)
+ if name == "-" {
+ continue
+ }
+ fv := v.Field(i)
+
+ // Embedded unnamed struct (or pointer to one): fields are inlined
+ // into the parent object; patch against the same map.
+ if name == "" && f.Anonymous {
+ patchSecrets(m, fv)
+ continue
+ }
+ if name == "" {
+ name = f.Name
+ }
+ child, exists := m[name]
+ if !exists {
+ continue // omitted (omitempty / empty) or custom-marshaled away
+ }
+
+ if fv.Type() == secretStringType {
+ m[name] = ucfg.SecretString(fv.String()).Get()
+ continue
+ }
+
+ switch fv.Kind() {
+ case reflect.Ptr, reflect.Interface, reflect.Struct:
+ patchSecrets(child, fv)
+ case reflect.Slice, reflect.Array:
+ arr, ok := child.([]interface{})
+ if !ok {
+ continue
+ }
+ for idx := 0; idx < fv.Len() && idx < len(arr); idx++ {
+ patchSecretElement(arr, idx, fv.Index(idx))
+ }
+ case reflect.Map:
+ cm, ok := child.(map[string]interface{})
+ if !ok {
+ continue
+ }
+ patchSecretMap(cm, fv)
+ }
+ }
+}
+
+func patchSecretElement(arr []interface{}, idx int, ev reflect.Value) {
+ if ev.Type() == secretStringType {
+ arr[idx] = ucfg.SecretString(ev.String()).Get()
+ return
+ }
+ switch ev.Kind() {
+ case reflect.Ptr, reflect.Interface, reflect.Struct:
+ patchSecrets(arr[idx], ev)
+ }
+}
+
+func patchSecretMap(cm map[string]interface{}, mv reflect.Value) {
+ elemType := mv.Type().Elem()
+ iter := mv.MapRange()
+ for iter.Next() {
+ key := iter.Key().String()
+ val := iter.Value()
+ cur, exists := cm[key]
+ if !exists {
+ continue
+ }
+ if elemType == secretStringType {
+ cm[key] = ucfg.SecretString(val.String()).Get()
+ continue
+ }
+ switch val.Kind() {
+ case reflect.Ptr, reflect.Interface, reflect.Struct:
+ patchSecrets(cur, val)
+ }
+ }
+}
+
+// jsonFieldName returns the effective JSON object key of a struct field
+// (the json tag name, or "" when the tag is absent).
+func jsonFieldName(f reflect.StructField) string {
+ tag := f.Tag.Get("json")
+ if tag == "" {
+ return ""
+ }
+ parts := strings.Split(tag, ",")
+ return parts[0]
+}
diff --git a/core/util/json_secrets_test.go b/core/util/json_secrets_test.go
new file mode 100644
index 000000000..a8ac8b47a
--- /dev/null
+++ b/core/util/json_secrets_test.go
@@ -0,0 +1,115 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package util
+
+import (
+ "encoding/json"
+ "testing"
+
+ "infini.sh/framework/lib/go-ucfg"
+)
+
+type secretNested struct {
+ Password ucfg.SecretString `json:"password"`
+ Tokens []ucfg.SecretString `json:"tokens,omitempty"`
+}
+
+type secretHolder struct {
+ Name string `json:"name"`
+ APIKey ucfg.SecretString `json:"api_key"`
+ Nested *secretNested `json:"nested,omitempty"`
+ Headers map[string]ucfg.SecretString `json:"headers,omitempty"`
+ plain string //nolint:unused // unexported on purpose
+}
+
+// TestMustToJSONBytesWithSecrets_RevealsPlainText verifies storage marshal
+// emits the resolved secret values (SecretString.MarshalJSON would emit the
+// "******" mask and destroy them on persist).
+func TestMustToJSONBytesWithSecrets_RevealsPlainText(t *testing.T) {
+ in := &secretHolder{
+ Name: "cluster-1",
+ APIKey: ucfg.SecretString("tok-plain"),
+ Nested: &secretNested{
+ Password: ucfg.SecretString("pw-plain"),
+ Tokens: []ucfg.SecretString{ucfg.SecretString("t1"), ucfg.SecretString("t2")},
+ },
+ Headers: map[string]ucfg.SecretString{"auth": ucfg.SecretString("h1")},
+ }
+
+ out := MustToJSONBytesWithSecrets(in)
+
+ var m map[string]interface{}
+ if err := json.Unmarshal(out, &m); err != nil {
+ t.Fatalf("unmarshal output: %v", err)
+ }
+ if m["api_key"] != "tok-plain" {
+ t.Fatalf("api_key = %v, want tok-plain", m["api_key"])
+ }
+ nested, _ := m["nested"].(map[string]interface{})
+ if nested["password"] != "pw-plain" {
+ t.Fatalf("nested.password = %v, want pw-plain", nested["password"])
+ }
+ tokens, _ := nested["tokens"].([]interface{})
+ if len(tokens) != 2 || tokens[0] != "t1" || tokens[1] != "t2" {
+ t.Fatalf("nested.tokens = %v, want [t1 t2]", nested["tokens"])
+ }
+ headers, _ := m["headers"].(map[string]interface{})
+ if headers["auth"] != "h1" {
+ t.Fatalf("headers.auth = %v, want h1", m["headers"])
+ }
+}
+
+// TestMustToJSONBytesWithSecrets_RoundTrip verifies a struct persisted with
+// the helper decodes back with intact Get() values, while normal
+// json.Marshal keeps displaying the mask.
+func TestMustToJSONBytesWithSecrets_RoundTrip(t *testing.T) {
+ in := &secretHolder{APIKey: ucfg.SecretString("real-secret")}
+
+ stored := MustToJSONBytesWithSecrets(in)
+ var loaded secretHolder
+ if err := json.Unmarshal(stored, &loaded); err != nil {
+ t.Fatalf("unmarshal stored: %v", err)
+ }
+ if loaded.APIKey.Get() != "real-secret" {
+ t.Fatalf("round-trip Get() = %q, want real-secret", loaded.APIKey.Get())
+ }
+
+ // Display marshal stays masked.
+ display, err := json.Marshal(&loaded)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !jsonContains(display, ucfg.SecretShadowText) {
+ t.Fatalf("display marshal should keep the mask, got %s", display)
+ }
+}
+
+// TestMustToJSONBytesWithSecrets_DoesNotMutateInput verifies the input
+// object is untouched (secrets revealed only in the marshaling copy).
+func TestMustToJSONBytesWithSecrets_DoesNotMutateInput(t *testing.T) {
+ in := &secretHolder{APIKey: ucfg.SecretString("real-secret")}
+ _ = MustToJSONBytesWithSecrets(in)
+
+ display, err := json.Marshal(in)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !jsonContains(display, ucfg.SecretShadowText) {
+ t.Fatalf("input was mutated: display = %s", display)
+ }
+}
+
+func jsonContains(b []byte, sub string) bool {
+ return len(sub) == 0 || stringContains(string(b), sub)
+}
+
+func stringContains(s, sub string) bool {
+ for i := 0; i+len(sub) <= len(s); i++ {
+ if s[i:i+len(sub)] == sub {
+ return true
+ }
+ }
+ return false
+}
diff --git a/docs/content.en/docs/release-notes/_index.md b/docs/content.en/docs/release-notes/_index.md
index 1a519e01b..7de56debc 100644
--- a/docs/content.en/docs/release-notes/_index.md
+++ b/docs/content.en/docs/release-notes/_index.md
@@ -22,6 +22,11 @@ Information about release notes of INFINI Framework is provided here.
- perf(sqlite): set per-connection PRAGMAs (WAL, busy_timeout, foreign_keys) via the DSN and enable 256 MiB mmap_size — applies to every pooled connection (not just one) and serves the metadata store via memory-mapped I/O instead of pread syscalls
- perf(sqlite): auto-create expression indexes from elastic_mapping tags, now including nested object fields via dotted `$.parent.child` paths — the SQLite ORM indexes `json_extract(raw,'$.field')` for keyword/date/long/integer/boolean/double fields, turning full-table scans into B-tree lookups with zero query or model changes
- feat: search provider injection + easysearch cluster CRUD module #398
+- feat(otel): add the canonical log data model (`core/otel`) aligned with the OpenTelemetry Logs Data Model — lowercase snake_case field naming (`log_level`, `host_name`, `trace_id`, ...), `LogRecord ↔ event.Event` interconversion (Attributes→Fields, Resource→`Meta["resource"]`), and the queue envelope codec byte-compatible with the agent LogEvent JSON
+- feat(pipeline): add the `for_each` processor — splits a message batch into records, runs a sub-chain per record with the `RecordContextKey` convention, honors drop markers (`drop_event`), and re-encodes in place
+- feat(pipeline): add processor config metadata registration (`RegisterProcessorPluginWithConfigMetadata`) and the `GET /pipeline/processors` discovery endpoint, so pipeline designer UIs can render configuration forms
+- feat(otlp): add the OTLP/gRPC transport — resource-grouped `ExportLogsServiceRequest` codec plus the `otlp_export` processor that ships batches to any OTLP collector (e.g. the gateway's intake on `:4317`) and keeps batches unacknowledged on failure so the local queue redelivers; the codec and export now live in the enterprise plugin tree (`plugins/enterprise/otlp`) with a `core/shipper` registry for queue-free direct shipping
+- feat(shipper): add the `core/shipper` direct-ship registry — producers with their own durable source (e.g. file tailing with offset checkpoints) can bypass the local queue and hand envelope batches straight to a registered shipper; implementations register a `Shipper` factory by name (same pattern as processor registration), keeping the open-source core free of transport dependencies
- feat(orm): cross-backend aggregation engine — metrics (min/max/sum/avg/value_count), bucket (terms/histogram/range) and pipeline aggregations run unchanged on the Elasticsearch and SQLite backends
- feat(crud): extend the generated CRUD with migration hooks driven by CocoAI's hand-written handlers — `ExtraOptions` per action (login, CORS, sensitive-field masking), custom `IDParam`, `CtxDecorate` orm-context markers, `UpdateMode` partial/full/`?replace=` with `ProtectedFields` + `PrepareUpdate`, best-effort `PostCreate/PostUpdate/PostDelete`, `PostGet` refinement, and `PrepareSearch`/`PostSearch` for injected filters and per-hit mapping; the MCP tool now registers on GET _search only (no duplicate tool names); `SkipActions` keeps hand-written endpoints where the generator cannot express the semantics (upsert/replace-keeping-system-fields, cache-first fetch); full-object update mode now merges through a map so `ProtectedFields` are restored from the loaded record, and `PrepareUpdate` receives the raw body as the delta
- feat(orm): fluent `SetAggs` query-builder API and a refactored SQLite SQL builder backing it
@@ -31,6 +36,7 @@ Information about release notes of INFINI Framework is provided here.
- fix: register elasticsearch instance even when version probe fails #393
- fix: health api requires a system cluster that may not exist #393
- fix: pipeline task not visible right after creation #393
+- fix: restore the lost `core/api/websocket/reverse` protocol helpers (reverse channel manager/protocol, from 77a377ad) that agent builds depend on
### ✈️ Improvements
- refactor: add EventSink support to overall utilization collector #387
diff --git a/go.mod b/go.mod
index b9018eee9..40520e344 100644
--- a/go.mod
+++ b/go.mod
@@ -59,7 +59,7 @@ require (
go.uber.org/zap v1.27.1
golang.org/x/crypto v0.53.0
golang.org/x/net v0.56.0
- golang.org/x/oauth2 v0.29.0
+ golang.org/x/oauth2 v0.36.0
golang.org/x/sys v0.47.0
golang.org/x/text v0.38.0
golang.org/x/time v0.11.0
@@ -82,15 +82,18 @@ require (
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/dgraph-io/ristretto/v2 v2.2.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+ github.com/dlclark/regexp2/v2 v2.5.2 // indirect
+ github.com/dop251/goja v0.0.0-20260806115107-493f22071ef6 // indirect
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/ebitengine/purego v0.10.0 // indirect
github.com/fatih/color v1.18.0 // indirect
github.com/fxamacker/cbor/v2 v2.7.0 // indirect
github.com/go-asn1-ber/asn1-ber v1.5.8-0.20250403174932-29230038a667 // indirect
github.com/go-ini/ini v1.67.0 // indirect
- github.com/go-logr/logr v1.4.2 // indirect
+ github.com/go-logr/logr v1.4.3 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/go-ole/go-ole v1.2.6 // indirect
+ github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect
github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
github.com/goccy/go-json v0.10.5 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
@@ -98,13 +101,16 @@ require (
github.com/google/flatbuffers v25.2.10+incompatible // indirect
github.com/google/go-querystring v1.1.0 // indirect
github.com/google/gofuzz v1.2.0 // indirect
+ github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gookit/filter v1.2.3 // indirect
github.com/gookit/goutil v0.7.1 // indirect
github.com/gorilla/securecookie v1.1.2 // indirect
+ github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
github.com/invopop/jsonschema v0.13.0 // indirect
github.com/josharian/intern v1.0.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
+ github.com/julienschmidt/httprouter v1.3.0 // indirect
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
github.com/mattn/go-colorable v0.1.13 // indirect
@@ -118,6 +124,7 @@ require (
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/mschoch/smat v0.2.0 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
+ github.com/oschwald/maxminddb-golang v1.13.1 // indirect
github.com/pelletier/go-toml/v2 v2.2.3 // indirect
github.com/pierrec/lz4/v4 v4.1.22 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
@@ -135,23 +142,28 @@ require (
github.com/tklauser/go-sysconf v0.3.16 // indirect
github.com/tklauser/numcpus v0.11.0 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
+ github.com/valyala/fasthttp v1.52.0 // indirect
github.com/vmihailenco/msgpack v4.0.4+incompatible // indirect
github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect
github.com/x448/float16 v0.8.4 // indirect
github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
github.com/yusufpapurcu/wmi v1.2.4 // indirect
github.com/zeebo/blake3 v0.2.4 // indirect
- go.opentelemetry.io/auto/sdk v1.1.0 // indirect
- go.opentelemetry.io/otel v1.35.0 // indirect
- go.opentelemetry.io/otel/metric v1.35.0 // indirect
- go.opentelemetry.io/otel/trace v1.35.0 // indirect
+ go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+ go.opentelemetry.io/otel v1.43.0 // indirect
+ go.opentelemetry.io/otel/metric v1.43.0 // indirect
+ go.opentelemetry.io/otel/trace v1.43.0 // indirect
+ go.opentelemetry.io/proto/otlp v1.9.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap/exp v0.3.0 // indirect
golang.org/x/mod v0.37.0 // indirect
golang.org/x/sync v0.21.0 // indirect
golang.org/x/term v0.44.0 // indirect
google.golang.org/appengine v1.6.6 // indirect
- google.golang.org/protobuf v1.36.6 // indirect
+ google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478 // indirect
+ google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect
+ google.golang.org/grpc v1.82.0 // indirect
+ google.golang.org/protobuf v1.36.11 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
k8s.io/klog/v2 v2.130.1 // indirect
diff --git a/go.sum b/go.sum
index 7515cff7f..663467007 100644
--- a/go.sum
+++ b/go.sum
@@ -47,7 +47,11 @@ github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da h1:aIftn67I1fkbMa5
github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/dlclark/regexp2/v2 v2.5.2 h1:HAsucWRhsqcDzl6Ua9aR8JwYOTzrZyPrF0/FNxJVAI0=
+github.com/dlclark/regexp2/v2 v2.5.2/go.mod h1:avUrQvPaLz2DrFNHJF0taWAFFX2C1GMSSoeiqFjcBmU=
github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
+github.com/dop251/goja v0.0.0-20260806115107-493f22071ef6 h1:Oh2rRG1un7tLlC3/NJDzKppZ4CeZGkVFJCUOTRwLpfw=
+github.com/dop251/goja v0.0.0-20260806115107-493f22071ef6/go.mod h1:LiIEzozrcvNXorsG/3+ypGqdTUAqZryhzSsqi0oU/Qg=
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU=
@@ -83,6 +87,7 @@ github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
@@ -93,6 +98,8 @@ github.com/go-openapi/spec v0.0.0-20160808142527-6aced65f8501/go.mod h1:J8+jY1nA
github.com/go-openapi/swag v0.0.0-20160704191624-1d0bd113de87/go.mod h1:DXUve3Dpr1UfpPtxFw+EFuQ41HhCWZfha5jSVRG7C7I=
github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible h1:W1iEw64niKVGogNgBN3ePyLFfuisuzeidWPMPWmECqU=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
github.com/go-stack/stack v1.6.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
github.com/go-viper/mapstructure/v2 v2.2.1 h1:ZAaOCxANMuZx5RCeg0mBdEZk7DZasvvZIxtHqx8aGss=
github.com/go-viper/mapstructure/v2 v2.2.1/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
@@ -154,6 +161,8 @@ github.com/gorilla/sessions v1.4.0/go.mod h1:FLWm50oby91+hl7p/wRxDth9bWSuk0qVL2e
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/gregjones/httpcache v0.0.0-20170920190843-316c5e0ff04e/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs=
github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-version v1.7.0 h1:5tqGy27NaOTB8yJKUZELlFAS/LTKJkrmONwQKeRZfjY=
@@ -188,6 +197,8 @@ github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCV
github.com/json-iterator/go v1.1.8/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/julienschmidt/httprouter v1.3.0 h1:U0609e9tgbseu3rBINet9P48AI/D3oJs4dN7jwJOQ1U=
+github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 h1:iQTw/8FWTuc7uiaSepXwyf3o52HaUYcV+Tu66S3F5GA=
github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0/go.mod h1:1NbS8ALrpOvjt0rHPNLyCIeMtbizbir8U//inJ+zuB8=
github.com/kardianos/service v1.2.2 h1:ZvePhAHfvo0A7Mftk/tEzqEZ7Q4lgnR8sGz4xu1YX60=
@@ -272,6 +283,8 @@ github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGV
github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/onsi/gomega v1.35.1 h1:Cwbd75ZBPxFSuZ6T+rN/WCb/gOc6YgFBXLlZLhC7Ds4=
github.com/onsi/gomega v1.35.1/go.mod h1:PvZbdDc8J6XJEpDK4HCuRBm8a6Fzp9/DmhC9C7yFlog=
+github.com/oschwald/maxminddb-golang v1.13.1 h1:G3wwjdN9JmIK2o/ermkHM+98oX5fS+k5MbwsmL4MRQE=
+github.com/oschwald/maxminddb-golang v1.13.1/go.mod h1:K4pgV9N/GcK694KSTmVSDTODk4IsCNThNdTmnaBZ/F8=
github.com/pelletier/go-toml v1.0.1-0.20170904195809-1d6b12b7cb29/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/pelletier/go-toml/v2 v2.2.3 h1:YmeHyLY8mFWbdkNWwpr+qIL2bEqT0o95WSdkNHvL12M=
github.com/pelletier/go-toml/v2 v2.2.3/go.mod h1:MfCQTFTvCcUyyvvwm1+G6H/jORL20Xlb6rzQu9GuUkc=
@@ -353,6 +366,8 @@ github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQ
github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
+github.com/valyala/fasthttp v1.52.0 h1:wqBQpxH71XW0e2g+Og4dzQM8pk34aFYlA1Ga8db7gU0=
+github.com/valyala/fasthttp v1.52.0/go.mod h1:hf5C4QnVMkNXMspnsUlfM3WitlgYflyhHYoKol/szxQ=
github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVSA8=
github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
github.com/vmihailenco/msgpack v4.0.4+incompatible h1:dSLoQfGFAo3F6OoNhwUmLwVgaUXK79GlxNBwueZn0xI=
@@ -379,12 +394,18 @@ github.com/zeebo/sbloom v0.0.0-20151106181526-405c65bd9be0 h1:EAluI/s9FYrMnDGmyX
github.com/zeebo/sbloom v0.0.0-20151106181526-405c65bd9be0/go.mod h1:J0OA/x7vNUsWZ88/oJ0BPtebbGfjvSW1lA07GinZNLM=
go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
+go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/otel v1.35.0 h1:xKWKPxrxB6OtMCbmMY021CqC45J+3Onta9MqjhnusiQ=
go.opentelemetry.io/otel v1.35.0/go.mod h1:UEqy8Zp11hpkUrL73gSlELM0DupHoiq72dR+Zqel/+Y=
+go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0=
go.opentelemetry.io/otel/metric v1.35.0 h1:0znxYu2SNyuMSQT4Y9WDWej0VpcsxkuklLa4/siN90M=
go.opentelemetry.io/otel/metric v1.35.0/go.mod h1:nKVFgxBZ2fReX6IlyW28MgZojkoAkJGaE8CpgeAU3oE=
+go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY=
go.opentelemetry.io/otel/trace v1.35.0 h1:dPpEfJu1sDIqruz7BHFG3c7528f6ddfSWfFDVt/xgMs=
go.opentelemetry.io/otel/trace v1.35.0/go.mod h1:WUk7DtFp1Aw2MkvqGdwiXYDZZNvA/1J8o6xRXLrIkyc=
+go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
+go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A=
+go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
@@ -416,6 +437,8 @@ golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
golang.org/x/oauth2 v0.0.0-20170912212905-13449ad91cb2/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
golang.org/x/oauth2 v0.29.0 h1:WdYw2tdTK1S8olAzWHdgeqfy+Mtm9XNhv/xJsY65d98=
golang.org/x/oauth2 v0.29.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8=
+golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
+golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
golang.org/x/sync v0.0.0-20170517211232-f52d1811a629/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -463,9 +486,18 @@ google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCID
google.golang.org/appengine v1.6.6 h1:lMO5rYAqUxkmaj76jAkRUvt5JZgFymx/+Q5Mzfivuhc=
google.golang.org/appengine v1.6.6/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
google.golang.org/genproto v0.0.0-20170918111702-1e559d0a00ee/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
+google.golang.org/genproto v0.0.0-20241118233622-e639e219e697 h1:ToEetK57OidYuqD4Q5w+vfEnPvPpuTwedCNVohYJfNk=
+google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478 h1:yQugLulqltosq0B/f8l4w9VryjV+N/5gcW0jQ3N8Qec=
+google.golang.org/genproto/googleapis/api v0.0.0-20260414002931-afd174a4e478/go.mod h1:C6ADNqOxbgdUUeRTU+LCHDPB9ttAMCTff6auwCVa4uc=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
google.golang.org/grpc v1.2.1-0.20170921194603-d4b75ebd4f9f/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
+google.golang.org/grpc v1.82.0 h1:vguDnZUPjE26w09A63VoxZPnvPjB5Riyc0mkXPFmAIU=
+google.golang.org/grpc v1.82.0/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
+google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc h1:2gGKlE2+asNV9m7xrywl36YYNnBG5ZQ0r/BOOxqPpmk=
gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc/go.mod h1:m7x9LTH6d71AHyAX77c9yqWCCa3UKHcVEj9y7hAtKDk=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/modules/configs/client/client.go b/modules/configs/client/client.go
index f863434b2..34546db56 100644
--- a/modules/configs/client/client.go
+++ b/modules/configs/client/client.go
@@ -53,6 +53,29 @@ import (
const bucketName = "instance_registered"
const configRegisterEnvKey = "CONFIG_MANAGED_SUCCESS"
+// managerTokenBucket persists the per-instance manager token minted at
+// registration (or exchange) so it survives restarts.
+const managerTokenBucket = "managed_manager_token"
+const managerTokenKey = "manager_token"
+
+// setManagerToken stores the instance-scoped manager token locally.
+func setManagerToken(token string) {
+ if token == "" {
+ return
+ }
+ _ = kv.AddValue(managerTokenBucket, []byte(managerTokenKey), []byte(token))
+}
+
+// getManagerToken returns the stored per-instance token ("" when the
+// manager has not minted one yet — e.g. pre-upgrade servers).
+func getManagerToken() string {
+ v, err := kv.GetValue(managerTokenBucket, []byte(managerTokenKey))
+ if err != nil || len(v) == 0 {
+ return ""
+ }
+ return string(v)
+}
+
func ConnectToManager() error {
if !global.Env().SystemConfig.Configs.Managed {
@@ -92,6 +115,16 @@ func ConnectToManager() error {
panic(err)
}
global.Register(configRegisterEnvKey, true)
+
+ // Capture the per-instance manager token the server minted (a
+ // no-token body means an older server — keep whatever we have).
+ var regResp struct {
+ ManagerToken string `json:"manager_token"`
+ }
+ if util.FromJSONBytes(res.Body, ®Resp) == nil && regResp.ManagerToken != "" {
+ setManagerToken(regResp.ManagerToken)
+ log.Info("received per-instance manager token from config manager")
+ }
}
} else {
log.Error("failed to register to config manager,", err, ",", server)
@@ -103,7 +136,13 @@ func submitRequestToManager(req *util.Request) (string, *util.Result, error) {
var err error
var res *util.Result
cfg := global.Env().SystemConfig.Configs
- if cfg.ManagerConfig.BasicAuth.Username != "" {
+ // Auth precedence: per-instance token (minted at registration) → static
+ // configured token (configs.manager.token) → legacy BasicAuth.
+ if t := getManagerToken(); t != "" {
+ req.AddHeader("Authorization", "Bearer "+t)
+ } else if tk := cfg.ManagerConfig.Token.Get(); tk != "" {
+ req.AddHeader("Authorization", "Bearer "+tk)
+ } else if cfg.ManagerConfig.BasicAuth.Username != "" {
req.SetBasicAuth(cfg.ManagerConfig.BasicAuth.Username, cfg.ManagerConfig.BasicAuth.Password.Get())
}
for _, server := range cfg.Servers {
@@ -170,6 +209,7 @@ func ListenConfigChanges() error {
obj := common.ConfigSyncResponse{}
err := util.FromJSONBytes(res.Body, &obj)
if err != nil {
+ log.Debugf("failed to parse config sync response, %v, %v", err, string(res.Body))
panic(err)
}
diff --git a/modules/configs/server/instance_token.go b/modules/configs/server/instance_token.go
new file mode 100644
index 000000000..655013e6e
--- /dev/null
+++ b/modules/configs/server/instance_token.go
@@ -0,0 +1,174 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "crypto/rand"
+ "crypto/sha256"
+ "crypto/subtle"
+ "encoding/hex"
+ "net/http"
+ "time"
+
+ httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/core/elastic"
+ "infini.sh/framework/core/orm"
+ "infini.sh/framework/core/util"
+)
+
+// ──────────────────────────────────────────────────────────────────────────
+// Per-instance manager tokens (console_design: managed-config-security.md §4).
+//
+// Lifecycle (portable adaptation of Console's managed token flow — the
+// final-state design from pr/framework-managed-token-flow-20260524 as
+// evolved on console_framework, simplified for the embedded server):
+//
+// 1. BOOTSTRAP: a fresh instance registers using one of the statically
+// configured tokens (configs.server.auth.tokens). The server mints an
+// instance-scoped token and returns it in the register response.
+// 2. STEADY STATE: the instance presents its per-instance token
+// (Authorization: Bearer) on every /configs/_sync.
+// 3. ROTATION: POST /instance/_exchange_token with the current token
+// mints a replacement; the previous token stays valid for 1h
+// (grace window for in-flight syncs).
+//
+// Storage: only the SHA-256 hash of each token is persisted (tokens are
+// bearer-equivalent secrets; hashing at rest means a database leak does not
+// leak credentials). Comparison is constant-time.
+// ──────────────────────────────────────────────────────────────────────────
+
+// rotationGrace is how long a superseded token remains valid after exchange.
+const rotationGrace = time.Hour
+
+// InstanceToken is the per-instance manager credential record.
+type InstanceToken struct {
+ orm.ORMObjectBase
+
+ InstanceID string `json:"instance_id" elastic_mapping:"instance_id:{type:keyword}"`
+ // TokenHash is sha256(token) of the current token.
+ TokenHash string `json:"token_hash" elastic_mapping:"token_hash:{type:keyword}"`
+ // PreviousHash is sha256(token) of the superseded token (rotation grace).
+ PreviousHash string `json:"previous_hash,omitempty" elastic_mapping:"previous_hash:{type:keyword}"`
+ // RotatedAt is when the current token was minted (grace window anchor).
+ RotatedAt time.Time `json:"rotated_at" elastic_mapping:"rotated_at:{type:date}"`
+}
+
+// MintInstanceToken creates (or rotates) the token record for an instance
+// and returns the plaintext token — the ONLY time it exists in the clear
+// outside the client's memory.
+func MintInstanceToken(ctx *orm.Context, instanceID string) (string, error) {
+ raw := make([]byte, 32)
+ if _, err := rand.Read(raw); err != nil {
+ return "", err
+ }
+ token := hex.EncodeToString(raw)
+
+ rec := loadInstanceToken(ctx, instanceID)
+ if rec == nil {
+ rec = &InstanceToken{InstanceID: instanceID}
+ rec.ID = util.GetUUID()
+ } else {
+ rec.PreviousHash = rec.TokenHash
+ }
+ rec.TokenHash = hashToken(token)
+ rec.RotatedAt = time.Now().UTC()
+ orm.WithModel(ctx, rec)
+ if err := orm.Save(ctx, rec); err != nil {
+ return "", err
+ }
+ return token, nil
+}
+
+// ValidateInstanceToken checks a presented token against the instance's
+// current record: current token always passes; the previous token passes
+// only inside the rotation grace window. Constant-time per comparison.
+func ValidateInstanceToken(ctx *orm.Context, instanceID, token string) bool {
+ if token == "" {
+ return false
+ }
+ rec := loadInstanceToken(ctx, instanceID)
+ if rec == nil {
+ return false
+ }
+ got := hashToken(token)
+ match := subtle.ConstantTimeCompare([]byte(got), []byte(rec.TokenHash))
+ if match == 1 {
+ return true
+ }
+ if rec.PreviousHash != "" && time.Since(rec.RotatedAt) < rotationGrace {
+ return subtle.ConstantTimeCompare([]byte(got), []byte(rec.PreviousHash)) == 1
+ }
+ return false
+}
+
+func loadInstanceToken(ctx *orm.Context, instanceID string) *InstanceToken {
+ orm.WithModel(ctx, &InstanceToken{})
+ qb := orm.NewQuery().
+ Filter(orm.TermQuery("instance_id", instanceID)).
+ Size(1)
+ res, err := orm.SearchV2(ctx, qb)
+ if err != nil || res == nil {
+ return nil
+ }
+ tokens, _, _ := elastic.DecodeHits[InstanceToken](res)
+ if len(tokens) == 0 {
+ return nil
+ }
+ return &tokens[0]
+}
+
+func hashToken(token string) string {
+ sum := sha256.Sum256([]byte(token))
+ return hex.EncodeToString(sum[:])
+}
+
+// ──────────────────────────────────────────────────────────────────────────
+// HTTP: token exchange
+// ──────────────────────────────────────────────────────────────────────────
+
+// exchangeTokenHandler — POST /instance/_exchange_token
+//
+// Body: {"instance_id": "..."} authenticated by the CURRENT token (Bearer).
+// Response: {"manager_token": "", "grace_seconds": 3600}
+func (h *APIHandler) exchangeTokenHandler(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ var body struct {
+ InstanceID string `json:"instance_id"`
+ }
+ if err := h.DecodeJSON(req, &body); err != nil || body.InstanceID == "" {
+ h.WriteError(w, "instance_id is required", http.StatusBadRequest)
+ return
+ }
+
+ presented := extractBearerToken(req)
+ if presented == "" {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+
+ ctx := orm.NewContextWithParent(req.Context()).DirectAccess()
+
+ // The caller must hold the instance's CURRENT token (or a static token —
+ // static holders are bootstrap admins and may also rotate).
+ ok := ValidateInstanceToken(ctx, body.InstanceID, presented)
+ if !ok {
+ ok = validateStaticToken(presented)
+ }
+ if !ok {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+
+ token, err := MintInstanceToken(ctx, body.InstanceID)
+ if err != nil {
+ h.WriteError(w, "mint token: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ h.WriteJSON(w, util.MapStr{
+ "manager_token": token,
+ "grace_seconds": int(rotationGrace.Seconds()),
+ }, http.StatusOK)
+}
diff --git a/modules/configs/server/server.go b/modules/configs/server/server.go
new file mode 100644
index 000000000..5f780e05c
--- /dev/null
+++ b/modules/configs/server/server.go
@@ -0,0 +1,450 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+// Package server implements the SERVER side of the standard managed-config
+// protocol: the counterpart of modules/configs/client that every framework
+// process runs when `configs.managed: true`.
+//
+// Routes (protocol-compatible with Console's managed plugin; Console keeps
+// its richer implementation — token exchange, script hooks, websocket
+// proxy — on its own product surface, this one carries the portable core
+// any product can embed):
+//
+// POST /instance/_register self-description registration (upsert)
+// POST /configs/_sync heartbeat + config diff delivery
+//
+// Sync semantics (mirrors the Console contract):
+// - the client posts its current managed configs + a hash of them
+// - the server replies {changed, configs:{created,updated,deleted}}
+// - unchanged content is skipped by version comparison; a config the
+// client marked Managed=false is never touched
+// - every sync refreshes the instance record (heartbeat via labels)
+package server
+
+import (
+ "crypto/subtle"
+ "encoding/json"
+ "io"
+ "net/http"
+ "strconv"
+ "strings"
+ "time"
+
+ "infini.sh/framework/core/api"
+ httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/core/elastic"
+ "infini.sh/framework/core/env"
+ log "infini.sh/framework/core/log"
+ "infini.sh/framework/core/model"
+ "infini.sh/framework/core/orm"
+ "infini.sh/framework/core/util"
+ "infini.sh/framework/lib/go-ucfg"
+ "infini.sh/framework/modules/configs/common"
+)
+
+// Config holds the server-side settings (configs.server.* in YAML).
+// Disabled by default; products opt in via Setup().
+type Config struct {
+ Enabled bool `config:"enabled"`
+
+ // Auth gates both protocol routes (_register/_sync) with the standard
+ // Bearer token mechanism: clients present
+ // Authorization: Bearer (or X-API-Token: )
+ // and the server constant-time-compares against the configured token
+ // list. Multiple tokens allow zero-downtime rotation (add the new
+ // token, roll the clients, remove the old one). Deployments MUST
+ // configure auth.tokens in production: without it any host that
+ // reaches the port can register instances and pull every assigned
+ // config. When unset the server logs a warning and runs in open dev
+ // mode. The framework client sends its token automatically via
+ // configs.manager.token.
+ Auth struct {
+ Tokens []ucfg.SecretString `config:"tokens"`
+ } `config:"auth"`
+}
+
+// ManagedConfig is one config file assigned to an instance (or "*", all
+// instances). Version bumps on every content change; clients apply
+// Created/Updated diffs by version comparison.
+type ManagedConfig struct {
+ orm.ORMObjectBase
+
+ InstanceID string `json:"instance_id" elastic_mapping:"instance_id:{type:keyword}"` // target instance id, or "*" for all
+ Name string `json:"name" elastic_mapping:"name:{type:keyword}"` // config file name, e.g. pipeline.yml
+ Location string `json:"location,omitempty" elastic_mapping:"location:{type:keyword}"`
+ Content string `json:"content,omitempty" elastic_mapping:"content:{type:text}"`
+ Version int64 `json:"version" elastic_mapping:"version:{type:long}"`
+ Readonly bool `json:"readonly,omitempty" elastic_mapping:"readonly:{type:boolean}"`
+}
+
+// Label keys for heartbeat state on the instance record (model.Instance
+// has no dedicated online-state fields; labels keep the wire type intact).
+const (
+ LabelLastSyncAt = "managed_last_sync_at"
+ LabelRegistered = "managed_registered"
+)
+
+// AllInstancesID assigns a ManagedConfig to every syncing instance.
+const AllInstancesID = "*"
+
+// instanceTokenExchangeAPI rotates an instance's manager token.
+const instanceTokenExchangeAPI = "/instance/_exchange_token"
+
+type APIHandler struct {
+ api.Handler
+}
+
+var handler = &APIHandler{}
+
+// Setup registers the ORM schemas and the protocol routes. Call once from
+// the product's module setup (e.g. logpilot's init). No-op when
+// configs.server.enabled is false in the product config.
+func Setup() {
+ cfg := Config{Enabled: true}
+ exists, err := env.ParseConfig("configs.server", &cfg)
+ if err != nil {
+ panic(err)
+ }
+ if exists && !cfg.Enabled {
+ log.Debug("configs server disabled by configuration")
+ return
+ }
+
+ orm.MustRegisterSchemaWithIndexName(model.Instance{}, "instance")
+ orm.MustRegisterSchemaWithIndexName(ManagedConfig{}, "managed-configs")
+ orm.MustRegisterSchemaWithIndexName(InstanceToken{}, "instance-tokens")
+
+ gate := newTokenGate(cfg.Auth.Tokens)
+ registerGate := gate
+ if len(cfg.Auth.Tokens) == 0 {
+ // Open mode is warned about at startup; the per-instance flow below
+ // still mints tokens so closing the server later does not strand
+ // instances that already registered.
+ registerGate = func(next httprouter.Handle) httprouter.Handle { return next }
+ }
+ api.HandleAPIMethod(api.POST, common.REGISTER_API, registerGate(handler.registerInstance))
+ api.HandleAPIMethod(api.POST, common.SYNC_API, gate(handler.syncConfigs))
+ api.HandleAPIMethod(api.POST, instanceTokenExchangeAPI, gate(handler.exchangeTokenHandler))
+
+ api.HandleUIMethod(api.POST, common.REGISTER_API, registerGate(handler.registerInstance))
+ api.HandleUIMethod(api.POST, common.SYNC_API, gate(handler.syncConfigs))
+ api.HandleUIMethod(api.POST, instanceTokenExchangeAPI, gate(handler.exchangeTokenHandler))
+
+ if len(cfg.Auth.Tokens) > 0 {
+ log.Infof("configs server ready: %s + %s (bearer token auth, %d token(s) accepted)", common.REGISTER_API, common.SYNC_API, len(cfg.Auth.Tokens))
+ } else {
+ log.Warnf("configs server ready in OPEN mode (no configs.server.auth.tokens configured) - " +
+ "any host reaching this port can register instances and pull assigned configs; configure auth.tokens in production")
+ }
+}
+
+// staticTokens holds the configured bootstrap/admin tokens (set in Setup).
+var staticTokens []string
+
+// validateStaticToken constant-time checks against the configured static
+// tokens (bootstrap admission + admin fallback).
+func validateStaticToken(token string) bool {
+ if token == "" || len(staticTokens) == 0 {
+ return false
+ }
+ matched := 0
+ for _, want := range staticTokens {
+ matched |= subtle.ConstantTimeCompare([]byte(token), []byte(want))
+ }
+ return matched == 1
+}
+
+// extractBearerToken reads the access token from the standard
+// Authorization: Bearer header, falling back to X-API-Token (the framework's
+// conventional token header).
+func extractBearerToken(req *http.Request) string {
+ if h := req.Header.Get("Authorization"); strings.HasPrefix(h, "Bearer ") {
+ return strings.TrimSpace(h[len("Bearer "):])
+ }
+ return strings.TrimSpace(req.Header.Get("X-API-Token"))
+}
+
+// newTokenGate wraps a protocol handler with constant-time Bearer-token
+// validation against the accepted token list. No tokens configured = open
+// pass-through (dev mode, loudly warned at startup).
+func newTokenGate(tokens []ucfg.SecretString) func(httprouter.Handle) httprouter.Handle {
+ if len(tokens) == 0 {
+ return func(next httprouter.Handle) httprouter.Handle { return next }
+ }
+ wants := make([]string, 0, len(tokens))
+ for _, t := range tokens {
+ if v := t.Get(); v != "" {
+ wants = append(wants, v)
+ }
+ }
+ staticTokens = wants
+ return func(next httprouter.Handle) httprouter.Handle {
+ return func(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+ got := extractBearerToken(req)
+ if got == "" {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ // Constant-time compare against EVERY accepted token (all of
+ // them, so the response time does not reveal which position
+ // matched); any match passes.
+ matched := 0
+ for _, want := range wants {
+ matched |= subtle.ConstantTimeCompare([]byte(got), []byte(want))
+ }
+ if matched != 1 {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ next(w, req, ps)
+ }
+ }
+}
+
+// ──────────────────────────────────────────────────────────────────────────
+// POST /instance/_register
+// ──────────────────────────────────────────────────────────────────────────
+
+// registerBody decodes both the wrapped form ({client:{...}}) and the
+// legacy plain model.Instance the framework client sends.
+type registerBody struct {
+ Client model.Instance `json:"client"`
+}
+
+func (h *APIHandler) registerInstance(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ var wrapped registerBody
+ if err := h.DecodeJSON(req, &wrapped); err != nil {
+ h.WriteError(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ instance := wrapped.Client
+ if instance.ID == "" {
+ // legacy plain payload
+ var plain model.Instance
+ body, readErr := readBody(req)
+ if readErr != nil {
+ h.WriteError(w, readErr.Error(), http.StatusBadRequest)
+ return
+ }
+ if err := util.FromJSONBytes(body, &plain); err != nil {
+ h.WriteError(w, "invalid registration payload: "+err.Error(), http.StatusBadRequest)
+ return
+ }
+ instance = plain
+ }
+ if instance.ID == "" {
+ h.WriteError(w, "instance id is required", http.StatusBadRequest)
+ return
+ }
+
+ // An EXISTING instance re-registering must prove identity with its own
+ // token (a static token also qualifies — bootstrap admin). A fresh
+ // instance is authenticated by the static gate already.
+ ormCtx := orm.NewContextWithParent(req.Context()).DirectAccess()
+ existingToken := loadInstanceToken(ormCtx, instance.ID)
+ if existingToken != nil {
+ presented := extractBearerToken(req)
+ if !ValidateInstanceToken(ormCtx, instance.ID, presented) && !validateStaticToken(presented) {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized: instance token required to re-register", http.StatusUnauthorized)
+ return
+ }
+ }
+
+ created, err := upsertInstance(&instance)
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+
+ // Mint (or rotate on re-register) the per-instance token; the response
+ // is the only place the plaintext ever appears.
+ token, err := MintInstanceToken(ormCtx, instance.ID)
+ if err != nil {
+ h.WriteError(w, "mint instance token: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+
+ resp := util.MapStr{"id": instance.ID, "manager_token": token}
+ if !created {
+ resp["exists"] = true // the framework client treats "exists" as success
+ } else {
+ resp["created"] = true
+ }
+ h.WriteJSON(w, resp, http.StatusOK)
+}
+
+// upsertInstance persists/refreshes the registration. Returns true when
+// the instance was newly created.
+func upsertInstance(instance *model.Instance) (bool, error) {
+ ctx := orm.NewContext().DirectAccess()
+ existing := model.Instance{}
+ existing.ID = instance.ID
+ exists, err := orm.GetV2(ctx, &existing)
+ if err != nil {
+ return false, err
+ }
+
+ now := strconv.FormatInt(time.Now().UnixMilli(), 10)
+ if instance.Labels == nil {
+ instance.Labels = map[string]string{}
+ }
+ instance.Labels[LabelRegistered] = now
+ instance.Labels[LabelLastSyncAt] = now
+
+ if exists {
+ // keep server-side timestamps; refresh the self-description
+ created := existing.Created
+ instanceCopy := *instance
+ instanceCopy.Created = created
+ return false, orm.Save(ctx, &instanceCopy)
+ }
+ created := time.Now().UTC()
+ instance.Created = &created
+ return true, orm.Save(ctx, instance)
+}
+
+// ──────────────────────────────────────────────────────────────────────────
+// POST /configs/_sync
+// ──────────────────────────────────────────────────────────────────────────
+
+func (h *APIHandler) syncConfigs(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ var obj common.ConfigSyncRequest
+ if err := h.DecodeJSON(req, &obj); err != nil {
+ h.WriteError(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ if obj.Client.ID == "" {
+ h.WriteError(w, "client.id is required", http.StatusBadRequest)
+ return
+ }
+
+ // Authentication: the static gate has already accepted the caller, but
+ // instances that hold per-instance tokens must be checked against them —
+ // a revoked static token must not keep a registered instance alive, and
+ // conversely a valid instance token must pass even if statics rotate.
+ if presentToken := loadInstanceToken(orm.NewContext().DirectAccess(), obj.Client.ID); presentToken != nil {
+ if !ValidateInstanceToken(orm.NewContext().DirectAccess(), obj.Client.ID, extractBearerToken(req)) {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ }
+
+ // Heartbeat: refresh the instance record on every sync.
+ if _, err := upsertInstance(&obj.Client); err != nil {
+ log.Debugf("configs server: heartbeat upsert failed for %s: %v", obj.Client.ID, err)
+ }
+
+ assigned := loadAssignedConfigs(obj.Client.ID)
+
+ // Fast path: identical hash and no forced sync → nothing changed.
+ serverHash := ConfigsHash(assigned)
+ if !obj.ForceSync && obj.Hash != "" && obj.Hash == serverHash {
+ h.WriteJSON(w, common.ConfigSyncResponse{Changed: false}, http.StatusOK)
+ return
+ }
+
+ resp := diffConfigs(assigned, obj.Configs.Configs)
+ h.WriteJSON(w, resp, http.StatusOK)
+}
+
+// loadAssignedConfigs returns the server-side config files assigned to the
+// instance (its own + the "*" catch-all), newest version per name.
+func loadAssignedConfigs(instanceID string) []common.ConfigFile {
+ ctx := orm.NewContext().DirectAccess()
+ orm.WithModel(ctx, &ManagedConfig{})
+
+ qb := orm.NewQuery().
+ Filter(orm.TermQuery("instance_id", instanceID)).
+ Filter(orm.TermQuery("instance_id", AllInstancesID)).
+ Size(1000)
+ res, err := orm.SearchV2(ctx, qb)
+ if err != nil || res == nil {
+ return nil
+ }
+ stored, _, _ := decodeManagedConfigs(res)
+
+ out := make([]common.ConfigFile, 0, len(stored))
+ for _, mc := range stored {
+ out = append(out, common.ConfigFile{
+ Name: mc.Name,
+ Location: mc.Location,
+ Content: mc.Content,
+ Version: mc.Version,
+ Managed: true,
+ Hash: util.MD5digest(mc.Content),
+ Size: int64(len(mc.Content)),
+ Updated: time.Now().UnixMilli(),
+ })
+ }
+ return out
+}
+
+// decodeManagedConfigs decodes search hits via the shared elastic mapper.
+func decodeManagedConfigs(res *orm.SearchResult) ([]ManagedConfig, int64, error) {
+ return elastic.DecodeHits[ManagedConfig](res)
+}
+
+// diffConfigs builds the protocol response: created (server-only),
+// updated (version newer than the client's), deleted (client-only).
+// Configs the client marked Managed=false are never touched.
+func diffConfigs(assigned []common.ConfigFile, clientConfigs map[string]common.ConfigFile) common.ConfigSyncResponse {
+ resp := common.ConfigSyncResponse{}
+ resp.Configs.CreatedConfigs = map[string]common.ConfigFile{}
+ resp.Configs.UpdatedConfigs = map[string]common.ConfigFile{}
+ resp.Configs.DeletedConfigs = map[string]common.ConfigFile{}
+
+ serverMap := map[string]common.ConfigFile{}
+ for _, c := range assigned {
+ serverMap[c.Name] = c
+ }
+
+ for name, sc := range serverMap {
+ cc, ok := clientConfigs[name]
+ if !ok {
+ resp.Configs.CreatedConfigs[name] = sc
+ continue
+ }
+ if !cc.Managed {
+ continue // client opted this file out of management
+ }
+ if sc.Version > cc.Version {
+ resp.Configs.UpdatedConfigs[name] = sc
+ }
+ }
+ for name, cc := range clientConfigs {
+ if _, ok := serverMap[name]; !ok {
+ if !cc.Managed {
+ continue
+ }
+ resp.Configs.DeletedConfigs[name] = cc
+ }
+ }
+
+ resp.Changed = len(resp.Configs.CreatedConfigs) > 0 ||
+ len(resp.Configs.UpdatedConfigs) > 0 ||
+ len(resp.Configs.DeletedConfigs) > 0
+ return resp
+}
+
+// ConfigsHash mirrors the framework client's hash: MD5 of the JSON of the
+// assigned config list, so both sides compare identical digests.
+func ConfigsHash(files []common.ConfigFile) string {
+ if len(files) == 0 {
+ return ""
+ }
+ b, err := json.Marshal(files)
+ if err != nil {
+ return ""
+ }
+ return util.MD5digest(string(b))
+}
+
+func readBody(req *http.Request) ([]byte, error) {
+ defer func() { _ = req.Body.Close() }()
+ return io.ReadAll(req.Body)
+}
diff --git a/modules/configs/server/server_test.go b/modules/configs/server/server_test.go
new file mode 100644
index 000000000..9948cd97e
--- /dev/null
+++ b/modules/configs/server/server_test.go
@@ -0,0 +1,227 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/lib/go-ucfg"
+ "infini.sh/framework/modules/configs/common"
+)
+
+func cfg(name string, version int64) common.ConfigFile {
+ return common.ConfigFile{Name: name, Content: "content-of-" + name, Version: version, Managed: true}
+}
+
+func TestDiffConfigs_AllStates(t *testing.T) {
+ assigned := []common.ConfigFile{cfg("a.yml", 1), cfg("b.yml", 2)}
+
+ t.Run("fresh client gets everything created", func(t *testing.T) {
+ resp := diffConfigs(assigned, nil)
+ if !resp.Changed {
+ t.Fatal("expected changed=true")
+ }
+ if len(resp.Configs.CreatedConfigs) != 2 {
+ t.Fatalf("created = %d, want 2", len(resp.Configs.CreatedConfigs))
+ }
+ if len(resp.Configs.UpdatedConfigs) != 0 || len(resp.Configs.DeletedConfigs) != 0 {
+ t.Fatal("no updates/deletes expected")
+ }
+ })
+
+ t.Run("same versions → no change", func(t *testing.T) {
+ client := map[string]common.ConfigFile{"a.yml": cfg("a.yml", 1), "b.yml": cfg("b.yml", 2)}
+ resp := diffConfigs(assigned, client)
+ if resp.Changed {
+ t.Fatalf("expected no change, got %+v", resp.Configs)
+ }
+ })
+
+ t.Run("server version bump → updated", func(t *testing.T) {
+ bumped := []common.ConfigFile{cfg("a.yml", 1), cfg("b.yml", 3)}
+ client := map[string]common.ConfigFile{"a.yml": cfg("a.yml", 1), "b.yml": cfg("b.yml", 2)}
+ resp := diffConfigs(bumped, client)
+ if !resp.Changed {
+ t.Fatal("expected changed")
+ }
+ if len(resp.Configs.UpdatedConfigs) != 1 || resp.Configs.UpdatedConfigs["b.yml"].Version != 3 {
+ t.Fatalf("updated = %+v", resp.Configs.UpdatedConfigs)
+ }
+ })
+
+ t.Run("client-only config → deleted", func(t *testing.T) {
+ client := map[string]common.ConfigFile{
+ "a.yml": cfg("a.yml", 1),
+ "b.yml": cfg("b.yml", 2),
+ "gone.yml": cfg("gone.yml", 1),
+ }
+ resp := diffConfigs(assigned, client)
+ if !resp.Changed {
+ t.Fatal("expected changed")
+ }
+ if len(resp.Configs.DeletedConfigs) != 1 || resp.Configs.DeletedConfigs["gone.yml"].Name != "gone.yml" {
+ t.Fatalf("deleted = %+v", resp.Configs.DeletedConfigs)
+ }
+ })
+
+ t.Run("client opts out via Managed=false → untouched", func(t *testing.T) {
+ localOnly := cfg("local.yml", 1)
+ localOnly.Managed = false
+ client := map[string]common.ConfigFile{
+ "a.yml": cfg("a.yml", 1),
+ "b.yml": cfg("b.yml", 2),
+ "local.yml": localOnly,
+ }
+ resp := diffConfigs(assigned, client)
+ if resp.Changed {
+ t.Fatal("unmanaged local config must not trigger deletion")
+ }
+
+ // server-side version bump on a config the client holds unmanaged:
+ // also skipped
+ unmanagedA := cfg("a.yml", 1)
+ unmanagedA.Managed = false
+ client2 := map[string]common.ConfigFile{"a.yml": unmanagedA, "b.yml": cfg("b.yml", 2)}
+ resp2 := diffConfigs(assigned, client2)
+ if _, touched := resp2.Configs.UpdatedConfigs["a.yml"]; touched {
+ t.Fatal("unmanaged config must not be updated")
+ }
+ })
+
+ t.Run("server removed everything → all client configs deleted", func(t *testing.T) {
+ client := map[string]common.ConfigFile{"a.yml": cfg("a.yml", 1)}
+ resp := diffConfigs(nil, client)
+ if !resp.Changed || len(resp.Configs.DeletedConfigs) != 1 {
+ t.Fatalf("expected deletion, got %+v", resp.Configs)
+ }
+ })
+
+ t.Run("empty both sides → no change", func(t *testing.T) {
+ resp := diffConfigs(nil, nil)
+ if resp.Changed {
+ t.Fatal("expected changed=false")
+ }
+ })
+}
+
+func TestConfigsHash(t *testing.T) {
+ files := []common.ConfigFile{cfg("a.yml", 1), cfg("b.yml", 2)}
+ h1 := ConfigsHash(files)
+ h2 := ConfigsHash(files)
+ if h1 == "" {
+ t.Fatal("hash must not be empty")
+ }
+ if h1 != h2 {
+ t.Fatal("hash must be stable")
+ }
+ // 顺序无关不应成立? 协议要求两侧列表一致 — 同序序列化, 顺序变化视为变更
+ reordered := []common.ConfigFile{files[1], files[0]}
+ if ConfigsHash(reordered) == h1 {
+ t.Log("note: hash is order-sensitive (both sides marshal the same list)")
+ }
+ if ConfigsHash(nil) != "" {
+ t.Fatal("empty list must hash to empty string")
+ }
+ if ConfigsHash([]common.ConfigFile{cfg("a.yml", 2)}) == h1 {
+ t.Fatal("version change must change the hash")
+ }
+}
+
+func TestTokenGate(t *testing.T) {
+ handler := func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ w.WriteHeader(http.StatusOK)
+ }
+ mkTokens := func(vals ...string) []ucfg.SecretString {
+ out := make([]ucfg.SecretString, len(vals))
+ for i, v := range vals {
+ out[i] = ucfg.SecretString(v)
+ }
+ return out
+ }
+ bearer := func(v string) *http.Request {
+ req := httptest.NewRequest("POST", "/x", nil)
+ req.Header.Set("Authorization", "Bearer "+v)
+ return req
+ }
+
+ t.Run("no tokens configured = open (dev mode)", func(t *testing.T) {
+ gate := newTokenGate(nil)
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, httptest.NewRequest("POST", "/x", nil), nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("open mode must pass, got %d", rec.Code)
+ }
+ })
+
+ t.Run("valid token passes, wrong/missing rejected", func(t *testing.T) {
+ gate := newTokenGate(mkTokens("s3cret"))
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, bearer("s3cret"), nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("valid token must pass, got %d", rec.Code)
+ }
+ rec2 := httptest.NewRecorder()
+ gate(handler)(rec2, bearer("wrong"), nil)
+ if rec2.Code != http.StatusUnauthorized {
+ t.Fatalf("wrong token must 401, got %d", rec2.Code)
+ }
+ rec3 := httptest.NewRecorder()
+ gate(handler)(rec3, httptest.NewRequest("POST", "/x", nil), nil)
+ if rec3.Code != http.StatusUnauthorized {
+ t.Fatalf("missing header must 401, got %d", rec3.Code)
+ }
+ })
+
+ t.Run("multiple tokens accepted (rotation window)", func(t *testing.T) {
+ gate := newTokenGate(mkTokens("old", "new"))
+ for _, tok := range []string{"old", "new"} {
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, bearer(tok), nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("token %q must pass, got %d", tok, rec.Code)
+ }
+ }
+ })
+
+ t.Run("X-API-Token header also accepted", func(t *testing.T) {
+ gate := newTokenGate(mkTokens("s3cret"))
+ req := httptest.NewRequest("POST", "/x", nil)
+ req.Header.Set("X-API-Token", "s3cret")
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, req, nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("X-API-Token must pass, got %d", rec.Code)
+ }
+ })
+
+ t.Run("validateStaticToken matches configured list only", func(t *testing.T) {
+ staticTokens = []string{"alpha", "beta"}
+ if !validateStaticToken("beta") {
+ t.Fatal("beta must validate")
+ }
+ if validateStaticToken("gamma") {
+ t.Fatal("gamma must not validate")
+ }
+ if validateStaticToken("") {
+ t.Fatal("empty must not validate")
+ }
+ })
+}
+
+func TestHashTokenOneWay(t *testing.T) {
+ h1 := hashToken("my-token")
+ if h1 == "my-token" || len(h1) != 64 { // sha256 hex
+ t.Fatalf("hash must be 64-hex and differ from input, got %q", h1)
+ }
+ if hashToken("my-token") != h1 {
+ t.Fatal("hash must be deterministic")
+ }
+ if hashToken("my-token2") == h1 {
+ t.Fatal("different inputs must hash differently")
+ }
+}
diff --git a/modules/easysearch/cluster_api.go b/modules/easysearch/cluster_api.go
index e757610a1..2974ee68a 100644
--- a/modules/easysearch/cluster_api.go
+++ b/modules/easysearch/cluster_api.go
@@ -10,8 +10,11 @@ import (
"infini.sh/framework/core/api/crud"
httprouter "infini.sh/framework/core/api/router"
"infini.sh/framework/core/elastic"
+ "infini.sh/framework/core/model"
+ "infini.sh/framework/core/orm"
"infini.sh/framework/core/security"
"infini.sh/framework/core/util"
+ "infini.sh/framework/lib/go-ucfg"
)
// ──────────────────────────────────────────────────────────────────────────
@@ -97,6 +100,7 @@ func registerClusterAPI() {
}
return nil
},
+ PrepareUpdate: preserveClusterSecrets,
})
// Pre-registration connectivity probe — deliberately bespoke.
@@ -162,3 +166,74 @@ func (h *ClusterAPI) testConnection(w http.ResponseWriter, req *http.Request, _
// ──────────────────────────────────────────────────────────────────────────
// Helpers
// ──────────────────────────────────────────────────────────────────────────
+
+// preserveClusterSecrets keeps stored cluster credentials intact when an
+// update request carries the display mask (or a blank / missing password
+// field): the edit form cannot recover the real password, so it sends
+// "******" back (or omits it), which would otherwise overwrite the stored
+// credential and break every connection that references the cluster.
+//
+// Works for both update modes:
+// - partial: the delta's basic_auth object would REPLACE the nested
+// stored object wholesale, so a delta without a real password must be
+// repaired with the stored value before it is applied;
+// - full: the decoded body has already overwritten obj, so the stored
+// record is re-read from the ORM and the masked fields restored.
+func preserveClusterSecrets(obj *elastic.ElasticsearchConfig, delta util.MapStr) error {
+ masked := func(s string) bool { return s == "" || s == ucfg.SecretShadowText }
+
+ // Determine what the request actually wants for each credential.
+ // Delta values (partial mode) take precedence over obj fields, since
+ // obj still holds the stored values until the delta is applied.
+ incomingPassword := ""
+ if obj.BasicAuth != nil {
+ incomingPassword = obj.BasicAuth.Password.Get()
+ }
+ if ba, ok := delta["basic_auth"].(map[string]interface{}); ok {
+ if pw, ok := ba["password"].(string); ok {
+ incomingPassword = pw
+ } else {
+ // basic_auth present without a password key: the nested
+ // replace would drop the stored password entirely.
+ incomingPassword = ""
+ }
+ }
+ incomingToken := obj.Token.Get()
+ if tok, ok := delta["token"].(string); ok {
+ incomingToken = tok
+ }
+
+ if !masked(incomingPassword) && !masked(incomingToken) {
+ return nil // real credentials submitted: nothing to preserve
+ }
+
+ stored := &elastic.ElasticsearchConfig{}
+ stored.ID = obj.ID // ID is promoted (embedded): not settable via struct literal
+ ctx := orm.NewContext().DirectAccess()
+ exists, err := orm.GetV2(ctx, stored)
+ if err != nil || !exists {
+ return nil // nothing stored to preserve
+ }
+
+ if masked(incomingPassword) && stored.BasicAuth != nil {
+ if pw := stored.BasicAuth.Password.Get(); pw != "" && !masked(pw) {
+ if obj.BasicAuth == nil {
+ obj.BasicAuth = &model.BasicAuth{Username: stored.BasicAuth.Username}
+ }
+ obj.BasicAuth.Password = ucfg.SecretString(pw)
+ if ba, ok := delta["basic_auth"].(map[string]interface{}); ok {
+ ba["password"] = pw
+ } else if _, hasDelta := delta["basic_auth"]; hasDelta {
+ // delta carried basic_auth in a non-map form: replace it
+ delta["basic_auth"] = util.MapStr{"username": stored.BasicAuth.Username, "password": pw}
+ }
+ }
+ }
+ if masked(incomingToken) {
+ if tok := stored.Token.Get(); tok != "" && !masked(tok) {
+ obj.Token = ucfg.SecretString(tok)
+ delta["token"] = tok
+ }
+ }
+ return nil
+}
diff --git a/modules/pipeline/for_each.go b/modules/pipeline/for_each.go
new file mode 100644
index 000000000..2364e0640
--- /dev/null
+++ b/modules/pipeline/for_each.go
@@ -0,0 +1,126 @@
+// 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 .
+
+/* ©INFINI, All Rights Reserved.
+ * mail: contact#infini.ltd */
+
+package pipeline
+
+import (
+ "fmt"
+
+ log "github.com/cihub/seelog"
+ "infini.sh/framework/core/config"
+ "infini.sh/framework/core/otel"
+ "infini.sh/framework/core/param"
+ "infini.sh/framework/core/pipeline"
+ "infini.sh/framework/core/queue"
+)
+
+// ForEachProcessor splits a batch of queue messages into individual
+// records and runs a sub-chain of processors on each record.
+//
+// Each message payload is decoded into an *event.Event (otel envelope),
+// exposed to the sub-chain under pipeline.RecordContextKey, mutated in
+// place, then re-encoded back into the message. Per-record transform
+// processors (dissect, field_standardize, ...) rely on this convention.
+//
+// Configuration:
+//
+// - for_each:
+// message_field: messages # where the consumer stored []queue.Message
+// processor: # sub-chain, executed per record
+// - dissect:
+// pattern: "%{log_level} %{message}"
+type ForEachConfig struct {
+ MessageField string `config:"message_field"`
+ Processors []*config.Config `config:"processor"`
+}
+
+type ForEachProcessor struct {
+ cfg ForEachConfig
+ sub *pipeline.Processors
+}
+
+func NewForEachProcessor(c *config.Config) (pipeline.Processor, error) {
+ cfg := ForEachConfig{MessageField: "messages"}
+ if err := c.Unpack(&cfg); err != nil {
+ return nil, fmt.Errorf("failed to unpack the configuration of for_each processor: %s", err)
+ }
+ if len(cfg.Processors) == 0 {
+ return nil, fmt.Errorf("for_each processor requires a non-empty processor sub-chain")
+ }
+ sub, err := pipeline.NewPipeline(cfg.Processors)
+ if err != nil {
+ return nil, fmt.Errorf("failed to compile for_each sub-chain: %s", err)
+ }
+ return &ForEachProcessor{cfg: cfg, sub: sub}, nil
+}
+
+func (p *ForEachProcessor) Name() string { return "for_each" }
+
+func (p *ForEachProcessor) Process(c *pipeline.Context) error {
+ v := c.Get(param.ParaKey(p.cfg.MessageField))
+ msgs, ok := v.([]queue.Message)
+ if !ok {
+ log.Debugf("for_each: no batch found under context key [%s], skip", p.cfg.MessageField)
+ return nil
+ }
+
+ for i := range msgs {
+ if c.IsCanceled() || !c.ShouldContinue() {
+ break
+ }
+ data := msgs[i].Data
+ if len(data) == 0 {
+ continue
+ }
+ rec, err := otel.DecodeEnvelope(data)
+ if err != nil {
+ log.Warnf("for_each: failed to decode message payload at offset %v: %v", msgs[i].Offset, err)
+ continue
+ }
+
+ c.Set(pipeline.RecordContextKey, rec)
+ if err := p.sub.Process(c); err != nil {
+ log.Warnf("for_each: sub-chain error on record at offset %v: %v", msgs[i].Offset, err)
+ }
+
+ // drop_event marks: clear the payload so downstream stages
+ // (otlp_export and friends) skip this record entirely.
+ if pipeline.IsDropped(rec) {
+ msgs[i].Data = nil
+ msgs[i].Size = 0
+ continue
+ }
+
+ encoded, err := otel.EncodeEnvelope(rec)
+ if err != nil {
+ log.Warnf("for_each: failed to encode record at offset %v: %v", msgs[i].Offset, err)
+ continue
+ }
+ msgs[i].Data = encoded
+ msgs[i].Size = len(encoded)
+ }
+ return nil
+}
diff --git a/modules/pipeline/module.go b/modules/pipeline/module.go
index b742e1b49..621422c97 100755
--- a/modules/pipeline/module.go
+++ b/modules/pipeline/module.go
@@ -83,6 +83,7 @@ func (module *PipeModule) Setup() {
pipeline.RegisterProcessorPlugin("dag", pipeline.NewDAGProcessor)
pipeline.RegisterProcessorPlugin("echo", NewEchoProcessor)
+ pipeline.RegisterProcessorPluginWithConfigMetadata("for_each", NewForEachProcessor, ForEachConfig{})
//TODO remove
api.HandleAPIMethod(api.GET, "/pipeline/tasks/", module.getRunningPipelineTasksHandler)
@@ -93,6 +94,9 @@ func (module *PipeModule) Setup() {
api.HandleAPIMethod(api.POST, "/pipeline/task/:id/_start", module.startPipelineTaskHandler)
api.HandleAPIMethod(api.POST, "/pipeline/task/:id/_stop", module.stopPipelineTaskHandler)
+ // processor discovery: names + config schemas for pipeline designers
+ api.HandleAPIMethod(api.GET, "/pipeline/processors", module.getProcessorsHandler)
+
//use pipelines to avoid naming conflicts
api.HandleUIMethod(api.POST, "/pipelines/_search", module.searchPipelineHandler, api.RequirePermission(security.GetOrInitPermission("generic", "pipeline", security.Search)))
api.HandleUIMethod(api.POST, "/pipelines/", module.createPipelineHandler, api.RequirePermission(security.GetOrInitPermission("generic", "pipeline", security.Create)))
diff --git a/modules/pipeline/processors_api.go b/modules/pipeline/processors_api.go
new file mode 100644
index 000000000..0ea4df7dc
--- /dev/null
+++ b/modules/pipeline/processors_api.go
@@ -0,0 +1,42 @@
+// 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 .
+
+/* ©INFINI, All Rights Reserved.
+ * mail: contact#infini.ltd */
+
+package pipeline
+
+import (
+ "net/http"
+
+ httprouter "infini.sh/framework/core/api/router"
+
+ "infini.sh/framework/core/pipeline"
+)
+
+// getProcessorsHandler serves GET /pipeline/processors: the registry of
+// available pipeline processors with their config schemas, so that
+// pipeline designer UIs can render configuration forms.
+func (module *PipeModule) getProcessorsHandler(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+ module.WriteJSON(w, pipeline.GetProcessorMetadata(), 200)
+}
diff --git a/modules/sqlite/orm.go b/modules/sqlite/orm.go
index 0926949a9..22192ea02 100644
--- a/modules/sqlite/orm.go
+++ b/modules/sqlite/orm.go
@@ -158,7 +158,11 @@ func (handler *SQLiteORM) Create(ctx *api.Context, o interface{}) error {
}
tableName := handler.GetIndexName(o)
- rawJSON := util.MustToJSONBytes(o)
+ // Persist with secrets resolved: SecretString.MarshalJSON emits the
+ // "******" mask for display, which would destroy plain passwords and
+ // tokens on storage round-trip (e.g. registered ES clusters losing
+ // their credentials between create and the next read).
+ rawJSON := util.MustToJSONBytesWithSecrets(o)
query := fmt.Sprintf("INSERT INTO [%s] (id, raw) VALUES (?, ?)", tableName)
if global.Env().IsDebug {
@@ -176,7 +180,7 @@ func (handler *SQLiteORM) Save(ctx *api.Context, o interface{}) error {
}
tableName := handler.GetIndexName(o)
- rawJSON := util.MustToJSONBytes(o)
+ rawJSON := util.MustToJSONBytesWithSecrets(o)
query := fmt.Sprintf("INSERT OR REPLACE INTO [%s] (id, raw) VALUES (?, ?)", tableName)
if global.Env().IsDebug {
@@ -194,7 +198,7 @@ func (handler *SQLiteORM) Update(ctx *api.Context, o interface{}) error {
}
tableName := handler.GetIndexName(o)
- rawJSON := util.MustToJSONBytes(o)
+ rawJSON := util.MustToJSONBytesWithSecrets(o)
query := fmt.Sprintf("UPDATE [%s] SET raw = ? WHERE id = ?", tableName)
if global.Env().IsDebug {
diff --git a/plugins/queue/queue_output/queue_output.go b/plugins/queue/queue_output/queue_output.go
new file mode 100644
index 000000000..128522c58
--- /dev/null
+++ b/plugins/queue/queue_output/queue_output.go
@@ -0,0 +1,81 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+// Package queue_output provides the "queue_output" pipeline processor: the
+// chain-tail companion of "consumer". It takes the message batch the
+// consumer exposed in the context (typically after a for_each transform
+// chain) and appends every record onto a target queue, enabling two-stage
+// pipelines: process on one queue, sink (e.g. bulk_indexing) from another.
+//
+// Configuration:
+//
+// - queue_output:
+// queue_name: indexing-my-stream # target queue (required)
+// message_field: messages # ctx batch key (default "messages")
+//
+// On queue push failure the processor returns an error so the consumer
+// does not commit the offset and the batch is redelivered (at-least-once).
+package queue_output
+
+import (
+ "fmt"
+
+ "infini.sh/framework/core/config"
+ "infini.sh/framework/core/param"
+ "infini.sh/framework/core/pipeline"
+ "infini.sh/framework/core/queue"
+ log "infini.sh/framework/core/log"
+)
+
+const name = "queue_output"
+
+type Config struct {
+ QueueName string `config:"queue_name"`
+ MessageField string `config:"message_field"`
+}
+
+type Processor struct {
+ cfg Config
+}
+
+func init() {
+ pipeline.RegisterProcessorPlugin(name, New)
+}
+
+func New(c *config.Config) (pipeline.Processor, error) {
+ cfg := Config{MessageField: "messages"}
+ if err := c.Unpack(&cfg); err != nil {
+ return nil, fmt.Errorf("failed to unpack the configuration of %s processor: %s", name, err)
+ }
+ if cfg.QueueName == "" {
+ return nil, fmt.Errorf("%s processor requires queue_name", name)
+ }
+ return &Processor{cfg: cfg}, nil
+}
+
+func (p *Processor) Name() string { return name }
+
+// Process appends every record of the context batch onto the target queue.
+func (p *Processor) Process(c *pipeline.Context) error {
+ v := c.Get(param.ParaKey(p.cfg.MessageField))
+ msgs, ok := v.([]queue.Message)
+ if !ok || len(msgs) == 0 {
+ return nil
+ }
+
+ qConfig := queue.GetOrInitConfig(p.cfg.QueueName)
+ pushed := 0
+ for i := range msgs {
+ if len(msgs[i].Data) == 0 {
+ continue // dropped records (drop_event) are skipped
+ }
+ if err := queue.Push(qConfig, msgs[i].Data); err != nil {
+ log.Errorf("%s: queue push failed after %d/%d records: %v", name, pushed, len(msgs), err)
+ return fmt.Errorf("%s: queue push failed after %d/%d records: %w",
+ name, pushed, len(msgs), err)
+ }
+ pushed++
+ }
+ return nil
+}