pulsartracing

README

Usage

Interceptors based solution

// create new tracer
// register tracer with GlobalTracer
opentracing.SetGlobalTracer(tracer)

Producer

tracingInterceptor := &pulsartracing.ProducerInterceptor{}

options := pulsar.ProducerOptions{
Topic:            topicName,
Interceptors:     pulsar.ProducerInterceptors{tracingInterceptor},
}

Consumer

tracingInterceptor := &pulsartracing.ConsumerInterceptor{}

options := pulsar.ConsumerOptions{
Topics:           topicName,
SubscriptionName: subscriptionName,
Type:             pulsar.Shared,
Interceptors:      pulsar.ConsumerInterceptors{tracingInterceptor},
}


// to create span with message as parent span
span := pulsartracing.CreateSpanFromMessage(message, tracer, "child_span")

License

Apache 2.0 License.

Documentation

Index

• func CreateSpanFromMessage(cm *pulsar.ConsumerMessage, tracer opentracing.Tracer, label string) opentracing.Span
• func ExtractSpanContextFromConsumerMessage(message pulsar.ConsumerMessage) opentracing.SpanContext
• func ExtractSpanContextFromProducerMessage(message *pulsar.ProducerMessage) opentracing.SpanContext
• func InjectConsumerMessageSpanContext(ctx context.Context, message pulsar.ConsumerMessage)
• func InjectProducerMessageSpanContext(ctx context.Context, message *pulsar.ProducerMessage)
• type ConsumerInterceptor
  • func (t *ConsumerInterceptor) BeforeConsume(message pulsar.ConsumerMessage)
  • func (t *ConsumerInterceptor) OnAcknowledge(consumer pulsar.Consumer, msgID pulsar.MessageID)
  • func (t *ConsumerInterceptor) OnNegativeAcksSend(consumer pulsar.Consumer, msgIDs []pulsar.MessageID)
• type ConsumerMessageExtractAdapter
  • func (a *ConsumerMessageExtractAdapter) ForeachKey(handler func(key, val string) error) error
  • func (a *ConsumerMessageExtractAdapter) Set(key, val string)
• type ConsumerMessageInjectAdapter
  • func (a *ConsumerMessageInjectAdapter) ForeachKey(handler func(key, val string) error) error
  • func (a *ConsumerMessageInjectAdapter) Set(key, val string)
• type ProducerInterceptor
  • func (t *ProducerInterceptor) BeforeSend(producer pulsar.Producer, message *pulsar.ProducerMessage)
  • func (t *ProducerInterceptor) OnSendAcknowledgement(producer pulsar.Producer, message *pulsar.ProducerMessage, ...)
• type ProducerMessageExtractAdapter
  • func (a *ProducerMessageExtractAdapter) ForeachKey(handler func(key, val string) error) error
  • func (a *ProducerMessageExtractAdapter) Set(key, val string)
• type ProducerMessageInjectAdapter
  • func (a *ProducerMessageInjectAdapter) ForeachKey(handler func(key, val string) error) error
  • func (a *ProducerMessageInjectAdapter) Set(key, val string)

Functions

func CreateSpanFromMessage

func CreateSpanFromMessage(cm *pulsar.ConsumerMessage, tracer opentracing.Tracer, label string) opentracing.Span

func ExtractSpanContextFromConsumerMessage

func ExtractSpanContextFromConsumerMessage(message pulsar.ConsumerMessage) opentracing.SpanContext

func ExtractSpanContextFromProducerMessage

func ExtractSpanContextFromProducerMessage(message *pulsar.ProducerMessage) opentracing.SpanContext

func InjectConsumerMessageSpanContext

func InjectConsumerMessageSpanContext(ctx context.Context, message pulsar.ConsumerMessage)

func InjectProducerMessageSpanContext

func InjectProducerMessageSpanContext(ctx context.Context, message *pulsar.ProducerMessage)

Types

type ConsumerInterceptor

type ConsumerInterceptor struct {
}

func (*ConsumerInterceptor) BeforeConsume

func (t *ConsumerInterceptor) BeforeConsume(message pulsar.ConsumerMessage)

func (*ConsumerInterceptor) OnAcknowledge

func (t *ConsumerInterceptor) OnAcknowledge(consumer pulsar.Consumer, msgID pulsar.MessageID)

func (*ConsumerInterceptor) OnNegativeAcksSend

func (t *ConsumerInterceptor) OnNegativeAcksSend(consumer pulsar.Consumer, msgIDs []pulsar.MessageID)

type ConsumerMessageExtractAdapter

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

ConsumerMessageExtractAdapter Implements TextMap Interface

func (*ConsumerMessageExtractAdapter) ForeachKey

func (a *ConsumerMessageExtractAdapter) ForeachKey(handler func(key, val string) error) error

func (*ConsumerMessageExtractAdapter) Set

func (a *ConsumerMessageExtractAdapter) Set(key, val string)

type ConsumerMessageInjectAdapter

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

ConsumerMessageInjectAdapter Implements TextMap Interface

func (*ConsumerMessageInjectAdapter) ForeachKey

func (a *ConsumerMessageInjectAdapter) ForeachKey(handler func(key, val string) error) error

func (*ConsumerMessageInjectAdapter) Set

func (a *ConsumerMessageInjectAdapter) Set(key, val string)

type ProducerInterceptor

type ProducerInterceptor struct {
}

func (*ProducerInterceptor) BeforeSend

func (t *ProducerInterceptor) BeforeSend(producer pulsar.Producer, message *pulsar.ProducerMessage)

func (*ProducerInterceptor) OnSendAcknowledgement

func (t *ProducerInterceptor) OnSendAcknowledgement(producer pulsar.Producer,
	message *pulsar.ProducerMessage,
	msgID pulsar.MessageID)

type ProducerMessageExtractAdapter

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

ProducerMessageExtractAdapter Implements TextMap Interface

func (*ProducerMessageExtractAdapter) ForeachKey

func (a *ProducerMessageExtractAdapter) ForeachKey(handler func(key, val string) error) error

func (*ProducerMessageExtractAdapter) Set

func (a *ProducerMessageExtractAdapter) Set(key, val string)

type ProducerMessageInjectAdapter

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

ProducerMessageInjectAdapter Implements TextMap Interface

func (*ProducerMessageInjectAdapter) ForeachKey

func (a *ProducerMessageInjectAdapter) ForeachKey(handler func(key, val string) error) error

func (*ProducerMessageInjectAdapter) Set

func (a *ProducerMessageInjectAdapter) Set(key, val string)