gerrit.opnfv Code Review - onosfw.git/blob

   1 /*
   2  * Copyright 2015 Open Networking Laboratory
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.onosproject.segmentrouting;
  17
  18 import org.apache.felix.scr.annotations.Activate;
  19 import org.apache.felix.scr.annotations.Component;
  20 import org.apache.felix.scr.annotations.Deactivate;
  21 import org.apache.felix.scr.annotations.Reference;
  22 import org.apache.felix.scr.annotations.ReferenceCardinality;
  23 import org.apache.felix.scr.annotations.Service;
  24 import org.onlab.packet.Ethernet;
  25 import org.onlab.packet.IPv4;
  26 import org.onlab.util.KryoNamespace;
  27 import org.onosproject.core.ApplicationId;
  28 import org.onosproject.core.CoreService;
  29 import org.onosproject.event.Event;
  30 import org.onosproject.net.config.ConfigFactory;
  31 import org.onosproject.net.config.NetworkConfigEvent;
  32 import org.onosproject.net.config.NetworkConfigRegistry;
  33 import org.onosproject.net.config.NetworkConfigListener;
  34 import org.onosproject.net.config.basics.SubjectFactories;
  35 import org.onosproject.segmentrouting.config.SegmentRoutingConfig;
  36 import org.onosproject.segmentrouting.grouphandler.DefaultGroupHandler;
  37 import org.onosproject.segmentrouting.grouphandler.NeighborSet;
  38 import org.onosproject.segmentrouting.grouphandler.NeighborSetNextObjectiveStoreKey;
  39 import org.onosproject.mastership.MastershipService;
  40 import org.onosproject.net.Device;
  41 import org.onosproject.net.DeviceId;
  42 import org.onosproject.net.Link;
  43 import org.onosproject.net.Port;
  44 import org.onosproject.net.device.DeviceEvent;
  45 import org.onosproject.net.device.DeviceListener;
  46 import org.onosproject.net.device.DeviceService;
  47 import org.onosproject.net.flowobjective.FlowObjectiveService;
  48 import org.onosproject.net.group.GroupKey;
  49 import org.onosproject.net.host.HostService;
  50 import org.onosproject.net.intent.IntentService;
  51 import org.onosproject.net.link.LinkEvent;
  52 import org.onosproject.net.link.LinkListener;
  53 import org.onosproject.net.link.LinkService;
  54 import org.onosproject.net.packet.InboundPacket;
  55 import org.onosproject.net.packet.PacketContext;
  56 import org.onosproject.net.packet.PacketProcessor;
  57 import org.onosproject.net.packet.PacketService;
  58 import org.onosproject.net.topology.TopologyService;
  59 import org.onosproject.store.service.EventuallyConsistentMap;
  60 import org.onosproject.store.service.EventuallyConsistentMapBuilder;
  61 import org.onosproject.store.service.StorageService;
  62 import org.onosproject.store.service.WallClockTimestamp;
  63 import org.slf4j.Logger;
  64 import org.slf4j.LoggerFactory;
  65
  66 import java.net.URI;
  67 import java.util.HashSet;
  68 import java.util.List;
  69 import java.util.Map;
  70 import java.util.concurrent.ConcurrentHashMap;
  71 import java.util.concurrent.ConcurrentLinkedQueue;
  72 import java.util.concurrent.Executors;
  73 import java.util.concurrent.ScheduledExecutorService;
  74 import java.util.concurrent.ScheduledFuture;
  75 import java.util.concurrent.TimeUnit;
  76
  77 @SuppressWarnings("ALL")
  78 @Service
  79 @Component(immediate = true)
  80 public class SegmentRoutingManager implements SegmentRoutingService {
  81
  82     private static Logger log = LoggerFactory
  83             .getLogger(SegmentRoutingManager.class);
  84
  85     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  86     protected CoreService coreService;
  87
  88     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  89     protected TopologyService topologyService;
  90
  91     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  92     protected PacketService packetService;
  93
  94     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  95     protected IntentService intentService;
  96
  97     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  98     protected HostService hostService;
  99
 100     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 101     protected DeviceService deviceService;
 102
 103     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 104     protected FlowObjectiveService flowObjectiveService;
 105
 106     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 107     protected LinkService linkService;
 108
 109     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 110     protected MastershipService mastershipService;
 111
 112     protected ArpHandler arpHandler = null;
 113     protected IcmpHandler icmpHandler = null;
 114     protected IpHandler ipHandler = null;
 115     protected RoutingRulePopulator routingRulePopulator = null;
 116     protected ApplicationId appId;
 117     protected DeviceConfiguration deviceConfiguration = null;
 118
 119     private DefaultRoutingHandler defaultRoutingHandler = null;
 120     private TunnelHandler tunnelHandler = null;
 121     private PolicyHandler policyHandler = null;
 122     private InternalPacketProcessor processor = null;
 123     private InternalLinkListener linkListener = null;
 124     private InternalDeviceListener deviceListener = null;
 125     private InternalEventHandler eventHandler = new InternalEventHandler();
 126
 127     private ScheduledExecutorService executorService = Executors
 128             .newScheduledThreadPool(1);
 129
 130     private static ScheduledFuture<?> eventHandlerFuture = null;
 131     private ConcurrentLinkedQueue<Event> eventQueue = new ConcurrentLinkedQueue<Event>();
 132     private Map<DeviceId, DefaultGroupHandler> groupHandlerMap = new ConcurrentHashMap<DeviceId, DefaultGroupHandler>();
 133     // Per device next objective ID store with (device id + neighbor set) as key
 134     private EventuallyConsistentMap<NeighborSetNextObjectiveStoreKey,
 135         Integer> nsNextObjStore = null;
 136     private EventuallyConsistentMap<String, Tunnel> tunnelStore = null;
 137     private EventuallyConsistentMap<String, Policy> policyStore = null;
 138
 139     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 140     protected StorageService storageService;
 141
 142     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 143     protected NetworkConfigRegistry cfgService;
 144
 145     private final InternalConfigListener cfgListener =
 146             new InternalConfigListener(this);
 147
 148     private final ConfigFactory cfgFactory =
 149             new ConfigFactory(SubjectFactories.DEVICE_SUBJECT_FACTORY,
 150                               SegmentRoutingConfig.class,
 151                               "segmentrouting") {
 152                 @Override
 153                 public SegmentRoutingConfig createConfig() {
 154                     return new SegmentRoutingConfig();
 155                 }
 156             };
 157
 158     private Object threadSchedulerLock = new Object();
 159     private static int numOfEventsQueued = 0;
 160     private static int numOfEventsExecuted = 0;
 161     private static int numOfHandlerExecution = 0;
 162     private static int numOfHandlerScheduled = 0;
 163
 164     private KryoNamespace.Builder kryoBuilder = null;
 165
 166     @Activate
 167     protected void activate() {
 168         appId = coreService
 169                 .registerApplication("org.onosproject.segmentrouting");
 170
 171         kryoBuilder = new KryoNamespace.Builder()
 172             .register(NeighborSetNextObjectiveStoreKey.class,
 173                     NeighborSet.class,
 174                     DeviceId.class,
 175                     URI.class,
 176                     WallClockTimestamp.class,
 177                     org.onosproject.cluster.NodeId.class,
 178                     HashSet.class,
 179                     Tunnel.class,
 180                     DefaultTunnel.class,
 181                     Policy.class,
 182                     TunnelPolicy.class,
 183                     Policy.Type.class
 184             );
 185
 186         log.debug("Creating EC map nsnextobjectivestore");
 187         EventuallyConsistentMapBuilder<NeighborSetNextObjectiveStoreKey, Integer>
 188                 nsNextObjMapBuilder = storageService.eventuallyConsistentMapBuilder();
 189
 190         nsNextObjStore = nsNextObjMapBuilder
 191                 .withName("nsnextobjectivestore")
 192                 .withSerializer(kryoBuilder)
 193                 .withTimestampProvider((k, v) -> new WallClockTimestamp())
 194                 .build();
 195         log.trace("Current size {}", nsNextObjStore.size());
 196
 197         EventuallyConsistentMapBuilder<String, Tunnel> tunnelMapBuilder =
 198                 storageService.eventuallyConsistentMapBuilder();
 199
 200         tunnelStore = tunnelMapBuilder
 201                 .withName("tunnelstore")
 202                 .withSerializer(kryoBuilder)
 203                 .withTimestampProvider((k, v) -> new WallClockTimestamp())
 204                 .build();
 205
 206         EventuallyConsistentMapBuilder<String, Policy> policyMapBuilder =
 207                 storageService.eventuallyConsistentMapBuilder();
 208
 209         policyStore = policyMapBuilder
 210                 .withName("policystore")
 211                 .withSerializer(kryoBuilder)
 212                 .withTimestampProvider((k, v) -> new WallClockTimestamp())
 213                 .build();
 214
 215         cfgService.addListener(cfgListener);
 216         cfgService.registerConfigFactory(cfgFactory);
 217
 218         processor = new InternalPacketProcessor();
 219         linkListener = new InternalLinkListener();
 220         deviceListener = new InternalDeviceListener();
 221
 222         packetService.addProcessor(processor, PacketProcessor.director(2));
 223         linkService.addListener(linkListener);
 224         deviceService.addListener(deviceListener);
 225
 226         cfgListener.configureNetwork();
 227
 228         log.info("Started");
 229     }
 230
 231     @Deactivate
 232     protected void deactivate() {
 233         cfgService.removeListener(cfgListener);
 234         cfgService.unregisterConfigFactory(cfgFactory);
 235
 236         packetService.removeProcessor(processor);
 237         linkService.removeListener(linkListener);
 238         deviceService.removeListener(deviceListener);
 239         processor = null;
 240         linkListener = null;
 241         deviceService = null;
 242
 243         groupHandlerMap.clear();
 244
 245         log.info("Stopped");
 246     }
 247
 248
 249     @Override
 250     public List<Tunnel> getTunnels() {
 251         return tunnelHandler.getTunnels();
 252     }
 253
 254     @Override
 255     public TunnelHandler.Result createTunnel(Tunnel tunnel) {
 256         return tunnelHandler.createTunnel(tunnel);
 257     }
 258
 259     @Override
 260     public TunnelHandler.Result removeTunnel(Tunnel tunnel) {
 261         for (Policy policy: policyHandler.getPolicies()) {
 262             if (policy.type() == Policy.Type.TUNNEL_FLOW) {
 263                 TunnelPolicy tunnelPolicy = (TunnelPolicy) policy;
 264                 if (tunnelPolicy.tunnelId().equals(tunnel.id())) {
 265                     log.warn("Cannot remove the tunnel used by a policy");
 266                     return TunnelHandler.Result.TUNNEL_IN_USE;
 267                 }
 268             }
 269         }
 270         return tunnelHandler.removeTunnel(tunnel);
 271     }
 272
 273     @Override
 274     public PolicyHandler.Result removePolicy(Policy policy) {
 275         return policyHandler.removePolicy(policy);
 276     }
 277
 278     @Override
 279     public PolicyHandler.Result createPolicy(Policy policy) {
 280         return policyHandler.createPolicy(policy);
 281     }
 282
 283     @Override
 284     public List<Policy> getPolicies() {
 285         return policyHandler.getPolicies();
 286     }
 287
 288     /**
 289      * Returns the tunnel object with the tunnel ID.
 290      *
 291      * @param tunnelId Tunnel ID
 292      * @return Tunnel reference
 293      */
 294     public Tunnel getTunnel(String tunnelId) {
 295         return tunnelHandler.getTunnel(tunnelId);
 296     }
 297
 298     /**
 299      * Returns the GrouopKey object for the device and the NighborSet given.
 300      *
 301      * @param ns NeightborSet object for the GroupKey
 302      * @return GroupKey object for the NeighborSet
 303      */
 304     public GroupKey getGroupKey(NeighborSet ns) {
 305         for (DefaultGroupHandler groupHandler : groupHandlerMap.values()) {
 306             return groupHandler.getGroupKey(ns);
 307         }
 308
 309         return null;
 310     }
 311
 312     /**
 313      * Returns the next objective ID for the NeighborSet given. If the nextObjectiveID does not exist,
 314      * a new one is created and returned.
 315      *
 316      * @param deviceId Device ID
 317      * @param ns NegighborSet
 318      * @return next objective ID
 319      */
 320     public int getNextObjectiveId(DeviceId deviceId, NeighborSet ns) {
 321         if (groupHandlerMap.get(deviceId) != null) {
 322             log.trace("getNextObjectiveId query in device {}", deviceId);
 323             return groupHandlerMap
 324                     .get(deviceId).getNextObjectiveId(ns);
 325         } else {
 326             log.warn("getNextObjectiveId query in device {} not found", deviceId);
 327             return -1;
 328         }
 329     }
 330
 331     private class InternalPacketProcessor implements PacketProcessor {
 332         @Override
 333         public void process(PacketContext context) {
 334
 335             if (context.isHandled()) {
 336                 return;
 337             }
 338
 339             InboundPacket pkt = context.inPacket();
 340             Ethernet ethernet = pkt.parsed();
 341
 342             if (ethernet.getEtherType() == Ethernet.TYPE_ARP) {
 343                 arpHandler.processPacketIn(pkt);
 344             } else if (ethernet.getEtherType() == Ethernet.TYPE_IPV4) {
 345                 IPv4 ipPacket = (IPv4) ethernet.getPayload();
 346                 ipHandler.addToPacketBuffer(ipPacket);
 347                 if (ipPacket.getProtocol() == IPv4.PROTOCOL_ICMP) {
 348                     icmpHandler.processPacketIn(pkt);
 349                 } else {
 350                     ipHandler.processPacketIn(pkt);
 351                 }
 352             }
 353         }
 354     }
 355
 356     private class InternalLinkListener implements LinkListener {
 357         @Override
 358         public void event(LinkEvent event) {
 359             if (event.type() == LinkEvent.Type.LINK_ADDED
 360                     || event.type() == LinkEvent.Type.LINK_REMOVED) {
 361                 log.debug("Event {} received from Link Service", event.type());
 362                 scheduleEventHandlerIfNotScheduled(event);
 363             }
 364         }
 365     }
 366
 367     private class InternalDeviceListener implements DeviceListener {
 368         @Override
 369         public void event(DeviceEvent event) {
 370             switch (event.type()) {
 371             case DEVICE_ADDED:
 372             case PORT_REMOVED:
 373             case DEVICE_UPDATED:
 374             case DEVICE_AVAILABILITY_CHANGED:
 375                 log.debug("Event {} received from Device Service", event.type());
 376                 scheduleEventHandlerIfNotScheduled(event);
 377                 break;
 378             default:
 379             }
 380         }
 381     }
 382
 383     private void scheduleEventHandlerIfNotScheduled(Event event) {
 384         synchronized (threadSchedulerLock) {
 385             eventQueue.add(event);
 386             numOfEventsQueued++;
 387
 388             if ((numOfHandlerScheduled - numOfHandlerExecution) == 0) {
 389                 //No pending scheduled event handling threads. So start a new one.
 390                 eventHandlerFuture = executorService
 391                         .schedule(eventHandler, 100, TimeUnit.MILLISECONDS);
 392                 numOfHandlerScheduled++;
 393             }
 394             log.trace("numOfEventsQueued {}, numOfEventHanlderScheduled {}",
 395                       numOfEventsQueued,
 396                       numOfHandlerScheduled);
 397         }
 398     }
 399
 400     private class InternalEventHandler implements Runnable {
 401         @Override
 402         public void run() {
 403             try {
 404                 while (true) {
 405                     Event event = null;
 406                     synchronized (threadSchedulerLock) {
 407                         if (!eventQueue.isEmpty()) {
 408                             event = eventQueue.poll();
 409                             numOfEventsExecuted++;
 410                         } else {
 411                             numOfHandlerExecution++;
 412                             log.debug("numOfHandlerExecution {} numOfEventsExecuted {}",
 413                                       numOfHandlerExecution, numOfEventsExecuted);
 414                             break;
 415                         }
 416                     }
 417                     if (event.type() == LinkEvent.Type.LINK_ADDED) {
 418                         processLinkAdded((Link) event.subject());
 419                     } else if (event.type() == LinkEvent.Type.LINK_REMOVED) {
 420                         processLinkRemoved((Link) event.subject());
 421                     } else if (event.type() == DeviceEvent.Type.DEVICE_ADDED ||
 422                             event.type() == DeviceEvent.Type.DEVICE_AVAILABILITY_CHANGED ||
 423                             event.type() == DeviceEvent.Type.DEVICE_UPDATED) {
 424                         if (deviceService.isAvailable(((Device) event.subject()).id())) {
 425                             processDeviceAdded((Device) event.subject());
 426                         }
 427                     } else if (event.type() == DeviceEvent.Type.PORT_REMOVED) {
 428                         processPortRemoved((Device) event.subject(),
 429                                            ((DeviceEvent) event).port());
 430                     } else {
 431                         log.warn("Unhandled event type: {}", event.type());
 432                     }
 433                 }
 434             } catch (Exception e) {
 435                 log.error("SegmentRouting event handler "
 436                         + "thread thrown an exception: {}", e);
 437             }
 438         }
 439     }
 440
 441     private void processLinkAdded(Link link) {
 442         log.debug("A new link {} was added", link.toString());
 443
 444         //Irrespective whether the local is a MASTER or not for this device,
 445         //create group handler instance and push default TTP flow rules.
 446         //Because in a multi-instance setup, instances can initiate
 447         //groups for any devices. Also the default TTP rules are needed
 448         //to be pushed before inserting any IP table entries for any device
 449         DefaultGroupHandler groupHandler = groupHandlerMap.get(link.src()
 450                 .deviceId());
 451         if (groupHandler != null) {
 452             groupHandler.linkUp(link);
 453         } else {
 454             Device device = deviceService.getDevice(link.src().deviceId());
 455             if (device != null) {
 456                 log.warn("processLinkAdded: Link Added "
 457                         + "Notification without Device Added "
 458                         + "event, still handling it");
 459                 processDeviceAdded(device);
 460                 groupHandler = groupHandlerMap.get(link.src()
 461                                                    .deviceId());
 462                 groupHandler.linkUp(link);
 463             }
 464         }
 465
 466         log.trace("Starting optimized route population process");
 467         defaultRoutingHandler.populateRoutingRulesForLinkStatusChange(null);
 468         //log.trace("processLinkAdded: re-starting route population process");
 469         //defaultRoutingHandler.startPopulationProcess();
 470     }
 471
 472     private void processLinkRemoved(Link link) {
 473         log.debug("A link {} was removed", link.toString());
 474         DefaultGroupHandler groupHandler = groupHandlerMap.get(link.src().deviceId());
 475         if (groupHandler != null) {
 476             groupHandler.portDown(link.src().port());
 477         }
 478         log.trace("Starting optimized route population process");
 479         defaultRoutingHandler.populateRoutingRulesForLinkStatusChange(link);
 480         //log.trace("processLinkRemoved: re-starting route population process");
 481         //defaultRoutingHandler.startPopulationProcess();
 482     }
 483
 484     private void processDeviceAdded(Device device) {
 485         log.debug("A new device with ID {} was added", device.id());
 486         //Irrespective whether the local is a MASTER or not for this device,
 487         //create group handler instance and push default TTP flow rules.
 488         //Because in a multi-instance setup, instances can initiate
 489         //groups for any devices. Also the default TTP rules are needed
 490         //to be pushed before inserting any IP table entries for any device
 491         DefaultGroupHandler dgh = DefaultGroupHandler.
 492                 createGroupHandler(device.id(),
 493                                    appId,
 494                                    deviceConfiguration,
 495                                    linkService,
 496                                    flowObjectiveService,
 497                                    nsNextObjStore);
 498         groupHandlerMap.put(device.id(), dgh);
 499         defaultRoutingHandler.populateTtpRules(device.id());
 500     }
 501
 502     private void processPortRemoved(Device device, Port port) {
 503         log.debug("Port {} was removed", port.toString());
 504         DefaultGroupHandler groupHandler = groupHandlerMap.get(device.id());
 505         if (groupHandler != null) {
 506             groupHandler.portDown(port.number());
 507         }
 508     }
 509
 510     private class InternalConfigListener implements NetworkConfigListener {
 511         SegmentRoutingManager segmentRoutingManager;
 512
 513         public InternalConfigListener(SegmentRoutingManager srMgr) {
 514             this.segmentRoutingManager = srMgr;
 515         }
 516
 517         public void configureNetwork() {
 518             deviceConfiguration = new DeviceConfiguration(segmentRoutingManager.cfgService);
 519
 520             arpHandler = new ArpHandler(segmentRoutingManager);
 521             icmpHandler = new IcmpHandler(segmentRoutingManager);
 522             ipHandler = new IpHandler(segmentRoutingManager);
 523             routingRulePopulator = new RoutingRulePopulator(segmentRoutingManager);
 524             defaultRoutingHandler = new DefaultRoutingHandler(segmentRoutingManager);
 525
 526             tunnelHandler = new TunnelHandler(linkService, deviceConfiguration,
 527                                               groupHandlerMap, tunnelStore);
 528             policyHandler = new PolicyHandler(appId, deviceConfiguration,
 529                                               flowObjectiveService,
 530                                               tunnelHandler, policyStore);
 531
 532             for (Device device : deviceService.getDevices()) {
 533                 //Irrespective whether the local is a MASTER or not for this device,
 534                 //create group handler instance and push default TTP flow rules.
 535                 //Because in a multi-instance setup, instances can initiate
 536                 //groups for any devices. Also the default TTP rules are needed
 537                 //to be pushed before inserting any IP table entries for any device
 538                 DefaultGroupHandler groupHandler = DefaultGroupHandler
 539                         .createGroupHandler(device.id(), appId,
 540                                             deviceConfiguration, linkService,
 541                                             flowObjectiveService,
 542                                             nsNextObjStore);
 543                 groupHandlerMap.put(device.id(), groupHandler);
 544                 defaultRoutingHandler.populateTtpRules(device.id());
 545             }
 546
 547             defaultRoutingHandler.startPopulationProcess();
 548         }
 549
 550         @Override
 551         public void event(NetworkConfigEvent event) {
 552             if (event.configClass().equals(SegmentRoutingConfig.class)) {
 553                 if (event.type() == NetworkConfigEvent.Type.CONFIG_ADDED) {
 554                     log.info("Network configuration added.");
 555                     configureNetwork();
 556                 }
 557                 if (event.type() == NetworkConfigEvent.Type.CONFIG_UPDATED) {
 558                     log.info("Network configuration updated.");
 559                     // TODO support dynamic configuration
 560                 }
 561             }
 562         }
 563     }
 564 }