feat: add WaitUntilEmpty to LogSender

[spikecurtis]

We'll need this to be able to tell when all outstanding logs have been sent, as part of graceful shutdown.

[spikecurtis]

• main
  • feat: send log limit exceeded in response, not error #12078
    • feat: add logSender for sending logs on agent v2 API #12046
      • chore: move LogSender to agentsdk #12158
        • feat: add WaitUntilEmpty to LogSender #12159 👈
      • feat: handle log limit exceeded in logSender #12079
        • feat: ensure that log batches don't exceed 1MiB in logSender #12048
          • feat: limit queued logs to database limit in agent #12067
      • feat: switch agent to use v2 API for sending logs #12068

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @spikecurtis and the rest of your teammates on Graphite

[mafredri]

Why are we duplicating logic from SendLoop here? Since this method doesn't attempt to send, it's quite pointless unless the signaling is happening from a running SendLoop anyway.

Edit: Ah, nevermind, just realized this is only here to handle the user provided context.

I think this could be greatly simplified:

func (l *LogSender) WaitUntilEmpty(ctx context.Context) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-l.allSent:
		return nil
	}
}

[spikecurtis]

The problem with an allSent channel is how to arrange when it should read. Closing the channel won't work, because you can't "unclose" it if more data gets queued.

Writing to the channel won't work if there are more than one caller to WaitUntilEmpty.

[mafredri]

True, it would work better/simpler if the send loop was channel-based as well. In that case, one approach could be this:

func (l *LogSender) WaitUntilEmpty(ctx context.Context) error {
	wait := make(chan struct{})
	l.waitUntilEmpty <- wait

	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-wait:
		return nil
	}
}

// SendLoop
	var waiters []chan struct{}
	for {
		select {
		case <-tick:
		case <-l.waitUntilEmpty:
			waiters = append(waiters, wait)
		}

		// ...

		if len(l.queues) == 0 {
			for _, wait := range waiters {
				close(wait)
			}
			waiters = nil
		}

But it's not quite as nice when retrofitted into the mutex style loop.

[spikecurtis]

The problem there is that it requires SendLoop to actually be running in order for WaitUntilEmpty() to return, which it might not be but we are still empty.

Channels are great for communicating between goroutines. Here what we really, actually want is to know when a condition is satisfied, regardless of other running goroutines, and for that sync.Cond is your friend.

[mafredri]

I had a suggestion for changing WaitUntilEmpty, but I'll leave it up to you whether or not to implement. Works as-is, too, although I feel mutexes are starting to make this service overly complex (vs using channel based communication).

[mafredri]

Suggested change

	if len(l.queues) == 0 {
	return nil
	}
	return ctx.Err()
	return ctx.Err()

We don't actually need this check, this way we give priority to the context cancellation, even if we happen to be done at the same time (this can be preferable in some cases).

[spikecurtis]

Merge activity

• Feb 20, 2:11 AM EST: @spikecurtis merged this pull request with Graphite.