DefaultAppScaleManager implements AppScaleManager { Observable<String> v3LiveStreamTargetUpdates() { return v3JobOperations.observeJobs() .filter(event -> { if (event instanceof JobUpdateEvent) { JobUpdateEvent jobUpdateEvent = (JobUpdateEvent) event; return jobUpdateEvent.getCurrent().getStatus().getState() != JobState.Finished; } return false; }) .cast(JobUpdateEvent.class) .flatMap(event -> appScalePolicyStore.retrievePoliciesForJob(event.getCurrent().getId()) .filter(autoScalingPolicy -> shouldRefreshScalableTargetForJob(autoScalingPolicy.getJobId(), getJobScalingConstraints(autoScalingPolicy.getRefId(), autoScalingPolicy.getJobId()))) .map(this::sendUpdateTargetAction) .map(AppScaleAction::getJobId) .doOnError(e -> logger.error("Exception in v3LiveStreamTargetUpdates -> ", e)) .onErrorResumeNext(e -> Observable.empty()) ); } @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 checkV3LiveStreamTargetUpdates() throws Exception { String jobIdOne = UUID.randomUUID().toString(); String jobIdTwo = UUID.randomUUID().toString(); InMemoryPolicyStore policyStore = new InMemoryPolicyStore(); V3JobOperations v3JobOperations = mockV3Operations(jobIdOne, jobIdTwo); AppScaleClientWithScalingPolicyConstraints appScalingClient = new AppScaleClientWithScalingPolicyConstraints(); DefaultAppScaleManager appScaleManager = new DefaultAppScaleManager(policyStore, new AutoScalingPolicyTests.MockAlarmClient(), appScalingClient, v3JobOperations, new DefaultRegistry(), AutoScalingPolicyTests.mockAppScaleManagerConfiguration(), Schedulers.immediate(), mock(TitusRuntime.class)); List<String> refIds = submitTwoJobs(appScaleManager, jobIdOne, jobIdTwo, policyStore); Assertions.assertThat(refIds.size()).isEqualTo(2); CountDownLatch latch = new CountDownLatch(1); Observable<String> jobIdTargetUpdates = appScaleManager.v3LiveStreamTargetUpdates(); List<String> targetsUpdated = new ArrayList<>(); jobIdTargetUpdates.subscribe(targetUpdated -> { log.info("Got ScalableTarget to be updated {}", targetUpdated); Assertions.assertThat(targetUpdated).isEqualTo(jobIdTwo); targetsUpdated.add(targetUpdated); latch.countDown(); }, e -> log.error("Error in v2 live stream for scalable target update", e), () -> log.info("Completed")); latch.await(60, TimeUnit.SECONDS); AutoScalingPolicyTests.waitForCondition(() -> { JobScalingConstraints jpc = appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdTwo); return jpc != null && jpc.getMinCapacity() == 5 && jpc.getMaxCapacity() == 15; }); Assertions.assertThat(targetsUpdated.size()).isEqualTo(1); Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdOne).getMinCapacity()).isEqualTo(1); Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdOne).getMaxCapacity()).isEqualTo(10); Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdTwo).getMinCapacity()).isEqualTo(5); Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdTwo).getMaxCapacity()).isEqualTo(15); }