diff --git a/src/main/java/ru/nbch/credit_tracker/config/mdc/ClearMdc.java b/src/main/java/ru/nbch/credit_tracker/config/mdc/ClearMdc.java new file mode 100644 index 0000000..5489ca5 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/mdc/ClearMdc.java @@ -0,0 +1,14 @@ +package ru.nbch.credit_tracker.config.mdc; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import ru.nbch.credit_tracker.utils.LogUtil; + +// Annotation to only clear the MDC key at method exit +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.METHOD) +public @interface ClearMdc { + String key() default LogUtil.mdcKey; +} diff --git a/src/main/java/ru/nbch/credit_tracker/config/mdc/MdcAspect.java b/src/main/java/ru/nbch/credit_tracker/config/mdc/MdcAspect.java index 756dc01..9f68cef 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/mdc/MdcAspect.java +++ b/src/main/java/ru/nbch/credit_tracker/config/mdc/MdcAspect.java @@ -1,18 +1,27 @@ package ru.nbch.credit_tracker.config.mdc; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.exception.ExceptionUtils; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; -import org.slf4j.MDC; import ru.nbch.credit_tracker.entities.app.UserPackage; import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest; +import ru.nbch.credit_tracker.utils.LogUtil; @Slf4j @Aspect public class MdcAspect { + // For methods that only require clearing the MDC at the end + @Around("@annotation(clearMdc)") + public Object clearMdcAdvice(ProceedingJoinPoint pjp, ClearMdc clearMdc) throws Throwable { + try { + return pjp.proceed(); + } finally { + LogUtil.removeMdcLogKey(clearMdc.key()); + } + } + // For methods that set an MDC value (from a method argument) and then clear it at the end @Around("@annotation(setAndClearMdc)") public Object setAndClearMdcAdvice(ProceedingJoinPoint pjp, SetAndClearMdc setAndClearMdc) throws Throwable { @@ -20,30 +29,18 @@ public class MdcAspect { int index = setAndClearMdc.argIndex(); Object obj = (args != null && args.length > index) ? args[index] : null; if (obj instanceof MonitoringRequest request) { - if (request.getPackageName() != null) { - putMdcValue(setAndClearMdc.key(), request.getPackageName()); - } + LogUtil.addMdcLogKey(setAndClearMdc.key(), request.getPackageName()); } else if (obj instanceof UserPackage userPackage) { - if (userPackage.getPackageName() != null) { - putMdcValue(setAndClearMdc.key(), userPackage.getPackageName()); - } + LogUtil.addMdcLogKey(setAndClearMdc.key(), userPackage.getPackageName()); } else if (obj instanceof String packageName) { - putMdcValue(setAndClearMdc.key(), packageName); + LogUtil.addMdcLogKey(setAndClearMdc.key(), packageName); } else { log.debug("no packageName found for for {}", pjp.toShortString()); } try { return pjp.proceed(); } finally { - MDC.remove(setAndClearMdc.key()); - } - } - - private void putMdcValue(String key, String value) { - try { - MDC.put(key, "[" + value + "] "); - } catch (Throwable e) { - log.error("{}", ExceptionUtils.getStackTrace(e)); + LogUtil.removeMdcLogKey(setAndClearMdc.key()); } } } diff --git a/src/main/java/ru/nbch/credit_tracker/config/mdc/SetAndClearMdc.java b/src/main/java/ru/nbch/credit_tracker/config/mdc/SetAndClearMdc.java index 8d8a14a..a832171 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/mdc/SetAndClearMdc.java +++ b/src/main/java/ru/nbch/credit_tracker/config/mdc/SetAndClearMdc.java @@ -4,12 +4,13 @@ import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; +import ru.nbch.credit_tracker.utils.LogUtil; // Annotation to set an MDC key based on a method argument and clear it after method execution @Retention(RetentionPolicy.RUNTIME) @Target(ElementType.METHOD) public @interface SetAndClearMdc { - String key() default "ru.nbch.credit_tracker.log.mdc.key"; + String key() default LogUtil.mdcKey; // The index of the method argument to use for the MDC value (defaults to the first argument) int argIndex() default 0; } diff --git a/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java b/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java index e26930b..02dd061 100644 --- a/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java +++ b/src/main/java/ru/nbch/credit_tracker/controller/SignalControllerV3.java @@ -10,6 +10,7 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.multipart.MultipartFile; +import ru.nbch.credit_tracker.config.mdc.ClearMdc; import ru.nbch.credit_tracker.entities.user_request.active_packages.MonitoringActivePackagesRequest; import ru.nbch.credit_tracker.entities.user_request.all_packages.MonitoringAllPackagesRequest; import ru.nbch.credit_tracker.entities.user_response.active_packages.MonitoringActivePackagesResponse; @@ -34,6 +35,7 @@ public class SignalControllerV3 { } @PostMapping(value = "/set", produces = MediaType.APPLICATION_XML_VALUE) + @ClearMdc public ResponseEntity setupMonitoring(@RequestParam("file") MultipartFile file) { signalFacadeV3.setupMonitoringRequest(file); return ResponseEntity.ok().body(new MonitoringSetupResponse()); diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java index 9f85af3..82f16eb 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -36,6 +36,7 @@ import ru.nbch.credit_tracker.service.scoring.AuthorizationService; import ru.nbch.credit_tracker.service.signal.SignalRestService; import ru.nbch.credit_tracker.service.task.PipelinePackageManager; import ru.nbch.credit_tracker.utils.CTUtil; +import ru.nbch.credit_tracker.utils.LogUtil; import ru.nbch.credit_tracker.utils.ZipUtils; import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp; import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp; @@ -114,6 +115,7 @@ public class SignalFacadeV3 { // } Long packageId = signalService.registerPackage(userId, packageName); + LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId)); log.info("Package registered. PackageId {}. Status: {}", packageId, Status.PROCESSING.name()); pipelinePackageManager.buildAndRun(packageId, filePath); } catch (Exception e) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java index e4819b9..8d46872 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Handler.java @@ -2,4 +2,5 @@ package ru.nbch.credit_tracker.service.task; public interface Handler { Z handle(X taskPayload); + default int preferredConcurrency() {return 10;} } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java index e774c1b..927c833 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java @@ -21,6 +21,7 @@ import ru.nbch.credit_tracker.service.task.impl.SaveSubjectHandler; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2; +import ru.nbch.credit_tracker.utils.LogUtil; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @Component @@ -85,8 +86,8 @@ public class PipelinePackageManager { .pollFinishingQueue(pipelineTerminatorFactory.get()) .build(); - executorService.submit(pipeline.getStarter()); - executorService.submit(pipeline.getTerminator()); + executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getStarter())); + executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getTerminator())); pipeline.getTerminator() .doneFuture() diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java index 03dd6de..97714a6 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/Stage.java @@ -13,6 +13,7 @@ import java.util.function.Consumer; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import ru.nbch.credit_tracker.model.StagePayload; +import ru.nbch.credit_tracker.utils.LogUtil; @Slf4j public class Stage implements Runnable { @@ -35,12 +36,13 @@ public class Stage implements Ru this.in = in; this.out = out; this.dispatcherName = dispatcher.getClass().getSimpleName(); - this.ioPool = new ThreadPoolExecutor(10, - 10, + var mdcMap = LogUtil.extractMdcMap(); + this.ioPool = new ThreadPoolExecutor(dispatcher.preferredConcurrency(), + dispatcher.preferredConcurrency(), 0L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(1), r -> { - Thread t = new Thread(r); + Thread t = new Thread(LogUtil.wrap(mdcMap, r)); t.setName("pipe-handler-%s-%d".formatted(dispatcherName, t.threadId())); return t; }, @@ -102,6 +104,9 @@ public class Stage implements Ru } ); inflight.add(f); + if ((counter % 20) == 0) { + inflight.removeIf(Future::isDone); + } } } catch (Exception e) { log.error("Error for {}: {}; {}", dispatcherName, info(), ExceptionUtils.getStackTrace(e)); diff --git a/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java new file mode 100644 index 0000000..de07d12 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/utils/LogUtil.java @@ -0,0 +1,85 @@ +package ru.nbch.credit_tracker.utils; + +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.MDC; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; + + +public class LogUtil { + private final static Logger log = LoggerFactory.getLogger(LogUtil.class); + public static final String mdcKey = "ru.nbch.credit_tracker.log.mdc.key"; + + public static void addMdcLogKey(String uuid) { + addMdcLogKey(mdcKey, uuid); + } + + + public static void removeMdcLogKey() { + removeMdcLogKey(mdcKey); + } + + public static void removeMdcLogKey(String key) { + try { + MDC.remove(key); + } catch (Throwable e) { + log.error(ExceptionUtils.getStackTrace(e)); + } + } + + public static void addMdcLogKey(String mdcKey, String uuid) { + if (uuid != null) { + try { + MDC.put(mdcKey, "[" + uuid + "] "); + } catch (Throwable e) { + log.error("{}", ExceptionUtils.getStackTrace(e)); + } + } + } + + public static Map extractMdcMap() { + try { + return MDC.getCopyOfContextMap(); + } catch (Throwable e) { + log.error(ExceptionUtils.getStackTrace(e)); + return null; + } + } + + public static Runnable wrap(Map contextMap, Runnable task) { + return () -> { + Map previous = null; + try { + try { + previous = MDC.getCopyOfContextMap(); + } catch (Throwable e) { + log.error(ExceptionUtils.getStackTrace(e)); + } + // ставим захваченный контекст (может быть null) + try { + if (contextMap == null) { + MDC.clear(); + } else { + MDC.setContextMap(contextMap); + } + } catch (Throwable e) { + log.error(ExceptionUtils.getStackTrace(e)); + } + + task.run(); + } finally { + // восстанавливаем предыдущий (может быть null) + try { + if (previous == null) { + MDC.clear(); + } else { + MDC.setContextMap(previous); + } + } catch (Throwable e) { + log.error(ExceptionUtils.getStackTrace(e)); + } + } + }; + } +} diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml index 40d7a8c..e0f5bce 100644 --- a/src/main/resources/logback.xml +++ b/src/main/resources/logback.xml @@ -8,7 +8,7 @@ - %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} %X{ru.nbch.credit_tracker.log.mdc.key} - %msg%n + %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} %X{ru.nbch.credit_tracker.log.mdc.key}- %msg%n utf-8 @@ -22,7 +22,7 @@ 100MB - %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} %X{ru.nbch.credit_tracker.log.mdc.key} - %msg%n + %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} %X{ru.nbch.credit_tracker.log.mdc.key}- %msg%n utf-8