Improve discovery process to update information about status and progress

src/main/java/com/czertainly/core/config/AsyncConfig.java

@@ -1,5 +1,8 @@
 package com.czertainly.core.config;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.scheduling.annotation.AsyncConfigurer;
 import org.springframework.scheduling.annotation.EnableAsync;
@@ -12,9 +15,16 @@
 @EnableAsync
 public class AsyncConfig implements AsyncConfigurer {
 
+    private static final Logger logger = LoggerFactory.getLogger(AsyncConfig.class);
+
     @Override
     public Executor getAsyncExecutor() {
         Executor virtualThreadExecutor = Executors.newVirtualThreadPerTaskExecutor();
         return new DelegatingSecurityContextExecutor(virtualThreadExecutor);
     }
+
+    @Override
+    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
+        return (ex, method, params) -> logger.error("Uncaught exception in async method {}.{}: {}", method.getDeclaringClass().getName(), method.getName(), ex.getMessage(), ex);
+    }
 }

src/main/java/com/czertainly/core/messaging/listeners/EventListener.java

@@ -1,27 +1,18 @@
 package com.czertainly.core.messaging.listeners;
 
-import com.czertainly.api.exception.AttributeException;
-import com.czertainly.api.exception.NotFoundException;
-import com.czertainly.api.exception.RuleException;
+import com.czertainly.api.model.core.auth.Resource;
 import com.czertainly.api.model.core.certificate.CertificateEvent;
 import com.czertainly.api.model.core.certificate.CertificateEventStatus;
-import com.czertainly.api.model.core.other.ResourceEvent;
 import com.czertainly.core.messaging.configuration.RabbitMQConstants;
 import com.czertainly.core.messaging.model.EventMessage;
 import com.czertainly.core.service.CertificateEventHistoryService;
-import com.czertainly.core.service.DiscoveryService;
-import com.czertainly.core.tasks.ScheduledJobInfo;
-import com.czertainly.core.util.AuthHelper;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.security.NoSuchAlgorithmException;
-import java.security.cert.CertificateException;
 import java.util.Objects;
 
 @Component
