kafka

Documentation

Index

• Variables
• type Consumer
  • func NewConsumer(ctx context.Context, logger logr.Logger, cli client.Client, brokers []string, ...) (Consumer, error)

Variables

var (
	// KafkaMessageGeneratedDiagnosisPrefix is the name prefix for diagnoses generated by kafka message.
	KafkaMessageGeneratedDiagnosisPrefix = "kafka-message"
	// KafkaMessageTopicAnnotation is the annotation used to store the kafka message topic.
	KafkaMessageTopicAnnotation = util.KubeDiagPrefix + "kafka-message-topic"
	// KafkaMessagePartitionAnnotation is the annotation used to store the kafka message partition.
	KafkaMessagePartitionAnnotation = util.KubeDiagPrefix + "kafka-message-partition"
	// KafkaMessageOffsetAnnotation is the annotation used to store the kafka message offset.
	KafkaMessageOffsetAnnotation = util.KubeDiagPrefix + "kafka-message-offset"
	// KafkaMessageKeyAnnotation is the annotation used to store the kafka message key.
	KafkaMessageKeyAnnotation = util.KubeDiagPrefix + "kafka-message-key"
	// KafkaMessageValueAnnotation is the annotation used to store the kafka message value.
	KafkaMessageValueAnnotation = util.KubeDiagPrefix + "kafka-message-value"
	// KafkaMessageHeadersAnnotation is the annotation used to store the kafka message headers.
	KafkaMessageHeadersAnnotation = util.KubeDiagPrefix + "kafka-message-headers"
	// KafkaMessageTimeAnnotation is the annotation used to store the kafka message time.
	KafkaMessageTimeAnnotation = util.KubeDiagPrefix + "kafka-message-time"
	// KafkaMessageOperationSetKey is the key to specify operationset in kafka messages.
	KafkaMessageOperationSetKey = "operationset"
	// KafkaMessageNodeNameKey is the key to specify node in kafka messages.
	KafkaMessageNodeNameKey = "node"
	// KafkaMessagePodNameKey is the key to specify pod name in kafka messages.
	KafkaMessagePodNameKey = "pod"
	// KafkaMessagePodNamespaceKey is the key to specify pod namespace in kafka messages.
	KafkaMessagePodNamespaceKey = "namespace"
	// KafkaMessageContainerKey is the key to specify container in kafka messages.
	KafkaMessageContainerKey = "container"
)

Types

type Consumer

type Consumer interface {
	// Run runs the Consumer.
	Run(<-chan struct{})
}

Consumer consumes kafka messages and create diagnoses from messages.

func NewConsumer

func NewConsumer(
	ctx context.Context,
	logger logr.Logger,
	cli client.Client,
	brokers []string,
	topic string,
	kafkaConsumerEnabled bool,
) (Consumer, error)

NewConsumer creates a new Consumer.