treshenkov/credit_tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

pipeline orchestrator [2]

Browse source
This commit is contained in:
ialbert 2025-12-02 14:53:42 +03:00
parent f2f7cfd5a1
commit 90259217b4
5 changed files with 48 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java
View file

@ -7,7 +7,6 @@ import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import javax.xml.stream.XMLStreamException;
import lombok.extern.slf4j.Slf4j;
@ -41,7 +40,8 @@ import ru.nbch.credit_tracker.service.UserPackageService;
import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.credit_tracker.service.task.impl.Saver;
@ -50,7 +50,6 @@ import ru.nbch.credit_tracker.utils.CtFileUtils;
import ru.nbch.credit_tracker.utils.EnumUtil;
import ru.nbch.credit_tracker.utils.LogUtil;
import ru.nbch.credit_tracker.utils.ZipUtils;
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp;
import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp;
@ -62,21 +61,22 @@ public class SignalFacadeV3 {
private final CreditTrackerSettings config;
private final UserPackageService userPackageService;
private final AuthorizationService authorizationService;
private final PipelinePackageManager pipelinePackageManager;
private final PipelinePackageFabric pipelinePackageManager;
private final SignalRestService signalRestService;
private final Saver saver;
private final ExecutorService pipelineExecutorService;
private final ExecutorService saverExecutorService;
private final PipelinePackageOrchestrator pipelineOrchestrator;
public SignalFacadeV3(MonitoringRequestValidator monitoringRequestValidator,
CreditTrackerSettings config,
UserPackageService userPackageService,
AuthorizationService authorizationService,
PipelinePackageManager pipelinePackageManager,
PipelinePackageFabric pipelinePackageManager,
SignalRestService signalRestService,
Saver saver,
ExecutorService pipelineExecutorService,
ExecutorService saverExecutorService) {
ExecutorService saverExecutorService, PipelinePackageOrchestrator pipelineOrchestrator) {
this.monitoringRequestValidator = monitoringRequestValidator;
this.config = config;
this.userPackageService = userPackageService;
@ -86,6 +86,7 @@ public class SignalFacadeV3 {
this.saverExecutorService = saverExecutorService;
this.pipelineExecutorService = pipelineExecutorService;
this.saver = saver;
this.pipelineOrchestrator = pipelineOrchestrator;
}
public void setupMonitoringRequest(MultipartFile file) {
@ -150,18 +151,11 @@ public class SignalFacadeV3 {
Long packageId = userPackageService.registerPackage(userId, packageName);
LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId));
log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name());
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT));
CompletableFuture
.runAsync(() -> {
try {
Pipeline pipeline = pipelinePackageManager.build(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT));
pipelineOrchestrator.startOrDelay(() -> {
saver.readAndSave(packageId, parsedFilePath);
CtFileUtils.deleteQuietly(parsedFilePath);
} catch (Throwable t) {
log.error(ExceptionUtils.getStackTrace(t));
pipeline.close();
}
}, saverExecutorService)
.thenRun(() -> pipeline.start(pipelineExecutorService));
}, pipeline, packageId);
} catch (Exception e) {
try {
Files.delete(parsedFilePath);

34
src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java
View file

@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.service;
import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
@ -13,7 +12,8 @@ import ru.nbch.credit_tracker.enums.Status;
import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
import ru.nbch.credit_tracker.service.task.impl.Saver;
@ -31,7 +31,8 @@ public class UpdateProcessingPackagesService {
private final UserPackageMapper userPackageMapper;
private final PhoneFidMapMapper phoneFidMapMapper;
private final PipelinePackageManager pipelinePackageManager;
private final PipelinePackageFabric pipelinePackageManager;
private final PipelinePackageOrchestrator pipelineOrchestrator;
private final CreditTrackerSettings creditTrackerSettings;
private final Saver saver;
private final ExecutorService pipelineExecutorService;
@ -39,7 +40,7 @@ public class UpdateProcessingPackagesService {
public UpdateProcessingPackagesService(UserPackageMapper userPackageMapper,
PhoneFidMapMapper phoneFidMapMapper,
PipelinePackageManager pipelinePackageManager,
PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator,
CreditTrackerSettings creditTrackerSettings,
Saver saver,
ExecutorService pipelineExecutorService,
@ -47,6 +48,7 @@ public class UpdateProcessingPackagesService {
this.userPackageMapper = userPackageMapper;
this.phoneFidMapMapper = phoneFidMapMapper;
this.pipelinePackageManager = pipelinePackageManager;
this.pipelineOrchestrator = pipelineOrchestrator;
this.creditTrackerSettings = creditTrackerSettings;
this.saver = saver;
this.pipelineExecutorService = pipelineExecutorService;
@ -68,17 +70,11 @@ public class UpdateProcessingPackagesService {
log.trace("Deleted {} rows", count);
Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage);
log.debug("Try to read and build pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath);
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT));
CompletableFuture
.runAsync(() -> {
try {
saver.readAndSave(userPackage.getId(), xmlReqPath);
} catch (Throwable t) {
log.error(ExceptionUtils.getStackTrace(t));
pipeline.close();
}
}, saverExecutorService)
.thenRun(() -> pipeline.start(pipelineExecutorService));
Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT));
pipelineOrchestrator.startOrDelay(
() -> saver.readAndSave(userPackage.getId(), xmlReqPath),
pipeline, userPackage.getId()
);
} catch (Exception e) {
log.error("Error build default pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
}
@ -91,8 +87,8 @@ public class UpdateProcessingPackagesService {
log.debug("Found {} packages in status PROCESSING and rejected_status = null", packages.size());
for (UserPackage userPackage : packages) {
try {
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE));
pipeline.start(pipelineExecutorService);
Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE));
pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
} catch (Exception e) {
log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
}
@ -104,8 +100,8 @@ public class UpdateProcessingPackagesService {
log.debug("Found {} packages in status PROCESSING_REPEATED", packages.size());
for (UserPackage userPackage : packages) {
try {
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE));
pipeline.start(pipelineExecutorService);
Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE));
pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
} catch (Exception e) {
log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
}

