TrafficController { public void bootstrap(String device, int rootBandwidthMbit, int yarnBandwidthMbit) throws ResourceHandlerException { if (device == null) { throw new ResourceHandlerException("device cannot be null!"); } String tmpDirBase = conf.get("hadoop.tmp.dir"); if (tmpDirBase == null) { throw new ResourceHandlerException("hadoop.tmp.dir not set!"); } tmpDirPath = tmpDirBase + "/nm-tc-rules"; File tmpDir = new File(tmpDirPath); if (!(tmpDir.exists() || tmpDir.mkdirs())) { LOG.warn("Unable to create directory: " + tmpDirPath); throw new ResourceHandlerException("Unable to create directory: " + tmpDirPath); } this.device = device; this.rootBandwidthMbit = rootBandwidthMbit; this.yarnBandwidthMbit = yarnBandwidthMbit; defaultClassBandwidthMbit = (rootBandwidthMbit - yarnBandwidthMbit) <= 0 ? rootBandwidthMbit : (rootBandwidthMbit - yarnBandwidthMbit); boolean recoveryEnabled = conf.getBoolean(YarnConfiguration .NM_RECOVERY_ENABLED, YarnConfiguration.DEFAULT_NM_RECOVERY_ENABLED); String state = null; if (!recoveryEnabled) { LOG.info("NM recovery is not enabled. We'll wipe tc state before proceeding."); } else { state = readState(); if (checkIfAlreadyBootstrapped(state)) { LOG.info("TC configuration is already in place. Not wiping state."); reacquireContainerClasses(state); return; } else { LOG.info("TC configuration is incomplete. Wiping tc state before proceeding"); } } wipeState(); initializeState(); } TrafficController(Configuration conf, PrivilegedOperationExecutor exec); void bootstrap(String device, int rootBandwidthMbit, int
      yarnBandwidthMbit); Map<Integer, Integer> readStats(); int getNextClassId(); void releaseClassId(int classId); String getStringForNetClsClassId(int classId); int getClassIdFromFileContents(String input);  }
@Test public void testBootstrapRecoveryDisabled() { conf.setBoolean(YarnConfiguration.NM_RECOVERY_ENABLED, false); TrafficController trafficController = new TrafficController(conf, privilegedOperationExecutorMock); try { trafficController .bootstrap(DEVICE, ROOT_BANDWIDTH_MBIT, YARN_BANDWIDTH_MBIT); ArgumentCaptor<PrivilegedOperation> opCaptor = ArgumentCaptor.forClass (PrivilegedOperation.class); verify(privilegedOperationExecutorMock, times(2)) .executePrivilegedOperation(opCaptor.capture(), eq(false)); List<PrivilegedOperation> ops = opCaptor.getAllValues(); verifyTrafficControlOperation(ops.get(0), PrivilegedOperation.OperationType.TC_MODIFY_STATE, Arrays.asList(WIPE_STATE_CMD)); verifyTrafficControlOperation(ops.get(1), PrivilegedOperation.OperationType.TC_MODIFY_STATE, Arrays.asList(ADD_ROOT_QDISC_CMD, ADD_CGROUP_FILTER_CMD, ADD_ROOT_CLASS_CMD, ADD_DEFAULT_CLASS_CMD, ADD_YARN_CLASS_CMD)); } catch (ResourceHandlerException | PrivilegedOperationException | IOException e) { LOG.error("Unexpected exception: " + e); Assert.fail("Caught unexpected exception: " + e.getClass().getSimpleName()); } }
@Test public void testBootstrapRecoveryEnabled() { conf.setBoolean(YarnConfiguration.NM_RECOVERY_ENABLED, true); TrafficController trafficController = new TrafficController(conf, privilegedOperationExecutorMock); try { when(privilegedOperationExecutorMock.executePrivilegedOperation( any(PrivilegedOperation.class), eq(true))) .thenReturn(DEFAULT_TC_STATE_EXAMPLE); trafficController .bootstrap(DEVICE, ROOT_BANDWIDTH_MBIT, YARN_BANDWIDTH_MBIT); ArgumentCaptor<PrivilegedOperation> readOpCaptor = ArgumentCaptor.forClass (PrivilegedOperation.class); verify(privilegedOperationExecutorMock, times(1)) .executePrivilegedOperation(readOpCaptor.capture(), eq(true)); List<PrivilegedOperation> readOps = readOpCaptor.getAllValues(); verifyTrafficControlOperation(readOps.get(0), PrivilegedOperation.OperationType.TC_READ_STATE, Arrays.asList(READ_QDISC_CMD, READ_FILTER_CMD, READ_CLASS_CMD)); ArgumentCaptor<PrivilegedOperation> writeOpCaptor = ArgumentCaptor .forClass(PrivilegedOperation.class); verify(privilegedOperationExecutorMock, times(2)) .executePrivilegedOperation(writeOpCaptor.capture(), eq(false)); List<PrivilegedOperation> writeOps = writeOpCaptor.getAllValues(); verifyTrafficControlOperation(writeOps.get(0), PrivilegedOperation.OperationType.TC_MODIFY_STATE, Arrays.asList(WIPE_STATE_CMD)); verifyTrafficControlOperation(writeOps.get(1), PrivilegedOperation.OperationType.TC_MODIFY_STATE, Arrays.asList(ADD_ROOT_QDISC_CMD, ADD_CGROUP_FILTER_CMD, ADD_ROOT_CLASS_CMD, ADD_DEFAULT_CLASS_CMD, ADD_YARN_CLASS_CMD)); } catch (ResourceHandlerException | PrivilegedOperationException | IOException e) { LOG.error("Unexpected exception: " + e); Assert.fail("Caught unexpected exception: " + e.getClass().getSimpleName()); } }
@Test public void testInvalidBuilder() { conf.setBoolean(YarnConfiguration.NM_RECOVERY_ENABLED, false); TrafficController trafficController = new TrafficController(conf, privilegedOperationExecutorMock); try { trafficController .bootstrap(DEVICE, ROOT_BANDWIDTH_MBIT, YARN_BANDWIDTH_MBIT); try { TrafficController.BatchBuilder invalidBuilder = trafficController. new BatchBuilder( PrivilegedOperation.OperationType.ADD_PID_TO_CGROUP); Assert.fail("Invalid builder check failed!"); } catch (ResourceHandlerException e) { } } catch (ResourceHandlerException e) { LOG.error("Unexpected exception: " + e); Assert.fail("Caught unexpected exception: " + e.getClass().getSimpleName()); } }