WIP: Add Bulk Profile changes in optimizer - #28
Draft
bharathappali wants to merge 9 commits into
Draft
Conversation
Signed-off-by: bharathappali <abharath@redhat.com>
Signed-off-by: bharathappali <abharath@redhat.com>
Signed-off-by: bharathappali <abharath@redhat.com>
Signed-off-by: bharathappali <abharath@redhat.com>
Signed-off-by: bharathappali <abharath@redhat.com>
Signed-off-by: bharathappali <abharath@redhat.com>
Signed-off-by: bharathappali <abharath@redhat.com>
Signed-off-by: bharathappali <abharath@redhat.com>
Reviewer's GuideIntroduces profile-based bulk scheduling by adding bulk profile models and services, a per-profile timer manager that triggers bulk jobs via Kruize, a webhook endpoint for profile updates, and a feature flag to switch between legacy fixed-schedule and new profile-driven behavior, including per-profile job metrics. Sequence diagram for profile-based bulk scheduling and job executionsequenceDiagram
participant BulkSchedulerService
participant ProfileTimerManager
participant BulkProfileService
participant KruizeClient
participant Scheduler
participant JobsService
BulkSchedulerService->>BulkSchedulerService: initialize()
alt profileBasedSchedulingEnabled
BulkSchedulerService->>ProfileTimerManager: initializeProfiles()
ProfileTimerManager->>BulkProfileService: getEnabledProfiles()
BulkProfileService->>KruizeClient: getBulkProfiles()
KruizeClient-->>BulkProfileService: bulkProfilesJson
BulkProfileService-->>ProfileTimerManager: List<BulkProfile>
loop for each BulkProfile
ProfileTimerManager->>ProfileTimerManager: scheduleProfile(profile)
ProfileTimerManager->>Scheduler: scheduleAtFixedRate(executeProfileJob, interval)
end
else legacyMode
BulkSchedulerService->>BulkSchedulerService: scheduledBulkApiCall()
BulkSchedulerService->>BulkSchedulerService: call legacy bulk API
end
rect rgb(230,230,230)
Scheduler->>ProfileTimerManager: executeProfileJob(profile)
ProfileTimerManager->>BulkProfileService: convertProfileToBulkJob(profile)
BulkProfileService-->>ProfileTimerManager: bulkJob
ProfileTimerManager->>KruizeClient: bulkCreateExperiments(bulkJob)
KruizeClient-->>ProfileTimerManager: response
ProfileTimerManager->>JobsService: incrementJobsTriggered(profileName)
end
Sequence diagram for bulk profile update webhook handlingsequenceDiagram
participant Kruize
participant WebhookResource
participant BulkSchedulerService
participant ProfileTimerManager
Kruize->>WebhookResource: POST /webhook/profile-update BulkProfile
WebhookResource->>WebhookResource: receiveProfileUpdate(profile)
alt invalidProfile
WebhookResource-->>Kruize: 400 Invalid profile update
else validProfile
WebhookResource->>BulkSchedulerService: handleProfileUpdate(profile)
alt profileBasedSchedulingEnabled
BulkSchedulerService->>ProfileTimerManager: updateProfileTimer(updatedProfile)
alt updatedProfile.enabled is false
ProfileTimerManager->>ProfileTimerManager: cancelProfileTimer(profileName)
else updatedProfile.enabled is true
ProfileTimerManager->>ProfileTimerManager: updateProfileTimer logic
end
else profileBasedSchedulingDisabled
BulkSchedulerService->>BulkSchedulerService: log ignore update
end
BulkSchedulerService-->>WebhookResource: void
WebhookResource-->>Kruize: 200 OK
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
bharathappali
changed the title
bharathappali
marked this pull request as draft
June 19, 2026 05:10
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- In JobsService,
incrementJobsTriggered(String profileName)is synchronized while also using aConcurrentHashMap, which is redundant; consider either removingsynchronizedand using atomic counters per profile, or using a plainHashMapunder synchronized access for consistency with the other counters. - ProfileTimerManager currently constructs its own
ScheduledThreadPoolExecutorwith a hard-coded pool size of 10; consider making the pool size configurable and using a namedThreadFactoryso these threads are easier to tune and observe in production. - BulkProfileService.getEnabledProfiles() logs and returns an empty list on failure to contact Kruize, which makes errors indistinguishable from 'no profiles'; consider surfacing the failure (e.g., via an exception or a result wrapper) so callers can react differently to transient errors versus a genuinely empty profile set.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In JobsService, `incrementJobsTriggered(String profileName)` is synchronized while also using a `ConcurrentHashMap`, which is redundant; consider either removing `synchronized` and using atomic counters per profile, or using a plain `HashMap` under synchronized access for consistency with the other counters.
- ProfileTimerManager currently constructs its own `ScheduledThreadPoolExecutor` with a hard-coded pool size of 10; consider making the pool size configurable and using a named `ThreadFactory` so these threads are easier to tune and observe in production.
- BulkProfileService.getEnabledProfiles() logs and returns an empty list on failure to contact Kruize, which makes errors indistinguishable from 'no profiles'; consider surfacing the failure (e.g., via an exception or a result wrapper) so callers can react differently to transient errors versus a genuinely empty profile set.
## Individual Comments
### Comment 1
<location path="src/main/java/com/kruize/optimizer/service/ProfileTimerManager.java" line_range="89-98" />
<code_context>
+ public void updateProfileTimer(BulkProfile updatedProfile) {
</code_context>
<issue_to_address>
**issue (bug_risk):** Timer update path always reschedules with immediate execution, making delay inspection/logs misleading.
Both branches of `updateProfileTimer` cancel the current timer and call `scheduleProfile(updatedProfile)`, which always uses an initial delay of `0`. As a result, the remaining delay from `currentTimer.getDelay(...)` is ignored and the “shortened” vs “changed” paths behave the same. Please either simplify by removing the unused delay calculation/logging, or update `scheduleProfile`/this method so the next run actually uses the computed delay.
</issue_to_address>
### Comment 2
<location path="src/main/java/com/kruize/optimizer/service/ProfileTimerManager.java" line_range="74-77" />
<code_context>
+ LOG.infof("Scheduling profile '%s' with interval: %s", profileName, interval);
+
+ // Schedule recurring task
+ ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(
+ () -> executeProfileJob(profile),
+ 0, // Initial delay = 0 (execute immediately)
+ interval.toMillis(),
+ TimeUnit.MILLISECONDS
+ );
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Running blocking HTTP calls inside fixed-rate scheduled tasks can cause timer drift and contention.
This uses `scheduleAtFixedRate` to trigger `executeProfileJob(profile)`, which in turn calls `kruizeClient.bulkCreateExperiments(...)` (likely a blocking HTTP call). With a fixed thread pool of 10, slow or stuck calls can cause overlapping runs for the same profile and eventually exhaust the scheduler threads. Consider switching to `scheduleWithFixedDelay` so each run waits for the previous one to finish, or move the HTTP call to a separate executor so the scheduled job stays lightweight. It would also be good to explicitly handle cases where execution time frequently exceeds the interval.
</issue_to_address>
### Comment 3
<location path="src/main/java/com/kruize/optimizer/service/BulkProfileService.java" line_range="59-68" />
<code_context>
+ *
+ * @return List of enabled bulk profiles
+ */
+ public List<BulkProfile> getEnabledProfiles() {
+ try {
+ String response = kruizeClient.getBulkProfiles();
+ List<BulkProfile> allProfiles = objectMapper.readValue(
+ response,
+ new TypeReference<List<BulkProfile>>() {}
+ );
+
+ List<BulkProfile> enabledProfiles = allProfiles.stream()
+ .filter(p -> p.getEnabled() != null && p.getEnabled())
+ .collect(Collectors.toList());
+
+ LOG.infof("Fetched %d enabled profiles out of %d total profiles",
+ enabledProfiles.size(), allProfiles.size());
+
+ return enabledProfiles;
+
+ } catch (Exception e) {
+ LOG.error("Failed to fetch bulk profiles from Kruize", e);
+ return Collections.emptyList();
+ }
+ }
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Swallowing errors and returning an empty profile list can silently disable profile-based scheduling.
Because any exception from `kruizeClient.getBulkProfiles()` or deserialization is mapped to `Collections.emptyList()`, callers like `initializeProfiles()` interpret this as “no enabled profiles” instead of a fetch failure, hiding real operational issues. Please propagate the failure (e.g., throw, or return a result type that distinguishes “0 profiles” from “fetch failed”) so the scheduler can react appropriately or emit a clearer signal.
Suggested implementation:
```java
/**
* Fetch all enabled profiles from Kruize
*
* @return List of enabled bulk profiles
*/
public List<BulkProfile> getEnabledProfiles() {
String response = kruizeClient.getBulkProfiles();
List<BulkProfile> allProfiles = objectMapper.readValue(
response,
new TypeReference<List<BulkProfile>>() {}
);
List<BulkProfile> enabledProfiles = allProfiles.stream()
.filter(p -> p.getEnabled() != null && p.getEnabled())
.collect(Collectors.toList());
LOG.infof("Fetched %d enabled profiles out of %d total profiles",
enabledProfiles.size(), allProfiles.size());
return enabledProfiles;
}
```
Callers such as `initializeProfiles()` should now be prepared for this method to throw exceptions (e.g., IO, HTTP, or deserialization errors) instead of always returning a list. If needed, you can add targeted exception handling at a higher level (e.g., around the scheduler initialization) to log and surface a clearer operational signal (metrics, alerts, or a specific failure state) when fetching profiles fails.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+89
to
+98
| public void updateProfileTimer(BulkProfile updatedProfile) { | ||
| String profileName = updatedProfile.getProfileName(); | ||
|
|
||
| if (updatedProfile.getEnabled() == null || !updatedProfile.getEnabled()) { | ||
| // Profile disabled - cancel timer | ||
| LOG.infof("Profile '%s' disabled, canceling timer", profileName); | ||
| cancelProfileTimer(profileName); | ||
| return; | ||
| } | ||
|
|
There was a problem hiding this comment.
issue (bug_risk): Timer update path always reschedules with immediate execution, making delay inspection/logs misleading.
Both branches of updateProfileTimer cancel the current timer and call scheduleProfile(updatedProfile), which always uses an initial delay of 0. As a result, the remaining delay from currentTimer.getDelay(...) is ignored and the “shortened” vs “changed” paths behave the same. Please either simplify by removing the unused delay calculation/logging, or update scheduleProfile/this method so the next run actually uses the computed delay.
Comment on lines
+74
to
+77
| ScheduledFuture<?> future = scheduler.scheduleAtFixedRate( | ||
| () -> executeProfileJob(profile), | ||
| 0, // Initial delay = 0 (execute immediately) | ||
| interval.toMillis(), |
There was a problem hiding this comment.
suggestion (bug_risk): Running blocking HTTP calls inside fixed-rate scheduled tasks can cause timer drift and contention.
This uses scheduleAtFixedRate to trigger executeProfileJob(profile), which in turn calls kruizeClient.bulkCreateExperiments(...) (likely a blocking HTTP call). With a fixed thread pool of 10, slow or stuck calls can cause overlapping runs for the same profile and eventually exhaust the scheduler threads. Consider switching to scheduleWithFixedDelay so each run waits for the previous one to finish, or move the HTTP call to a separate executor so the scheduled job stays lightweight. It would also be good to explicitly handle cases where execution time frequently exceeds the interval.
Comment on lines
+59
to
+68
| public List<BulkProfile> getEnabledProfiles() { | ||
| try { | ||
| String response = kruizeClient.getBulkProfiles(); | ||
| List<BulkProfile> allProfiles = objectMapper.readValue( | ||
| response, | ||
| new TypeReference<List<BulkProfile>>() {} | ||
| ); | ||
|
|
||
| List<BulkProfile> enabledProfiles = allProfiles.stream() | ||
| .filter(p -> p.getEnabled() != null && p.getEnabled()) |
There was a problem hiding this comment.
suggestion (bug_risk): Swallowing errors and returning an empty profile list can silently disable profile-based scheduling.
Because any exception from kruizeClient.getBulkProfiles() or deserialization is mapped to Collections.emptyList(), callers like initializeProfiles() interpret this as “no enabled profiles” instead of a fetch failure, hiding real operational issues. Please propagate the failure (e.g., throw, or return a result type that distinguishes “0 profiles” from “fetch failed”) so the scheduler can react appropriately or emit a clearer signal.
Suggested implementation:
/**
* Fetch all enabled profiles from Kruize
*
* @return List of enabled bulk profiles
*/
public List<BulkProfile> getEnabledProfiles() {
String response = kruizeClient.getBulkProfiles();
List<BulkProfile> allProfiles = objectMapper.readValue(
response,
new TypeReference<List<BulkProfile>>() {}
);
List<BulkProfile> enabledProfiles = allProfiles.stream()
.filter(p -> p.getEnabled() != null && p.getEnabled())
.collect(Collectors.toList());
LOG.infof("Fetched %d enabled profiles out of %d total profiles",
enabledProfiles.size(), allProfiles.size());
return enabledProfiles;
}Callers such as initializeProfiles() should now be prepared for this method to throw exceptions (e.g., IO, HTTP, or deserialization errors) instead of always returning a list. If needed, you can add targeted exception handling at a higher level (e.g., around the scheduler initialization) to log and surface a clearer operational signal (metrics, alerts, or a specific failure state) when fetching profiles fails.
Signed-off-by: bharathappali <abharath@redhat.com>
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
This PR adds the bulk profile support for the optimizer
Summary by Sourcery
Introduce profile-based bulk scheduling alongside the existing fixed-interval mode, driven by bulk profiles fetched from Kruize and managed via per-profile timers.
New Features:
Enhancements: