Add Bulk Scheduler & Webhook updates

[bharathappali]

This PR is built on top of #35

#35 needs to be merged before merging this PR

This PR adds the bulk scheduler changes

Summary by Sourcery

Introduce profile-based bulk scheduling driven by Kruize bulk profiles and integrate it with existing bulk scheduler and webhook handling.

New Features:

• Add profile-based bulk scheduling using per-profile timers that trigger bulk jobs based on recommendation settings from Kruize.
• Expose client APIs and models to fetch and consume Kruize bulk profiles, including cluster and recommendation configuration.
• Track jobs triggered per bulk profile and expose per-profile job counts via the jobs service.
• Support webhook-driven updates of bulk profiles to dynamically adjust or cancel profile timers.

Enhancements:

• Gate legacy fixed-interval bulk scheduling behind a configuration flag and clarify related configuration properties in application.yml.
• Add lifecycle management for profile timers, including initialization at startup and graceful shutdown.

[sourcery-ai]

Reviewer's Guide

Introduce profile-based bulk scheduling driven by Kruize bulk profiles, including per-profile timers, new profile/webhook handling paths, and basic per-profile job metrics, while keeping the existing fixed-interval scheduler as a fallback mode.

Sequence diagram for profile-based bulk scheduling and execution

sequenceDiagram
    participant BulkSchedulerService
    participant ProfileTimerManager
    participant BulkProfileService
    participant KruizeClient
    participant Scheduler
    participant JobsService

    BulkSchedulerService->>ProfileTimerManager: initializeProfiles()
    ProfileTimerManager->>BulkProfileService: getEnabledProfiles()
    BulkProfileService->>KruizeClient: getBulkProfiles()
    KruizeClient-->>BulkProfileService: JSON BulkProfile[]
    BulkProfileService-->>ProfileTimerManager: List<BulkProfile>
    loop for each BulkProfile
        ProfileTimerManager->>ProfileTimerManager: scheduleProfile(profile)
        ProfileTimerManager->>Scheduler: scheduleAtFixedRate(executeProfileJob)
    end

    rect rgb(230,230,230)
    Scheduler->>ProfileTimerManager: executeProfileJob(profile)
    ProfileTimerManager->>BulkProfileService: convertProfileToBulkJob(profile)
    BulkProfileService-->>ProfileTimerManager: Map bulkJob
    ProfileTimerManager->>KruizeClient: bulkCreateExperiments(bulkJob)
    KruizeClient-->>ProfileTimerManager: response
    ProfileTimerManager->>JobsService: incrementJobsTriggered(profileName)
    end

Loading

File-Level Changes

Change	Details	Files
Add profile-based bulk scheduling with per-profile timers managed independently from the legacy fixed-interval scheduler.	Inject ProfileTimerManager and a feature flag into the bulk scheduler service to toggle between profile-based and legacy scheduling On initialization, refresh state as before and, when enabled, fetch and schedule timers for all active profiles via ProfileTimerManager instead of relying on the fixed cron/duration scheduler Short-circuit the scheduled bulk API call when profile-based scheduling is enabled so that bulk jobs are only triggered via profile timers	src/main/java/com/kruize/optimizer/service/BulkSchedulerService.java src/main/java/com/kruize/optimizer/service/ProfileTimerManager.java src/main/resources/application.yml
Introduce bulk profile domain model, REST client integrations, and conversion logic to generate bulk jobs from profiles.	Define BulkProfile, ClusterConfig, and RecommendationSettings model classes matching Kruize bulk profile JSON structure Extend KruizeClient with endpoints to fetch one or all bulk profiles from Kruize Implement BulkProfileService to fetch enabled profiles, parse human-readable scheduling strings into Durations, and convert profile configs into bulk job request payloads including filters, datasource, metadata profile, measurement duration, and webhook URL	src/main/java/com/kruize/optimizer/model/kruize/BulkProfile.java src/main/java/com/kruize/optimizer/model/kruize/ClusterConfig.java src/main/java/com/kruize/optimizer/model/kruize/RecommendationSettings.java src/main/java/com/kruize/optimizer/client/KruizeClient.java src/main/java/com/kruize/optimizer/service/BulkProfileService.java
Enable dynamic profile update handling and tracking of jobs per profile.	Add a handleProfileUpdate method to bulk scheduler service to receive profile update webhooks and delegate timer updates to ProfileTimerManager when profile-based scheduling is enabled Implement logic in ProfileTimerManager to create, update, and cancel scheduled tasks per profile, execute bulk jobs via KruizeClient using the converted job payload, and cleanly shut down timers on application shutdown Extend JobsService to keep per-profile job trigger counts and expose them via a read-only map	src/main/java/com/kruize/optimizer/service/BulkSchedulerService.java src/main/java/com/kruize/optimizer/service/ProfileTimerManager.java src/main/java/com/kruize/optimizer/service/JobsService.java
Wire configuration and constants for bulk profile features.	Add configuration properties to toggle profile-based scheduling and clarify that legacy scheduler interval/measurement-duration apply only when profile mode is disabled Introduce constants for bulk profiles endpoint and profile_name query parameter used by the Kruize client	src/main/resources/application.yml src/main/java/com/kruize/optimizer/utils/OptimizerConstants.java

