diff --git a/fr-batch-service/docs/ADR/001-dynamic-job-loading-system.md b/fr-batch-service/docs/ADR/001-dynamic-job-loading-system.md index 26396365..fdbb1524 100644 --- a/fr-batch-service/docs/ADR/001-dynamic-job-loading-system.md +++ b/fr-batch-service/docs/ADR/001-dynamic-job-loading-system.md @@ -351,6 +351,13 @@ CREATE TABLE job_executions_audit ( ); ``` +`job_executions_audit` has a single writer: `JobExecutionAuditListener`, which +`PluginRegistryService` appends to every job (static and dynamic) right before +`JobRegistry.register(job)`, so every launch path, sync or async, is audited. `beforeJob` inserts +the row with the start time; `afterJob` adds the finish time, duration, and the outcome as the raw +`BatchStatus` name (`COMPLETED`, `FAILED`, `STOPPED`, ...). Jobs without a `job_definitions` row +are not audited, and audit failures never fail the job. + ### 3. Dynamic ClassLoader Strategy **Parent-Last Delegation Model:** @@ -1235,7 +1242,8 @@ results. Fixed in `DynamicJobLoaderService.unloadJob()` with try-catch guard. Se unit tests - [x] **Audit service + JobExecutionListener** — `AuditEventType` enum (8 lifecycle events), `AuditEvent` record, `AuditService` with SLF4J structured JSON logging + MDC correlation IDs; - `JobExecutionAuditListener` populates `job_executions_audit` on job start/end; `createdBy` + `JobExecutionAuditListener` populates `job_executions_audit` on job start/end (attached to every + job at registration since #111, the table's only writer); `createdBy` populated via `SecurityContext` resolver with `"system"` fallback; wired into `JarUploadService`, `DynamicJobLoaderService`, `JobManagementController` - [x] **8 Snyk CVEs resolved** — `spring-boot-starter-security` fixed `CVE-2024-38807` (DoS in diff --git a/fr-batch-service/docs/job-plugins/developer-guide.md b/fr-batch-service/docs/job-plugins/developer-guide.md index 7ef59aa8..32e1592f 100644 --- a/fr-batch-service/docs/job-plugins/developer-guide.md +++ b/fr-batch-service/docs/job-plugins/developer-guide.md @@ -333,6 +333,16 @@ curl -u admin:admin123 \ | `execution_attempt_number` | Yes | Unique attempt number | | `description` | No | Non-identifying descriptor | +**Execution audit history**: every launch path above (single or all, sync or async) writes exactly +one `job_executions_audit` row per job execution. The only writer is `JobExecutionAuditListener`, +which the platform attaches to every plugin job at registration time (alongside any listeners the +plugin attached itself). The row records the start time when the job starts, then the finish time, +duration in milliseconds, and the outcome, which is the execution's Spring Batch `BatchStatus` name +(`COMPLETED`, `FAILED`, `STOPPED`, ...). Jobs without a `job_definitions` row (for example, +classpath-only jobs) are not audited. Audit write failures are logged and never fail the job. A +plugin whose `configureJob(...)` returns a `Job` that is not a Spring Batch `AbstractJob` still +registers, but its executions are not audited and a `WARN` is logged. + ______________________________________________________________________ ## JAR Signing Configuration diff --git a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/JobsManagerService.java b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/JobsManagerService.java index 504b6020..434e7ad6 100644 --- a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/JobsManagerService.java +++ b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/JobsManagerService.java @@ -3,16 +3,10 @@ import com.fronzec.api.BatchJobPlugin; import com.fronzec.frbatchservice.batchjobs.plugins.PluginRegistryService; -import com.fronzec.frbatchservice.batchjobs.plugins.entity.JobDefinitionEntity; -import com.fronzec.frbatchservice.batchjobs.plugins.entity.JobExecutionAuditEntity; import com.fronzec.frbatchservice.batchjobs.plugins.metrics.PluginMetrics; -import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobDefinitionRepository; -import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobExecutionAuditRepository; import com.fronzec.frbatchservice.utils.JsonUtils; import com.fronzec.frbatchservice.web.SingleJobDataRequest; -import java.time.Duration; import java.time.LocalDate; -import java.time.LocalDateTime; import java.util.HashMap; import java.util.Map; import java.util.Objects; @@ -44,8 +38,6 @@ public class JobsManagerService { private final PluginRegistryService pluginRegistryService; private final JobRepository jobRepository; private final PluginMetrics pluginMetrics; - private final JobDefinitionRepository jobDefinitionRepository; - private final JobExecutionAuditRepository auditRepository; public JobsManagerService( JobOperator asyncJobLauncher, @@ -53,17 +45,13 @@ public JobsManagerService( JobRegistry jobRegistry, PluginRegistryService pluginRegistryService, JobRepository jobRepository, - PluginMetrics pluginMetrics, - JobDefinitionRepository jobDefinitionRepository, - JobExecutionAuditRepository auditRepository) { + PluginMetrics pluginMetrics) { this.asyncJobLauncher = asyncJobLauncher; this.syncJobLauncher = syncJobLauncher; this.jobRegistry = jobRegistry; this.pluginRegistryService = pluginRegistryService; this.jobRepository = jobRepository; this.pluginMetrics = pluginMetrics; - this.jobDefinitionRepository = jobDefinitionRepository; - this.auditRepository = auditRepository; } public HashMap launchAllNonAutoDetectedJobsAsync( @@ -243,7 +231,6 @@ public Map syncRunJobWithParams(SingleJobDataRequest request) { "parameters", JsonUtils.parseObject2Json(jobParametersBuilder.toJobParameters())); launchedJobMetadata.put("result", run.getExitStatus().toString()); - saveAudit(run, request.getJobBeanName()); } catch (JobExecutionAlreadyRunningException e) { logger.error("already running job with the same params", e); launchedJobMetadata.put("error", e.getMessage()); @@ -304,25 +291,4 @@ public HashMap> getRunning() { }); return running; } - - private void saveAudit(JobExecution execution, String jobName) { - try { - JobDefinitionEntity def = jobDefinitionRepository.findByJobName(jobName).orElse(null); - if (def == null) return; - JobExecutionAuditEntity entity = new JobExecutionAuditEntity(); - entity.setJobDefinitionId(def.getId()); - entity.setJobExecutionId(execution.getId()); - entity.setJobVersion(def.getVersion()); - long durationMs = Duration.between(execution.getStartTime(), - execution.getEndTime() != null ? execution.getEndTime() : LocalDateTime.now()) - .toMillis(); - entity.setExecutionMetadata( - "{\"outcome\":\"" + execution.getExitStatus().getExitCode() + "\"," - + "\"durationMs\":" + durationMs + "}"); - auditRepository.save(entity); - logger.debug("Audit saved for job '{}': {}", jobName, execution.getExitStatus().getExitCode()); - } catch (Exception e) { - logger.warn("Failed to save audit for '{}': {}", jobName, e.getMessage()); - } - } } diff --git a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java index 4f8ef71d..f95acbae 100644 --- a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java +++ b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryService.java @@ -2,6 +2,7 @@ import com.fronzec.api.BatchJobPlugin; import com.fronzec.frbatchservice.batchjobs.plugins.entity.JobDefinitionEntity; +import com.fronzec.frbatchservice.batchjobs.plugins.listener.JobExecutionAuditListener; import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobDefinitionRepository; import jakarta.annotation.PostConstruct; import java.util.Collection; @@ -16,6 +17,7 @@ import org.slf4j.LoggerFactory; import org.springframework.batch.core.configuration.DuplicateJobException; import org.springframework.batch.core.configuration.JobRegistry; +import org.springframework.batch.core.job.AbstractJob; import org.springframework.batch.core.job.Job; import org.springframework.batch.core.repository.JobRepository; import org.springframework.context.ApplicationContext; @@ -28,6 +30,9 @@ * *

Registration runs in {@link PostConstruct}, which Spring guarantees completes before any * consumer bean that constructor-injects this service is made available to the HTTP layer. + * + *

Every job, static or dynamic, receives the {@link JobExecutionAuditListener} right before it + * enters the {@link JobRegistry}; that listener is the single writer of execution audit history. */ @Service public class PluginRegistryService { @@ -40,6 +45,7 @@ public class PluginRegistryService { private final PlatformTransactionManager transactionManager; private final ApplicationContext applicationContext; private final JobDefinitionRepository jobDefinitionRepository; + private final JobExecutionAuditListener auditListener; /** Built during {@link PostConstruct} and dynamic registration; mutable. */ private final Map pluginsByJobName = new LinkedHashMap<>(); @@ -51,13 +57,15 @@ public PluginRegistryService( JobRepository jobRepository, PlatformTransactionManager transactionManager, ApplicationContext applicationContext, - JobDefinitionRepository jobDefinitionRepository) { + JobDefinitionRepository jobDefinitionRepository, + JobExecutionAuditListener auditListener) { this.plugins = plugins; this.jobRegistry = jobRegistry; this.jobRepository = jobRepository; this.transactionManager = transactionManager; this.applicationContext = applicationContext; this.jobDefinitionRepository = jobDefinitionRepository; + this.auditListener = auditListener; } @PostConstruct @@ -101,7 +109,8 @@ public void registerAll() { e); } - // 3. Register into the shared JobRegistry + // 3. Attach the execution audit listener, then register into the shared JobRegistry + attachAuditListener(job); try { jobRegistry.register(job); } catch (DuplicateJobException e) { @@ -204,6 +213,7 @@ public void registerDynamicPlugin(BatchJobPlugin plugin) { e); } + attachAuditListener(job); try { jobRegistry.register(job); } catch (DuplicateJobException e) { @@ -221,6 +231,31 @@ public void registerDynamicPlugin(BatchJobPlugin plugin) { } } + /** + * Appends the shared {@link JobExecutionAuditListener} to the job so every launch path writes + * its execution audit row through one writer. + * + *

Uses {@link AbstractJob#registerJobExecutionListener}, which appends and keeps the + * listeners the plugin attached itself (never {@code setJobExecutionListeners}, which replaces + * them). Spring Batch ignores a listener that is already registered on the same job instance, + * so re-registering a job cannot attach the audit listener twice. A {@link Job} that is not an + * {@link AbstractJob} cannot accept listeners: it is still registered, but its executions are + * not audited. + * + * @param job the job produced by the plugin's {@code configureJob(...)} + */ + private void attachAuditListener(Job job) { + if (job instanceof AbstractJob abstractJob) { + abstractJob.registerJobExecutionListener(auditListener); + } else { + log.warn( + "Job '{}' ({}) is not an AbstractJob; registering it without the execution audit" + + " listener, so its executions will not be audited", + job.getName(), + job.getClass().getName()); + } + } + /** * Unregisters a dynamically-registered plugin from the internal map and the * shared {@link JobRegistry}. diff --git a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/listener/JobExecutionAuditListener.java b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/listener/JobExecutionAuditListener.java index b15b463f..a8628799 100644 --- a/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/listener/JobExecutionAuditListener.java +++ b/fr-batch-service/src/main/java/com/fronzec/frbatchservice/batchjobs/plugins/listener/JobExecutionAuditListener.java @@ -9,25 +9,30 @@ import java.time.LocalDateTime; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.job.JobExecution; import org.springframework.batch.core.listener.JobExecutionListener; -import org.springframework.context.event.EventListener; import org.springframework.stereotype.Component; /** * Spring Batch listener that captures job execution lifecycle events into the * {@code job_executions_audit} table. * + *

This listener is the single writer of execution audit history. {@code PluginRegistryService} + * attaches it to every plugin job right before the job enters the {@code JobRegistry}, so every + * launch path (synchronous or asynchronous, single job or launch-all) is audited the same way. + * *

Each execution pair (before → after) produces one audit row: *

    *
  • {@link #beforeJob(JobExecution)} — creates the row with start metadata
    • - *
  • {@link #afterJob(JobExecution)} — updates the row with outcome, duration
    • + *
  • {@link #afterJob(JobExecution)} — updates the row with finish time, duration, and the + * execution's raw {@link org.springframework.batch.core.BatchStatus} name as the outcome
    • *
* *

The {@link JobDefinitionEntity} is resolved by the job name from the execution's * job instance. If no matching definition exists (e.g., a classpath-registered job - * with no DB row), the execution is silently skipped. + * with no DB row), the execution is not audited: the audit row requires a definition. + * + *

Audit persistence is best-effort: any failure is logged and never fails the job. */ @Component public class JobExecutionAuditListener implements JobExecutionListener { @@ -40,6 +45,7 @@ public class JobExecutionAuditListener implements JobExecutionListener { private final JobExecutionAuditRepository auditRepository; private final JobDefinitionRepository jobDefinitionRepository; + /** Creates a listener that persists audit rows for jobs with a matching job definition. */ public JobExecutionAuditListener( JobExecutionAuditRepository auditRepository, JobDefinitionRepository jobDefinitionRepository) { @@ -47,38 +53,61 @@ public JobExecutionAuditListener( this.jobDefinitionRepository = jobDefinitionRepository; } + /** + * Creates the audit row for a starting execution and stores its ID in the execution context. + * + *

Any audit ID inherited from a previous execution (restarts copy the execution context) is + * removed first, so {@link #afterJob(JobExecution)} can never update another execution's row. + * + * @param jobExecution the starting execution + */ @Override public void beforeJob(JobExecution jobExecution) { + jobExecution.getExecutionContext().remove(AUDIT_ENTITY_ID_KEY); String jobName = jobExecution.getJobInstance().getJobName(); + try { + // Only audit plugin-managed definitions (resolvable by name in our registry) + JobDefinitionEntity def = + jobDefinitionRepository.findByJobName(jobName).orElse(null); + if (def == null) { + log.debug( + "No job definition found for '{}' — skipping execution audit", jobName); + return; + } + + JobExecutionAuditEntity entity = new JobExecutionAuditEntity(); + entity.setJobDefinitionId(def.getId()); + entity.setJobExecutionId(jobExecution.getId()); + entity.setJobVersion(def.getVersion()); + entity.setExecutionMetadata( + metadataJson(jobExecution.getStartTime(), null, null, null)); + + JobExecutionAuditEntity saved = auditRepository.save(entity); + + // Stash the row ID in the execution context so afterJob can find it + jobExecution.getExecutionContext().put(AUDIT_ENTITY_ID_KEY, saved.getId()); - // Only audit plugin-managed definitions (resolvable by name in our registry) - JobDefinitionEntity def = - jobDefinitionRepository.findByJobName(jobName).orElse(null); - if (def == null) { log.debug( - "No job definition found for '{}' — skipping execution audit", jobName); - return; + "Audit row created: id={}, jobName={}, executionId={}", + saved.getId(), + jobName, + jobExecution.getId()); + } catch (RuntimeException e) { + log.warn( + "Failed to create execution audit row for job '{}' (executionId={}); the job continues" + + " unaudited", + jobName, + jobExecution.getId(), + e); } - - JobExecutionAuditEntity entity = new JobExecutionAuditEntity(); - entity.setJobDefinitionId(def.getId()); - entity.setJobExecutionId(jobExecution.getId()); - entity.setJobVersion(def.getVersion()); - entity.setExecutionMetadata( - metadataJson(jobExecution.getStartTime(), null, null, null)); - - JobExecutionAuditEntity saved = auditRepository.save(entity); - - // Stash the row ID in the execution context so afterJob can find it - jobExecution.getExecutionContext().put(AUDIT_ENTITY_ID_KEY, saved.getId()); - - log.debug( - "Audit row created: id={}, jobName={}, executionId={}", - saved.getId(), - jobName, - jobExecution.getId()); } + /** + * Completes the audit row created by {@link #beforeJob(JobExecution)} with the finish time, + * duration, and outcome. Spring Batch calls this for completed, failed, and stopped executions. + * + * @param jobExecution the finished execution + */ @Override public void afterJob(JobExecution jobExecution) { Object rawId = jobExecution.getExecutionContext().get(AUDIT_ENTITY_ID_KEY); @@ -89,53 +118,51 @@ public void afterJob(JobExecution jobExecution) { return; } - auditRepository - .findById(entityId) - .ifPresentOrElse( - entity -> { - long durationMs = - Duration.between( - jobExecution.getStartTime(), - jobExecution.getEndTime() != null - ? jobExecution.getEndTime() - : jobExecution.getStartTime()) - .toMillis(); - String outcome = mapOutcome(jobExecution.getExitStatus()); - - entity.setExecutionMetadata( - metadataJson( - jobExecution.getStartTime(), - jobExecution.getEndTime(), - durationMs, - outcome)); - auditRepository.save(entity); - - log.debug( - "Audit row updated: id={}, outcome={}, durationMs={}", - entityId, - outcome, - durationMs); - }, - () -> - log.warn( - "Audit entity id={} not found in DB for after-job update (executionId={})", + try { + auditRepository + .findById(entityId) + .ifPresentOrElse( + entity -> { + long durationMs = + Duration.between( + jobExecution.getStartTime(), + jobExecution.getEndTime() != null + ? jobExecution.getEndTime() + : jobExecution.getStartTime()) + .toMillis(); + String outcome = jobExecution.getStatus().name(); + + entity.setExecutionMetadata( + metadataJson( + jobExecution.getStartTime(), + jobExecution.getEndTime(), + durationMs, + outcome)); + auditRepository.save(entity); + + log.debug( + "Audit row updated: id={}, outcome={}, durationMs={}", entityId, - jobExecution.getId())); + outcome, + durationMs); + }, + () -> + log.warn( + "Audit entity id={} not found in DB for after-job update (executionId={})", + entityId, + jobExecution.getId())); + } catch (RuntimeException e) { + log.warn( + "Failed to complete execution audit row id={} for job '{}' (executionId={})", + entityId, + jobExecution.getJobInstance().getJobName(), + jobExecution.getId(), + e); + } } // ── helpers ──────────────────────────────────────────────────────────────── - /** - * Maps Spring Batch's {@link ExitStatus} to a simple outcome string. - * - *

{@code COMPLETED} → {@code SUCCESS}; anything else → {@code FAILURE}. - */ - static String mapOutcome(ExitStatus exitStatus) { - return ExitStatus.COMPLETED.getExitCode().equals(exitStatus.getExitCode()) - ? "SUCCESS" - : "FAILURE"; - } - /** * Builds a compact JSON snapshot for {@code execution_metadata}. * diff --git a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/JobsManagerServiceTest.java b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/JobsManagerServiceTest.java index 39f87697..8211b2e2 100644 --- a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/JobsManagerServiceTest.java +++ b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/JobsManagerServiceTest.java @@ -49,7 +49,7 @@ void setUp() { jobsManagerService = new JobsManagerService( jobOperator, jobOperator, jobRegistry, pluginRegistryService, null, - pluginMetrics, null, null); + pluginMetrics); } @Test diff --git a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java index 895c4aac..11c258a8 100644 --- a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java +++ b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/PluginRegistryServiceTest.java @@ -10,20 +10,33 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import com.fronzec.api.BatchJobPlugin; +import com.fronzec.frbatchservice.batchjobs.plugins.listener.JobExecutionAuditListener; import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobDefinitionRepository; import java.util.List; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InOrder; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import org.slf4j.LoggerFactory; import org.springframework.batch.core.configuration.JobRegistry; import org.springframework.batch.core.job.Job; +import org.springframework.batch.core.job.JobExecution; +import org.springframework.batch.core.job.JobInstance; +import org.springframework.batch.core.job.SimpleJob; +import org.springframework.batch.core.job.parameters.JobParameters; +import org.springframework.batch.core.listener.JobExecutionListener; import org.springframework.batch.core.repository.JobRepository; import org.springframework.context.ApplicationContext; import org.springframework.transaction.PlatformTransactionManager; @@ -36,6 +49,7 @@ class PluginRegistryServiceTest { @Mock private PlatformTransactionManager transactionManager; @Mock private ApplicationContext applicationContext; @Mock private JobDefinitionRepository jobDefinitionRepository; + @Mock private JobExecutionAuditListener auditListener; private BatchJobPlugin stubPlugin(String jobName, Job producedJob) { BatchJobPlugin plugin = mock(BatchJobPlugin.class); @@ -58,7 +72,7 @@ void happyPath_singlePlugin_isDiscoveredAndRegistered() throws Exception { PluginRegistryService service = new PluginRegistryService( List.of(plugin), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); service.registerAll(); assertEquals(1, service.getPlugins().size()); @@ -80,7 +94,7 @@ void duplicateJobName_throwsPluginRegistrationException_withBothClassNames() { PluginRegistryService service = new PluginRegistryService( List.of(pluginA, pluginB), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); PluginRegistrationException ex = assertThrows( PluginRegistrationException.class, service::registerAll); @@ -103,7 +117,7 @@ void configureJobThrows_wrapsInPluginRegistrationException_withPluginClassAndCau PluginRegistryService service = new PluginRegistryService( List.of(plugin), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); PluginRegistrationException ex = assertThrows( PluginRegistrationException.class, service::registerAll); @@ -124,7 +138,7 @@ void registerDynamicPlugin_addsToMapAndJobRegistry() throws Exception { PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); service.registerDynamicPlugin(plugin); @@ -140,7 +154,7 @@ void registerDynamicPlugin_doesNotRetainClassloaderOwnership() throws Exception PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); service.registerDynamicPlugin(plugin); @@ -159,7 +173,7 @@ void registerDynamicPlugin_throwsOnDuplicate() throws Exception { PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); service.registerDynamicPlugin(plugin1); @@ -178,7 +192,7 @@ void unregisterDynamicPlugin_removesFromMapAndJobRegistry() throws Exception { PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); service.registerDynamicPlugin(plugin); service.unregisterDynamicPlugin("dynamicJob"); @@ -195,7 +209,7 @@ void unregisterDynamicPlugin_propagatesRegistryFailureAndPreservesPluginMap() th PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); service.registerDynamicPlugin(plugin); RuntimeException failure = new RuntimeException("registry unavailable"); @@ -213,7 +227,7 @@ void unregisterDynamicPlugin_propagatesRegistryFailureAndPreservesPluginMap() th void unregisterDynamicPlugin_unknownName_logsWarning() throws Exception { PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); // Should not throw — just logs a warning service.unregisterDynamicPlugin("nonexistent"); @@ -226,8 +240,110 @@ void unregisterDynamicPlugin_unknownName_logsWarning() throws Exception { void getPluginBySource_returnsNullForUnknownJob() throws Exception { PluginRegistryService service = new PluginRegistryService( List.of(), jobRegistry, jobRepository, transactionManager, - applicationContext, jobDefinitionRepository); + applicationContext, jobDefinitionRepository, auditListener); assertNull(service.getPluginBySource("nonexistent")); } + + // ── Execution audit listener wiring (#111) ───────────────────────────── + + /** Builds a real {@link SimpleJob} that already carries a plugin-provided listener. */ + private SimpleJob realJobWithExistingListener(String name, JobExecutionListener existing) + throws Exception { + SimpleJob job = new SimpleJob(name); + job.setJobRepository(jobRepository); + job.registerJobExecutionListener(existing); + job.afterPropertiesSet(); + return job; + } + + /** Runs the job once and verifies both listeners saw the full lifecycle, existing one first. */ + private void assertBothListenersInvoked(SimpleJob job, JobExecutionListener existing) + throws Exception { + JobExecution execution = + new JobExecution(1L, new JobInstance(1L, job.getName()), new JobParameters()); + job.execute(execution); + + InOrder order = inOrder(existing, auditListener); + order.verify(existing).beforeJob(execution); + order.verify(auditListener).beforeJob(execution); + verify(existing).afterJob(execution); + verify(auditListener).afterJob(execution); + } + + @Test + void registerAll_attachesAuditListener_andKeepsExistingListeners() throws Exception { + JobExecutionListener existing = mock(JobExecutionListener.class); + SimpleJob job = realJobWithExistingListener("job1", existing); + BatchJobPlugin plugin = stubPlugin("job1", job); + + PluginRegistryService service = new PluginRegistryService( + List.of(plugin), jobRegistry, jobRepository, transactionManager, + applicationContext, jobDefinitionRepository, auditListener); + service.registerAll(); + + verify(jobRegistry).register(job); + assertBothListenersInvoked(job, existing); + } + + @Test + void registerDynamicPlugin_attachesAuditListener_andKeepsExistingListeners() throws Exception { + JobExecutionListener existing = mock(JobExecutionListener.class); + SimpleJob job = realJobWithExistingListener("dynamicJob", existing); + BatchJobPlugin plugin = stubPlugin("dynamicJob", job); + + PluginRegistryService service = new PluginRegistryService( + List.of(), jobRegistry, jobRepository, transactionManager, + applicationContext, jobDefinitionRepository, auditListener); + service.registerDynamicPlugin(plugin); + + verify(jobRegistry).register(job); + assertBothListenersInvoked(job, existing); + } + + @Test + void registerDynamicPlugin_reRegisteringTheSameJobInstance_attachesAuditListenerOnce() + throws Exception { + JobExecutionListener existing = mock(JobExecutionListener.class); + SimpleJob job = realJobWithExistingListener("dynamicJob", existing); + BatchJobPlugin plugin = stubPlugin("dynamicJob", job); + + PluginRegistryService service = new PluginRegistryService( + List.of(), jobRegistry, jobRepository, transactionManager, + applicationContext, jobDefinitionRepository, auditListener); + service.registerDynamicPlugin(plugin); + service.unregisterDynamicPlugin("dynamicJob"); + service.registerDynamicPlugin(plugin); + + // verify(...) defaults to times(1): a duplicate attachment would fire beforeJob twice + assertBothListenersInvoked(job, existing); + } + + @Test + void registerDynamicPlugin_nonAbstractJob_registersUnauditedWithWarning() throws Exception { + Job job = stubJob("customJob"); + BatchJobPlugin plugin = stubPlugin("customJob", job); + Logger logger = (Logger) LoggerFactory.getLogger(PluginRegistryService.class); + ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + + try { + PluginRegistryService service = new PluginRegistryService( + List.of(), jobRegistry, jobRepository, transactionManager, + applicationContext, jobDefinitionRepository, auditListener); + service.registerDynamicPlugin(plugin); + + verify(jobRegistry).register(job); + assertTrue(service.getRegisteredJobNames().contains("customJob")); + assertTrue( + appender.list.stream() + .anyMatch(event -> event.getLevel() == Level.WARN + && event.getFormattedMessage().contains("customJob") + && event.getFormattedMessage().contains(job.getClass().getName())), + "a WARN naming the job and its class should be logged"); + } finally { + logger.detachAppender(appender); + } + } } diff --git a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/listener/JobExecutionAuditListenerTest.java b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/listener/JobExecutionAuditListenerTest.java new file mode 100644 index 00000000..cbf35700 --- /dev/null +++ b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/listener/JobExecutionAuditListenerTest.java @@ -0,0 +1,178 @@ +/* 2024-2026 */ +package com.fronzec.frbatchservice.batchjobs.plugins.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.fronzec.frbatchservice.batchjobs.plugins.entity.JobDefinitionEntity; +import com.fronzec.frbatchservice.batchjobs.plugins.entity.JobExecutionAuditEntity; +import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobDefinitionRepository; +import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobExecutionAuditRepository; +import java.time.LocalDateTime; +import java.util.Optional; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.job.JobExecution; +import org.springframework.batch.core.job.JobInstance; +import org.springframework.batch.core.job.parameters.JobParameters; + +@ExtendWith(MockitoExtension.class) +class JobExecutionAuditListenerTest { + + private static final String JOB_NAME = "audited-job"; + private static final String AUDIT_ENTITY_ID_KEY = "_auditEntityId"; + private static final LocalDateTime STARTED_AT = LocalDateTime.of(2026, 10, 16, 10, 0, 0); + private static final LocalDateTime FINISHED_AT = STARTED_AT.plusSeconds(3); + + @Mock private JobExecutionAuditRepository auditRepository; + @Mock private JobDefinitionRepository jobDefinitionRepository; + + private JobExecutionAuditListener listener; + private JobExecution execution; + + @BeforeEach + void setUp() { + listener = new JobExecutionAuditListener(auditRepository, jobDefinitionRepository); + execution = new JobExecution(77L, new JobInstance(5L, JOB_NAME), new JobParameters()); + execution.setStartTime(STARTED_AT); + } + + private JobDefinitionEntity definition() { + JobDefinitionEntity def = new JobDefinitionEntity(); + def.setId(9L); + def.setJobName(JOB_NAME); + def.setVersion("1.2.3"); + return def; + } + + private void stubSaveAssigningId(long id) { + when(auditRepository.save(any(JobExecutionAuditEntity.class))) + .thenAnswer( + invocation -> { + JobExecutionAuditEntity entity = invocation.getArgument(0); + if (entity.getId() == null) { + entity.setId(id); + } + return entity; + }); + } + + @Test + void beforeJob_withDefinition_savesRowAndStoresItsIdInContext() { + when(jobDefinitionRepository.findByJobName(JOB_NAME)).thenReturn(Optional.of(definition())); + stubSaveAssigningId(42L); + + listener.beforeJob(execution); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(JobExecutionAuditEntity.class); + verify(auditRepository).save(captor.capture()); + JobExecutionAuditEntity saved = captor.getValue(); + assertThat(saved.getJobDefinitionId()).isEqualTo(9L); + assertThat(saved.getJobExecutionId()).isEqualTo(77L); + assertThat(saved.getJobVersion()).isEqualTo("1.2.3"); + assertThat(saved.getExecutionMetadata()).isEqualTo("{\"startedAt\":\"" + STARTED_AT + "\"}"); + assertThat(execution.getExecutionContext().get(AUDIT_ENTITY_ID_KEY)).isEqualTo(42L); + } + + @Test + void beforeJob_withoutDefinition_savesNothing() { + when(jobDefinitionRepository.findByJobName(JOB_NAME)).thenReturn(Optional.empty()); + + listener.beforeJob(execution); + + verify(auditRepository, never()).save(any()); + assertThat(execution.getExecutionContext().containsKey(AUDIT_ENTITY_ID_KEY)).isFalse(); + } + + @Test + void beforeJob_clearsStaleAuditIdSoAfterJobCannotUpdateAnOldRow() { + // A restarted execution inherits the previous execution's context, including the old row id. + execution.getExecutionContext().put(AUDIT_ENTITY_ID_KEY, 99L); + when(jobDefinitionRepository.findByJobName(JOB_NAME)).thenReturn(Optional.empty()); + + listener.beforeJob(execution); + listener.afterJob(execution); + + assertThat(execution.getExecutionContext().containsKey(AUDIT_ENTITY_ID_KEY)).isFalse(); + verify(auditRepository, never()).findById(any()); + verify(auditRepository, never()).save(any()); + } + + @Test + void beforeJob_replacesStaleAuditIdWithTheNewRowId() { + execution.getExecutionContext().put(AUDIT_ENTITY_ID_KEY, 99L); + when(jobDefinitionRepository.findByJobName(JOB_NAME)).thenReturn(Optional.of(definition())); + stubSaveAssigningId(43L); + + listener.beforeJob(execution); + + assertThat(execution.getExecutionContext().get(AUDIT_ENTITY_ID_KEY)).isEqualTo(43L); + } + + @Test + void beforeJob_repositoryFailure_doesNotFailTheJob() { + when(jobDefinitionRepository.findByJobName(JOB_NAME)).thenReturn(Optional.of(definition())); + when(auditRepository.save(any(JobExecutionAuditEntity.class))) + .thenThrow(new IllegalStateException("database unavailable")); + + assertThatCode(() -> listener.beforeJob(execution)).doesNotThrowAnyException(); + assertThat(execution.getExecutionContext().containsKey(AUDIT_ENTITY_ID_KEY)).isFalse(); + } + + @Test + void beforeJob_definitionLookupFailure_doesNotFailTheJob() { + when(jobDefinitionRepository.findByJobName(JOB_NAME)) + .thenThrow(new IllegalStateException("database unavailable")); + + assertThatCode(() -> listener.beforeJob(execution)).doesNotThrowAnyException(); + } + + @Test + void afterJob_repositoryFailure_doesNotFailTheJob() { + execution.getExecutionContext().put(AUDIT_ENTITY_ID_KEY, 42L); + execution.setEndTime(FINISHED_AT); + execution.setStatus(BatchStatus.COMPLETED); + when(auditRepository.findById(42L)).thenThrow(new IllegalStateException("database unavailable")); + + assertThatCode(() -> listener.afterJob(execution)).doesNotThrowAnyException(); + } + + @ParameterizedTest + @EnumSource( + value = BatchStatus.class, + names = {"COMPLETED", "FAILED", "STOPPED"}) + void afterJob_writesFinishTimeDurationAndRawBatchStatusOutcome(BatchStatus status) { + JobExecutionAuditEntity row = new JobExecutionAuditEntity(); + row.setId(42L); + row.setExecutionMetadata("{\"startedAt\":\"" + STARTED_AT + "\"}"); + execution.getExecutionContext().put(AUDIT_ENTITY_ID_KEY, 42L); + execution.setEndTime(FINISHED_AT); + execution.setStatus(status); + when(auditRepository.findById(42L)).thenReturn(Optional.of(row)); + + listener.afterJob(execution); + + verify(auditRepository).save(row); + assertThat(row.getExecutionMetadata()) + .isEqualTo( + "{\"startedAt\":\"" + + STARTED_AT + + "\",\"finishedAt\":\"" + + FINISHED_AT + + "\",\"durationMs\":3000,\"outcome\":\"" + + status.name() + + "\"}"); + } +} diff --git a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoadingIntegrationTest.java b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoadingIntegrationTest.java index 668ef6cd..3768c231 100644 --- a/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoadingIntegrationTest.java +++ b/fr-batch-service/src/test/java/com/fronzec/frbatchservice/batchjobs/plugins/loader/DynamicJobLoadingIntegrationTest.java @@ -13,6 +13,7 @@ import com.fronzec.frbatchservice.batchjobs.JobsManagerService; import com.fronzec.frbatchservice.batchjobs.plugins.PluginRegistryService; import com.fronzec.frbatchservice.batchjobs.plugins.entity.JobDefinitionEntity; +import com.fronzec.frbatchservice.batchjobs.plugins.entity.JobExecutionAuditEntity; import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobDefinitionRepository; import com.fronzec.frbatchservice.batchjobs.plugins.repository.JobExecutionAuditRepository; import com.fronzec.frbatchservice.restclients.ApiClient; @@ -22,13 +23,20 @@ import java.math.BigDecimal; import java.net.URL; import java.nio.file.*; +import java.time.Duration; import java.time.LocalDate; import java.util.Comparator; +import java.util.List; import java.util.Map; import java.util.jar.JarEntry; import java.util.jar.JarOutputStream; import org.junit.jupiter.api.*; +import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.job.JobExecution; +import org.springframework.batch.core.job.parameters.JobParameters; +import org.springframework.batch.core.job.parameters.JobParametersBuilder; +import org.springframework.batch.core.repository.JobRepository; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.http.MediaType; @@ -84,6 +92,7 @@ class DynamicJobLoadingIntegrationTest { @Autowired private PluginRegistryService pluginRegistryService; @Autowired private DynamicJobLoaderService loaderService; @Autowired private JobExecutionAuditRepository jobExecutionAuditRepository; + @Autowired private JobRepository jobRepository; @MockitoBean private ApiClient apiClient; @@ -191,6 +200,43 @@ private MockMultipartFile testJar() throws IOException { Files.readAllBytes(testJarPath)); } + /* ── Helpers: execution audit lookups ──────────────────────────────────── */ + + private static JobParameters identifyingParams(String attemptNumber) { + return new JobParametersBuilder() + .addString("DATE", LocalDate.now().toString()) + .addString("ATTEMPT_NUMBER", attemptNumber) + .toJobParameters(); + } + + private List auditRowsFor(long jobExecutionId) { + return jobExecutionAuditRepository.findByJobDefinitionId(definitionId).stream() + .filter(row -> row.getJobExecutionId() == jobExecutionId) + .toList(); + } + + private JobExecutionAuditEntity singleAuditRow(long jobExecutionId) { + List rows = auditRowsFor(jobExecutionId); + assertThat(rows).as("audit rows for job execution %d", jobExecutionId).hasSize(1); + return rows.get(0); + } + + /** Polls (bounded) until the async execution's single audit row carries its finish time. */ + private JobExecutionAuditEntity awaitFinishedAuditRow(long jobExecutionId) + throws InterruptedException { + long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos(); + while (System.nanoTime() < deadline) { + List rows = auditRowsFor(jobExecutionId); + if (rows.size() == 1 + && rows.get(0).getExecutionMetadata() != null + && rows.get(0).getExecutionMetadata().contains("finishedAt")) { + return rows.get(0); + } + Thread.sleep(100); + } + return singleAuditRow(jobExecutionId); // fails with a descriptive message, or returns the row + } + /* ── 1. Upload valid plugin ────────────────────────────────────────────── */ @Test @@ -372,12 +418,41 @@ void dynamicPlugin_isExecutable_viaJobsManagerService() { assertEquals(ExitStatus.COMPLETED.toString(), result.get("result"), "Dynamic plugin job should complete successfully"); + + // The audit listener attached at registration is the single writer: exactly one finished row. + JobExecution execution = jobRepository.getLastJobExecution(JOB_NAME, identifyingParams("1")); + assertThat(execution).isNotNull(); + JobExecutionAuditEntity row = singleAuditRow(execution.getId()); + assertThat(row.getJobDefinitionId()).isEqualTo(definitionId); + assertThat(row.getExecutionMetadata()).contains("finishedAt", "durationMs", "COMPLETED"); } - /* ── 12. Load-all → loads unloaded definitions ─────────────────────────── */ + /* ── 12. Async launch of the dynamic plugin is audited exactly once ────── */ @Test @Order(12) + void dynamicPlugin_asyncLaunch_writesExactlyOneFinishedAuditRow() throws InterruptedException { + SingleJobDataRequest request = new SingleJobDataRequest(); + request.setJobBeanName(JOB_NAME); + request.setParam("DATE", LocalDate.now().toString()); + request.setParam("ATTEMPT_NUMBER", "2"); + + Map result = jobsManagerService.asyncRunJobWithParams(request); + assertThat(result).doesNotContainKey("error"); + + JobExecution execution = jobRepository.getLastJobExecution(JOB_NAME, identifyingParams("2")); + assertThat(execution).as("async launch must create a job execution").isNotNull(); + JobExecutionAuditEntity row = awaitFinishedAuditRow(execution.getId()); + assertThat(row.getJobDefinitionId()).isEqualTo(definitionId); + assertThat(row.getExecutionMetadata()).contains("durationMs", "COMPLETED"); + assertThat(jobRepository.getJobExecution(execution.getId()).getStatus()) + .isEqualTo(BatchStatus.COMPLETED); + } + + /* ── 13. Load-all → loads unloaded definitions ─────────────────────────── */ + + @Test + @Order(13) void loadAll_loadsUnloadedDefinitions() throws Exception { // First, reload to ensure the definition is LOADED, then unload it so // load-all has something to pick up. @@ -399,10 +474,10 @@ void loadAll_loadsUnloadedDefinitions() throws Exception { assertThat(body).contains("\"message\":\"Successfully loaded\""); } - /* ── 13. Auto-load re-instantiates definitions with stale LOADED status ── */ + /* ── 14. Auto-load re-instantiates definitions with stale LOADED status ── */ @Test - @Order(13) + @Order(14) void loadAllEnabled_reloadsDefinitionWithStaleLoadedStatus() { // Simulate a backend restart: the database still says LOADED from a previous // JVM, but the in-memory registry is empty. Auto-load must actually load the @@ -437,12 +512,12 @@ void loadAllEnabled_reloadsDefinitionWithStaleLoadedStatus() { assertThat(pluginRegistryService.getRegisteredJobNames()).contains(JOB_NAME); } - /* ── 14. DELETE with execution history → 409 before unload, nothing changes ── */ + /* ── 15. DELETE with execution history → 409 before unload, nothing changes ── */ @Test - @Order(14) + @Order(15) void deleteWithExecutionHistory_returnsConflict_andKeepsPluginRowAndJar() throws Exception { - // Test 11 executed the job, so job_executions_audit rows reference this definition. + // Tests 11 and 12 executed the job, so job_executions_audit rows reference this definition. // The history guard runs before unload, so even force=true must leave everything intact. assertThat(jobExecutionAuditRepository.existsByJobDefinitionId(definitionId)).isTrue(); assertThat(pluginRegistryService.getRegisteredJobNames()).contains(JOB_NAME); @@ -462,12 +537,12 @@ void deleteWithExecutionHistory_returnsConflict_andKeepsPluginRowAndJar() throws assertThat(pluginRegistryService.getRegisteredJobNames()).contains(JOB_NAME); } - /* ── 15. DELETE a loaded plugin → 204, unregistered, row and JAR removed ── */ + /* ── 16. DELETE a loaded plugin → 204, unregistered, row and JAR removed ── */ @Test - @Order(15) + @Order(16) void deleteLoadedPlugin_returnsNoContent_andUnregistersJob() throws Exception { - // Remove the execution history that blocked test 14 so this test isolates the + // Remove the execution history that blocked test 15 so this test isolates the // lifecycle behaviour of deleting a loaded plugin. jobExecutionAuditRepository.deleteAll( jobExecutionAuditRepository.findByJobDefinitionId(definitionId)); @@ -488,10 +563,10 @@ void deleteLoadedPlugin_returnsNoContent_andUnregistersJob() throws Exception { assertThat(storedJar).doesNotExist(); } - /* ── 16. DELETE an unknown definition → 404 error envelope ────────────── */ + /* ── 17. DELETE an unknown definition → 404 error envelope ────────────── */ @Test - @Order(16) + @Order(17) void deleteNonExistentDefinition_returnsNotFound() throws Exception { mockMvc .perform(delete(JOBS_BASE + "/definitions/99999"))