Merge pull request #2963 from telstra/fix_single_switch_flow_migration

Fix for single switch flow

services/src/kilda-core/kilda-persistence-api/src/main/java/org/openkilda/persistence/repositories/FlowRepository.java

@@ -51,6 +51,8 @@
 
     Collection<Flow> findByEndpointSwitch(SwitchId switchId);
 
+    Collection<Flow> findByEndpointSwitchWithMultiTableSupport(SwitchId switchId);
+
     Collection<Flow> findDownFlows();
 
     Optional<String> getOrCreateFlowGroupId(String flowId);

services/src/kilda-core/kilda-persistence-neo4j/src/main/java/org/openkilda/persistence/repositories/impl/Neo4jFlowRepository.java

@@ -169,6 +169,19 @@ public boolean exists(String flowId) {
                 .collect(Collectors.toList());
     }
 
+    @Override
+    public Collection<Flow> findByEndpointSwitchWithMultiTableSupport(SwitchId switchId) {
+        Filter srcSwitchFilter = createSrcSwitchFilter(switchId);
+        Filter srcMultiTableFilter = new Filter(SRC_MULTI_TABLE_PROPERTY_NAME, ComparisonOperator.IS_TRUE);
+        Filter dstSwitchFilter = createDstSwitchFilter(switchId);
+        Filter dstMultiTableFilter = new Filter(DST_MULTI_TABLE_PROPERTY_NAME, ComparisonOperator.IS_TRUE);
+
+        return Stream.concat(
+                loadAll(srcSwitchFilter.and(srcMultiTableFilter)).stream(),
+                loadAll(dstSwitchFilter.and(dstMultiTableFilter)).stream())
+                .collect(Collectors.toList());
+    }
+
     @Override
     public Collection<Flow> findDownFlows() {
         Filter flowStatusDown = new Filter(STATUS_PROPERTY_NAME, ComparisonOperator.EQUALS, FlowStatus.DOWN);

services/src/kilda-core/kilda-persistence-neo4j/src/test/java/org/openkilda/persistence/repositories/impl/Neo4jFlowRepositoryTest.java

@@ -218,6 +218,17 @@ public void shouldFindFlowBySwitchEndpoint() {
         assertThat(foundFlowIds, Matchers.hasSize(1));
     }
 
+    @Test
+    public void shouldFindFlowBySwitchEndpointWithMultiTable() {
+        Flow flow = buildTestFlow(TEST_FLOW_ID, switchA, switchB);
+        flow.setSrcWithMultiTable(true);
+        flowRepository.createOrUpdate(flow);
+
+        Collection<Flow> foundFlows = flowRepository.findByEndpointSwitchWithMultiTableSupport(TEST_SWITCH_A_ID);
+        Set<String> foundFlowIds = foundFlows.stream().map(foundFlow -> flow.getFlowId()).collect(Collectors.toSet());
+        assertThat(foundFlowIds, Matchers.hasSize(1));
+    }
+
     @Test
     public void shouldFindDownFlowIdsByEndpoint() {
         Flow flow = buildTestFlow(TEST_FLOW_ID, switchA, switchB);

services/wfm/src/main/java/org/openkilda/wfm/topology/switchmanager/fsm/SwitchValidateFsm.java

@@ -49,6 +49,7 @@
 import org.openkilda.model.SwitchId;
 import org.openkilda.model.SwitchProperties;
 import org.openkilda.persistence.repositories.FlowPathRepository;
+import org.openkilda.persistence.repositories.FlowRepository;
 import org.openkilda.persistence.repositories.IslRepository;
 import org.openkilda.persistence.repositories.RepositoryFactory;
 import org.openkilda.persistence.repositories.SwitchPropertiesRepository;
@@ -104,6 +105,7 @@ public SwitchValidateFsm(SwitchManagerCarrier carrier, String key, SwitchValidat
         this.switchId = request.getSwitchId();
         this.flowPorts = new ArrayList<>();
         FlowPathRepository flowPathRepository = repositoryFactory.createFlowPathRepository();
+        FlowRepository flowRepository = repositoryFactory.createFlowRepository();
         Collection<FlowPath> flowPaths = flowPathRepository.findBySrcSwitch(switchId);
         for (FlowPath flowPath : flowPaths) {
             if (flowPath.isForward() && flowPath.getFlow().isSrcWithMultiTable()) {
@@ -113,7 +115,8 @@ public SwitchValidateFsm(SwitchManagerCarrier carrier, String key, SwitchValidat
             }
         }
 
-        hasMultiTableFlows = !flowPathRepository.findBySegmentSwitchWithMultiTable(switchId, true).isEmpty();
+        hasMultiTableFlows = !flowPathRepository.findBySegmentSwitchWithMultiTable(switchId, true).isEmpty()
+                             || !flowRepository.findByEndpointSwitchWithMultiTableSupport(switchId).isEmpty();
 
         SwitchPropertiesRepository switchPropertiesRepository = repositoryFactory.createSwitchPropertiesRepository();
         this.switchProperties = switchPropertiesRepository.findBySwitchId(switchId).orElse(null);

services/wfm/src/test/java/org/openkilda/wfm/topology/switchmanager/service/impl/SwitchValidateServiceImplTest.java

@@ -46,6 +46,7 @@
 import org.openkilda.model.SwitchProperties;
 import org.openkilda.persistence.PersistenceManager;
 import org.openkilda.persistence.repositories.FlowPathRepository;
+import org.openkilda.persistence.repositories.FlowRepository;
 import org.openkilda.persistence.repositories.IslRepository;
 import org.openkilda.persistence.repositories.RepositoryFactory;
 import org.openkilda.persistence.repositories.SwitchPropertiesRepository;
@@ -97,6 +98,7 @@ public void setUp() {
         when(carrier.getTopologyConfig()).thenReturn(topologyConfig);
         RepositoryFactory repositoryFactory = Mockito.mock(RepositoryFactory.class);
         FlowPathRepository flowPathRepository = Mockito.mock(FlowPathRepository.class);
+        FlowRepository flowRepository = Mockito.mock(FlowRepository.class);
         SwitchPropertiesRepository switchPropertiesRepository = mock(SwitchPropertiesRepository.class);
         when(switchPropertiesRepository.findBySwitchId(any(SwitchId.class))).thenAnswer((invocation) ->
                 Optional.of(SwitchProperties.builder()
@@ -109,6 +111,7 @@ public void setUp() {
                 Collections.emptyList());
         when(repositoryFactory.createIslRepository()).thenReturn(islRepository);
         when(repositoryFactory.createFlowPathRepository()).thenReturn(flowPathRepository);
+        when(repositoryFactory.createFlowRepository()).thenReturn(flowRepository);
         when(persistenceManager.getRepositoryFactory()).thenReturn(repositoryFactory);
 
         service = new SwitchValidateServiceImpl(carrier, persistenceManager);