DefaultAgentManagementService implements AgentManagementService { @Override public Completable scaleUp(String instanceGroupId, int scaleUpCount) { if (!configuration.isInstanceGroupUpdateCapacityEnabled()) { return Completable.complete(); } return Observable.fromCallable(() -> agentCache.getInstanceGroup(instanceGroupId)) .flatMap(instanceGroup -> { Completable cloudUpdate = instanceCloudConnector.scaleUp(instanceGroup.getId(), scaleUpCount); Completable cacheUpdate = agentCache.updateInstanceGroupStoreAndSyncCloud( instanceGroup.getId(), ig -> ig.toBuilder().withDesired(ig.getDesired() + scaleUpCount).build() ).toCompletable(); return cloudUpdate.concatWith(cacheUpdate).toObservable(); } ) .toCompletable(); } @Inject  DefaultAgentManagementService(AgentManagementConfiguration configuration,
                                         @Named(KUBE_SCHEDULER_INSTANCE_GROUP_PREDICATE) Predicate<AgentInstanceGroup> kubeSchedulerInstanceGroupPredicate,
                                         InstanceCloudConnector instanceCloudConnector,
                                         AgentCache AgentCache,
                                         ServerInfoResolver serverInfoResolver,
                                         TitusRuntime titusRuntime); @Activator void enterActiveMode(); @PreDestroy void shutdown(); @Override boolean isOwnedByFenzo(AgentInstanceGroup instanceGroup); @Override boolean isOwnedByFenzo(AgentInstance instance); @Override List<AgentInstanceGroup> getInstanceGroups(); @Override AgentInstanceGroup getInstanceGroup(String instanceGroupId); @Override Optional<AgentInstanceGroup> findInstanceGroup(String instanceGroupId); @Override AgentInstance getAgentInstance(String instanceId); @Override Optional<AgentInstance> findAgentInstance(String instanceId); @Override List<AgentInstance> getAgentInstances(String instanceGroupId); @Override List<Pair<AgentInstanceGroup, List<AgentInstance>>> findAgentInstances(Predicate<Pair<AgentInstanceGroup, AgentInstance>> filter); @Override ResourceDimension getResourceLimits(String instanceType); @Override Optional<ResourceDimension> findResourceLimits(String instanceType); @Override Completable updateInstanceGroupTier(String instanceGroupId, Tier tier); @Override Completable updateInstanceGroupLifecycle(String instanceGroupId, InstanceGroupLifecycleStatus instanceGroupLifecycleStatus); @Override Completable updateInstanceGroupAttributes(String instanceGroupId, Map<String, String> attributes); @Override Completable deleteInstanceGroupAttributes(String instanceGroupId, Set<String> keys); @Override Completable updateAgentInstanceAttributes(String instanceId, Map<String, String> attributes); @Override Completable deleteAgentInstanceAttributes(String instanceId, Set<String> keys); @Override Completable updateCapacity(String instanceGroupId, Optional<Integer> min, Optional<Integer> desired); @Override Completable scaleUp(String instanceGroupId, int scaleUpCount); @Override Observable<List<Either<Boolean, Throwable>>> terminateAgents(String instanceGroupId, List<String> instanceIds, boolean shrink); @Override Observable<AgentEvent> events(boolean includeSnapshot); static final String KUBE_SCHEDULER_INSTANCE_GROUP_PREDICATE; }
@Test public void testScaleUp() { ExtTestSubscriber<Object> testSubscriber = new ExtTestSubscriber<>(); AgentInstanceGroup instanceGroup = instanceGroups.get(0); String instanceGroupId = instanceGroup.getId(); when(agentCache.updateInstanceGroupStoreAndSyncCloud(eq(instanceGroupId), any())) .thenReturn(Single.just(instanceGroup)); service.scaleUp(instanceGroup.getId(), 500).toObservable().subscribe(testSubscriber); verify(connector, times(1)).scaleUp(instanceGroup.getId(), 500); verify(agentCache, times(1)).updateInstanceGroupStoreAndSyncCloud(eq(instanceGroupId), any()); }