fix: revise kafka sink conf/metrics

extensions/impl/kafka/metrics.go

@@ -21,9 +21,8 @@ import (
 )
 
 const (
-	LblRequest   = "req"
+	LblWriteMsgs = "write"
 	LblMessage   = "message"
-	LblException = "exception"
 )
 
 var (
@@ -32,12 +31,18 @@ var (
 		Subsystem: "io",
 		Name:      "kafka_count",
 		Help:      "counter of Kafka IO",
-	}, []string{metrics.LblType, metrics.LblIOType, metrics.LblRuleIDType, metrics.LblOpIDType})
+	}, []string{metrics.LblType, metrics.LblIOType, metrics.LblStatusType, metrics.LblRuleIDType, metrics.LblOpIDType})
 
-	KafkaHist = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+	KafkaDurationHist = prometheus.NewHistogramVec(prometheus.HistogramOpts{
 		Namespace: "kuiper",
 		Subsystem: "io",
 		Name:      "kafka_duration",
 		Help:      "Historgram of Kafka IO",
+		Buckets:   prometheus.ExponentialBuckets(10, 2, 20), // 10us ~ 5s
 	}, []string{metrics.LblType, metrics.LblIOType, metrics.LblRuleIDType, metrics.LblOpIDType})
 )
+
+func init() {
+	prometheus.MustRegister(KafkaCounter)
+	prometheus.MustRegister(KafkaDurationHist)
+}

extensions/impl/kafka/sink.go

@@ -26,6 +26,7 @@
 	kafkago "github.com/segmentio/kafka-go"
 	"github.com/segmentio/kafka-go/sasl"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/util"
 	"github.com/lf-edge/ekuiper/v2/metrics"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
@@ -40,9 +41,11 @@
 	headerTemplate string
 	saslConf       *saslConf
 	mechanism      sasl.Mechanism
+	LastStats      kafkago.WriterStats
 }
 
 type kafkaConf struct {
+	kafkaWriterConf
 	Brokers      string      `json:"brokers"`
 	Topic        string      `json:"topic"`
 	MaxAttempts  int         `json:"maxAttempts"`
@@ -54,6 +57,12 @@
 	Compression string `json:"compression"`
 }
 
