package com.hubspot.singularity.scheduler;

import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import javax.inject.Singleton;

import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.inject.Inject;
import com.hubspot.singularity.SingularityAbort;
import com.hubspot.singularity.SingularityDeployStatistics;
import com.hubspot.singularity.SingularityRequestWithState;
import com.hubspot.singularity.SingularityTaskId;
import com.hubspot.singularity.config.SingularityConfiguration;
import com.hubspot.singularity.data.DeployManager;
import com.hubspot.singularity.data.RequestManager;
import com.hubspot.singularity.data.TaskManager;
import com.hubspot.singularity.mesos.SingularityMesosSchedulerDelegator;
import com.hubspot.singularity.sentry.SingularityExceptionNotifier;
import com.hubspot.singularity.smtp.SingularityMailer;

@Singleton
public class SingularityScheduledJobPoller extends SingularityLeaderOnlyPoller {

  private static final Logger LOG = LoggerFactory.getLogger(SingularityScheduledJobPoller.class);

  private final TaskManager taskManager;
  private final RequestManager requestManager;
  private final DeployManager deployManager;
  private final SingularityMailer mailer;
  private final SingularityConfiguration configuration;
  private final SingularityExceptionNotifier exceptionNotifier;

  @Inject
  public SingularityScheduledJobPoller(final LeaderLatch leaderLatch, final SingularityMesosSchedulerDelegator mesosScheduler, SingularityExceptionNotifier exceptionNotifier, TaskManager taskManager,
      SingularityConfiguration configuration, SingularityAbort abort, RequestManager requestManager, DeployManager deployManager, SingularityMailer mailer) {
    super(leaderLatch, mesosScheduler, exceptionNotifier, abort, configuration.getCheckScheduledJobsEveryMillis(), TimeUnit.MILLISECONDS, SchedulerLockType.NO_LOCK);

    this.taskManager = taskManager;
    this.deployManager = deployManager;
    this.configuration = configuration;
    this.exceptionNotifier = exceptionNotifier;
    this.requestManager = requestManager;
    this.mailer = mailer;
  }

  @Override
  public void runActionOnPoll() {
    final long start = System.currentTimeMillis();

    final List<SingularityTaskId> activeTaskIds = taskManager.getActiveTaskIds();
    final List<String> requestIdsToLookup = Lists.newArrayListWithCapacity(activeTaskIds.size());

    for (SingularityTaskId taskId : activeTaskIds) {
      if (start - taskId.getStartedAt() < configuration.getWarnIfScheduledJobIsRunningForAtLeastMillis()) {
        continue;
      }

      requestIdsToLookup.add(taskId.getRequestId());
    }

    final List<SingularityRequestWithState> requests = requestManager.getRequests(requestIdsToLookup);
    final Map<String, SingularityRequestWithState> idToRequest = Maps.uniqueIndex(requests, SingularityRequestWithState.REQUEST_STATE_TO_REQUEST_ID);

    for (SingularityTaskId taskId : activeTaskIds) {
      SingularityRequestWithState request = idToRequest.get(taskId.getRequestId());

      if (request == null || !request.getRequest().isScheduled() || taskManager.hasNotifiedOverdue(taskId)) {
        continue;
      }

      final long runtime = start - taskId.getStartedAt();
      final Optional<Long> expectedRuntime = getExpectedRuntime(request, taskId);

      if (!expectedRuntime.isPresent()) {
        continue;
      }

      final int overDuePct = (int) (100 * (runtime / (float) expectedRuntime.get()));

      if (overDuePct > configuration.getWarnIfScheduledJobIsRunningPastNextRunPct()) {
        LOG.info("{} is overdue by {}% (expectedRunTime: {}, warnIfScheduledJobIsRunningPastNextRunPct: {})", taskId, overDuePct, expectedRuntime.get(), configuration.getWarnIfScheduledJobIsRunningPastNextRunPct());

        mailer.sendTaskOverdueMail(taskId, request.getRequest(), runtime, expectedRuntime.get());

        taskManager.saveNotifiedOverdue(taskId);
      }
    }
  }

  private Optional<Long> getExpectedRuntime(SingularityRequestWithState request, SingularityTaskId taskId) {
    if (request.getRequest().getScheduledExpectedRuntimeMillis().isPresent()) {
      return request.getRequest().getScheduledExpectedRuntimeMillis();
    } else {
      final Optional<SingularityDeployStatistics> deployStatistics = deployManager.getDeployStatistics(taskId.getRequestId(), taskId.getDeployId());

      if (deployStatistics.isPresent() && deployStatistics.get().getAverageRuntimeMillis().isPresent()) {
        return deployStatistics.get().getAverageRuntimeMillis();
      }

      final CronExpression cronExpression;

      try {
        cronExpression = new CronExpression(request.getRequest().getQuartzScheduleSafe());
      } catch (ParseException e) {
        LOG.warn("Unable to parse cron for {} ({})", taskId, request.getRequest().getQuartzScheduleSafe(), e);
        exceptionNotifier.notify(e);
        return Optional.absent();
      }

      final Date startDate = new Date(taskId.getStartedAt());
      final Date nextRunAtDate = cronExpression.getNextValidTimeAfter(startDate);

      if (nextRunAtDate == null) {
        String msg = String.format("No next run date found for %s (%s)", taskId, request.getRequest().getQuartzScheduleSafe());
        LOG.warn(msg);
        exceptionNotifier.notify(msg);
        return Optional.absent();
      }

      return Optional.of(nextRunAtDate.getTime() - taskId.getStartedAt());
    }
  }

}

@@ -4214,16 +4214,228 @@
 askId);%0A
+      %7D else %7B%0A        LOG.trace(%22%7B%7D is not overdue yet - runtime %7B%7D, expected %7B%7D, (%7B%7D%25 %3C %7B%7D%25)%22, taskId, runtime, expectedRuntime.get(), overDuePct, configuration.getWarnIfScheduledJobIsRunningPastNextRunPct());%0A
       %7D%0A

add trace stmt