feat(pipeline): unified pipeline engine, distributed deployment, and lane-based bulk indexing

.github/workflows/unit_test.yml

@@ -49,4 +49,70 @@ jobs:
                 # for unit test
                 cd $WORK
                 echo Testing code at $PWD ...
+                OFFLINE_BUILD=true CI=true make tidy test
+
+    enterprise-test:
+        timeout-minutes: 60
+        runs-on: ubuntu-latest
+        env:
+            GO_VERSION: 1.23.4
+        steps:
+            - name: Checkout current repository
+              uses: actions/checkout@v4
+              with:
+                path: framework
+
+            - name: Set up go toolchain
+              uses: actions/setup-go@v5
+              with:
+                go-version: ${{ env.GO_VERSION }}
+                check-latest: false
+                cache: true
+
+            # Enterprise plugins overlay (private repo): prefer the branch
+            # matching this PR's branch name, fall back to the default branch
+            # when it does not exist there; the license module is a build
+            # dependency of the overlay.
+            - name: Checkout enterprise plugins overlay
+              env:
+                CI_GH_TOKEN: ${{ secrets.CI_TOKEN || secrets.DOCS_DEPLOYMENT_TOKEN }}
+                OVERLAY_REF: ${{ github.head_ref }}
+              run: |
+                set -e
+                REPO="https://x-access-token:${CI_GH_TOKEN}@github.com/infinilabs/framework-enterprise-plugins.git"
+                if [ -n "$OVERLAY_REF" ] && git ls-remote --exit-code --heads "$REPO" "refs/heads/${OVERLAY_REF}" >/dev/null 2>&1; then
+                    git clone --depth 1 --branch "${OVERLAY_REF}" "$REPO" framework/plugins/enterprise
+                else
+                    git clone --depth 1 "$REPO" framework/plugins/enterprise
+                fi
+                git -C framework/plugins/enterprise log --oneline -1
+
+            - name: Checkout license module
+              uses: actions/checkout@v4
+              with:
+                token: ${{ secrets.CI_TOKEN || secrets.DOCS_DEPLOYMENT_TOKEN }}
+                repository: infinilabs/license
+                ref: main
+                path: license
+
+            - name: Check go toolchain
+              run: go version
+
+            - name: Unit test with enterprise overlay
+              env:
+                GOFLAGS: -tags=ci
+              run: |
+                echo Home path is $HOME
+                export WORKBASE=$HOME/go/src/infini.sh
+                export WORK=$WORKBASE/framework
+
+                # for test workspace
+                mkdir -p $HOME/go/src/
+                ln -s $GITHUB_WORKSPACE $WORKBASE
+
+                # the overlay's umbrella packages need the private infini.sh/license module (kept out of go.mod)
+                cd $WORK
+                go mod edit -replace infini.sh/license=../license
+
+                echo "Testing code at $PWD (with enterprise overlay) ..."
                 OFFLINE_BUILD=true CI=true make tidy test

.gitignore

@@ -62,3 +62,4 @@ generated_*.go
 
 # keystore runtime artifacts
 core/elastic/.keystore/
+.mimosa/

README.md

@@ -1,5 +1,22 @@
 # INFINI Framework
 
+## Pipeline Terminology (glossary)
+
+The data-processing subsystems use a fixed vocabulary. The short version:
+
+| Term | Meaning |
+|------|---------|
+| **message** | A queue's unit of storage — opaque bytes. Queues never interpret payloads. |
+| **batch** | The pipeline's native input: a slice of messages, the *batch scope* all historical processors operate on. |
+| **record** | A typed, decoded event inside a record host's sub-chain — the *record scope*; mutable in place, never leaks downstream. |
+| **record host** | A splitter processor that turns a batch into records and back: `for_each`, `lane_consumer`, or a custom host. |
+| **codec** | Pluggable byte↔record converter (`otel` default, `coco_document`, ...). |
+| **namespace** | A record family (`document`, `attachment`) used in compile-time contract checks. |
+| **queue** | Durable at-least-once **byte** channel; cross-process safe. |
+| **lane** | In-process at-most-once **record** channel (zero serialization, pointer hand-off). |
+
+Full definitions and the queue-vs-lane selection guide live in the docs: [Glossary](docs/content.en/docs/references/glossary.md), [Pipeline Record Processing](docs/content.en/docs/references/pipeline_record.md), [Lane](docs/content.en/docs/references/lane.md).
+
 ## Requirements
 - Go 1.21+
 

