DefaultAgentManagementService implements AgentManagementService { @Override public List<Pair<AgentInstanceGroup, List<AgentInstance>>> findAgentInstances(Predicate<Pair<AgentInstanceGroup, AgentInstance>> filter) { return agentCache.getInstanceGroups().stream() .map(instanceGroup -> { List<AgentInstance> matchingAgents = agentCache.getAgentInstances(instanceGroup.getId()).stream() .filter(agent -> filter.test(Pair.of(instanceGroup, agent))) .collect(Collectors.toList()); return Pair.of(instanceGroup, matchingAgents); }) .filter(pair -> !pair.getRight().isEmpty()) .collect(Collectors.toList()); } @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 testFindAgentInstances() { List<AgentInstanceGroup> serverGroups = service.getInstanceGroups(); assertThat(serverGroups).hasSize(2); String serverGroupId0 = serverGroups.get(0).getId(); List<Pair<AgentInstanceGroup, List<AgentInstance>>> result = service.findAgentInstances(pair -> pair.getLeft().getId().equals(serverGroupId0)); assertThat(result).hasSize(1); assertThat(result.get(0).getLeft().getId()).isEqualTo(serverGroups.get(0).getId()); }