Possibly linked issues

• #: They match: PR adds Bulk Profile API client, profile timers, webhooks, stats, and feature-flagged legacy mode.

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've found 2 issues, and left some high level feedback:

• In BulkProfileService.convertProfileToBulkJob() and ProfileTimerManager.scheduleProfile()/updateProfileTimer(), consider adding defensive null checks for recommendationSettings and its fields (e.g., getScheduling(), getMeasurementDuration()), since these come from an external API and can currently cause NPEs or IllegalArgumentExceptions that abort initialization or webhook handling.
• The logic in ProfileTimerManager.updateProfileTimer() for handling interval changes can be simplified: both branches cancel and reschedule the timer, so you can remove the delayMillis > newInterval.toMillis() condition unless you plan to treat shorter vs. longer intervals differently.
• In ProfileTimerManager, the ScheduledExecutorService is created with a hardcoded pool size of 10; consider either making this configurable or using a container-managed executor so the thread usage can be tuned for different deployment environments.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- In `BulkProfileService.convertProfileToBulkJob()` and `ProfileTimerManager.scheduleProfile()/updateProfileTimer()`, consider adding defensive null checks for `recommendationSettings` and its fields (e.g., `getScheduling()`, `getMeasurementDuration()`), since these come from an external API and can currently cause NPEs or `IllegalArgumentException`s that abort initialization or webhook handling.
- The logic in `ProfileTimerManager.updateProfileTimer()` for handling interval changes can be simplified: both branches cancel and reschedule the timer, so you can remove the `delayMillis > newInterval.toMillis()` condition unless you plan to treat shorter vs. longer intervals differently.
- In `ProfileTimerManager`, the `ScheduledExecutorService` is created with a hardcoded pool size of 10; consider either making this configurable or using a container-managed executor so the thread usage can be tuned for different deployment environments.

## Individual Comments

