logging statements

aggregator/src/main/java/com/redhat/cloud/notifications/DailyEmailAggregationJob.java

@@ -80,10 +80,14 @@ public void processDailyEmail() {
             aggregationOrgConfigRepository.createMissingDefaultConfiguration(defaultDailyDigestTime);
             List<AggregationCommand> aggregationCommands = processAggregateEmailsWithOrgPref(now, registry);
             Log.infof("found %s commands", aggregationCommands.size());
+            Log.debugf("Aggregation commands: %s", aggregationCommands);
+
             aggregationCommands.stream().collect(Collectors.groupingBy(AggregationCommand::getOrgId))
                 .values().forEach(this::sendIt);
 
             List<String> orgIdsToUpdate = aggregationCommands.stream().map(agc -> agc.getAggregationKey().getOrgId()).collect(Collectors.toList());
+            Log.debugf("Found following org IDs to update: %s", orgIdsToUpdate);
+
             aggregationOrgConfigRepository.updateLastCronJobRunAccordingOrgPref(orgIdsToUpdate, now);
 
             Gauge lastSuccess = Gauge

engine/src/main/java/com/redhat/cloud/notifications/events/EndpointProcessor.java

@@ -17,6 +17,7 @@
 import com.redhat.cloud.notifications.processors.webhooks.WebhookTypeProcessor;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.quarkus.logging.Log;
 import jakarta.annotation.PostConstruct;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -89,7 +90,11 @@ public void process(Event event) {
 
             endpoints = List.of(endpoint);
         } else if (isAggregatorEvent(event)) {
+            Log.debugf("[org_id: %s] Processing aggregation event: %s", event.getOrgId(), event);
+
             endpoints = List.of(endpointRepository.getOrCreateDefaultSystemSubscription(event.getAccountId(), event.getOrgId(), EndpointType.EMAIL_SUBSCRIPTION));
+
+            Log.debugf("[org_id: %s] Found %s endpoints for the aggregation event: %s", event.getOrgId(), endpoints.size(), event);
         } else {
             endpoints = endpointRepository.getTargetEndpoints(event.getOrgId(), event.getEventType());
         }
@@ -124,8 +129,10 @@ public void process(Event event) {
                             break;
                         case EMAIL_SUBSCRIPTION:
                             if (isAggregatorEvent(event)) {
+                                Log.debugf("[org_id: %s] Sending event through the aggregator processor: %s", event.getOrgId(), event);
                                 emailAggregationProcessor.processAggregation(event);
                             } else {
+                                Log.debugf("[org_id: %s] Sending event through the email connector: %s", event.getOrgId(), event);
                                 emailConnectorProcessor.process(event, endpointsByTypeEntry.getValue());
                             }
                             break;

engine/src/main/java/com/redhat/cloud/notifications/events/EventConsumer.java

@@ -141,6 +141,7 @@ public CompletionStage<Void> process(Message<String> message) {
                  * The message ID is already known which means we already processed the current
                  * message and sent notifications. The message is therefore ignored.
                  */
+                Log.debugf(" [kafka_message_id: %s] Duplicated Kafka message ignored", messageId);
                 duplicateCounter.increment();
             } else {
                 /*

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailAggregationProcessor.java

@@ -264,6 +264,8 @@ private void processBundleAggregation(List<AggregationCommand> aggregationComman
         Map<User, List<ApplicationAggregatedData>> userData = new HashMap<>();
 
         for (AggregationCommand applicationAggregationCommand : aggregationCommands) {
+            Log.debugf("Processing aggregation command: %s", applicationAggregationCommand);
+
             try {
                 Application app = applicationRepository.getApplication(applicationAggregationCommand.getAggregationKey().getBundle(), applicationAggregationCommand.getAggregationKey().getApplication())
                         .orElseThrow(() -> {
@@ -289,6 +291,8 @@ private void processBundleAggregation(List<AggregationCommand> aggregationComman
         Map<List<ApplicationAggregatedData>, Set<User>> usersWithSameAggregatedData = userData.keySet().stream()
             .collect(Collectors.groupingBy(userData::get, Collectors.toSet()));
 
+        Log.debugf("Users with same aggregated data: %s", usersWithSameAggregatedData);
+
         String emailTitle = "Daily digest - " + bundle.getDisplayName();
 
         // for each set of users, generate email subject + body and send it to email connector
@@ -348,6 +352,8 @@ private void processBundleAggregation(List<AggregationCommand> aggregationComman
                 );
 
                 connectorSender.send(aggregatorEvent, endpoint, JsonObject.mapFrom(emailNotification));
+
+                Log.debugf("Sent email notification to connector: %s", emailNotification);
             }
         });
 

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailProcessor.java

@@ -121,6 +121,8 @@ public void process(final Event event, final List<Endpoint> endpoints) {
             externalAuthorizationCriteriaExtractor.extract(event)
         );
 
+        Log.debugf("[org_id: %s] Sending email notification to connector", emailNotification);
+
         final JsonObject payload = JsonObject.mapFrom(emailNotification);
 
         final Endpoint endpoint = endpointRepository.getOrCreateDefaultSystemSubscription(event.getAccountId(), event.getOrgId(), EMAIL_SUBSCRIPTION);