Support micrometer context-propagation

[chickenchickenlove]

Motivation:

• Related Issue : Support micrometer-context-propagation #5145
• Armeria already support context-propagation to maintain RequestContext during executing Reactor code. How it requires maintenance.
• Reactor integrate micro-meter:context-propagation to do context-propagation during Flux, Mono officially. thus, it would be better to migrate from RequestContextHook to RequestContextPropagationHooks because it can reduce maintenance cost.

Modifications:

• Add new Hook for Reactor.
• Add new ThreadLocalAccessor for micro-meter:context-propagation to main RequestContext during executing Reactor code like Mono, Flux.
• Add new config enableContextPropagation to integrate micro-meter:context-propagation with spring-boot3.

Result:

• Closes Support micrometer-context-propagation #5145
• If user want to use micrometer:context-propagation to maintain RequestContext during executing Reactor code like Mono, Flux, just call RequestContextPropagationHook.enable().

[chickenchickenlove]

FYI, flow of micro-meter/context-propagation.

1. If we enable context-propagation, hook for context-propagation will be added Reactor's hooks.

2. if subscribe() are called, hook like captureThreadLocals() are called.

3. and then, ContextSnapshot are created to propagate context. in this time, accessor are used to capture state of ThreadLocals.

4. When Mono or Flux are executed in Scheduler's executor, scope are gotten by calling setThreadLocals(). Scope instance are returned, and it is concrete class of AutoClosable.

5. As you can see, Context-Propagation is restore thread local state at end of try ~ resource.

6. On the other hand, setThreadLocals() be about to restore Threadlocal state from ContextSnapshot before runnable.run().

IMHO, it is very similar to RequestContextHook on armeria.

[github-actions]

🔍 Build Scan® (commit: 33535c0)

Job name	Status	Build Scan®
build-ubicloud-standard-8-jdk-8	✅	https://ge.armeria.dev/s/ct6aj6fgbdouy
build-ubicloud-standard-8-jdk-21-snapshot-blockhound	✅	https://ge.armeria.dev/s/c3llwp5ozt6jm
build-ubicloud-standard-8-jdk-17-min-java-17-coverage	✅	https://ge.armeria.dev/s/2vten2bzyboqw
build-ubicloud-standard-8-jdk-17-min-java-11	✅	https://ge.armeria.dev/s/r67agmbioeg74
build-ubicloud-standard-8-jdk-17-leak	✅	https://ge.armeria.dev/s/d6nxajnvtnuzu
build-ubicloud-standard-8-jdk-11	✅	https://ge.armeria.dev/s/32w7yrkucpxpq
build-macos-12-jdk-21	✅	https://ge.armeria.dev/s/26jcb5cmax4a4

[chickenchickenlove]

I investigate that internal of Context-Propagation.

Case 1. publisher.subscribe(...). downstream -> upstream flow

• When publisher.subscribe(...) are called and publisher is instance of ContextWriteRestoringThreadLocals, it creates ContextSnapshot.Scope.in this sequence, read the Key-Value stored in ReactorContext, then use the ThreadLocalAccessor instance to store the value from ReactorContext into ThreadLocal. Additionally, keep the previously stored values in ThreadLocal as PreviousValues, and use this values to revert the state of ThreadLocal to its previous state when the ContextSnapshot.Scope ends. (FYI, ContextSnapshot.Scope implement AutoClosable).

Case 2. subscriber.onSubscribe(...). upstream -> downstream flow

• When subscriber.onSubscribe(...) are called and subscriber is instance of ContextWriteRestoringThreadLocal, it creates ContextSnapshot.Scope as well. it means that the values in ReactorContext will be stored to ThreadLocal at the start of subscriber.onSubscribe() and ThreadLocal will be revert to its previous state at the end of the function.

Case 3. subscription.request(...). downstream -> upstream flow

