Skip to content

Commit 5ac6390

Browse files
kaituokaituo
authored
Refactoring task cache manager for forecasting (#982)
* Refactoring Shared Functionality for Broader Task Compatibility This PR extracts shared components from `ADTaskCacheManager` into `TaskCacheManager` to support the requirements of the forecasting feature. **Renamings**: - **Method-level in `ADTaskCacheManager`**: - `addDeletedDetector` to `addDeletedConfig` - `addDeletedDetectorTask` to `addDeletedTask` - `hasDeletedDetectorTask` to `hasDeletedTask` - `pollDeletedDetector` to `pollDeletedConfig` - `pollDeletedDetectorTask` to `pollDeletedTask` - **Variable-level in `AnomalyDetectorSettings`**: - `CHECKPOINT_MAINTAIN_QUEUE_MAX_HEAP_PERCENT` to `AD_CHECKPOINT_MAINTAIN_QUEUE_MAX_HEAP_PERCENT` - `CHECKPOINT_READ_QUEUE_MAX_HEAP_PERCENT` to `AD_CHECKPOINT_READ_QUEUE_MAX_HEAP_PERCENT` - `CHECKPOINT_SAVING_FREQ` to `AD_CHECKPOINT_SAVING_FREQ` - `CHECKPOINT_TTL` to `AD_CHECKPOINT_TTL` - `CHECKPOINT_WRITE_QUEUE_MAX_HEAP_PERCENT` to `AD_CHECKPOINT_WRITE_QUEUE_MAX_HEAP_PERCENT` - `COLD_ENTITY_QUEUE_MAX_HEAP_PERCENT` to `AD_COLD_ENTITY_QUEUE_MAX_HEAP_PERCENT` - `DEDICATED_CACHE_SIZE` to `AD_DEDICATED_CACHE_SIZE` - `ENTITY_COLD_START_QUEUE_CONCURRENCY` to `AD_ENTITY_COLD_START_QUEUE_CONCURRENCY` - `ENTITY_COLD_START_QUEUE_MAX_HEAP_PERCENT` to `AD_ENTITY_COLD_START_QUEUE_MAX_HEAP_PERCENT` - `EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS` to `AD_EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS` - `FILTER_BY_BACKEND_ROLES` to `AD_FILTER_BY_BACKEND_ROLES` - `MAX_ENTITIES_PER_QUERY` to `AD_MAX_ENTITIES_PER_QUERY` - `MAX_MODEL_SIZE_PER_NODE` to `AD_MAX_MODEL_SIZE_PER_NODE` - `MAX_MULTI_ENTITY_ANOMALY_DETECTORS` to `AD_MAX_HC_ANOMALY_DETECTORS` - `MAX_RETRY_FOR_END_RUN_EXCEPTION` to `AD_MAX_RETRY_FOR_END_RUN_EXCEPTION` - `MAX_SINGLE_ENTITY_ANOMALY_DETECTORS` to `AD_MAX_SINGLE_ENTITY_ANOMALY_DETECTORS` - `MODEL_MAX_SIZE_PERCENTAGE` to `AD_MODEL_MAX_SIZE_PERCENTAGE` - `PAGE_SIZE` to `AD_PAGE_SIZE` - `RESULT_WRITE_QUEUE_MAX_HEAP_PERCENT` to `AD_RESULT_WRITE_QUEUE_MAX_HEAP_PERCENT` - **Class-level**: - `ADRealtimeTaskCache` renamed to `RealtimeTaskCache` - **Package-level**: - Shifted from `org.opensearch.ad.breaker` to `org.opensearch.timeseries.breaker` **Migrations**: - Variables transferred from `AnomalyDetectorSettings` to `TimeSeriesSettings`: - `BATCH_BOUNDING_BOX_CACHE_RATIO` - `CHECKPOINT_MAINTAIN_REQUEST_SIZE_IN_BYTES` - `CHECKPOINT_WRITE_QUEUE_SIZE_IN_BYTES` - `DEFAULT_AD_JOB_LOC_DURATION_SECONDS` (renamed to `DEFAULT_JOB_LOC_DURATION_SECONDS`) - `ENTITY_REQUEST_SIZE_IN_BYTES` - `HOURLY_MAINTENANCE` - `INTERVAL_RATIO_FOR_REQUESTS` - `LOW_SEGMENT_PRUNE_RATIO` - `MAINTENANCE_FREQ_CONSTANT` - `MAX_COLD_START_ROUNDS` - `MAX_QUEUED_TASKS_RATIO` - `MAX_TOTAL_RCF_SERIALIZATION_BUFFERS` - `MAX_CHECKPOINT_BYTES` - `MEDIUM_SEGMENT_PRUNE_RATIO` - `NUM_MIN_SAMPLES` - `NUM_SAMPLES_PER_TREE` - `NUM_TREES` - `QUEUE_MAINTENANCE` - `RESULT_WRITE_QUEUE_SIZE_IN_BYTES` - `SERIALIZATION_BUFFER_BYTES` - `THRESHOLD_MIN_PVALUE` - `TIME_DECAY` **Deletions**: - Obsolete settings and methods: - `DESIRED_MODEL_SIZE_PERCENTAGE` in `AnomalyDetectorSettings` - `getDesiredModelSize` in `MemoryTracker` **Modifications**: - `MemoryTracker` enums renamed for clear differentiation between real-time and historical memory usage, adding `REAL_TIME_FORECASTER` for a harmonized single-stream and HC analysis approach. **Tests**: - Changes validated with a successful Gradle build. Signed-off-by: Kaituo Li <kaituo@amazon.com> * improve comments Signed-off-by: Kaituo Li <kaituo@amazon.com> * improve comments Signed-off-by: Kaituo Li <kaituo@amazon.com> --------- Signed-off-by: Kaituo Li <kaituo@amazon.com>
1 parent 0c5b4b9 commit 5ac6390

File tree

95 files changed

+1090
-1163
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

95 files changed

+1090
-1163
lines changed

‎src/main/java/org/opensearch/ad/AnomalyDetectorJobRunner.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,7 @@ public void setThreadPool(ThreadPool threadPool) {
112112

113113
public void setSettings(Settings settings) {
114114
this.settings = settings;
115-
this.maxRetryForEndRunException = AnomalyDetectorSettings.MAX_RETRY_FOR_END_RUN_EXCEPTION.get(settings);
115+
this.maxRetryForEndRunException = AnomalyDetectorSettings.AD_MAX_RETRY_FOR_END_RUN_EXCEPTION.get(settings);
116116
}
117117

118118
public void setAdTaskManager(ADTaskManager adTaskManager) {

‎src/main/java/org/opensearch/ad/AnomalyDetectorProfileRunner.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,6 @@
3939
import org.opensearch.ad.model.DetectorState;
4040
import org.opensearch.ad.model.InitProgressProfile;
4141
import org.opensearch.ad.settings.ADNumericSetting;
42-
import org.opensearch.ad.settings.AnomalyDetectorSettings;
4342
import org.opensearch.ad.task.ADTaskManager;
4443
import org.opensearch.ad.transport.ProfileAction;
4544
import org.opensearch.ad.transport.ProfileRequest;
@@ -70,6 +69,7 @@
7069
import org.opensearch.timeseries.constant.CommonName;
7170
import org.opensearch.timeseries.model.IntervalTimeConfiguration;
7271
import org.opensearch.timeseries.model.Job;
72+
import org.opensearch.timeseries.settings.TimeSeriesSettings;
7373
import org.opensearch.timeseries.util.DiscoveryNodeFilterer;
7474
import org.opensearch.timeseries.util.ExceptionUtil;
7575
import org.opensearch.timeseries.util.MultiResponsesDelegateActionListener;
@@ -105,7 +105,7 @@ public AnomalyDetectorProfileRunner(
105105
}
106106
this.transportService = transportService;
107107
this.adTaskManager = adTaskManager;
108-
this.maxTotalEntitiesToTrack = AnomalyDetectorSettings.MAX_TOTAL_ENTITIES_TO_TRACK;
108+
this.maxTotalEntitiesToTrack = TimeSeriesSettings.MAX_TOTAL_ENTITIES_TO_TRACK;
109109
}
110110

111111
public void profile(String detectorId, ActionListener<DetectorProfile> listener, Set<DetectorProfileName> profilesToCollect) {

‎src/main/java/org/opensearch/ad/caching/CacheBuffer.java

+6-6
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,6 @@
2525
import org.apache.commons.lang.builder.HashCodeBuilder;
2626
import org.apache.logging.log4j.LogManager;
2727
import org.apache.logging.log4j.Logger;
28-
import org.opensearch.ad.MemoryTracker;
29-
import org.opensearch.ad.MemoryTracker.Origin;
3028
import org.opensearch.ad.ml.EntityModel;
3129
import org.opensearch.ad.ml.ModelState;
3230
import org.opensearch.ad.model.InitProgressProfile;
@@ -36,6 +34,8 @@
3634
import org.opensearch.ad.ratelimit.RequestPriority;
3735
import org.opensearch.ad.util.DateUtils;
3836
import org.opensearch.timeseries.ExpiringState;
37+
import org.opensearch.timeseries.MemoryTracker;
38+
import org.opensearch.timeseries.MemoryTracker.Origin;
3939

4040
/**
4141
* We use a layered cache to manage active entities’ states. We have a two-level
@@ -159,7 +159,7 @@ private void put(String entityModelId, ModelState<EntityModel> value, float prio
159159
// Since we have already considered them while allocating CacheBuffer,
160160
// skip bookkeeping.
161161
if (!sharedCacheEmpty()) {
162-
memoryTracker.consumeMemory(memoryConsumptionPerEntity, false, Origin.HC_DETECTOR);
162+
memoryTracker.consumeMemory(memoryConsumptionPerEntity, false, Origin.REAL_TIME_DETECTOR);
163163
}
164164
} else {
165165
update(entityModelId);
@@ -267,7 +267,7 @@ public ModelState<EntityModel> remove(String keyToRemove, boolean saveCheckpoint
267267
if (valueRemoved != null) {
268268
if (!reserved) {
269269
// release in shared memory
270-
memoryTracker.releaseMemory(memoryConsumptionPerEntity, false, Origin.HC_DETECTOR);
270+
memoryTracker.releaseMemory(memoryConsumptionPerEntity, false, Origin.REAL_TIME_DETECTOR);
271271
}
272272

273273
EntityModel modelRemoved = valueRemoved.getModel();
@@ -460,9 +460,9 @@ public void clear() {
460460
// not a problem as we are releasing memory in MemoryTracker.
461461
// The newly added one loses references and soon GC will collect it.
462462
// We have memory tracking correction to fix incorrect memory usage record.
463-
memoryTracker.releaseMemory(getReservedBytes(), true, Origin.HC_DETECTOR);
463+
memoryTracker.releaseMemory(getReservedBytes(), true, Origin.REAL_TIME_DETECTOR);
464464
if (!sharedCacheEmpty()) {
465-
memoryTracker.releaseMemory(getBytesInSharedCache(), false, Origin.HC_DETECTOR);
465+
memoryTracker.releaseMemory(getBytesInSharedCache(), false, Origin.REAL_TIME_DETECTOR);
466466
}
467467
items.clear();
468468
priorityTracker.clearPriority();

‎src/main/java/org/opensearch/ad/caching/PriorityCache.java

+8-8
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,8 @@
1111

1212
package org.opensearch.ad.caching;
1313

14-
import static org.opensearch.ad.settings.AnomalyDetectorSettings.DEDICATED_CACHE_SIZE;
15-
import static org.opensearch.ad.settings.AnomalyDetectorSettings.MODEL_MAX_SIZE_PERCENTAGE;
14+
import static org.opensearch.ad.settings.AnomalyDetectorSettings.AD_DEDICATED_CACHE_SIZE;
15+
import static org.opensearch.ad.settings.AnomalyDetectorSettings.AD_MODEL_MAX_SIZE_PERCENTAGE;
1616

1717
import java.time.Clock;
1818
import java.time.Duration;
@@ -38,8 +38,6 @@
3838
import org.apache.logging.log4j.LogManager;
3939
import org.apache.logging.log4j.Logger;
4040
import org.apache.logging.log4j.message.ParameterizedMessage;
41-
import org.opensearch.ad.MemoryTracker;
42-
import org.opensearch.ad.MemoryTracker.Origin;
4341
import org.opensearch.ad.ml.CheckpointDao;
4442
import org.opensearch.ad.ml.EntityModel;
4543
import org.opensearch.ad.ml.ModelManager.ModelType;
@@ -57,6 +55,8 @@
5755
import org.opensearch.core.action.ActionListener;
5856
import org.opensearch.core.common.Strings;
5957
import org.opensearch.threadpool.ThreadPool;
58+
import org.opensearch.timeseries.MemoryTracker;
59+
import org.opensearch.timeseries.MemoryTracker.Origin;
6060
import org.opensearch.timeseries.TimeSeriesAnalyticsPlugin;
6161
import org.opensearch.timeseries.common.exception.LimitExceededException;
6262
import org.opensearch.timeseries.common.exception.TimeSeriesException;
@@ -116,12 +116,12 @@ public PriorityCache(
116116

117117
this.activeEnities = new ConcurrentHashMap<>();
118118
this.dedicatedCacheSize = dedicatedCacheSize;
119-
clusterService.getClusterSettings().addSettingsUpdateConsumer(DEDICATED_CACHE_SIZE, (it) -> {
119+
clusterService.getClusterSettings().addSettingsUpdateConsumer(AD_DEDICATED_CACHE_SIZE, (it) -> {
120120
this.dedicatedCacheSize = it;
121121
this.setDedicatedCacheSizeListener();
122122
this.tryClearUpMemory();
123123
}, this::validateDedicatedCacheSize);
124-
clusterService.getClusterSettings().addSettingsUpdateConsumer(MODEL_MAX_SIZE_PERCENTAGE, it -> this.tryClearUpMemory());
124+
clusterService.getClusterSettings().addSettingsUpdateConsumer(AD_MODEL_MAX_SIZE_PERCENTAGE, it -> this.tryClearUpMemory());
125125

126126
this.memoryTracker = memoryTracker;
127127
this.maintenanceLock = new ReentrantLock();
@@ -461,7 +461,7 @@ private CacheBuffer computeBufferIfAbsent(AnomalyDetector detector, String detec
461461
if (buffer == null) {
462462
long requiredBytes = getRequiredMemory(detector, dedicatedCacheSize);
463463
if (memoryTracker.canAllocateReserved(requiredBytes)) {
464-
memoryTracker.consumeMemory(requiredBytes, true, Origin.HC_DETECTOR);
464+
memoryTracker.consumeMemory(requiredBytes, true, Origin.REAL_TIME_DETECTOR);
465465
long intervalSecs = detector.getIntervalInSeconds();
466466

467467
buffer = new CacheBuffer(
@@ -621,7 +621,7 @@ private void recalculateUsedMemory() {
621621
reserved += buffer.getReservedBytes();
622622
shared += buffer.getBytesInSharedCache();
623623
}
624-
memoryTracker.syncMemoryState(Origin.HC_DETECTOR, reserved + shared, reserved);
624+
memoryTracker.syncMemoryState(Origin.REAL_TIME_DETECTOR, reserved + shared, reserved);
625625
}
626626

627627
/**

‎src/main/java/org/opensearch/ad/ml/ModelManager.java

+2-3
Original file line numberDiff line numberDiff line change
@@ -33,17 +33,16 @@
3333
import org.apache.logging.log4j.Logger;
3434
import org.apache.logging.log4j.message.ParameterizedMessage;
3535
import org.opensearch.ad.DetectorModelSize;
36-
import org.opensearch.ad.MemoryTracker;
3736
import org.opensearch.ad.constant.ADCommonMessages;
3837
import org.opensearch.ad.feature.FeatureManager;
3938
import org.opensearch.ad.model.AnomalyDetector;
40-
import org.opensearch.ad.settings.AnomalyDetectorSettings;
4139
import org.opensearch.ad.util.DateUtils;
4240
import org.opensearch.cluster.service.ClusterService;
4341
import org.opensearch.common.settings.Setting;
4442
import org.opensearch.common.settings.Settings;
4543
import org.opensearch.common.unit.TimeValue;
4644
import org.opensearch.core.action.ActionListener;
45+
import org.opensearch.timeseries.MemoryTracker;
4746
import org.opensearch.timeseries.common.exception.ResourceNotFoundException;
4847
import org.opensearch.timeseries.ml.SingleStreamModelIdMapper;
4948
import org.opensearch.timeseries.model.Entity;
@@ -624,7 +623,7 @@ public List<ThresholdingResult> getPreviewResults(double[][] dataPoints, int shi
624623
.parallelExecutionEnabled(false)
625624
.compact(true)
626625
.precision(Precision.FLOAT_32)
627-
.boundingBoxCacheFraction(AnomalyDetectorSettings.BATCH_BOUNDING_BOX_CACHE_RATIO)
626+
.boundingBoxCacheFraction(TimeSeriesSettings.BATCH_BOUNDING_BOX_CACHE_RATIO)
628627
.shingleSize(shingleSize)
629628
.anomalyRate(1 - this.thresholdMinPvalue)
630629
.transformMethod(TransformMethod.NORMALIZE)

‎src/main/java/org/opensearch/ad/ml/TRCFMemoryAwareConcurrentHashmap.java

+4-4
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,8 @@
1313

1414
import java.util.concurrent.ConcurrentHashMap;
1515

16-
import org.opensearch.ad.MemoryTracker;
17-
import org.opensearch.ad.MemoryTracker.Origin;
16+
import org.opensearch.timeseries.MemoryTracker;
17+
import org.opensearch.timeseries.MemoryTracker.Origin;
1818

1919
import com.amazon.randomcutforest.parkservices.ThresholdedRandomCutForest;
2020

@@ -37,7 +37,7 @@ public ModelState<ThresholdedRandomCutForest> remove(Object key) {
3737
ModelState<ThresholdedRandomCutForest> deletedModelState = super.remove(key);
3838
if (deletedModelState != null && deletedModelState.getModel() != null) {
3939
long memoryToRelease = memoryTracker.estimateTRCFModelSize(deletedModelState.getModel());
40-
memoryTracker.releaseMemory(memoryToRelease, true, Origin.SINGLE_ENTITY_DETECTOR);
40+
memoryTracker.releaseMemory(memoryToRelease, true, Origin.REAL_TIME_DETECTOR);
4141
}
4242
return deletedModelState;
4343
}
@@ -47,7 +47,7 @@ public ModelState<ThresholdedRandomCutForest> put(K key, ModelState<ThresholdedR
4747
ModelState<ThresholdedRandomCutForest> previousAssociatedState = super.put(key, value);
4848
if (value != null && value.getModel() != null) {
4949
long memoryToConsume = memoryTracker.estimateTRCFModelSize(value.getModel());
50-
memoryTracker.consumeMemory(memoryToConsume, true, Origin.SINGLE_ENTITY_DETECTOR);
50+
memoryTracker.consumeMemory(memoryToConsume, true, Origin.REAL_TIME_DETECTOR);
5151
}
5252
return previousAssociatedState;
5353
}

‎src/main/java/org/opensearch/ad/ratelimit/BatchWorker.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,14 @@
1919
import org.apache.logging.log4j.LogManager;
2020
import org.apache.logging.log4j.Logger;
2121
import org.opensearch.action.support.ThreadedActionListener;
22-
import org.opensearch.ad.breaker.ADCircuitBreakerService;
2322
import org.opensearch.cluster.service.ClusterService;
2423
import org.opensearch.common.settings.Setting;
2524
import org.opensearch.common.settings.Settings;
2625
import org.opensearch.core.action.ActionListener;
2726
import org.opensearch.threadpool.ThreadPool;
2827
import org.opensearch.timeseries.NodeStateManager;
2928
import org.opensearch.timeseries.TimeSeriesAnalyticsPlugin;
29+
import org.opensearch.timeseries.breaker.CircuitBreakerService;
3030

3131
/**
3232
*
@@ -46,7 +46,7 @@ public BatchWorker(
4646
Setting<Float> maxHeapPercentForQueueSetting,
4747
ClusterService clusterService,
4848
Random random,
49-
ADCircuitBreakerService adCircuitBreakerService,
49+
CircuitBreakerService adCircuitBreakerService,
5050
ThreadPool threadPool,
5151
Settings settings,
5252
float maxQueuedTaskRatio,

‎src/main/java/org/opensearch/ad/ratelimit/CheckpointMaintainWorker.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,13 @@
2323

2424
import org.apache.logging.log4j.LogManager;
2525
import org.apache.logging.log4j.Logger;
26-
import org.opensearch.ad.breaker.ADCircuitBreakerService;
2726
import org.opensearch.ad.settings.AnomalyDetectorSettings;
2827
import org.opensearch.cluster.service.ClusterService;
2928
import org.opensearch.common.settings.Setting;
3029
import org.opensearch.common.settings.Settings;
3130
import org.opensearch.threadpool.ThreadPool;
3231
import org.opensearch.timeseries.NodeStateManager;
32+
import org.opensearch.timeseries.breaker.CircuitBreakerService;
3333

3434
public class CheckpointMaintainWorker extends ScheduledWorker<CheckpointMaintainRequest, CheckpointWriteRequest> {
3535
private static final Logger LOG = LogManager.getLogger(CheckpointMaintainWorker.class);
@@ -43,7 +43,7 @@ public CheckpointMaintainWorker(
4343
Setting<Float> maxHeapPercentForQueueSetting,
4444
ClusterService clusterService,
4545
Random random,
46-
ADCircuitBreakerService adCircuitBreakerService,
46+
CircuitBreakerService adCircuitBreakerService,
4747
ThreadPool threadPool,
4848
Settings settings,
4949
float maxQueuedTaskRatio,

‎src/main/java/org/opensearch/ad/ratelimit/CheckpointReadWorker.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import org.opensearch.action.get.MultiGetItemResponse;
3333
import org.opensearch.action.get.MultiGetRequest;
3434
import org.opensearch.action.get.MultiGetResponse;
35-
import org.opensearch.ad.breaker.ADCircuitBreakerService;
3635
import org.opensearch.ad.caching.CacheProvider;
3736
import org.opensearch.ad.constant.ADCommonName;
3837
import org.opensearch.ad.indices.ADIndex;
@@ -53,6 +52,7 @@
5352
import org.opensearch.threadpool.ThreadPool;
5453
import org.opensearch.timeseries.AnalysisType;
5554
import org.opensearch.timeseries.NodeStateManager;
55+
import org.opensearch.timeseries.breaker.CircuitBreakerService;
5656
import org.opensearch.timeseries.common.exception.EndRunException;
5757
import org.opensearch.timeseries.constant.CommonMessages;
5858
import org.opensearch.timeseries.model.Config;
@@ -91,7 +91,7 @@ public CheckpointReadWorker(
9191
Setting<Float> maxHeapPercentForQueueSetting,
9292
ClusterService clusterService,
9393
Random random,
94-
ADCircuitBreakerService adCircuitBreakerService,
94+
CircuitBreakerService adCircuitBreakerService,
9595
ThreadPool threadPool,
9696
Settings settings,
9797
float maxQueuedTaskRatio,

‎src/main/java/org/opensearch/ad/ratelimit/CheckpointWriteWorker.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030
import org.opensearch.action.bulk.BulkRequest;
3131
import org.opensearch.action.bulk.BulkResponse;
3232
import org.opensearch.action.update.UpdateRequest;
33-
import org.opensearch.ad.breaker.ADCircuitBreakerService;
3433
import org.opensearch.ad.ml.CheckpointDao;
3534
import org.opensearch.ad.ml.EntityModel;
3635
import org.opensearch.ad.ml.ModelState;
@@ -43,6 +42,7 @@
4342
import org.opensearch.threadpool.ThreadPool;
4443
import org.opensearch.timeseries.AnalysisType;
4544
import org.opensearch.timeseries.NodeStateManager;
45+
import org.opensearch.timeseries.breaker.CircuitBreakerService;
4646
import org.opensearch.timeseries.model.Config;
4747
import org.opensearch.timeseries.util.ExceptionUtil;
4848

@@ -60,7 +60,7 @@ public CheckpointWriteWorker(
6060
Setting<Float> maxHeapPercentForQueueSetting,
6161
ClusterService clusterService,
6262
Random random,
63-
ADCircuitBreakerService adCircuitBreakerService,
63+
CircuitBreakerService adCircuitBreakerService,
6464
ThreadPool threadPool,
6565
Settings settings,
6666
float maxQueuedTaskRatio,

‎src/main/java/org/opensearch/ad/ratelimit/ColdEntityWorker.java

+5-5
Original file line numberDiff line numberDiff line change
@@ -12,21 +12,21 @@
1212
package org.opensearch.ad.ratelimit;
1313

1414
import static org.opensearch.ad.settings.AnomalyDetectorSettings.AD_CHECKPOINT_READ_QUEUE_BATCH_SIZE;
15-
import static org.opensearch.ad.settings.AnomalyDetectorSettings.EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS;
15+
import static org.opensearch.ad.settings.AnomalyDetectorSettings.AD_EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS;
1616

1717
import java.time.Clock;
1818
import java.time.Duration;
1919
import java.util.List;
2020
import java.util.Random;
2121
import java.util.stream.Collectors;
2222

23-
import org.opensearch.ad.breaker.ADCircuitBreakerService;
2423
import org.opensearch.ad.settings.AnomalyDetectorSettings;
2524
import org.opensearch.cluster.service.ClusterService;
2625
import org.opensearch.common.settings.Setting;
2726
import org.opensearch.common.settings.Settings;
2827
import org.opensearch.threadpool.ThreadPool;
2928
import org.opensearch.timeseries.NodeStateManager;
29+
import org.opensearch.timeseries.breaker.CircuitBreakerService;
3030

3131
/**
3232
* A queue slowly releasing low-priority requests to CheckpointReadQueue
@@ -52,7 +52,7 @@ public ColdEntityWorker(
5252
Setting<Float> maxHeapPercentForQueueSetting,
5353
ClusterService clusterService,
5454
Random random,
55-
ADCircuitBreakerService adCircuitBreakerService,
55+
CircuitBreakerService adCircuitBreakerService,
5656
ThreadPool threadPool,
5757
Settings settings,
5858
float maxQueuedTaskRatio,
@@ -87,12 +87,12 @@ public ColdEntityWorker(
8787
this.batchSize = AD_CHECKPOINT_READ_QUEUE_BATCH_SIZE.get(settings);
8888
clusterService.getClusterSettings().addSettingsUpdateConsumer(AD_CHECKPOINT_READ_QUEUE_BATCH_SIZE, it -> this.batchSize = it);
8989

90-
this.expectedExecutionTimeInMilliSecsPerRequest = AnomalyDetectorSettings.EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS
90+
this.expectedExecutionTimeInMilliSecsPerRequest = AnomalyDetectorSettings.AD_EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS
9191
.get(settings);
9292
clusterService
9393
.getClusterSettings()
9494
.addSettingsUpdateConsumer(
95-
EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS,
95+
AD_EXPECTED_COLD_ENTITY_EXECUTION_TIME_IN_MILLISECS,
9696
it -> this.expectedExecutionTimeInMilliSecsPerRequest = it
9797
);
9898
}

‎src/main/java/org/opensearch/ad/ratelimit/ConcurrentWorker.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,13 @@
1919

2020
import org.apache.logging.log4j.LogManager;
2121
import org.apache.logging.log4j.Logger;
22-
import org.opensearch.ad.breaker.ADCircuitBreakerService;
2322
import org.opensearch.cluster.service.ClusterService;
2423
import org.opensearch.common.settings.Setting;
2524
import org.opensearch.common.settings.Settings;
2625
import org.opensearch.threadpool.ThreadPool;
2726
import org.opensearch.timeseries.NodeStateManager;
2827
import org.opensearch.timeseries.TimeSeriesAnalyticsPlugin;
28+
import org.opensearch.timeseries.breaker.CircuitBreakerService;
2929

3030
/**
3131
* A queue to run concurrent requests (either batch or single request).
@@ -74,7 +74,7 @@ public ConcurrentWorker(
7474
Setting<Float> maxHeapPercentForQueueSetting,
7575
ClusterService clusterService,
7676
Random random,
77-
ADCircuitBreakerService adCircuitBreakerService,
77+
CircuitBreakerService adCircuitBreakerService,
7878
ThreadPool threadPool,
7979
Settings settings,
8080
float maxQueuedTaskRatio,

0 commit comments

Comments
 (0)