feat: Define the metrics data model and modify the plugin manager to support procress the metrics data

.gitignore

@@ -47,6 +47,7 @@ e2e-test
 # vscode
 .vscode
 !.vscode/settings.json
+!.vscode/launch.json
 output
 dist
 
@@ -57,4 +58,5 @@ dist
 .coretestCoverage.txt
 .testCoverage.txt
 plugin_main/*.dll
-plugin_main/*.so
+plugin_main/*.so
+plugin_main/*.json

.vscode/launch.json

@@ -12,4 +12,4 @@
             }
         }
     ]
-}
+}

aggregator.go

@@ -15,6 +15,7 @@
 package ilogtail
 
 import (
+	"github.com/alibaba/ilogtail/pkg/models"
 	"github.com/alibaba/ilogtail/pkg/protocol"
 )
 
@@ -30,12 +31,23 @@ type Aggregator interface {
 	// Description returns a one-sentence description on the Input.
 	Description() string
 
-	// Add the metric to the aggregator.
-	Add(log *protocol.Log, ctx map[string]interface{}) error
-
-	// Flush pushes the current aggregates to the accumulator.
-	Flush() []*protocol.LogGroup
-
 	// Reset resets the aggregators caches and aggregates.
 	Reset()
 }
+
+type SlsAggregator interface {
+	Aggregator
+	// AddLogs the metric to the aggregator.
+	AddLogs(log *protocol.Log, ctx map[string]interface{}) error
+
+	// FlushLogs pushes the current aggregates to the accumulator.
+	FlushLogs() []*protocol.LogGroup
+}
+
+type PipelineAggregator interface {
+	Aggregator
+	// Add the metric to the aggregator.
+	Add(*models.PipelineGroupEvents, PipelineContext) error
+	// Flush pushes the current aggregates to the accumulator.
+	Flush(PipelineContext) error
+}

docs/cn/developer-guide/plugin-development/how-to-write-input-plugins.md

@@ -22,7 +22,7 @@ type MetricInput interface {
 
     // Collect takes in an accumulator and adds the metrics that the Input
     // gathers. This is called every "interval"
-    Collect(Collector) error
+    CollectLogs(Collector) error
 }
 ```
 

docs/en/guides/How-to-write-input-plugins.md

@@ -28,7 +28,7 @@ Description() string
 
 // Collect takes in an accumulator and adds the metrics that the Input
 // gathers. This is called every "interval"
-Collect(Collector) error
+CollectLogs(Collector) error
 }
 ```
 

external/github.com/elastic/beats/v7/winlogbeat/sys/wineventlog/winmeta.go

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build windows
 // +build windows
 
 package wineventlog

flusher.go

@@ -15,6 +15,7 @@
 package ilogtail
 
 import (
+	"github.com/alibaba/ilogtail/pkg/models"
 	"github.com/alibaba/ilogtail/pkg/protocol"
 )
 
@@ -34,11 +35,6 @@ type Flusher interface {
 	//   stopped gracefully.
 	IsReady(projectName string, logstoreName string, logstoreKey int64) bool
 
-	// Flush flushes data to destination, such as SLS, console, file, etc.
-	// It is expected to return no error at most time because IsReady will be called
-	// before it to make sure there is space for next data.
-	Flush(projectName string, logstoreName string, configName string, logGroupList []*protocol.LogGroup) error
-
 	// SetUrgent indicates the flusher that it will be destroyed soon.
 	// @flag indicates if main program (Logtail mostly) will exit after calling this.
 	//
@@ -60,3 +56,19 @@ type Flusher interface {
 	// In a word, flusher should only have things that can be recycled by GC after this.
 	Stop() error
 }
+
+type SlsFlusher interface {
+	Flusher
+	// FlushLogs flushes data to destination, such as SLS, console, file, etc.
+	// It is expected to return no error at most time because IsReady will be called
+	// before it to make sure there is space for next data.
+	FlushLogs(projectName string, logstoreName string, configName string, logGroupList []*protocol.LogGroup) error
+}
+
+type PipelineFlusher interface {
+	Flusher
+	// FlushLogs flushes data to destination, such as SLS, console, file, etc.
+	// It is expected to return no error at most time because IsReady will be called
+	// before it to make sure there is space for next data.
+	Flush([]*models.PipelineGroupEvents, PipelineContext) error
+}

helper/math_helper.go

@@ -0,0 +1,33 @@
+// Copyright 2021 iLogtail Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package helper
+
+type IntUintFloat interface {
+	~int | ~int8 | ~int16 | ~int32 | ~int64 | ~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64 | ~float32 | ~float64
+}
+
+func Max[T IntUintFloat](x T, y T) T {
+	if x > y {
+		return x
+	}
+	return y
+}
+
+func Min[T IntUintFloat](x T, y T) T {
+	if x < y {
+		return x
+	}
+	return y
+}

input.go

@@ -22,10 +22,20 @@ type MetricInput interface {
 
 	// Description returns a one-sentence description on the Input
 	Description() string
+}
+
+type SlsMetricInput interface {
+	MetricInput
+	// CollectLogs takes in an accumulator and adds the metrics that the Input
+	// gathers. This is called every "interval"
+	CollectLogs(Collector) error
+}
 
+type PipelineMetricInput interface {
+	MetricInput
 	// Collect takes in an accumulator and adds the metrics that the Input
 	// gathers. This is called every "interval"
-	Collect(Collector) error
+	Collect(PipelineContext) error
 }
 
 // ServiceInput ...
@@ -37,9 +47,18 @@ type ServiceInput interface {
 	// Description returns a one-sentence description on the Input
 	Description() string
 
-	// Start starts the ServiceInput's service, whatever that may be
-	Start(Collector) error
-
 	// Stop stops the services and closes any necessary channels and connections
 	Stop() error
 }
+
+type SlsServiceInput interface {
+	ServiceInput
+	// StartCollectLogs starts the ServiceInput's service, whatever that may be
+	StartCollectLogs(Collector) error
+}
+
+type PipelineServiceInput interface {
+	ServiceInput
+	// Start starts the ServiceInput's service, whatever that may be
+	Start(PipelineContext) error
+}

pipeline.go

@@ -0,0 +1,128 @@
+// Copyright 2022 iLogtail Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ilogtail
+
+import (
+	"github.com/alibaba/ilogtail/pkg/models"
+)
+
+type PipelineContext interface {
+	Collector() PipelineCollector
+}
+
+type PipelineCollector interface {
+	Collect(group *models.GroupInfo, events ...models.PipelineEvent)
+
+	Dump() []*models.PipelineGroupEvents
+
+	Observe() chan *models.PipelineGroupEvents
+}
+
+type observePipeCollector struct {
+	groupChan chan *models.PipelineGroupEvents
+}
+
+func (p *observePipeCollector) Collect(group *models.GroupInfo, events ...models.PipelineEvent) {
+	if len(events) == 0 {
+		return
+	}
+	p.groupChan <- &models.PipelineGroupEvents{
+		Group:  group,
+		Events: events,
+	}
+}
+
+func (p *observePipeCollector) Dump() []*models.PipelineGroupEvents {
+	return nil
+}
+
+func (p *observePipeCollector) Observe() chan *models.PipelineGroupEvents {
+	return p.groupChan
+}
+
+type dumpPipeCollector struct {
+	groupEvents map[*models.GroupInfo][]models.PipelineEvent
+}
+
+func (p *dumpPipeCollector) Collect(group *models.GroupInfo, events ...models.PipelineEvent) {
+	store, has := p.groupEvents[group]
+	if !has {
+		store = make([]models.PipelineEvent, 0)
+	}
+	p.groupEvents[group] = append(store, events...)
+}
+
+func (p *dumpPipeCollector) Dump() []*models.PipelineGroupEvents {
+	len, idx := len(p.groupEvents), 0
+	results := make([]*models.PipelineGroupEvents, len)
+	if len == 0 {
+		return results
+	}
+	for group, events := range p.groupEvents {
+		results[idx] = &models.PipelineGroupEvents{
+			Group:  group,
+			Events: events,
+		}
+		idx++
+	}
+	p.groupEvents = make(map[*models.GroupInfo][]models.PipelineEvent)
+	return results
+}
+
+func (p *dumpPipeCollector) Observe() chan *models.PipelineGroupEvents {
+	return nil
+}
+
+type voidPipeCollector struct {
+}
+
+func (p *voidPipeCollector) Collect(group *models.GroupInfo, events ...models.PipelineEvent) {
+}
+
+func (p *voidPipeCollector) Dump() []*models.PipelineGroupEvents {
+	return nil
+}
+
+func (p *voidPipeCollector) Observe() chan *models.PipelineGroupEvents {
+	return nil
+}
+
+type defaultPipelineContext struct {
+	collector PipelineCollector
+}
+
+func (p *defaultPipelineContext) Collector() PipelineCollector {
+	return p.collector
+}
+
+func NewObservePipelineConext(queueSize int) PipelineContext {
+	return newPipelineConext(&observePipeCollector{
+		groupChan: make(chan *models.PipelineGroupEvents, queueSize),
+	})
+}
+
+func NewDumpPipelineConext() PipelineContext {
+	return newPipelineConext(&dumpPipeCollector{
+		groupEvents: make(map[*models.GroupInfo][]models.PipelineEvent),
+	})
+}
+
+func NewVoidPipelineConext() PipelineContext {
+	return newPipelineConext(&voidPipeCollector{})
+}
+
+func newPipelineConext(collector PipelineCollector) PipelineContext {
+	return &defaultPipelineContext{collector: collector}
+}