Support eventing metrics

pkg/channel/consolidated/dispatcher/consumer_message_handler.go

@@ -23,10 +23,13 @@ import (
  "github.com/Shopify/sarama"
  protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
  "github.com/cloudevents/sdk-go/v2/binding"
+ "github.com/cloudevents/sdk-go/v2/binding/buffering"
  "go.uber.org/zap"
  "knative.dev/eventing-kafka/pkg/common/consumer"
  "knative.dev/eventing-kafka/pkg/common/tracing"
  eventingchannels "knative.dev/eventing/pkg/channel"
+ fanout "knative.dev/eventing/pkg/channel/fanout"
+ "knative.dev/eventing/pkg/kncloudevents"
 )
 
 type consumerMessageHandler struct {
@@ -35,6 +38,8 @@ type consumerMessageHandler struct {
  dispatcher *eventingchannels.MessageDispatcherImpl
  kafkaSubscription *KafkaSubscription
  consumerGroup string
+ reporter eventingchannels.StatsReporter
+ channelNs string
 }
 
 var _ consumer.KafkaConsumerHandler = (*consumerMessageHandler)(nil)
@@ -69,16 +74,33 @@ func (c consumerMessageHandler) Handle(ctx context.Context, consumerMessage *sar
  ctx, span := tracing.StartTraceFromMessage(c.logger, ctx, message, consumerMessage.Topic)
  defer span.End()
 
- _, err := c.dispatcher.DispatchMessageWithRetries(
+ te := kncloudevents.TypeExtractorTransformer("")
+
+ bufferedMessage, err := buffering.CopyMessage(ctx, message, &te)
+
+ if err != nil {
+ return false, err
+ }
+
+ args := eventingchannels.ReportArgs{
+ Ns: c.channelNs,
+ EventType: string(te),
+ }
+
+ _ = message.Finish(nil)
+
+ dispatchExecutionInfo, err := c.dispatcher.DispatchMessageWithRetries(
  ctx,
- message,
+ bufferedMessage,
  nil,
  c.sub.Subscriber,
  c.sub.Reply,
  c.sub.DeadLetter,
  c.sub.RetryConfig,
  )
 
+ _ = fanout.ParseDispatchResultAndReportMetrics(fanout.NewDispatchResult(err, dispatchExecutionInfo), c.reporter, args)
+
  // NOTE: only return `true` here if DispatchMessage actually delivered the message.
  return err == nil, err
 }

pkg/channel/consolidated/dispatcher/dispatcher.go

@@ -18,12 +18,9 @@ package dispatcher
 import (
  "context"
  "fmt"
- nethttp "net/http"
  "strings"
  "sync"
 
- "knative.dev/eventing-kafka/pkg/common/config"
-
  "github.com/Shopify/sarama"
  protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
  "github.com/cloudevents/sdk-go/v2/binding"
@@ -32,11 +29,14 @@ import (
  "go.uber.org/zap"
  "k8s.io/apimachinery/pkg/types"
  "k8s.io/apimachinery/pkg/util/sets"
+ "knative.dev/eventing-kafka/pkg/common/config"
  "knative.dev/pkg/logging"
 
  eventingchannels "knative.dev/eventing/pkg/channel"
  "knative.dev/pkg/kmeta"
 
+ nethttp "net/http"
+
  "knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
  "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
  "knative.dev/eventing-kafka/pkg/common/consumer"
@@ -58,6 +58,7 @@ type KafkaDispatcherArgs struct {
 type KafkaDispatcher struct {
  receiver *eventingchannels.MessageReceiver
  dispatcher *eventingchannels.MessageDispatcherImpl
+ reporter eventingchannels.StatsReporter
 
  // Receiver data structures
  // map[string]eventingchannels.ChannelReference
@@ -112,6 +113,7 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
  return nil, err
  }
  reporter := eventingchannels.NewStatsReporter(containerName, kmeta.ChildName(podName, uuid.New().String()))
+ dispatcher.reporter = reporter
  receiverFunc, err := eventingchannels.NewMessageReceiver(
  func(ctx context.Context, channel eventingchannels.ChannelReference, message binding.Message, transformers []binding.Transformer, _ nethttp.Header) error {
  kafkaProducerMessage := sarama.ProducerMessage{
@@ -303,6 +305,8 @@ func (d *KafkaDispatcher) subscribe(channelRef types.NamespacedName, sub Subscri
  d.dispatcher,
  kafkaSubscription,
  groupID,
+ d.reporter,
+ channelRef.Namespace,
  }
  d.logger.Debugw("Starting consumer group", zap.Any("channelRef", channelRef),
  zap.Any("subscription", sub.UID), zap.String("topic", topicName), zap.String("consumer group", groupID))