added tracing log

src/main/java/it/pagopa/interop/signalhub/pull/service/config/ReactorConfiguration.java

@@ -0,0 +1,20 @@
+package it.pagopa.interop.signalhub.pull.service.config;
+
+
+import it.pagopa.interop.signalhub.pull.service.logging.ContextLifter;
+import it.pagopa.interop.signalhub.pull.service.logging.MdcContextLifter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import reactor.core.publisher.Hooks;
+import reactor.core.publisher.Operators;
+
+@Configuration
+public class ReactorConfiguration {
+
+
+    @Bean
+    public void contextLifterConfiguration() {
+        Hooks.onEachOperator(MdcContextLifter.class.getSimpleName(),
+                Operators.lift((sc, sub) -> new ContextLifter<>(sub)));
+    }
+}

src/main/java/it/pagopa/interop/signalhub/pull/service/config/SignalHubPullConfig.java

@@ -14,4 +14,6 @@ public class SignalHubPullConfig {
     private String id;
     private String audience;
     private Long maxNumberPage;
+    private String headerTraceIdKey;
+
 }

src/main/java/it/pagopa/interop/signalhub/pull/service/filter/TraceIdFilter.java

@@ -0,0 +1,42 @@
+package it.pagopa.interop.signalhub.pull.service.filter;
+
+import it.pagopa.interop.signalhub.pull.service.config.SignalHubPullConfig;
+import lombok.AllArgsConstructor;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpHeaders;
+import org.springframework.stereotype.Component;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.WebFilter;
+import org.springframework.web.server.WebFilterChain;
+import reactor.core.publisher.Mono;
+import reactor.util.context.Context;
+
+import java.util.Collections;
+import java.util.UUID;
+
+import static it.pagopa.interop.signalhub.pull.service.utils.Const.TRACE_ID_KEY;
+
+
+@Component
+@Order(Ordered.HIGHEST_PRECEDENCE)
+@AllArgsConstructor
+public class TraceIdFilter implements WebFilter {
+    private final SignalHubPullConfig cfg;
+
+
+    @Override
+    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
+        HttpHeaders headers = exchange.getRequest().getHeaders();
+        var traceId = UUID.randomUUID().toString();
+        if (headers.containsKey(cfg.getHeaderTraceIdKey())) {
+            traceId = headers.getFirst(cfg.getHeaderTraceIdKey());
+        }
+        exchange.getResponse().getHeaders()
+                .putIfAbsent(cfg.getHeaderTraceIdKey(), Collections.singletonList(traceId));
+
+
+        String finalTraceId = traceId;
+        return chain.filter(exchange).contextWrite(Context.of(TRACE_ID_KEY, finalTraceId));
+    }
+}

src/main/java/it/pagopa/interop/signalhub/pull/service/logging/ContextLifter.java

@@ -0,0 +1,55 @@
+package it.pagopa.interop.signalhub.pull.service.logging;
+
+import org.reactivestreams.Subscription;
+import reactor.core.CoreSubscriber;
+import reactor.util.context.Context;
+
+public class ContextLifter<T> implements CoreSubscriber<T> {
+    private final CoreSubscriber<T> actualSubscriber;
+    private final Context context;
+
+    public ContextLifter(CoreSubscriber<T> actualSubscriber) {
+        this.actualSubscriber = actualSubscriber;
+        this.context = actualSubscriber.currentContext();
+    }
+
+    @Override
+    public Context currentContext() {
+        return context;
+    }
+
+    @Override
+    public void onSubscribe(Subscription subscription) {
+        actualSubscriber.onSubscribe(subscription);
+    }
+
+    @Override
+    public void onNext(T t) {
+        MdcContextLifter.setContextToMdc(context);
+        try {
+            actualSubscriber.onNext(t);
+        } finally {
+            MdcContextLifter.clearMdc();
+        }
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        MdcContextLifter.setContextToMdc(context);
+        try {
+            actualSubscriber.onError(throwable);
+        } finally {
+            MdcContextLifter.clearMdc();
+        }
+    }
+
+    @Override
+    public void onComplete() {
+        MdcContextLifter.setContextToMdc(context);
+        try {
+            actualSubscriber.onComplete();
+        } finally {
+            MdcContextLifter.clearMdc();
+        }
+    }
+}

src/main/java/it/pagopa/interop/signalhub/pull/service/logging/MdcContextLifter.java

@@ -0,0 +1,40 @@
+package it.pagopa.interop.signalhub.pull.service.logging;
+
+import org.slf4j.MDC;
+import reactor.core.publisher.Signal;
+import reactor.util.context.Context;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+import static it.pagopa.interop.signalhub.pull.service.utils.Const.TRACE_ID_KEY;
+
+public class MdcContextLifter implements Consumer<Signal<?>> {
+
+    @Override
+    public void accept(Signal<?> signal) {
+        if (!signal.isOnComplete() && !signal.isOnError()) {
+            Optional<Map.Entry<Object, Object>> context = signal.getContextView().stream()
+                    .filter(cxt -> cxt.getKey().equals(TRACE_ID_KEY))
+                    .findFirst();
+
+            context.ifPresent(ctx -> MDC.put(TRACE_ID_KEY, (String)ctx.getValue()));
+        } else {
+            MDC.clear();
+        }
+    }
+
+    public static void setContextToMdc(Context context) {
+        context.stream().forEach(entry -> {
+            if (entry.getKey().equals(TRACE_ID_KEY)){
+                MDC.put(TRACE_ID_KEY, (String) entry.getValue());
+            }
+        });
+    }
+
+    public static void clearMdc(){
+        MDC.clear();
+    }
+
+}

src/main/java/it/pagopa/interop/signalhub/pull/service/utils/Const.java

@@ -3,5 +3,7 @@
 public class Const {
     public static final String STATE_PUBLISHED = "PUBLISHED";
     public static final String STATE_ACTIVE = "ACTIVE";
+    public static final String TRACE_ID_KEY = "traceId";
+
 
 }

src/main/resources/application.properties

@@ -1,5 +1,7 @@
 logging.level.root=INFO
 
+logging.pattern.level="%2p [%X{traceId:-}]"
+
 spring.main.web-application-type=reactive
 management.endpoint.health.show-details=always
 cloud.aws.stack.auto=false
@@ -19,6 +21,8 @@ pdnd.client.event.endpoint-url=https://api.uat.interop.pagopa.it/1.0
 application.eservice.pull.max-number-page=100
 application.eservice.pull.id=a2b84a6e-34cf-44ca-85a4-de21fd232668
 application.eservice.pull.audience=interop-signalhub-pull-signal
+application.eservice.pull.header-trace-id-key=X-Trace-Id
+
 
 database.name=${DB_NAME:signal-hub}
 database.host=${DATABASE_READER_HOST:localhost}