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) error
  • func (e *Extractor) Init(ctx context.Context, config plugins.Config) error

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"`
}

type Config

type Config struct {
	Broker string     `json:"broker" yaml:"broker" mapstructure:"broker" validate:"required"`
	Auth   AuthConfig `json:"auth_config" yaml:"auth_config" 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) 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