13
src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java
View file

@ -14,7 +14,8 @@ import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.Pipeline;
import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
import ru.nbch.credit_tracker.service.task.SetupParam;
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
@ -33,7 +34,8 @@ public class UpdatingFlagsService {
private final Long daysToUpdateFlags;
private final PhoneFidMapMapper phoneFidMapMapper;
private final UserPackageMapper userPackageMapper;
private final PipelinePackageManager pipelinePackageManager;
private final PipelinePackageFabric pipelinePackageManager;
private final PipelinePackageOrchestrator pipelineOrchestrator;
private final ExecutorService pipelineExecutorService;
public UpdatingFlagsService(SignalService signalService,
@ -41,7 +43,7 @@ public class UpdatingFlagsService {
MonitoringErrorProcesingService monitoringErrorProcesingService,
CreditTrackerSettings config,
PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper,
PipelinePackageManager pipelinePackageManager,
PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator,
ExecutorService pipelineExecutorService) {
this.signalService = signalService;
this.signalRestService = signalRestService;
@ -50,6 +52,7 @@ public class UpdatingFlagsService {
this.phoneFidMapMapper = phoneFidMapMapper;
this.userPackageMapper = userPackageMapper;
this.pipelinePackageManager = pipelinePackageManager;
this.pipelineOrchestrator = pipelineOrchestrator;
this.pipelineExecutorService = pipelineExecutorService;
}
@ -65,8 +68,8 @@ public class UpdatingFlagsService {
if (hasOldRow) {
try {
userPackageMapper.updatePackageStatus(userPackage.getId(), Status.PROCESSING_REPEATED.name());
Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT));
pipeline.start(pipelineExecutorService);
Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT));
pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
} catch (Exception e) {
log.error("Failed to build pipeline", e);
}

15
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java → src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java
View file

@ -8,23 +8,19 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
import ru.nbch.credit_tracker.config.task.StageQueueFactory;
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
import ru.nbch.credit_tracker.model.impl.PartFilePayload;
import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
import ru.nbch.credit_tracker.service.MonitoringErrorProcesingService;
import ru.nbch.credit_tracker.service.signal.SignalRestService;
import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher;
import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher;
import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
import ru.nbch.credit_tracker.service.task.impl.Saver;
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;
@Component
@Slf4j
public class PipelinePackageManager implements InitializingBean {
public class PipelinePackageFabric implements InitializingBean {
private final FlagsEnricher flagsEnricherV2;
private final StageQueueFactory<SubjectPayload> idEnricherQueueFactory;
private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory;
@ -36,18 +32,15 @@ public class PipelinePackageManager implements InitializingBean {
private final CreditTrackerSettings creditTrackerSettings;
private final PipelineBackoff pipelineBackoff;
public PipelinePackageManager(SignalRestService signalRestService,
@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
public PipelinePackageFabric(@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
@Qualifier("nameEnricherQueueFactory") StageQueueFactory<SubjectPayload> nameEnricherQueueFactory,
@Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
UserPackageMapper userPackageMapper,
FidIdEnricher fidIdEnricherHandler,
Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher,
CreditTrackerSettings creditTrackerSettings,
PipelineBackoff pipelineBackoff, Saver saver,
MonitoringErrorProcesingService monitoringErrorProcesingService) {
PipelineBackoff pipelineBackoff) {
this.idEnricherQueueFactory = idEnricherQueueFactory;
this.nameEnricherQueueFactory = nameEnricherQueueFactory;
this.sendSignalQueueFactory = sendSignalQueueFactory;
@ -60,7 +53,7 @@ public class PipelinePackageManager implements InitializingBean {
this.pipelineBackoff = pipelineBackoff;
}
public Pipeline buildAndRun(SetupParam setupParam) throws Exception {
public Pipeline build(SetupParam setupParam) throws Exception {
return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
.handler(fidNameEnricher)
.handlerResult(flagsEnricherV2)

5
src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java
View file

@ -62,7 +62,7 @@ public class PipelinePackageOrchestrator {
innerStartOrDelay(before, pipeline, packageId);
}
public void innerStartOrDelay(Runnable before, Pipeline pipeline, Long packageId) {
private void innerStartOrDelay(Runnable before, Pipeline pipeline, Long packageId) {
try {
if (before != null) {
CompletableFuture
@ -89,7 +89,7 @@ public class PipelinePackageOrchestrator {
}
}
public void addCallback(Pipeline pipeline, Long packageId) {
private void addCallback(Pipeline pipeline, Long packageId) {
pipeline.getTerminator()
.doneFuture()
.orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS)
@ -134,7 +134,6 @@ public class PipelinePackageOrchestrator {
return;
}
}
log.info("delayed pipeline found: packageId={}. continue work...", delayedLaunch.packageId);
innerStartOrDelay(delayedLaunch.before, delayedLaunch.pipeline, delayedLaunch.packageId);
}