OfferStrategy { protected OfferResult evaluate(Protos.Offer offer) { final Optional<OfferRule> decline = acceptanceRules.stream().filter(offerRule -> offerRule.rule.accepts(offer)).limit(1).findFirst(); if (decline.isPresent()) { return OfferResult.decline(decline.get().declineReason); } LOGGER.info("Accepted offer: " + offer.getHostname()); return OfferResult.accept(); } protected  OfferStrategy(Configuration configuration, ClusterState clusterState);  static final int DUMMY_PORT; }
@Test public void willDeclineIfHostIsAlreadyRunningTask() throws Exception { when(clusterState.getTaskList()).thenReturn(singletonList(createTask("host1"))); final OfferStrategyNormal.OfferResult result = offerStrategy.evaluate(validOffer("host1")); assertFalse(result.acceptable); assertEquals("Host already running task", result.reason.get()); }
@Test public void willDeclineIfClusterSizeIsFulfilled() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"), createTask("host3"))); when(configuration.getElasticsearchNodes()).thenReturn(3); final OfferStrategyNormal.OfferResult result = offerStrategy.evaluate(validOffer("host4")); assertFalse(result.acceptable); assertEquals("Cluster size already fulfilled", result.reason.get()); }
@Test public void willDeclineIfOfferDoesNotHaveTwoPorts() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(9200, 9200, configuration.getFrameworkRole())) .build()); assertFalse(offerResult.acceptable); assertEquals("Offer did not have 2 ports", offerResult.reason.get()); }
@Test public void shouldDeclineIfUserPortsNotAvailable() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); when(configuration.getElasticsearchPorts()).thenReturn(Arrays.asList(9200, 9300)); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(31000, 32000, configuration.getFrameworkRole())) .build()); assertFalse(offerResult.acceptable); assertEquals("The offer does not contain the user specified ports", offerResult.reason.get()); }
@Test public void shouldDeclineIfOneOfUserPortsNotAvailable() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); when(configuration.getElasticsearchPorts()).thenReturn(Arrays.asList(31000, 9300)); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(31000, 32000, configuration.getFrameworkRole())) .build()); assertFalse(offerResult.acceptable); assertEquals("The offer does not contain the user specified ports", offerResult.reason.get()); }
@Test public void shouldAcceptIfUserPortsAreAvailable() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); when(configuration.getElasticsearchPorts()).thenReturn(Arrays.asList(9200, 9300)); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(31000, 32000, configuration.getFrameworkRole())) .addResources(portRange(9200, 9300, configuration.getFrameworkRole())) .addResources(cpus(configuration.getCpus(), configuration.getFrameworkRole())) .addResources(mem(configuration.getMem(), configuration.getFrameworkRole())) .addResources(disk(configuration.getDisk(), configuration.getFrameworkRole())) .build()); assertTrue(offerResult.acceptable); }
@Test public void willDeclineIfOfferDoesNotEnoughCpu() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); when(configuration.getCpus()).thenReturn(1.0); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(9200, 9200, configuration.getFrameworkRole())) .addResources(portRange(9300, 9300, configuration.getFrameworkRole())) .addResources(cpus(0.1, configuration.getFrameworkRole())) .build()); assertFalse(offerResult.acceptable); assertEquals("Offer did not have enough CPU resources", offerResult.reason.get()); }
@Test public void willDeclineIfOfferDoesNotEnoughMem() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); when(configuration.getMem()).thenReturn(100.0); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(9200, 9200, configuration.getFrameworkRole())) .addResources(portRange(9300, 9300, configuration.getFrameworkRole())) .addResources(cpus(10.0, configuration.getFrameworkRole())) .addResources(mem(10, configuration.getFrameworkRole())) .build()); assertFalse(offerResult.acceptable); assertEquals("Offer did not have enough RAM resources", offerResult.reason.get()); }
@Test public void willDeclineIfOfferDoesNotEnoughDisk() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); when(configuration.getDisk()).thenReturn(100.0); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(9200, 9200, configuration.getFrameworkRole())) .addResources(portRange(9300, 9300, configuration.getFrameworkRole())) .addResources(cpus(10.0, configuration.getFrameworkRole())) .addResources(mem(100, configuration.getFrameworkRole())) .addResources(disk(10, configuration.getFrameworkRole())) .build()); assertFalse(offerResult.acceptable); assertEquals("Offer did not have enough disk resources", offerResult.reason.get()); }
@Test public void willAcceptValidOffer() throws Exception { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(9200, 9200, configuration.getFrameworkRole())) .addResources(portRange(9300, 9300, configuration.getFrameworkRole())) .addResources(cpus(configuration.getCpus(), configuration.getFrameworkRole())) .addResources(mem(configuration.getMem(), configuration.getFrameworkRole())) .addResources(disk(configuration.getDisk(), configuration.getFrameworkRole())) .build()); assertTrue(offerResult.acceptable); assertFalse(offerResult.reason.isPresent()); }
@Test public void shouldDeclineWhenHostIsUnresolveable() throws InvalidProtocolBufferException { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(9200, 9200, configuration.getFrameworkRole())) .addResources(portRange(9300, 9300, configuration.getFrameworkRole())) .addResources(cpus(configuration.getCpus(), configuration.getFrameworkRole())) .addResources(mem(configuration.getMem(), configuration.getFrameworkRole())) .addResources(disk(configuration.getDisk(), configuration.getFrameworkRole())) .setHostname("NonResolvableHostname") .build()); assertFalse(offerResult.acceptable); }
@Test public void shouldAcceptWhenHostIsResolveable() throws InvalidProtocolBufferException { when(clusterState.getTaskList()).thenReturn(asList(createTask("host1"), createTask("host2"))); when(configuration.getElasticsearchNodes()).thenReturn(3); final OfferStrategyNormal.OfferResult offerResult = offerStrategy.evaluate(baseOfferBuilder("host3") .addResources(portRange(9200, 9200, configuration.getFrameworkRole())) .addResources(portRange(9300, 9300, configuration.getFrameworkRole())) .addResources(cpus(configuration.getCpus(), configuration.getFrameworkRole())) .addResources(mem(configuration.getMem(), configuration.getFrameworkRole())) .addResources(disk(configuration.getDisk(), configuration.getFrameworkRole())) .build()); assertTrue(offerResult.acceptable); }