+type kafkaWriterConf struct {
+	BatchSize    int           `json:"batchSize"`
+	BatchTimeout time.Duration `json:"-"`
+	BatchBytes   int64         `json:"batchBytes"`
+}
+
 func (c *kafkaConf) validate() error {
 	if c.Topic == "" {
 		return fmt.Errorf("topic can not be empty")
@@ -65,11 +74,8 @@
 }
 
 func (k *KafkaSink) Provision(ctx api.StreamContext, configs map[string]any) error {
-	c := &kafkaConf{
-		RequiredACKs: -1,
-		MaxAttempts:  1,
-	}
-	err := cast.MapToStruct(configs, c)
+	c := getDefaultKafkaConf()
+	err := c.configure(configs)
 	failpoint.Inject("kafkaErr", func(val failpoint.Value) {
 		err = mockKakfaSourceErr(val.(int), castConfErr)
 	})
@@ -149,7 +155,9 @@
 		AllowAutoTopicCreation: true,
 		MaxAttempts:            k.kc.MaxAttempts,
 		RequiredAcks:           kafkago.RequiredAcks(k.kc.RequiredACKs),
-		BatchSize:              1,
+		BatchSize:              k.kc.BatchSize,
+		BatchBytes:             k.kc.BatchBytes,
+		BatchTimeout:           k.kc.BatchTimeout,
 		Transport: &kafkago.Transport{
 			SASL: k.mechanism,
 			TLS:  k.tlsConfig,
@@ -176,29 +184,20 @@
 
 func (k *KafkaSink) Collect(ctx api.StreamContext, item api.MessageTuple) (err error) {
 	defer func() {
-		if err != nil {
-			KafkaCounter.WithLabelValues(LblException, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Inc()
-		}
+		KafkaCounter.WithLabelValues(LblMessage, metrics.LblSinkIO, metrics.GetStatusValue(err), ctx.GetRuleId(), ctx.GetOpId()).Inc()
 	}()
 	msgs, err := k.collect(ctx, item)
 	if err != nil {
 		return err
 	}
-	KafkaCounter.WithLabelValues(LblRequest, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Inc()
-	KafkaCounter.WithLabelValues(LblMessage, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Add(float64(len(msgs)))
 	start := time.Now()
 	defer func() {
-		KafkaHist.WithLabelValues(LblRequest, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
+		KafkaDurationHist.WithLabelValues(LblWriteMsgs, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
 	}()
 	return k.writer.WriteMessages(ctx, msgs...)
 }
 
 func (k *KafkaSink) CollectList(ctx api.StreamContext, items api.MessageTupleList) (err error) {
-	defer func() {
-		if err != nil {
-			KafkaCounter.WithLabelValues(LblException, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Inc()
-		}
-	}()
 	allMsgs := make([]kafkago.Message, 0)
 	items.RangeOfTuples(func(index int, tuple api.MessageTuple) bool {
 		msgs, err := k.collect(ctx, tuple)
@@ -208,13 +207,14 @@
 		allMsgs = append(allMsgs, msgs...)
 		return true
 	})
-	KafkaCounter.WithLabelValues(LblMessage, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Add(float64(len(allMsgs)))
-	KafkaCounter.WithLabelValues(LblRequest, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Inc()
 	start := time.Now()
 	defer func() {
-		KafkaHist.WithLabelValues(LblRequest, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
+		conf.Log.Infof("send kafka cost %v", time.Since(start).String())
+		KafkaDurationHist.WithLabelValues(LblWriteMsgs, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
 	}()
-	return k.writer.WriteMessages(ctx, allMsgs...)
+	err = k.writer.WriteMessages(ctx, allMsgs...)
+	KafkaCounter.WithLabelValues(LblMessage, metrics.LblSinkIO, metrics.GetStatusValue(err), ctx.GetRuleId(), ctx.GetOpId()).Add(float64(len(allMsgs)))
+	return err
 }
 
 func (k *KafkaSink) collect(ctx api.StreamContext, item api.MessageTuple) ([]kafkago.Message, error) {
@@ -338,3 +338,26 @@
 	_ api.TupleCollector = &KafkaSink{}
 	_ util.PingableConn  = &KafkaSink{}
 )
+
+func getDefaultKafkaConf() *kafkaConf {
+	c := &kafkaConf{
+		RequiredACKs: -1,
+		MaxAttempts:  1,
+	}
+	c.kafkaWriterConf = kafkaWriterConf{
+		BatchSize:    100,
+		BatchTimeout: time.Microsecond,
+		BatchBytes:   10485760, // 10MB
+	}
+	return c
+}
+
+func (kc *kafkaConf) configure(props map[string]interface{}) error {
+	if err := cast.MapToStruct(props, kc); err != nil {
+		return err
+	}
+	if err := cast.MapToStruct(props, &kc.kafkaWriterConf); err != nil {
+		return err
+	}
+	return nil
+}

extensions/impl/kafka/source.go

@@ -194,7 +194,7 @@
 			ingestError(ctx, err)
 			continue
 		}
-		KafkaCounter.WithLabelValues(LblMessage, metrics.LblSourceIO, ctx.GetRuleId(), ctx.GetOpId()).Inc()
+		KafkaCounter.WithLabelValues(LblMessage, metrics.LblSourceIO, metrics.LblSuccess, ctx.GetRuleId(), ctx.GetOpId()).Inc()
 		ingest(ctx, msg.Value, nil, timex.GetNow())
 	}
 }

extensions/impl/sql/metrics.go

@@ -34,10 +34,15 @@ var (
 		Help:      "counter of SQL IO",
 	}, []string{metrics.LblType, metrics.LblIOType, metrics.LblRuleIDType, metrics.LblOpIDType})
 
-	SQLHist = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+	SQLDurationHist = prometheus.NewHistogramVec(prometheus.HistogramOpts{
 		Namespace: "kuiper",
 		Subsystem: "io",
 		Name:      "sql_duration",
 		Help:      "Historgram of Kafka IO",
 	}, []string{metrics.LblType, metrics.LblIOType, metrics.LblRuleIDType, metrics.LblOpIDType})
 )
+
+func init() {
+	prometheus.MustRegister(SQLCounter)
+	prometheus.MustRegister(SQLDurationHist)
+}

extensions/impl/sql/sink.go

@@ -298,7 +298,7 @@ func (s *SQLSinkConnector) writeToDB(ctx api.StreamContext, sqlStr string) error
 		s.needReconnect = true
 		return errorx.NewIOErr(err.Error())
 	}
-	SQLHist.WithLabelValues(LblRequest, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
+	SQLDurationHist.WithLabelValues(LblRequest, metrics.LblSinkIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
 	s.needReconnect = false
 	d, err := r.RowsAffected()
 	if err != nil {

extensions/impl/sql/source.go

@@ -160,7 +160,7 @@ func (s *SQLSourceConnector) queryData(ctx api.StreamContext, rcvTime time.Time,
 	failpoint.Inject("QueryErr", func() {
 		err = errors.New("QueryErr")
 	})
-	SQLHist.WithLabelValues(LblRequest, metrics.LblSourceIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
+	SQLDurationHist.WithLabelValues(LblRequest, metrics.LblSourceIO, ctx.GetRuleId(), ctx.GetOpId()).Observe(float64(time.Since(start).Microseconds()))
 	if err != nil {
 		logger.Errorf("query sql error %v", err)
 		s.needReconnect = true

internal/topo/planner/planner_sink.go

@@ -170,7 +170,6 @@ func findTemplateProps(props map[string]any) []string {
 func splitSink(tp *topo.Topo, s api.Sink, sinkName string, options *def.RuleOption, sc *node.SinkConf, templates []string) ([]node.TopNode, error) {
 	index := 0
 	result := make([]node.TopNode, 0)
-	// Batch enabled
 	if sc.BatchSize > 0 || sc.LingerInterval > 0 {
 		batchOp, err := node.NewBatchOp(fmt.Sprintf("%s_%d_batch", sinkName, index), options, sc.BatchSize, time.Duration(sc.LingerInterval))
 		if err != nil {

metrics/metrics.go

@@ -27,8 +27,17 @@
 	LblRuleStop    = "stop"
 	LblSourceIO    = "source"
 	LblSinkIO      = "sink"
+	LblException   = "err"
+	LblSuccess     = "success"
 )
 
+func GetStatusValue(err error) string {
+	if err == nil {
+		return LblSuccess
+	}
+	return LblException
+}
+
 var (
 	RuleStatusCountGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: "kuiper",