feat: add ext_default_encoder with prometheus encoder

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.4
 	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/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
+}