http://jira.mfd.msk:8088/browse/BKI-3215 Use MDC for monitoring, updating flags, setup monitoring; Update ValidationUtil to a service

src/main/java/ru/nbch/credit_tracker/config/mdc/MdcAspect.java

@@ -21,11 +21,11 @@ public class MdcAspect {
         Object obj = (args != null && args.length > index) ? args[index] : null;
         if (obj instanceof MonitoringRequest request) {
             if (request.getPackageName() != null) {
-                putMdcValue(request.getPackageName(), setAndClearMdc.key());
+                putMdcValue(setAndClearMdc.key(), request.getPackageName());
             }
         } else if (obj instanceof UserPackage userPackage) {
             if (userPackage.getPackageName() != null) {
-                putMdcValue(userPackage.getPackageName(), setAndClearMdc.key());
+                putMdcValue(setAndClearMdc.key(), userPackage.getPackageName());
             }
         } else if (obj instanceof String packageName) {
             putMdcValue(setAndClearMdc.key(), packageName);
@@ -39,7 +39,7 @@ public class MdcAspect {
         }
     }
 
-    private void putMdcValue(String value, String key) {
+    private void putMdcValue(String key, String value) {
         try {
             MDC.put(key, "[" + value + "] ");
         } catch (Throwable e) {

src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java

@@ -1,12 +1,16 @@
 package ru.nbch.credit_tracker.facade;
 
 import lombok.extern.slf4j.Slf4j;
+import org.slf4j.MDC;
+import org.springframework.aop.framework.AopContext;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.core.io.FileUrlResource;
 import org.springframework.core.io.Resource;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
+import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.dto.UserPackage;
 import ru.nbch.credit_tracker.entities.user_request.active_packages.MonitoringActivePackagesRequest;
@@ -31,7 +35,7 @@ import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 import ru.nbch.credit_tracker.utils.CTUtil;
-import ru.nbch.credit_tracker.utils.ValidationUtil;
+import ru.nbch.credit_tracker.utils.ValidationService;
 import ru.nbch.credit_tracker.utils.ZipUtils;
 
 import javax.xml.stream.XMLStreamException;
@@ -43,6 +47,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
@@ -54,12 +59,17 @@ import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp;
 
 @Slf4j
 @Service
+@EnableAspectJAutoProxy(exposeProxy = true)
 public class SignalFacade {
+
+    private SignalFacade self;
+
     private final CreditTrackerSettings config;
     private final AuthorizationService authorizationService;
     private final SignalService signalService;
     private final SignalRestService signalRestService;
     private final MonitoringErrorProcesingService monitoringErrorProcesingService;
+    private final ValidationService validationService;
     private final StaxXmlProcessor staxXmlProcessor;
 
     private final ExecutorService executor;
@@ -71,6 +81,7 @@ public class SignalFacade {
                         SignalService signalService,
                         SignalRestService signalRestService,
                         MonitoringErrorProcesingService monitoringErrorProcesingService,
+                        ValidationService validationService,
                         StaxXmlProcessor staxXmlProcessor,
                         @Qualifier("executorService") ExecutorService executor) {
         this.config = config;
@@ -78,6 +89,7 @@ public class SignalFacade {
         this.signalService = signalService;
         this.signalRestService = signalRestService;
         this.monitoringErrorProcesingService = monitoringErrorProcesingService;
+        this.validationService = validationService;
         this.staxXmlProcessor = staxXmlProcessor;
         this.executor = executor;
         this.maxXmlFileSize = config.getMaxXmlFileSize() * 1024 * 1024;
@@ -164,7 +176,7 @@ public class SignalFacade {
         }
 
         // в отдельном потоке запускаем регистрацию субъектов, поиск фидов и расчет флагов просрочек
-        registerFidsAndDebts(packageId, monitoringRequest.getPackageName(), filePath);
+        getCurrentProxy().registerFidsAndDebts(packageId, monitoringRequest.getPackageName(), filePath);
 
         return ResponseEntity.ok().body(new MonitoringSetupResponse());
     }
@@ -175,7 +187,7 @@ public class SignalFacade {
             monitoringRequest = staxXmlProcessor.read(in, MonitoringRequest.class,
                     request -> {
                         try {
-                            ValidationUtil.validateMonitoringRequest(request);
+                            validationService.validateMonitoringRequest(request);
                         } catch (SignalClientException e) {
                             if (request != null) {
                                 signalService.rejectPackage(request);
@@ -211,8 +223,13 @@ public class SignalFacade {
         }
     }
 
-    private void registerFidsAndDebts(Integer packageId, String packageName, String filePath) {
+    @SetAndClearMdc(argIndex = 1)
+    public void registerFidsAndDebts(Integer packageId, String packageName, String filePath) {
+        Map<String, String> context = MDC.getCopyOfContextMap();
         executor.execute(() -> {
+            if (context != null) {
+                MDC.setContextMap(context);
+            }
             try {
                 signalService.registerSubjects(filePath, packageId);
                 CTUtil.removeFile(filePath);
@@ -408,4 +425,11 @@ public class SignalFacade {
             throw serverExcp(SignalError.ERROR_013, XmlErrorFormat.MonitoringDownloadError, e);
         }
     }
+
+    private SignalFacade getCurrentProxy() {
+        if (self == null) {
+            self = (SignalFacade) AopContext.currentProxy();
+        }
+        return self;
+    }
 }

src/main/java/ru/nbch/credit_tracker/service/MonitoringErrorProcesingService.java

@@ -2,15 +2,21 @@ package ru.nbch.credit_tracker.service;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.aop.framework.AopContext;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc;
 import ru.nbch.credit_tracker.entities.dto.UserPackage;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 import ru.nbch.credit_tracker.utils.CTUtil;
 
 @Slf4j
 @Service
+@EnableAspectJAutoProxy(exposeProxy = true)
 public class MonitoringErrorProcesingService {
 
+    private MonitoringErrorProcesingService self;
+
     private final SignalService signalService;
     private final SignalRestService signalRestService;
 
@@ -21,6 +27,11 @@ public class MonitoringErrorProcesingService {
 
     public void findErrors(Integer packageId) {
         UserPackage userPackage = signalService.findPackage(packageId);
+        getCurrentProxy().processPackage(userPackage, packageId);
+    }
+
+    @SetAndClearMdc
+    public void processPackage(UserPackage userPackage, Integer packageId) {
         if (userPackage != null) {
             String packCode = userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName();
             log.info("Проверка на ошибки при постановке на мониторинг. packCode={}, packageId={}", packCode, packageId);
@@ -38,4 +49,11 @@ public class MonitoringErrorProcesingService {
             }
         }
     }
+
+    private MonitoringErrorProcesingService getCurrentProxy() {
+        if (self == null) {
+            self = (MonitoringErrorProcesingService) AopContext.currentProxy();
+        }
+        return self;
+    }
 }

src/main/java/ru/nbch/credit_tracker/service/MonitoringService.java

@@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.service;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.MDC;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@@ -69,83 +70,89 @@ public class MonitoringService {
         log.info("Start monitoring");
         List<UserPackage> packages = signalService.findPackagesOnMonitoring();
         for (UserPackage userPackage : packages) {
-            log.info("Processing package {}", userPackage.getPackageName());
+            MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() +"]");
-            if (!signalService.hasNonNullfids(userPackage.getId())) {
+            processPackage(userPackage);
-                log.info("Package {} has no fids. Package will be skipped.", userPackage.getPackageName());
+            MDC.remove("ru.nbch.credit_tracker.log.mdc.key");
-                continue;
-            }
-            SgnlOnlineReport onlineReports = signalRestService.getReports(userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName());
-
-            if (onlineReports == null) {
-                log.info("No reports found for package {}", userPackage.getPackageName());
-                continue;
-            }
-
-            // Формируем список имен файлов, по которым сработали сигналы. Сюда входят только те имена, у которых парсится дата формирования отчета
-            List<String> fileNames = onlineReports.getOperationReport().getReports().getReports().stream().map(Report::getFilename)
-                    .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName) != null).toList();
-
-            if (fileNames.isEmpty()) {
-                log.info("No new report found for package {}", userPackage.getPackageName());
-                continue;
-            }
-
-            LocalDateTime lastDownloadedReportDate = CTUtil.parseDateFromReportFileName(userPackage.getLastDownloadedReport());
-
-            if (lastDownloadedReportDate != null) {
-                // Отбираем только те отчеты, дата которых позже даты последнего загруженного отчета
-                fileNames = fileNames.stream()
-                        .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName).isAfter(lastDownloadedReportDate)) // отфильтровали только те имена, которые парсятся в дату, так что null тут не придет
-                        .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате
-                        .toList();
-            } else {
-                fileNames = fileNames.stream()
-                        .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате
-                        .toList();
-            }
-
-            if (StringUtils.isNotBlank(userPackage.getLastDownloadedReport())) {
-                log.info("Last downloaded report found for package {}. Last Downloaded Report Name: {}", userPackage.getPackageName(), userPackage.getLastDownloadedReport());
-            } else {
-                log.info("Last Downloaded Report Name is empty");
-            }
-
-            for (String newReport : fileNames) {
-                try {
-                    String loadedReportPath = null;
-                    if (StringUtils.isBlank(userPackage.getLastDownloadedReport())) {
-                        loadedReportPath = loadReport(userPackage.getId(), newReport); // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным
-                    } else if (lastDownloadedReportDate != null) {
-                        LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(newReport);
-                        if (newReportDate.isAfter(lastDownloadedReportDate)) {
-                            loadedReportPath = loadReport(userPackage.getId(), newReport); // полученный отчет новее записанного в базу
-                        }
-                    }
-                    if (loadedReportPath == null) {
-                        log.info("Report was not loaded for package {}", userPackage.getPackageName());
-                        continue;
-                    }
-
-                    String responsePath = generateResponse(userPackage, loadedReportPath);
-                    String fileName = null;
-                    if (responsePath != null) {
-                        fileName = createResponseFile(userPackage, responsePath);
-                    }
-                    if (fileName != null) {
-                        log.info("New report fileName: {}", fileName);
-                        updateSignalHits(userPackage, responsePath, fileName);
-                        signalService.updateLastDownloadedReport(userPackage.getId(), newReport);
-                    } else {
-                        log.info("Report was not created for package {}", userPackage.getPackageName());
-                    }
-                } catch (Exception e) {
-                    log.error(e.getMessage(), e);
-                }
-            }
         }
         log.info("Finish monitoring");
     }
 
+    private void processPackage(UserPackage userPackage) {
+        log.info("Processing package");
+        if (!signalService.hasNonNullfids(userPackage.getId())) {
+            log.info("Package has no fids. Package will be skipped.");
+            return;
+        }
+        SgnlOnlineReport onlineReports = signalRestService.getReports(userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName());
+
+        if (onlineReports == null) {
+            log.info("No reports found for package");
+            return;
+        }
+
+        // Формируем список имен файлов, по которым сработали сигналы. Сюда входят только те имена, у которых парсится дата формирования отчета
+        List<String> fileNames = onlineReports.getOperationReport().getReports().getReports().stream().map(Report::getFilename)
+                .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName) != null).toList();
+
+        if (fileNames.isEmpty()) {
+            log.info("No new report found for package");
+            return;
+        }
+
+        LocalDateTime lastDownloadedReportDate = CTUtil.parseDateFromReportFileName(userPackage.getLastDownloadedReport());
+
+        if (lastDownloadedReportDate != null) {
+            // Отбираем только те отчеты, дата которых позже даты последнего загруженного отчета
+            fileNames = fileNames.stream()
+                    .filter(fileName -> CTUtil.parseDateFromReportFileName(fileName).isAfter(lastDownloadedReportDate)) // отфильтровали только те имена, которые парсятся в дату, так что null тут не придет
+                    .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате
+                    .toList();
+        } else {
+            fileNames = fileNames.stream()
+                    .sorted(Comparator.comparing(CTUtil::parseDateFromReportFileName)) // Сортируем по дате
+                    .toList();
+        }
+
+        if (StringUtils.isNotBlank(userPackage.getLastDownloadedReport())) {
+            log.info("Last downloaded report found for package. Last Downloaded Report Name: {}", userPackage.getLastDownloadedReport());
+        } else {
+            log.info("Last Downloaded Report Name is empty");
+        }
+
+        for (String newReport : fileNames) {
+            try {
+                String loadedReportPath = null;
+                if (StringUtils.isBlank(userPackage.getLastDownloadedReport())) {
+                    loadedReportPath = loadReport(userPackage.getId(), newReport); // по этому пакету еще не было загруженных отчетов, продолжаем работу с полученным
+                } else if (lastDownloadedReportDate != null) {
+                    LocalDateTime newReportDate = CTUtil.parseDateFromReportFileName(newReport);
+                    if (newReportDate.isAfter(lastDownloadedReportDate)) {
+                        loadedReportPath = loadReport(userPackage.getId(), newReport); // полученный отчет новее записанного в базу
+                    }
+                }
+                if (loadedReportPath == null) {
+                    log.info("Report was not loaded for package");
+                    continue;
+                }
+
+                String responsePath = generateResponse(userPackage, loadedReportPath);
+                String fileName = null;
+                if (responsePath != null) {
+                    fileName = createResponseFile(userPackage, responsePath);
+                }
+                if (fileName != null) {
+                    log.info("New report fileName: {}", fileName);
+                    updateSignalHits(userPackage, responsePath, fileName);
+                    signalService.updateLastDownloadedReport(userPackage.getId(), newReport);
+                } else {
+                    log.info("Report was not created for package");
+                }
+            } catch (Exception e) {
+                log.error(e.getMessage(), e);
+            }
+        }
+    }
+
     /**
      * Формирует результирующий xml файл для конечного пользователя на основе файла из официального сервиса сигналов
      * @return Путь к временному рещультирующему файлу
@@ -297,7 +304,7 @@ public class MonitoringService {
                     10000,
                     "/MonitoringSignalResponse/Subjects/Subject");
         } catch (IOException e) {
-            throw new RuntimeException("Error while and updating hits", e);
+            throw new RuntimeException("Error while updating hits", e);
         }
 
         CTUtil.removeFile(responsePath); // удалям результирующий xml файл с диска (Он уже заархивирован к этому моменту)

src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java

@@ -1,7 +1,10 @@
 package ru.nbch.credit_tracker.service;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.aop.framework.AopContext;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc;
 import ru.nbch.credit_tracker.entities.dto.UserPackage;
 import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
@@ -14,8 +17,11 @@ import java.util.List;
  */
 @Slf4j
 @Service
+@EnableAspectJAutoProxy(exposeProxy = true)
 public class UpdateProcessingPackagesService {
 
+    private UpdateProcessingPackagesService self;
+
     private final SignalService signalService;
     private final SignalRestService signalRestService;
     private final MonitoringErrorProcesingService monitoringErrorProcesingService;
@@ -33,25 +39,26 @@ public class UpdateProcessingPackagesService {
         List<UserPackage> packages = signalService.findPackagesOnProcessing();
         for (UserPackage userPackage : packages) {
             log.info("Updating flags for Package {}", userPackage.getPackageName());
-            process(userPackage.getId(), userPackage.getPackageName());
+            getCurrentProxy().process(userPackage.getId(), userPackage.getPackageName());
         }
         log.info("Finish updating processing packages");
     }
 
-    private void process(Integer packageId, String packageName) {
+    @SetAndClearMdc(argIndex = 1)
+    public void process(Integer packageId, String packageName) {
         try {
             SignalService.UpdateResult updateResult = signalService.updateSubjects(packageId, null);
             if (updateResult.isNoFidsFound()) {
-                log.info("No fids found for PackageId: {} PackageName: {}. Package status ON_MONITORING. No registration in signal service.", packageId, packageName);
+                log.info("No fids found for PackageId: {} . Package status ON_MONITORING. No registration in signal service.", packageId);
                 signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
                 return;
             }
             if (signalRestService.setupMonitoring(packageId)) {
-                log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName);
+                log.info("Package was successfully registered in signal service. PackageId: {}", packageId);
                 signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
                 monitoringErrorProcesingService.findErrors(packageId);
             } else {
-                log.info("Package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName);
+                log.info("Package registration was rejected by signal service. PackageId: {}", packageId);
                 signalService.updatePackageStatus(packageId, Status.ERROR);
             }
         } catch (Exception e) {
@@ -59,4 +66,11 @@ public class UpdateProcessingPackagesService {
             log.error("Ошибка при поиске фидов или расчете флагов просрочек. Package_id {}", packageId, e);
         }
     }
+
+    private UpdateProcessingPackagesService getCurrentProxy() {
+        if (self == null) {
+            self = (UpdateProcessingPackagesService) AopContext.currentProxy();
+        }
+        return self;
+    }
 }

src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java

@@ -1,6 +1,7 @@
 package ru.nbch.credit_tracker.service;
 
 import lombok.extern.slf4j.Slf4j;
+import org.slf4j.MDC;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@@ -52,7 +53,9 @@ public class UpdatingFlagsService {
         List<UserPackage> packages = signalService.findPackagesOnMonitoring();
         for (UserPackage userPackage : packages) {
             log.info("Updating flags for Package {}", userPackage.getPackageName());
+            MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() +"]");
             updateFlags(userPackage.getId(), userPackage.getPackageName());
+            MDC.remove("ru.nbch.credit_tracker.log.mdc.key");
         }
         log.info("Finish updating flags");
     }

src/main/java/ru/nbch/credit_tracker/utils/ValidationService.java

@@ -1,6 +1,8 @@
 package ru.nbch.credit_tracker.utils;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.Subject;
 import ru.nbch.credit_tracker.enums.SignalError;
@@ -12,13 +14,15 @@ import java.util.regex.Pattern;
 import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp;
 
 @Slf4j
-public class ValidationUtil {
+@Service
+public class ValidationService {
 
     private static final  String phoneRegex = "^\\d{9}$";
     private static final  String packageNameRegex = "^[A-Z0-9_]{1,22}$";
     private static final  String npIdRegex = "^\\S{1,36}$";
 
-    public static void validateMonitoringRequest(MonitoringRequest request) throws SignalClientException {
+    @SetAndClearMdc
+    public void validateMonitoringRequest(MonitoringRequest request) throws SignalClientException {
         if (request == null) {
             throw clientExcp(SignalError.ERROR_011, XmlErrorFormat.MonitoringError);
         }

src/test/java/ru/nbch/credit_tracker/utils/ValidationServiceTest.java

@@ -28,14 +28,17 @@ import java.util.zip.ZipInputStream;
 
 import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp;
 
-@SpringBootTest(classes = {ValidationUtilTest.TestConfig.class})
+@SpringBootTest(classes = {ValidationServiceTest.TestConfig.class})
-public class ValidationUtilTest {
+public class ValidationServiceTest {
 
     private final StaxXmlProcessor staxXmlProcessor;
+    private final ValidationService validationService;
 
     @Autowired
-    public ValidationUtilTest(StaxXmlProcessor staxXmlProcessor) {
+    public ValidationServiceTest(StaxXmlProcessor staxXmlProcessor,
+                                 ValidationService validationService) {
         this.staxXmlProcessor = staxXmlProcessor;
+        this.validationService = validationService;
     }
 
     @Configuration
@@ -46,6 +49,11 @@ public class ValidationUtilTest {
             staxReader.add(MonitoringRequestStaxReader::new);
             return new XmlProcessorStaxImpl(staxReader);
         }
+
+        @Bean
+        public ValidationService validationService() {
+            return new ValidationService();
+        }
     }
 
     private static Stream<Arguments> validationTestCases() {
@@ -107,7 +115,7 @@ public class ValidationUtilTest {
 
         if(ex == null) {
             try {
-                ValidationUtil.validateMonitoringRequest(monitoringRequest);
+                validationService.validateMonitoringRequest(monitoringRequest);
             } catch (SignalClientException e) {
                 ex = e;
             }