fkpulsarreceiver

README

Pulsar Receiver

Status
Stability	alpha: metrics, traces, logs
Distributions	contrib
Issues

Pulsar receiver receives logs, metrics, and traces from Pulsar.

Getting Started

The following settings can be optionally configured:

• endpoint (default = pulsar://localhost:6650): The url of pulsar cluster.
• topic (default = otlp_spans for traces, otlp_metrics for metrics, otlp_logs for logs): The name of the pulsar topic to consume from.
• encoding (default = otlp_proto): The encoding of the payload sent to pulsar. Available encodings:
  • otlp_proto: the payload is deserialized to ExportTraceServiceRequest.
  • jaeger_proto: the payload is deserialized to a single Jaeger proto Span.
  • jaeger_json: the payload is deserialized to a single Jaeger JSON Span using jsonpb.
  • zipkin_proto: the payload is deserialized into a list of Zipkin proto spans.
  • zipkin_json: the payload is deserialized into a list of Zipkin V2 JSON spans.
  • zipkin_thrift: the payload is deserialized into a list of Zipkin Thrift spans.
• consumer_name: specifies the consumer name.
• auth
  • tls
    • cert_file:
    • key_file:
  • token
    • token
  • oauth2
    • issuer_url:
    • client_id:
    • audience:
  • athenz
    • provider_domain:
    • tenant_domain:
    • tenant_service:
    • private_key:
    • key_id:
    • principal_header:
    • zts_url:
• subscription (default = otlp_subscription): the subscription name of consumer.
• tls_trust_certs_file_path: path to the CA cert. For a client this verifies the server certificate. Should only be used if insecure is set to true.
• tls_allow_insecure_connection: configure whether the Pulsar client accept untrusted TLS certificate from broker (default: false)

Example configuration:

receivers:
  pulsar:
    endpoint: pulsar://localhost:6650
    topic: otlp-spans
    subscription: otlp_spans_sub
    consumer_name: otlp_spans_sub_1
    encoding: otlp_proto
    auth:
      tls:
        cert_file: cert.pem
        key_file: key.pem
    tls_allow_insecure_connection: false
    tls_trust_certs_file_path: ca.pem

Documentation

Index

• Constants
• func GeneratePrivateKey(clientId, clientSecret, issuerUrl string) (string, error)
• func NewFactory(options ...FactoryOption) receiver.Factory
• type Athenz
• type Authentication
• type Config
  • func (cfg *Config) Validate() error
• type FactoryOption
  • func WithLogsUnmarshalers(logsUnmarshalers ...LogsUnmarshaler) FactoryOption
  • func WithMetricsUnmarshalers(metricsUnmarshalers ...MetricsUnmarshaler) FactoryOption
  • func WithTracesUnmarshalers(tracesUnmarshalers ...TracesUnmarshaler) FactoryOption
• type LogsUnmarshaler
• type MetricsUnmarshaler
• type OAuth2
• type TLS
• type Token
• type TracesUnmarshaler

Constants

const (
	Type             = "fkpulsar"
	MetricsStability = component.StabilityLevelAlpha
	TracesStability  = component.StabilityLevelAlpha
	LogsStability    = component.StabilityLevelAlpha
)

Functions

func GeneratePrivateKey

func GeneratePrivateKey(clientId, clientSecret, issuerUrl string) (string, error)

func NewFactory

func NewFactory(options ...FactoryOption) receiver.Factory

NewFactory creates Pulsar receiver factory.

Types

type Athenz

type Athenz struct {
	ProviderDomain  string              `mapstructure:"provider_domain"`
	TenantDomain    string              `mapstructure:"tenant_domain"`
	TenantService   string              `mapstructure:"tenant_service"`
	PrivateKey      configopaque.String `mapstructure:"private_key"`
	KeyID           string              `mapstructure:"key_id"`
	PrincipalHeader string              `mapstructure:"principal_header"`
	ZtsURL          string              `mapstructure:"zts_url"`
}

type Authentication

type Authentication struct {
	TLS    *TLS    `mapstructure:"tls"`
	Token  *Token  `mapstructure:"token"`
	Athenz *Athenz `mapstructure:"athenz"`
	OAuth2 *OAuth2 `mapstructure:"oauth2"`
}

type Config

type Config struct {
	// Configure the service URL for the Pulsar service.
	Endpoint string `mapstructure:"endpoint"`
	// The topic of pulsar to consume logs,metrics,traces. (default = "otlp_traces" for traces,
	// "otlp_metrics" for metrics, "otlp_logs" for logs)
	Topic string `mapstructure:"topic"`
	// The Subscription that receiver will be consuming messages from (default "otlp_subscription")
	Subscription string `mapstructure:"subscription"`
	// Encoding of the messages (default "otlp_proto")
	Encoding string `mapstructure:"encoding"`
	// Name specifies the consumer name.
	ConsumerName string `mapstructure:"consumer_name"`
	// Set the path to the trusted TLS certificate file
	TLSTrustCertsFilePath string `mapstructure:"tls_trust_certs_file_path"`
	// Configure whether the Pulsar client accept untrusted TLS certificate from broker (default: false)
	TLSAllowInsecureConnection bool           `mapstructure:"tls_allow_insecure_connection"`
	Authentication             Authentication `mapstructure:"auth"`
}

func (*Config) Validate

func (cfg *Config) Validate() error

Validate checks the receiver configuration is valid

type FactoryOption

type FactoryOption func(factory *pulsarReceiverFactory)

FactoryOption applies changes to PulsarExporterFactory.

func WithLogsUnmarshalers

func WithLogsUnmarshalers(logsUnmarshalers ...LogsUnmarshaler) FactoryOption

WithLogsUnmarshalers adds LogsUnmarshalers.

func WithMetricsUnmarshalers

func WithMetricsUnmarshalers(metricsUnmarshalers ...MetricsUnmarshaler) FactoryOption

WithMetricsUnmarshalers adds MetricsUnmarshalers.

func WithTracesUnmarshalers

func WithTracesUnmarshalers(tracesUnmarshalers ...TracesUnmarshaler) FactoryOption

WithTracesUnmarshalers adds Unmarshalers.

type LogsUnmarshaler

type LogsUnmarshaler interface {
	// Unmarshal deserializes the message body into tracesConsumer.
	Unmarshal([]byte) (plog.Logs, error)

	// Encoding of the serialized messages.
	Encoding() string
}

LogsUnmarshaler deserializes the message body.

type MetricsUnmarshaler

type MetricsUnmarshaler interface {
	// Unmarshal deserializes the message body into tracesConsumer
	Unmarshal([]byte) (pmetric.Metrics, error)

	// Encoding of the serialized messages
	Encoding() string
}

MetricsUnmarshaler deserializes the message body

type OAuth2

type OAuth2 struct {
	IssuerURL    string `mapstructure:"issuer_url"`
	ClientID     string `mapstructure:"client_id"`
	Audience     string `mapstructure:"audience"`
	ClientSecret string `mapstructure:"client_secret"`
	Scope        string `mapstructure:"scope"`
}

type TLS

type TLS struct {
	CertFile string `mapstructure:"cert_file"`
	KeyFile  string `mapstructure:"key_file"`
}

type Token

type Token struct {
	Token configopaque.String `mapstructure:"token"`
}

type TracesUnmarshaler

type TracesUnmarshaler interface {
	// Unmarshal deserializes the message body into tracesConsumer.
	Unmarshal([]byte) (ptrace.Traces, error)

	// Encoding of the serialized messages.
	Encoding() string
}

TracesUnmarshaler deserializes the message body.