traefik/vendor/github.com/Microsoft/ApplicationInsights-Go/appinsights/inmemorychannel.go
2018-07-11 09:08:03 +02:00

408 lines
10 KiB
Go

package appinsights
import (
"sync"
"time"
"code.cloudfoundry.org/clock"
)
var (
submit_retries = []time.Duration{time.Duration(10 * time.Second), time.Duration(30 * time.Second), time.Duration(60 * time.Second)}
)
type TelemetryBufferItems []Telemetry
type InMemoryChannel struct {
endpointAddress string
isDeveloperMode bool
collectChan chan Telemetry
controlChan chan *inMemoryChannelControl
batchSize int
batchInterval time.Duration
waitgroup sync.WaitGroup
throttle *throttleManager
transmitter transmitter
}
type inMemoryChannelControl struct {
// If true, flush the buffer.
flush bool
// If true, stop listening on the channel. (Flush is required if any events are to be sent)
stop bool
// If stopping and flushing, this specifies whether to retry submissions on error.
retry bool
// If retrying, what is the max time to wait before finishing up?
timeout time.Duration
// If specified, a message will be sent on this channel when all pending telemetry items have been submitted
callback chan struct{}
}
func NewInMemoryChannel(config *TelemetryConfiguration) *InMemoryChannel {
channel := &InMemoryChannel{
endpointAddress: config.EndpointUrl,
collectChan: make(chan Telemetry),
controlChan: make(chan *inMemoryChannelControl),
batchSize: config.MaxBatchSize,
batchInterval: config.MaxBatchInterval,
throttle: newThrottleManager(),
transmitter: newTransmitter(config.EndpointUrl),
}
go channel.acceptLoop()
return channel
}
func (channel *InMemoryChannel) EndpointAddress() string {
return channel.endpointAddress
}
func (channel *InMemoryChannel) Send(item Telemetry) {
if item != nil && channel.collectChan != nil {
channel.collectChan <- item
}
}
func (channel *InMemoryChannel) Flush() {
if channel.controlChan != nil {
channel.controlChan <- &inMemoryChannelControl{
flush: true,
}
}
}
func (channel *InMemoryChannel) Stop() {
if channel.controlChan != nil {
channel.controlChan <- &inMemoryChannelControl{
stop: true,
}
}
}
func (channel *InMemoryChannel) IsThrottled() bool {
return channel.throttle != nil && channel.throttle.IsThrottled()
}
func (channel *InMemoryChannel) Close(timeout ...time.Duration) <-chan struct{} {
if channel.controlChan != nil {
callback := make(chan struct{})
ctl := &inMemoryChannelControl{
stop: true,
flush: true,
retry: false,
callback: callback,
}
if len(timeout) > 0 {
ctl.retry = true
ctl.timeout = timeout[0]
}
channel.controlChan <- ctl
return callback
} else {
return nil
}
}
func (channel *InMemoryChannel) acceptLoop() {
channelState := newInMemoryChannelState(channel)
for !channelState.stopping {
channelState.start()
}
channelState.stop()
}
// Data shared between parts of a channel
type inMemoryChannelState struct {
channel *InMemoryChannel
stopping bool
buffer TelemetryBufferItems
retry bool
retryTimeout time.Duration
callback chan struct{}
timer clock.Timer
}
func newInMemoryChannelState(channel *InMemoryChannel) *inMemoryChannelState {
return &inMemoryChannelState{
channel: channel,
buffer: make(TelemetryBufferItems, 0, 16),
stopping: false,
timer: currentClock.NewTimer(channel.batchInterval),
}
}
// Part of channel accept loop: Initialize buffer and accept first message, handle controls.
func (state *inMemoryChannelState) start() bool {
if len(state.buffer) > 16 {
// Start out with the size of the previous buffer
state.buffer = make(TelemetryBufferItems, 0, cap(state.buffer))
} else if len(state.buffer) > 0 {
// Start out with at least 16 slots
state.buffer = make(TelemetryBufferItems, 0, 16)
}
// Wait for an event
select {
case event := <-state.channel.collectChan:
if event == nil {
// Channel closed? Not intercepted by Send()?
panic("Received nil event")
}
state.buffer = append(state.buffer, event)
case ctl := <-state.channel.controlChan:
// The buffer is empty, so there would be no point in flushing
state.channel.signalWhenDone(ctl.callback)
if ctl.stop {
state.stopping = true
return false
}
}
if len(state.buffer) == 0 {
return true
}
return state.waitToSend()
}
// Part of channel accept loop: Wait for buffer to fill, timeout to expire, or flush
func (state *inMemoryChannelState) waitToSend() bool {
// Things that are used by the sender if we receive a control message
state.retryTimeout = 0
state.retry = true
state.callback = nil
// Delay until timeout passes or buffer fills up
state.timer.Reset(state.channel.batchInterval)
for {
select {
case event := <-state.channel.collectChan:
if event == nil {
// Channel closed? Not intercepted by Send()?
panic("Received nil event")
}
state.buffer = append(state.buffer, event)
if len(state.buffer) >= state.channel.batchSize {
return state.send()
}
case ctl := <-state.channel.controlChan:
if ctl.stop {
state.stopping = true
state.retry = ctl.retry
if !ctl.flush {
// No flush? Just exit.
state.channel.signalWhenDone(ctl.callback)
return false
}
}
if ctl.flush {
state.retryTimeout = ctl.timeout
state.callback = ctl.callback
return state.send()
}
case _ = <-state.timer.C():
// Timeout expired
return state.send()
}
}
}
// Part of channel accept loop: Check and wait on throttle, submit pending telemetry
func (state *inMemoryChannelState) send() bool {
// Hold up transmission if we're being throttled
if !state.stopping && state.channel.throttle.IsThrottled() {
if !state.waitThrottle() {
// Stopped
return false
}
}
// Send
if len(state.buffer) > 0 {
state.channel.waitgroup.Add(1)
// If we have a callback, wait on the waitgroup now that it's
// incremented.
state.channel.signalWhenDone(state.callback)
go func(buffer TelemetryBufferItems, retry bool, retryTimeout time.Duration) {
defer state.channel.waitgroup.Done()
state.channel.transmitRetry(buffer, retry, retryTimeout)
}(state.buffer, state.retry, state.retryTimeout)
} else if state.callback != nil {
state.channel.signalWhenDone(state.callback)
}
return true
}
// Part of channel accept loop: Wait for throttle to expire while dropping messages
func (state *inMemoryChannelState) waitThrottle() bool {
// Channel is currently throttled. Once the buffer fills, messages will
// be lost... If we're exiting, then we'll just try to submit anyway. That
// request may be throttled and transmitRetry will perform the backoff correctly.
diagnosticsWriter.Write("Channel is throttled, events may be dropped.")
throttleDone := state.channel.throttle.NotifyWhenReady()
dropped := 0
defer diagnosticsWriter.Printf("Channel dropped %d events while throttled", dropped)
for {
select {
case <-throttleDone:
close(throttleDone)
return true
case event := <-state.channel.collectChan:
// If there's still room in the buffer, then go ahead and add it.
if len(state.buffer) < state.channel.batchSize {
state.buffer = append(state.buffer, event)
} else {
if dropped == 0 {
diagnosticsWriter.Write("Buffer is full, dropping further events.")
}
dropped++
}
case ctl := <-state.channel.controlChan:
if ctl.stop {
state.stopping = true
state.retry = ctl.retry
if !ctl.flush {
state.channel.signalWhenDone(ctl.callback)
return false
} else {
// Make an exception when stopping
return true
}
}
// Cannot flush
// TODO: Figure out what to do about callback?
if ctl.flush {
state.channel.signalWhenDone(ctl.callback)
}
}
}
}
// Part of channel accept loop: Clean up and close telemetry channel
func (state *inMemoryChannelState) stop() {
close(state.channel.collectChan)
close(state.channel.controlChan)
state.channel.collectChan = nil
state.channel.controlChan = nil
// Throttle can't close until transmitters are done using it.
state.channel.waitgroup.Wait()
state.channel.throttle.Stop()
state.channel.throttle = nil
}
func (channel *InMemoryChannel) transmitRetry(items TelemetryBufferItems, retry bool, retryTimeout time.Duration) {
payload := items.serialize()
retryTimeRemaining := retryTimeout
for _, wait := range submit_retries {
result, err := channel.transmitter.Transmit(payload, items)
if err == nil && result != nil && result.IsSuccess() {
return
}
if !retry {
diagnosticsWriter.Write("Refusing to retry telemetry submission (retry==false)")
return
}
// Check for success, determine if we need to retry anything
if result != nil {
if result.CanRetry() {
// Filter down to failed items
payload, items = result.GetRetryItems(payload, items)
if len(payload) == 0 || len(items) == 0 {
return
}
} else {
diagnosticsWriter.Write("Cannot retry telemetry submission")
return
}
// Check for throttling
if result.IsThrottled() {
if result.retryAfter != nil {
diagnosticsWriter.Printf("Channel is throttled until %s", *result.retryAfter)
channel.throttle.RetryAfter(*result.retryAfter)
} else {
// TODO: Pick a time
}
}
}
if retryTimeout > 0 {
// We're on a time schedule here. Make sure we don't try longer
// than we have been allowed.
if retryTimeRemaining < wait {
// One more chance left -- we'll wait the max time we can
// and then retry on the way out.
currentClock.Sleep(retryTimeRemaining)
break
} else {
// Still have time left to go through the rest of the regular
// retry schedule
retryTimeRemaining -= wait
}
}
diagnosticsWriter.Printf("Waiting %s to retry submission", wait)
currentClock.Sleep(wait)
// Wait if the channel is throttled and we're not on a schedule
if channel.IsThrottled() && retryTimeout == 0 {
diagnosticsWriter.Printf("Channel is throttled; extending wait time.")
ch := channel.throttle.NotifyWhenReady()
result := <-ch
close(ch)
if !result {
return
}
}
}
// One final try
_, err := channel.transmitter.Transmit(payload, items)
if err != nil {
diagnosticsWriter.Write("Gave up transmitting payload; exhausted retries")
}
}
func (channel *InMemoryChannel) signalWhenDone(callback chan struct{}) {
if callback != nil {
go func() {
channel.waitgroup.Wait()
close(callback)
}()
}
}