defaultone

Documentation

Index

• Constants
• type AggregatorDefault
  • func NewAggregatorDefault() *AggregatorDefault
  • func (p *AggregatorDefault) Add(log *protocol.Log) error
  • func (*AggregatorDefault) Description() string
  • func (p *AggregatorDefault) Flush() []*protocol.LogGroup
  • func (p *AggregatorDefault) Init(context ilogtail.Context, que ilogtail.LogGroupQueue) (int, error)
  • func (p *AggregatorDefault) InitInner(packFlag bool, packString string, lock *sync.Mutex, logstore string, ...)
  • func (p *AggregatorDefault) Reset()

Constants

const (
	MaxLogCount     = 1024
	MaxLogGroupSize = 3 * 1024 * 1024
)

Types

type AggregatorDefault

type AggregatorDefault struct {
	MaxLogGroupCount int    // the maximum log group count to trigger flush operation
	MaxLogCount      int    // the maximum log in a log group
	PackFlag         bool   // whether to add config name as a tag
	Topic            string // the output topic

	Lock *sync.Mutex
	// contains filtered or unexported fields
}

AggregatorDefault is the default aggregator in plugin system.

It has two usages: 1. If there is no specific aggregator in plugin config, it will be added. 2. Other aggregators can use it as base aggregator.

For inner usage, note about following information. There is a quick flush design in AggregatorDefault, which is implemented in Add method (search p.queue.Add in current file). Therefore, not all LogGroups are returned through Flush method. If you want to do some operations (such as adding tags) on LogGroups returned by AggregatorDefault in your own aggregator, you should do some extra works, just see the sample code in doc.go.

func NewAggregatorDefault

func NewAggregatorDefault() *AggregatorDefault

NewAggregatorDefault create a default aggregator with default value.

func (*AggregatorDefault) Add

func (p *AggregatorDefault) Add(log *protocol.Log) error

Add adds @log to aggregator. It uses defaultLogGroup to store log groups which contain logs as following: defaultLogGroup => [LG1: log1->log2->log3] -> [LG2: log1->log2->log3] -> .. The last log group is set as nowLogGroup, @log will be appended to nowLogGroup if the size and log count of the log group don't exceed limits (MaxLogCount and MAX_LOG_GROUP_SIZE). When nowLogGroup exceeds limits, Add creates a new log group and switch nowLogGroup to it, then append @log to it. When the count of log group reaches MaxLogGroupCount, the first log group will be popped from defaultLogGroup list and add to queue (after adding pack_id tag). Add returns any error encountered, nil means success.

@return error. **For inner usage, must handle this error!!!!**

func (*AggregatorDefault) Description

func (*AggregatorDefault) Description() string

func (*AggregatorDefault) Flush

func (p *AggregatorDefault) Flush() []*protocol.LogGroup

Flush ...

func (*AggregatorDefault) Init

func (p *AggregatorDefault) Init(context ilogtail.Context, que ilogtail.LogGroupQueue) (int, error)

Init method would be trigger before working. 1. context store the metadata of this Logstore config 2. que is a transfer channel for flushing LogGroup when reaches the maximum in the cache.

func (*AggregatorDefault) InitInner

func (p *AggregatorDefault) InitInner(packFlag bool, packString string, lock *sync.Mutex, logstore string, topic string, maxLogCount int, maxLoggroupCount int)

InitInner initializes instance for other aggregators.

func (*AggregatorDefault) Reset

func (p *AggregatorDefault) Reset()

Reset ...