• When subscription.request(...) are called and subscriptionis instance ofContextWriteRestoringThreadLocal, it creates ContextSnapshot.scope` as well.

ContextWriteRestoringThreadLocals operator are integrated when both [Mono|Flux]#contextCapture() and [Mono|Flux]#contextWrite() are called. It means that Reactor Context are essential to propagate RequestContext to each ThreadLocal during Reactor Operations.

In practice, it works like this:

1. Reads all values accessible by the KEY of ThreadLocalAccessor in the Reactor Context.
2. Stores the read values in the Threadlocal by using ThreadLocalAccessor. At this time, the values that previously existed in the Threadllocal are maintained in a Map called PreviousValues.
3. When the Scope ends, it restores the PreviousValues back to the ThreadLocal.

It means that Reactor Context for writing, ThreadLocals for reading.

[chickenchickenlove]

@trustin nim, There are also major differences in the test.
I would like to inform you about them.

The test utility function addCallbacks() should be change.

private static <T> Mono<T> addCallbacks(Mono<T> mono, ClientRequestContext ctx) {
        return mono.doFirst(() -> assertThat(ctxExists(ctx)).isTrue())
                   .doOnSubscribe(s -> assertThat(ctxExists(ctx)).isTrue())
                   .doOnRequest(l -> assertThat(ctxExists(ctx)).isTrue())
                   .doOnNext(foo -> assertThat(ctxExists(ctx)).isTrue())
                   .doOnSuccess(t -> assertThat(ctxExists(ctx)).isTrue())
                   .doOnEach(s -> assertThat(ctxExists(ctx)).isTrue())
                   .doOnError(t -> assertThat(ctxExists(ctx)).isTrue())
                   .doAfterTerminate(() -> assertThat(ctxExists(ctx)).isTrue())
                   // I added contextWrite(...)
                   .contextWrite(Context.of(RequestContextAccessor.getInstance().key(), ctx));
        // doOnCancel and doFinally do not have context because we cannot add a hook to the cancel.
    }

contextWrite(Context.of(RequestContextAccessor.getInstance().key(), ctx)); are added. As we know, micro-meter:context-propagation require Reactor Context to propagate context during reactor operations. thus, i added this method.

// Before : StepVerifier.create(mono1)
// After : Add initiali Reactor Context  to StepVerifier. 
StepVerifier.create(mono1, initialReactorContext(ctx))
                    .expectSubscriptionMatches(s -> ctxExists(ctx))
                    .expectNextMatches(s -> ctxExists(ctx) && "baz".equals(s))
                    .verifyComplete();

In previous test code, StepVerifier.create(mono1) is enough. however, it is not enough to micro-meter:context-propagation.
StepVerifier create DefaultVerifySubscriber to subscribe Flux|Mono and valid result is correct. however, DefaultVerifySubscriber has only empty Reactor Context. it means that .expectSubscriptionMatches(s -> ctxExists(ctx)) should be failed.

micro-meter:context-propagation is used to read the values from the Reactor Context and restore the state of ThreadLocal. however, since the DefaultStepVerifierSubscriber has an Empty Reactor Context, the RequestContext stored in ThreadLocal will become Null.

Thus, initial Reactor Context should be include to StepVerifier as well.

[trustin]

Could you fix the build failures before getting reviews?

[chickenchickenlove]

Hi @ikhoon nim, orry to bother you.
When you have time, could you take a look this PR?
Thanks in advanced 🙇‍♂️

[chickenchickenlove]

It has been a long time since the last review.
Please let me know if there is any additional work I should to do. 🙇‍♂️

[minwoox]

Hi! @chickenchickenlove. Apologies for the late reply.

The current implementation has a limitation—it doesn’t support the context hook due to the design of the Micrometer context propagation API. Specifically, the setValue() method doesn’t allow returning a Closeable instance. Because of this limitation, we cannot place these classes in the core module.

Before we can address this issue, we need to discuss potential solutions with the Micrometer maintainers. In the meantime, we thought it would be a good idea to move these classes to a separate module, like micrometer-context, so users who don’t need the context hooks can still use them.

Would you mind creating another module for this?

[chickenchickenlove]

@minwoox -nim, thanks for your comments 🙇‍♂️
Would it work if I do it this way?

1. Create new module micrometer-context by usning gradle.
2. Move these codes from core module to micrometeter-contenxt.

[minwoox]

Would it work if I do it this way?
Create new module micrometer-context by usning gradle.
Move these codes from core module to micrometeter-contenxt.

Yeah, please go ahead. 😉

[CLAassistant]

All committers have signed the CLA.

[chickenchickenlove]

@minwoox -nim, Thanks for your comments!
I made a new commit to apply your comments.
When you have time, please take another look. 🙇‍♂️

[minwoox]

Would you exclude this micrometer-context module from reactor3?
micrometer-context has a limitation which isn't supporting the context hook so we cannot include it in reactor3 module by default.

Here's the suggestion:

// reactor3/build.gradle
dependencies {
    api libs.reactor.core
}

// micrometer-context/build.gradle
dependencies {
    implementation libs.context.propagation
    testImplementation project(':reactor3')
}

and move all test classes into micrometer-context module.

[chickenchickenlove]

@minwoox , thanks for your comments!
I fixed it!
Please take another look. 🙇‍♂️

[minwoox]

Looks great. 👍
Thanks a lot for your hard work. 🙇‍♂️

[jrhee17]

👍 👍

[ikhoon]

🙇‍♂️🙇‍♂️

[ikhoon]

It is okay to merge this PR when CI builds pass.