From 054e41c2a86b10b940c599a179291299ce530629 Mon Sep 17 00:00:00 2001 From: kun Date: Mon, 5 Mar 2018 16:48:07 +0800 Subject: [PATCH] support jaeger agent export --- exporter/jaeger/agent.go | 90 +++++++ exporter/jaeger/example_test.go | 15 +- .../jaeger/internal/gen-go/jaeger/agent.go | 244 ++++++++++++++++++ exporter/jaeger/jaeger.go | 47 +++- 4 files changed, 386 insertions(+), 10 deletions(-) create mode 100644 exporter/jaeger/agent.go create mode 100644 exporter/jaeger/internal/gen-go/jaeger/agent.go diff --git a/exporter/jaeger/agent.go b/exporter/jaeger/agent.go new file mode 100644 index 000000000..58bdb157f --- /dev/null +++ b/exporter/jaeger/agent.go @@ -0,0 +1,90 @@ +// Copyright 2018, OpenCensus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package jaeger contains an OpenCensus tracing exporter for Jaeger. +package jaeger // import "go.opencensus.io/exporter/jaeger" + +import ( + "fmt" + "io" + "net" + + "git.apache.org/thrift.git/lib/go/thrift" + gen "go.opencensus.io/exporter/jaeger/internal/gen-go/jaeger" +) + +// udpPacketMaxLength is the max size of UDP packet we want to send, synced with jaeger-agent +const udpPacketMaxLength = 65000 + +// agentClientUDP is a UDP client to Jaeger agent that implements gen.Agent interface. +type agentClientUDP struct { + gen.Agent + io.Closer + + connUDP *net.UDPConn + client *gen.AgentClient + maxPacketSize int // max size of datagram in bytes + thriftBuffer *thrift.TMemoryBuffer // buffer used to calculate byte size of a span +} + +// newAgentClientUDP creates a client that sends spans to Jaeger Agent over UDP. +func newAgentClientUDP(hostPort string, maxPacketSize int) (*agentClientUDP, error) { + if maxPacketSize == 0 { + maxPacketSize = udpPacketMaxLength + } + + thriftBuffer := thrift.NewTMemoryBufferLen(maxPacketSize) + protocolFactory := thrift.NewTCompactProtocolFactory() + client := gen.NewAgentClientFactory(thriftBuffer, protocolFactory) + + destAddr, err := net.ResolveUDPAddr("udp", hostPort) + if err != nil { + return nil, err + } + + connUDP, err := net.DialUDP(destAddr.Network(), nil, destAddr) + if err != nil { + return nil, err + } + if err := connUDP.SetWriteBuffer(maxPacketSize); err != nil { + return nil, err + } + + clientUDP := &agentClientUDP{ + connUDP: connUDP, + client: client, + maxPacketSize: maxPacketSize, + thriftBuffer: thriftBuffer} + return clientUDP, nil +} + +// EmitBatch implements EmitBatch() of Agent interface +func (a *agentClientUDP) EmitBatch(batch *gen.Batch) error { + a.thriftBuffer.Reset() + a.client.SeqId = 0 // we have no need for distinct SeqIds for our one-way UDP messages + if err := a.client.EmitBatch(batch); err != nil { + return err + } + if a.thriftBuffer.Len() > a.maxPacketSize { + return fmt.Errorf("Data does not fit within one UDP packet; size %d, max %d, spans %d", + a.thriftBuffer.Len(), a.maxPacketSize, len(batch.Spans)) + } + _, err := a.connUDP.Write(a.thriftBuffer.Bytes()) + return err +} + +// Close implements Close() of io.Closer and closes the underlying UDP connection. +func (a *agentClientUDP) Close() error { + return a.connUDP.Close() +} diff --git a/exporter/jaeger/example_test.go b/exporter/jaeger/example_test.go index 8e4c4c5c8..7865d6ef5 100644 --- a/exporter/jaeger/example_test.go +++ b/exporter/jaeger/example_test.go @@ -21,7 +21,7 @@ import ( "go.opencensus.io/trace" ) -func Example() { +func ExampleNewExporter_collector() { // Register the Jaeger exporter to be able to retrieve // the collected spans. exporter, err := jaeger.NewExporter(jaeger.Options{ @@ -33,3 +33,16 @@ func Example() { } trace.RegisterExporter(exporter) } + +func ExampleNewExporter_agent() { + // Register the Jaeger exporter to be able to retrieve + // the collected spans. + exporter, err := jaeger.NewExporter(jaeger.Options{ + AgentEndpoint: "localhost:6831", + ServiceName: "trace-demo", + }) + if err != nil { + log.Fatal(err) + } + trace.RegisterExporter(exporter) +} diff --git a/exporter/jaeger/internal/gen-go/jaeger/agent.go b/exporter/jaeger/internal/gen-go/jaeger/agent.go new file mode 100644 index 000000000..56c46cb08 --- /dev/null +++ b/exporter/jaeger/internal/gen-go/jaeger/agent.go @@ -0,0 +1,244 @@ +// Autogenerated by Thrift Compiler (0.9.3) +// DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING + +package jaeger + +import ( + "bytes" + "context" + "fmt" + + "git.apache.org/thrift.git/lib/go/thrift" +) + +// (needed to ensure safety because of naive import list construction.) +var _ = thrift.ZERO +var _ = fmt.Printf +var _ = bytes.Equal + +type Agent interface { + // Parameters: + // - Batch + EmitBatch(batch *Batch) (err error) +} + +type AgentClient struct { + Transport thrift.TTransport + ProtocolFactory thrift.TProtocolFactory + InputProtocol thrift.TProtocol + OutputProtocol thrift.TProtocol + SeqId int32 +} + +func NewAgentClientFactory(t thrift.TTransport, f thrift.TProtocolFactory) *AgentClient { + return &AgentClient{Transport: t, + ProtocolFactory: f, + InputProtocol: f.GetProtocol(t), + OutputProtocol: f.GetProtocol(t), + SeqId: 0, + } +} + +func NewAgentClientProtocol(t thrift.TTransport, iprot thrift.TProtocol, oprot thrift.TProtocol) *AgentClient { + return &AgentClient{Transport: t, + ProtocolFactory: nil, + InputProtocol: iprot, + OutputProtocol: oprot, + SeqId: 0, + } +} + +// Parameters: +// - Batch +func (p *AgentClient) EmitBatch(batch *Batch) (err error) { + if err = p.sendEmitBatch(batch); err != nil { + return + } + return +} + +func (p *AgentClient) sendEmitBatch(batch *Batch) (err error) { + oprot := p.OutputProtocol + if oprot == nil { + oprot = p.ProtocolFactory.GetProtocol(p.Transport) + p.OutputProtocol = oprot + } + p.SeqId++ + if err = oprot.WriteMessageBegin("emitBatch", thrift.ONEWAY, p.SeqId); err != nil { + return + } + args := AgentEmitBatchArgs{ + Batch: batch, + } + if err = args.Write(oprot); err != nil { + return + } + if err = oprot.WriteMessageEnd(); err != nil { + return + } + return oprot.Flush() +} + +type AgentProcessor struct { + processorMap map[string]thrift.TProcessorFunction + handler Agent +} + +func (p *AgentProcessor) AddToProcessorMap(key string, processor thrift.TProcessorFunction) { + p.processorMap[key] = processor +} + +func (p *AgentProcessor) GetProcessorFunction(key string) (processor thrift.TProcessorFunction, ok bool) { + processor, ok = p.processorMap[key] + return processor, ok +} + +func (p *AgentProcessor) ProcessorMap() map[string]thrift.TProcessorFunction { + return p.processorMap +} + +func NewAgentProcessor(handler Agent) *AgentProcessor { + + self0 := &AgentProcessor{handler: handler, processorMap: make(map[string]thrift.TProcessorFunction)} + self0.processorMap["emitBatch"] = &agentProcessorEmitBatch{handler: handler} + return self0 +} + +func (p *AgentProcessor) Process(iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) { + ctx := context.Background() + name, _, seqId, err := iprot.ReadMessageBegin() + if err != nil { + return false, err + } + if processor, ok := p.GetProcessorFunction(name); ok { + return processor.Process(ctx, seqId, iprot, oprot) + } + iprot.Skip(thrift.STRUCT) + iprot.ReadMessageEnd() + x1 := thrift.NewTApplicationException(thrift.UNKNOWN_METHOD, "Unknown function "+name) + oprot.WriteMessageBegin(name, thrift.EXCEPTION, seqId) + x1.Write(oprot) + oprot.WriteMessageEnd() + oprot.Flush() + return false, x1 +} + +type agentProcessorEmitBatch struct { + handler Agent +} + +func (p *agentProcessorEmitBatch) Process(ctx context.Context, seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) { + args := AgentEmitBatchArgs{} + if err = args.Read(iprot); err != nil { + iprot.ReadMessageEnd() + return false, err + } + + iprot.ReadMessageEnd() + var err2 error + if err2 = p.handler.EmitBatch(args.Batch); err2 != nil { + return true, err2 + } + return true, nil +} + +// HELPER FUNCTIONS AND STRUCTURES + +// Attributes: +// - Batch +type AgentEmitBatchArgs struct { + Batch *Batch `thrift:"batch,1" json:"batch"` +} + +func NewAgentEmitBatchArgs() *AgentEmitBatchArgs { + return &AgentEmitBatchArgs{} +} + +var AgentEmitBatchArgs_Batch_DEFAULT *Batch + +func (p *AgentEmitBatchArgs) GetBatch() *Batch { + if !p.IsSetBatch() { + return AgentEmitBatchArgs_Batch_DEFAULT + } + return p.Batch +} +func (p *AgentEmitBatchArgs) IsSetBatch() bool { + return p.Batch != nil +} + +func (p *AgentEmitBatchArgs) Read(iprot thrift.TProtocol) error { + if _, err := iprot.ReadStructBegin(); err != nil { + return thrift.PrependError(fmt.Sprintf("%T read error: ", p), err) + } + + for { + _, fieldTypeId, fieldId, err := iprot.ReadFieldBegin() + if err != nil { + return thrift.PrependError(fmt.Sprintf("%T field %d read error: ", p, fieldId), err) + } + if fieldTypeId == thrift.STOP { + break + } + switch fieldId { + case 1: + if err := p.readField1(iprot); err != nil { + return err + } + default: + if err := iprot.Skip(fieldTypeId); err != nil { + return err + } + } + if err := iprot.ReadFieldEnd(); err != nil { + return err + } + } + if err := iprot.ReadStructEnd(); err != nil { + return thrift.PrependError(fmt.Sprintf("%T read struct end error: ", p), err) + } + return nil +} + +func (p *AgentEmitBatchArgs) readField1(iprot thrift.TProtocol) error { + p.Batch = &Batch{} + if err := p.Batch.Read(iprot); err != nil { + return thrift.PrependError(fmt.Sprintf("%T error reading struct: ", p.Batch), err) + } + return nil +} + +func (p *AgentEmitBatchArgs) Write(oprot thrift.TProtocol) error { + if err := oprot.WriteStructBegin("emitBatch_args"); err != nil { + return thrift.PrependError(fmt.Sprintf("%T write struct begin error: ", p), err) + } + if err := p.writeField1(oprot); err != nil { + return err + } + if err := oprot.WriteFieldStop(); err != nil { + return thrift.PrependError("write field stop error: ", err) + } + if err := oprot.WriteStructEnd(); err != nil { + return thrift.PrependError("write struct stop error: ", err) + } + return nil +} + +func (p *AgentEmitBatchArgs) writeField1(oprot thrift.TProtocol) (err error) { + if err := oprot.WriteFieldBegin("batch", thrift.STRUCT, 1); err != nil { + return thrift.PrependError(fmt.Sprintf("%T write field begin error 1:batch: ", p), err) + } + if err := p.Batch.Write(oprot); err != nil { + return thrift.PrependError(fmt.Sprintf("%T error writing struct: ", p.Batch), err) + } + if err := oprot.WriteFieldEnd(); err != nil { + return thrift.PrependError(fmt.Sprintf("%T write field end error 1:batch: ", p), err) + } + return err +} + +func (p *AgentEmitBatchArgs) String() string { + if p == nil { + return "" + } + return fmt.Sprintf("AgentEmitBatchArgs(%+v)", *p) +} diff --git a/exporter/jaeger/jaeger.go b/exporter/jaeger/jaeger.go index 43384a90a..46395ee50 100644 --- a/exporter/jaeger/jaeger.go +++ b/exporter/jaeger/jaeger.go @@ -39,6 +39,10 @@ type Options struct { // For example, http://localhost:14268. Endpoint string + // AgentEndpoint instructs exporter to send spans to jaeger-agent at this address. + // For example, localhost:6831. + AgentEndpoint string + // OnError is the hook to be called when there is // an error occurred when uploading the stats data. // If no custom hook is set, errors are logged. @@ -61,10 +65,20 @@ type Options struct { // the collected spans to Jaeger. func NewExporter(o Options) (*Exporter, error) { endpoint := o.Endpoint - if endpoint == "" { + if endpoint == "" && o.AgentEndpoint == "" { return nil, errors.New("missing endpoint for Jaeger exporter") } - endpoint = endpoint + "/api/traces?format=jaeger.thrift" + + var client *agentClientUDP + var err error + if endpoint != "" { + endpoint = endpoint + "/api/traces?format=jaeger.thrift" + } else { + client, err = newAgentClientUDP(o.AgentEndpoint, udpPacketMaxLength) + if err != nil { + return nil, err + } + } onError := func(err error) { if o.OnError != nil { o.OnError(err) @@ -77,10 +91,12 @@ func NewExporter(o Options) (*Exporter, error) { service = defaultServiceName } e := &Exporter{ - endpoint: endpoint, - username: o.Username, - password: o.Password, - service: service, + endpoint: endpoint, + agentEndpoint: o.AgentEndpoint, + client: client, + username: o.Username, + password: o.Password, + service: service, } bundler := bundler.NewBundler((*gen.Span)(nil), func(bundle interface{}) { if err := e.upload(bundle.([]*gen.Span)); err != nil { @@ -93,9 +109,11 @@ func NewExporter(o Options) (*Exporter, error) { // Exporter is an implementation of trace.Exporter that uploads spans to Jaeger. type Exporter struct { - endpoint string - service string - bundler *bundler.Bundler + endpoint string + agentEndpoint string + service string + bundler *bundler.Bundler + client *agentClientUDP username, password string } @@ -188,6 +206,17 @@ func (e *Exporter) upload(spans []*gen.Span) error { ServiceName: e.service, }, } + if e.endpoint != "" { + return e.uploadCollector(batch) + } + return e.uploadAgent(batch) +} + +func (e *Exporter) uploadAgent(batch *gen.Batch) error { + return e.client.EmitBatch(batch) +} + +func (e *Exporter) uploadCollector(batch *gen.Batch) error { body, err := serialize(batch) if err != nil { return err