Skip to content

Optionally compress files prior to uploading them in SingularityS3Uploader. #2195

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pschoenfelder merged 7 commits into from
Apr 12, 2021
Merged

Optionally compress files prior to uploading them in SingularityS3Uploader. #2195

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f28bef6
Run a final logrotate on hourly and size based files as well
ssalinas Mar 18, 2021
62af120
Optionally compress files prior to uploading them.
baconmania Apr 5, 2021
62863b0
File glob logic needs to be adjusted as well, see comments for details.
baconmania Apr 5, 2021
ded3a70
Comment.
baconmania Apr 5, 2021
1304ff7
Merge branch 'executor_rotate_on_shutdown' into compress-before-upload
ssalinas Apr 7, 2021
707824f
Fix merge with rotate on shutdown
ssalinas Apr 7, 2021
a5e3028
Fix in more places
ssalinas Apr 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion SingularityBase/src/main/java/com/hubspot/singularity/SingularityS3UploaderFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ public class SingularityS3UploaderFile {
private final Optional<String> s3StorageClass;
private final Optional<Long> applyS3StorageClassAfterBytes;
private final boolean checkSubdirectories;
private final boolean compressBeforeUpload;

@JsonCreator
public static SingularityS3UploaderFile fromString(String value) {
Expand All @@ -28,6 +29,7 @@ public static SingularityS3UploaderFile fromString(String value) {
Optional.empty(),
Optional.empty(),
Optional.empty(),
Optional.empty(),
Optional.empty()
);
}
Expand All @@ -43,7 +45,8 @@ public SingularityS3UploaderFile(
@JsonProperty(
"applyS3StorageClassAfterBytes"
) Optional<Long> applyS3StorageClassAfterBytes,
@JsonProperty("checkSubdirectories") Optional<Boolean> checkSubdirectories
@JsonProperty("checkSubdirectories") Optional<Boolean> checkSubdirectories,
@JsonProperty("compressBeforeUpload") Optional<Boolean> compressBeforeUpload
) {
this.filename = filename;
this.s3UploaderBucket = s3UploaderBucket;
Expand All @@ -53,6 +56,7 @@ public SingularityS3UploaderFile(
this.s3StorageClass = s3StorageClass;
this.applyS3StorageClassAfterBytes = applyS3StorageClassAfterBytes;
this.checkSubdirectories = checkSubdirectories.orElse(false);
this.compressBeforeUpload = compressBeforeUpload.orElse(false);
}

@Schema(description = "The name of the file")
Expand Down Expand Up @@ -121,6 +125,11 @@ public boolean isCheckSubdirectories() {
return checkSubdirectories;
}

@Schema(title = "Compress to GZIP before uploading", defaultValue = "false")
public boolean isCompressBeforeUpload() {
return compressBeforeUpload;
}

@Override
public String toString() {
return (
Expand Down
72 changes: 56 additions & 16 deletions ...rc/main/java/com/hubspot/singularity/executor/task/SingularityExecutorTaskLogManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,10 +169,23 @@ private boolean writeS3MetadataFileForRotatedFiles(boolean finished) {
continue;
}

String fileGlob = additionalFile.getFilename() != null &&
additionalFile.getFilename().contains("*")
? additionalFile.getFilename()
: String.format("%s*.[gb]z*", additionalFile.getFilename());
String fileGlob;

if (additionalFile.isCompressBeforeUpload()) {
// We have to compress it before the upload, so just look for the filename prefix
// If it's already compressed, we'll skip compression in SingularityS3Uploader and upload it as-is.
fileGlob = String.format("%s*", additionalFile.getFilename());
} else if (
additionalFile.getFilename() != null && additionalFile.getFilename().contains("*")
) {
// There's already a wildcard in the filename, so pass the glob through as-is
fileGlob = additionalFile.getFilename();
} else {
// We're not compressing it ourselves in SingularityS3Uploader, and there's no wildcard,
// so look for a compressed file with this filename prefix
fileGlob = String.format("%s*.[gb]z*", additionalFile.getFilename());
}

result =
result &&
writeS3MetadataFile(
Expand All @@ -190,7 +203,8 @@ private boolean writeS3MetadataFileForRotatedFiles(boolean finished) {
additionalFile.getApplyS3StorageClassAfterBytes().isPresent()
? additionalFile.getApplyS3StorageClassAfterBytes()
: taskDefinition.getExecutorData().getApplyS3StorageClassAfterBytes(),
additionalFile.isCheckSubdirectories()
additionalFile.isCheckSubdirectories(),
additionalFile.isCompressBeforeUpload()
);
index++;
handledLogs.add(additionalFile.getFilename());
Expand All @@ -212,6 +226,7 @@ private boolean writeS3MetadataFileForRotatedFiles(boolean finished) {
finished,
taskDefinition.getExecutorData().getS3StorageClass(),
taskDefinition.getExecutorData().getApplyS3StorageClassAfterBytes(),
false,
false
);
}
Expand Down Expand Up @@ -383,6 +398,7 @@ public boolean teardown() {
true,
taskDefinition.getExecutorData().getS3StorageClass(),
taskDefinition.getExecutorData().getApplyS3StorageClassAfterBytes(),
false,
false
);
}
Expand Down Expand Up @@ -512,21 +528,43 @@ public boolean removeLogrotateFile() {
}

public boolean manualLogrotate() {
if (!Files.exists(getLogrotateConfPath())) {
log.info("{} did not exist, skipping manual logrotation", getLogrotateConfPath());
Set<Path> frequencyBasedPaths = getCronFakedLogrotateAdditionalFileFrequencies()
.stream()
.map(this::getLogrotateHourlyConfPath)
.collect(Collectors.toSet());
boolean regularConfExists = Files.exists(getLogrotateConfPath());
boolean hourlyConfExists = frequencyBasedPaths
.stream()
.anyMatch(p -> Files.exists(p));
boolean sizeConfExists = Files.exists(getLogrotateSizeBasedConfPath());
if (!sizeConfExists && !hourlyConfExists && !regularConfExists) {
log.info(
"{}/{}/{} did not exist, skipping manual logrotation",
getLogrotateConfPath(),
frequencyBasedPaths,
getLogrotateSizeBasedConfPath()
);
return true;
}

final List<String> command = ImmutableList.of(
configuration.getLogrotateCommand(),
"-f",
"-s",
taskDefinition.getLogrotateStateFilePath().toString(),
getLogrotateConfPath().toString()
);
final ImmutableList.Builder<String> command = ImmutableList.builder();
command.add(configuration.getLogrotateCommand());
command.add("-f");
command.add("-s");
command.add(taskDefinition.getLogrotateStateFilePath().toString());

if (regularConfExists) {
command.add(getLogrotateConfPath().toString());
}
if (hourlyConfExists) {
frequencyBasedPaths.forEach(p -> command.add(p.toString()));
}
if (sizeConfExists) {
command.add(getLogrotateSizeBasedConfPath().toString());
}

try {
new SimpleProcessManager(log).runCommand(command);
new SimpleProcessManager(log).runCommand(command.build());
return true;
} catch (Throwable t) {
log.warn(
Expand Down Expand Up @@ -643,7 +681,8 @@ private boolean writeS3MetadataFile(
boolean finished,
Optional<String> s3StorageClass,
Optional<Long> applyS3StorageClassAfterBytes,
boolean checkSubdirectories
boolean checkSubdirectories,
boolean compressBeforeUpload
) {
final String s3UploaderBucket = s3Bucket.orElse(
taskDefinition.getExecutorData().getDefaultS3Bucket()
Expand Down Expand Up @@ -674,6 +713,7 @@ private boolean writeS3MetadataFile(
applyS3StorageClassAfterBytes,
Optional.of(finished),
Optional.of(checkSubdirectories),
Optional.of(compressBeforeUpload),
Optional.empty(),
Collections.emptyMap(),
Optional.empty(),
Expand Down
7 changes: 7 additions & 0 deletions ...RunnerBase/src/main/java/com/hubspot/singularity/runner/base/shared/S3UploadMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ public class S3UploadMetadata {
private final Optional<Long> applyStorageClassIfOverBytes;
private final Optional<Boolean> uploadImmediately;
private final boolean checkSubdirectories;
private final boolean compressBeforeUpload;
private final SingularityUploaderType uploaderType;
private final Map<String, Object> gcsCredentials;
private final Optional<String> gcsStorageClass;
Expand All @@ -80,6 +81,7 @@ public S3UploadMetadata(
) Optional<Long> applyStorageClassIfOverBytes,
@JsonProperty("uploadImmediately") Optional<Boolean> uploadImmediately,
@JsonProperty("checkSubdirectories") Optional<Boolean> checkSubdirectories,
@JsonProperty("compressBeforeUpload") Optional<Boolean> compressBeforeUpload,
@JsonProperty("uploaderType") Optional<SingularityUploaderType> uploaderType,
@JsonProperty("gcsCredentials") Map<String, Object> gcsCredentials,
@JsonProperty("gcsStorageClass") Optional<String> gcsStorageClass,
Expand All @@ -105,6 +107,7 @@ public S3UploadMetadata(
this.applyStorageClassIfOverBytes = applyStorageClassIfOverBytes;
this.uploadImmediately = uploadImmediately;
this.checkSubdirectories = checkSubdirectories.orElse(false);
this.compressBeforeUpload = compressBeforeUpload.orElse(false);
this.uploaderType = uploaderType.orElse(SingularityUploaderType.S3);
this.gcsCredentials =
gcsCredentials != null ? gcsCredentials : Collections.emptyMap();
Expand Down Expand Up @@ -211,6 +214,10 @@ public boolean isCheckSubdirectories() {
return checkSubdirectories;
}

public boolean isCompressBeforeUpload() {
return compressBeforeUpload;
}

public SingularityUploaderType getUploaderType() {
return uploaderType;
}
Expand Down
35 changes: 34 additions & 1 deletion ...arityS3Uploader/src/main/java/com/hubspot/singularity/s3uploader/SingularityUploader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
import com.google.common.collect.Sets;
import com.hubspot.mesos.JavaUtils;
import com.hubspot.singularity.runner.base.sentry.SingularityRunnerExceptionNotifier;
import com.hubspot.singularity.runner.base.shared.CompressionType;
import com.hubspot.singularity.runner.base.shared.ProcessFailedException;
import com.hubspot.singularity.runner.base.shared.S3UploadMetadata;
import com.hubspot.singularity.runner.base.shared.SimpleProcessManager;
import com.hubspot.singularity.s3uploader.config.SingularityS3UploaderConfiguration;
Expand Down Expand Up @@ -255,7 +257,38 @@ private AtomicInteger handleFile(Path path, boolean isFinished, List<Path> toUpl

found.incrementAndGet();

toUpload.add(path);
if (uploadMetadata.isCompressBeforeUpload() && !path.toString().endsWith(".gz")) {
Copy link
Member

@ssalinas ssalinas Apr 6, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if we are checking for 'isAlreadyGzipped' here, should we also be checking that in the file glob section? E.g. if something else already zipped it, it should still match the toUpload glob as well

Copy link
Member

@ssalinas ssalinas Apr 12, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nevermind, stared down the section above a bit more, we are catching those other cases

try {
LOG.trace("{} Compressing {}...", logIdentifier, path);
long start = -1;
if (LOG.isTraceEnabled()) {
start = System.currentTimeMillis();
}

new SimpleProcessManager(LOG)
.runCommand(ImmutableList.of(CompressionType.GZIP.getCommand(), path.toString()));

if (LOG.isTraceEnabled()) {
LOG.trace(
"{} Compressed {} in {}ms",
logIdentifier,
path,
System.currentTimeMillis() - start
);
}

toUpload.add(Paths.get(path.toAbsolutePath().toString() + ".gz"));
} catch (InterruptedException | ProcessFailedException e) {
LOG.warn(
"{} Skipping {} because we were unable to compress it",
logIdentifier,
path,
e
);
}
} else {
toUpload.add(path);
}

return found;
}
Expand Down