Skip to content

[GOBBLIN-1823] Improving Container Calculation and Allocation Methodology #3692

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
ZihanLi58 merged 7 commits into from
May 22, 2023
Merged

[GOBBLIN-1823] Improving Container Calculation and Allocation Methodology #3692

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6880626
address comments
Mar 25, 2022
065f7d9
use connectionmanager when httpclient is not cloesable
Mar 25, 2022
bfb45d3
[GOBBLIN-1823] Improving Container Calculation and Allocation Methodo…
Apr 21, 2023
6c09c43
improve code style
May 4, 2023
748e55f
add more un-retriable status that we want to log out
May 5, 2023
af29927
address comments
May 11, 2023
915f450
add log when mismatch happens for debuggability
May 19, 2023
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
6 changes: 3 additions & 3 deletions gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnAutoScalingManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ public class YarnAutoScalingManager extends AbstractIdleService {

private final String AUTO_SCALING_WINDOW_SIZE = AUTO_SCALING_PREFIX + "windowSize";

private final static int DEFAULT_MAX_IDLE_TIME_BEFORE_SCALING_DOWN_MINUTES = 10;
public final static int DEFAULT_MAX_CONTAINER_IDLE_TIME_BEFORE_SCALING_DOWN_MINUTES = 10;

private final Config config;
private final HelixManager helixManager;
Expand All @@ -97,9 +97,9 @@ public class YarnAutoScalingManager extends AbstractIdleService {
private final int partitionsPerContainer;
private final double overProvisionFactor;
private final SlidingWindowReservoir slidingFixedSizeWindow;
private static int maxIdleTimeInMinutesBeforeScalingDown = DEFAULT_MAX_IDLE_TIME_BEFORE_SCALING_DOWN_MINUTES;
private static int maxIdleTimeInMinutesBeforeScalingDown = DEFAULT_MAX_CONTAINER_IDLE_TIME_BEFORE_SCALING_DOWN_MINUTES;
private static final HashSet<TaskPartitionState>
UNUSUAL_HELIX_TASK_STATES = Sets.newHashSet(TaskPartitionState.ERROR, TaskPartitionState.DROPPED);
UNUSUAL_HELIX_TASK_STATES = Sets.newHashSet(TaskPartitionState.ERROR, TaskPartitionState.DROPPED, TaskPartitionState.COMPLETED, TaskPartitionState.TIMED_OUT);

public YarnAutoScalingManager(GobblinApplicationMaster appMaster) {
this.config = appMaster.getConfig();
Expand Down
46 changes: 43 additions & 3 deletions gobblin-yarn/src/main/java/org/apache/gobblin/yarn/YarnService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,8 @@ public class YarnService extends AbstractIdleService {
private final boolean isPurgingOfflineHelixInstancesEnabled;
private final long helixPurgeLaggingThresholdMs;
private final long helixPurgeStatusPollingRateMs;
private final ConcurrentMap<ContainerId, Long> containerIdleSince = Maps.newConcurrentMap();
private final ConcurrentMap<ContainerId, String> removedContainerID = Maps.newConcurrentMap();

private volatile YarnContainerRequestBundle yarnContainerRequest;
private final AtomicInteger priorityNumGenerator = new AtomicInteger(0);
Expand Down Expand Up @@ -473,6 +475,17 @@ public synchronized boolean requestTargetNumberOfContainers(YarnContainerRequest
return false;
}

//Correct the containerMap first as there is cases that handleContainerCompletion() is called before onContainersAllocated()
for (ContainerId removedId :this.removedContainerID.keySet()) {

@homatthew homatthew May 8, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

nit: whitespace after :

ContainerInfo containerInfo = this.containerMap.remove(removedId);
if (containerInfo != null) {
String helixTag = containerInfo.getHelixTag();
allocatedContainerCountMap.putIfAbsent(helixTag, new AtomicInteger(0));
this.allocatedContainerCountMap.get(helixTag).decrementAndGet();

@homatthew homatthew May 8, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

put if absent 0 and then decrementing means the resulting value would be -1. That does not seem correct to me

@ZihanLi58 ZihanLi58 May 10, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I don't think we will put it though because in this case, onContainerAllocated should have already been called and we definitely have one entry in the map. The worst case is if we call onContainerAllocated and this method concurrently, then we might end up decreasing the value and then increasing it immediately.

@homatthew homatthew May 19, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

When Yarn allocates "ghost containers" without calling the onContainerAllocated() method and when the container is eventually released, onContainersCompleted() is called, container numbers mismatches can occur.
In the onContainerAllocated() method, we add the container to the containerMap using the container ID as the key, and increase the count for the specific tag.

This from the description makes it sound like it is not guaranteed to be called. Either way, I think if we do need to put this line, we should put if absent 1 and then decrement

this.removedContainerID.remove(removedId);
}
}

int numTargetContainers = yarnContainerRequestBundle.getTotalContainers();
// YARN can allocate more than the requested number of containers, compute additional allocations and deallocations
// based on the max of the requested and actual allocated counts
Expand Down Expand Up @@ -501,12 +514,30 @@ public synchronized boolean requestTargetNumberOfContainers(YarnContainerRequest
}
}

//We go through all the containers we have now and check whether the assigned participant is still alive, if not, we should put them in idle container Map

@homatthew homatthew May 8, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

"Check whether assigned participant is still alive". There is nothing here to suggest that these instances are actually "assigned" anything.

I think a more accurate comment is instead something like:

iterate through all containers allocated and check whether the corresponding helix instance is still LIVE within the helix cluster. A container that has a bad connection to zookeeper will be dropped from the Helix cluster if the disconnection is greater than the specified timeout. In these cases, we want to release the container to get a new container because these containers won't be assigned tasks by Helix

//And we will release the container if the assigned participant still offline after a given time

List<Container> containersToRelease = new ArrayList<>();
for (Map.Entry<ContainerId, ContainerInfo> entry : this.containerMap.entrySet()) {
ContainerInfo containerInfo = entry.getValue();
if (!HelixUtils.isInstanceLive(helixManager, containerInfo.getHelixParticipantId())) {
containerIdleSince.putIfAbsent(entry.getKey(), System.currentTimeMillis());
if (System.currentTimeMillis() - containerIdleSince.get(entry.getKey())
>= TimeUnit.MINUTES.toMillis(YarnAutoScalingManager.DEFAULT_MAX_CONTAINER_IDLE_TIME_BEFORE_SCALING_DOWN_MINUTES)) {
LOGGER.info("Releasing Container {} because the assigned participant {} has been in-active for more than {} minutes",
entry.getKey(), containerInfo.getHelixParticipantId(), YarnAutoScalingManager.DEFAULT_MAX_CONTAINER_IDLE_TIME_BEFORE_SCALING_DOWN_MINUTES);
containersToRelease.add(containerInfo.getContainer());
}
} else {
containerIdleSince.remove(entry.getKey());
}
}

// If the total desired is lower than the currently allocated amount then release free containers.
// This is based on the currently allocated amount since containers may still be in the process of being allocated
// and assigned work. Resizing based on numRequestedContainers at this point may release a container right before
// or soon after it is assigned work.
if (numTargetContainers < totalAllocatedContainers) {
List<Container> containersToRelease = new ArrayList<>();
if (containersToRelease.isEmpty() && numTargetContainers < totalAllocatedContainers) {

@homatthew homatthew May 8, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Why do we entirely skip this block if there are already containers to release? Hopefully the previous block doesn't happen that often, but this if statement is still a bit strange to read.

To me, this should instead be:

if (numTargetContainers < totalAllocatedContainers - containersToRelease.size())

@ZihanLi58 ZihanLi58 May 10, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Because at this point we still hold reference to those bad containers, and we might end up with releasing those containers again in this block. We can consider changing the algorism to have a hash set of containerIDtoRelease and collect the ContainerIdToRelease first and at the end of the call calculate the containerToRelease.

int numToShutdown = totalAllocatedContainers - numTargetContainers;

LOGGER.info("Shrinking number of containers by {} because numTargetContainers < totalAllocatedContainers ({} < {})",
Expand All @@ -525,7 +556,9 @@ public synchronized boolean requestTargetNumberOfContainers(YarnContainerRequest
}

LOGGER.info("Shutting down {} containers. containersToRelease={}", containersToRelease.size(), containersToRelease);
}

if (!containersToRelease.isEmpty()) {
this.eventBus.post(new ContainerReleaseRequest(containersToRelease));
}
this.yarnContainerRequest = yarnContainerRequestBundle;
Expand Down Expand Up @@ -721,9 +754,16 @@ protected void handleContainerCompletion(ContainerStatus containerStatus) {
//Get the Helix instance name for the completed container. Because callbacks are processed asynchronously, we might
//encounter situations where handleContainerCompletion() is called before onContainersAllocated(), resulting in the
//containerId missing from the containersMap.
// We use removedContainerID to remember these containers and remove them from containerMap later when we call requestTargetNumberOfContainers method
if (completedContainerInfo == null) {
removedContainerID.putIfAbsent(containerStatus.getContainerId(), "");
}
String completedInstanceName = completedContainerInfo == null? UNKNOWN_HELIX_INSTANCE : completedContainerInfo.getHelixParticipantId();

String helixTag = completedContainerInfo == null ? helixInstanceTags : completedContainerInfo.getHelixTag();
allocatedContainerCountMap.get(helixTag).decrementAndGet();
if (completedContainerInfo != null) {
allocatedContainerCountMap.get(helixTag).decrementAndGet();
}

LOGGER.info(String.format("Container %s running Helix instance %s with tag %s has completed with exit status %d",
containerStatus.getContainerId(), completedInstanceName, helixTag, containerStatus.getExitStatus()));
Expand Down