DefaultAppScaleManager implements AppScaleManager { @Override public Observable<String> createAutoScalingPolicy(AutoScalingPolicy autoScalingPolicy) { if (autoScalingPolicy.getJobId() == null || autoScalingPolicy.getPolicyConfiguration() == null) { return Observable.error(AutoScalePolicyException.invalidScalingPolicy(autoScalingPolicy.getRefId(), String.format("JobID Or PolicyConfiguration missing for %s", autoScalingPolicy.getRefId()))); } return appScalePolicyStore.storePolicy(autoScalingPolicy) .map(policyRefId -> { addScalableTargetIfNew(autoScalingPolicy.getJobId()); AutoScalingPolicy newPolicy = AutoScalingPolicy.newBuilder().withAutoScalingPolicy(autoScalingPolicy).withRefId(policyRefId).build(); sendCreatePolicyAction(newPolicy); metrics.reportPolicyStatusTransition(newPolicy, PolicyStatus.Pending); return policyRefId; }); } @Inject  DefaultAppScaleManager(AppScalePolicyStore appScalePolicyStore, CloudAlarmClient cloudAlarmClient,
                                  AppAutoScalingClient applicationAutoScalingClient,
                                  V3JobOperations v3JobOperations,
                                  Registry registry,
                                  AppScaleManagerConfiguration appScaleManagerConfiguration,
                                  TitusRuntime titusRuntime); private  DefaultAppScaleManager(AppScalePolicyStore appScalePolicyStore, CloudAlarmClient cloudAlarmClient,
                                   AppAutoScalingClient applicationAutoScalingClient,
                                   V3JobOperations v3JobOperations,
                                   Registry registry,
                                   AppScaleManagerConfiguration appScaleManagerConfiguration,
                                   ExecutorService awsInteractionExecutor,
                                   TitusRuntime titusRuntime); @VisibleForTesting  DefaultAppScaleManager(AppScalePolicyStore appScalePolicyStore, CloudAlarmClient cloudAlarmClient,
                                  AppAutoScalingClient applicationAutoScalingClient,
                                  V3JobOperations v3JobOperations,
                                  Registry registry,
                                  AppScaleManagerConfiguration appScaleManagerConfiguration,
                                  Scheduler awsInteractionScheduler,
                                  TitusRuntime titusRuntime); @Activator Completable enterActiveMode(); @PreDestroy void shutdown(); @Override Observable<String> createAutoScalingPolicy(AutoScalingPolicy autoScalingPolicy); @Override Completable updateAutoScalingPolicy(AutoScalingPolicy autoScalingPolicy); @Override Observable<AutoScalingPolicy> getScalingPoliciesForJob(String jobId); @Override Observable<AutoScalingPolicy> getScalingPolicy(String policyRefId); @Override Observable<AutoScalingPolicy> getAllScalingPolicies(); @Override Completable removeAutoScalingPolicy(String policyRefId); static final long INITIAL_RETRY_DELAY_SEC; static final long MAX_RETRY_DELAY_SEC; }
@Test public void checkBulkPolicyCreations() { AutoScalingPolicyTests.MockAlarmClient mockAlarmClient = new AutoScalingPolicyTests.MockAlarmClient(); AutoScalingPolicyTests.MockAppAutoScalingClient mockAppAutoScalingClient = new AutoScalingPolicyTests.MockAppAutoScalingClient(); InMemoryPolicyStore policyStore = new InMemoryPolicyStore(); final int totalJobs = 300; List<String> jobIds = new ArrayList<>(totalJobs); for (int i = 0; i < totalJobs; i++) { String jobId = UUID.randomUUID().toString(); jobIds.add(jobId); } V3JobOperations v3JobOperations = mockV3OperationsForJobs(jobIds); DefaultAppScaleManager appScaleManager = new DefaultAppScaleManager(policyStore, mockAlarmClient, mockAppAutoScalingClient, v3JobOperations, new DefaultRegistry(), AutoScalingPolicyTests.mockAppScaleManagerConfiguration(), Schedulers.computation(), mock(TitusRuntime.class)); final CountDownLatch latch = new CountDownLatch(totalJobs); for (int i = 0; i < totalJobs; i++) { final AutoScalingPolicy autoScalingPolicy = AutoScalingPolicyTests.buildStepScalingPolicy(jobIds.get(i)); appScaleManager.createAutoScalingPolicy(autoScalingPolicy).subscribe(pid -> { log.info("Policy created {}", pid); latch.countDown(); }); } await().timeout(30, TimeUnit.SECONDS).until(() -> mockAppAutoScalingClient.getNumPolicies() == totalJobs); }