incubator/net/src/main/java/org/onosproject/incubator/net/virtual/impl/provider/DefaultVirtualFlowRuleProvider.java

/*
 * Copyright 2016-present Open Networking Laboratory
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.onosproject.incubator.net.virtual.impl.provider;

import com.google.common.collect.HashBasedTable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.collect.Table;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Modified;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.packet.VlanId;
import org.onosproject.core.ApplicationId;
import org.onosproject.core.CoreService;
import org.onosproject.incubator.net.virtual.NetworkId;
import org.onosproject.incubator.net.virtual.VirtualNetworkService;
import org.onosproject.incubator.net.virtual.VirtualPort;
import org.onosproject.incubator.net.virtual.VirtualLink;
import org.onosproject.incubator.net.virtual.provider.AbstractVirtualProvider;
import org.onosproject.incubator.net.virtual.provider.InternalRoutingAlgorithm;
import org.onosproject.incubator.net.virtual.provider.VirtualFlowRuleProvider;
import org.onosproject.incubator.net.virtual.provider.VirtualFlowRuleProviderService;
import org.onosproject.incubator.net.virtual.provider.VirtualProviderRegistryService;
import org.onosproject.net.ConnectPoint;
import org.onosproject.net.DeviceId;
import org.onosproject.net.Link;
import org.onosproject.net.Path;
import org.onosproject.net.PortNumber;
import org.onosproject.net.device.DeviceService;
import org.onosproject.net.flow.CompletedBatchOperation;
import org.onosproject.net.flow.DefaultFlowEntry;
import org.onosproject.net.flow.DefaultFlowRule;
import org.onosproject.net.flow.DefaultTrafficSelector;
import org.onosproject.net.flow.DefaultTrafficTreatment;
import org.onosproject.net.flow.FlowEntry;
import org.onosproject.net.flow.FlowRule;
import org.onosproject.net.flow.FlowRuleBatchEntry;
import org.onosproject.net.flow.FlowRuleBatchOperation;
import org.onosproject.net.flow.FlowRuleEvent;
import org.onosproject.net.flow.FlowRuleListener;
import org.onosproject.net.flow.FlowRuleService;
import org.onosproject.net.flow.TrafficSelector;
import org.onosproject.net.flow.TrafficTreatment;
import org.onosproject.net.flow.criteria.Criterion;
import org.onosproject.net.flow.criteria.PortCriterion;
import org.onosproject.net.flow.instructions.Instruction;
import org.onosproject.net.flow.instructions.Instructions;
import org.onosproject.net.provider.ProviderId;
import org.onosproject.net.topology.TopologyService;
import org.osgi.service.component.ComponentContext;
import org.slf4j.Logger;

import java.util.Dictionary;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.List;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.ImmutableSet.copyOf;
import static org.slf4j.LoggerFactory.getLogger;

/**
 * Provider that translate virtual flow rules into physical rules.
 * Current implementation is based on FlowRules.
 * This virtualize and de-virtualize virtual flow rules into physical flow rules.
 * {@link org.onosproject.net.flow.FlowRule}
 */
