ConfigurationNodeAttributesProvider extends NodeAttributesProvider { @VisibleForTesting public Set<NodeAttribute> parseAttributes(String config) throws IOException { if (Strings.isNullOrEmpty(config)) { return ImmutableSet.of(); } Set<NodeAttribute> attributeSet = new HashSet<>(); String[] attributeStrs = config.split(NODE_ATTRIBUTES_DELIMITER); for (String attributeStr : attributeStrs) { String[] fields = attributeStr.split(NODE_ATTRIBUTE_DELIMITER); if (fields.length != 3) { throw new IOException("Invalid value for " + YarnConfiguration.NM_PROVIDER_CONFIGURED_NODE_ATTRIBUTES + "=" + config); } if (fields[0].contains("/")) { throw new IOException("Node attribute set in " + YarnConfiguration.NM_PROVIDER_CONFIGURED_NODE_ATTRIBUTES + " should not contain any prefix."); } if (!EnumUtils.isValidEnum(NodeAttributeType.class, fields[1])) { throw new IOException("Invalid node attribute type: " + fields[1] + ", valid values are " + Arrays.asList(NodeAttributeType.values())); } NodeAttribute na = NodeAttribute.newInstance( NodeAttribute.PREFIX_DISTRIBUTED, fields[0], NodeAttributeType.valueOf(fields[1]), fields[2]); if (!attributeSet.add(na)) { throw new IOException("Ambiguous node attribute is found: " + na.toString() + ", a same attribute already exists"); } } try { NodeLabelUtil.validateNodeAttributes(attributeSet); } catch (IOException e) { throw new IOException("Node attributes set by configuration property: " + YarnConfiguration.NM_PROVIDER_CONFIGURED_NODE_ATTRIBUTES + " is not valid. Detail message: " + e.getMessage()); } return attributeSet; } ConfigurationNodeAttributesProvider(); @VisibleForTesting Set<NodeAttribute> parseAttributes(String config); @Override TimerTask createTimerTask();  }
@Test(timeout=30000L) public void testNodeAttributesFetchInterval() throws IOException, InterruptedException { Set<NodeAttribute> expectedAttributes1 = new HashSet<>(); expectedAttributes1.add(NodeAttribute .newInstance("test.io", "host", NodeAttributeType.STRING, "host1")); Configuration conf = new Configuration(); conf.setLong( YarnConfiguration.NM_NODE_ATTRIBUTES_PROVIDER_FETCH_INTERVAL_MS, 1000); ConfigurationNodeAttributesProvider spyProvider = Mockito.spy(nodeAttributesProvider); Mockito.when(spyProvider.parseAttributes(Mockito.any())) .thenReturn(expectedAttributes1); spyProvider.init(conf); spyProvider.start(); Assert.assertEquals(expectedAttributes1, spyProvider.getDescriptors()); Set<NodeAttribute> expectedAttributes2 = new HashSet<>(); expectedAttributes2.add(NodeAttribute .newInstance("test.io", "os", NodeAttributeType.STRING, "windows")); Mockito.when(spyProvider.parseAttributes(Mockito.any())) .thenReturn(expectedAttributes2); ArrayList<String> keysMet = new ArrayList<>(); int numOfOldValue = 0; int numOfNewValue = 0; int times=5; while(times>0) { Set<NodeAttribute> current = spyProvider.getDescriptors(); Assert.assertEquals(1, current.size()); String attributeName = current.iterator().next().getAttributeKey().getAttributeName(); if ("host".equals(attributeName)){ numOfOldValue++; } else if ("os".equals(attributeName)) { numOfNewValue++; } Thread.sleep(500); times--; } Assert.assertEquals(5, numOfNewValue + numOfOldValue); Assert.assertTrue(numOfOldValue > 0); Assert.assertTrue(numOfNewValue > 0); }
@Test public void testDisableFetchNodeAttributes() throws IOException, InterruptedException { Set<NodeAttribute> expectedAttributes1 = new HashSet<>(); expectedAttributes1.add(NodeAttribute .newInstance("test.io", "host", NodeAttributeType.STRING, "host1")); Configuration conf = new Configuration(); conf.setLong( YarnConfiguration.NM_NODE_ATTRIBUTES_PROVIDER_FETCH_INTERVAL_MS, -1); ConfigurationNodeAttributesProvider spyProvider = Mockito.spy(nodeAttributesProvider); Mockito.when(spyProvider.parseAttributes(Mockito.any())) .thenReturn(expectedAttributes1); spyProvider.init(conf); spyProvider.start(); Assert.assertEquals(expectedAttributes1, spyProvider.getDescriptors()); Set<NodeAttribute> expectedAttributes2 = new HashSet<>(); expectedAttributes2.add(NodeAttribute .newInstance("test.io", "os", NodeAttributeType.STRING, "windows")); Mockito.when(spyProvider.parseAttributes(Mockito.anyString())) .thenReturn(expectedAttributes2); try { GenericTestUtils.waitFor(() -> { Set<NodeAttribute> attributes = spyProvider.getDescriptors(); return "os".equalsIgnoreCase(attributes .iterator().next().getAttributeKey().getAttributeName()); }, 500, 1000); } catch (Exception e) { Assert.assertTrue(e instanceof TimeoutException); return; } Assert.fail("Expecting a failure in previous check!"); }
@Test public void testParseConfiguration() throws IOException { String attributesStr = "hostname,STRING,host1234:uptime,STRING,321543"; Set<NodeAttribute> attributes = nodeAttributesProvider .parseAttributes(attributesStr); Assert.assertEquals(2, attributes.size()); Iterator<NodeAttribute> ait = attributes.iterator(); while(ait.hasNext()) { NodeAttribute attr = ait.next(); NodeAttributeKey at = attr.getAttributeKey(); if (at.getAttributeName().equals("hostname")) { Assert.assertEquals("hostname", at.getAttributeName()); Assert.assertEquals(NodeAttribute.PREFIX_DISTRIBUTED, at.getAttributePrefix()); Assert.assertEquals(NodeAttributeType.STRING, attr.getAttributeType()); Assert.assertEquals("host1234", attr.getAttributeValue()); } else if (at.getAttributeName().equals("uptime")) { Assert.assertEquals("uptime", at.getAttributeName()); Assert.assertEquals(NodeAttribute.PREFIX_DISTRIBUTED, at.getAttributePrefix()); Assert.assertEquals(NodeAttributeType.STRING, attr.getAttributeType()); Assert.assertEquals("321543", attr.getAttributeValue()); } else { Assert.fail("Unexpected attribute"); } } attributesStr = "hostname,host1234"; try { nodeAttributesProvider.parseAttributes(attributesStr); Assert.fail("Expecting a parsing failure"); } catch (IOException e) { Assert.assertNotNull(e); Assert.assertTrue(e.getMessage().contains("Invalid value")); } attributesStr = "prefix/hostname,STRING,host1234"; try { nodeAttributesProvider.parseAttributes(attributesStr); Assert.fail("Expecting a parsing failure"); } catch (IOException e) { Assert.assertNotNull(e); Assert.assertTrue(e.getMessage() .contains("should not contain any prefix.")); } attributesStr = "hostname,T,host1234"; try { nodeAttributesProvider.parseAttributes(attributesStr); Assert.fail("Expecting a parsing failure"); } catch (IOException e) { e.printStackTrace(); Assert.assertNotNull(e); Assert.assertTrue(e.getMessage() .contains("Invalid node attribute type")); } }