mirror of
https://github.com/iflytek/skillhub.git
synced 2026-10-07 02:57:51 +00:00
feat(observability): propagate async trace context
Signed-off-by: XiaoSeS <87064762+XiaoSeS@users.noreply.github.com>
This commit is contained in:
XiaoSeS
parent
f805076c5c
commit
91fb155ef1
7 changed files with 604 additions and 3 deletions
|
|
@ -2,6 +2,7 @@ package com.iflytek.skillhub.config;
|
|||
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.task.support.ContextPropagatingTaskDecorator;
|
||||
import org.springframework.scheduling.annotation.EnableAsync;
|
||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||
|
|
@ -19,12 +20,15 @@ import java.util.concurrent.ThreadPoolExecutor;
|
|||
public class AsyncConfig {
|
||||
|
||||
@Bean(name = "skillhubEventExecutor")
|
||||
public Executor skillhubEventExecutor() {
|
||||
public Executor skillhubEventExecutor(
|
||||
ContextPropagatingTaskDecorator contextPropagatingTaskDecorator
|
||||
) {
|
||||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||||
executor.setCorePoolSize(2);
|
||||
executor.setMaxPoolSize(4);
|
||||
executor.setQueueCapacity(100);
|
||||
executor.setThreadNamePrefix("skillhub-event-");
|
||||
executor.setTaskDecorator(contextPropagatingTaskDecorator);
|
||||
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
|
||||
executor.initialize();
|
||||
return executor;
|
||||
|
|
|
|||
|
|
@ -26,7 +26,10 @@ public class SkillScannerConfig {
|
|||
|
||||
@Bean
|
||||
@ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true")
|
||||
public HttpClient scannerHttpClient(SkillScannerProperties properties) {
|
||||
public HttpClient scannerHttpClient(
|
||||
WebClient.Builder webClientBuilder,
|
||||
SkillScannerProperties properties
|
||||
) {
|
||||
int readTimeoutMs = properties.getReadTimeoutMs();
|
||||
int connectTimeoutMs = properties.getConnectTimeoutMs();
|
||||
|
||||
|
|
@ -52,7 +55,7 @@ public class SkillScannerConfig {
|
|||
.codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(100 * 1024 * 1024))
|
||||
.build();
|
||||
|
||||
WebClient webClient = WebClient.builder()
|
||||
WebClient webClient = webClientBuilder.clone()
|
||||
.clientConnector(new ReactorClientHttpConnector(reactorClient))
|
||||
.exchangeStrategies(strategies)
|
||||
.build();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,37 @@
|
|||
package com.iflytek.skillhub.observability;
|
||||
|
||||
import io.micrometer.context.ThreadLocalAccessor;
|
||||
|
||||
/**
|
||||
* Captures and restores the authoritative Request ID scope for asynchronous execution.
|
||||
*/
|
||||
public final class RequestIdThreadLocalAccessor implements ThreadLocalAccessor<String> {
|
||||
|
||||
public static final String KEY = "skillhub.request-id";
|
||||
|
||||
private final RequestIdAccessor requestIdAccessor;
|
||||
|
||||
public RequestIdThreadLocalAccessor(RequestIdAccessor requestIdAccessor) {
|
||||
this.requestIdAccessor = requestIdAccessor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object key() {
|
||||
return KEY;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getValue() {
|
||||
return requestIdAccessor.current();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setValue(String value) {
|
||||
requestIdAccessor.replace(value);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setValue() {
|
||||
requestIdAccessor.replace(null);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
package com.iflytek.skillhub.observability;
|
||||
|
||||
import com.iflytek.skillhub.observability.tracing.SkillHubObservabilityProperties;
|
||||
import com.iflytek.skillhub.observability.tracing.TracingMode;
|
||||
import io.micrometer.context.ContextRegistry;
|
||||
import io.micrometer.context.ContextSnapshotFactory;
|
||||
import io.micrometer.observation.ObservationRegistry;
|
||||
import io.micrometer.tracing.Tracer;
|
||||
import io.micrometer.tracing.contextpropagation.ObservationAwareSpanThreadLocalAccessor;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.task.support.ContextPropagatingTaskDecorator;
|
||||
|
||||
/**
|
||||
* Defines the context captured by SkillHub-managed asynchronous executors.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class SkillHubContextPropagationConfiguration {
|
||||
|
||||
@Bean
|
||||
ContextRegistry skillHubContextRegistry(
|
||||
RequestIdAccessor requestIdAccessor,
|
||||
SkillHubObservabilityProperties observabilityProperties,
|
||||
ObservationRegistry observationRegistry,
|
||||
Tracer tracer
|
||||
) {
|
||||
ContextRegistry registry = new ContextRegistry()
|
||||
.loadContextAccessors()
|
||||
.loadThreadLocalAccessors();
|
||||
registry.registerThreadLocalAccessor(
|
||||
new RequestIdThreadLocalAccessor(requestIdAccessor)
|
||||
);
|
||||
if (observabilityProperties.getTracingMode() == TracingMode.OTEL_SDK) {
|
||||
registry.registerThreadLocalAccessor(
|
||||
new ObservationAwareSpanThreadLocalAccessor(observationRegistry, tracer)
|
||||
);
|
||||
}
|
||||
return registry;
|
||||
}
|
||||
|
||||
@Bean
|
||||
ContextPropagatingTaskDecorator skillHubContextPropagatingTaskDecorator(
|
||||
ContextRegistry skillHubContextRegistry
|
||||
) {
|
||||
ContextSnapshotFactory snapshotFactory = ContextSnapshotFactory.builder()
|
||||
.contextRegistry(skillHubContextRegistry)
|
||||
.clearMissing(true)
|
||||
.build();
|
||||
return new ContextPropagatingTaskDecorator(snapshotFactory);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,320 @@
|
|||
package com.iflytek.skillhub.observability;
|
||||
|
||||
import com.iflytek.skillhub.config.AsyncConfig;
|
||||
import com.iflytek.skillhub.observability.tracing.SkillHubTracingConfiguration;
|
||||
import io.micrometer.observation.Observation;
|
||||
import io.micrometer.observation.ObservationRegistry;
|
||||
import io.micrometer.tracing.Span;
|
||||
import io.micrometer.tracing.Tracer;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.slf4j.MDC;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.core.task.TaskDecorator;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.FutureTask;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
class ContextPropagationConfigurationTest {
|
||||
|
||||
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(TestApplication.class)
|
||||
.withPropertyValues(
|
||||
"spring.flyway.enabled=false",
|
||||
"spring.jpa.hibernate.ddl-auto=none"
|
||||
);
|
||||
|
||||
@Test
|
||||
void requestIdShouldPropagateRestoreNestedScopeAndNotLeakOnThreadReuse() {
|
||||
contextRunner
|
||||
.withPropertyValues("skillhub.observability.tracing-mode=none")
|
||||
.run(context -> {
|
||||
RequestIdAccessor requestIdAccessor =
|
||||
context.getBean(RequestIdAccessor.class);
|
||||
TaskDecorator taskDecorator = context.getBean(TaskDecorator.class);
|
||||
ExecutorService worker = Executors.newSingleThreadExecutor();
|
||||
try {
|
||||
ContextValues propagated;
|
||||
try (RequestIdAccessor.Scope ignored =
|
||||
requestIdAccessor.open("request-one")) {
|
||||
propagated = execute(worker, taskDecorator, () -> {
|
||||
assertThat(requestIdAccessor.current())
|
||||
.isEqualTo("request-one");
|
||||
try (RequestIdAccessor.Scope nested =
|
||||
requestIdAccessor.open("nested")) {
|
||||
assertThat(requestIdAccessor.current())
|
||||
.isEqualTo("nested");
|
||||
}
|
||||
return currentValues(requestIdAccessor, null);
|
||||
});
|
||||
}
|
||||
|
||||
assertThat(propagated.requestId()).isEqualTo("request-one");
|
||||
assertThat(propagated.mdcRequestId()).isEqualTo("request-one");
|
||||
|
||||
try (RequestIdAccessor.Scope ignored =
|
||||
requestIdAccessor.open("request-failure")) {
|
||||
assertThatThrownBy(() -> execute(
|
||||
worker,
|
||||
taskDecorator,
|
||||
() -> {
|
||||
throw new IllegalStateException("expected failure");
|
||||
}
|
||||
)).hasCauseInstanceOf(IllegalStateException.class);
|
||||
}
|
||||
|
||||
ContextValues clean = execute(
|
||||
worker,
|
||||
taskDecorator,
|
||||
() -> currentValues(requestIdAccessor, null)
|
||||
);
|
||||
assertThat(clean.requestId()).isNull();
|
||||
assertThat(clean.mdcRequestId()).isNull();
|
||||
} finally {
|
||||
worker.shutdownNow();
|
||||
MDC.clear();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void configuredEventExecutorShouldPropagateRequestId() {
|
||||
contextRunner
|
||||
.withPropertyValues("skillhub.observability.tracing-mode=none")
|
||||
.run(context -> {
|
||||
RequestIdAccessor requestIdAccessor =
|
||||
context.getBean(RequestIdAccessor.class);
|
||||
Executor executor = context.getBean("skillhubEventExecutor", Executor.class);
|
||||
try (RequestIdAccessor.Scope ignored =
|
||||
requestIdAccessor.open("configured-executor")) {
|
||||
FutureTask<ContextValues> task = new FutureTask<>(
|
||||
() -> currentValues(requestIdAccessor, null)
|
||||
);
|
||||
executor.execute(task);
|
||||
|
||||
assertThat(task.get(5, TimeUnit.SECONDS).requestId())
|
||||
.isEqualTo("configured-executor");
|
||||
} finally {
|
||||
MDC.clear();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void callerRunsPolicyShouldRestoreCallerScopeAndLeaveWorkerClean() {
|
||||
contextRunner
|
||||
.withPropertyValues("skillhub.observability.tracing-mode=none")
|
||||
.run(context -> {
|
||||
RequestIdAccessor requestIdAccessor =
|
||||
context.getBean(RequestIdAccessor.class);
|
||||
TaskDecorator taskDecorator = context.getBean(TaskDecorator.class);
|
||||
ThreadPoolTaskExecutor executor = callerRunsExecutor(taskDecorator);
|
||||
CountDownLatch workerStarted = new CountDownLatch(1);
|
||||
CountDownLatch releaseWorker = new CountDownLatch(1);
|
||||
FutureTask<Void> blockingTask = new FutureTask<>(() -> {
|
||||
workerStarted.countDown();
|
||||
releaseWorker.await(5, TimeUnit.SECONDS);
|
||||
return null;
|
||||
});
|
||||
try {
|
||||
executor.execute(blockingTask);
|
||||
assertThat(workerStarted.await(5, TimeUnit.SECONDS)).isTrue();
|
||||
|
||||
AtomicReference<ContextValues> callerRunValues =
|
||||
new AtomicReference<>();
|
||||
String callerThread = Thread.currentThread().getName();
|
||||
try (RequestIdAccessor.Scope ignored =
|
||||
requestIdAccessor.open("caller-request")) {
|
||||
executor.execute(() -> {
|
||||
assertThat(Thread.currentThread().getName())
|
||||
.isEqualTo(callerThread);
|
||||
callerRunValues.set(currentValues(
|
||||
requestIdAccessor,
|
||||
null
|
||||
));
|
||||
});
|
||||
assertThat(requestIdAccessor.current())
|
||||
.isEqualTo("caller-request");
|
||||
assertThat(MDC.get(RequestIdAccessor.MDC_KEY))
|
||||
.isEqualTo("caller-request");
|
||||
}
|
||||
|
||||
assertThat(callerRunValues.get().requestId())
|
||||
.isEqualTo("caller-request");
|
||||
releaseWorker.countDown();
|
||||
blockingTask.get(5, TimeUnit.SECONDS);
|
||||
|
||||
FutureTask<ContextValues> cleanTask = new FutureTask<>(
|
||||
() -> currentValues(requestIdAccessor, null)
|
||||
);
|
||||
executor.execute(cleanTask);
|
||||
ContextValues clean = cleanTask.get(5, TimeUnit.SECONDS);
|
||||
assertThat(clean.requestId()).isNull();
|
||||
assertThat(clean.mdcRequestId()).isNull();
|
||||
} finally {
|
||||
releaseWorker.countDown();
|
||||
executor.shutdown();
|
||||
MDC.clear();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void otelSpanShouldPropagateAndBeClearedAfterTask() {
|
||||
contextRunner
|
||||
.withPropertyValues(
|
||||
"skillhub.observability.tracing-mode=otel-sdk",
|
||||
"management.tracing.sampling.probability=1.0"
|
||||
)
|
||||
.run(context -> {
|
||||
RequestIdAccessor requestIdAccessor =
|
||||
context.getBean(RequestIdAccessor.class);
|
||||
TaskDecorator taskDecorator = context.getBean(TaskDecorator.class);
|
||||
Tracer tracer = context.getBean(Tracer.class);
|
||||
ExecutorService worker = Executors.newSingleThreadExecutor();
|
||||
Span span = tracer.nextSpan().name("parent").start();
|
||||
try {
|
||||
ContextValues propagated;
|
||||
try (Tracer.SpanInScope ignored = tracer.withSpan(span)) {
|
||||
propagated = execute(
|
||||
worker,
|
||||
taskDecorator,
|
||||
() -> currentValues(requestIdAccessor, tracer)
|
||||
);
|
||||
}
|
||||
|
||||
assertThat(propagated.traceId())
|
||||
.isEqualTo(span.context().traceId());
|
||||
assertThat(propagated.mdcTraceId())
|
||||
.isEqualTo(span.context().traceId());
|
||||
|
||||
ContextValues clean = execute(
|
||||
worker,
|
||||
taskDecorator,
|
||||
() -> currentValues(requestIdAccessor, tracer)
|
||||
);
|
||||
assertThat(clean.traceId()).isNull();
|
||||
assertThat(clean.mdcTraceId()).isNull();
|
||||
} finally {
|
||||
span.end();
|
||||
worker.shutdownNow();
|
||||
MDC.clear();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void otelObservationShouldPropagateItsTraceAndRestoreWorker() {
|
||||
contextRunner
|
||||
.withPropertyValues(
|
||||
"skillhub.observability.tracing-mode=otel-sdk",
|
||||
"management.tracing.sampling.probability=1.0"
|
||||
)
|
||||
.run(context -> {
|
||||
RequestIdAccessor requestIdAccessor =
|
||||
context.getBean(RequestIdAccessor.class);
|
||||
TaskDecorator taskDecorator = context.getBean(TaskDecorator.class);
|
||||
Tracer tracer = context.getBean(Tracer.class);
|
||||
ObservationRegistry observationRegistry =
|
||||
context.getBean(ObservationRegistry.class);
|
||||
ExecutorService worker = Executors.newSingleThreadExecutor();
|
||||
Observation observation = Observation
|
||||
.createNotStarted("parent-observation", observationRegistry)
|
||||
.start();
|
||||
try {
|
||||
ContextValues propagated;
|
||||
String parentTraceId;
|
||||
try (Observation.Scope ignored = observation.openScope()) {
|
||||
assertThat(tracer.currentSpan()).isNotNull();
|
||||
parentTraceId = tracer.currentSpan().context().traceId();
|
||||
propagated = execute(
|
||||
worker,
|
||||
taskDecorator,
|
||||
() -> currentValues(requestIdAccessor, tracer)
|
||||
);
|
||||
}
|
||||
|
||||
assertThat(propagated.traceId()).isEqualTo(parentTraceId);
|
||||
assertThat(propagated.mdcTraceId()).isEqualTo(parentTraceId);
|
||||
|
||||
ContextValues clean = execute(
|
||||
worker,
|
||||
taskDecorator,
|
||||
() -> currentValues(requestIdAccessor, tracer)
|
||||
);
|
||||
assertThat(clean.traceId()).isNull();
|
||||
assertThat(clean.mdcTraceId()).isNull();
|
||||
} finally {
|
||||
observation.stop();
|
||||
worker.shutdownNow();
|
||||
MDC.clear();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private ContextValues currentValues(
|
||||
RequestIdAccessor requestIdAccessor,
|
||||
Tracer tracer
|
||||
) {
|
||||
Span currentSpan = tracer == null ? null : tracer.currentSpan();
|
||||
return new ContextValues(
|
||||
requestIdAccessor.current(),
|
||||
MDC.get(RequestIdAccessor.MDC_KEY),
|
||||
currentSpan == null ? null : currentSpan.context().traceId(),
|
||||
MDC.get("traceId")
|
||||
);
|
||||
}
|
||||
|
||||
private <T> T execute(
|
||||
Executor executor,
|
||||
TaskDecorator taskDecorator,
|
||||
Callable<T> action
|
||||
) throws Exception {
|
||||
FutureTask<T> task = new FutureTask<>(action);
|
||||
executor.execute(taskDecorator.decorate(task));
|
||||
return task.get(5, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
private ThreadPoolTaskExecutor callerRunsExecutor(TaskDecorator taskDecorator) {
|
||||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||||
executor.setCorePoolSize(1);
|
||||
executor.setMaxPoolSize(1);
|
||||
executor.setQueueCapacity(0);
|
||||
executor.setTaskDecorator(taskDecorator);
|
||||
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
|
||||
executor.initialize();
|
||||
return executor;
|
||||
}
|
||||
|
||||
private record ContextValues(
|
||||
String requestId,
|
||||
String mdcRequestId,
|
||||
String traceId,
|
||||
String mdcTraceId
|
||||
) {
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableAutoConfiguration
|
||||
@Import({
|
||||
SkillHubTracingConfiguration.class,
|
||||
SkillHubContextPropagationConfiguration.class,
|
||||
RequestIdAccessor.class,
|
||||
AsyncConfig.class
|
||||
})
|
||||
static class TestApplication {
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,178 @@
|
|||
package com.iflytek.skillhub.observability.tracing;
|
||||
|
||||
import com.iflytek.skillhub.auth.oauth.GitLabClaimsExtractor;
|
||||
import com.iflytek.skillhub.auth.oauth.OAuthClaims;
|
||||
import com.iflytek.skillhub.config.SkillScannerConfig;
|
||||
import com.iflytek.skillhub.config.SkillScannerProperties;
|
||||
import com.iflytek.skillhub.infra.http.HttpClient;
|
||||
import com.sun.net.httpserver.HttpExchange;
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
import io.micrometer.tracing.Span;
|
||||
import io.micrometer.tracing.Tracer;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.security.oauth2.client.registration.ClientRegistration;
|
||||
import org.springframework.security.oauth2.client.userinfo.OAuth2UserRequest;
|
||||
import org.springframework.security.oauth2.core.AuthorizationGrantType;
|
||||
import org.springframework.security.oauth2.core.OAuth2AccessToken;
|
||||
import org.springframework.security.oauth2.core.user.DefaultOAuth2User;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class HttpTracePropagationTest {
|
||||
|
||||
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(TestApplication.class)
|
||||
.withPropertyValues(
|
||||
"spring.flyway.enabled=false",
|
||||
"spring.jpa.hibernate.ddl-auto=none",
|
||||
"skillhub.observability.tracing-mode=otel-sdk",
|
||||
"management.tracing.sampling.probability=1.0",
|
||||
"skillhub.security.scanner.enabled=true"
|
||||
);
|
||||
|
||||
@Test
|
||||
void shouldPropagateW3cContextToScannerButNotExternalGitLab() throws Exception {
|
||||
try (HeaderCaptureServer scannerServer =
|
||||
new HeaderCaptureServer("text/plain", "scanner-ok");
|
||||
HeaderCaptureServer gitLabServer =
|
||||
new HeaderCaptureServer(
|
||||
"application/json",
|
||||
"""
|
||||
[
|
||||
{
|
||||
"email": "alice@gitlab.example",
|
||||
"confirmed_at": "2026-04-16T08:00:00Z"
|
||||
}
|
||||
]
|
||||
"""
|
||||
)) {
|
||||
contextRunner.run(context -> {
|
||||
Tracer tracer = context.getBean(Tracer.class);
|
||||
HttpClient scannerClient =
|
||||
context.getBean("scannerHttpClient", HttpClient.class);
|
||||
GitLabClaimsExtractor gitLabClaimsExtractor =
|
||||
context.getBean(GitLabClaimsExtractor.class);
|
||||
Span span = tracer.nextSpan().name("outbound-boundary").start();
|
||||
try (Tracer.SpanInScope ignored = tracer.withSpan(span)) {
|
||||
assertThat(scannerClient.get(
|
||||
scannerServer.url("/health"),
|
||||
String.class
|
||||
)).isEqualTo("scanner-ok");
|
||||
|
||||
OAuthClaims claims = gitLabClaimsExtractor.extract(
|
||||
gitLabRequest(gitLabServer.url("/api/v4/user")),
|
||||
new DefaultOAuth2User(
|
||||
List.of(),
|
||||
Map.of(
|
||||
"id", 42,
|
||||
"username", "alice",
|
||||
"email", "alice+pending@gitlab.example"
|
||||
),
|
||||
"username"
|
||||
)
|
||||
);
|
||||
assertThat(claims.email())
|
||||
.isEqualTo("alice@gitlab.example");
|
||||
} finally {
|
||||
span.end();
|
||||
}
|
||||
|
||||
String scannerTraceparent = scannerServer.traceparent();
|
||||
assertThat(scannerTraceparent)
|
||||
.matches("^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$");
|
||||
assertThat(scannerTraceparent.substring(3, 35))
|
||||
.isEqualTo(span.context().traceId());
|
||||
assertThat(gitLabServer.traceparent()).isNull();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private OAuth2UserRequest gitLabRequest(String userInfoUri) {
|
||||
ClientRegistration registration = ClientRegistration
|
||||
.withRegistrationId("gitlab")
|
||||
.clientId("client-id")
|
||||
.clientSecret("client-secret")
|
||||
.authorizationGrantType(AuthorizationGrantType.AUTHORIZATION_CODE)
|
||||
.redirectUri("{baseUrl}/login/oauth2/code/{registrationId}")
|
||||
.scope("read_user", "email")
|
||||
.authorizationUri("https://gitlab.example/oauth/authorize")
|
||||
.tokenUri("https://gitlab.example/oauth/token")
|
||||
.userInfoUri(userInfoUri)
|
||||
.userNameAttributeName("username")
|
||||
.clientName("GitLab")
|
||||
.build();
|
||||
OAuth2AccessToken accessToken = new OAuth2AccessToken(
|
||||
OAuth2AccessToken.TokenType.BEARER,
|
||||
"test-token",
|
||||
Instant.now(),
|
||||
Instant.now().plusSeconds(3600)
|
||||
);
|
||||
return new OAuth2UserRequest(registration, accessToken);
|
||||
}
|
||||
|
||||
private static final class HeaderCaptureServer implements AutoCloseable {
|
||||
|
||||
private final HttpServer server;
|
||||
private final AtomicReference<String> traceparent = new AtomicReference<>();
|
||||
|
||||
private HeaderCaptureServer(String contentType, String body) throws IOException {
|
||||
byte[] response = body.getBytes(StandardCharsets.UTF_8);
|
||||
server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
|
||||
server.createContext("/", exchange -> respond(
|
||||
exchange,
|
||||
contentType,
|
||||
response
|
||||
));
|
||||
server.start();
|
||||
}
|
||||
|
||||
private void respond(
|
||||
HttpExchange exchange,
|
||||
String contentType,
|
||||
byte[] response
|
||||
) throws IOException {
|
||||
traceparent.set(exchange.getRequestHeaders().getFirst("traceparent"));
|
||||
exchange.getResponseHeaders().set("Content-Type", contentType);
|
||||
exchange.sendResponseHeaders(200, response.length);
|
||||
try (var responseBody = exchange.getResponseBody()) {
|
||||
responseBody.write(response);
|
||||
}
|
||||
}
|
||||
|
||||
private String url(String path) {
|
||||
return "http://127.0.0.1:" + server.getAddress().getPort() + path;
|
||||
}
|
||||
|
||||
private String traceparent() {
|
||||
return traceparent.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
server.stop(0);
|
||||
}
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableAutoConfiguration
|
||||
@Import({
|
||||
SkillHubTracingConfiguration.class,
|
||||
SkillScannerConfig.class,
|
||||
SkillScannerProperties.class,
|
||||
GitLabClaimsExtractor.class
|
||||
})
|
||||
static class TestApplication {
|
||||
}
|
||||
}
|
||||
|
|
@ -27,6 +27,14 @@ public class GitLabClaimsExtractor implements OAuthClaimsExtractor {
|
|||
|
||||
private final RestClient restClient;
|
||||
|
||||
/**
|
||||
* Uses an external-service client that is intentionally not customized with application
|
||||
* tracing. Trace context must not be propagated to a user-configured GitLab host.
|
||||
*/
|
||||
public GitLabClaimsExtractor() {
|
||||
this(RestClient.builder());
|
||||
}
|
||||
|
||||
public GitLabClaimsExtractor(RestClient.Builder restClientBuilder) {
|
||||
this.restClient = restClientBuilder
|
||||
.defaultHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue