@Test public void testRerouteInactivePinnedFlowsOneFailedSegment() { pinnedFlow.setStatus(FlowStatus.DOWN); for (FlowPath flowPath : pinnedFlow.getPaths()) { flowPath.setStatus(FlowPathStatus.INACTIVE); for (PathSegment pathSegment : flowPath.getSegments()) { if (pathSegment.containsNode(SWITCH_ID_A, PORT)) { pathSegment.setFailed(true); } } } RepositoryFactory repositoryFactory = mock(RepositoryFactory.class); FlowRepository flowRepository = mock(FlowRepository.class); when(flowRepository.findInactiveFlows()) .thenReturn(Collections.singletonList(pinnedFlow)); doAnswer(invocation -> { FlowStatus status = invocation.getArgument(1); pinnedFlow.setStatus(status); return null; }).when(flowRepository).updateStatusSafe(eq(pinnedFlow), any(), any()); when(repositoryFactory.createFlowRepository()).thenReturn(flowRepository); FlowPathRepository pathRepository = mock(FlowPathRepository.class); when(repositoryFactory.createFlowPathRepository()).thenReturn(pathRepository); PathSegmentRepository pathSegmentRepository = mock(PathSegmentRepository.class); when(repositoryFactory.createPathSegmentRepository()).thenReturn(pathSegmentRepository); MessageSender messageSender = mock(MessageSender.class); PersistenceManager persistenceManager = mock(PersistenceManager.class); when(persistenceManager.getRepositoryFactory()).thenReturn(repositoryFactory); TransactionManager transactionManager = mock(TransactionManager.class); doAnswer(invocation -> { TransactionCallbackWithoutResult arg = invocation.getArgument(0); arg.doInTransaction(); return null; }).when(transactionManager).doInTransaction(Mockito.<TransactionCallbackWithoutResult>any()); when(persistenceManager.getTransactionManager()).thenReturn(transactionManager); RerouteService rerouteService = new RerouteService(persistenceManager); rerouteService.rerouteInactiveFlows(messageSender, CORRELATION_ID, REROUTE_INACTIVE_FLOWS_COMMAND); assertEquals(FlowStatus.UP, pinnedFlow.getStatus()); for (FlowPath fp : pinnedFlow.getPaths()) { assertEquals(FlowPathStatus.ACTIVE, fp.getStatus()); for (PathSegment ps : fp.getSegments()) { if (ps.containsNode(SWITCH_ID_A, PORT)) { assertFalse(ps.isFailed()); } } } }
public void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); transactionManager.doInTransaction(() -> { Map<Flow, Set<PathId>> flowsForRerouting = getInactiveFlowsForRerouting(); for (Entry<Flow, Set<PathId>> entry : flowsForRerouting.entrySet()) { Flow flow = entry.getKey(); Set<IslEndpoint> allAffectedIslEndpoints = new HashSet<>(); for (FlowPath flowPath : flow.getPaths()) { Set<IslEndpoint> affectedIslEndpoints = new HashSet<>(); PathSegment firstSegment = null; int failedSegmentsCount = 0; for (PathSegment pathSegment : flowPath.getSegments()) { if (firstSegment == null) { firstSegment = pathSegment; } if (pathSegment.isFailed()) { affectedIslEndpoints.add(new IslEndpoint( pathSegment.getSrcSwitchId(), pathSegment.getSrcPort())); affectedIslEndpoints.add(new IslEndpoint( pathSegment.getDestSwitchId(), pathSegment.getDestPort())); if (pathSegment.containsNode(switchId, port)) { pathSegment.setFailed(false); pathSegmentRepository.updateFailedStatus(flowPath, pathSegment, false); } else { failedSegmentsCount++; } } } if (flowPath.getStatus().equals(FlowPathStatus.INACTIVE) && failedSegmentsCount == 0) { updateFlowPathStatus(flowPath, FlowPathStatus.ACTIVE); if (affectedIslEndpoints.isEmpty() && firstSegment != null) { affectedIslEndpoints.add(new IslEndpoint( firstSegment.getSrcSwitchId(), firstSegment.getSrcPort())); } } allAffectedIslEndpoints.addAll(affectedIslEndpoints); } FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute(attempt to restore inactive flows) request for {} (affected ISL " + "endpoints: {})", flow.getFlowId(), allAffectedIslEndpoints); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(allAffectedIslEndpoints) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } }); }
RerouteService { public void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); transactionManager.doInTransaction(() -> { Map<Flow, Set<PathId>> flowsForRerouting = getInactiveFlowsForRerouting(); for (Entry<Flow, Set<PathId>> entry : flowsForRerouting.entrySet()) { Flow flow = entry.getKey(); Set<IslEndpoint> allAffectedIslEndpoints = new HashSet<>(); for (FlowPath flowPath : flow.getPaths()) { Set<IslEndpoint> affectedIslEndpoints = new HashSet<>(); PathSegment firstSegment = null; int failedSegmentsCount = 0; for (PathSegment pathSegment : flowPath.getSegments()) { if (firstSegment == null) { firstSegment = pathSegment; } if (pathSegment.isFailed()) { affectedIslEndpoints.add(new IslEndpoint( pathSegment.getSrcSwitchId(), pathSegment.getSrcPort())); affectedIslEndpoints.add(new IslEndpoint( pathSegment.getDestSwitchId(), pathSegment.getDestPort())); if (pathSegment.containsNode(switchId, port)) { pathSegment.setFailed(false); pathSegmentRepository.updateFailedStatus(flowPath, pathSegment, false); } else { failedSegmentsCount++; } } } if (flowPath.getStatus().equals(FlowPathStatus.INACTIVE) && failedSegmentsCount == 0) { updateFlowPathStatus(flowPath, FlowPathStatus.ACTIVE); if (affectedIslEndpoints.isEmpty() && firstSegment != null) { affectedIslEndpoints.add(new IslEndpoint( firstSegment.getSrcSwitchId(), firstSegment.getSrcPort())); } } allAffectedIslEndpoints.addAll(affectedIslEndpoints); } FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute(attempt to restore inactive flows) request for {} (affected ISL " + "endpoints: {})", flow.getFlowId(), allAffectedIslEndpoints); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(allAffectedIslEndpoints) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } }); } }
RerouteService { public void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); transactionManager.doInTransaction(() -> { Map<Flow, Set<PathId>> flowsForRerouting = getInactiveFlowsForRerouting(); for (Entry<Flow, Set<PathId>> entry : flowsForRerouting.entrySet()) { Flow flow = entry.getKey(); Set<IslEndpoint> allAffectedIslEndpoints = new HashSet<>(); for (FlowPath flowPath : flow.getPaths()) { Set<IslEndpoint> affectedIslEndpoints = new HashSet<>(); PathSegment firstSegment = null; int failedSegmentsCount = 0; for (PathSegment pathSegment : flowPath.getSegments()) { if (firstSegment == null) { firstSegment = pathSegment; } if (pathSegment.isFailed()) { affectedIslEndpoints.add(new IslEndpoint( pathSegment.getSrcSwitchId(), pathSegment.getSrcPort())); affectedIslEndpoints.add(new IslEndpoint( pathSegment.getDestSwitchId(), pathSegment.getDestPort())); if (pathSegment.containsNode(switchId, port)) { pathSegment.setFailed(false); pathSegmentRepository.updateFailedStatus(flowPath, pathSegment, false); } else { failedSegmentsCount++; } } } if (flowPath.getStatus().equals(FlowPathStatus.INACTIVE) && failedSegmentsCount == 0) { updateFlowPathStatus(flowPath, FlowPathStatus.ACTIVE); if (affectedIslEndpoints.isEmpty() && firstSegment != null) { affectedIslEndpoints.add(new IslEndpoint( firstSegment.getSrcSwitchId(), firstSegment.getSrcPort())); } } allAffectedIslEndpoints.addAll(affectedIslEndpoints); } FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute(attempt to restore inactive flows) request for {} (affected ISL " + "endpoints: {})", flow.getFlowId(), allAffectedIslEndpoints); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(allAffectedIslEndpoints) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } }); } RerouteService(PersistenceManager persistenceManager); }
RerouteService { public void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); transactionManager.doInTransaction(() -> { Map<Flow, Set<PathId>> flowsForRerouting = getInactiveFlowsForRerouting(); for (Entry<Flow, Set<PathId>> entry : flowsForRerouting.entrySet()) { Flow flow = entry.getKey(); Set<IslEndpoint> allAffectedIslEndpoints = new HashSet<>(); for (FlowPath flowPath : flow.getPaths()) { Set<IslEndpoint> affectedIslEndpoints = new HashSet<>(); PathSegment firstSegment = null; int failedSegmentsCount = 0; for (PathSegment pathSegment : flowPath.getSegments()) { if (firstSegment == null) { firstSegment = pathSegment; } if (pathSegment.isFailed()) { affectedIslEndpoints.add(new IslEndpoint( pathSegment.getSrcSwitchId(), pathSegment.getSrcPort())); affectedIslEndpoints.add(new IslEndpoint( pathSegment.getDestSwitchId(), pathSegment.getDestPort())); if (pathSegment.containsNode(switchId, port)) { pathSegment.setFailed(false); pathSegmentRepository.updateFailedStatus(flowPath, pathSegment, false); } else { failedSegmentsCount++; } } } if (flowPath.getStatus().equals(FlowPathStatus.INACTIVE) && failedSegmentsCount == 0) { updateFlowPathStatus(flowPath, FlowPathStatus.ACTIVE); if (affectedIslEndpoints.isEmpty() && firstSegment != null) { affectedIslEndpoints.add(new IslEndpoint( firstSegment.getSrcSwitchId(), firstSegment.getSrcPort())); } } allAffectedIslEndpoints.addAll(affectedIslEndpoints); } FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute(attempt to restore inactive flows) request for {} (affected ISL " + "endpoints: {})", flow.getFlowId(), allAffectedIslEndpoints); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(allAffectedIslEndpoints) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } }); } RerouteService(PersistenceManager persistenceManager); void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command); void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId,
                                             SwitchId switchId); void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command); Collection<FlowPath> getAffectedFlowPaths(SwitchId switchId, int port); List<FlowPath> getPathsForSwapping(Collection<FlowPath> paths); List<FlowWithAffectedPaths> groupPathsForRerouting(Collection<FlowPath> paths); Set<Flow> groupAffectedPinnedFlows(Collection<FlowPath> paths); Map<Flow, Set<PathId>> getInactiveFlowsForRerouting(); Set<Flow> getAffectedInactiveFlowsForRerouting(SwitchId switchId); void processManualRerouteRequest(MessageSender sender, String correlationId, FlowRerouteRequest request); void processSingleSwitchFlowStatusUpdate(SwitchStateChanged request); }
RerouteService { public void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); transactionManager.doInTransaction(() -> { Map<Flow, Set<PathId>> flowsForRerouting = getInactiveFlowsForRerouting(); for (Entry<Flow, Set<PathId>> entry : flowsForRerouting.entrySet()) { Flow flow = entry.getKey(); Set<IslEndpoint> allAffectedIslEndpoints = new HashSet<>(); for (FlowPath flowPath : flow.getPaths()) { Set<IslEndpoint> affectedIslEndpoints = new HashSet<>(); PathSegment firstSegment = null; int failedSegmentsCount = 0; for (PathSegment pathSegment : flowPath.getSegments()) { if (firstSegment == null) { firstSegment = pathSegment; } if (pathSegment.isFailed()) { affectedIslEndpoints.add(new IslEndpoint( pathSegment.getSrcSwitchId(), pathSegment.getSrcPort())); affectedIslEndpoints.add(new IslEndpoint( pathSegment.getDestSwitchId(), pathSegment.getDestPort())); if (pathSegment.containsNode(switchId, port)) { pathSegment.setFailed(false); pathSegmentRepository.updateFailedStatus(flowPath, pathSegment, false); } else { failedSegmentsCount++; } } } if (flowPath.getStatus().equals(FlowPathStatus.INACTIVE) && failedSegmentsCount == 0) { updateFlowPathStatus(flowPath, FlowPathStatus.ACTIVE); if (affectedIslEndpoints.isEmpty() && firstSegment != null) { affectedIslEndpoints.add(new IslEndpoint( firstSegment.getSrcSwitchId(), firstSegment.getSrcPort())); } } allAffectedIslEndpoints.addAll(affectedIslEndpoints); } FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute(attempt to restore inactive flows) request for {} (affected ISL " + "endpoints: {})", flow.getFlowId(), allAffectedIslEndpoints); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(allAffectedIslEndpoints) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } }); } RerouteService(PersistenceManager persistenceManager); void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command); void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId,
                                             SwitchId switchId); void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command); Collection<FlowPath> getAffectedFlowPaths(SwitchId switchId, int port); List<FlowPath> getPathsForSwapping(Collection<FlowPath> paths); List<FlowWithAffectedPaths> groupPathsForRerouting(Collection<FlowPath> paths); Set<Flow> groupAffectedPinnedFlows(Collection<FlowPath> paths); Map<Flow, Set<PathId>> getInactiveFlowsForRerouting(); Set<Flow> getAffectedInactiveFlowsForRerouting(SwitchId switchId); void processManualRerouteRequest(MessageSender sender, String correlationId, FlowRerouteRequest request); void processSingleSwitchFlowStatusUpdate(SwitchStateChanged request);  }