@Component(immediate = true)
@Service
public class DefaultVirtualFlowRuleProvider extends AbstractVirtualProvider
        implements VirtualFlowRuleProvider {

    private static final String APP_ID_STR = "org.onosproject.virtual.vnet-flow_";

    private final Logger log = getLogger(getClass());

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected TopologyService topologyService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected VirtualNetworkService vnService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected CoreService coreService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected FlowRuleService flowRuleService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected DeviceService deviceService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected VirtualProviderRegistryService providerRegistryService;

    private InternalRoutingAlgorithm internalRoutingAlgorithm;
    private InternalVirtualFlowRuleManager frm;
    private ApplicationId appId;
    private FlowRuleListener flowRuleListener;

    /**
     * Creates a provider with the supplied identifier.
     */
    public DefaultVirtualFlowRuleProvider() {
        super(new ProviderId("vnet-flow", "org.onosproject.virtual.vnet-flow"));
    }


    @Activate
    public void activate() {
        appId = coreService.registerApplication(APP_ID_STR);

        providerRegistryService.registerProvider(this);

        flowRuleListener = new InternalFlowRuleListener();
        flowRuleService.addListener(flowRuleListener);

        internalRoutingAlgorithm = new DefaultInternalRoutingAlgorithm();
        frm = new InternalVirtualFlowRuleManager();

        log.info("Started");
    }

    @Deactivate
    public void deactivate() {
        flowRuleService.removeListener(flowRuleListener);
        providerRegistryService.unregisterProvider(this);
    }

    @Modified
    protected void modified(ComponentContext context) {
        Dictionary<?, ?> properties = context.getProperties();
    }

    @Override
    public void applyFlowRule(NetworkId networkId, FlowRule... flowRules) {
        for (FlowRule flowRule : flowRules) {
            devirtualize(networkId, flowRule).forEach(
                    r -> {
                        flowRuleService.applyFlowRules(r);
                    }
            );
        }
    }

    @Override
    public void removeFlowRule(NetworkId networkId, FlowRule... flowRules) {
        for (FlowRule flowRule : flowRules) {
            devirtualize(networkId, flowRule).forEach(
                    r -> {
                        flowRuleService.removeFlowRules(r);
                    }
            );
        }
    }

    @Override
    public void executeBatch(NetworkId networkId, FlowRuleBatchOperation batch) {
        checkNotNull(batch);

        for (FlowRuleBatchEntry fop : batch.getOperations()) {
            devirtualize(networkId, fop.target())
                    .forEach(f -> flowRuleService.applyFlowRules(f));
        }

        //FIXME: check the success of the all batch operations
        CompletedBatchOperation status =
                new CompletedBatchOperation(true, Sets.newConcurrentHashSet(),
                                            batch.deviceId());

        VirtualFlowRuleProviderService providerService =
                (VirtualFlowRuleProviderService) providerRegistryService
                        .getProviderService(networkId,
                                            VirtualFlowRuleProvider.class);
        providerService.batchOperationCompleted(batch.id(), status);
    }

    public void setEmbeddingAlgorithm(InternalRoutingAlgorithm
                                              internalRoutingAlgorithm) {
        this.internalRoutingAlgorithm = internalRoutingAlgorithm;
    }

    /**
     * Translate the requested physical flow rules into virtual flow rules.
     *
     * @param flowRule A virtual flow rule to be translated
     * @return A flow rule for a specific virtual network
     */
    private FlowRule virtualizeFlowRule(FlowRule flowRule) {
        return frm.getVirtualRule(flowRule);
    }

    private FlowEntry virtualize(FlowEntry flowEntry) {
        FlowRule vRule = virtualizeFlowRule(flowEntry);
        FlowEntry vEntry = new DefaultFlowEntry(vRule, flowEntry.state(),
                                                flowEntry.life(),
                                                flowEntry.packets(),
                                                flowEntry.bytes());

        return vEntry;
    }

    /**
     * Translate the requested virtual flow rules into physical flow rules.
     * The translation could be one to many.
     *
     * @param flowRule A flow rule from underlying data plane to be translated
     * @return A set of flow rules for physical network
     */
    private Set<FlowRule> devirtualize(NetworkId networkId, FlowRule flowRule) {

        Set<FlowRule> outRules = new HashSet<>();

        Set<ConnectPoint> ingressPoints = extractIngressPoints(networkId,
                                                               flowRule.deviceId(),
                                                               flowRule.selector());

        ConnectPoint egressPoint = extractEgressPoints(networkId,
                                                         flowRule.deviceId(),
                                                         flowRule.treatment());

        if (egressPoint == null) {
            return outRules;
        }

        TrafficSelector.Builder commonSelectorBuilder
                = DefaultTrafficSelector.builder();
        flowRule.selector().criteria().stream()
                .filter(c -> c.type() != Criterion.Type.IN_PORT)
                .forEach(c -> commonSelectorBuilder.add(c));
        TrafficSelector commonSelector = commonSelectorBuilder.build();

        TrafficTreatment.Builder commonTreatmentBuilder
                = DefaultTrafficTreatment.builder();
        flowRule.treatment().allInstructions().stream()
                .filter(i -> i.type() != Instruction.Type.OUTPUT)
                .forEach(i -> commonTreatmentBuilder.add(i));
        TrafficTreatment commonTreatment = commonTreatmentBuilder.build();

        for (ConnectPoint ingressPoint : ingressPoints) {
            if (egressPoint.port() == PortNumber.FLOOD) {
                Set<ConnectPoint> outPoints = vnService
                        .getVirtualPorts(networkId, flowRule.deviceId())
                        .stream()
                        .map(VirtualPort::realizedBy)
                        .filter(p -> !p.equals(ingressPoint))
                        .collect(Collectors.toSet());

                for (ConnectPoint outPoint : outPoints) {
                    outRules.addAll(generateRules(networkId, ingressPoint, outPoint,
                                                  commonSelector, commonTreatment, flowRule));
                }
            } else {
                outRules.addAll(generateRules(networkId, ingressPoint, egressPoint,
                                              commonSelector, commonTreatment, flowRule));
            }
        }

        return outRules;
    }

    /**
     * Extract ingress connect points of the physical network
     * from the requested traffic selector.
     *
     * @param networkId the virtual network identifier
     * @param deviceId the virtual device identifier
     * @param selector the traffic selector to extract ingress point
     * @return the set of ingress connect points of the physical network
     */
    private Set<ConnectPoint> extractIngressPoints(NetworkId networkId,
                                                   DeviceId deviceId,
                                                   TrafficSelector selector) {

        Set<ConnectPoint> ingressPoints = new HashSet<>();

        Set<VirtualPort> vPorts = vnService
                .getVirtualPorts(networkId, deviceId);

        PortCriterion portCriterion = ((PortCriterion) selector
                .getCriterion(Criterion.Type.IN_PORT));

        if (portCriterion != null) {
            PortNumber vInPortNum = portCriterion.port();

            Optional<ConnectPoint> optionalCp =  vPorts.stream()
                    .filter(v -> v.number().equals(vInPortNum))
                    .map(VirtualPort::realizedBy).findFirst();
            if (!optionalCp.isPresent()) {
                log.warn("Port {} is not realized yet, in Network {}, Device {}",
                         vInPortNum, networkId, deviceId);
                return ingressPoints;
            }

            ingressPoints.add(optionalCp.get());
        } else {
            for (VirtualPort vPort : vPorts) {
                if (vPort.realizedBy() != null) {
                    ingressPoints.add(vPort.realizedBy());
                } else {
                    log.warn("Port {} is not realized yet, in Network {}, " +
                                     "Device {}",
                             vPort, networkId, deviceId);
                }
            }
        }

        return ingressPoints;
    }

    /**
     * Extract egress connect point of the physical network
     * from the requested traffic treatment.
     *
     * @param networkId the virtual network identifier
     * @param deviceId the virtual device identifier
     * @param treatment the traffic treatment to extract ingress point
     * @return the egress connect point of the physical network
     */
    private ConnectPoint extractEgressPoints(NetworkId networkId,
                                                  DeviceId deviceId,
                                                  TrafficTreatment treatment) {

        Set<VirtualPort> vPorts = vnService
                .getVirtualPorts(networkId, deviceId);

        PortNumber vOutPortNum = treatment.allInstructions().stream()
                .filter(i -> i.type() == Instruction.Type.OUTPUT)
                .map(i -> ((Instructions.OutputInstruction) i).port())
                .findFirst().get();

        Optional<ConnectPoint> optionalCpOut = vPorts.stream()
                .filter(v -> v.number().equals(vOutPortNum))
                .map(VirtualPort::realizedBy)
                .findFirst();

        if (!optionalCpOut.isPresent()) {
            if (vOutPortNum.isLogical()) {
                return new ConnectPoint(DeviceId.deviceId("vNet"), vOutPortNum);
            }

            log.warn("Port {} is not realized yet, in Network {}, Device {}",
                     vOutPortNum, networkId, deviceId);
            return null;
        }

        return optionalCpOut.get();
    }


    /**
     * Generates the corresponding flow rules for the physical network.
     *
     * @param networkId The virtual network identifier
     * @param ingressPoint The ingress point of the physical network
     * @param egressPoint The egress point of the physical network
     * @param commonSelector A common traffic selector between the virtual
     *                       and physical flow rules
     * @param commonTreatment A common traffic treatment between the virtual
     *                        and physical flow rules
     * @param flowRule The virtual flow rule to be translated
     * @return A set of flow rules for the physical network
     */
    private Set<FlowRule> generateRules(NetworkId networkId,
                                        ConnectPoint ingressPoint,
                                        ConnectPoint egressPoint,
                                        TrafficSelector commonSelector,
                                        TrafficTreatment commonTreatment,
                                        FlowRule flowRule) {

        Set<FlowRule> outRules = new HashSet<>();

        if (ingressPoint.deviceId().equals(egressPoint.deviceId()) ||
                egressPoint.port().isLogical()) {
            //Traffic is handled inside a single physical switch

            TrafficSelector.Builder selectorBuilder = DefaultTrafficSelector
                    .builder(commonSelector)
                    .matchInPort(ingressPoint.port());

            TrafficTreatment.Builder treatmentBuilder = DefaultTrafficTreatment
                    .builder(commonTreatment);

            VirtualPort virtualIngressPort = vnService
                    .getVirtualPorts(networkId, flowRule.deviceId())
                    .stream()
                    .filter(p -> p.realizedBy().equals(ingressPoint))
                    .findFirst()
                    .get();

            VirtualPort virtualEgressPort = vnService
                    .getVirtualPorts(networkId, flowRule.deviceId())
                    .stream()
                    .filter(p -> p.realizedBy().equals(egressPoint))
                    .findFirst()
                    .get();

            ConnectPoint ingressCp = new ConnectPoint(virtualIngressPort.element().id(), virtualIngressPort.number());
            ConnectPoint egressCp = new ConnectPoint(virtualEgressPort.element().id(), virtualEgressPort.number());

            Optional<VirtualLink> optionalIngressLink = vnService
                    .getVirtualLinks(networkId)
                    .stream()
                    .filter(l -> l.dst().equals(ingressCp))
                    .findFirst();

            Optional<VirtualLink> optionalEgressLink = vnService
                    .getVirtualLinks(networkId)
                    .stream()
                    .filter(l -> l.src().equals(egressCp))
                    .findFirst();

            //Isolate traffic from different virtual networks with VLAN
            if (!optionalIngressLink.isPresent() && !optionalEgressLink.isPresent()) {
                treatmentBuilder.setOutput(egressPoint.port());
            } else if (optionalIngressLink.isPresent() && !optionalEgressLink.isPresent()) {
                selectorBuilder.matchVlanId(VlanId.vlanId(networkId.id().shortValue()));
                treatmentBuilder.popVlan();
                treatmentBuilder.setOutput(egressPoint.port());
            } else if (!optionalIngressLink.isPresent() && optionalEgressLink.isPresent()) {
                outRules.addAll(generateRulesOnPath(networkId, optionalEgressLink.get(),
                        commonSelector, commonTreatment, flowRule));
                treatmentBuilder.pushVlan()
                        .setVlanId(VlanId.vlanId(networkId.id().shortValue()));
                treatmentBuilder.setOutput(egressPoint.port());
            } else if (optionalIngressLink.isPresent() && optionalEgressLink.isPresent()) {
                outRules.addAll(generateRulesOnPath(networkId, optionalEgressLink.get(),
                        commonSelector, commonTreatment, flowRule));
                selectorBuilder.matchVlanId(VlanId.vlanId(networkId.id().shortValue()));
                treatmentBuilder.setOutput(egressPoint.port());
            }

            FlowRule.Builder ruleBuilder = DefaultFlowRule.builder()
                    .fromApp(vnService.getVirtualNetworkApplicationId(networkId))
                    .forDevice(ingressPoint.deviceId())
                    .withSelector(selectorBuilder.build())
                    .withTreatment(treatmentBuilder.build())
                    .withPriority(flowRule.priority());

            if (flowRule.isPermanent()) {
                ruleBuilder.makePermanent();
            } else {
                ruleBuilder.makeTemporary(flowRule.timeout());
            }

            FlowRule rule = ruleBuilder.build();
            frm.addIngressRule(flowRule, rule, networkId);
            outRules.add(rule);
        } else {
            //Traffic is handled by multiple physical switches
            //A tunnel is needed.

            Path internalPath = internalRoutingAlgorithm
                    .findPath(ingressPoint, egressPoint);
            checkNotNull(internalPath, "No path between " +
                    ingressPoint.toString() + " " + egressPoint.toString());

            ConnectPoint outCp = internalPath.links().get(0).src();

            //ingress point of tunnel
            TrafficSelector.Builder selectorBuilder =
                    DefaultTrafficSelector.builder(commonSelector);
            selectorBuilder.matchInPort(ingressPoint.port());

            TrafficTreatment.Builder treatmentBuilder =
                    DefaultTrafficTreatment.builder(commonTreatment);
            //TODO: add the logic to check host location
            treatmentBuilder.pushVlan()
                    .setVlanId(VlanId.vlanId(networkId.id().shortValue()));
            treatmentBuilder.setOutput(outCp.port());

            FlowRule.Builder ruleBuilder = DefaultFlowRule.builder()
                    .fromApp(vnService.getVirtualNetworkApplicationId(networkId))
                    .forDevice(ingressPoint.deviceId())
                    .withSelector(selectorBuilder.build())
                    .withTreatment(treatmentBuilder.build())
                    .withPriority(flowRule.priority());

            if (flowRule.isPermanent()) {
                ruleBuilder.makePermanent();
            } else {
                ruleBuilder.makeTemporary(flowRule.timeout());
            }

            FlowRule rule = ruleBuilder.build();
            frm.addIngressRule(flowRule, rule, networkId);
            outRules.add(rule);

            //routing inside tunnel
            ConnectPoint inCp = internalPath.links().get(0).dst();

            if (internalPath.links().size() > 1) {
                for (Link l : internalPath.links()
                        .subList(1, internalPath.links().size())) {

                    outCp = l.src();

                    selectorBuilder = DefaultTrafficSelector
                            .builder(commonSelector)
                            .matchVlanId(VlanId.vlanId(networkId.id().shortValue()))
                            .matchInPort(inCp.port());

                    treatmentBuilder = DefaultTrafficTreatment
                            .builder(commonTreatment)
                            .setOutput(outCp.port());

                    ruleBuilder = DefaultFlowRule.builder()
                            .fromApp(vnService.getVirtualNetworkApplicationId(networkId))
                            .forDevice(inCp.deviceId())
                            .withSelector(selectorBuilder.build())
                            .withTreatment(treatmentBuilder.build())
                            .withPriority(flowRule.priority());

                    if (flowRule.isPermanent()) {
                        ruleBuilder.makePermanent();
                    } else {
                        ruleBuilder.makeTemporary(flowRule.timeout());
                    }

                    outRules.add(ruleBuilder.build());
                    inCp = l.dst();
                }
            }

            //egress point of tunnel
            selectorBuilder = DefaultTrafficSelector.builder(commonSelector)
                    .matchVlanId(VlanId.vlanId(networkId.id().shortValue()))
                    .matchInPort(inCp.port());

            treatmentBuilder = DefaultTrafficTreatment.builder(commonTreatment)
                    .popVlan()
                    .setOutput(egressPoint.port());

            ruleBuilder = DefaultFlowRule.builder()
                    .fromApp(appId)
                    .forDevice(egressPoint.deviceId())
                    .withSelector(selectorBuilder.build())
                    .withTreatment(treatmentBuilder.build())
                    .withPriority(flowRule.priority());

            if (flowRule.isPermanent()) {
                ruleBuilder.makePermanent();
            } else {
                ruleBuilder.makeTemporary(flowRule.timeout());
            }

            outRules.add(ruleBuilder.build());
        }

        return outRules;
    }

    /**
     * Generate flow rules to the intermediate nodes on the physical path for a virtual link.
     *
     * @param networkId The virtual network identifier
     * @param virtualLink A virtual link
     * @param commonSelector A common traffic selector between the virtual
     *                       and physical flow rules
     * @param commonTreatment A common traffic treatment between the virtual
     *                        and physical flow rules
     * @param flowRule The virtual flow rule to be translated
     * @return A set of flow rules for the path on physical network
     */
    private Set<FlowRule> generateRulesOnPath(NetworkId networkId,
                                              VirtualLink virtualLink,
                                              TrafficSelector commonSelector,
                                              TrafficTreatment commonTreatment,
                                              FlowRule flowRule) {

        VirtualPort srcVirtualPort = vnService
                .getVirtualPorts(networkId, virtualLink.src().deviceId())
                .stream()
                .filter(p -> p.number().equals(virtualLink.src().port()))
                .findFirst()
                .get();

        VirtualPort dstVirtualPort = vnService
                .getVirtualPorts(networkId, virtualLink.dst().deviceId())
                .stream()
                .filter(p -> p.number().equals(virtualLink.dst().port()))
                .findFirst()
                .get();
        Set<FlowRule> outRules = new HashSet<>();
        ConnectPoint srcCp = srcVirtualPort.realizedBy();
        ConnectPoint dstCp = dstVirtualPort.realizedBy();

        Path internalPath = internalRoutingAlgorithm
                .findPath(srcCp, dstCp);
        List<Link> links = internalPath.links();
        if (internalPath != null && links.size() > 1) {
            for (int i = 0; i < links.size() - 1; i++) {
                ConnectPoint inCp = links.get(i).dst();
                ConnectPoint outCp = links.get(i + 1).src();
                TrafficSelector.Builder linkSelectorBuilder = DefaultTrafficSelector
                        .builder(commonSelector)
                        .matchVlanId(VlanId.vlanId(networkId.id().shortValue()))
                        .matchInPort(inCp.port());

                TrafficTreatment.Builder linkTreatmentBuilder = DefaultTrafficTreatment
                        .builder(commonTreatment)
                        .setOutput(outCp.port());

                FlowRule.Builder ruleBuilder = DefaultFlowRule.builder()
                        .fromApp(vnService.getVirtualNetworkApplicationId(networkId))
                        .forDevice(inCp.deviceId())
                        .withSelector(linkSelectorBuilder.build())
                        .withTreatment(linkTreatmentBuilder.build())
                        .withPriority(flowRule.priority());

                if (flowRule.isPermanent()) {
                    ruleBuilder.makePermanent();
                } else {
                    ruleBuilder.makeTemporary(flowRule.timeout());
                }

                outRules.add(ruleBuilder.build());
            }
        }
        return outRules;
    }

    private class InternalFlowRuleListener implements FlowRuleListener {
        @Override
        public void event(FlowRuleEvent event) {
            if ((event.type() == FlowRuleEvent.Type.RULE_ADDED) ||
                    (event.type() == FlowRuleEvent.Type.RULE_UPDATED)) {
                if (frm.isVirtualIngressRule(event.subject())) {
                    NetworkId networkId = frm.getVirtualNetworkId(event.subject());
                    FlowEntry vEntry = getVirtualFlowEntry(event.subject());
                    frm.addOrUpdateFlowEntry(networkId, vEntry.deviceId(), vEntry);

                    VirtualFlowRuleProviderService providerService =
                            (VirtualFlowRuleProviderService) providerRegistryService
                                    .getProviderService(networkId,
                                                        VirtualFlowRuleProvider.class);

                    ImmutableList.Builder<FlowEntry> builder = ImmutableList.builder();
                    builder.addAll(frm.getFlowEntries(networkId, vEntry.deviceId()));

                    providerService.pushFlowMetrics(vEntry.deviceId(), builder.build());
                }
            } else if (event.type() == FlowRuleEvent.Type.RULE_REMOVED) {
                if (frm.isVirtualIngressRule(event.subject())) {
                    //FIXME confirm all physical rules are removed
                    NetworkId networkId = frm.getVirtualNetworkId(event.subject());
                    FlowEntry vEntry = getVirtualFlowEntry(event.subject());

                    VirtualFlowRuleProviderService providerService =
                            (VirtualFlowRuleProviderService) providerRegistryService
                                    .getProviderService(networkId,
                                                        VirtualFlowRuleProvider.class);
                    providerService.flowRemoved(vEntry);
                }
            }
        }

        private FlowEntry getVirtualFlowEntry(FlowRule rule) {
            FlowEntry entry = null;
            for (FlowEntry fe :
                    flowRuleService.getFlowEntries(rule.deviceId())) {
                if (rule.exactMatch(fe)) {
                    entry = fe;
                }
            }

            FlowRule vRule = virtualize(entry);
            FlowEntry vEntry = new DefaultFlowEntry(vRule, entry.state(),
                                                    entry.life(), entry.packets(),
                                                    entry.bytes());

            return vEntry;
        }
    }

    private class InternalVirtualFlowRuleManager {
        /** <Virtual Network ID, Virtual Device ID, Virtual Flow Rules>.*/
        final Table<NetworkId, DeviceId, Set<FlowRule>> flowRuleTable
                = HashBasedTable.create();

        /** <Virtual Network ID, Virtual Device ID, Virtual Flow Entries>.*/
        final Table<NetworkId, DeviceId, Set<FlowEntry>> flowEntryTable
                = HashBasedTable.create();

        /** <Physical Flow Rule, Virtual Network ID>.*/
        final Map<FlowRule, NetworkId> ingressRuleMap = Maps.newHashMap();

        /** <Physical Flow Rule, Virtual Virtual Flow Rule>.*/
        final Map<FlowRule, FlowRule> virtualizationMap = Maps.newHashMap();

        private Iterable<FlowRule> getFlowRules(NetworkId networkId,
                                                DeviceId deviceId) {
            return flowRuleTable.get(networkId, deviceId);
        }

        private Iterable<FlowEntry> getFlowEntries(NetworkId networkId,
                                                   DeviceId deviceId) {
            return flowEntryTable.get(networkId, deviceId);
        }

        private void addFlowRule(NetworkId networkId, DeviceId deviceId,
                                 FlowRule flowRule) {
            Set<FlowRule> set = flowRuleTable.get(networkId, deviceId);
            if (set == null) {
                set = Sets.newHashSet();
                flowRuleTable.put(networkId, deviceId, set);
            }
            set.add(flowRule);
        }

        private void removeFlowRule(NetworkId networkId, DeviceId deviceId,
                                    FlowRule flowRule) {
            Set<FlowRule> set = flowRuleTable.get(networkId, deviceId);
            if (set == null) {
                return;
            }
            set.remove(flowRule);
        }

        private void addOrUpdateFlowEntry(NetworkId networkId, DeviceId deviceId,
                                  FlowEntry flowEntry) {
            Set<FlowEntry> set = flowEntryTable.get(networkId, deviceId);
            if (set == null) {
                set = Sets.newConcurrentHashSet();
                flowEntryTable.put(networkId, deviceId, set);
            }

            //Replace old entry with new one
            set.stream().filter(fe -> fe.exactMatch(flowEntry))
                    .forEach(set::remove);
            set.add(flowEntry);
        }

        private void removeFlowEntry(NetworkId networkId, DeviceId deviceId,
                                     FlowEntry flowEntry) {
            Set<FlowEntry> set = flowEntryTable.get(networkId, deviceId);
            if (set == null) {
                return;
            }
            set.remove(flowEntry);
        }

        private void addIngressRule(FlowRule virtualRule, FlowRule physicalRule,
                                    NetworkId networkId) {
            ingressRuleMap.put(physicalRule, networkId);
            virtualizationMap.put(physicalRule, virtualRule);
        }

        private FlowRule getVirtualRule(FlowRule physicalRule) {
                return virtualizationMap.get(physicalRule);
        }

        private void removeIngressRule(FlowRule physicalRule) {
            ingressRuleMap.remove(physicalRule);
            virtualizationMap.remove(physicalRule);
        }

        private Set<FlowRule> getAllPhysicalRule() {
            return copyOf(virtualizationMap.keySet());
        }

        private NetworkId getVirtualNetworkId(FlowRule physicalRule) {
            return ingressRuleMap.get(physicalRule);
        }

        /**
         * Test the rule is the ingress rule for virtual rules.
         *
         * @param flowRule A flow rule from underlying data plane to be translated
         * @return True when the rule is for ingress point for a virtual switch
         */
        private boolean isVirtualIngressRule(FlowRule flowRule) {
            return ingressRuleMap.containsKey(flowRule);
        }
    }

    private class DefaultInternalRoutingAlgorithm
            implements InternalRoutingAlgorithm {

        @Override
        public Path findPath(ConnectPoint src, ConnectPoint dst) {
            Set<Path> paths =
                    topologyService.getPaths(topologyService.currentTopology(),
                                             src.deviceId(),
                                             dst.deviceId());

            if (paths.isEmpty()) {
                return null;
            }

            //TODO the logic find the best path
            return (Path) paths.toArray()[0];
        }
    }
}