Skip to content

Commit 5a1f0b6

Browse files
tburchtburch
committed
break out cleanup into several Runnables
1 parent 21dc07f commit 5a1f0b6

File tree

4 files changed

+133
-65
lines changed

4 files changed

+133
-65
lines changed

‎src/main/java/com/lowtuna/jsonblob/core/BlobCleanupJob.java‎

Lines changed: 14 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,16 @@
11
package com.lowtuna.jsonblob.core;
22

33
import com.fasterxml.jackson.databind.ObjectMapper;
4-
import com.google.common.base.Function;
5-
import com.google.common.base.Stopwatch;
64
import com.google.common.collect.Lists;
7-
import com.google.common.collect.Maps;
8-
import com.google.common.collect.Sets;
95
import io.dropwizard.util.Duration;
106
import lombok.RequiredArgsConstructor;
117
import lombok.extern.slf4j.Slf4j;
12-
import org.joda.time.DateTime;
138

14-
import javax.annotation.Nullable;
159
import java.io.File;
16-
import java.io.IOException;
1710
import java.nio.file.Files;
1811
import java.nio.file.Path;
19-
import java.util.Arrays;
2012
import java.util.List;
21-
import java.util.Map;
22-
import java.util.Set;
23-
import java.util.concurrent.TimeUnit;
24-
import java.util.concurrent.atomic.AtomicInteger;
25-
import java.util.stream.Collectors;
13+
import java.util.concurrent.ExecutorService;
2614

