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.net.flowobjective.impl;
  17
  18 import com.google.common.collect.Maps;
  19 import com.google.common.collect.Sets;
  20 import org.apache.felix.scr.annotations.Activate;
  21 import org.apache.felix.scr.annotations.Component;
  22 import org.apache.felix.scr.annotations.Deactivate;
  23 import org.apache.felix.scr.annotations.Reference;
  24 import org.apache.felix.scr.annotations.ReferenceCardinality;
  25 import org.apache.felix.scr.annotations.Service;
  26 import org.onlab.osgi.DefaultServiceDirectory;
  27 import org.onlab.osgi.ServiceDirectory;
  28 import org.onlab.util.ItemNotFoundException;
  29 import org.onosproject.cluster.ClusterService;
  30 import org.onosproject.mastership.MastershipEvent;
  31 import org.onosproject.mastership.MastershipListener;
  32 import org.onosproject.mastership.MastershipService;
  33 import org.onosproject.net.DeviceId;
  34 import org.onosproject.net.behaviour.Pipeliner;
  35 import org.onosproject.net.behaviour.PipelinerContext;
  36 import org.onosproject.net.device.DeviceEvent;
  37 import org.onosproject.net.device.DeviceListener;
  38 import org.onosproject.net.device.DeviceService;
  39 import org.onosproject.net.driver.DefaultDriverProviderService;
  40 import org.onosproject.net.driver.DriverHandler;
  41 import org.onosproject.net.driver.DriverService;
  42 import org.onosproject.net.flow.FlowRuleService;
  43 import org.onosproject.net.flowobjective.FilteringObjective;
  44 import org.onosproject.net.flowobjective.FlowObjectiveService;
  45 import org.onosproject.net.flowobjective.FlowObjectiveStore;
  46 import org.onosproject.net.flowobjective.FlowObjectiveStoreDelegate;
  47 import org.onosproject.net.flowobjective.ForwardingObjective;
  48 import org.onosproject.net.flowobjective.NextObjective;
  49 import org.onosproject.net.flowobjective.Objective;
  50 import org.onosproject.net.flowobjective.ObjectiveError;
  51 import org.onosproject.net.flowobjective.ObjectiveEvent;
  52 import org.onosproject.net.group.GroupService;
  53 import org.slf4j.Logger;
  54 import org.slf4j.LoggerFactory;
  55
  56 import java.util.Map;
  57 import java.util.Set;
  58 import java.util.concurrent.ExecutorService;
  59
  60 import static com.google.common.base.Preconditions.checkNotNull;
  61 import static java.util.concurrent.Executors.newFixedThreadPool;
  62 import static org.onlab.util.Tools.groupedThreads;
  63 import static org.onosproject.security.AppGuard.checkPermission;
  64 import static org.onosproject.security.AppPermission.Type.*;
  65
  66
  67
  68 /**
  69  * Provides implementation of the flow objective programming service.
  70  */
  71 @Component(immediate = true)
  72 @Service
  73 public class FlowObjectiveManager implements FlowObjectiveService {
  74
  75     public static final int INSTALL_RETRY_ATTEMPTS = 5;
  76     public static final long INSTALL_RETRY_INTERVAL = 1000; // ms
  77
  78     private final Logger log = LoggerFactory.getLogger(getClass());
  79
  80     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  81     protected DriverService driverService;
  82
  83     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  84     protected DeviceService deviceService;
  85
  86     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  87     protected MastershipService mastershipService;
  88
  89     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  90     protected ClusterService clusterService;
  91
  92     // Note: The following dependencies are added on behalf of the pipeline
  93     // driver behaviours to assure these services are available for their
  94     // initialization.
  95     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  96     protected FlowRuleService flowRuleService;
  97
  98     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  99     protected GroupService groupService;
 100
 101     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 102     protected FlowObjectiveStore flowObjectiveStore;
 103
 104     // Note: This must remain an optional dependency to allow re-install of default drivers.
 105     // Note: For now disabled until we can move to OPTIONAL_UNARY dependency
 106     // @Reference(cardinality = ReferenceCardinality.OPTIONAL_UNARY, policy = ReferencePolicy.DYNAMIC)
 107     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 108     protected DefaultDriverProviderService defaultDriverService;
 109
 110     private final FlowObjectiveStoreDelegate delegate = new InternalStoreDelegate();
 111
 112     private final Map<DeviceId, DriverHandler> driverHandlers = Maps.newConcurrentMap();
 113     private final Map<DeviceId, Pipeliner> pipeliners = Maps.newConcurrentMap();
 114
 115     private final PipelinerContext context = new InnerPipelineContext();
 116     private final MastershipListener mastershipListener = new InnerMastershipListener();
 117     private final DeviceListener deviceListener = new InnerDeviceListener();
 118
 119     protected ServiceDirectory serviceDirectory = new DefaultServiceDirectory();
 120
 121     private Map<Integer, Set<PendingNext>> pendingForwards = Maps.newConcurrentMap();
 122
 123     private ExecutorService executorService;
 124
 125     @Activate
 126     protected void activate() {
 127         executorService = newFixedThreadPool(4, groupedThreads("onos/objective-installer", "%d"));
 128         flowObjectiveStore.setDelegate(delegate);
 129         mastershipService.addListener(mastershipListener);
 130         deviceService.addListener(deviceListener);
 131         deviceService.getDevices().forEach(device -> setupPipelineHandler(device.id()));
 132         log.info("Started");
 133     }
 134
 135     @Deactivate
 136     protected void deactivate() {
 137         flowObjectiveStore.unsetDelegate(delegate);
 138         mastershipService.removeListener(mastershipListener);
 139         deviceService.removeListener(deviceListener);
 140         executorService.shutdown();
 141         pipeliners.clear();
 142         driverHandlers.clear();
 143         log.info("Stopped");
 144     }
 145
 146     /**
 147      * Task that passes the flow objective down to the driver. The task will
 148      * make a few attempts to find the appropriate driver, then eventually give
 149      * up and report an error if no suitable driver could be found.
 150      */
 151     private class ObjectiveInstaller implements Runnable {
 152         private final DeviceId deviceId;
 153         private final Objective objective;
 154
 155         private final int numAttempts;
 156
 157         public ObjectiveInstaller(DeviceId deviceId, Objective objective) {
 158             this(deviceId, objective, 1);
 159         }
 160
 161         public ObjectiveInstaller(DeviceId deviceId, Objective objective, int attemps) {
 162             this.deviceId = checkNotNull(deviceId);
 163             this.objective = checkNotNull(objective);
 164             this.numAttempts = checkNotNull(attemps);
 165         }
 166
 167         @Override
 168         public void run() {
 169             try {
 170                 Pipeliner pipeliner = getDevicePipeliner(deviceId);
 171
 172                 if (pipeliner != null) {
 173                     if (objective instanceof NextObjective) {
 174                         pipeliner.next((NextObjective) objective);
 175                     } else if (objective instanceof ForwardingObjective) {
 176                         pipeliner.forward((ForwardingObjective) objective);
 177                     } else {
 178                         pipeliner.filter((FilteringObjective) objective);
 179                     }
 180                 } else if (numAttempts < INSTALL_RETRY_ATTEMPTS) {
 181                     Thread.sleep(INSTALL_RETRY_INTERVAL);
 182                     executorService.submit(new ObjectiveInstaller(deviceId, objective, numAttempts + 1));
 183                 } else {
 184                     // Otherwise we've tried a few times and failed, report an
 185                     // error back to the user.
 186                     objective.context().ifPresent(
 187                             c -> c.onError(objective, ObjectiveError.DEVICEMISSING));
 188                 }
 189             } catch (Exception e) {
 190                 log.warn("Exception while installing flow objective", e);
 191             }
 192         }
 193     }
 194
 195     @Override
 196     public void filter(DeviceId deviceId, FilteringObjective filteringObjective) {
 197         checkPermission(FLOWRULE_WRITE);
 198         executorService.submit(new ObjectiveInstaller(deviceId, filteringObjective));
 199     }
 200
 201     @Override
 202     public void forward(DeviceId deviceId, ForwardingObjective forwardingObjective) {
 203         checkPermission(FLOWRULE_WRITE);
 204         if (queueObjective(deviceId, forwardingObjective)) {
 205             return;
 206         }
 207         executorService.submit(new ObjectiveInstaller(deviceId, forwardingObjective));
 208     }
 209
 210     @Override
 211     public void next(DeviceId deviceId, NextObjective nextObjective) {
 212         checkPermission(FLOWRULE_WRITE);
 213         executorService.submit(new ObjectiveInstaller(deviceId, nextObjective));
 214     }
 215
 216     @Override
 217     public int allocateNextId() {
 218         checkPermission(FLOWRULE_WRITE);
 219         return flowObjectiveStore.allocateNextId();
 220     }
 221
 222     @Override
 223     public void initPolicy(String policy) {}
 224
 225     private boolean queueObjective(DeviceId deviceId, ForwardingObjective fwd) {
 226         if (fwd.nextId() != null &&
 227                 flowObjectiveStore.getNextGroup(fwd.nextId()) == null) {
 228             log.trace("Queuing forwarding objective for nextId {}", fwd.nextId());
 229             if (pendingForwards.putIfAbsent(fwd.nextId(),
 230                                             Sets.newHashSet(new PendingNext(deviceId, fwd))) != null) {
 231                 Set<PendingNext> pending = pendingForwards.get(fwd.nextId());
 232                 pending.add(new PendingNext(deviceId, fwd));
 233             }
 234             return true;
 235         }
 236         return false;
 237     }
 238
 239     // Retrieves the device pipeline behaviour from the cache.
 240     private Pipeliner getDevicePipeliner(DeviceId deviceId) {
 241         return pipeliners.get(deviceId);
 242     }
 243
 244     private void setupPipelineHandler(DeviceId deviceId) {
 245         if (defaultDriverService == null) {
 246             // We're not ready to go to work yet.
 247             return;
 248         }
 249
 250         // Attempt to lookup the handler in the cache
 251         DriverHandler handler = driverHandlers.get(deviceId);
 252         cTime = now();
 253
 254         if (handler == null) {
 255             try {
 256                 // Otherwise create it and if it has pipeline behaviour, cache it
 257                 handler = driverService.createHandler(deviceId);
 258                 dTime = now();
 259                 if (!handler.driver().hasBehaviour(Pipeliner.class)) {
 260                     log.warn("Pipeline behaviour not supported for device {}",
 261                              deviceId);
 262                     return;
 263                 }
 264             } catch (ItemNotFoundException e) {
 265                 log.warn("No applicable driver for device {}", deviceId);
 266                 return;
 267             }
 268
 269             driverHandlers.put(deviceId, handler);
 270             eTime = now();
 271         }
 272
 273         // Always (re)initialize the pipeline behaviour
 274         log.info("Driver {} bound to device {} ... initializing driver",
 275                  handler.driver().name(), deviceId);
 276         hTime = now();
 277         Pipeliner pipeliner = handler.behaviour(Pipeliner.class);
 278         hbTime = now();
 279         pipeliner.init(deviceId, context);
 280         pipeliners.putIfAbsent(deviceId, pipeliner);
 281     }
 282
 283     // Triggers driver setup when the local node becomes a device master.
 284     private class InnerMastershipListener implements MastershipListener {
 285         @Override
 286         public void event(MastershipEvent event) {
 287             switch (event.type()) {
 288                 case MASTER_CHANGED:
 289                     log.debug("mastership changed on device {}", event.subject());
 290                     start = now();
 291                     if (deviceService.isAvailable(event.subject())) {
 292                         setupPipelineHandler(event.subject());
 293                     }
 294                     stopWatch();
 295                     break;
 296                 case BACKUPS_CHANGED:
 297                     break;
 298                 default:
 299                     break;
 300             }
 301         }
 302     }
 303
 304     // Triggers driver setup when a device is (re)detected.
 305     private class InnerDeviceListener implements DeviceListener {
 306         @Override
 307         public void event(DeviceEvent event) {
 308             switch (event.type()) {
 309                 case DEVICE_ADDED:
 310                 case DEVICE_AVAILABILITY_CHANGED:
 311                     log.debug("Device either added or availability changed {}",
 312                               event.subject().id());
 313                     start = now();
 314                     if (deviceService.isAvailable(event.subject().id())) {
 315                         log.debug("Device is now available {}", event.subject().id());
 316                         setupPipelineHandler(event.subject().id());
 317                     }
 318                     stopWatch();
 319                     break;
 320                 case DEVICE_UPDATED:
 321                     break;
 322                 case DEVICE_REMOVED:
 323                     break;
 324                 case DEVICE_SUSPENDED:
 325                     break;
 326                 case PORT_ADDED:
 327                     break;
 328                 case PORT_UPDATED:
 329                     break;
 330                 case PORT_REMOVED:
 331                     break;
 332                 default:
 333                     break;
 334             }
 335         }
 336     }
 337
 338     // Temporary mechanism to monitor pipeliner setup time-cost; there are
 339     // intermittent time where this takes in excess of 2 seconds. Why?
 340     private long start = 0, totals = 0, count = 0;
 341     private long cTime, dTime, eTime, hTime, hbTime;
 342     private static final long LIMIT = 500;
 343
 344     private long now() {
 345         return System.currentTimeMillis();
 346     }
 347
 348     private void stopWatch() {
 349         long duration = System.currentTimeMillis() - start;
 350         totals += duration;
 351         count += 1;
 352         if (duration > LIMIT) {
 353             log.info("Pipeline setup took {} ms; avg {} ms; cTime={}, dTime={}, eTime={}, hTime={}, hbTime={}",
 354                      duration, totals / count, diff(cTime), diff(dTime), diff(eTime), diff(hTime), diff(hbTime));
 355         }
 356     }
 357
 358     private long diff(long bTime) {
 359         long diff = bTime - start;
 360         return diff < 0 ? 0 : diff;
 361     }
 362
 363     // Processing context for initializing pipeline driver behaviours.
 364     private class InnerPipelineContext implements PipelinerContext {
 365         @Override
 366         public ServiceDirectory directory() {
 367             return serviceDirectory;
 368         }
 369
 370         @Override
 371         public FlowObjectiveStore store() {
 372             return flowObjectiveStore;
 373         }
 374     }
 375
 376     private class InternalStoreDelegate implements FlowObjectiveStoreDelegate {
 377         @Override
 378         public void notify(ObjectiveEvent event) {
 379             log.debug("Received notification of obj event {}", event);
 380             Set<PendingNext> pending = pendingForwards.remove(event.subject());
 381
 382             if (pending == null) {
 383                 log.debug("Nothing pending for this obj event");
 384                 return;
 385             }
 386
 387             log.debug("Processing pending forwarding objectives {}", pending.size());
 388
 389             pending.forEach(p -> getDevicePipeliner(p.deviceId())
 390                     .forward(p.forwardingObjective()));
 391
 392         }
 393     }
 394
 395     /**
 396      * Data class used to hold a pending forwarding objective that could not
 397      * be processed because the associated next object was not present.
 398      */
 399     private class PendingNext {
 400         private final DeviceId deviceId;
 401         private final ForwardingObjective fwd;
 402
 403         public PendingNext(DeviceId deviceId, ForwardingObjective fwd) {
 404             this.deviceId = deviceId;
 405             this.fwd = fwd;
 406         }
 407
 408         public DeviceId deviceId() {
 409             return deviceId;
 410         }
 411
 412         public ForwardingObjective forwardingObjective() {
 413             return fwd;
 414         }
 415     }
 416 }