@@ -31,37 +22,18 @@ public class EventListener {
     private static final Logger logger = LoggerFactory.getLogger(EventListener.class);
 
     private CertificateEventHistoryService certificateEventHistoryService;
-    private DiscoveryService discoveryService;
-    private AuthHelper authHelper;
-    private final ObjectMapper mapper = new ObjectMapper();
-
-    @Autowired
-    public void setAuthHelper(AuthHelper authHelper) {
-        this.authHelper = authHelper;
-    }
-
-    @Autowired
-    public void setDiscoveryService(DiscoveryService discoveryService) {
-        this.discoveryService = discoveryService;
-    }
 
     @Autowired
     public void setCertificateEventHistoryService(CertificateEventHistoryService certificateEventHistoryService) {
         this.certificateEventHistoryService = certificateEventHistoryService;
     }
 
     @RabbitListener(queues = RabbitMQConstants.QUEUE_EVENTS_NAME, messageConverter = "jsonMessageConverter", concurrency = "3")
-    public void processMessage(EventMessage eventMessage) throws NotFoundException, CertificateException, NoSuchAlgorithmException, RuleException, AttributeException {
-        switch (eventMessage.getResource()) {
-            case CERTIFICATE ->
-                    certificateEventHistoryService.addEventHistory(eventMessage.getResourceUUID(), CertificateEvent.findByCode(eventMessage.getEventName()), CertificateEventStatus.valueOf(eventMessage.getEventStatus()), eventMessage.getEventMessage(), eventMessage.getEventDetail());
-            case DISCOVERY -> {
-                authHelper.authenticateAsUser(eventMessage.getUserUuid());
-                if (Objects.equals(eventMessage.getEventName(), ResourceEvent.DISCOVERY_FINISHED.getCode())) {
-                    discoveryService.evaluateDiscoveryTriggers(eventMessage.getResourceUUID(), eventMessage.getUserUuid(), mapper.convertValue(eventMessage.getEventData(), ScheduledJobInfo.class));
-                }
-            }
-            default -> logger.warn("Event handling is supported only for certificates for now");
+    public void processMessage(EventMessage eventMessage) {
+        if (Objects.requireNonNull(eventMessage.getResource()) == Resource.CERTIFICATE) {
+            certificateEventHistoryService.addEventHistory(eventMessage.getResourceUUID(), CertificateEvent.findByCode(eventMessage.getEventName()), CertificateEventStatus.valueOf(eventMessage.getEventStatus()), eventMessage.getEventMessage(), eventMessage.getEventDetail());
+        } else {
+            logger.warn("Event handling is supported only for certificates for now");
         }
     }
 

src/main/java/com/czertainly/core/messaging/listeners/ValidationListener.java

@@ -6,12 +6,12 @@
 import com.czertainly.core.messaging.configuration.RabbitMQConstants;
 import com.czertainly.core.messaging.model.ValidationMessage;
 import com.czertainly.core.service.handler.CertificateHandler;
-import jakarta.transaction.Transactional;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 

src/main/java/com/czertainly/core/messaging/producers/EventProducer.java

@@ -4,10 +4,8 @@
 import com.czertainly.api.model.core.auth.Resource;
 import com.czertainly.api.model.core.certificate.CertificateEvent;
 import com.czertainly.api.model.core.certificate.CertificateEventStatus;
-import com.czertainly.api.model.core.other.ResourceEvent;
 import com.czertainly.core.messaging.configuration.RabbitMQConstants;
 import com.czertainly.core.messaging.model.EventMessage;
-import com.czertainly.core.tasks.ScheduledJobInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -44,10 +42,4 @@ public void produceCertificateStatusChangeEventMessage(final UUID certificateUUI
         produceMessage(eventMessage);
     }
 
-    public void produceDiscoveryFinishedEventMessage(final UUID discoveryUuid, final UUID userUuid, final ResourceEvent event, final ScheduledJobInfo scheduledJobInfo) {
-
-        final EventMessage eventMessage = new EventMessage(Resource.DISCOVERY, discoveryUuid, event.getCode(), null, null, null, userUuid, scheduledJobInfo);
-        produceMessage(eventMessage);
-    }
-
 }

src/main/java/com/czertainly/core/service/DiscoveryService.java

@@ -11,8 +11,6 @@
 import com.czertainly.core.security.authz.SecurityFilter;
 import com.czertainly.core.tasks.ScheduledJobInfo;
 
-import java.security.NoSuchAlgorithmException;
-import java.security.cert.CertificateException;
 import java.util.List;
 import java.util.UUID;
 
@@ -51,6 +49,4 @@ public interface DiscoveryService extends ResourceExtensionService {
 
     List<SearchFieldDataByGroupDto> getSearchableFieldInformationByGroup();
 
-    void evaluateDiscoveryTriggers(UUID discoveryUuid, UUID userUuid, ScheduledJobInfo scheduledJobInfo) throws NotFoundException, RuleException, CertificateException, NoSuchAlgorithmException, AttributeException;
-
-}
+}

src/main/java/com/czertainly/core/service/handler/CertificateHandler.java

@@ -21,10 +21,10 @@
 import com.czertainly.core.dao.entity.workflows.TriggerHistory;
 import com.czertainly.core.dao.repository.CertificateRepository;
 import com.czertainly.core.dao.repository.DiscoveryCertificateRepository;
+import com.czertainly.core.dao.repository.DiscoveryRepository;
 import com.czertainly.core.dao.repository.workflows.TriggerAssociationRepository;
 import com.czertainly.core.evaluator.CertificateRuleEvaluator;
 import com.czertainly.core.event.transaction.CertificateValidationEvent;
-import com.czertainly.core.event.transaction.DiscoveryProgressEvent;
 import com.czertainly.core.messaging.model.ValidationMessage;
 import com.czertainly.core.messaging.producers.ValidationProducer;
 import com.czertainly.core.service.CertificateEventHistoryService;
@@ -37,7 +37,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Isolation;
 import org.springframework.transaction.annotation.Propagation;
@@ -57,7 +56,6 @@ public class CertificateHandler {
 
     private AttributeEngine attributeEngine;
     private CertificateRuleEvaluator certificateRuleEvaluator;
-    private ApplicationEventPublisher applicationEventPublisher;
     private ValidationProducer validationProducer;
 
     private TriggerService triggerService;
@@ -66,6 +64,7 @@ public class CertificateHandler {
     private CertificateEventHistoryService certificateEventHistoryService;
 
     private CertificateRepository certificateRepository;
+    private DiscoveryRepository discoveryRepository;
     private DiscoveryCertificateRepository discoveryCertificateRepository;
     private TriggerAssociationRepository triggerAssociationRepository;
 
@@ -79,11 +78,6 @@ public void setCertificateRuleEvaluator(CertificateRuleEvaluator certificateRule
         this.certificateRuleEvaluator = certificateRuleEvaluator;
     }
 
-    @Autowired
-    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
-        this.applicationEventPublisher = applicationEventPublisher;
-    }
-
     @Autowired
     public void setValidationProducer(ValidationProducer validationProducer) {
         this.validationProducer = validationProducer;
@@ -114,6 +108,11 @@ public void setCertificateRepository(CertificateRepository certificateRepository
         this.certificateRepository = certificateRepository;
     }
 
+    @Autowired
+    public void setDiscoveryRepository(DiscoveryRepository discoveryRepository) {
+        this.discoveryRepository = discoveryRepository;
+    }
+
     @Autowired
     public void setDiscoveryCertificateRepository(DiscoveryCertificateRepository discoveryCertificateRepository) {
         this.discoveryCertificateRepository = discoveryCertificateRepository;
@@ -128,7 +127,7 @@ public void setTriggerAssociationRepository(TriggerAssociationRepository trigger
     public void validate(Certificate certificate) {
         certificateService.validate(certificate);
         try {
-            if(certificate.getRaProfileUuid() != null) {
+            if (certificate.getRaProfileUuid() != null) {
                 complianceService.checkComplianceOfCertificate(certificate);
             }
         } catch (ConnectorException e) {
@@ -176,17 +175,20 @@ public void createDiscoveredCertificate(String batch, DiscoveryHistory discovery
             }
         }
 
-        applicationEventPublisher.publishEvent(new DiscoveryProgressEvent(discovery.getUuid(), discovery.getTotalCertificatesDiscovered(), true));
+        // report progress
+        long currentCount = discoveryCertificateRepository.countByDiscovery(discovery);
+        discovery.setMessage(String.format("Downloaded %d %% of discovered certificates from provider (%d / %d)", (int) ((currentCount / (double) discovery.getConnectorTotalCertificatesDiscovered()) * 100), currentCount, discovery.getConnectorTotalCertificatesDiscovered()));
+        discoveryRepository.save(discovery);
     }
 
     @Transactional(propagation = Propagation.REQUIRES_NEW, isolation = Isolation.DEFAULT)
     public void processDiscoveredCertificate(int certIndex, int totalCount, DiscoveryHistory discovery, DiscoveryCertificate discoveryCertificate) {
         // Get X509 from discovered certificate and create certificate entity, do not save in database yet
-        Certificate entry;
+        Certificate certificate;
         X509Certificate x509Cert;
         try {
             x509Cert = CertificateUtil.parseCertificate(discoveryCertificate.getCertificateContent().getContent());
-            entry = certificateService.createCertificateEntity(x509Cert);
+            certificate = certificateService.createCertificateEntity(x509Cert);
         } catch (Exception e) {
             logger.error("Unable to create certificate from discovery certificate with UUID {}: {}", discoveryCertificate.getUuid(), e.getMessage());
             discoveryCertificate.setProcessed(true);
@@ -213,60 +215,68 @@ public void processDiscoveredCertificate(int certIndex, int totalCount, Discover
         }
 
         try {
-            // First, check the triggers that have action with action type set to ignore
-            boolean ignored = false;
-            List<TriggerHistory> ignoreTriggerHistories = new ArrayList<>();
-            for (Trigger trigger : ignoreTriggers) {
-                TriggerHistory triggerHistory = triggerService.createTriggerHistory(OffsetDateTime.now(), trigger.getUuid(), discovery.getUuid(), null, discoveryCertificate.getUuid());
-                if (certificateRuleEvaluator.evaluateRules(trigger.getRules(), entry, triggerHistory)) {
-                    ignored = true;
-                    triggerHistory.setConditionsMatched(true);
-                    triggerHistory.setActionsPerformed(true);
-                    break;
-                } else {
-                    triggerHistory.setConditionsMatched(false);
-                    triggerHistory.setActionsPerformed(false);
-                }
-                ignoreTriggerHistories.add(triggerHistory);
-            }
+            processTriggers(discovery.getUuid(), certificate, discoveryCertificate, ignoreTriggers, orderedTriggers);
+        } catch (RuleException e) {
+            logger.error("Unable to process trigger on certificate {} from discovery certificate with UUID {}. Message: {}", certificate.getUuid(), discoveryCertificate.getUuid(), e.getMessage());
+        }
 
-            // If some trigger ignored this certificate, certificate is not saved and continue with next one
-            if (ignored) {
-                return;
-            }
+        updateDiscoveredCertificate(discovery, certificate, discoveryCertificate.getMeta());
+        discoveryCertificate.setProcessed(true);
+
+        discoveryCertificateRepository.save(discoveryCertificate);
 
-            // Save certificate to database
-            certificateService.updateCertificateEntity(entry);
+        // report progress
+        if (certIndex % 2 == 0) {
+            long currentCount = discoveryCertificateRepository.countByDiscoveryAndNewlyDiscoveredAndProcessed(discovery, true, true);
+            discovery.setMessage(String.format("Processed %d %% of newly discovered certificates (%d / %d)", (int) ((currentCount / (double) totalCount) * 100), currentCount, totalCount));
+            discoveryRepository.save(discovery);
+        }
+    }
 
-            // update objectUuid of not ignored certs
-            for (TriggerHistory ignoreTriggerHistory : ignoreTriggerHistories) {
-                ignoreTriggerHistory.setObjectUuid(entry.getUuid());
+    private void processTriggers(UUID discoveryUuid, Certificate certificate, DiscoveryCertificate discoveryCertificate, List<Trigger> ignoreTriggers, List<Trigger> orderedTriggers) throws RuleException {
+        // First, check the triggers that have action with action type set to ignore
+        boolean ignored = false;
+        List<TriggerHistory> ignoreTriggerHistories = new ArrayList<>();
+        for (Trigger trigger : ignoreTriggers) {
+            TriggerHistory triggerHistory = triggerService.createTriggerHistory(OffsetDateTime.now(), trigger.getUuid(), discoveryUuid, null, discoveryCertificate.getUuid());
+            if (certificateRuleEvaluator.evaluateRules(trigger.getRules(), certificate, triggerHistory)) {
+                ignored = true;
+                triggerHistory.setConditionsMatched(true);
+                triggerHistory.setActionsPerformed(true);
+                break;
+            } else {
+                triggerHistory.setConditionsMatched(false);
+                triggerHistory.setActionsPerformed(false);
             }
+            ignoreTriggerHistories.add(triggerHistory);
+        }
 
-            // Evaluate rest of the triggers in given order
-            for (Trigger trigger : orderedTriggers) {
-                // Create trigger history entry
-                TriggerHistory triggerHistory = triggerService.createTriggerHistory(OffsetDateTime.now(), trigger.getUuid(), discovery.getUuid(), entry.getUuid(), discoveryCertificate.getUuid());
-                // If rules are satisfied, perform defined actions
-                if (certificateRuleEvaluator.evaluateRules(trigger.getRules(), entry, triggerHistory)) {
-                    triggerHistory.setConditionsMatched(true);
-                    certificateRuleEvaluator.performActions(trigger, entry, triggerHistory);
-                    triggerHistory.setActionsPerformed(triggerHistory.getRecords().isEmpty());
-                } else {
-                    triggerHistory.setConditionsMatched(false);
-                    triggerHistory.setActionsPerformed(false);
-                }
-            }
-        } catch (RuleException e) {
-            logger.error("Unable to process trigger on certificate {} from discovery certificate with UUID {}. Message: {}", entry.getUuid(), discoveryCertificate.getUuid(), e.getMessage());
+        // If some trigger ignored this certificate, certificate is not saved and continue with next one
+        if (ignored) {
+            return;
         }
 
-        updateDiscoveredCertificate(discovery, entry, discoveryCertificate.getMeta());
-        discoveryCertificate.setProcessed(true);
-        discoveryCertificateRepository.save(discoveryCertificate);
+        // Save certificate to database
+        certificateService.updateCertificateEntity(certificate);
 
-        if (certIndex % 100 == 0) {
-            applicationEventPublisher.publishEvent(new DiscoveryProgressEvent(discovery.getUuid(), totalCount, false));
+        // update objectUuid of not ignored certs
+        for (TriggerHistory ignoreTriggerHistory : ignoreTriggerHistories) {
+            ignoreTriggerHistory.setObjectUuid(certificate.getUuid());
+        }
+
+        // Evaluate rest of the triggers in given order
+        for (Trigger trigger : orderedTriggers) {
+            // Create trigger history entry
+            TriggerHistory triggerHistory = triggerService.createTriggerHistory(OffsetDateTime.now(), trigger.getUuid(), discoveryUuid, certificate.getUuid(), discoveryCertificate.getUuid());
+            // If rules are satisfied, perform defined actions
+            if (certificateRuleEvaluator.evaluateRules(trigger.getRules(), certificate, triggerHistory)) {
+                triggerHistory.setConditionsMatched(true);
+                certificateRuleEvaluator.performActions(trigger, certificate, triggerHistory);
+                triggerHistory.setActionsPerformed(triggerHistory.getRecords().isEmpty());
+            } else {
+                triggerHistory.setConditionsMatched(false);
+                triggerHistory.setActionsPerformed(false);
+            }
         }
     }