[BUG] BlockingBuffer.bufferUsage metric does not include records in-flight

[dlvenable]

Describe the bug

The BlockingBuffer.bufferUsage metric is inaccurate. It indicates that it is the percentage of the buffer used. However, it is only the percentage of the buffer used for messages that are waiting. It does not include in-flight messages.

Expected behavior

I expect this metric to include both the records waiting and the records-in-flight when calculating the usage.

Screenshots

We sent these metrics to AWS CloudWatch. You can see in the screenshot below that the sum of recordsInBuffer and recordsInFlight (the line in blue) reaches near the maximum defined size of 1,000,000 records. However, the bufferUsage metric is around 70% at the time that happens.

Additional context

This can lead to confusion when trying to see why events are not writing to the buffer. The metrics indicate that there is capacity, but we fail to write to the buffer due to a timeout exception.

You can see that below.

data-prepper/data-prepper-plugins/blocking-buffer/src/main/java/org/opensearch/dataprepper/plugins/buffer/blockingbuffer/BlockingBuffer.java

Lines 120 to 125 in b0d253c

	final boolean permitAcquired = capacitySemaphore.tryAcquire(size, timeoutInMillis, TimeUnit.MILLISECONDS);
	if (!permitAcquired) {
	throw new TimeoutException(
	format("Pipeline [%s] - Buffer does not have enough capacity left for the number of records: %d, " +
	"timed out waiting for slots.",
	pipelineName, size));

This code is what appears to be incorrect. It does not account for the records in-flight.

data-prepper/data-prepper-plugins/blocking-buffer/src/main/java/org/opensearch/dataprepper/plugins/buffer/blockingbuffer/BlockingBuffer.java

Lines 202 to 205 in b0d253c

	final Double nonNegativeTotalRecords = recordsInBuffer.doubleValue() < 0 ? 0 : recordsInBuffer.doubleValue();
	final Double boundedTotalRecords = nonNegativeTotalRecords > bufferCapacity ? bufferCapacity : nonNegativeTotalRecords;
	final Double usage = boundedTotalRecords / bufferCapacity * 100;
	bufferUsage.set(usage);

[dlvenable]

An alternative solution is to also update the recordsInBuffer metric to more accurately reflect that it is the totality of all records. Then we could have a recordsWaitingInBuffer.

Or we could add a new metric that expresses both: recordsTotalInBuffer. This would be equal to the sum of both recordsInBuffer and recordsInFlight.

[graytaylor0]

I think there is still value in knowing the number of recordsInBuffer from a debugging standpoint

[kkondaka]

As discussed offline, I think it is better to have a metric that directly corresponds to the semaphore count that controls the capacity

[dlvenable]

I'm thinking of making two changes:

1. Create a new gauge metric on the semaphore's availablePermits() to track the actual buffer capacity - BlockingBuffer.capacityUsed.
2. Replace the logic for creating the existing bufferUsage such that it is effectively availablePermits() / bufferCapacity.