feat: add ext_default_encoder with prometheus encoder

docs/cn/SUMMARY.md

@@ -115,6 +115,12 @@
  * [Pulsar](plugins/flusher/flusher-pulsar.md)
  * [HTTP](plugins/flusher/flusher-http.md)
  * [Loki](plugins/flusher/loki.md)
+* [扩展](plugins/extension/README.md)
+ * [BasicAuth鉴权](plugins/extension/ext-basicauth.md)
+ * [协议解码/反序列化](plugins/extension/ext-default-decoder.md)
+ * [协议编码/序列化](plugins/extension/ext-default-encoder.md)
+ * [数据筛选](plugins/extension/ext-groupinfo-filter.md)
+ * [请求熔断](plugins/extension/ext-request-breaker.md)
 
 ## 工作原理 <a href="#principle" id="principle"></a>
 

docs/cn/plugins/extension/README.md

@@ -0,0 +1,3 @@
+# 扩展
+
+扩展插件用于对其它插件能力的补充（e.g. 鉴权、编解码、熔断、限流……）

docs/cn/plugins/extension/ext-default-encoder.md

@@ -0,0 +1,37 @@
+# DefaultEncoder Encoder扩展
+
+## 简介
+
+[ext_default_encoder](https://github.com/alibaba/ilogtail/blob/main/plugins/extension/default_encoder/default_encoder.go)
+扩展，实现了 [Encoder](https://github.com/alibaba/ilogtail/blob/main/pkg/pipeline/extensions/encoder.go) 接口，可以用在
+`flusher_http` 等插件中用于序列化不同的协议数据。
+
+## 版本
+
+[Alpha](../stability-level.md)
+
+## 配置参数
+
+| 参数 | 类型 | 是否必选 | 说明 |
+|-------------|--------|------|-----------------------------------------------------------------------------------------------------------|
+| Format | String | 是 | 具体的协议，[查看支持的具体协议列表](https://github.com/alibaba/ilogtail/blob/master/pkg/protocol/encoder/common/comon.go) |
+| SeriesLimit | Int | 否 | 触发序列化时序切片的最大长度，默认 1000，仅针对 Format=prometheus 时有效 |
+
+## 样例
+
+使用 `flusher_http` flusher 插件，配置发送 `prometheus` 协议数据。
+
+```yaml
+enable: true
+flushers:
+- Type: flusher_http
+ ...
+ Encoder:
+ Type: ext_default_encoder/prometheus
+ ...
+...
+extensions:
+- Type: ext_default_encoder/prometheus
+ Format: 'prometheus'
+ SeriesLimit: 1024
+```

docs/cn/plugins/overview.md

@@ -115,3 +115,9 @@
 | 名称 | 提供方 | 简介 |
 |----------------------------------------------------------------------------|-------------------------------------------------|-----------------------------|
 | [`ext_default_decoder`](extension/ext-default-decoder.md)<br> 默认的decoder扩展 | 社区<br>[`snakorse`](https://github.com/snakorse) | 将内置支持的Format以Decoder扩展的形式封装 |
+
+### Encoder
+
+| 名称 | 提供方 | 简介 |
+|----------------------------------------------------------------------------|--------------------------------------------------------|-----------------------------|
+| [`ext_default_encoder`](extension/ext-default-encoder.md)<br> 默认的encoder扩展 | 社区<br>[`yuanshuai.1900`](https://github.com/aiops1900) | 将内置支持的Format以Encoder扩展的形式封装 |

go.mod

@@ -5,7 +5,7 @@ go 1.19
 require (
  github.com/ClickHouse/clickhouse-go/v2 v2.6.0
  github.com/IBM/sarama v1.42.2
- github.com/VictoriaMetrics/VictoriaMetrics v1.83.1
+ github.com/VictoriaMetrics/VictoriaMetrics v1.83.0
  github.com/alibaba/ilogtail/pkg v0.0.0
  github.com/apache/pulsar-client-go v0.10.0
  github.com/buger/jsonparser v1.1.1
@@ -44,7 +44,7 @@ require (
  github.com/prometheus/client_golang v1.14.0
  github.com/prometheus/common v0.42.0
  github.com/prometheus/procfs v0.8.0
- github.com/pyroscope-io/pyroscope v0.37.2
+ github.com/pyroscope-io/pyroscope v1.5.0
  github.com/shirou/gopsutil v3.21.11+incompatible
  github.com/sirupsen/logrus v1.8.1
  github.com/smartystreets/goconvey v1.7.2

pkg/go.mod

@@ -4,6 +4,7 @@ go 1.19
 
 require (
  github.com/Microsoft/go-winio v0.5.2
+ github.com/VictoriaMetrics/VictoriaMetrics v1.83.0
  github.com/cespare/xxhash v1.1.0
  github.com/cespare/xxhash/v2 v2.2.0
  github.com/cihub/seelog v0.0.0-20170130134532-f561c5e57575
@@ -23,7 +24,7 @@ require (
  github.com/prometheus/common v0.42.0
  github.com/prometheus/prometheus v1.8.2-0.20210430082741-2a4b8e12bbf2
  github.com/pyroscope-io/jfr-parser v0.6.0
- github.com/pyroscope-io/pyroscope v0.0.0-00010101000000-000000000000
+ github.com/pyroscope-io/pyroscope v1.5.0
  github.com/richardartoul/molecule v1.0.0
  github.com/smartystreets/goconvey v1.7.2
  github.com/stretchr/testify v1.8.2

pkg/go.sum

@@ -783,6 +783,8 @@ github.com/hashicorp/serf v0.9.5/go.mod h1:UWDWwZeL5cuWDJdl0C6wrvrUwEqtQ4ZKBKKEN
 github.com/hetznercloud/hcloud-go v1.24.0/go.mod h1:3YmyK8yaZZ48syie6xpm3dt26rtB6s65AisBHylXYFA=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/hudl/fargo v1.3.0/go.mod h1:y3CKSmjA+wD2gak7sUSXTAoopbhU08POFhmITJgmKTg=
+github.com/iLogtail/VictoriaMetrics v1.83.4-ilogtail h1:LRDJt9eUKKhHdwPJRbC6tgtiMs/0XTjlCz1dl2pzRt0=
+github.com/iLogtail/VictoriaMetrics v1.83.4-ilogtail/go.mod h1:JagjwAO58g1WNpyr6x/lrQqMTf99d/WU/yxjADxBz8E=
 github.com/iLogtail/handy v0.0.0-20230327021402-6a47ec586270/go.mod h1:6ai2R0qBm3xL13e10jwvyIf91Spxvo/yREZE9KOz7so=
 github.com/iLogtail/jfr-parser v0.6.0 h1:dNaQ0Ng2BLE5uxrhUQwtx1q7O9LIQFpMthl3SV326AU=
 github.com/iLogtail/jfr-parser v0.6.0/go.mod h1:ZMcbJjfDkOwElEK8CvUJbpetztRWRXszCmf5WU0erV8=

pkg/pipeline/extensions/encoder.go

@@ -0,0 +1,53 @@
+// Copyright 2024 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 extensions
+
+import (
+ "github.com/alibaba/ilogtail/pkg/models"
+ "github.com/alibaba/ilogtail/pkg/pipeline"
+ "github.com/alibaba/ilogtail/pkg/protocol"
+)
+
+// Encoder encodes data of iLogtail data models into bytes.
+// Different drivers with different encoding protocols implement Encoder interface.
+//
+// drivers: raw, influxdb, prometheus, sls, ...
+type Encoder interface {
+ EncoderV1
+ EncoderV2
+}
+
+// EncoderV1 supports v1 pipeline plugin interface,
+// encodes data of v1 model into bytes.
+//
+// drivers: sls, influxdb, ...
+type EncoderV1 interface {
+ EncodeV1(*protocol.LogGroup) ([][]byte, error)
+ EncodeBatchV1([]*protocol.LogGroup) ([][]byte, error)
+}
+
+// EncoderV2 supports v2 pipeline plugin interface,
+// encodes data of v2 model into bytes.
+//
+// drivers: raw, influxdb, prometheus, ...
+type EncoderV2 interface {
+ EncodeV2(*models.PipelineGroupEvents) ([][]byte, error)
+ EncodeBatchV2([]*models.PipelineGroupEvents) ([][]byte, error)
+}
+
+type EncoderExtension interface {
+ Encoder
+ pipeline.Extension
+}

pkg/protocol/encoder/common/common.go

@@ -0,0 +1,19 @@
+// Copyright 2024 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 common
+
+const (
+ ProtocolPrometheus = "prometheus"
+)

pkg/protocol/encoder/encoder.go

@@ -0,0 +1,39 @@
+// Copyright 2024 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 encoder
+
+import (
+ "fmt"
+ "strings"
+
+ "github.com/mitchellh/mapstructure"
+
+ "github.com/alibaba/ilogtail/pkg/pipeline/extensions"
+ "github.com/alibaba/ilogtail/pkg/protocol/encoder/prometheus"
+)
+
+func NewEncoder(format string, options map[string]any) (extensions.Encoder, error) {
+ switch strings.TrimSpace(strings.ToLower(format)) {
+ case ProtocolPrometheus:
+ var opt prometheus.Option
+ if err := mapstructure.Decode(options, &opt); err != nil {
+ return nil, err
+ }
+ return prometheus.NewPromEncoder(opt.SeriesLimit), nil
+
+ default:
+ return nil, fmt.Errorf("not supported encode format: %s", format)
+ }
+}

pkg/protocol/encoder/prometheus/encoder_prometheus.go

@@ -0,0 +1,126 @@
+// Copyright 2024 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 prometheus
+
+import (
+ "context"
+ "errors"
+
+ "github.com/alibaba/ilogtail/pkg/logger"
+ "github.com/alibaba/ilogtail/pkg/models"
+ "github.com/alibaba/ilogtail/pkg/pipeline/extensions"
+ "github.com/alibaba/ilogtail/pkg/protocol"
+)
+
+const defaultSeriesLimit = 1000
+
+var errNilOrZeroGroupEvents = errors.New("nil or zero group events")
+
+type Option struct {
+ SeriesLimit int // config for prometheus encoder
+}
+
+func NewPromEncoder(seriesLimit int) extensions.Encoder {
+ return newPromEncoder(seriesLimit)
+}
+
+type Encoder struct {
+ SeriesLimit int
+}
+
+func newPromEncoder(seriesLimit int) *Encoder {
+ if seriesLimit <= 0 {
+ seriesLimit = defaultSeriesLimit
+ }
+
+ return &Encoder{
+ SeriesLimit: seriesLimit,
+ }
+}
+
+func (p *Encoder) EncodeV1(logGroups *protocol.LogGroup) ([][]byte, error) {
+ // TODO implement me
+ return nil, nil
+}
+
+func (p *Encoder) EncodeBatchV1(logGroups []*protocol.LogGroup) ([][]byte, error) {
+ // TODO implement me
+ return nil, nil
+}
+
+func (p *Encoder) EncodeV2(groupEvents *models.PipelineGroupEvents) ([][]byte, error) {
+ if groupEvents == nil || len(groupEvents.Events) == 0 {
+ return nil, errNilOrZeroGroupEvents
+ }
+
+ var res [][]byte
+
+ wr := getWriteRequest(p.SeriesLimit)
+ defer putWriteRequest(wr)
+
+ for _, event := range groupEvents.Events {
+ if event == nil {
+ logger.Debugf(context.Background(), "nil event")
+ continue
+ }
+
+ if event.GetType() != models.EventTypeMetric {
+ logger.Debugf(context.Background(), "event type (%s) not metric", event.GetName())
+ continue
+ }
+
+ metricEvent, ok := event.(*models.Metric)
+ if !ok {
+ logger.Debugf(context.Background(), "assert metric event type (%s) failed", event.GetName())
+ continue
+ }
+
+ wr.Timeseries = append(wr.Timeseries, genPromRemoteWriteTimeseries(metricEvent))
+ if len(wr.Timeseries) >= p.SeriesLimit {
+ res = append(res, marshalBatchTimeseriesData(wr))
+ wr.Timeseries = wr.Timeseries[:0]
+ }
+ }
+
+ if len(wr.Timeseries) > 0 {
+ res = append(res, marshalBatchTimeseriesData(wr))
+ wr.Timeseries = wr.Timeseries[:0]
+ }
+
+ return res, nil
+}
+
+func (p *Encoder) EncodeBatchV2(groupEventsSlice []*models.PipelineGroupEvents) ([][]byte, error) {
+ if groupEventsSlice == nil || len(groupEventsSlice) == 0 {
+ return nil, errNilOrZeroGroupEvents
+ }
+
+ var res [][]byte
+
+ for _, groupEvents := range groupEventsSlice {
+ bytes, err := p.EncodeV2(groupEvents)
+ if err != nil {
+ continue
+ }
+
+ res = append(res, bytes...)
+ }
+
+ if res == nil {
+ return nil, errNilOrZeroGroupEvents
+ }
+
+ return res, nil
+}