<commit_before>package uk.gov.ons.ctp.response.action.export.scheduled;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import javax.inject.Named;

import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.scheduling.annotation.Scheduled;

import lombok.extern.slf4j.Slf4j;
import uk.gov.ons.ctp.common.distributed.DistributedInstanceManager;
import uk.gov.ons.ctp.common.distributed.DistributedLatchManager;
import uk.gov.ons.ctp.common.distributed.DistributedLockManager;
import uk.gov.ons.ctp.common.error.CTPException;
import uk.gov.ons.ctp.response.action.export.domain.ActionRequestInstruction;
import uk.gov.ons.ctp.response.action.export.domain.ExportMessage;
import uk.gov.ons.ctp.response.action.export.message.SftpServicePublisher;
import uk.gov.ons.ctp.response.action.export.service.ActionRequestService;
import uk.gov.ons.ctp.response.action.export.service.TemplateMappingService;
import uk.gov.ons.ctp.response.action.export.service.TransformationService;

/**
 * This class will be responsible for the scheduling of export actions
 *
 */
@Named
@Slf4j
public class ExportScheduler implements HealthIndicator {

  private static final String DATE_FORMAT_IN_FILE_NAMES = "ddMMyyyy_HHmm";
  private static final String DISTRIBUTED_OBJECT_KEY_LATCH = "latch";
  private static final String DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT = "scheduler";

  @Inject
  private TransformationService transformationService;

  @Inject
  private TemplateMappingService templateMappingService;

  @Inject
  private SftpServicePublisher sftpService;

  @Inject
  private ActionRequestService actionRequestService;

  @Inject
  private DistributedLockManager actionExportLockManager;

  @Inject
  private DistributedInstanceManager actionExportInstanceManager;

  @Inject
  private DistributedLatchManager actionExportLatchManager;

  @Inject
  private ExportInfo exportInfo;

  @Override
  public Health health() {
    return Health.up()
        .withDetail("exportInfo", exportInfo)
        .build();
  }

  /**
   * Initialise export scheduler
   *
   */
  @PostConstruct
  public void init() {
    actionExportInstanceManager.incrementInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
    log.info("{} {} instance/s running",
        actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT),
        DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
  }

  /**
   * Clean up scheduler on bean destruction
   *
   */
  @PreDestroy
  public void cleanUp() {
    actionExportInstanceManager.decrementInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
    // Make sure no locks if interrupted in middle of run
    actionExportLockManager.unlockInstanceLocks();
    log.info("{} {} instance/s running",
        actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT),
        DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
  }

