kafka

README

kafka

Usage

source:
  name: kafka
  config:
    broker: "localhost:9092"
    auth_config:
      tls:
        enabled: true
        insecure_skip_verify: false
        cert_file: "/opt/client.cer.pem"
        key_file: "/opt/client.key.pem"
        ca_file: "/opt/caCertFile.cer.pem"

Inputs

Key	Value	Example	Description
broker	string	localhost:9092	Kafka broker's host	required
auth_config.tls.enabled	boolean	false	config to enable tls auth	optional
auth_config.tls.insecure_skip_verify	boolean	false	InsecureSkipVerify controls whether a client verifies the server's certificate chain and host name	optional
auth_config.tls.cert_file	string	/opt/client.cer.pem	certificate file for client authentication	optional
auth_config.tls.key_file	string	/opt/client.key.pem	key file for client authentication	optional
auth_config.tls.ca_file	string	/opt/caCertFile.cer.pem	certificate authority file for TLS client authentication	optional

Outputs

Field	Sample Value
resource.urn	my-topic
resource.name	my-topic
resource.service	kafka

Contributing

Refer to the contribution guidelines for information on contributing to this module.

Documentation

Index

• type AuthConfig
• type Config
• type Extractor
  • func New(logger log.Logger) *Extractor
  • func (e *Extractor) Extract(ctx context.Context, emit plugins.Emit) (err error)
  • func (e *Extractor) Init(ctx context.Context, config plugins.Config) error
• type KubernetesTokenProvider
  • func NewKubernetesTokenProvider(opts ...TokenProviderOption) *KubernetesTokenProvider
  • func (tp *KubernetesTokenProvider) Token() (*sarama.AccessToken, error)
• type TokenProviderOption
  • func WithTokenFilePath(path string) TokenProviderOption
• type TokenProviderOptions

Types

type AuthConfig

type AuthConfig struct {
	TLS struct {
		// Whether to use TLS when connecting to the broker
		// (defaults to false).
		Enabled bool `mapstructure:"enabled"`

		// controls whether a client verifies the server's certificate chain and host name
		// defaults to false
		InsecureSkipVerify bool `mapstructure:"insecure_skip_verify"`

		// certificate file for client authentication
		CertFile string `mapstructure:"cert_file"`

		// key file for client authentication
		KeyFile string `mapstructure:"key_file"`

		// certificate authority file for TLS client authentication
		CAFile string `mapstructure:"ca_file"`
	} `mapstructure:"tls"`

	SASL struct {
		Enabled   bool   `mapstructure:"enabled"`
		Mechanism string `mapstructure:"mechanism"`
	}
}

type Config

type Config struct {
	Broker string     `mapstructure:"broker" validate:"required"`
	Auth   AuthConfig `mapstructure:"auth_config"`
}

Config holds the set of configuration for the kafka extractor

type Extractor

type Extractor struct {
	plugins.BaseExtractor
	// contains filtered or unexported fields
}

Extractor manages the extraction of data from a kafka broker

func New

func New(logger log.Logger) *Extractor

New returns a pointer to an initialized Extractor Object

func (*Extractor) Extract

func (e *Extractor) Extract(ctx context.Context, emit plugins.Emit) (err error)

Extract checks if the extractor is ready to extract if so, then extracts metadata from the kafka broker

func (*Extractor) Init

func (e *Extractor) Init(ctx context.Context, config plugins.Config) error

Init initializes the extractor

type KubernetesTokenProvider

type KubernetesTokenProvider struct {
	// contains filtered or unexported fields
}

func NewKubernetesTokenProvider

func NewKubernetesTokenProvider(opts ...TokenProviderOption) *KubernetesTokenProvider

NewKubernetesTokenProvider creates a new TokenProvider that reads the token from kubernetes pod service account token file. By default, the token file path for kafka is stored in `/var/run/secrets/kafka/serviceaccount/token`. User need to make sure there a valid projected service account token on that path.

func (*KubernetesTokenProvider) Token

func (tp *KubernetesTokenProvider) Token() (*sarama.AccessToken, error)

Token returns the token from the service account token file.

type TokenProviderOption

type TokenProviderOption func(*TokenProviderOptions)

func WithTokenFilePath

func WithTokenFilePath(path string) TokenProviderOption

WithTokenFilePath sets the file path to the token.

type TokenProviderOptions

type TokenProviderOptions struct {
	// FilePath is the path to the file containing the token.
	FilePath string
}