renderers/ofoverlay/src/main/java/org/opendaylight/groupbasedpolicy/renderer/ofoverlay/PolicyManager.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.groupbasedpolicy.renderer.ofoverlay;
  10
  11 import java.io.Closeable;
  12 import java.io.IOException;
  13 import java.util.ArrayList;
  14 import java.util.Collection;
  15 import java.util.List;
  16 import java.util.concurrent.Callable;
  17 import java.util.concurrent.CompletionService;
  18 import java.util.concurrent.ExecutionException;
  19 import java.util.concurrent.ExecutorCompletionService;
  20 import java.util.concurrent.ScheduledExecutorService;
  21 import java.util.concurrent.TimeUnit;
  22
  23 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  24 import org.opendaylight.controller.md.sal.binding.api.DataTreeChangeListener;
  25 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
  26 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
  27 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  28 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  29 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  30 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  31 import org.opendaylight.groupbasedpolicy.dto.EgKey;
  32 import org.opendaylight.groupbasedpolicy.dto.EpKey;
  33 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.endpoint.EndpointManager;
  34 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.DestinationMapper;
  35 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.EgressNatMapper;
  36 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.ExternalMapper;
  37 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.FlowUtils;
  38 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.GroupTable;
  39 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.IngressNatMapper;
  40 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.OfTable;
  41 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.PolicyEnforcer;
  42 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.PortSecurity;
  43 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.SourceMapper;
  44 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.node.SwitchListener;
  45 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.node.SwitchManager;
  46 import org.opendaylight.groupbasedpolicy.util.DataStoreHelper;
  47 import org.opendaylight.groupbasedpolicy.util.IidFactory;
  48 import org.opendaylight.groupbasedpolicy.util.SingletonTask;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.ofoverlay.rev140528.OfOverlayConfig.LearningMode;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.renderer.rev151103.renderers.renderer.interests.followed.tenants.followed.tenant.FollowedEndpointGroup;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.renderer.rev151103.renderers.renderer.interests.followed.tenants.followed.tenant.FollowedEndpointGroupBuilder;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.resolved.policy.rev150828.ResolvedPolicies;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.resolved.policy.rev150828.resolved.policies.ResolvedPolicy;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId;
  57 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  58 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  59 import org.slf4j.Logger;
  60 import org.slf4j.LoggerFactory;
  61
  62 import com.google.common.base.Function;
  63 import com.google.common.base.Optional;
  64 import com.google.common.collect.ImmutableList;
  65 import com.google.common.util.concurrent.AsyncFunction;
  66 import com.google.common.util.concurrent.Futures;
  67 import com.google.common.util.concurrent.ListenableFuture;
  68
  69 /**
  70  * Manage policies on switches by subscribing to updates from the
  71  * policy resolver and information about endpoints from the endpoint
  72  * registry
  73  */
  74 public class PolicyManager
  75      implements SwitchListener, EndpointListener, DataTreeChangeListener<ResolvedPolicy>, Closeable {
  76     private static final Logger LOG =
  77             LoggerFactory.getLogger(PolicyManager.class);
  78
  79     private short tableOffset;
  80     private static final short TABLEID_PORTSECURITY = 0;
  81     private static final short TABLEID_INGRESS_NAT =  1;
  82     private static final short TABLEID_SOURCE_MAPPER = 2;
  83     private static final short TABLEID_DESTINATION_MAPPER = 3;
  84     private static final short TABLEID_POLICY_ENFORCER = 4;
  85     private static final short TABLEID_EGRESS_NAT = 5;
  86     private static final short TABLEID_EXTERNAL_MAPPER = 6;
  87
  88     private final SwitchManager switchManager;
  89     private final EndpointManager endpointManager;
  90
  91     private final ListenerRegistration<PolicyManager> registerDataTreeChangeListener;
  92
  93     private final ScheduledExecutorService executor;
  94     private final SingletonTask flowUpdateTask;
  95     private final DataBroker dataBroker;
  96
  97     /**
  98      * The delay before triggering the flow update task in response to an
  99      * event in milliseconds.
 100      */
 101     private final static int FLOW_UPDATE_DELAY = 250;
 102
 103     public PolicyManager(DataBroker dataBroker,
 104                          SwitchManager switchManager,
 105                          EndpointManager endpointManager,
 106                          RpcProviderRegistry rpcRegistry,
 107                          ScheduledExecutorService executor,
 108                          short tableOffset) {
 109         super();
 110         this.switchManager = switchManager;
 111         this.executor = executor;
 112         this.dataBroker = dataBroker;
 113         this.tableOffset = tableOffset;
 114         try {
 115             // to validate against model
 116             verifyMaxTableId(tableOffset);
 117         } catch (IllegalArgumentException e) {
 118             throw new IllegalArgumentException("Failed to start OF-Overlay renderer\n."
 119                     + "Max. table ID would be out of range. Check config-subsystem.\n{}", e);
 120         }
 121
 122         if (dataBroker != null) {
 123             registerDataTreeChangeListener = dataBroker.registerDataTreeChangeListener(
 124                     new DataTreeIdentifier<>(LogicalDatastoreType.OPERATIONAL,
 125                             InstanceIdentifier.builder(ResolvedPolicies.class).child(ResolvedPolicy.class).build()),
 126                     this);
 127         } else {
 128             registerDataTreeChangeListener = null;
 129             LOG.error("DataBroker is null. Listener for {} was not registered.",
 130                     ResolvedPolicy.class.getCanonicalName());
 131         }
 132         if (switchManager != null)
 133             switchManager.registerListener(this);
 134         this.endpointManager = endpointManager;
 135         endpointManager.registerListener(this);
 136
 137         flowUpdateTask = new SingletonTask(executor, new FlowUpdateTask());
 138         scheduleUpdate();
 139
 140         LOG.debug("Initialized OFOverlay policy manager");
 141     }
 142
 143     private List<? extends OfTable> createFlowPipeline(OfContext ofCtx) {
 144         // TODO - PORTSECURITY is kept in table 0.
 145         // According to openflow spec,processing on vSwitch always starts from table 0.
 146         // Packets will be droped if table 0 is empty.
 147         // Alternative workaround - table-miss flow entries in table 0.
 148         return ImmutableList.of(new PortSecurity(ofCtx, (short) 0),
 149                                         new GroupTable(ofCtx),
 150                                         new IngressNatMapper(ofCtx, getTABLEID_INGRESS_NAT()),
 151                                         new SourceMapper(ofCtx, getTABLEID_SOURCE_MAPPER()),
 152                                         new DestinationMapper(ofCtx, getTABLEID_DESTINATION_MAPPER()),
 153                                         new PolicyEnforcer(ofCtx, getTABLEID_POLICY_ENFORCER()),
 154                                         new EgressNatMapper(ofCtx, getTABLEID_EGRESS_NAT()),
 155                                         new ExternalMapper(ofCtx, getTABLEID_EXTERNAL_MAPPER())
 156                                         );
 157     }
 158
 159     /**
 160      * @param tableOffset the new offset value
 161      * @return {@link ListenableFuture} to indicate that tables have been synced
 162      */
 163     public ListenableFuture<Void> changeOpenFlowTableOffset(final short tableOffset) {
 164         try {
 165             verifyMaxTableId(tableOffset);
 166         } catch (IllegalArgumentException e) {
 167             LOG.error("Cannot update table offset. Max. table ID would be out of range.\n{}", e);
 168             // TODO - invalid offset value remains in conf DS
 169             // It's not possible to validate offset value by using constrains in model,
 170             // because number of tables in pipeline varies.
 171             return Futures.immediateFuture(null);
 172         }
 173         List<Short> tableIDs = getTableIDs();
 174         this.tableOffset = tableOffset;
 175         return Futures.transform(removeUnusedTables(tableIDs), new Function<Void, Void>() {
 176
 177             @Override
 178             public Void apply(Void tablesRemoved) {
 179                 scheduleUpdate();
 180                 return null;
 181             }
 182         });
 183     }
 184
 185     /**
 186      * @param  tableIDs - IDs of tables to delete
 187      * @return ListenableFuture<Void> - which will be filled when clearing is done
 188      */
 189     private ListenableFuture<Void> removeUnusedTables(final List<Short> tableIDs) {
 190         List<ListenableFuture<Void>> checkList = new ArrayList<>();
 191         final ReadWriteTransaction rwTx = dataBroker.newReadWriteTransaction();
 192         for (Short tableId : tableIDs) {
 193             for (NodeId nodeId : switchManager.getReadySwitches()) {
 194                 final InstanceIdentifier<Table> tablePath = FlowUtils.createTablePath(nodeId, tableId);
 195                 checkList.add(deteleTableIfExists(rwTx, tablePath));
 196             }
 197         }
 198         ListenableFuture<List<Void>> allAsListFuture = Futures.allAsList(checkList);
 199         return Futures.transform(allAsListFuture, new AsyncFunction<List<Void>, Void>() {
 200
 201             @Override
 202             public ListenableFuture<Void> apply(List<Void> readyToSubmit) {
 203                 return rwTx.submit();
 204             }
 205         });
 206     }
 207
 208     private List<Short> getTableIDs() {
 209         List<Short> tableIds = new ArrayList<>();
 210         tableIds.add(getTABLEID_PORTSECURITY());
 211         tableIds.add(getTABLEID_INGRESS_NAT());
 212         tableIds.add(getTABLEID_SOURCE_MAPPER());
 213         tableIds.add(getTABLEID_DESTINATION_MAPPER());
 214         tableIds.add(getTABLEID_POLICY_ENFORCER());
 215         tableIds.add(getTABLEID_EGRESS_NAT());
 216         tableIds.add(getTABLEID_EXTERNAL_MAPPER());
 217         return tableIds;
 218     }
 219
 220     private ListenableFuture<Void> deteleTableIfExists(final ReadWriteTransaction rwTx, final InstanceIdentifier<Table> tablePath){
 221     return Futures.transform(rwTx.read(LogicalDatastoreType.CONFIGURATION, tablePath), new Function<Optional<Table>, Void>() {
 222
 223         @Override
 224         public Void apply(Optional<Table> optTable) {
 225             if(optTable.isPresent()){
 226                 rwTx.delete(LogicalDatastoreType.CONFIGURATION, tablePath);
 227             }
 228             return null;
 229         }});
 230     }
 231
 232     // **************
 233     // SwitchListener
 234     // **************
 235
 236     public short getTABLEID_PORTSECURITY() {
 237         return (short)(tableOffset+TABLEID_PORTSECURITY);
 238     }
 239
 240
 241     public short getTABLEID_INGRESS_NAT() {
 242         return (short)(tableOffset+TABLEID_INGRESS_NAT);
 243     }
 244
 245
 246     public short getTABLEID_SOURCE_MAPPER() {
 247         return (short)(tableOffset+TABLEID_SOURCE_MAPPER);
 248     }
 249
 250
 251     public short getTABLEID_DESTINATION_MAPPER() {
 252         return (short)(tableOffset+TABLEID_DESTINATION_MAPPER);
 253     }
 254
 255
 256     public short getTABLEID_POLICY_ENFORCER() {
 257         return (short)(tableOffset+TABLEID_POLICY_ENFORCER);
 258     }
 259
 260
 261     public short getTABLEID_EGRESS_NAT() {
 262         return (short)(tableOffset+TABLEID_EGRESS_NAT);
 263     }
 264
 265
 266     public short getTABLEID_EXTERNAL_MAPPER() {
 267         return (short)(tableOffset+TABLEID_EXTERNAL_MAPPER);
 268     }
 269
 270
 271     public TableId verifyMaxTableId(short tableOffset) {
 272         return new TableId((short)(tableOffset+TABLEID_EXTERNAL_MAPPER));
 273     }
 274
 275     @Override
 276     public void switchReady(final NodeId nodeId) {
 277         scheduleUpdate();
 278     }
 279
 280     @Override
 281     public void switchRemoved(NodeId sw) {
 282         // XXX TODO purge switch flows
 283         scheduleUpdate();
 284     }
 285
 286     @Override
 287     public void switchUpdated(NodeId sw) {
 288         scheduleUpdate();
 289     }
 290
 291     // ****************
 292     // EndpointListener
 293     // ****************
 294
 295     @Override
 296     public void endpointUpdated(EpKey epKey) {
 297         scheduleUpdate();
 298     }
 299
 300     @Override
 301     public void nodeEndpointUpdated(NodeId nodeId, EpKey epKey){
 302         scheduleUpdate();
 303     }
 304
 305     @Override
 306     public void groupEndpointUpdated(EgKey egKey, EpKey epKey) {
 307         // TODO a renderer should remove followed-EPG and followed-tenant at some point
 308         if (dataBroker == null) {
 309             LOG.error("DataBroker is null. Cannot write followed-epg {}", epKey);
 310             return;
 311         }
 312         WriteTransaction wTx = dataBroker.newWriteOnlyTransaction();
 313         FollowedEndpointGroup followedEpg = new FollowedEndpointGroupBuilder().setId(egKey.getEgId()).build();
 314         wTx.put(LogicalDatastoreType.OPERATIONAL, IidFactory.followedEndpointgroupIid(OFOverlayRenderer.RENDERER_NAME,
 315                 egKey.getTenantId(), egKey.getEgId()), followedEpg, true);
 316         DataStoreHelper.submitToDs(wTx);
 317         scheduleUpdate();
 318     }
 319
 320     // **************
 321     // DataTreeChangeListener<ResolvedPolicy>
 322     // **************
 323
 324     @Override
 325     public void onDataTreeChanged(Collection<DataTreeModification<ResolvedPolicy>> changes) {
 326         scheduleUpdate();
 327     }
 328
 329     // *************
 330     // PolicyManager
 331     // *************
 332
 333     /**
 334      * Set the learning mode to the specified value
 335      * @param learningMode the learning mode to set
 336      */
 337     public void setLearningMode(LearningMode learningMode) {
 338         // No-op for now
 339     }
 340
 341     // **************
 342     // Implementation
 343     // **************
 344
 345     private void scheduleUpdate() {
 346         if (switchManager != null) {
 347             LOG.trace("Scheduling flow update task");
 348             flowUpdateTask.reschedule(FLOW_UPDATE_DELAY, TimeUnit.MILLISECONDS);
 349         }
 350     }
 351
 352     /**
 353      * Update the flows on a particular switch
 354      */
 355     private class SwitchFlowUpdateTask implements Callable<Void> {
 356         private final OfWriter ofWriter;
 357
 358         public SwitchFlowUpdateTask(OfWriter ofWriter) {
 359             this.ofWriter = ofWriter;
 360         }
 361
 362         @Override
 363         public Void call() throws Exception {
 364             OfContext ofCtx = new OfContext(dataBroker, PolicyManager.this, switchManager, endpointManager, executor);
 365             if (ofCtx.getCurrentPolicy() == null)
 366                 return null;
 367             List<? extends OfTable> flowPipeline = createFlowPipeline(ofCtx);
 368             for (NodeId node : switchManager.getReadySwitches()) {
 369                 for (OfTable table : flowPipeline) {
 370                     try {
 371                         table.sync(node, ofWriter);
 372                     } catch (Exception e) {
 373                         LOG.error("Failed to write Openflow table {}",
 374                                 table.getClass().getSimpleName(), e);
 375                     }
 376                 }
 377             }
 378             return null;
 379         }
 380     }
 381
 382     /**
 383      * Update all flows on all switches as needed.  Note that this will block
 384      * one of the threads on the executor.
 385      */
 386     private class FlowUpdateTask implements Runnable {
 387         @Override
 388         public void run() {
 389             LOG.debug("Beginning flow update task");
 390
 391             CompletionService<Void> ecs
 392                 = new ExecutorCompletionService<>(executor);
 393             int n = 0;
 394
 395             OfWriter ofWriter = new OfWriter();
 396
 397             SwitchFlowUpdateTask swut = new SwitchFlowUpdateTask(ofWriter);
 398             ecs.submit(swut);
 399             n+=1;
 400
 401             for (int i = 0; i < n; i++) {
 402                 try {
 403                     ecs.take().get();
 404                     ofWriter.commitToDataStore(dataBroker);
 405                 } catch (InterruptedException | ExecutionException e) {
 406                     LOG.error("Failed to update flow tables", e);
 407                 }
 408             }
 409             LOG.debug("Flow update completed");
 410         }
 411     }
 412
 413     @Override
 414     public void close() throws IOException {
 415         if (registerDataTreeChangeListener != null)
 416             registerDataTreeChangeListener.close();
 417         // TODO unregister classifier and action instance validators
 418     }
 419
 420 }