Statsd meter recordings are dropped when submitted from parallel threads

[hdv]

Describe the bug
The following test fails when added to StatsdMeterRegistryPublishTest:

    @ParameterizedTest
    @EnumSource(StatsdProtocol.class)
    void receiveParallelMetricsSuccessfully(StatsdProtocol protocol) throws InterruptedException {
        final int N = 10;

        skipUdsTestOnWindows(protocol);
        serverLatch = new CountDownLatch(N);
        server = startServer(protocol, 0);
        final int port = getPort(protocol);

        meterRegistry = new StatsdMeterRegistry(getUnbufferedConfig(protocol, port), Clock.SYSTEM);
        startRegistryAndWaitForClient();
        Counter counter = Counter.builder("my.counter").register(meterRegistry);

        IntStream.range(0, N)
                .parallel()
                .forEach(ignored -> counter.increment());

        assertThat(serverLatch.await(3, TimeUnit.SECONDS)).isTrue();
    }

It succeeds when .parallel() line is commented out.

Environment

• Micrometer version [e.g. 1.7.1]: latest Micrometer from main branch
• Micrometer registry [e.g. prometheus]: statsd
• OS: [e.g. macOS] macOS
• Java version: [e.g. output of java -version]
  ~/projects/micrometer #v1.8.0 *3 !1 ❯ java --version  base
  openjdk 11.0.2 2019-01-15
  OpenJDK Runtime Environment 18.9 (build 11.0.2+9)
  OpenJDK 64-Bit Server VM 18.9 (build 11.0.2+9, mixed mode)

To Reproduce
How to reproduce the bug:
See the test above

Expected behavior
Test should succeed

Additional context

• Similar test succeeds for 1.7.6 release
• Looks like this change might be related: Upgrade to Reactor 2020.0 #2588
• increasing N to 1000 makes the test fail even without .parallel()

[hdv]

Looks like the following change in StatsdCounter will make make things work for N=10:

    @Override
    public void increment(double amount) {
        if (!shutdown && amount > 0) {
            count.add(amount);
            
            // sink.tryEmitNext(lineBuilder.count((long) amount));

            sink.emitNext(lineBuilder.count((long) amount), (signal, failure) -> {
                System.out.println(failure);
                return failure == Sinks.EmitResult.FAIL_NON_SERIALIZED;
            });
    }

Problem is that we try to emit events to a sink from different threads and the current behavior of Sinks.many() sinks is to fail fast if events are not serialized. Apparently it was different with Processors before.
The danger of the busy-loop approach above is that it might increase CPU load as it will produce lots of retries.

This is all based on https://stackoverflow.com/questions/65379026/reactor-handle-fail-non-serialized-error-from-a-many-sink?noredirect=1&lq=1
Also see reactor/reactor-core#2382

[hdv]

This discovery was a little problematic as LogbackMetricsSuppressingManySink is masking the problem :-(

    @Override
    public Sinks.EmitResult tryEmitNext(String s) {
        LogbackMetrics.ignoreMetrics(() -> delegate.tryEmitNext(s));
        // We want to silently drop the element if not emitted for any reason
        // We do not use the returned result
        return Sinks.EmitResult.OK;
    }

It claims that everything is great while it isn't. And doesn't even debug log anything.
It should probably be smarter about it.

[hdv]

Actually, aforementioned busy-loop solution doesn't really work if I try to set N to 1000 in a test. Too much busy work is happening.

[shakuzen]

Thank you for the detailed investigation. That is very helpful. We will discuss with the Reactor team how we should handle this.

[hdv]

Noticed the title change.
Just to make sure we're on the same page, the problem doesn't seem to be specific to counters but rather the way samples are fed to sink. All StatsD meters are affected.

[simonbasle]

can you try this as a workaround: change the underlying Sinks.Many<T> flavor to be:

Sinks.unsafe().many().unicast().onBackpressureBuffer(
    Queues.unboundedMultiproducer().get());

If I understand correctly, the sink is already subscribed to only once so Unicast is an option.

The above code will create an "unsafe" implementation of the sink, meaning there will be no attempt at detecting concurrent usage. Instead, it will rely on the MPSC properties of the unboundedMultiproducer queue for serialization.

[jonatan-ivanov]

For the sake of documentation:

• unicast does not work since we are re-subscribing in case of connection issues
• I tried Simon's SinksMultiproducer see 1eedca4
• It seems it works (you need to build reactor-core locally and make MultiproducerEmitResult public)
• See connected reactor issue: Explore support for multiproducer Sinks.Many with heavy contention reactor/reactor-core#2850

[jonatan-ivanov]

I moved the code back to the deprecated DirectProcessor. Merging this until we find out what to do can be an option: main...jonatan-ivanov:revert-to-use-processor

[shakuzen]

I've updated the 1.8.0 release notes calling this out as a known issue at the top. We'll work on getting a release out in the next day or two with the fix going back to the DirectProcessor for now. Once we get a snapshot published, could you please try that out and let us know if it looks fixed on your end @hdv?

[hdv]

Yes, I can do that. Thank you!

[shakuzen]

The 1.8.1-SNAPSHOT now contains the change reverting to the deprecated API. Could you please let us know if it looks alright before we do the release @hdv? Thanks for reporting the issue and all the investigation including a test case. Much appreciated and sorry for the regression.

[hdv]

@shakuzen I'll report back shortly after I try it in a live test system.

However, I'd like to point one more thing out. If we set N equal to 1000 in the test, it still fails for both parallel and sequential cases. Any idea why that could happen?

[hdv]

Also I noticed that TcpClient batches samples disregarding the buffering setting. Is that intentional?

[hdv]

@shakuzen I can see that 1.8.1-SNAPSHOT behavior is back on par with 1.7.6! Confirmed in my test environment.

Please note, the questions above are still standing.

[jonatan-ivanov]

@hdv Thank you very much confirming it!
I'm going to release a new patch round today to fix the most pressing/apparent issue and we will look into the other issues around this after that.

[jonatan-ivanov]

@hdv New patch releases are out, you can use a release version now: https://github.com/micrometer-metrics/micrometer/releases/tag/v1.8.1

Thank you very much for reporting and investigating this!
Next steps: fixing the issues you mentioned above and migrating back to the Sink API.

[shakuzen]

However, I'd like to point one more thing out. If we set N equal to 1000 in the test, it still fails for both parallel and sequential cases. Any idea why that could happen?

Yes, we noticed that and have some ideas on why it is happening but need to investigate further. For what it's worth, it is the same behavior on 1.7.x, so it is pre-existing behavior, which is why we're considering it separate from this regression and did not block the patch release on it.

Also I noticed that TcpClient batches samples disregarding the buffering setting. Is that intentional?

Looking at the packet splitting in the TCP case is one of the things we need to look into more.