@@ -48,6 +48,10 @@ type (
4848 logger * log.Logger
4949 stats * RecorderStats
5050 statsOnce sync.Once
51+
52+ concurrencyLevel int
53+ workerJobs chan * workerJob
54+ workerResults chan * workerResult
5155 }
5256 RecorderStats struct {
5357 totalSpans int64
@@ -66,6 +70,18 @@ type (
6670
6771 PayloadSpan map [string ]interface {}
6872 PayloadEvent map [string ]interface {}
73+
74+ workerJob struct {
75+ spans []PayloadSpan
76+ totalSpans int
77+ events []PayloadEvent
78+ totalEvents int
79+ }
80+ workerResult struct {
81+ workerId int
82+ error error
83+ shouldExit bool
84+ }
6985)
7086
7187func NewSpanRecorder (agent * Agent ) * SpanRecorder {
@@ -79,9 +95,12 @@ func NewSpanRecorder(agent *Agent) *SpanRecorder {
7995 r .metadata = agent .metadata
8096 r .logger = agent .logger
8197 r .flushFrequency = agent .flushFrequency
98+ r .concurrencyLevel = agent .concurrencyLevel
8299 r .url = agent .getUrl ("api/agent/ingest" )
83100 r .client = & http.Client {}
84101 r .stats = & RecorderStats {}
102+ r .logger .Printf ("recorder frequency: %v" , agent .flushFrequency )
103+ r .logger .Printf ("recorder concurrency level: %v" , agent .concurrencyLevel )
85104 r .t .Go (r .loop )
86105 return r
87106}
@@ -103,8 +122,18 @@ func (r *SpanRecorder) RecordSpan(span tracer.RawSpan) {
103122
104123func (r * SpanRecorder ) loop () error {
105124 defer func () {
125+ close (r .workerJobs )
126+ close (r .workerResults )
106127 r .logger .Println ("recorder has been stopped." )
107128 }()
129+
130+ // start workers
131+ r .workerJobs = make (chan * workerJob , r .concurrencyLevel )
132+ r .workerResults = make (chan * workerResult , r .concurrencyLevel )
133+ for i := 0 ; i < r .concurrencyLevel ; i ++ {
134+ go r .worker (i + 1 )
135+ }
136+
108137 ticker := time .NewTicker (1 * time .Second )
109138 cTime := time .Now ()
110139 for {
@@ -144,51 +173,104 @@ func (r *SpanRecorder) sendSpans() (error, bool) {
144173 atomic .AddInt64 (& r .stats .sendSpansCalls , 1 )
145174 const batchSize = 1000
146175 var lastError error
176+ var jobs int
147177 for {
148178 spans , spMore , spTotal := r .popPayloadSpan (batchSize )
149179 events , evMore , evTotal := r .popPayloadEvents (batchSize )
150180
151- payload := map [ string ] interface {} {
152- "metadata" : r . metadata ,
153- "spans" : spans ,
154- " events" : events ,
155- tags . AgentID : r . agentId ,
181+ r . workerJobs <- & workerJob {
182+ spans : spans ,
183+ totalSpans : spTotal ,
184+ events : events ,
185+ totalEvents : evTotal ,
156186 }
157- buf , err := encodePayload (payload )
158- if err != nil {
159- atomic .AddInt64 (& r .stats .sendSpansKo , 1 )
160- atomic .AddInt64 (& r .stats .spansNotSent , int64 (len (spans )))
161- return err , false
162- }
163-
164- var testSpans int64
165- for _ , span := range spans {
166- if isTestSpan (span ) {
167- testSpans ++
187+ jobs ++
188+
189+ if len (r .workerResults ) > 0 {
190+ // We check if a previous result call the cancellation of the send
191+ result := <- r .workerResults
192+ lastError = result .error
193+ jobs --
194+ if result .shouldExit {
195+ r .logger .Printf ("worker %d: received a should exit response" , result .workerId )
196+ for i := 0 ; i < jobs ; i ++ {
197+ <- r .workerResults
198+ }
199+ return result .error , result .shouldExit
168200 }
169201 }
170202
171- r .logger .Printf ("sending %d/%d spans with %d/%d events" , len (spans ), spTotal , len (events ), evTotal )
172- statusCode , err := r .callIngest (buf )
173- if err != nil {
174- atomic .AddInt64 (& r .stats .sendSpansKo , 1 )
175- atomic .AddInt64 (& r .stats .spansNotSent , int64 (len (spans )))
176- atomic .AddInt64 (& r .stats .testSpansNotSent , testSpans )
177- } else {
178- atomic .AddInt64 (& r .stats .sendSpansOk , 1 )
179- atomic .AddInt64 (& r .stats .spansSent , int64 (len (spans )))
180- atomic .AddInt64 (& r .stats .testSpansSent , testSpans )
203+ if ! spMore && ! evMore {
204+ break
181205 }
182- if statusCode == 401 {
183- return err , true
206+ }
207+ shouldExit := false
208+ for i := 0 ; i < jobs ; i ++ {
209+ result := <- r .workerResults
210+ lastError = result .error
211+ if result .shouldExit {
212+ r .logger .Printf ("worker %d: received a should exit response" , result .workerId )
213+ shouldExit = true
184214 }
185- lastError = err
215+ }
216+ return lastError , shouldExit
217+ }
186218
187- if ! spMore && ! evMore {
188- break
219+ func (r * SpanRecorder ) worker (id int ) {
220+ for {
221+ select {
222+ case j , ok := <- r .workerJobs :
223+ if ! ok {
224+ if r .debugMode {
225+ r .logger .Printf ("exiting from worker: %d" , id )
226+ }
227+ return
228+ }
229+
230+ payload := map [string ]interface {}{
231+ "metadata" : r .metadata ,
232+ "spans" : j .spans ,
233+ "events" : j .events ,
234+ tags .AgentID : r .agentId ,
235+ }
236+
237+ buf , err := encodePayload (payload )
238+ if err != nil {
239+ atomic .AddInt64 (& r .stats .sendSpansKo , 1 )
240+ atomic .AddInt64 (& r .stats .spansNotSent , int64 (len (j .spans )))
241+ r .workerResults <- & workerResult {
242+ workerId : id ,
243+ error : err ,
244+ shouldExit : false ,
245+ }
246+ continue
247+ }
248+
249+ var testSpans int64
250+ for _ , span := range j .spans {
251+ if isTestSpan (span ) {
252+ testSpans ++
253+ }
254+ }
255+
256+ r .logger .Printf ("worker %d: sending %d/%d spans with %d/%d events" , id , len (j .spans ), j .totalSpans , len (j .events ), j .totalEvents )
257+ statusCode , err := r .callIngest (buf )
258+ if err != nil {
259+ atomic .AddInt64 (& r .stats .sendSpansKo , 1 )
260+ atomic .AddInt64 (& r .stats .spansNotSent , int64 (len (j .spans )))
261+ atomic .AddInt64 (& r .stats .testSpansNotSent , testSpans )
262+ } else {
263+ atomic .AddInt64 (& r .stats .sendSpansOk , 1 )
264+ atomic .AddInt64 (& r .stats .spansSent , int64 (len (j .spans )))
265+ atomic .AddInt64 (& r .stats .testSpansSent , testSpans )
266+ }
267+ r .workerResults <- & workerResult {
268+ workerId : id ,
269+ error : err ,
270+ shouldExit : statusCode == 401 ,
271+ }
189272 }
190273 }
191- return lastError , false
192274}
193275
194276// Stop recorder
0 commit comments