Skip to content

Various bug fixes and stability improvements #2281

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
ssalinas merged 10 commits into from
Apr 6, 2022
Merged

Various bug fixes and stability improvements #2281

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
1e2255d
Fix lb deletes that will never finish
ssalinas Apr 5, 2022
f45b1cb
race condition causing NPE
ssalinas Apr 5, 2022
d0c71f7
curator cleanup and avoid nulls
ssalinas Apr 5, 2022
fc3568f
Fix possible missing pending tasks on startup
ssalinas Apr 5, 2022
ecf12ba
Skip tasks which have no active task data present
ssalinas Apr 5, 2022
1349516
More sentry fix
ssalinas Apr 5, 2022
c042582
run later in startup
ssalinas Apr 5, 2022
8d9f55d
Fix test for new missing task data handling
ssalinas Apr 5, 2022
c796146
Add option for additional debug verification of task writes
ssalinas Apr 6, 2022
d4b6b68
consistent sort value
ssalinas Apr 6, 2022
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
14 changes: 9 additions & 5 deletions SingularityBase/src/main/java/com/hubspot/singularity/SingularityTaskIdHistory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ public class SingularityTaskIdHistory implements Comparable<SingularityTaskIdHis
private final Optional<ExtendedTaskState> lastTaskState;
private final Optional<String> runId;

