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