handler concurrency & logs

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;
+}

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) {
+            LogUtil.addMdcLogKey(setAndClearMdc.key(), request.getPackageName());
-                putMdcValue(setAndClearMdc.key(), request.getPackageName());
-            }
         } else if (obj instanceof UserPackage userPackage) {
-            if (userPackage.getPackageName() != null) {
+            LogUtil.addMdcLogKey(setAndClearMdc.key(), userPackage.getPackageName());
-                putMdcValue(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());
+            LogUtil.removeMdcLogKey(setAndClearMdc.key());
-        }
-    }
-
-    private void putMdcValue(String key, String value) {
-        try {
-            MDC.put(key, "[" + value + "] ");
-        } catch (Throwable e) {
-            log.error("{}", ExceptionUtils.getStackTrace(e));
         }
     }
 }

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;
 }

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<MonitoringSetupResponse> setupMonitoring(@RequestParam("file") MultipartFile file) {
         signalFacadeV3.setupMonitoringRequest(file);
         return ResponseEntity.ok().body(new MonitoringSetupResponse());

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) {

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<X, Z> {
     Z handle(X taskPayload);
+    default int preferredConcurrency() {return 10;}
 }

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(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getStarter()));
-            executorService.submit(pipeline.getTerminator());
+            executorService.submit(LogUtil.wrap(LogUtil.extractMdcMap(), pipeline.getTerminator()));
 
             pipeline.getTerminator()
                 .doneFuture()

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<X extends StagePayload, Z extends StagePayload> implements Runnable {
@@ -35,12 +36,13 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         this.in = in;
         this.out = out;
         this.dispatcherName = dispatcher.getClass().getSimpleName();
-        this.ioPool = new ThreadPoolExecutor(10,
+        var mdcMap = LogUtil.extractMdcMap();
-            10,
+        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<X extends StagePayload, Z extends StagePayload> 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));

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<String, String> extractMdcMap() {
+        try {
+            return MDC.getCopyOfContextMap();
+        } catch (Throwable e) {
+            log.error(ExceptionUtils.getStackTrace(e));
+            return null;
+        }
+    }
+
+    public static Runnable wrap(Map<String, String> contextMap, Runnable task) {
+        return () -> {
+            Map<String, String> 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));
+                }
+            }
+        };
+    }
+}