kafkatarget

Documentation

Index

• Constants
• Variables
• func LabelsToMetric(ls labels.Labels) model.Metric
• func NewOAuthProvider(opts OAuthConfig) (sarama.AccessTokenProvider, error)
• func StringsContain(values []string, search string) bool
• type Authentication
• type AuthenticationType
• type Config
• type ConsumerDetails
  • func (c ConsumerDetails) String() string
• type KafkaTarget
  • func NewKafkaTarget(logger log.Logger, session sarama.ConsumerGroupSession, ...) *KafkaTarget
  • func (t *KafkaTarget) Details() interface{}
  • func (t *KafkaTarget) DiscoveredLabels() model.LabelSet
  • func (t *KafkaTarget) Labels() model.LabelSet
  • func (t *KafkaTarget) Ready() bool
  • func (t *KafkaTarget) Type() target.TargetType
• type KafkaTargetMessageParser
  • func (p *KafkaTargetMessageParser) Parse(message *sarama.ConsumerMessage, labels model.LabelSet, ...) ([]loki.Entry, error)
• type MessageParser
• type OAuthConfig
• type RunnableTarget
• type SASLConfig
• type TargetConfig
• type TargetDiscoverer
• type TargetSyncer
  • func NewSyncer(reg prometheus.Registerer, logger log.Logger, cfg Config, ...) (*TargetSyncer, error)
  • func (c *TargetSyncer) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *TargetSyncer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (ts *TargetSyncer) NewTarget(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) (RunnableTarget, error)
  • func (c *TargetSyncer) Setup(session sarama.ConsumerGroupSession) error
  • func (ts *TargetSyncer) Stop() error
• type TokenProviderAzure
  • func (t *TokenProviderAzure) Token() (*sarama.AccessToken, error)
• type TokenProviderType
• type TopicManager
• type XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Constants

const (
	// AuthenticationTypeNone represents using no authentication
	AuthenticationTypeNone = "none"
	// AuthenticationTypeSSL represents using SSL/TLS to authenticate
	AuthenticationTypeSSL = "ssl"
	// AuthenticationTypeSASL represents using SASL to authenticate
	AuthenticationTypeSASL = "sasl"
)

Variables

var (
	SHA256 scram.HashGeneratorFcn = sha256.New
	SHA512 scram.HashGeneratorFcn = sha512.New
)

copied from https://github.com/Shopify/sarama/blob/44627b731c60bb90efe25573e7ef2b3f8df3fa23/examples/sasl_scram_client/scram_client.go

var TopicPollInterval = 30 * time.Second

Functions

func LabelsToMetric

func LabelsToMetric(ls labels.Labels) model.Metric

LabelsToMetric converts a Labels to Metric Don't do this on any performance sensitive paths.

func NewOAuthProvider

func NewOAuthProvider(opts OAuthConfig) (sarama.AccessTokenProvider, error)

func StringsContain

func StringsContain(values []string, search string) bool

StringsContain returns true if the search value is within the list of input values.

Types

type Authentication

type Authentication struct {
	// Type is authentication type
	// Possible values: none, sasl and ssl (defaults to none).
	Type AuthenticationType `yaml:"type"`

	// TLSConfig is used for TLS encryption and authentication with Kafka brokers
	TLSConfig promconfig.TLSConfig `yaml:"tls_config,omitempty"`

	// SASLConfig is used for SASL authentication with Kafka brokers
	SASLConfig SASLConfig `yaml:"sasl_config,omitempty"`
}

Authentication describe the configuration for authentication with Kafka brokers

type AuthenticationType

type AuthenticationType string

AuthenticationType specifies method to authenticate with Kafka brokers

type Config

type Config struct {
	KafkaConfig    TargetConfig      `mapstructure:"kafka,omitempty" yaml:"kafka,omitempty"`
	RelabelConfigs []*relabel.Config `mapstructure:"relabel_configs,omitempty" yaml:"relabel_configs,omitempty"`
}

Config describes a job to scrape.

type ConsumerDetails

type ConsumerDetails struct {

	// MemberID returns the cluster member ID.
	MemberID string

	// GenerationID returns the current generation ID.
	GenerationID int32

	Topic         string
	Partition     int32
	InitialOffset int64
}

func (ConsumerDetails) String

func (c ConsumerDetails) String() string

type KafkaTarget

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

func NewKafkaTarget

func NewKafkaTarget(
	logger log.Logger,
	session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim,
	discoveredLabels, lbs model.LabelSet,
	relabelConfig []*relabel.Config,
	client loki.EntryHandler,
	useIncomingTimestamp bool,
	messageParser MessageParser,
) *KafkaTarget

func (*KafkaTarget) Details

func (t *KafkaTarget) Details() interface{}

Details returns target-specific details.

func (*KafkaTarget) DiscoveredLabels

