DefaultAgentManagementService implements AgentManagementService { @Override public Completable updateInstanceGroupAttributes(String instanceGroupId, Map<String, String> attributes) { return agentCache.updateInstanceGroupStore( instanceGroupId, instanceGroup -> { Map<String, String> updatedAttributes = CollectionsExt.merge(instanceGroup.getAttributes(), attributes); return instanceGroup.toBuilder().withAttributes(updatedAttributes).build(); } ).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 testUpdateInstanceGroupAttributes() { ExtTestSubscriber<Object> testSubscriber = new ExtTestSubscriber<>(); AgentInstanceGroup instanceGroup = instanceGroups.get(0); String instanceGroupId = instanceGroup.getId(); Map<String, String> attributes = Collections.singletonMap("a", "1"); when(agentCache.updateInstanceGroupStore(eq(instanceGroupId), any())) .thenReturn(Single.just(instanceGroup.toBuilder().withAttributes(attributes).build())); service.updateInstanceGroupAttributes(instanceGroupId, attributes).toObservable().subscribe(testSubscriber); verify(agentCache, times(1)).updateInstanceGroupStore(eq(instanceGroupId), any()); }