### Comment 1
<location path="src/main/java/com/kruize/optimizer/service/BulkProfileService.java" line_range="177-178" />
<code_context>
+            bulkJob.put("metadata_profile", cluster.getMetadataProfile());
+        }
+
+        // Add measurement duration from recommendation settings
+        String measurementDuration = profile.getRecommendationSettings().getMeasurementDuration();
+        if (measurementDuration != null && !measurementDuration.isEmpty()) {
+            bulkJob.put("measurement_duration", measurementDuration);
</code_context>
<issue_to_address>
**issue (bug_risk):** Guard against null recommendation settings when building bulk job requests

This dereferences `profile.getRecommendationSettings()` without a null check. If Kruize returns a profile without `recommendation_settings`, this will throw a `NullPointerException` and prevent bulk job creation for that profile. Add a null check before calling `getMeasurementDuration()`, and consider logging and skipping `measurement_duration` when settings are missing.
</issue_to_address>

### Comment 2
<location path="src/main/java/com/kruize/optimizer/service/BulkProfileService.java" line_range="92-101" />
<code_context>
+    public Duration parseScheduling(String scheduling) {
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider making parseScheduling more robust to unexpected formats to avoid bringing down callers

`parseScheduling` currently throws `IllegalArgumentException` for null/empty/invalid strings, and callers like `scheduleProfile`/`updateProfileTimer` don’t catch it. A single bad value can therefore abort timer initialization or fail a webhook. Consider either handling these exceptions at the call sites (e.g., log and skip scheduling for that profile) or changing this method to return a default/optional value so callers can decide how to recover.

Suggested implementation:

```java
import java.time.Duration;
import java.util.Optional;

```

```java
     * @param scheduling Scheduling string
     * @return Optional Duration object, empty if the scheduling string is null/empty/invalid
     */
    public Optional<Duration> parseScheduling(String scheduling) {
        if (scheduling == null || scheduling.trim().isEmpty()) {
            LOG.warn("Skipping scheduling: scheduling string is null or empty");
            return Optional.empty();
        }

        Matcher matcher = SCHEDULING_PATTERN.matcher(scheduling.trim().toLowerCase());
        if (!matcher.matches()) {
            LOG.warn("Skipping scheduling: invalid scheduling format '{}'", scheduling);
            return Optional.empty();
        }

        int value = Integer.parseInt(matcher.group(1));

```

To fully implement the suggestion and avoid unexpected exceptions propagating to callers, you should also:

1. Ensure a logger is available in this class:
   - If not already present, add something like `private static final Logger LOG = LoggerFactory.getLogger(BulkProfileService.class);` and the corresponding SLF4J imports.
2. Update all call sites in this file (and others) from:
   - `Duration duration = parseScheduling(profile.getScheduling());`
   - to something that safely handles missing/invalid values, for example:
     - `Optional<Duration> maybeDuration = parseScheduling(profile.getScheduling());`
     - `maybeDuration.ifPresent(duration -> scheduleProfile(profile, duration));`
     - or provide a fallback/default duration where appropriate.
3. Adjust method signatures for `scheduleProfile` / `updateProfileTimer` (or similar) if they currently assume a non-null `Duration`, so they can either:
   - Accept an `Optional<Duration>`, or
   - Simply not be called when `parseScheduling` returns `Optional.empty()`.
4. If this class is part of a public API, update any interfaces or tests that reference `parseScheduling` to use `Optional<Duration>` and assert on the empty case instead of expecting exceptions.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

issue (bug_risk): Guard against null recommendation settings when building bulk job requests

This dereferences profile.getRecommendationSettings() without a null check. If Kruize returns a profile without recommendation_settings, this will throw a NullPointerException and prevent bulk job creation for that profile. Add a null check before calling getMeasurementDuration(), and consider logging and skipping measurement_duration when settings are missing.

[sourcery-ai]

suggestion (bug_risk): Consider making parseScheduling more robust to unexpected formats to avoid bringing down callers

parseScheduling currently throws IllegalArgumentException for null/empty/invalid strings, and callers like scheduleProfile/updateProfileTimer don’t catch it. A single bad value can therefore abort timer initialization or fail a webhook. Consider either handling these exceptions at the call sites (e.g., log and skip scheduling for that profile) or changing this method to return a default/optional value so callers can decide how to recover.

Suggested implementation:

import java.time.Duration;
import java.util.Optional;

     * @param scheduling Scheduling string
     * @return Optional Duration object, empty if the scheduling string is null/empty/invalid
     */
    public Optional<Duration> parseScheduling(String scheduling) {
        if (scheduling == null || scheduling.trim().isEmpty()) {
            LOG.warn("Skipping scheduling: scheduling string is null or empty");
            return Optional.empty();
        }

        Matcher matcher = SCHEDULING_PATTERN.matcher(scheduling.trim().toLowerCase());
        if (!matcher.matches()) {
            LOG.warn("Skipping scheduling: invalid scheduling format '{}'", scheduling);
            return Optional.empty();
        }

        int value = Integer.parseInt(matcher.group(1));

To fully implement the suggestion and avoid unexpected exceptions propagating to callers, you should also:

1. Ensure a logger is available in this class:
   • If not already present, add something like private static final Logger LOG = LoggerFactory.getLogger(BulkProfileService.class); and the corresponding SLF4J imports.
2. Update all call sites in this file (and others) from:
   • Duration duration = parseScheduling(profile.getScheduling());
   • to something that safely handles missing/invalid values, for example:
     • Optional<Duration> maybeDuration = parseScheduling(profile.getScheduling());
     • maybeDuration.ifPresent(duration -> scheduleProfile(profile, duration));
     • or provide a fallback/default duration where appropriate.
3. Adjust method signatures for scheduleProfile / updateProfileTimer (or similar) if they currently assume a non-null Duration, so they can either:
   • Accept an Optional<Duration>, or
   • Simply not be called when parseScheduling returns Optional.empty().
4. If this class is part of a public API, update any interfaces or tests that reference parseScheduling to use Optional<Duration> and assert on the empty case instead of expecting exceptions.