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

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