pulsarreceiver

README

Pulsar Receiver

The Pulsar Receiver receives logs, metrics, and traces from Apache Pulsar.

Status
Stability	alpha: metrics, traces, logs
Unsupported Platforms	aix
Distributions	contrib
Issues
Code coverage
Code Owners	@dao-jun
Emeritus	@dmitryax

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:
    • scope:
    • private_key: Path to the private client credentials json file. Must contain type, client_id, client_secret and issuer_url fields.
  • 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

• func NewFactory(options ...FactoryOption) receiver.Factory
• type Athenz
• type Authentication
• type Config
  • func (*Config) Validate() error
• type FactoryOption
• type LogsUnmarshaler
• type MetricsUnmarshaler
• type OAuth2
• type TLS
• type Token
• type TracesUnmarshaler

Functions

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    configoptional.Optional[TLS]    `mapstructure:"tls"`
	Token  configoptional.Optional[Token]  `mapstructure:"token"`
	Athenz configoptional.Optional[Athenz] `mapstructure:"athenz"`
	OAuth2 configoptional.Optional[OAuth2] `mapstructure:"oauth2"`
	// contains filtered or unexported fields
}

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 (*Config) Validate() error

Validate checks the receiver configuration is valid

type FactoryOption

type FactoryOption func(factory *pulsarReceiverFactory)

FactoryOption applies changes to PulsarExporterFactory.

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"`
	PrivateKey string `mapstructure:"private_key"`
	Scope      string `mapstructure:"scope"`
}

type TLS

type TLS struct {
	CertFile string `mapstructure:"cert_file"`
	KeyFile  string `mapstructure:"key_file"`
	// contains filtered or unexported fields
}

type Token

type Token struct {
	Token configopaque.String `mapstructure:"token"`
	// contains filtered or unexported fields
}

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.