Skip to content

Log scheduler stats for Pulsar Functions #7474

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
jerrypeng merged 2 commits into from
Jul 8, 2020
Merged

Log scheduler stats for Pulsar Functions #7474

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2e385f2
Log scheduler stats for Pulsar Functions
Jul 8, 2020
f546598
addressing comments
Jul 8, 2020
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
1 change: 0 additions & 1 deletion ...tions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -813,7 +813,6 @@ private void addAssignment(Assignment assignment) {
}

private void startFunctionInstance(Assignment assignment) {
log.info("infos: {}", functionRuntimeInfos.getAll());
String fullyQualifiedInstanceId = FunctionCommon.getFullyQualifiedInstanceId(assignment.getInstance());
FunctionRuntimeInfo functionRuntimeInfo = _getFunctionRuntimeInfo(fullyQualifiedInstanceId);

Expand Down
10 changes: 9 additions & 1 deletion ...-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MembershipManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,7 @@ public void checkFailures(FunctionMetaDataManager functionMetaDataManager,
// check unassigned
Collection<Function.Instance> needSchedule = new LinkedList<>();
Collection<Function.Assignment> needRemove = new LinkedList<>();
Map<String, Integer> numRemoved = new HashMap<>();
for (Map.Entry<Function.Instance, Long> entry : this.unsignedFunctionDurations.entrySet()) {
Function.Instance instance = entry.getKey();
long unassignedDurationMs = entry.getValue();
Expand All @@ -217,6 +218,12 @@ public void checkFailures(FunctionMetaDataManager functionMetaDataManager,
Function.Assignment assignment = assignmentMap.get(FunctionCommon.getFullyQualifiedInstanceId(instance));
if (assignment != null) {
needRemove.add(assignment);

Integer count = numRemoved.get(assignment.getWorkerId());
if (count == null) {
count = 0;
}
numRemoved.put(assignment.getWorkerId(), count + 1);
}
triggerScheduler = true;
}
Expand All @@ -225,7 +232,8 @@ public void checkFailures(FunctionMetaDataManager functionMetaDataManager,
functionRuntimeManager.removeAssignments(needRemove);
}
if (triggerScheduler) {
log.info("Functions that need scheduling/rescheduling: {}", needSchedule);
log.info("Failure check - Total number of instances that need to be scheduled/rescheduled: {} | Number of unassigned instances that need to be scheduled: {} | Number of instances on dead workers that need to be reassigned {}",

@srkukarni srkukarni Jul 8, 2020

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.

Maybe a better wording here?

@jerrypeng jerrypeng Jul 8, 2020

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.

Do you have a suggestion?

@srkukarni srkukarni Jul 8, 2020

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.

I actually liked the old message. What was the reason for the change?

@jerrypeng jerrypeng Jul 8, 2020

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.

Added "Failure check" in the beginning so it will be easier to search for

needSchedule.size(), needSchedule.size() - needRemove.size(), numRemoved);
schedulerManager.schedule();
}
}
Expand Down
146 changes: 143 additions & 3 deletions ...r-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,14 @@
*/
package org.apache.pulsar.functions.worker;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.Builder;
import lombok.Data;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
Expand All @@ -34,6 +38,7 @@
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.common.util.Reflections;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.Assignment;
Expand Down Expand Up @@ -239,7 +244,8 @@ public Future<?> rebalanceIfNotInprogress() {

@VisibleForTesting
void invokeScheduler() {

long startTime = System.nanoTime();

Set<String> currentMembership = membershipManager.getCurrentMembership()
.stream().map(workerInfo -> workerInfo.getWorkerId()).collect(Collectors.toSet());

Expand All @@ -248,11 +254,13 @@ void invokeScheduler() {
Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager
.getCurrentAssignments();

// initialize stats collection
SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, currentMembership);

//delete assignments of functions and instances that don't exist anymore
Iterator<Map.Entry<String, Map<String, Assignment>>> it = workerIdToAssignments.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<String, Map<String, Assignment>> workerIdToAssignmentEntry = it.next();
String workerId = workerIdToAssignmentEntry.getKey();
Map<String, Assignment> functionMap = workerIdToAssignmentEntry.getValue();

// remove instances that don't exist anymore
Expand All @@ -268,6 +276,8 @@ void invokeScheduler() {
functionRuntimeManager.deleteAssignment(fullyQualifiedInstanceId);
// update message id associated with current view of assignments map
lastMessageProduced = messageId;
// update stats
schedulerStats.removedAssignment(assignment);
}
return deleted;
});
Expand All @@ -288,6 +298,8 @@ void invokeScheduler() {
functionRuntimeManager.processAssignment(newAssignment);
// update message id associated with current view of assignments map
lastMessageProduced = messageId;
//update stats
schedulerStats.updatedAssignment(newAssignment);
}
if (functionMap.isEmpty()) {
it.remove();
Expand Down Expand Up @@ -331,16 +343,26 @@ void invokeScheduler() {
functionRuntimeManager.processAssignment(assignment);
// update message id associated with current view of assignments map
lastMessageProduced = messageId;
// update stats
schedulerStats.newAssignment(assignment);
}

log.info("Schedule summary - execution time: {} sec | total unassigned: {} | stats: {}\n{}",
(System.nanoTime() - startTime) / Math.pow(10, 9),
unassignedInstances.getLeft().size(), schedulerStats.getSummary(), schedulerStats);
}

private void invokeRebalance() {
long startTime = System.nanoTime();

Set<String> currentMembership = membershipManager.getCurrentMembership()
.stream().map(workerInfo -> workerInfo.getWorkerId()).collect(Collectors.toSet());

Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();

// initialize stats collection
SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, currentMembership);

// filter out assignments of workers that are not currently in the active membership
List<Assignment> currentAssignments = workerIdToAssignments
.entrySet()
Expand Down Expand Up @@ -368,8 +390,13 @@ private void invokeRebalance() {
functionRuntimeManager.processAssignment(assignment);
// update message id associated with current view of assignments map
lastMessageProduced = messageId;
// update stats
schedulerStats.newAssignment(assignment);
}
log.info("Rebalance - Total number of new assignments computed: {}", rebalancedAssignments.size());

log.info("Rebalance summary - execution time: {} sec | stats: {}\n{}",

@merlimat merlimat Jul 8, 2020

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.

Can we also report this stats to prometheus?

@jerrypeng jerrypeng Jul 8, 2020

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.

Yes I was going to work on that in the near future. Stats like scheduler execution time can be in prometheus. However, a breakdown of what how many instances moved grouped by worker is not suitable to be put into prometheus

(System.nanoTime() - startTime) / Math.pow(10, 9), schedulerStats.getSummary(), schedulerStats);

rebalanceInProgess.set(false);
}

Expand Down Expand Up @@ -526,4 +553,117 @@ static String checkHeartBeatFunction(Instance funInstance) {

public static class RebalanceInProgressException extends RuntimeException {
}

private static class SchedulerStats {

@Builder
@Data
private static class WorkerStats {
private int originalNumAssignments;
private int finalNumAssignments;
private int instancesAdded;
private int instancesRemoved;
private int instancesUpdated;
private boolean alive;
}

private Map<String, WorkerStats> workerStatsMap = new HashMap<>();

private Map<String, String> instanceToWorkerId = new HashMap<>();

public SchedulerStats(Map<String, Map<String, Assignment>> workerIdToAssignments, Set<String> workers) {

for(String workerId : workers) {
WorkerStats.WorkerStatsBuilder workerStats = WorkerStats.builder().alive(true);
Map<String, Assignment> assignmentMap = workerIdToAssignments.get(workerId);
if (assignmentMap != null) {
workerStats.originalNumAssignments(assignmentMap.size());
workerStats.finalNumAssignments(assignmentMap.size());

for (String fullyQualifiedInstanceId : assignmentMap.keySet()) {
instanceToWorkerId.put(fullyQualifiedInstanceId, workerId);
}
} else {
workerStats.originalNumAssignments(0);
workerStats.finalNumAssignments(0);
}

workerStatsMap.put(workerId, workerStats.build());
}

// workers with assignments that are dead
for (Map.Entry<String, Map<String, Assignment>> entry : workerIdToAssignments.entrySet()) {
String workerId = entry.getKey();
Map<String, Assignment> assignmentMap = entry.getValue();
if (!workers.contains(workerId)) {
WorkerStats workerStats = WorkerStats.builder()
.alive(false)
.originalNumAssignments(assignmentMap.size())
.finalNumAssignments(assignmentMap.size())
.build();
workerStatsMap.put(workerId, workerStats);
}
}
}

public void removedAssignment(Assignment assignment) {
String workerId = assignment.getWorkerId();
WorkerStats stats = workerStatsMap.get(workerId);
Preconditions.checkNotNull(stats);

stats.instancesRemoved++;
stats.finalNumAssignments--;
}

public void newAssignment(Assignment assignment) {
String fullyQualifiedInstanceId = FunctionCommon.getFullyQualifiedInstanceId(assignment.getInstance());
String newWorkerId = assignment.getWorkerId();
String oldWorkerId = instanceToWorkerId.get(fullyQualifiedInstanceId);
if (oldWorkerId != null) {
WorkerStats oldWorkerStats = workerStatsMap.get(oldWorkerId);
Preconditions.checkNotNull(oldWorkerStats);

oldWorkerStats.instancesRemoved++;
oldWorkerStats.finalNumAssignments--;
}

WorkerStats newWorkerStats = workerStatsMap.get(newWorkerId);
Preconditions.checkNotNull(newWorkerStats);

newWorkerStats.instancesAdded++;
newWorkerStats.finalNumAssignments++;
}

public void updatedAssignment(Assignment assignment) {
String workerId = assignment.getWorkerId();
WorkerStats stats = workerStatsMap.get(workerId);
Preconditions.checkNotNull(stats);

stats.instancesUpdated++;
}

public String getSummary() {
int totalAdded = 0;
int totalUpdated = 0;
int totalRemoved = 0;

for (Map.Entry<String, WorkerStats> entry : workerStatsMap.entrySet()) {
WorkerStats workerStats = entry.getValue();
totalAdded += workerStats.instancesAdded;
totalUpdated += workerStats.instancesUpdated;
totalRemoved += workerStats.instancesRemoved;
}

return String.format("{\"Added\": %d, \"Updated\": %d, \"removed\": %d}", totalAdded, totalUpdated, totalRemoved);
}

@Override
public String toString() {
try {
return ObjectMapperFactory.getThreadLocal().writerWithDefaultPrettyPrinter().writeValueAsString(workerStatsMap);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
}
}
Loading