You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/09/24 18:48:22 UTC
[GitHub] [pulsar] kaushik-develop opened a new pull request #12178: Support for draining workers
kaushik-develop opened a new pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178
<!--
### Contribution Checklist
- Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
Skip *Issue XYZ* if there is no associated github issue for this pull request.
Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
*(If this PR fixes a github issue, please add `Fixes #<xyz>`.)*
Fixes #<xyz>
*(or if this PR is one task of a github issue, please add `Master Issue: #<xyz>` to link to the master issue.)*
Master Issue: #<xyz>
### Motivation
Support for draining function workers, to scale in a cluster.
*Explain here the context, and why you're making that change. What is the problem you're trying to solve.*
### Modifications
*Describe the modifications you've done.*
- added new endpoints to drain function assignments off a worker
- added new endpoints to get the status of a drain operation
### Verifying this change
- [X] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
- Added the following to SchedulerManagerTest: testDrain, testGetDrainStatus, testDrainExceptions, testUpdateWorkerDrainMap
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (yes)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
- The rest endpoints: (yes)
- The admin cli options: (no)
- Anything that affects deployment: (no)
### Documentation
Check the box below and label this PR (if you have committer privilege).
Need to update docs?
- [ ] doc-required
(If you need help on updating docs, create a doc issue)
- [ ] no-need-doc
(Please explain why)
- [ ] doc
(If this PR contains doc changes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716938186
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -215,28 +239,147 @@ public synchronized void initialize(Producer<byte[]> exclusiveProducer) {
}
public Future<?> rebalanceIfNotInprogress() {
- if (rebalanceInProgess.compareAndSet(false, true)) {
+ if (rebalanceInProgress.compareAndSet(false, true)) {
+ val numWorkers = getCurrentAvailableNumWorkers();
+ if (numWorkers <= 1) {
+ rebalanceInProgress.set(false);
+ throw new TooFewWorkersException();
+ }
currentRebalanceFuture = rebalance();
return currentRebalanceFuture;
} else {
throw new RebalanceInProgressException();
}
}
- @VisibleForTesting
- void invokeScheduler() {
+ private Future<?> drain(String workerId) {
+ return scheduleInternal(() -> {
+ workerStatsManager.drainTotalExecTimeStart();
+ currentPostDrainAssignments = invokeDrain(workerId);
+ workerStatsManager.drainTotalExecTimeEnd();
+ }, "Encountered error when invoking drain");
+ }
+
+ public Future<?> drainIfNotInProgress(String workerId) {
+ if (drainInProgressFlag.compareAndSet(false, true)) {
+ try {
+ val availableWorkers = getCurrentAvailableWorkers();
+ if (availableWorkers.size() <= 1) {
+ throw new TooFewWorkersException();
+ }
+
+ // A worker must be specified at this point. This would be set up by the caller.
+ Preconditions.checkNotNull(workerId);
+
+ // [We can get stricter, and require that every drain op be followed up with a cleanup of the
+ // corresponding worker before any other drain op, so that the drainOpStatusMap should be empty
+ // at the next drain operation.]
+ if (drainOpStatusMap.containsKey(workerId)) {
+ String warnString = "Worker " + workerId
+ + " was not removed yet from SchedulerManager after previous drain op";
+ log.warn(warnString);
+ throw new WorkerNotRemovedAfterPriorDrainException();
+ }
+
+ if (!availableWorkers.contains(workerId)) {
+ log.info("invokeDrain was called for a worker={} which is not currently active", workerId);
+ throw new UnknownWorkerException();
+ }
+
+ currentDrainFuture = drain(workerId);
+ return currentDrainFuture;
+ } finally {
+ drainInProgressFlag.set(false);
+ }
+ } else {
+ throw new DrainInProgressException();
+ }
+ }
+
+ public LongRunningProcessStatus getDrainStatus(String workerId) {
long startTime = System.nanoTime();
+ String errString;
+ LongRunningProcessStatus retVal = new LongRunningProcessStatus();
+ try {
+ val workerStatus = drainOpStatusMap.get(workerId);
+ if (workerStatus == null) {
+ errString = "Worker " + workerId + " not found in drain records";
+ retVal = LongRunningProcessStatus.forError(errString);
+ } else {
+ switch (workerStatus) {
+ default:
+ errString = "getDrainStatus: Unexpected status " + workerStatus + " found for worker " + workerId;
+ retVal = LongRunningProcessStatus.forError(errString);
+ break;
+ case DrainCompleted:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.SUCCESS);
+ break;
+ case DrainInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.RUNNING);
+ break;
+ case DrainNotInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.NOT_RUN);
+ break;
+ }
+ }
+ } finally {
+ log.info("Get drain status for worker {} - execution time: {} sec; returning status={}, error={}",
+ workerId, (System.nanoTime() - startTime) / Math.pow(10, 9), retVal.status, retVal.lastError);
+ return retVal;
+ }
+ }
+
+ @VisibleForTesting
+ void clearDrainOpsStatus() {
+ drainOpStatusMap.clear();
+ log.warn("Cleared drain op status map");
+ }
+
+ @VisibleForTesting
+ void setDrainOpsStatus(final String workerId, final DrainOpStatus dStatus) {
+ drainOpStatusMap.put(workerId, dStatus);
+ log.warn("setDrainOpsStatus: updated drain status of worker {} to {}", workerId, dStatus);
+ }
+ @VisibleForTesting
+ ConcurrentHashMap<String, DrainOpStatus> getDrainOpsStatusMap() {
+ val retVal = new ConcurrentHashMap<String, DrainOpStatus>(drainOpStatusMap);
+ return retVal;
+ }
+
+ private synchronized int getCurrentAvailableNumWorkers() {
+ return getCurrentAvailableWorkers().size();
+ }
+
+ private synchronized Set <String> getCurrentAvailableWorkers() {
Set<String> currentMembership = membershipManager.getCurrentMembership()
.stream().map(workerInfo -> workerInfo.getWorkerId()).collect(Collectors.toSet());
+ val unavailableWorkers = new ArrayList<String>();
+ // iterate the set, instead of the concurrent hashmap
+ for (String worker : currentMembership) {
Review comment:
We don't need two loops to remove. You can use an iterator to remove with iterating the set.
https://mkyong.com/java/java-how-to-remove-items-from-a-list-while-iterating/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r718020310
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Worker.java
##########
@@ -127,6 +129,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
Addressed in the next iteration.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
Review comment:
Addressed in the next iteration.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public LongRunningProcessStatus getDrainStatus(@PathParam("drain") String workerId) {
+ return workers().getDrainStatus(uri.getRequestUri(), clientAppId(), true);
Review comment:
Addressed in the next iteration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719759074
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
Review comment:
@kaushik-develop did you address this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719741667
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
Review comment:
What if the worker is restarting during when this happens? The worker can temporarily have disappeared from the the membership. Shouldn't we proceed to removing the assignments from the worker and still mark it as in progress of being drained.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r717748796
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I didn't quite understand the question: "Is there a reason we should synchronize this with drain operation?"
updateWorkerDrainMap() does a periodic cleanup of stale information about drained workers. We need to do the operation some time after the drain has finished, when the drained worker is removed from the cluster. Since there is currently no hook (that I know of) into the SchedulerManager when a worker is added to, or removed from the cluster, the cleanup is done through a periodic poll (updateWorkerDrainMap).
The drain operation adds a record into the concurrent map [drainOpStatusMap] when a worker is drained. An implicit assumption in the system is that the drained worker will be removed from the system, soon, by an external orchestrator. When the drained worker is seen to be removed from the system, the drainOpStatusMap is cleaned up of stale information in the updateWorkerDrainMap() code.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I didn't quite understand the question: "Is there a reason we should synchronize this with drain operation?"
updateWorkerDrainMap() does a periodic cleanup of stale information about drained workers. We need to do the operation some time after the drain has finished, when the drained worker is removed from the cluster. Since there is currently no hook (that I know of) into the SchedulerManager when a worker is added to, or removed from the cluster, the cleanup is done through a periodic poll (updateWorkerDrainMap).
The drain operation adds a record into the concurrent map [drainOpStatusMap] when a worker is drained. An implicit assumption in the system is that the drained worker will be removed from the system, soon, by an external orchestrator. When the drained worker is seen to be removed from the system, the drainOpStatusMap is cleaned up of stale information in the updateWorkerDrainMap() code.
PLMK if I misunderstood the qs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Anonymitaet commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
Anonymitaet commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716156705
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -71,7 +75,7 @@
* The scheduler manager is used to compute scheduling of function instances
* Only the leader computes new schedulings and writes assignments to the assignment topic
* The lifecyle of this class is the following:
- * 1. When worker becomes leader, this class with me initialized
+ * 1. When worker becomes leader, this class with be initialized
Review comment:
```suggestion
* 1. When a worker becomes a leader, this class is initialized
```
Do you intend to mean this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng merged pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng merged pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719764714
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
Review comment:
Sounds fair; I will remove the precondition check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r718020257
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
Addressed in the next iteration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716906045
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
I think a better name for the URL parameter is "workerId".
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public LongRunningProcessStatus getDrainStatus(@PathParam("drain") String workerId) {
+ return workers().getDrainStatus(uri.getRequestUri(), clientAppId(), true);
Review comment:
Why are you passing the URI into "getDrainStatus()" and later trying to parse the workerId from the URI instead of of just passing in the workerId directly to "getDrainStatus()"?
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
Review comment:
I think a better name for the URL parameter is "workerId".
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Worker.java
##########
@@ -127,6 +129,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
Please refer to my comments in WorkerApiV2Resource.java
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
Review comment:
Since this gets run periodically, lets make sure method cannot somehow get invoked concurrently.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I know drainOpStatusMap is a concurrent map but we are reading and writing from different different threads. This makes me nervous.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I know drainOpStatusMap is a concurrent map but we are reading and writing from different different threads. This makes me nervous. Is there a reason we should synchronize this with drain operation?
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -215,28 +239,147 @@ public synchronized void initialize(Producer<byte[]> exclusiveProducer) {
}
public Future<?> rebalanceIfNotInprogress() {
- if (rebalanceInProgess.compareAndSet(false, true)) {
+ if (rebalanceInProgress.compareAndSet(false, true)) {
+ val numWorkers = getCurrentAvailableNumWorkers();
+ if (numWorkers <= 1) {
+ rebalanceInProgress.set(false);
+ throw new TooFewWorkersException();
+ }
currentRebalanceFuture = rebalance();
return currentRebalanceFuture;
} else {
throw new RebalanceInProgressException();
}
}
- @VisibleForTesting
- void invokeScheduler() {
+ private Future<?> drain(String workerId) {
+ return scheduleInternal(() -> {
+ workerStatsManager.drainTotalExecTimeStart();
+ currentPostDrainAssignments = invokeDrain(workerId);
+ workerStatsManager.drainTotalExecTimeEnd();
+ }, "Encountered error when invoking drain");
+ }
+
+ public Future<?> drainIfNotInProgress(String workerId) {
+ if (drainInProgressFlag.compareAndSet(false, true)) {
+ try {
+ val availableWorkers = getCurrentAvailableWorkers();
+ if (availableWorkers.size() <= 1) {
+ throw new TooFewWorkersException();
+ }
+
+ // A worker must be specified at this point. This would be set up by the caller.
+ Preconditions.checkNotNull(workerId);
+
+ // [We can get stricter, and require that every drain op be followed up with a cleanup of the
+ // corresponding worker before any other drain op, so that the drainOpStatusMap should be empty
+ // at the next drain operation.]
+ if (drainOpStatusMap.containsKey(workerId)) {
+ String warnString = "Worker " + workerId
+ + " was not removed yet from SchedulerManager after previous drain op";
+ log.warn(warnString);
+ throw new WorkerNotRemovedAfterPriorDrainException();
+ }
+
+ if (!availableWorkers.contains(workerId)) {
+ log.info("invokeDrain was called for a worker={} which is not currently active", workerId);
+ throw new UnknownWorkerException();
+ }
+
+ currentDrainFuture = drain(workerId);
+ return currentDrainFuture;
+ } finally {
+ drainInProgressFlag.set(false);
+ }
+ } else {
+ throw new DrainInProgressException();
+ }
+ }
+
+ public LongRunningProcessStatus getDrainStatus(String workerId) {
long startTime = System.nanoTime();
+ String errString;
+ LongRunningProcessStatus retVal = new LongRunningProcessStatus();
+ try {
+ val workerStatus = drainOpStatusMap.get(workerId);
+ if (workerStatus == null) {
+ errString = "Worker " + workerId + " not found in drain records";
+ retVal = LongRunningProcessStatus.forError(errString);
+ } else {
+ switch (workerStatus) {
+ default:
+ errString = "getDrainStatus: Unexpected status " + workerStatus + " found for worker " + workerId;
+ retVal = LongRunningProcessStatus.forError(errString);
+ break;
+ case DrainCompleted:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.SUCCESS);
+ break;
+ case DrainInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.RUNNING);
+ break;
+ case DrainNotInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.NOT_RUN);
+ break;
+ }
+ }
+ } finally {
+ log.info("Get drain status for worker {} - execution time: {} sec; returning status={}, error={}",
+ workerId, (System.nanoTime() - startTime) / Math.pow(10, 9), retVal.status, retVal.lastError);
+ return retVal;
+ }
+ }
+
+ @VisibleForTesting
+ void clearDrainOpsStatus() {
+ drainOpStatusMap.clear();
+ log.warn("Cleared drain op status map");
+ }
+
+ @VisibleForTesting
+ void setDrainOpsStatus(final String workerId, final DrainOpStatus dStatus) {
+ drainOpStatusMap.put(workerId, dStatus);
+ log.warn("setDrainOpsStatus: updated drain status of worker {} to {}", workerId, dStatus);
+ }
+ @VisibleForTesting
+ ConcurrentHashMap<String, DrainOpStatus> getDrainOpsStatusMap() {
+ val retVal = new ConcurrentHashMap<String, DrainOpStatus>(drainOpStatusMap);
+ return retVal;
+ }
+
+ private synchronized int getCurrentAvailableNumWorkers() {
+ return getCurrentAvailableWorkers().size();
+ }
+
+ private synchronized Set <String> getCurrentAvailableWorkers() {
Set<String> currentMembership = membershipManager.getCurrentMembership()
.stream().map(workerInfo -> workerInfo.getWorkerId()).collect(Collectors.toSet());
+ val unavailableWorkers = new ArrayList<String>();
+ // iterate the set, instead of the concurrent hashmap
+ for (String worker : currentMembership) {
Review comment:
We don't need two loops to remove. You can use an iterator to remove with iterating the set.
https://mkyong.com/java/java-how-to-remove-items-from-a-list-while-iterating/
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -215,28 +239,147 @@ public synchronized void initialize(Producer<byte[]> exclusiveProducer) {
}
public Future<?> rebalanceIfNotInprogress() {
- if (rebalanceInProgess.compareAndSet(false, true)) {
+ if (rebalanceInProgress.compareAndSet(false, true)) {
+ val numWorkers = getCurrentAvailableNumWorkers();
+ if (numWorkers <= 1) {
+ rebalanceInProgress.set(false);
+ throw new TooFewWorkersException();
+ }
currentRebalanceFuture = rebalance();
return currentRebalanceFuture;
} else {
throw new RebalanceInProgressException();
}
}
- @VisibleForTesting
- void invokeScheduler() {
+ private Future<?> drain(String workerId) {
+ return scheduleInternal(() -> {
+ workerStatsManager.drainTotalExecTimeStart();
+ currentPostDrainAssignments = invokeDrain(workerId);
+ workerStatsManager.drainTotalExecTimeEnd();
+ }, "Encountered error when invoking drain");
+ }
+
+ public Future<?> drainIfNotInProgress(String workerId) {
+ if (drainInProgressFlag.compareAndSet(false, true)) {
+ try {
+ val availableWorkers = getCurrentAvailableWorkers();
+ if (availableWorkers.size() <= 1) {
+ throw new TooFewWorkersException();
+ }
+
+ // A worker must be specified at this point. This would be set up by the caller.
+ Preconditions.checkNotNull(workerId);
+
+ // [We can get stricter, and require that every drain op be followed up with a cleanup of the
+ // corresponding worker before any other drain op, so that the drainOpStatusMap should be empty
+ // at the next drain operation.]
+ if (drainOpStatusMap.containsKey(workerId)) {
+ String warnString = "Worker " + workerId
+ + " was not removed yet from SchedulerManager after previous drain op";
+ log.warn(warnString);
+ throw new WorkerNotRemovedAfterPriorDrainException();
+ }
+
+ if (!availableWorkers.contains(workerId)) {
+ log.info("invokeDrain was called for a worker={} which is not currently active", workerId);
+ throw new UnknownWorkerException();
+ }
+
+ currentDrainFuture = drain(workerId);
+ return currentDrainFuture;
+ } finally {
+ drainInProgressFlag.set(false);
+ }
+ } else {
+ throw new DrainInProgressException();
+ }
+ }
+
+ public LongRunningProcessStatus getDrainStatus(String workerId) {
long startTime = System.nanoTime();
+ String errString;
+ LongRunningProcessStatus retVal = new LongRunningProcessStatus();
+ try {
+ val workerStatus = drainOpStatusMap.get(workerId);
+ if (workerStatus == null) {
+ errString = "Worker " + workerId + " not found in drain records";
+ retVal = LongRunningProcessStatus.forError(errString);
+ } else {
+ switch (workerStatus) {
+ default:
+ errString = "getDrainStatus: Unexpected status " + workerStatus + " found for worker " + workerId;
+ retVal = LongRunningProcessStatus.forError(errString);
+ break;
+ case DrainCompleted:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.SUCCESS);
+ break;
+ case DrainInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.RUNNING);
+ break;
+ case DrainNotInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.NOT_RUN);
+ break;
+ }
+ }
+ } finally {
+ log.info("Get drain status for worker {} - execution time: {} sec; returning status={}, error={}",
+ workerId, (System.nanoTime() - startTime) / Math.pow(10, 9), retVal.status, retVal.lastError);
+ return retVal;
+ }
+ }
+
+ @VisibleForTesting
Review comment:
Can you add a comment to "clearDrainOpsStatus", "setDrainOpsStatus", and "getDrainOpsStatusMap" that these methods are only used for testing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719765116
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
+
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainInProgress);
+
+ val workerRemovealSuccessful = availableWorkers.remove(workerId);
+ Preconditions.checkState(workerRemovealSuccessful);
+
+ List<FunctionMetaData> allFunctions = functionMetaDataManager.getAllFunctionMetaData();
+ Map<String, Function.Instance> allInstances =
+ computeAllInstances(allFunctions, functionRuntimeManager.getRuntimeFactory().externallyManaged());
+
+ // Pretend that current assignments is empty, and get the instances to assign.
+ Pair<List<Function.Instance>, List<Assignment>> instancesToAssign
+ = getUnassignedFunctionInstances(null, allInstances);
+
+ workerStatsManager.drainTotalExecTimeStart();
+ // Try to schedule the instances on the workers remaining available after "workerId" is removed.
+ try {
+ postDrainAssignments = scheduler.schedule(instancesToAssign.getLeft(), null, availableWorkers);
+ } catch (Exception e) {
+ log.info("invokeDrain: Got exception from schedule: ", e);
+ }
+ workerStatsManager.drainTotalExecTimeEnd();
+
+ if (postDrainAssignments != null) {
+ // log.info("drain: postDrainAssignments={}", postDrainAssignments);
Review comment:
ok.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
+
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainInProgress);
+
+ val workerRemovealSuccessful = availableWorkers.remove(workerId);
+ Preconditions.checkState(workerRemovealSuccessful);
+
+ List<FunctionMetaData> allFunctions = functionMetaDataManager.getAllFunctionMetaData();
+ Map<String, Function.Instance> allInstances =
+ computeAllInstances(allFunctions, functionRuntimeManager.getRuntimeFactory().externallyManaged());
+
+ // Pretend that current assignments is empty, and get the instances to assign.
+ Pair<List<Function.Instance>, List<Assignment>> instancesToAssign
+ = getUnassignedFunctionInstances(null, allInstances);
+
+ workerStatsManager.drainTotalExecTimeStart();
+ // Try to schedule the instances on the workers remaining available after "workerId" is removed.
+ try {
+ postDrainAssignments = scheduler.schedule(instancesToAssign.getLeft(), null, availableWorkers);
+ } catch (Exception e) {
+ log.info("invokeDrain: Got exception from schedule: ", e);
+ }
+ workerStatsManager.drainTotalExecTimeEnd();
+
+ if (postDrainAssignments != null) {
+ // log.info("drain: postDrainAssignments={}", postDrainAssignments);
+ for (Assignment assignment : postDrainAssignments) {
+ MessageId messageId = publishNewAssignment(assignment, false);
+ // Directly update in memory assignment cache since I am leader
+ // log.info("Drain worker {} - new assignment: {}", workerId, assignment);
Review comment:
ok.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#issuecomment-929834997
/pulsarbot run-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719766751
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
+
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainInProgress);
+
+ val workerRemovealSuccessful = availableWorkers.remove(workerId);
+ Preconditions.checkState(workerRemovealSuccessful);
+
+ List<FunctionMetaData> allFunctions = functionMetaDataManager.getAllFunctionMetaData();
+ Map<String, Function.Instance> allInstances =
+ computeAllInstances(allFunctions, functionRuntimeManager.getRuntimeFactory().externallyManaged());
+
+ // Pretend that current assignments is empty, and get the instances to assign.
+ Pair<List<Function.Instance>, List<Assignment>> instancesToAssign
+ = getUnassignedFunctionInstances(null, allInstances);
Review comment:
Sounds fair. I will fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716906045
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
I think a better name for the URL parameter is "workerId".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719761657
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
That is by design, and should be ok. PLMK if you foresee problems in any specific scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716930912
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I know drainOpStatusMap is a concurrent map but we are reading and writing from different different threads. This makes me nervous.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716907130
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public LongRunningProcessStatus getDrainStatus(@PathParam("drain") String workerId) {
+ return workers().getDrainStatus(uri.getRequestUri(), clientAppId(), true);
Review comment:
Why are you passing the URI into "getDrainStatus()" and later trying to parse the workerId from the URI instead of of just passing in the workerId directly to "getDrainStatus()"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#issuecomment-926939451
/pulsarbot run-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#issuecomment-932308387
/pulsarbot run-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719761657
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
That is by design, and should be ok.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716930123
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
Review comment:
Since this gets run periodically, lets make sure method cannot somehow get invoked concurrently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r717748796
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I didn't quite understand the question: "Is there a reason we should synchronize this with drain operation?"
updateWorkerDrainMap() does a periodic cleanup of stale information about drained workers. We need to do the operation some time after the drain has finished, when the drained worker is removed from the cluster. Since there is currently no hook (that I know of) into the SchedulerManager when a worker is added to, or removed from the cluster, the cleanup is done through a periodic poll (updateWorkerDrainMap).
The drain operation adds a record into the concurrent map [drainOpStatusMap] when a worker is drained. An implicit assumption in the system is that the drained worker will be removed from the system, soon, by an external orchestrator. When the drained worker is seen to be removed from the system, the drainOpStatusMap is cleaned up of stale information in the updateWorkerDrainMap() code.
PLMK if I misunderstood the qs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r718020257
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
Addressed in the next iteration.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Worker.java
##########
@@ -127,6 +129,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
Addressed in the next iteration.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
Review comment:
Addressed in the next iteration.
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
+ workers().drain(uri.getRequestUri(), clientAppId(), true);
+ }
+
+ @PUT
+ @ApiOperation(
+ value = "Drains this worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/drain")
+ public void drain() {
+ workers().drain(uri.getRequestUri(), clientAppId(), false);
+ }
+
+ @GET
+ @ApiOperation(
+ value = "Get the status of any ongoing drain operation at the specified worker",
+ response = LongRunningProcessStatus.class
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public LongRunningProcessStatus getDrainStatus(@PathParam("drain") String workerId) {
+ return workers().getDrainStatus(uri.getRequestUri(), clientAppId(), true);
Review comment:
Addressed in the next iteration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#issuecomment-929834997
/pulsarbot run-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716939382
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -215,28 +239,147 @@ public synchronized void initialize(Producer<byte[]> exclusiveProducer) {
}
public Future<?> rebalanceIfNotInprogress() {
- if (rebalanceInProgess.compareAndSet(false, true)) {
+ if (rebalanceInProgress.compareAndSet(false, true)) {
+ val numWorkers = getCurrentAvailableNumWorkers();
+ if (numWorkers <= 1) {
+ rebalanceInProgress.set(false);
+ throw new TooFewWorkersException();
+ }
currentRebalanceFuture = rebalance();
return currentRebalanceFuture;
} else {
throw new RebalanceInProgressException();
}
}
- @VisibleForTesting
- void invokeScheduler() {
+ private Future<?> drain(String workerId) {
+ return scheduleInternal(() -> {
+ workerStatsManager.drainTotalExecTimeStart();
+ currentPostDrainAssignments = invokeDrain(workerId);
+ workerStatsManager.drainTotalExecTimeEnd();
+ }, "Encountered error when invoking drain");
+ }
+
+ public Future<?> drainIfNotInProgress(String workerId) {
+ if (drainInProgressFlag.compareAndSet(false, true)) {
+ try {
+ val availableWorkers = getCurrentAvailableWorkers();
+ if (availableWorkers.size() <= 1) {
+ throw new TooFewWorkersException();
+ }
+
+ // A worker must be specified at this point. This would be set up by the caller.
+ Preconditions.checkNotNull(workerId);
+
+ // [We can get stricter, and require that every drain op be followed up with a cleanup of the
+ // corresponding worker before any other drain op, so that the drainOpStatusMap should be empty
+ // at the next drain operation.]
+ if (drainOpStatusMap.containsKey(workerId)) {
+ String warnString = "Worker " + workerId
+ + " was not removed yet from SchedulerManager after previous drain op";
+ log.warn(warnString);
+ throw new WorkerNotRemovedAfterPriorDrainException();
+ }
+
+ if (!availableWorkers.contains(workerId)) {
+ log.info("invokeDrain was called for a worker={} which is not currently active", workerId);
+ throw new UnknownWorkerException();
+ }
+
+ currentDrainFuture = drain(workerId);
+ return currentDrainFuture;
+ } finally {
+ drainInProgressFlag.set(false);
+ }
+ } else {
+ throw new DrainInProgressException();
+ }
+ }
+
+ public LongRunningProcessStatus getDrainStatus(String workerId) {
long startTime = System.nanoTime();
+ String errString;
+ LongRunningProcessStatus retVal = new LongRunningProcessStatus();
+ try {
+ val workerStatus = drainOpStatusMap.get(workerId);
+ if (workerStatus == null) {
+ errString = "Worker " + workerId + " not found in drain records";
+ retVal = LongRunningProcessStatus.forError(errString);
+ } else {
+ switch (workerStatus) {
+ default:
+ errString = "getDrainStatus: Unexpected status " + workerStatus + " found for worker " + workerId;
+ retVal = LongRunningProcessStatus.forError(errString);
+ break;
+ case DrainCompleted:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.SUCCESS);
+ break;
+ case DrainInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.RUNNING);
+ break;
+ case DrainNotInProgress:
+ retVal = LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.NOT_RUN);
+ break;
+ }
+ }
+ } finally {
+ log.info("Get drain status for worker {} - execution time: {} sec; returning status={}, error={}",
+ workerId, (System.nanoTime() - startTime) / Math.pow(10, 9), retVal.status, retVal.lastError);
+ return retVal;
+ }
+ }
+
+ @VisibleForTesting
Review comment:
Can you add a comment to "clearDrainOpsStatus", "setDrainOpsStatus", and "getDrainOpsStatusMap" that these methods are only used for testing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#issuecomment-927171630
/pulsarbot run-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719734979
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
+
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainInProgress);
+
+ val workerRemovealSuccessful = availableWorkers.remove(workerId);
+ Preconditions.checkState(workerRemovealSuccessful);
+
+ List<FunctionMetaData> allFunctions = functionMetaDataManager.getAllFunctionMetaData();
+ Map<String, Function.Instance> allInstances =
+ computeAllInstances(allFunctions, functionRuntimeManager.getRuntimeFactory().externallyManaged());
+
+ // Pretend that current assignments is empty, and get the instances to assign.
+ Pair<List<Function.Instance>, List<Assignment>> instancesToAssign
+ = getUnassignedFunctionInstances(null, allInstances);
Review comment:
This will cause getUnassignedFunctionInstances to just return all instances as unassigned. Later in the code, the scheduler will create a scheduling with the assumption that everything is unassigned. This is more disruptive than necessary. A drain operation does not entail a complete re-scheduling and moving instances from workers that not the target of the drain. The unassigned instances should just be the instances currently assigned to the worker that is being drained.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716919241
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Worker.java
##########
@@ -127,6 +129,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
Review comment:
Please refer to my comments in WorkerApiV2Resource.java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719761657
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
That is by design, and should be ok. PLMK if you foresee any specific issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719744518
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
+
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainInProgress);
+
+ val workerRemovealSuccessful = availableWorkers.remove(workerId);
+ Preconditions.checkState(workerRemovealSuccessful);
+
+ List<FunctionMetaData> allFunctions = functionMetaDataManager.getAllFunctionMetaData();
+ Map<String, Function.Instance> allInstances =
+ computeAllInstances(allFunctions, functionRuntimeManager.getRuntimeFactory().externallyManaged());
+
+ // Pretend that current assignments is empty, and get the instances to assign.
+ Pair<List<Function.Instance>, List<Assignment>> instancesToAssign
+ = getUnassignedFunctionInstances(null, allInstances);
+
+ workerStatsManager.drainTotalExecTimeStart();
+ // Try to schedule the instances on the workers remaining available after "workerId" is removed.
+ try {
+ postDrainAssignments = scheduler.schedule(instancesToAssign.getLeft(), null, availableWorkers);
+ } catch (Exception e) {
+ log.info("invokeDrain: Got exception from schedule: ", e);
+ }
+ workerStatsManager.drainTotalExecTimeEnd();
+
+ if (postDrainAssignments != null) {
+ // log.info("drain: postDrainAssignments={}", postDrainAssignments);
Review comment:
can we remove this commented out comment?
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +548,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
+
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainInProgress);
+
+ val workerRemovealSuccessful = availableWorkers.remove(workerId);
+ Preconditions.checkState(workerRemovealSuccessful);
+
+ List<FunctionMetaData> allFunctions = functionMetaDataManager.getAllFunctionMetaData();
+ Map<String, Function.Instance> allInstances =
+ computeAllInstances(allFunctions, functionRuntimeManager.getRuntimeFactory().externallyManaged());
+
+ // Pretend that current assignments is empty, and get the instances to assign.
+ Pair<List<Function.Instance>, List<Assignment>> instancesToAssign
+ = getUnassignedFunctionInstances(null, allInstances);
+
+ workerStatsManager.drainTotalExecTimeStart();
+ // Try to schedule the instances on the workers remaining available after "workerId" is removed.
+ try {
+ postDrainAssignments = scheduler.schedule(instancesToAssign.getLeft(), null, availableWorkers);
+ } catch (Exception e) {
+ log.info("invokeDrain: Got exception from schedule: ", e);
+ }
+ workerStatsManager.drainTotalExecTimeEnd();
+
+ if (postDrainAssignments != null) {
+ // log.info("drain: postDrainAssignments={}", postDrainAssignments);
+ for (Assignment assignment : postDrainAssignments) {
+ MessageId messageId = publishNewAssignment(assignment, false);
+ // Directly update in memory assignment cache since I am leader
+ // log.info("Drain worker {} - new assignment: {}", workerId, assignment);
Review comment:
same here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719726628
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
@kaushik-develop there are two independent threads read and updating drainOpStatusMap. One is a schedule periodic task:
https://github.com/apache/pulsar/pull/12178/files#diff-343b3460561c5e794ce7351d663880e37784d37e8c0a877c9e4845b5209a8c84R568
The other is when a drain operation is triggered. Two independent actors can be read and modifying the map concurrently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r719761657
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
That is by design. The external orchestrator is expected to trigger the draining of a worker, and check for drain status. It is also expected to remove the worker after a drain operation completes, and not re-add a worker with the same name as the drained/removed worker for one period (configurable, nominally 60 seconds) after the worker-removal. This will ensure that the two actors work on different entries of the concurrent map. But PLMK if you foresee problems in any specific scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716883153
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -403,6 +545,82 @@ private void scheduleCompaction(ScheduledExecutorService executor, long schedule
}
}
+ @VisibleForTesting
+ List<Assignment> getCurrentPostDrainAssignments() {
+ return currentPostDrainAssignments;
+ }
+
+ @VisibleForTesting
+ void clearCurrentPostDrainAssignments() {
+ currentPostDrainAssignments = null;
+ }
+
+ @VisibleForTesting
+ List<Assignment> invokeDrain(String workerId) {
+
+ long startTime = System.nanoTime();
+
+ Set<String> availableWorkers = getCurrentAvailableWorkers();
+
+ // workerIdToAssignments is a map of the form {workerId : {FullyQualifiedInstanceId : Assignment}}
+ Map<String, Map<String, Assignment>> workerIdToAssignments = functionRuntimeManager.getCurrentAssignments();
+
+ // initialize stats collection
+ SchedulerStats schedulerStats = new SchedulerStats(workerIdToAssignments, availableWorkers);
+
+ boolean drainSuccessful = false;
+ List<Assignment> postDrainAssignments = null;
+
+ try {
+ // The worker was checked to be among the available ones, by the caller.
+ Preconditions.checkState(availableWorkers.contains(workerId));
+
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainInProgress);
+
+ val workerRemovealSuccessful = availableWorkers.remove(workerId);
+ Preconditions.checkState(workerRemovealSuccessful);
+
+ List<FunctionMetaData> allFunctions = functionMetaDataManager.getAllFunctionMetaData();
+ Map<String, Function.Instance> allInstances =
+ computeAllInstances(allFunctions, functionRuntimeManager.getRuntimeFactory().externallyManaged());
+
+ // Pretend that current assignments is empty, and get the instances to assign.
+ Pair<List<Function.Instance>, List<Assignment>> instancesToAssign
+ = getUnassignedFunctionInstances(null, allInstances);
+
+ workerStatsManager.drainTotalExecTimeStart();
+ // Try to schedule the instances on the workers remaining available after "workerId" is removed.
+ try {
+ postDrainAssignments = scheduler.schedule(instancesToAssign.getLeft(), null, availableWorkers);
+ } catch (Exception e) {
+ log.info("invokeDrain: Got exception from schedule: ", e);
+ }
+ workerStatsManager.drainTotalExecTimeEnd();
+
+ if (postDrainAssignments != null) {
+ // log.info("drain: postDrainAssignments={}", postDrainAssignments);
+ for (Assignment assignment : postDrainAssignments) {
+ MessageId messageId = publishNewAssignment(assignment, false);
+ // Directly update in memory assignment cache since I am leader
+ // log.info("Drain worker {} - new assignment: {}", workerId, assignment);
+ functionRuntimeManager.processAssignment(assignment);
+ // update message id associated with current view of assignments map
+ lastMessageProduced = messageId;
+ // update stats
+ schedulerStats.newAssignment(assignment);
+ }
+ }
+ drainOpStatusMap.put(workerId, DrainOpStatus.DrainCompleted);
+ drainSuccessful = true;
+ } finally {
+ log.info("Draining worker {} was {}successful; summary [] - execution time: {} sec | stats: {}\n{}",
Review comment:
Missing space "{}" and "successful"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716930912
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I know drainOpStatusMap is a concurrent map but we are reading and writing from different different threads. This makes me nervous. Is there a reason we should synchronize this with drain operation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] kaushik-develop commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
kaushik-develop commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r717748796
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
##########
@@ -415,6 +633,34 @@ private void compactAssignmentTopic() {
}
}
+ protected int updateWorkerDrainMap() {
+ long startTime = System.nanoTime();
+ int numRemovedWorkerIds = 0;
+
+ if (drainOpStatusMap.size() > 0) {
Review comment:
I didn't quite understand the question: "Is there a reason we should synchronize this with drain operation?"
updateWorkerDrainMap() does a periodic cleanup of stale information about drained workers. We need to do the operation some time after the drain has finished, when the drained worker is removed from the cluster. Since there is currently no hook (that I know of) into the SchedulerManager when a worker is added to, or removed from the cluster, the cleanup is done through a periodic poll (updateWorkerDrainMap).
The drain operation adds a record into the concurrent map [drainOpStatusMap] when a worker is drained. An implicit assumption in the system is that the drained worker will be removed from the system, soon, by an external orchestrator. When the drained worker is seen to be removed from the system, the drainOpStatusMap is cleaned up of stale information in the updateWorkerDrainMap() code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #12178: Support for draining workers
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #12178:
URL: https://github.com/apache/pulsar/pull/12178#discussion_r716908422
##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
##########
@@ -156,6 +160,65 @@ public void rebalance() {
workers().rebalance(uri.getRequestUri(), clientAppId());
}
+ @PUT
+ @ApiOperation(
+ value = "Drains the specified worker, i.e., moves its work-assignments to other workers"
+ )
+ @ApiResponses(value = {
+ @ApiResponse(code = 400, message = "Invalid request"),
+ @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+ @ApiResponse(code = 408, message = "Request timeout"),
+ @ApiResponse(code = 409, message = "Drain already in progress"),
+ @ApiResponse(code = 503, message = "Worker service is not ready")
+ })
+ @Path("/leader/{drain}")
+ public void drainAtLeader(@PathParam("drain") String workerId) {
Review comment:
I think a better name for the URL parameter is "workerId".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org