func (t *KafkaTarget) DiscoveredLabels() model.LabelSet

func (*KafkaTarget) Labels

func (t *KafkaTarget) Labels() model.LabelSet

func (*KafkaTarget) Ready

func (t *KafkaTarget) Ready() bool

func (*KafkaTarget) Type

func (t *KafkaTarget) Type() target.TargetType

type KafkaTargetMessageParser

type KafkaTargetMessageParser struct{}

KafkaTargetMessageParser implements MessageParser. It doesn't modify the content of the original `message.Value`.

func (*KafkaTargetMessageParser) Parse

func (p *KafkaTargetMessageParser) Parse(message *sarama.ConsumerMessage, labels model.LabelSet, relabels []*relabel.Config, useIncomingTimestamp bool) ([]loki.Entry, error)

type MessageParser

type MessageParser interface {
	Parse(message *sarama.ConsumerMessage, labels model.LabelSet, relabels []*relabel.Config, useIncomingTimestamp bool) ([]loki.Entry, error)
}

MessageParser defines parsing for each incoming message

type OAuthConfig

type OAuthConfig struct {
	// TokenProvider is used for resolving the OAuth access token
	TokenProvider TokenProviderType `yaml:"token_provider,omitempty"`

	Scopes []string
}

type RunnableTarget

type RunnableTarget interface {
	target.Target
	// contains filtered or unexported methods
}

type SASLConfig

type SASLConfig struct {
	// SASL mechanism. Supports PLAIN, SCRAM-SHA-256 and SCRAM-SHA-512
	Mechanism sarama.SASLMechanism `yaml:"mechanism"`

	// SASL Username
	User string `yaml:"user"`

	// SASL Password for the User
	Password flagext.Secret `yaml:"password"`

	// UseTLS sets whether TLS is used with SASL
	UseTLS bool `yaml:"use_tls"`

	// TLSConfig is used for SASL over TLS. It is used only when UseTLS is true
	TLSConfig promconfig.TLSConfig `yaml:",inline"`

	// OAuthConfig is used for configuring the token provider
	OAuthConfig OAuthConfig `yaml:"oauth_provider_config,omitempty"`
}

KafkaSASLConfig describe the SASL configuration for authentication with Kafka brokers

type TargetConfig

type TargetConfig struct {
	// Labels optionally holds labels to associate with each log line.
	Labels model.LabelSet `yaml:"labels"`

	// UseIncomingTimestamp sets the timestamp to the incoming kafka messages
	// timestamp if it's set.
	UseIncomingTimestamp bool `yaml:"use_incoming_timestamp"`

	// The list of brokers to connect to kafka (Required).
	Brokers []string `yaml:"brokers"`

	// The consumer group id (Required).
	GroupID string `yaml:"group_id"`

	// Kafka Topics to consume (Required).
	Topics []string `yaml:"topics"`

	// Kafka version. Default to 2.2.1
	Version string `yaml:"version"`

	// Rebalancing strategy to use. (e.g. sticky, roundrobin or range)
	Assignor string `yaml:"assignor"`

	// Authentication strategy with Kafka brokers
	Authentication Authentication `yaml:"authentication"`

	MessageParser MessageParser
}

type TargetDiscoverer

type TargetDiscoverer interface {
	NewTarget(sarama.ConsumerGroupSession, sarama.ConsumerGroupClaim) (RunnableTarget, error)
}

type TargetSyncer

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

func NewSyncer

func NewSyncer(
	reg prometheus.Registerer,
	logger log.Logger,
	cfg Config,
	pushClient loki.EntryHandler,
	messageParser MessageParser,
) (*TargetSyncer, error)

func (*TargetSyncer) Cleanup

func (c *TargetSyncer) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*TargetSyncer) ConsumeClaim

func (c *TargetSyncer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim creates a target for the given received claim and start reading message from it.

func (*TargetSyncer) NewTarget

func (ts *TargetSyncer) NewTarget(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) (RunnableTarget, error)

NewTarget creates a new targets based on the current kafka claim and group session.

func (*TargetSyncer) Setup

func (c *TargetSyncer) Setup(session sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

func (*TargetSyncer) Stop

func (ts *TargetSyncer) Stop() error

type TokenProviderAzure

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

TokenProviderAzure implements sarama.AccessTokenProvider

func (*TokenProviderAzure) Token

func (t *TokenProviderAzure) Token() (*sarama.AccessToken, error)

Token returns a new *sarama.AccessToken or an error

type TokenProviderType

type TokenProviderType string

TokenProviderType specifies the provider used for resolving the access token

const (
	// TokenProviderTypeAzure represents using the Azure as the token provider
	TokenProviderTypeAzure TokenProviderType = "azure"
)

type TopicManager

type TopicManager interface {
	Topics() ([]string, error)
}

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

XDGSCRAMClient implements sarama.SCRAMClient

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)