core/config/config_test.go

@@ -971,3 +971,47 @@ func TestLoadFileExpandsConfigTemplatesPlain(t *testing.T) {
 	assert.Equal(t, out.Entry[0].Name, "my_entry_test_entry")
 	assert.Equal(t, out.Entry[0].Network.Binding, "127.0.0.1:8002")
 }
+
+// A processor entry with an empty body (`- echo: {}`) must keep its key
+// through LoadPath + Unpack. Empty dicts used to reify to nil, which made
+// the KEY vanish from map[string]interface{} targets and broke every
+// config-file-loaded pipeline (managed instances saw "each processor must
+// have exactly one action, but found 0 actions").
+func TestLoadPathKeepsEmptyProcessorKeys(t *testing.T) {
+	dir := t.TempDir()
+	yml := "pipeline:\n" +
+		"  - name: t\n" +
+		"    auto_start: false\n" +
+		"    processor:\n" +
+		"      - echo: {}\n" +
+		"      - node_stamp:\n" +
+		"          field: x\n"
+	if err := os.WriteFile(filepath.Join(dir, "p.yml"), []byte(yml), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	cfg, err := LoadPath(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	child, err := cfg.Child("pipeline", -1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	type rawPipeline struct {
+		Processor []map[string]interface{} `config:"processor"`
+	}
+	list := []rawPipeline{}
+	if err := child.Unpack(&list); err != nil {
+		t.Fatal(err)
+	}
+	if len(list) != 1 || len(list[0].Processor) != 2 {
+		t.Fatalf("unexpected unpack result: %#v", list)
+	}
+	if _, ok := list[0].Processor[0]["echo"]; !ok {
+		t.Fatalf("empty processor body lost its key: %#v", list[0].Processor[0])
+	}
+	if _, ok := list[0].Processor[1]["node_stamp"]; !ok {
+		t.Fatalf("node_stamp lost its key: %#v", list[0].Processor[1])
+	}
+}

core/elastic/cluster_secrets.go

@@ -107,22 +107,25 @@ func HydrateClusterSecrets(cfg *ElasticsearchConfig) {
 		v, ok := loadClusterSecret(ClusterBasicAuthPasswordKey(cfg.ID))
 		if ok {
 			if v == "" {
-				log.Warnf("cluster [%s]'s basic auth password is empty: %v", cfg.ID)
+				log.Warnf("cluster [%s]'s basic auth password is empty", cfg.ID)
 			}
 			cfg.BasicAuth.Password = ucfg.SecretString(v)
 		} else {
-			log.Warnf("cluster [%s]'s basic auth password not found in the keystore: %v", cfg.ID)
+			log.Warnf("cluster [%s]'s basic auth password not found in the keystore", cfg.ID)
 		}
 	}
-	if !isRealSecret(cfg.Token.Get()) {
+	// Empty token = the cluster never had one (no-auth) — nothing to
+	// hydrate and nothing to warn about. Only a MASKED token (record says a
+	// token exists, value stashed elsewhere) requires the keystore lookup.
+	if tok := cfg.Token.Get(); tok != "" && !isRealSecret(tok) {
 		v, ok := loadClusterSecret(ClusterTokenKey(cfg.ID))
 		if ok {
 			if v == "" {
-				log.Warnf("cluster [%s]'s token is empty: %v", cfg.ID)
+				log.Warnf("cluster [%s]'s token is empty", cfg.ID)
 			}
 			cfg.Token = ucfg.SecretString(v)
 		} else {
-			log.Warnf("cluster [%s]'s token not found in the keystore: %v", cfg.ID)
+			log.Warnf("cluster [%s]'s token not found in the keystore", cfg.ID)
 		}
 	}
 }
@@ -171,6 +174,13 @@ func isRealSecret(s string) bool {
 	return s != "" && s != ucfg.SecretShadowText
 }
 
+// IsRealSecret is the exported form of isRealSecret, for app-side code
+// building credential payloads from ORM-loaded cluster records (deciding
+// whether a hydrated value is safe to ship).
+func IsRealSecret(s string) bool {
+	return isRealSecret(s)
+}
+
 func asMapStr(v interface{}) map[string]interface{} {
 	switch m := v.(type) {
 	case map[string]interface{}:

core/elastic/cluster_secrets_test.go

@@ -9,8 +9,8 @@ import (
 	"os"
 	"testing"
 
-	"infini.sh/framework/core/model"
 	"infini.sh/framework/core/keystore"
+	"infini.sh/framework/core/model"
 	ucfg "infini.sh/framework/lib/go-ucfg"
 )
 

core/env/version.go

@@ -28,13 +28,17 @@
 package env
 
 type Version struct {
-	VersionNumber                  string `json:"number,omitempty"`
-	FrameworkBuildCommitHash       string `json:"framework_hash,omitempty"`
-	FrameworkVendorBuildCommitHash string `json:"vendor_hash,omitempty"`
-	BuildCommitHash                string `json:"build_hash,omitempty"`
-	BuildDate                      string `json:"build_date,omitempty"`
-	BuildNumber                    string `json:"build_number,omitempty"`
-	EolDate                        string `json:"eol_date,omitempty"`
+	VersionNumber                  string `json:"number,omitempty" elastic_mapping:"number: { type: keyword }"`
+	FrameworkBuildCommitHash       string `json:"framework_hash,omitempty" elastic_mapping:"framework_hash: { type: keyword }"`
+	FrameworkVendorBuildCommitHash string `json:"vendor_hash,omitempty" elastic_mapping:"vendor_hash: { type: keyword }"`
+	BuildCommitHash                string `json:"build_hash,omitempty" elastic_mapping:"build_hash: { type: keyword }"`
+	// BuildDate/EolDate must map to keyword, never a dynamic date: binaries
+	// built without version ldflags report placeholders ("N/A"), which a
+	// date mapping rejects with document_parsing_exception — killing
+	// instance heartbeats on indices whose mapping predates this pin.
+	BuildDate   string `json:"build_date,omitempty" elastic_mapping:"build_date: { type: keyword }"`
+	BuildNumber string `json:"build_number,omitempty" elastic_mapping:"build_number: { type: keyword }"`
+	EolDate     string `json:"eol_date,omitempty" elastic_mapping:"eol_date: { type: keyword }"`
 }
 
 func (env *Env) GetVersionInfo() Version {

core/event/event.go

@@ -47,6 +47,13 @@ type Event struct {
 	Meta       util.MapStr `json:"-"`
 	Private    interface{} `json:"-"` // for beats private use
 	TimeSeries bool        `json:"-"` // true if the event contains timeseries data
+
+	// pooled links to the record-pool wrapper while the event came from
+	// AcquireEvent; managed by pool.go, never serialized. Held as
+	// interface{} on purpose: a typed *pooledEvent back-reference would
+	// create a recursive type (Event -> pooledEvent -> Event) and blow
+	// the reflection walkers (GetTagsByTagName) with a stack overflow.
+	pooled interface{}
 }
 
 type EventMetadata struct {

core/event/pool.go

@@ -0,0 +1,138 @@
+// Copyright (C) INFINI Labs & INFINI LIMITED.
+//
+// The INFINI Framework is offered under the GNU Affero General Public License v3.0
+// and as commercial software.
+//
+// For commercial licensing, contact us at:
+//   - Website: infinilabs.com
+//   - Email: hello#infini.ltd
+//
+// Open Source licensed under AGPL V3:
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+// See the GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+/* ©INFINI, All Rights Reserved.
+ * mail: contact#infini.ltd */
+
+package event
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// The record pool recycles *Event across record-mode batches: decode-heavy
+// hosts (for_each) otherwise allocate a fresh Event, its field maps and the
+// whole decoded tree per record only to drop them after encode.
+//
+// Contract:
+//   - AcquireEvent returns a zeroed event; Decode-style producers write into
+//     it as if it were freshly allocated.
+//   - ReleaseEvent returns the event to the pool. It must be called only
+//     after the record has been fully encoded (or dropped) AND no downstream
+//     processor retains a reference — the host owns the record lifetime.
+//   - ReleaseEvent is a no-op on events that did not come from the pool
+//     (codec- or clone-allocated), so a host can release every record
+//     unconditionally.
+//   - A double release is rejected atomically (second call returns false)
+//     instead of pooling the same event twice.
+//
+// Reset policy: scalar fields are zeroed; the hot maps (Fields, Metadata
+// labels, Meta) are cleared in place — their bucket arrays are reused —
+// unless they grew past maxRecycledEntries, in which case they are dropped
+// so one giant record cannot pin memory in the pool forever.
+
+// maxRecycledEntries caps in-place map reuse; bigger maps are released to
+// the GC instead of living in the pool.
+const maxRecycledEntries = 128
+
+type pooledEvent struct {
+	live atomic.Bool
+	ev   Event
+}
+
+var recordPool = sync.Pool{New: func() interface{} { return &pooledEvent{} }}
+
+var (
+	poolAcquires atomic.Int64
+	poolReleases atomic.Int64
+)
+
+// AcquireEvent returns an event from the record pool (zeroed).
+func AcquireEvent() *Event {
+	pe := recordPool.Get().(*pooledEvent)
+	pe.live.Store(true)
+	pe.ev.pooled = pe
+	poolAcquires.Add(1)
+	return &pe.ev
+}
+
+// ReleaseEvent recycles a pooled event; returns true when the event went
+// back to the pool, false for non-pooled events or a rejected double
+// release.
+func ReleaseEvent(e *Event) bool {
+	if e == nil {
+		return false
+	}
+	pe, ok := e.pooled.(*pooledEvent)
+	if !ok {
+		return false
+	}
+	if !pe.live.CompareAndSwap(true, false) {
+		return false
+	}
+	e.resetForPool()
+	e.pooled = nil
+	poolReleases.Add(1)
+	recordPool.Put(pe)
+	return true
+}
+
+// PoolStats reports cumulative acquire/release counts; the difference is
+// the number of pooled events currently in use (leak detection).
+func PoolStats() (acquired, released int64) {
+	return poolAcquires.Load(), poolReleases.Load()
+}
+
+// resetForPool clears an event for reuse; see the pool doc for the
+// map-retention policy.
+func (e *Event) resetForPool() {
+	e.Fields = recycleMap(e.Fields)
+	e.Metadata.Labels = recycleMap(e.Metadata.Labels)
+	e.Meta = recycleMap(e.Meta)
+	e.Metadata.Category = ""
+	e.Metadata.Name = ""
+	e.Metadata.Version = ""
+	e.Metadata.Datatype = ""
+	e.Agent = nil
+	e.Private = nil
+	e.QueueName = ""
+	e.TimeSeries = false
+	e.Timestamp = time.Time{}
+}
+
+// recycleMap clears a map in place when it is small enough to be worth
+// keeping, else drops it.
+func recycleMap(m map[string]interface{}) map[string]interface{} {
+	if m == nil {
+		return nil
+	}
+	if len(m) > maxRecycledEntries {
+		return nil
+	}
+	for k := range m {
+		delete(m, k)
+	}
+	return m
+}