Support dynamic job scaling rules #756
Open
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Andyz26
requested review from
calvin681,
hmitnflx,
fdc-ntflx and
dtrager02
as code owners
Andyz26
requested a deployment
to
Integrate Pull Request
GitHub Actions
Waiting
Test Results645 tests +23 635 ✅ +23 8m 36s ⏱️ +35s Results for commit 5cbfd7f. ± Comparison against base commit 1ffc3b9. This pull request removes 2 and adds 25 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
Andyz26
force-pushed
the
andyz/dynamicJobScaler1
branch
from
c8f9ba9 to
c0744fe
Compare
Andyz26
requested a deployment
to
Integrate Pull Request
GitHub Actions
Waiting
Andyz26
temporarily deployed
to
Integrate Pull Request
GitHub Actions
Inactive
Uploaded ArtifactsTo use these artifacts in your Gradle project, paste the following lines in your build.gradle. |
Andyz26
requested a deployment
to
Integrate Pull Request
GitHub Actions
Waiting
Andyz26
force-pushed
the
andyz/dynamicJobScaler1
branch
from
bc4b4c7 to
4fd427a
Compare
Andyz26
requested a deployment
to
Integrate Pull Request
GitHub Actions
Waiting
fdc-ntflx
reviewed
Mar 6, 2025
Comment on lines
+19
to
+21
| public static final String TRIGGER_TYPE_SCHEDULE = "schedule"; | ||
| public static final String TRIGGER_TYPE_PERPETUAL = "perpetual"; | ||
| public static final String TRIGGER_TYPE_CUSTOM = "custom"; |
There was a problem hiding this comment.
tbh i found using enum on proto class and tests to be quite verbose and error prone comparing to const strings.
|
|
||
| @Builder | ||
| @Value | ||
| public class JobScalingRule implements Serializable { |
There was a problem hiding this comment.
I believe this model is quite flexible, but I'm concerned it could lead to confusion about which policies are currently playing a major role or influencing decisions. Some policies might even contradict each other, like "scaling up vs. scaling down," potentially causing the algorithm to oscillate.
There was a problem hiding this comment.
I also added a new metric to report current active rule too. Will leave the logic as is in beta and revisit later from user feedback.
| import io.mantisrx.server.core.NamedJobInfo; | ||
| import io.mantisrx.server.core.PostJobStatusRequest; | ||
| import io.mantisrx.server.core.Status; | ||
| import io.mantisrx.server.core.*; |
There was a problem hiding this comment.
nit: don't use wildcard imports.
There was a problem hiding this comment.
I don't have a strong opinion on this and i would like to just apply current default profile.
| @@ -745,6 +741,52 @@ public Observable<JobSchedulingInfo> schedulingChanges(final String jobId) { | |||
| ; | |||
| } | |||
|
|
|||
| public Observable<JobScalerRuleInfo> jobScalerRulesStream(final String jobId) { | |||
There was a problem hiding this comment.
nit: Consider adding docstrings here to provide a quick understanding of the purpose of this long function.
There was a problem hiding this comment.
Side note: These functions have similar verbose code for subscribing and streaming events from master (or at least that's my understanding of this code). Consider extracting that into a separate auxiliary function.
| import io.mantisrx.server.core.JobSchedulingInfo; | ||
| import io.mantisrx.server.core.NamedJobInfo; | ||
| import io.mantisrx.server.core.Status; | ||
| import io.mantisrx.server.core.*; |
Comment on lines
+20
to
+21
| String jobId; | ||
| boolean jobCompleted; |
There was a problem hiding this comment.
Interesting. I didn't expect these to be part of this message. Can you explain?
There was a problem hiding this comment.
This is in the same semantics as other stream APIs and this mechanism is used to gracefully close the connection if the current job is terminated to avoid connection leak.
| @@ -308,6 +315,9 @@ private Receive getInitializingBehavior() { | |||
| .match(ScaleStageRequest.class, (x) -> getSender().tell(new ScaleStageResponse(x.requestId, CLIENT_ERROR, genUnexpectedMsg(x.toString(), state), 0), getSelf())) | |||
| .match(ResubmitWorkerRequest.class, (x) -> getSender().tell(new ResubmitWorkerResponse(x.requestId, CLIENT_ERROR, genUnexpectedMsg(x.toString(), state)), getSelf())) | |||
| .match(WorkerEvent.class, (x) -> logger.warn(genUnexpectedMsg(x.toString(), state))) | |||
| .match(JobClusterScalerRuleProto.CreateScalerRuleRequest.class, (x) -> getSender().tell(JobClusterScalerRuleProto.CreateScalerRuleResponse.builder().requestId(x.requestId).responseCode(CLIENT_ERROR_NOT_FOUND).message(genUnexpectedMsg(x.toString(), state)).build(), getSelf())) | |||
| .match(JobClusterScalerRuleProto.DeleteScalerRuleRequest.class, (x) -> getSender().tell(JobClusterScalerRuleProto.DeleteScalerRuleResponse.builder().requestId(x.requestId).responseCode(CLIENT_ERROR_NOT_FOUND).message(genUnexpectedMsg(x.toString(), state)).build(), getSelf())) | |||
| .match(JobClusterScalerRuleProto.GetScalerRulesRequest.class, (x) -> getSender().tell(JobClusterScalerRuleProto.GetScalerRulesResponse.builder().requestId(x.requestId).responseCode(CLIENT_ERROR_NOT_FOUND).message(genUnexpectedMsg(x.toString(), state)).build(), getSelf())) | |||
There was a problem hiding this comment.
do you accept GetJobScalerRuleStreamRequest during initialization?
There was a problem hiding this comment.
Good catch! updated.
Andyz26
temporarily deployed
to
Integrate Pull Request
GitHub Actions
Inactive
Andyz26
force-pushed
the
andyz/dynamicJobScaler1
branch
from
6838080 to
a216116
Compare
Andyz26
requested a deployment
to
Integrate Pull Request
GitHub Actions
Waiting
Andyz26
temporarily deployed
to
Integrate Pull Request
GitHub Actions
Inactive
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
[Reviewer Guidance]
The following is a summary of critical classes for review.
Runtime
Introduce new v2 job master service with refactor to current job scaler to allow ScalerControllerActor to recreate/shutdown running scaler. Support rule actors under coordinator actor.
API
New control plane API to support CRUD on scaling rules.
ControlPlane
Control plane implementation to support new APIs and the scaling changes stream API.
Persistency
E2E Test
Dynamic Job Scaling Rules Implementation
Problem Statement
In order to effectively manage job scaling in various scenarios, particularly during periods of bursty traffic, there is a need to dynamically update job auto scaler settings such as minimum, maximum, step-size, and burst-change. The goal is to consolidate the APIs required to modify a job's auto scaler and to allow changes to the job scaler without necessitating job resubmission.
Solution
This pull request introduces support for dynamic job scaling rules that override the default job scaler configuration. It allows multiple rules to coexist using different modes, providing flexibility in handling diverse scaling requirements.
Key Features:
JobScalingRuleCustomTriggerinterface to customize the triggering conditions and actions.Internals
ruleId, with newer rules taking precedence over older ones./jobScalerRules/stream API to listen for rule changes, ensuring that the system is responsive to updates.schedulingInfoscaler config is treated as the default rule with an ID of "-1".JM v2
This implementation provides a robust framework for managing job scaling dynamically, enhancing the system's ability to handle varying traffic patterns efficiently. The introduction of multiple rule types and the ability to customize scaling behavior offers significant flexibility and control to users.
API Reference for Job Cluster Scaler Rules
This document provides details on the APIs available under
api/v1/jobClusters/{}/scalerRules. These APIs allow you to manage scaler rules for job clusters, including creating, retrieving, and deleting scaler rules.Endpoints
1. Create Scaler Rule
POST /api/v1/jobClusters/{clusterName}/scalerRulesRequest Schema
Sample Request Payload
{ "jobClusterName": "exampleCluster", "scalerConfig": { "type": "standard", "scalingPolicies": [ { "stage": 1, "min": 1, "max": 5, "increment": 1, "decrement": 1, "coolDownSecs": 300, "strategies": { "CPU": { "reason": "CPU", "scaleDownBelowPct": 20.0, "scaleUpAbovePct": 80.0, "rollingCount": { "count": 3, "of": 5 } } }, "allowAutoScaleManager": true } ], "stageDesireSize": { "1": 3 } }, "triggerConfig": { "triggerType": "schedule", "scheduleCron": "0 0 15 * * ?", "scheduleDuration": "1h", "customTrigger": "" }, "metadata": { "createdBy": "admin" } }