fix removal of rules in a distributed setting

Change-Id: I44cb49990b8051f5f1542c11cbda6846049906e3

fix removal of rules in a distributed setting
Change-Id: I44cb49990b8051f5f1542c11cbda6846049906e3
alshabib · Gerrit Code Review
Commit 6cb86367396b18eebaf1e5a69c5e924b7ef5a310 6cb86367 1 parent 881f29f3
Showing 4 changed files with 127 additions and 75 deletions
apps/olt/app/src/main/java/org/onosproject/olt/impl/Olt.java
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseState.java
core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabaseState.java
drivers/default/src/main/java/org/onosproject/driver/pipeline/OltPipeline.java
--- a/apps/olt/app/src/main/java/org/onosproject/olt/impl/Olt.java
View file @6cb8636
+++ b/apps/olt/app/src/main/java/org/onosproject/olt/impl/Olt.java
View file @6cb8636
@@ -16,7 +16,6 @@
 package org.onosproject.olt.impl;
 import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -61,6 +60,9 @@ import org.onosproject.olt.AccessDeviceData;
 import org.onosproject.olt.AccessDeviceEvent;
 import org.onosproject.olt.AccessDeviceListener;
 import org.onosproject.olt.AccessDeviceService;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
 import org.osgi.service.component.ComponentContext;
 import org.slf4j.Logger;
@@ -69,7 +71,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
@@ -90,6 +91,7 @@ public class Olt
         implements AccessDeviceService {
     private static final short DEFAULT_VLAN = 0;
+    private static final String SUBSCRIBERS = "existing-subscribers";
     private final Logger log = getLogger(getClass());
@@ -108,6 +110,8 @@ public class Olt
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected ComponentConfigService componentConfigService;
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected StorageService storageService;
     @Property(name = "defaultVlan", intValue = DEFAULT_VLAN,
             label = "Default VLAN RG<->ONU traffic")
@@ -123,10 +127,7 @@ public class Olt
     private Map<DeviceId, AccessDeviceData> oltData = new ConcurrentHashMap<>();
-    private Map<ConnectPoint, Set<ForwardingObjective.Builder>> objectives =
+    private Map<ConnectPoint, VlanId> subscribers;
-            Maps.newConcurrentMap();
-
-    private Map<ConnectPoint, VlanId> subscribers = Maps.newConcurrentMap();
     private InternalNetworkConfigListener configListener =
             new InternalNetworkConfigListener();
@@ -172,6 +173,11 @@ public class Olt
                 .forEach(p -> processFilteringObjectives((DeviceId) p.element().id(),
                                                          p.number(), true));
+        subscribers = storageService.<ConnectPoint, VlanId>consistentMapBuilder()
+                .withName(SUBSCRIBERS)
+                .withSerializer(Serializer.using(KryoNamespaces.API))
+                .build().asJavaMap();
+
         deviceService.addListener(deviceListener);
         log.info("Started with Application ID {}", appId.id());
@@ -220,7 +226,15 @@ public class Olt
             return;
         }
-        unprovisionSubscriber(olt.deviceId(), olt.uplink(), port.port(), olt.vlan());
+        VlanId subscriberVlan = subscribers.remove(port);
+
+        if (subscriberVlan == null) {
+            log.warn("Unknown subscriber at location {}", port);
+            return;
+        }
+
+        unprovisionSubscriber(olt.deviceId(), olt.uplink(), port.port(), subscriberVlan,
+                              olt.vlan(), olt.defaultVlan());
     }
