Fix undefined switch logic when reporter is closed

[black-adder]

There's a race condition in remoteReporter.Report() where if both r.closed and r.queue channels are closed, the switch stmt will pick one arbitrarily causing a panic if Report is called post close. This PR keeps r.queue open so that even if somehow the queue is filled after close() is called, it will never be flushed. An atomic flag might've worked here but didn't want to add extra sync given the channels already do that for us.

[codecov]

Codecov Report

Merging #239 into master will increase coverage by 0.06%.
The diff coverage is 71.42%.

@@            Coverage Diff            @@
##           master    #239      +/-   ##
=========================================
+ Coverage   83.24%   83.3%   +0.06%     
=========================================
  Files          51      51              
  Lines        2691    2696       +5     
=========================================
+ Hits         2240    2246       +6     
  Misses        326     326              
+ Partials      125     124       -1

Impacted Files	Coverage Δ
transport/zipkin/http.go	78.68% <50%> (+3.24%)	⬆️
reporter.go	69.9% <76.47%> (+0.29%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 617d557...be27f64. Read the comment docs.

[black-adder]

@alexeykudinkin can you take a look?

[isaachier]

Not a huge fan of assigning to built-in function name close. Also, can't you just do defer func() {...}() in this case?

[yurishkuro]

+1, defer func() {...}() should do it

[yurishkuro]

+1, defer func() {...}() should do it

[yurishkuro]

why not close r.queue when exiting processQueue()?

[black-adder]

then we run into the same race condition in Report()

[alexeykudinkin]

@black-adder i think we can do it much simpler and just order the checks, splitting the select in 2:

select {
    case <-r.closed:
        r.reporterOptions.logger.Infof("Span not reported since Reporter is already closed: %+v", span)
    case r.queue <- span:
        atomic.AddInt64(&r.queueLength, 1)
    // ...
}

into

select {
    case <-r.closed:
        r.reporterOptions.logger.Infof("Span not reported since Reporter is already closed: %+v", span)
        return
    default:
}

select{
    case r.queue <- span:
        atomic.AddInt64(&r.queueLength, 1)
    // ...
}

[black-adder]

I still think we could potentially run into a race condition (I maybe completely wrong). I'm inlining the original Close() function and the newly suggested Report() function to show a potential race condition (although highly unlikely):

// From Report()
select {
    case <-r.closed:
        r.reporterOptions.logger.Infof("Span not reported since Reporter is already closed: %+v", span)
        return
    default:
}

// From Close()
close(r.closed)
r.queueDrained.Add(1)
close(r.queue)

// From Report(), this will panic
select{
    case r.queue <- span:
        atomic.AddInt64(&r.queueLength, 1)
    // ...
}

[alexeykudinkin]

@black-adder you're right, this very well might happen. If we absolutely need to close the r.queue we need to have the same reportersDrained wait group which would allow to drain all of the reports in-flight before closing the r.queue

[alexeykudinkin]

This is a dead code, given that we're not closing r.queue anymore

[black-adder]

I don't really see need to close the queue channel apart from being a bad "citizen"

[yurishkuro]

don't need to read ok in this case, can remove if ok

[yurishkuro]

I don't really see need to close the queue channel apart from being a bad "citizen"

fwiw, the Ticker we're using for periodic flushes also doesn't close the channel when you close the ticker.

[black-adder]

I somehow introduced some flakiness, will address

[black-adder]

Hmm, I figured out the cause of the flakiness. The flakiness fix has made me reconsider the changes this PR made.

essentially what's happening was:

span1.Finish()
tracer.Close()
span2.Finish() 

From a user's perspective, this should save span1 and it's understandable that span2 isn't saved. However with the changes in this PR, span1 isn't guaranteed to be persisted. I might have to add a mutex around the whole class which is unfortunate...

[yurishkuro]

can you explain why span1 may not be saved? the close() call has a waitgroup that ensures that the queue is drained.

[alexeykudinkin]

@yurishkuro b/c of what i was talking about: Close doesn't wait for outstanding Reports to finish.

[isaachier]

@alexeykudinkin IDK what you mean. Report calls r.queueDrained.Done() once the report is finished. Close blocks on r.queueDrained.Wait() until the aforementioned Done is called.

[alexeykudinkin]

@isaachier processQueue does that, not Report

[isaachier]

My mistake. Do you think adding a wait group to Report would alleviate the problem?

[black-adder]

I'm giving up on this solution and gonna just add a mutex: https://github.com/jaegertracing/jaeger-client-go/pull/240/files

[alexeykudinkin]

@black-adder putting mutex will make every Report serialized. I will put up a diff w/o mutex

[yurishkuro]

I was just looking at lightstep implementation, it's not using channel for the span queue, but an explicit buffer protected by a mutex (mutex on the tracer, which allows things like checking if it's closed). The side effect is that they don't have eager flushing when the packet size is reached, instead they run a flush loop at 500ms frequency.

[isaachier]

@yurishkuro did the same for C++ client but I feel it is very "un-Go" to avoid the channel altogether.

[yurishkuro]

it is very "un-Go" to avoid the channel altogether.

Channels are not some free magic, they are typically more expensive than a similar explicit code with a mutex. Their benefit is their use in select, since Go's mutex does not provide any signaling primitives (like Object.notify() in Java).

Anyway, I am not suggesting we change the implementation completely. We could simplify the logic by using just a single queue with a command pattern. That is, instead of chan *Span we use chan spanQueueCommand where

type spanQueueCommand struct {
    span *Span
    close *WaitGroup
    flush *WaitGroup // may replace the flushSignal queue, although not necessary
}

This way we don't need to close the queue from the Close() method, we just send a command with close field set to a WG. Any spans that already passed through the Report method will be guaranteed to get processed before the close command is read. One gotcha is that because the queue is bounded, the Close() may need to try inserting the close command repeatedly in a loop, with a small sleep:

func (r *remoteReporter) Close() {
  queueDrained := *sync.WaitGroup()
  queueDrained.Add(1)
  iterations := r.reporterOptions.CloseTimeout / time.MILLISECOND
  for i := 0; i < iterations, i++ {
    select {
       case r.queue <- spanQueueCommand{ close: queueDrained }: // no allocation, I assume
           queueDrained.Wait()
           r.sender.Close()
           return
       default:
           time.Sleep(time.MILLISECOND)
    }
  }
  // maybe log a warning that wasn't able to finish flushing
}

[alexeykudinkin]

@black-adder @yurishkuro addressed in #241