  /**
   * Carry out scheduled actions according to configured cron expression
   *
   */
  @Scheduled(cron = "#{appConfig.exportSchedule.cronExpression}")
  public void scheduleExport() {

    log.info("Scheduled run start");

    // Warn if Mapping document cannot deal with all ActionRequests stored
    List<String> storedActionTypes = actionRequestService.retrieveActionTypes();
    List<String> mappedActionTypes = templateMappingService.retrieveActionTypes();
    storedActionTypes.forEach((actionType) -> {
      if (!mappedActionTypes.contains(actionType)) {
        log.warn("No mapping for actionType : {}.", actionType);
      }
    });

    // Process templateMappings by file, have to as may be many actionTypes in
    // one file. Does not assume actionTypes in the same file use the same
    // template even so.
    String timeStamp = new SimpleDateFormat(DATE_FORMAT_IN_FILE_NAMES).format(Calendar.getInstance().getTime());
    actionExportLatchManager.setCountDownLatch(DISTRIBUTED_OBJECT_KEY_LATCH,
        actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT));
    templateMappingService.retrieveAllTemplateMappingsByFilename()
        .forEach((fileName, templatemappings) -> {
          log.info("Lock test {} {}", fileName, actionExportLockManager.isLocked(fileName));
          if (!actionExportLockManager.isLocked(fileName)) {
            if (actionExportLockManager.lock(fileName)) {
              log.info("Lock file {} {}", fileName, actionExportLockManager.isLocked(fileName));
              ExportMessage message = new ExportMessage();
              // process Collection of templateMappings
              templatemappings.forEach((templateMapping) -> {
                List<ActionRequestInstruction> requests = actionRequestService
                    .findByDateSentIsNullAndActionType(templateMapping.getActionType());
                if (requests.isEmpty()) {
                  log.info("No requests for actionType {} to process", templateMapping.getActionType());
                } else {
                  try {
                    transformationService.processActionRequests(message, requests);
                  } catch (CTPException e) {
                    // Error retrieving TemplateMapping in
                    // transformationService
                    log.error("Scheduled run error transforming ActionRequests");
                  }
                }
              });
              if (!message.isEmpty()) {
                sftpService.sendMessage(fileName + "_" + timeStamp + ".csv", message.getMergedActionRequestIds(),
                    message.getMergedOutputStreams());
              }
            }
          }
        });
    // Wait for all instances to finish to synchronise the removal of locks
    try {
      actionExportLatchManager.countDown(DISTRIBUTED_OBJECT_KEY_LATCH);
      if (!actionExportLatchManager.awaitCountDownLatch(DISTRIBUTED_OBJECT_KEY_LATCH)) {
        log.error("Scheduled run error countdownlatch timed out, should be {} instances running");
      }
    } catch (InterruptedException e) {
      log.error("Scheduled run error waiting for countdownlatch: {}", e.getMessage());
    } finally {
      actionExportLockManager.unlockInstanceLocks();
      actionExportLatchManager.deleteCountDownLatch(DISTRIBUTED_OBJECT_KEY_LATCH);
      log.info("{} {} instance/s running",
          actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT),
          DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
    }
  }
}
<commit_msg>CTPA-1022 Add instance count to an error log line.<commit_after>package uk.gov.ons.ctp.response.action.export.scheduled;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import javax.inject.Named;

import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.scheduling.annotation.Scheduled;

import lombok.extern.slf4j.Slf4j;
import uk.gov.ons.ctp.common.distributed.DistributedInstanceManager;
import uk.gov.ons.ctp.common.distributed.DistributedLatchManager;
import uk.gov.ons.ctp.common.distributed.DistributedLockManager;
import uk.gov.ons.ctp.common.error.CTPException;
import uk.gov.ons.ctp.response.action.export.domain.ActionRequestInstruction;
import uk.gov.ons.ctp.response.action.export.domain.ExportMessage;
import uk.gov.ons.ctp.response.action.export.message.SftpServicePublisher;
import uk.gov.ons.ctp.response.action.export.service.ActionRequestService;
import uk.gov.ons.ctp.response.action.export.service.TemplateMappingService;
import uk.gov.ons.ctp.response.action.export.service.TransformationService;

/**
 * This class will be responsible for the scheduling of export actions
 *
 */
@Named
@Slf4j
public class ExportScheduler implements HealthIndicator {

  private static final String DATE_FORMAT_IN_FILE_NAMES = "ddMMyyyy_HHmm";
  private static final String DISTRIBUTED_OBJECT_KEY_LATCH = "latch";
  private static final String DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT = "scheduler";

  @Inject
  private TransformationService transformationService;

  @Inject
  private TemplateMappingService templateMappingService;

  @Inject
  private SftpServicePublisher sftpService;

  @Inject
  private ActionRequestService actionRequestService;

  @Inject
  private DistributedLockManager actionExportLockManager;

  @Inject
  private DistributedInstanceManager actionExportInstanceManager;

  @Inject
  private DistributedLatchManager actionExportLatchManager;

  @Inject
  private ExportInfo exportInfo;

  @Override
  public Health health() {
    return Health.up()
        .withDetail("exportInfo", exportInfo)
        .build();
  }

  /**
   * Initialise export scheduler
   *
   */
  @PostConstruct
  public void init() {
    actionExportInstanceManager.incrementInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
    log.info("{} {} instance/s running",
        actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT),
        DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
  }