@@ -230,39 +244,43 @@ public class Olt
     }
     private void unprovisionSubscriber(DeviceId deviceId, PortNumber uplink,
-                                       PortNumber subscriberPort, VlanId deviceVlan) {
+                                       PortNumber subscriberPort, VlanId subscriberVlan,
-
+                                       VlanId deviceVlan, Optional<VlanId> defaultVlan) {
-        //FIXME: This method is slightly ugly but it'll do until we have a better
-        // way to remove flows from the flow store.
         CompletableFuture<ObjectiveError> downFuture = new CompletableFuture();
         CompletableFuture<ObjectiveError> upFuture = new CompletableFuture();
-        ConnectPoint cp = new ConnectPoint(deviceId, subscriberPort);
+        ForwardingObjective.Builder upFwd = upBuilder(uplink, subscriberPort,
-
+                                                      subscriberVlan, deviceVlan,
-        VlanId subscriberVlan = subscribers.remove(cp);
+                                                      defaultVlan);
+        ForwardingObjective.Builder downFwd = downBuilder(uplink, subscriberPort,
+                                                          subscriberVlan, deviceVlan,
+                                                          defaultVlan);
-        Set<ForwardingObjective.Builder> fwds = objectives.remove(cp);
-        if (fwds == null || fwds.size() != 2) {
+        flowObjectiveService.forward(deviceId, upFwd.remove(new ObjectiveContext() {
-            log.warn("Unknown or incomplete subscriber at {}", cp);
+            @Override
-            return;
+            public void onSuccess(Objective objective) {
-        }
+                upFuture.complete(null);
+            }
+            @Override
+            public void onError(Objective objective, ObjectiveError error) {
+                upFuture.complete(error);
+            }
+        }));
-        fwds.stream().forEach(
+        flowObjectiveService.forward(deviceId, downFwd.remove(new ObjectiveContext() {
-                fwd -> flowObjectiveService.forward(deviceId,
+            @Override
-                                                    fwd.remove(new ObjectiveContext() {
+            public void onSuccess(Objective objective) {
-                                                        @Override
+                downFuture.complete(null);
-                                                        public void onSuccess(Objective objective) {
+            }
-                                                            upFuture.complete(null);
-                                                        }
-                                                        @Override
+            @Override
-                                                        public void onError(Objective objective, ObjectiveError error) {
+            public void onError(Objective objective, ObjectiveError error) {
-                                                            upFuture.complete(error);
+                downFuture.complete(error);
-                                                        }
+            }
-                                                    })));
+        }));
         upFuture.thenAcceptBothAsync(downFuture, (upStatus, downStatus) -> {
             if (upStatus == null && downStatus == null) {
@@ -291,53 +309,17 @@ public class Olt
         CompletableFuture<ObjectiveError> downFuture = new CompletableFuture();
         CompletableFuture<ObjectiveError> upFuture = new CompletableFuture();
-        TrafficSelector upstream = DefaultTrafficSelector.builder()
+        ForwardingObjective.Builder upFwd = upBuilder(uplinkPort, subscriberPort,
-                .matchVlanId(defaultVlan.orElse(VlanId.vlanId((short) this.defaultVlan)))
+                                                      subscriberVlan, deviceVlan,
-                .matchInPort(subscriberPort)
+                                                      defaultVlan);
-                .build();
-
-        TrafficSelector downstream = DefaultTrafficSelector.builder()
-                .matchVlanId(deviceVlan)
-                .matchInPort(uplinkPort)
-                .matchInnerVlanId(subscriberVlan)
-                .build();
-
-        TrafficTreatment upstreamTreatment = DefaultTrafficTreatment.builder()
-                .pushVlan()
-                .setVlanId(subscriberVlan)
-                .pushVlan()
-                .setVlanId(deviceVlan)
-                .setOutput(uplinkPort)
-                .build();
-        TrafficTreatment downstreamTreatment = DefaultTrafficTreatment.builder()
-                .popVlan()
-                .setVlanId(defaultVlan.orElse(VlanId.vlanId((short) this.defaultVlan)))
-                .setOutput(subscriberPort)
-                .build();
-
+        ForwardingObjective.Builder downFwd = downBuilder(uplinkPort, subscriberPort,
-        ForwardingObjective.Builder upFwd = DefaultForwardingObjective.builder()
+                                                          subscriberVlan, deviceVlan,
-                .withFlag(ForwardingObjective.Flag.VERSATILE)
+                                                          defaultVlan);
-                .withPriority(1000)
-                .makePermanent()
-                .withSelector(upstream)
-                .fromApp(appId)
-                .withTreatment(upstreamTreatment);
-
-
-        ForwardingObjective.Builder downFwd = DefaultForwardingObjective.builder()
-                .withFlag(ForwardingObjective.Flag.VERSATILE)
-                .withPriority(1000)
-                .makePermanent()
-                .withSelector(downstream)
-                .fromApp(appId)
-                .withTreatment(downstreamTreatment);
         ConnectPoint cp = new ConnectPoint(deviceId, subscriberPort);
-
         subscribers.put(cp, subscriberVlan);
-        objectives.put(cp, Sets.newHashSet(upFwd, downFwd));
         flowObjectiveService.forward(deviceId, upFwd.add(new ObjectiveContext() {
             @Override
@@ -383,6 +365,60 @@ public class Olt
     }
+    private ForwardingObjective.Builder downBuilder(PortNumber uplinkPort,
+                                                    PortNumber subscriberPort,
+                                                    VlanId subscriberVlan,
+                                                    VlanId deviceVlan,
+                                                    Optional<VlanId> defaultVlan) {
+        TrafficSelector downstream = DefaultTrafficSelector.builder()
+                .matchVlanId(deviceVlan)
+                .matchInPort(uplinkPort)
+                .matchInnerVlanId(subscriberVlan)
+                .build();
+
+        TrafficTreatment downstreamTreatment = DefaultTrafficTreatment.builder()
+                .popVlan()
+                .setVlanId(defaultVlan.orElse(VlanId.vlanId((short) this.defaultVlan)))
+                .setOutput(subscriberPort)
+                .build();
+
+        return DefaultForwardingObjective.builder()
+                .withFlag(ForwardingObjective.Flag.VERSATILE)
+                .withPriority(1000)
+                .makePermanent()
+                .withSelector(downstream)
+                .fromApp(appId)
+                .withTreatment(downstreamTreatment);
+    }
+
+    private ForwardingObjective.Builder upBuilder(PortNumber uplinkPort,
+                                                  PortNumber subscriberPort,
+                                                  VlanId subscriberVlan,
+                                                  VlanId deviceVlan,
+                                                  Optional<VlanId> defaultVlan) {
+        TrafficSelector upstream = DefaultTrafficSelector.builder()
+                .matchVlanId(defaultVlan.orElse(VlanId.vlanId((short) this.defaultVlan)))
+                .matchInPort(subscriberPort)
+                .build();
+
+
+        TrafficTreatment upstreamTreatment = DefaultTrafficTreatment.builder()
+                .pushVlan()
+                .setVlanId(subscriberVlan)
+                .pushVlan()
+                .setVlanId(deviceVlan)
+                .setOutput(uplinkPort)
+                .build();
+
+        return DefaultForwardingObjective.builder()
+                .withFlag(ForwardingObjective.Flag.VERSATILE)
+                .withPriority(1000)
+                .makePermanent()
+                .withSelector(upstream)
+                .fromApp(appId)
+                .withTreatment(upstreamTreatment);
+    }
+
     private void processFilteringObjectives(DeviceId devId, PortNumber port, boolean install) {
         DefaultFilteringObjective.Builder builder = DefaultFilteringObjective.builder();
@@ -430,9 +466,11 @@ public class Olt
                     break;
                 case PORT_REMOVED:
                     AccessDeviceData olt = oltData.get(devId);
+                    VlanId vlan = subscribers.get(new ConnectPoint(devId,
+                                                                   event.port().number()));
                     unprovisionSubscriber(devId, olt.uplink(),
                                           event.port().number(),
-                                          olt.vlan());
+                                          vlan, olt.vlan(), olt.defaultVlan());
                     if (!oltData.get(devId).uplink().equals(event.port().number()) &&
                             event.port().isEnabled()) {
                         processFilteringObjectives(devId, event.port().number(), false);
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseState.java
View file @6cb8636
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DatabaseState.java
View file @6cb8636
@@ -104,6 +104,9 @@ public interface DatabaseState<K, V> {
   Long counterGet(String counterName);
   @Command
+  void counterSet(String counterName, long value);
+
+  @Command
   CommitResponse prepareAndCommit(Transaction transaction);
   @Command
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabaseState.java
View file @6cb8636
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultDatabaseState.java
View file @6cb8636
@@ -207,6 +207,11 @@ public class DefaultDatabaseState implements DatabaseState<String, byte[]> {
     }
     @Override
+    public void counterSet(String counterName, long value) {
+        getCounter(counterName).set(value);
+    }
+
+    @Override
     public Long queueSize(String queueName) {
         return Long.valueOf(getQueue(queueName).size());
     }
--- a/drivers/default/src/main/java/org/onosproject/driver/pipeline/OltPipeline.java
View file @6cb8636
+++ b/drivers/default/src/main/java/org/onosproject/driver/pipeline/OltPipeline.java
View file @6cb8636
@@ -38,6 +38,7 @@ import org.onosproject.net.driver.AbstractHandlerBehaviour;
 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.FlowRuleOperations;
 import org.onosproject.net.flow.FlowRuleOperationsContext;
@@ -592,7 +593,12 @@ public class OltPipeline extends AbstractHandlerBehaviour implements Pipeliner {
                 builder.add(inner.build()).add(outer.build());
                 break;
             case REMOVE:
-                builder.remove(inner.build()).remove(outer.build());
+                Iterable<FlowEntry> flows = flowRuleService.getFlowEntries(deviceId);
+                for (FlowEntry fe : flows) {
+                    if (fe.equals(inner.build()) || fe.equals(outer.build())) {
+                        builder.remove(fe);
+                    }
+                }
                 break;
             case ADD_TO_EXISTING:
                 break;