@SuppressFBWarnings("NP_NULL_PARAM_DEREF")
public static SingularityTaskIdHistory fromTaskIdAndTaskAndUpdates(
SingularityTaskId taskId,
SingularityTask task,
Expand All @@ -26,10 +25,15 @@ public static SingularityTaskIdHistory fromTaskIdAndTaskAndUpdates(
ExtendedTaskState lastTaskState = null;
long updatedAt = taskId.getStartedAt();

if (updates != null && !updates.isEmpty()) {
SingularityTaskHistoryUpdate lastUpdate = Collections.max(updates);
lastTaskState = lastUpdate.getTaskState();
updatedAt = lastUpdate.getTimestamp();
if (updates != null) {
Optional<SingularityTaskHistoryUpdate> maybeLastUpdate = updates
.stream()
.filter(Objects::nonNull)
.max(SingularityTaskHistoryUpdate::compareTo);
if (maybeLastUpdate.isPresent()) {
lastTaskState = maybeLastUpdate.get().getTaskState();
updatedAt = maybeLastUpdate.get().getTimestamp();
}
}

return new SingularityTaskIdHistory(
Expand Down
9 changes: 9 additions & 0 deletions ...ularityService/src/main/java/com/hubspot/singularity/config/SingularityConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -458,6 +458,7 @@ public class SingularityConfiguration extends Configuration {
private boolean skipPersistingTooLongTaskIds = false;

private boolean allowEmptyRequestInstances = false;
private boolean verifyTaskDataWrites = false;

public long getAskDriverToKillTasksAgainAfterMillis() {
return askDriverToKillTasksAgainAfterMillis;
Expand Down Expand Up @@ -2151,4 +2152,12 @@ public boolean allowEmptyRequestInstances() {
public void setAllowEmptyRequestInstances(boolean allowEmptyRequestInstances) {
this.allowEmptyRequestInstances = allowEmptyRequestInstances;
}

public boolean isVerifyTaskDataWrites() {
return verifyTaskDataWrites;
}

public void setVerifyTaskDataWrites(boolean verifyTaskDataWrites) {
this.verifyTaskDataWrites = verifyTaskDataWrites;
}
}
259 changes: 100 additions & 159 deletions SingularityService/src/main/java/com/hubspot/singularity/data/CuratorAsyncManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
Expand Down Expand Up @@ -57,22 +58,16 @@ private <T> List<T> getAsyncChildrenThrows(
final Transcoder<T> transcoder
)
throws Exception {
try {
List<String> children = getChildren(parent);
final List<String> paths = Lists.newArrayListWithCapacity(children.size());

for (String child : children) {
paths.add(ZKPaths.makePath(parent, child));
}

List<T> result = new ArrayList<>(
getAsyncThrows(parent, paths, transcoder, Optional.empty()).values()
);
List<String> children = getChildren(parent);
final List<String> paths = Lists.newArrayListWithCapacity(children.size());

return result;
} catch (Throwable t) {
throw t;
for (String child : children) {
paths.add(ZKPaths.makePath(parent, child));
}

return new ArrayList<>(
getAsyncThrows(parent, paths, transcoder, Optional.empty()).values()
);
}

private <T> Map<String, T> getAsyncThrows(
Expand Down Expand Up @@ -104,29 +99,22 @@ private <T> Map<String, T> getAsyncThrows(
final CountDownLatch latch = new CountDownLatch(paths.size());
final AtomicInteger bytes = new AtomicInteger();

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}

bytes.getAndAdd(event.getData().length);
final T object = transcoder.fromBytes(event.getData());
synchronizedObjects.put(event.getPath(), object);

if (cache.isPresent()) {
cache.get().set(event.getPath(), object);
}
} catch (Exception e) {
LOG.error("Exception processing curator result", e);
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}

bytes.getAndAdd(event.getData().length);
final T object = transcoder.fromBytes(event.getData());
synchronizedObjects.put(event.getPath(), object);

cache.ifPresent(tZkCache -> tZkCache.set(event.getPath(), object));
} catch (Exception e) {
LOG.error("Exception processing curator result", e);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -169,25 +157,21 @@ private <T extends SingularityId> List<T> getChildrenAsIdsForParentsThrows(

final CountDownLatch latch = new CountDownLatch(parents.size());

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getChildren() == null || event.getChildren().size() == 0) {
LOG.trace("Expected children for node {} - but found none", event.getPath());
return;
}
synchronizedObjects.addAll(
Lists.transform(
event.getChildren(),
Transcoders.getFromStringFunction(idTranscoder)
)
);
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getChildren() == null || event.getChildren().size() == 0) {
LOG.trace("Expected children for node {} - but found none", event.getPath());
return;
}
synchronizedObjects.addAll(
event
.getChildren()
.stream()
.map(Transcoders.getFromStringFunction(idTranscoder))
.collect(Collectors.toList())
);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -218,10 +202,10 @@ protected <T extends SingularityId> List<T> getChildrenAsIds(
final String rootPath,
final IdTranscoder<T> idTranscoder
) {
return Lists.transform(
getChildren(rootPath),
Transcoders.getFromStringFunction(idTranscoder)
);
return getChildren(rootPath)
.stream()
.map(Transcoders.getFromStringFunction(idTranscoder))
.collect(Collectors.toList());
}

private <T extends SingularityId> List<T> existsThrows(
Expand All @@ -238,22 +222,17 @@ private <T extends SingularityId> List<T> existsThrows(

final CountDownLatch latch = new CountDownLatch(paths.size());

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getStat() != null) {
objects.add(
Transcoders
.getFromStringFunction(idTranscoder)
.apply(ZKPaths.getNodeFromPath(event.getPath()))
);
}
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getStat() != null) {
objects.add(
Transcoders
.getFromStringFunction(idTranscoder)
.apply(ZKPaths.getNodeFromPath(event.getPath()))
);
}
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -293,18 +272,13 @@ private <T extends SingularityId> List<T> notExistsThrows(

final CountDownLatch latch = new CountDownLatch(pathsMap.size());

final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getStat() == null) {
objects.add(pathsMap.get(event.getPath()));
}
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getStat() == null) {
objects.add(pathsMap.get(event.getPath()));
}
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -352,25 +326,7 @@ protected <T> List<T> getAsync(
) {
try {
return new ArrayList<>(
getAsyncThrows(pathNameForLogs, paths, transcoder, Optional.<ZkCache<T>>empty())
.values()
);
} catch (Throwable t) {
throw new RuntimeException(t);
}
}

protected <T> Map<String, T> getAsyncWithPath(
final String pathNameForLogs,
final Collection<String> paths,
final Transcoder<T> transcoder
) {
try {
return getAsyncThrows(
pathNameForLogs,
paths,
transcoder,
Optional.<ZkCache<T>>empty()
getAsyncThrows(pathNameForLogs, paths, transcoder, Optional.empty()).values()
);
} catch (Throwable t) {
throw new RuntimeException(t);
Expand Down Expand Up @@ -404,24 +360,19 @@ protected <T> List<T> getAsyncNestedChildrenAsListThrows(
final List<T> results = new ArrayList<>();
final CountDownLatch latch = new CountDownLatch(allPaths.size());
final AtomicInteger bytes = new AtomicInteger();
final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final T object = transcoder.fromBytes(event.getData());

results.add(object);
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final T object = transcoder.fromBytes(event.getData());

results.add(object);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -468,27 +419,22 @@ protected <T, Q> Map<T, List<Q>> getAsyncNestedChildDataAsMapThrows(
final ConcurrentHashMap<T, List<Q>> resultsMap = new ConcurrentHashMap<>();
final CountDownLatch latch = new CountDownLatch(allPathsMap.size());
final AtomicInteger bytes = new AtomicInteger();
final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final Q object = transcoder.fromBytes(event.getData());

if (allPathsMap.get(event.getPath()) != null) {
resultsMap.putIfAbsent(allPathsMap.get(event.getPath()), new ArrayList<Q>());
resultsMap.get(allPathsMap.get(event.getPath())).add(object);
}
} finally {
latch.countDown();
final BackgroundCallback callback = (client, event) -> {
try {
if (event.getData() == null || event.getData().length == 0) {
LOG.trace("Expected active node {} but it wasn't there", event.getPath());
return;
}
bytes.getAndAdd(event.getData().length);

final Q object = transcoder.fromBytes(event.getData());

if (allPathsMap.get(event.getPath()) != null) {
resultsMap.putIfAbsent(allPathsMap.get(event.getPath()), new ArrayList<>());
resultsMap.get(allPathsMap.get(event.getPath())).add(object);
}
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -535,24 +481,19 @@ protected <T extends SingularityId> List<T> getAsyncNestedChildIdsAsListThrows(
final List<T> results = new ArrayList<>();
final CountDownLatch latch = new CountDownLatch(allPaths.size());
final AtomicInteger bytes = new AtomicInteger();
final BackgroundCallback callback = new BackgroundCallback() {

@Override
public void processResult(CuratorFramework client, CuratorEvent event)
throws Exception {
try {
event
.getChildren()
.forEach(
child -> {
final T object = transcoder.fromString(child);
bytes.getAndAdd(child.getBytes().length);
results.add(object);
}
);
} finally {
latch.countDown();
}
final BackgroundCallback callback = (client, event) -> {
try {
event
.getChildren()
.forEach(
child -> {
final T object = transcoder.fromString(child);
bytes.getAndAdd(child.getBytes().length);
results.add(object);
}
);
} finally {
latch.countDown();
}
};

Expand Down Expand Up @@ -612,7 +553,7 @@ private <T> T queryAndReturnResultsThrows(
log(
method.operationType,
Optional.of(paths.size()),
bytes.get() > 0 ? Optional.of(bytes.get()) : Optional.<Integer>empty(),
bytes.get() > 0 ? Optional.of(bytes.get()) : Optional.empty(),
start,
pathNameForLogs
);
Expand Down
Loading