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

   1 package org.opendaylight.groupbasedpolicy.renderer.ofoverlay;
   2
   3 import java.util.HashSet;
   4 import java.util.List;
   5 import java.util.Map.Entry;
   6 import java.util.Set;
   7 import java.util.concurrent.ConcurrentHashMap;
   8 import java.util.concurrent.ConcurrentMap;
   9 import java.util.concurrent.ExecutorService;
  10 import java.util.concurrent.Future;
  11
  12 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  13 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
  14 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  15 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  16 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
  17 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  18 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  19 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.sf.ChainAction;
  20 import org.opendaylight.groupbasedpolicy.resolver.PolicyResolver;
  21 import org.opendaylight.sfc.provider.SfcProviderRpc;
  22 import org.opendaylight.sfc.provider.api.SfcProviderServiceChainAPI;
  23 import org.opendaylight.sfc.provider.api.SfcProviderServicePathAPI;
  24 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.sfc.common.rev151017.SfcName;
  25 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.sfc.rsp.rev140701.ReadRenderedServicePathFirstHopInputBuilder;
  26 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.sfc.rsp.rev140701.ReadRenderedServicePathFirstHopOutput;
  27 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.sfc.rsp.rev140701.rendered.service.path.first.hop.info.RenderedServicePathFirstHop;
  28 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.sfc.sfc.rev140701.service.function.chain.grouping.ServiceFunctionChain;
  29 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.sfc.sfp.rev140701.ServiceFunctionPaths;
  30 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.sfc.sfp.rev140701.service.function.paths.ServiceFunctionPath;
  31 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.IpAddress;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.common.rev140421.ActionDefinitionId;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.SubjectFeatureDefinitions;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.Tenants;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.subject.feature.definitions.ActionDefinition;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.subject.feature.definitions.ActionDefinitionKey;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.subject.feature.instance.ParameterValue;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.tenants.Tenant;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.tenants.tenant.SubjectFeatureInstances;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.groupbasedpolicy.policy.rev140421.tenants.tenant.subject.feature.instances.ActionInstance;
  41 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  42 import org.opendaylight.yangtools.yang.binding.DataObject;
  43 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  44 import org.opendaylight.yangtools.yang.common.RpcResult;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 import com.google.common.base.Optional;
  49 import com.google.common.util.concurrent.FutureCallback;
  50 import com.google.common.util.concurrent.Futures;
  51 import com.google.common.util.concurrent.ListenableFuture;
  52
  53 /**
  54  * Manage the state exchanged with SFC
  55  *
  56  * For the Proof of Concept, this manages the
  57  * RenderedServicePathFirstHop elements that
  58  * are retrieved from SFC.
  59  *
  60  */
  61 public class SfcManager implements AutoCloseable, DataChangeListener {
  62     private static final Logger LOG =
  63             LoggerFactory.getLogger(SfcManager.class);
  64
  65     private final DataBroker dataBroker;
  66     private final ExecutorService executor;
  67     private final InstanceIdentifier<ActionInstance> allActionInstancesIid;
  68     private final ListenerRegistration<DataChangeListener> actionListener;
  69
  70     /*
  71      * local cache of the RSP first hops that we've requested from SFC,
  72      * keyed by RSP name
  73      */
  74     private final ConcurrentMap<String, RenderedServicePathFirstHop> rspMap;
  75
  76     /*
  77      *  TODO: these two String defs should move to the common
  78      *        "chain" action, once we have it.
  79      */
  80     // the chain action
  81     public static final String SFC_CHAIN_ACTION = "chain";
  82     // the parameter used for storing the chain name
  83     public static final String SFC_CHAIN_NAME = "sfc-chain-name";
  84
  85     private static enum ActionState {
  86         ADD("add"),
  87         CHANGE("change"),
  88         DELETE("delete");
  89
  90         private String state;
  91
  92         ActionState(String state) {
  93             this.state = state;
  94         }
  95
  96         @Override
  97         public String toString() {
  98             return this.state;
  99         }
 100     }
 101
 102
 103     public SfcManager(DataBroker dataBroker,
 104                       PolicyResolver policyResolver,
 105                       RpcProviderRegistry rpcRegistry,
 106                       ExecutorService executor) {
 107         this.dataBroker = dataBroker;
 108         this.executor = executor;
 109         /*
 110          * Use thread-safe type only because we use an executor
 111          */
 112         this.rspMap = new ConcurrentHashMap<String, RenderedServicePathFirstHop>();
 113
 114         /*
 115          * For now, listen to all changes in rules
 116          */
 117         allActionInstancesIid =
 118                 InstanceIdentifier.builder(Tenants.class)
 119                     .child(Tenant.class)
 120                     .child(SubjectFeatureInstances.class)
 121                     .child(ActionInstance.class)
 122                     .build();
 123         actionListener = dataBroker.registerDataChangeListener(LogicalDatastoreType.CONFIGURATION,
 124                 allActionInstancesIid, this, DataChangeScope.ONE);
 125         LOG.debug("SfcManager: Started");
 126     }
 127
 128     public Set<IpAddress> getSfcSourceIps() {
 129         if (rspMap.isEmpty()) return null;
 130
 131         Set<IpAddress> ipAddresses = new HashSet<IpAddress>();
 132         for (RenderedServicePathFirstHop rsp: rspMap.values()) {
 133             if (rsp.getIp() != null) {
 134                 ipAddresses.add(rsp.getIp());
 135             }
 136         }
 137         if (ipAddresses.isEmpty()) return null;
 138         return ipAddresses;
 139     }
 140
 141     @Override
 142     public void onDataChanged(
 143             AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> actionInstanceNotification) {
 144
 145         for (DataObject dao : actionInstanceNotification.getCreatedData().values()) {
 146             if (dao instanceof ActionInstance) {
 147                 ActionInstance ai = (ActionInstance)dao;
 148                 LOG.debug("New ActionInstance created");
 149                 executor.execute(new MatchActionDefTask(ai, null,
 150                         ActionState.ADD));
 151             }
 152         }
 153
 154         for (InstanceIdentifier<?> iid : actionInstanceNotification.getRemovedPaths()) {
 155             DataObject old = actionInstanceNotification.getOriginalData().get(iid);
 156             if (old instanceof ActionInstance) {
 157                 ActionInstance ai = (ActionInstance)old;
 158                 executor.execute(new MatchActionDefTask(null, ai,
 159                         ActionState.DELETE));
 160             }
 161         }
 162
 163         for (Entry<InstanceIdentifier<?>, DataObject> entry:
 164             actionInstanceNotification.getUpdatedData().entrySet()) {
 165             DataObject dao = entry.getValue();
 166             if (dao instanceof ActionInstance) {
 167                 ActionInstance nai = (ActionInstance)dao;
 168                 ActionInstance oai = null;
 169                 InstanceIdentifier<?> iid = entry.getKey();
 170                 DataObject orig = actionInstanceNotification.getOriginalData().get(iid);
 171                 if (orig != null) {
 172                     oai = (ActionInstance)orig;
 173                     /*
 174                      * We may have some cleanup here.  If the reference to
 175                      * the Action Definition changed, or if the Action Instance's
 176                      * chain parameter  then we're no longer
 177                      * an action, and we may need to remove the RSP.
 178                      */
 179                 }
 180
 181                 executor.execute(new MatchActionDefTask(nai, oai,
 182                         ActionState.CHANGE));
 183             }
 184         }
 185     }
 186
 187     /**
 188      * Private internal class that gets the action definition
 189      * referenced by the instance. If the definition has an
 190      * action of "chain" (or whatever we decide to use
 191      * here), then we need to invoke the SFC API to go
 192      * get the chain information, which we'll eventually
 193      * use during policy resolution.
 194      *
 195      */
 196     private class MatchActionDefTask implements Runnable,
 197                      FutureCallback<Optional<ActionDefinition>> {
 198         private final ActionState state;
 199         private final ActionInstance actionInstance;
 200         private final ActionInstance originalInstance;
 201         private final InstanceIdentifier<ActionDefinition> adIid;
 202         private final ActionDefinitionId id;
 203
 204         public MatchActionDefTask(ActionInstance actionInstance,
 205                 ActionInstance originalInstance, ActionState state) {
 206             this.actionInstance = actionInstance;
 207             this.originalInstance = originalInstance;
 208             if (actionInstance != null) {
 209                 this.id = actionInstance.getActionDefinitionId();
 210             } else {
 211                 this.id = null;
 212             }
 213             this.state = state;
 214
 215             adIid = InstanceIdentifier.builder(SubjectFeatureDefinitions.class)
 216                                       .child(ActionDefinition.class,
 217                                              new ActionDefinitionKey(this.id))
 218                                       .build();
 219
 220         }
 221
 222         /**
 223          * Create read transaction with callback to look up
 224          * the Action Definition that the Action Instance
 225          * references.
 226          */
 227         @Override
 228         public void run() {
 229             ReadOnlyTransaction rot = dataBroker.newReadOnlyTransaction();
 230             ListenableFuture<Optional<ActionDefinition>> dao =
 231                     rot.read(LogicalDatastoreType.OPERATIONAL, adIid);
 232             Futures.addCallback(dao, this, executor);
 233
 234         }
 235
 236         @Override
 237         public void onFailure(Throwable arg0) {
 238             LOG.error("Failure reading ActionDefinition {}", id.getValue());
 239         }
 240
 241         /**
 242          * An Action Definition exists - now we need to see
 243          * if the Action Definition is for a chain action,
 244          * and implement the appropriate behavior. If it's
 245          * not a chain action, then we can ignore it.
 246          *
 247          * @param dao
 248          */
 249         @Override
 250         public void onSuccess(Optional<ActionDefinition> dao) {
 251             LOG.debug("Found ActionDefinition {}", id.getValue());
 252             if (!dao.isPresent()) return;
 253
 254             ActionDefinition ad = dao.get();
 255             if (ad.getId().getValue().equals(ChainAction.ID.getValue())) {
 256                 /*
 257                  * We have the state we need:
 258                  *  1) it's a "CHAIN" action
 259                  *  2) the name is defined in the ActionInstance
 260                  */
 261                 switch (state) {
 262                 case ADD:
 263                     /*
 264                      * Go get the RSP First Hop
 265                      */
 266                     getSfcChain();
 267                     break;
 268                 case CHANGE:
 269                     /*
 270                      * We only care if the named chain changes
 271                      */
 272                     changeSfcRsp();
 273                     break;
 274                 case DELETE:
 275                     /*
 276                      * If the instance is deleted, we need to remove
 277                      * it from our map.
 278                      */
 279                     deleteSfcRsp();
 280                     break;
 281                 default:
 282                     break;
 283                 }
 284             }
 285         }
 286
 287         private ParameterValue getChainNameParameter(List<ParameterValue> pvl) {
 288             if (pvl == null) return null;
 289             for (ParameterValue pv: actionInstance.getParameterValue()) {
 290                 if (pv.getName().getValue().equals(SFC_CHAIN_NAME)) {
 291                     return pv;
 292                 }
 293             }
 294             return null;
 295         }
 296
 297         private void changeSfcRsp() {
 298             ParameterValue newPv =
 299                     getChainNameParameter(actionInstance.getParameterValue());
 300             ParameterValue origPv =
 301                     getChainNameParameter(originalInstance.getParameterValue());
 302             if (!newPv.getStringValue().equals(origPv.getStringValue())) {
 303                 if (rspMap.containsKey(origPv.getStringValue())) {
 304                     /*
 305                      * Flow cleanup will happen as part of the
 306                      * resolved policy
 307                      *
 308                      * TODO: can we guarantee that this
 309                      *       happens after we remove the RSP?).
 310                      */
 311                     rspMap.remove(origPv.getStringValue());
 312                 }
 313                 addSfcRsp();
 314             }
 315         }
 316
 317         private void deleteSfcRsp() {
 318             ParameterValue pv =
 319                     getChainNameParameter(originalInstance.getParameterValue());
 320             if (pv == null) return;
 321             rspMap.remove(pv.getStringValue());
 322         }
 323
 324         /**
 325          * Get the RenderedServicePathFirstHop from SFC
 326          *
 327          * TODO: what if SFC state isn't available at the time of
 328          *       this call, but becomes available later?  Do we want
 329          *       or need some sort of notification handler for this?
 330          */
 331         private void addSfcRsp() {
 332             ParameterValue pv =
 333                     getChainNameParameter(actionInstance.getParameterValue());
 334             if (pv == null) return;
 335
 336             LOG.trace("Invoking RPC for chain {}", pv.getStringValue());
 337             ReadRenderedServicePathFirstHopInputBuilder builder =
 338                 new ReadRenderedServicePathFirstHopInputBuilder()
 339                        .setName(pv.getStringValue());
 340             // TODO: make async
 341             Future<RpcResult<ReadRenderedServicePathFirstHopOutput>> result =
 342                 SfcProviderRpc.getSfcProviderRpc()
 343                               .readRenderedServicePathFirstHop(builder.build());
 344
 345             try {
 346                 RpcResult<ReadRenderedServicePathFirstHopOutput> output =
 347                         result.get();
 348                 if (output.isSuccessful()) {
 349                     LOG.trace("RPC for chain {} succeeded!", pv.getStringValue());
 350                     RenderedServicePathFirstHop rspFirstHop =
 351                         output.getResult().getRenderedServicePathFirstHop();
 352                     /*
 353                      * We won't retry installation in the map
 354                      * because the presumption is it's either
 355                      * the same object or contain the same
 356                      * state.
 357                      */
 358                     rspMap.putIfAbsent(pv.getStringValue(), rspFirstHop);
 359                 }
 360             } catch (Exception e) {
 361                 LOG.warn("Failed ReadRenderedServicePathFirstHop RPC: {}", e);
 362                 // TODO: proper exception handling
 363             }
 364         }
 365
 366         private void getSfcChain() {
 367             ParameterValue pv =
 368                     getChainNameParameter(actionInstance.getParameterValue());
 369             if (pv == null) return;
 370
 371             LOG.trace("Invoking RPC for chain {}", pv.getStringValue());
 372             SfcName chainName=new SfcName(pv.getStringValue());
 373             ServiceFunctionChain chain = SfcProviderServiceChainAPI.readServiceFunctionChain(chainName);
 374             ServiceFunctionPaths paths = SfcProviderServicePathAPI.readAllServiceFunctionPaths();
 375             for(ServiceFunctionPath path: paths.getServiceFunctionPath()) {
 376                 if(path.getServiceChainName().equals(chainName)) {
 377                     LOG.info("Found path {} for chain {}",path.getName(),path.getServiceChainName());
 378                 }
 379             }
 380         }
 381     }
 382
 383     /**
 384      * Return the first hop information for the Rendered Service Path
 385      *
 386      * @param rspName the Rendered Service Path
 387      * @return the first hop information for the Rendered Service Path
 388      */
 389     public RenderedServicePathFirstHop getRspFirstHop(String rspName) {
 390         return rspMap.get(rspName);
 391     }
 392
 393     @Override
 394     public void close() throws Exception {
 395         if (actionListener != null) actionListener.close();
 396
 397     }
 398 }
 399