package com.hubspot.singularity.executor.task;

import java.lang.ProcessBuilder.Redirect;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;

import org.slf4j.Logger;

import com.google.common.collect.ImmutableList;
import com.hubspot.singularity.SingularityS3FormatHelper;
import com.hubspot.singularity.SingularityTaskId;
import com.hubspot.singularity.executor.SimpleProcessManager;
import com.hubspot.singularity.executor.TemplateManager;
import com.hubspot.singularity.executor.config.SingularityExecutorConfiguration;
import com.hubspot.singularity.executor.models.LogrotateTemplateContext;
import com.hubspot.singularity.runner.base.shared.JsonObjectFileHelper;
import com.hubspot.singularity.runner.base.shared.S3UploadMetadata;
import com.hubspot.singularity.runner.base.shared.TailMetadata;

public class SingularityExecutorTaskLogManager extends SimpleProcessManager {

  private final SingularityExecutorTaskDefinition taskDefinition;
  private final TemplateManager templateManager;
  private final SingularityExecutorConfiguration configuration;
  private final Logger log;
  private final JsonObjectFileHelper jsonObjectFileHelper;
  
  public SingularityExecutorTaskLogManager(SingularityExecutorTaskDefinition taskDefinition, TemplateManager templateManager, SingularityExecutorConfiguration configuration, Logger log, JsonObjectFileHelper jsonObjectFileHelper) {
    super(log);
    this.log = log;
    this.taskDefinition = taskDefinition;
    this.templateManager = templateManager;
    this.configuration = configuration;
    this.jsonObjectFileHelper = jsonObjectFileHelper;
  }

  public void setup() {
    ensureServiceOutExists();
    writeLogrotateFile();
    writeTailMetadata(false);
    writeS3MetadataFile(false);
  }
  
  private void writeLogrotateFile() {
    log.info("Writing logrotate configuration file to {}", getLogrotateConfPath());
    templateManager.writeLogrotateFile(getLogrotateConfPath(), new LogrotateTemplateContext(configuration, taskDefinition));
  }
   
  public boolean teardown() {
    boolean writeTailMetadataSuccess = writeTailMetadata(true);
    
    ensureServiceOutExists();
    copyLogTail();
    
    if (manualLogrotate()) {
      boolean removeLogRotateFileSuccess = removeLogrotateFile();
      boolean writeS3MetadataFileSuccess = writeS3MetadataFile(true);
      
      return writeTailMetadataSuccess && removeLogRotateFileSuccess && writeS3MetadataFileSuccess;
    } else {
      return false;
    }
  }
  
  private void copyLogTail() {
    if (configuration.getTailLogLinesToSave() <= 0) {
      return;
    }
   
    final List<String> cmd = ImmutableList.of(
        "tail", 
        "-n", 
        Integer.toString(configuration.getTailLogLinesToSave()), 
        taskDefinition.getServiceLogOut());
    
    try {
      super.runCommand(cmd, Redirect.to(taskDefinition.getTaskDirectoryPath().resolve(configuration.getServiceFinishedTailLog()).toFile()));
    } catch (Throwable t) {
      log.error("Failed saving tail of log {} to {}", taskDefinition.getServiceLogOut(), configuration.getServiceFinishedTailLog(), t);
    }
  }
  
  private boolean removeLogrotateFile() {
    boolean deleted = false;
    try {
      deleted = Files.deleteIfExists(getLogrotateConfPath());
    } catch (Throwable t) {
      log.trace("Couldn't delete {}", getLogrotateConfPath(), t);
      return false;
    }
    log.trace("Deleted {} : {}", getLogrotateConfPath(), deleted);
    return true;
  }
  
  public boolean manualLogrotate() {
    final List<String> command = ImmutableList.of(
        configuration.getLogrotateCommand(), 
        "-f",
        "-s",
        taskDefinition.getLogrotateStateFilePath().toString(),
        getLogrotateConfPath().toString());
    
    try {
      super.runCommand(command);
      return true;
    } catch (Throwable t) {
      log.warn("Tried to manually logrotate using {}, but caught", getLogrotateConfPath(), t);
      return false;
    }
  }
  
  private void ensureServiceOutExists() {
    try {
      if (!Files.exists(taskDefinition.getServiceLogOutPath())) {
        Files.createFile(taskDefinition.getServiceLogOutPath());
      }
    } catch (FileAlreadyExistsException faee) {
      log.debug("Executor out {} already existed", taskDefinition.getServiceLogOut());
    } catch (Throwable t) {
      log.error("Failed creating executor out {}", taskDefinition.getServiceLogOut(), t);
    }
  }
  
  private boolean writeTailMetadata(boolean finished) {
    if (!taskDefinition.getExecutorData().getLoggingTag().isPresent()) {
      if (!finished) {
        log.warn("Not writing logging metadata because logging tag is absent");
      }
      return true;
    }
    
    final TailMetadata tailMetadata = new TailMetadata(taskDefinition.getServiceLogOut(), taskDefinition.getExecutorData().getLoggingTag().get(), taskDefinition.getExecutorData().getLoggingExtraFields(), finished);
    final Path path = TailMetadata.getTailMetadataPath(configuration.getLogMetadataDirectory(), configuration.getLogMetadataSuffix(), tailMetadata);
    
    return jsonObjectFileHelper.writeObject(tailMetadata, path, log);
  }
  
  private String getS3Glob() {
    return String.format("%s*.gz*", taskDefinition.getServiceLogOutPath().getFileName());
  }
  
  private String getS3KeyPattern() {
    String s3KeyPattern = configuration.getS3KeyPattern();
    
    final SingularityTaskId singularityTaskId = getSingularityTaskId();
    
    return SingularityS3FormatHelper.getS3KeyFormat(s3KeyPattern, singularityTaskId, taskDefinition.getExecutorData().getLoggingTag());
  }
  
  private SingularityTaskId getSingularityTaskId() {
    return SingularityTaskId.fromString(taskDefinition.getTaskId());
  }
  
  public Path getLogrotateConfPath() {
    return configuration.getLogrotateConfDirectory().resolve(taskDefinition.getTaskId());
  }

  private boolean writeS3MetadataFile(boolean finished) {
    Path logrotateDirectory = taskDefinition.getServiceLogOutPath().getParent().resolve(configuration.getLogrotateToDirectory());
    
    S3UploadMetadata s3UploadMetadata = new S3UploadMetadata(logrotateDirectory.toString(), getS3Glob(), configuration.getS3Bucket(), getS3KeyPattern(), finished);
    
    String s3UploadMetadatafilename = String.format("%s%s", taskDefinition.getTaskId(), configuration.getS3MetadataSuffix());
    
    Path s3UploadMetadataPath = configuration.getS3MetadataDirectory().resolve(s3UploadMetadatafilename);
    
    return jsonObjectFileHelper.writeObject(s3UploadMetadata, s3UploadMetadataPath, log);
  }
  
}

@@ -3563,24 +3563,192 @@
 grotate() %7B%0A
+    if (!Files.exists(getLogrotateConfPath())) %7B%0A      log.info(%22%7B%7D did not exist, skipping manual logrotation%22, getLogrotateConfPath());%0A      return true;%0A    %7D%0A    %0A
     final Li

fix minor cleanup issue