2715
@Slf4j
2816
@RequiredArgsConstructor
@@ -32,69 +20,31 @@ public class BlobCleanupJob implements Runnable {
3220
private final FileSystemJsonBlobManager fileSystemJsonBlobManager;
3321
private final ObjectMapper om;
3422
private final boolean deleteEnabled;
23+
private final ExecutorService executorService;
3524

3625
@Override
3726
public void run() {
38-
Stopwatch stopwatch = new Stopwatch().start();
39-
AtomicInteger blobsRemoved = new AtomicInteger(0);
4027
try {
41-
Set<String> blobsToDelete = Sets.newCopyOnWriteArraySet();
28+
List<String> dataDirs = Lists.newCopyOnWriteArrayList();
29+
4230
Files.walk(blobDirectory)
4331
.parallel()
4432
.filter(p -> !p.toFile().isDirectory())
4533
.map(Path::getParent)
4634
.distinct()
47-
.forEach(dataDir -> {
48-
log.info("Checking for blobs not accessed in the last {} in {}", blobAccessTtl, dataDir.toAbsolutePath());
49-
if (!dataDir.toFile().exists() || !dataDir.toFile().isDirectory()) {
50-
return;
51-
}
52-
try {
53-
List<File> files = Arrays.asList(dataDir.toFile().listFiles()).parallelStream().filter(File::exists).collect(Collectors.toList());
54-
Set<String> blobs = Sets
55-
.newHashSet(Lists.transform(files, f -> f.getName().split("\\.", 2)[0]))
56-
.parallelStream()
57-
.filter(f -> fileSystemJsonBlobManager.resolveTimestamp(f).isPresent()).collect(Collectors.toSet());
58-
log.info("Identified {} blobs in {}", blobs.size(), dataDir);
59-
Map<String, DateTime> lastAccessed = Maps.newHashMap(Maps.asMap(blobs, new Function<String, DateTime>() {
60-
@Nullable
61-
@Override
62-
public DateTime apply(@Nullable String input) {
63-
return fileSystemJsonBlobManager.resolveTimestamp(input).get();
64-
}
65-
}));
66-
log.debug("Completed building map of {} last accessed timestamps in {}", lastAccessed.size(), dataDir);
35+
.forEach(dataDir -> dataDirs.add(dataDir.toFile().getAbsolutePath()));
6736

68-
File metadataFile = fileSystemJsonBlobManager.getMetaDataFile(dataDir.toFile());
69-
try {
70-
BlobMetadataContainer metadataContainer = metadataFile.exists() ? om.readValue(fileSystemJsonBlobManager.readFile(metadataFile), BlobMetadataContainer.class) : new BlobMetadataContainer();
71-
log.debug("Adding {} last accessed timestamp from metadata {}", metadataContainer.getLastAccessedByBlobId().size(), metadataFile.getAbsolutePath());
72-
lastAccessed.putAll(metadataContainer.getLastAccessedByBlobId());
73-
log.debug("Determining which blobs to remove from {}", dataDir);
74-
Map<String, DateTime> toRemove = Maps.filterEntries(lastAccessed, input -> input.getValue().plusMillis((int) blobAccessTtl.toMilliseconds()).isBefore(DateTime.now()));
75-
log.info("Identified {} blobs to remove in {}", toRemove.size(), dataDir);
76-
blobsToDelete.addAll(toRemove.keySet());
77-
} catch (IOException e) {
78-
log.warn("Couldn't load metadata file from {}", dataDir.toAbsolutePath(), e);
79-
}
80-
} catch (Exception e) {
81-
log.warn("Caught Exception while trying to remove un-accessed blobs in {}", dataDir, e);
82-
}
83-
});
37+
log.debug("Found {} data directories", dataDirs.size());
8438

85-
log.info("Deleting {} blobs", blobsToDelete.size());
86-
blobsToDelete.parallelStream().forEach(blobId -> {
87-
if (deleteEnabled) {
88-
log.debug("Deleting blob with id {}", blobId);
89-
try {
90-
fileSystemJsonBlobManager.deleteBlob(blobId);
91-
blobsRemoved.incrementAndGet();
92-
} catch (BlobNotFoundException e) {
93-
log.debug("Couldn't delete blobId {} because it's already been deleted", blobId);
94-
}
39+
for(String dataDirPath: dataDirs) {
40+
File dir = new File(dataDirPath);
41+
if (dir.listFiles().length == 0) {
42+
dir.delete();
9543
}
96-
});
97-
log.info("Completed cleanup of {} blobs in {}ms", blobsRemoved.get(), stopwatch.elapsed(TimeUnit.MILLISECONDS));
44+
45+
log.debug("Submitting DataDirectoryCleanupJob for {}", dataDirPath);
46+
executorService.submit(new DataDirectoryCleanupJob(dataDirPath, executorService, fileSystemJsonBlobManager,blobAccessTtl, om, deleteEnabled));
47+
}
9848
} catch (Exception e) {
9949
log.warn("Couldn't remove old blobs", e);
10050
}

‎src/main/java/com/lowtuna/jsonblob/core/BulkBlobDeleteJob.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
package com.lowtuna.jsonblob.core;
2+
3+
import lombok.Data;
4+
import lombok.extern.slf4j.Slf4j;
5+
6+
import java.util.Set;
7+
8+
/**
9+
* Created by tburch on 8/16/17.
10+
*/
11+
@Data
12+
@Slf4j
13+
public class BulkBlobDeleteJob implements Runnable {
14+
private final Set<String> blobsToDelete;
15+
private final FileSystemJsonBlobManager fileSystemJsonBlobManager;
16+
private final boolean deleteEnabled;
17+
18+
@Override
19+
public void run() {
20+
try {
21+
log.info("Deleting {} blobs", blobsToDelete.size());
22+
blobsToDelete.parallelStream().forEach(blobId -> {
23+
if (deleteEnabled) {
24+
log.debug("Deleting blob with id {}", blobId);
25+
try {
26+
fileSystemJsonBlobManager.deleteBlob(blobId);
27+
} catch (BlobNotFoundException e) {
28+
log.debug("Couldn't delete blobId {} because it's already been deleted", blobId);
29+
}
30+
}
31+
});
32+
} catch (Exception e) {
33+
log.warn("Caught exception while trying to delete {} blobs", blobsToDelete.size(), e);
34+
}
35+
}
36+
}

‎src/main/java/com/lowtuna/jsonblob/core/DataDirectoryCleanupJob.java‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
package com.lowtuna.jsonblob.core;
2+
3+
import com.fasterxml.jackson.databind.ObjectMapper;
4+
import com.google.common.base.Function;
5+
import com.google.common.collect.Lists;
6+
import com.google.common.collect.Maps;
7+
import com.google.common.collect.Sets;
8+
import io.dropwizard.util.Duration;
9+
import lombok.Data;
10+
import lombok.extern.slf4j.Slf4j;
11+
import org.joda.time.DateTime;
12+
13+
import javax.annotation.Nullable;
14+
import java.io.File;
15+
import java.io.IOException;
16+
import java.util.Arrays;
17+
import java.util.List;
18+
import java.util.Map;
19+
import java.util.Set;
20+
import java.util.concurrent.ExecutorService;
21+
import java.util.stream.Collectors;
22+
23+
/**
24+
* Created by tburch on 8/16/17.
25+
*/
26+
@Data
27+
@Slf4j
28+
public class DataDirectoryCleanupJob implements Runnable {
29+
private final String dataDirPath;
30+
private final ExecutorService executorService;
31+
private final FileSystemJsonBlobManager fileSystemJsonBlobManager;
32+
private final Duration blobAccessTtl;
33+
private final ObjectMapper om;
34+
private final boolean deleteEnabled;
35+
36+
@Override
37+
public void run() {
38+
File dir = new File(dataDirPath);
39+
40+
log.info("Checking for blobs not accessed in the last {} in {}", blobAccessTtl, dir.getAbsolutePath());
41+
if (!dir.exists() || !dir.isDirectory()) {
42+
return;
43+
}
44+
try {
45+
List<File> files = Arrays.asList(dir.listFiles()).parallelStream().filter(File::exists).collect(Collectors.toList());
46+
Set<String> blobs = Sets
47+
.newHashSet(Lists.transform(files, f -> f.getName().split("\\.", 2)[0]))
48+
.parallelStream()
49+
.filter(f -> fileSystemJsonBlobManager.resolveTimestamp(f).isPresent()).collect(Collectors.toSet());
50+
log.info("Identified {} blobs in {}", blobs.size(), dir);
51+
Map<String, DateTime> lastAccessed = Maps.newHashMap(Maps.asMap(blobs, new Function<String, DateTime>() {
52+
@Nullable
53+
@Override
54+
public DateTime apply(@Nullable String input) {
55+
return fileSystemJsonBlobManager.resolveTimestamp(input).get();
56+
}
57+
}));
58+
log.debug("Completed building map of {} last accessed timestamps in {}", lastAccessed.size(), dir);
59+
60+
File metadataFile = fileSystemJsonBlobManager.getMetaDataFile(dir);
61+
try {
62+
BlobMetadataContainer metadataContainer = metadataFile.exists() ? om.readValue(fileSystemJsonBlobManager.readFile(metadataFile), BlobMetadataContainer.class) : new BlobMetadataContainer();
63+
log.debug("Adding {} last accessed timestamp from metadata {}", metadataContainer.getLastAccessedByBlobId().size(), metadataFile.getAbsolutePath());
64+
lastAccessed.putAll(metadataContainer.getLastAccessedByBlobId());
65+
log.debug("Determining which blobs to remove from {}", dir);
66+
Map<String, DateTime> toRemove = Maps.filterEntries(lastAccessed, input -> input.getValue().plusMillis((int) blobAccessTtl.toMilliseconds()).isBefore(DateTime.now()));
67+
log.info("Identified {} blobs to remove in {}", toRemove.size(), dir);
68+
69+
if (toRemove.keySet().isEmpty()) {
70+
return;
71+
}
72+
73+
log.debug("Submitting BulkBlobDeleteJob for {} blobs in {}", toRemove.size(), dir);
74+
executorService.submit(new BulkBlobDeleteJob(toRemove.keySet(), fileSystemJsonBlobManager, deleteEnabled));
75+
} catch (IOException e) {
76+
log.warn("Couldn't load metadata file from {}", dir.getAbsolutePath(), e);
77+
}
78+
} catch (Exception e) {
79+
log.warn("Caught Exception while trying to remove un-accessed blobs in {}", dir, e);
80+
}
81+
}
82+
}

‎src/main/java/com/lowtuna/jsonblob/core/FileSystemJsonBlobManager.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -304,7 +304,7 @@ public void start() throws Exception {
304304
scheduledExecutorService.scheduleWithFixedDelay(this, 1, 1, TimeUnit.MINUTES);
305305

306306
log.info("Scheduling blob cleanup job");
307-
scheduledExecutorService.scheduleWithFixedDelay(new BlobCleanupJob(blobDataDirectory.toPath(), blobAccessTtl, this, objectMapper, deleteEnabled), 0, 1, TimeUnit.DAYS);
307+
scheduledExecutorService.scheduleWithFixedDelay(new BlobCleanupJob(blobDataDirectory.toPath(), blobAccessTtl, this, objectMapper, deleteEnabled, scheduledExecutorService), 0, 1, TimeUnit.DAYS);
308308
}
309309

310310
@Override

0 commit comments

Comments
 (0)