  /**
   * Clean up scheduler on bean destruction
   *
   */
  @PreDestroy
  public void cleanUp() {
    actionExportInstanceManager.decrementInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
    // Make sure no locks if interrupted in middle of run
    actionExportLockManager.unlockInstanceLocks();
    log.info("{} {} instance/s running",
        actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT),
        DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
  }

  /**
   * Carry out scheduled actions according to configured cron expression
   *
   */
  @Scheduled(cron = "#{appConfig.exportSchedule.cronExpression}")
  public void scheduleExport() {

    log.info("Scheduled run start");

    // Warn if Mapping document cannot deal with all ActionRequests stored
    List<String> storedActionTypes = actionRequestService.retrieveActionTypes();
    List<String> mappedActionTypes = templateMappingService.retrieveActionTypes();
    storedActionTypes.forEach((actionType) -> {
      if (!mappedActionTypes.contains(actionType)) {
        log.warn("No mapping for actionType : {}.", actionType);
      }
    });

    // Process templateMappings by file, have to as may be many actionTypes in
    // one file. Does not assume actionTypes in the same file use the same
    // template even so.
    String timeStamp = new SimpleDateFormat(DATE_FORMAT_IN_FILE_NAMES).format(Calendar.getInstance().getTime());
    actionExportLatchManager.setCountDownLatch(DISTRIBUTED_OBJECT_KEY_LATCH,
        actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT));
    templateMappingService.retrieveAllTemplateMappingsByFilename()
        .forEach((fileName, templatemappings) -> {
          log.info("Lock test {} {}", fileName, actionExportLockManager.isLocked(fileName));
          if (!actionExportLockManager.isLocked(fileName)) {
            if (actionExportLockManager.lock(fileName)) {
              log.info("Lock file {} {}", fileName, actionExportLockManager.isLocked(fileName));
              ExportMessage message = new ExportMessage();
              // process Collection of templateMappings
              templatemappings.forEach((templateMapping) -> {
                List<ActionRequestInstruction> requests = actionRequestService
                    .findByDateSentIsNullAndActionType(templateMapping.getActionType());
                if (requests.isEmpty()) {
                  log.info("No requests for actionType {} to process", templateMapping.getActionType());
                } else {
                  try {
                    transformationService.processActionRequests(message, requests);
                  } catch (CTPException e) {
                    // Error retrieving TemplateMapping in
                    // transformationService
                    log.error("Scheduled run error transforming ActionRequests");
                  }
                }
              });
              if (!message.isEmpty()) {
                sftpService.sendMessage(fileName + "_" + timeStamp + ".csv", message.getMergedActionRequestIds(),
                    message.getMergedOutputStreams());
              }
            }
          }
        });
    // Wait for all instances to finish to synchronise the removal of locks
    try {
      actionExportLatchManager.countDown(DISTRIBUTED_OBJECT_KEY_LATCH);
      if (!actionExportLatchManager.awaitCountDownLatch(DISTRIBUTED_OBJECT_KEY_LATCH)) {
        log.error("Scheduled run error countdownlatch timed out, should be {} instances running",
            actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT));
      }
    } catch (InterruptedException e) {
      log.error("Scheduled run error waiting for countdownlatch: {}", e.getMessage());
    } finally {
      actionExportLockManager.unlockInstanceLocks();
      actionExportLatchManager.deleteCountDownLatch(DISTRIBUTED_OBJECT_KEY_LATCH);
      log.info("{} {} instance/s running",
          actionExportInstanceManager.getInstanceCount(DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT),
          DISTRIBUTED_OBJECT_KEY_INSTANCE_COUNT);
    }
  }
}

{'fraction_non_alphanumeric': 0.08793334325249219, 'fraction_numerical': 0.00074393691414968, 'mean_word_length': 3.649948114839156, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2775541', 'n_tokens_mistral': 3616, 'n_tokens_neox': 3604, 'n_words': 693}