applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/AbstractListeningCommiter.java

   1 /*
   2  * Copyright (c) 2014, 2017 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 package org.opendaylight.openflowplugin.applications.frm.impl;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.util.concurrent.FutureCallback;
  12 import com.google.common.util.concurrent.Futures;
  13 import com.google.common.util.concurrent.MoreExecutors;
  14 import java.util.Collection;
  15 import org.eclipse.jdt.annotation.Nullable;
  16 import org.opendaylight.mdsal.binding.api.DataBroker;
  17 import org.opendaylight.mdsal.binding.api.DataObjectModification;
  18 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  19 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  20 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  21 import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesCommiter;
  22 import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
  23 import org.opendaylight.openflowplugin.applications.frm.NodeConfigurator;
  24 import org.opendaylight.serviceutils.srm.RecoverableListener;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  26 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  27 import org.opendaylight.yangtools.yang.binding.DataObject;
  28 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * AbstractChangeListner implemented basic {@link org.opendaylight.mdsal.binding.api.DataTreeModification}
  34  * processing for flow node subDataObject (flows, groups and meters).
  35  */
  36 public abstract class AbstractListeningCommiter<T extends DataObject>
  37         implements ForwardingRulesCommiter<T>, RecoverableListener {
  38
  39     private static final Logger LOG = LoggerFactory.getLogger(AbstractListeningCommiter.class);
  40     final ForwardingRulesManager provider;
  41     NodeConfigurator nodeConfigurator;
  42     protected final DataBroker dataBroker;
  43     protected final ListenerRegistrationHelper registrationHelper;
  44     protected ListenerRegistration<AbstractListeningCommiter> listenerRegistration;
  45
  46     public AbstractListeningCommiter(final ForwardingRulesManager provider, final DataBroker dataBroker,
  47                                      final ListenerRegistrationHelper registrationHelper) {
  48         this.provider = Preconditions.checkNotNull(provider, "ForwardingRulesManager can not be null!");
  49         this.nodeConfigurator = Preconditions.checkNotNull(provider.getNodeConfigurator(),
  50                 "NodeConfigurator can not be null!");
  51         this.dataBroker = Preconditions.checkNotNull(dataBroker, "DataBroker can not be null!");
  52         this.registrationHelper = Preconditions.checkNotNull(registrationHelper, "registrationHelper can not be null!");
  53         registerListener();
  54         provider.addRecoverableListener(this);
  55     }
  56
  57     @SuppressWarnings("checkstyle:IllegalCatch")
  58     @Override
  59     public void onDataTreeChanged(Collection<DataTreeModification<T>> changes) {
  60         Preconditions.checkNotNull(changes, "Changes may not be null!");
  61         LOG.trace("Received data changes :{}", changes);
  62
  63         for (DataTreeModification<T> change : changes) {
  64             final InstanceIdentifier<T> key = change.getRootPath().getRootIdentifier();
  65             final DataObjectModification<T> mod = change.getRootNode();
  66             final InstanceIdentifier<FlowCapableNode> nodeIdent =
  67                     key.firstIdentifierOf(FlowCapableNode.class);
  68             try {
  69                 if (preConfigurationCheck(nodeIdent)) {
  70                     switch (mod.getModificationType()) {
  71                         case DELETE:
  72                             remove(key, mod.getDataBefore(), nodeIdent);
  73                             break;
  74                         case SUBTREE_MODIFIED:
  75                             update(key, mod.getDataBefore(), mod.getDataAfter(), nodeIdent);
  76                             break;
  77                         case WRITE:
  78                             if (mod.getDataBefore() == null) {
  79                                 add(key, mod.getDataAfter(), nodeIdent);
  80                             } else {
  81                                 update(key, mod.getDataBefore(), mod.getDataAfter(), nodeIdent);
  82                             }
  83                             break;
  84                         default:
  85                             throw new
  86                                     IllegalArgumentException("Unhandled modification type "
  87                                     + mod.getModificationType());
  88                     }
  89                 } else {
  90                     if (provider.isStaleMarkingEnabled()) {
  91                         LOG.info("Stale-Marking ENABLED and switch {} is NOT connected, storing stale entities",
  92                                 nodeIdent.toString());
  93                         // Switch is NOT connected
  94                         switch (mod.getModificationType()) {
  95                             case DELETE:
  96                                 createStaleMarkEntity(key, mod.getDataBefore(), nodeIdent);
  97                                 break;
  98                             case SUBTREE_MODIFIED:
  99                                 break;
 100                             case WRITE:
 101                                 break;
 102                             default:
 103                                 throw new
 104                                         IllegalArgumentException("Unhandled modification type "
 105                                         + mod.getModificationType());
 106                         }
 107                     }
 108                 }
 109             } catch (RuntimeException e) {
 110                 LOG.error("Failed to handle event {} key {} due to error ", mod.getModificationType(), key, e);
 111             }
 112         }
 113     }
 114
 115     @Override
 116     public void registerListener() {
 117         final DataTreeIdentifier<T> treeId =
 118                 DataTreeIdentifier.create(LogicalDatastoreType.CONFIGURATION, getWildCardPath());
 119         Futures.addCallback(registrationHelper.checkedRegisterListener(treeId, this),
 120                 new FutureCallback<ListenerRegistration<AbstractListeningCommiter>>() {
 121                     @Override
 122                     public void onSuccess(
 123                             @Nullable ListenerRegistration<AbstractListeningCommiter> flowListenerRegistration) {
 124                         LOG.info("{} registered successfully", flowListenerRegistration.getInstance());
 125                         listenerRegistration = flowListenerRegistration;
 126                     }
 127
 128                     @Override
 129                     public void onFailure(Throwable throwable) {
 130                         LOG.error("Registration failed ", throwable);
 131                     }
 132                 }, MoreExecutors.directExecutor());
 133     }
 134
 135     /**
 136      * Method return wildCardPath for Listener registration
 137      * and for identify the correct KeyInstanceIdentifier from data.
 138      */
 139     protected abstract InstanceIdentifier<T> getWildCardPath();
 140
 141     private boolean preConfigurationCheck(final InstanceIdentifier<FlowCapableNode> nodeIdent) {
 142         Preconditions.checkNotNull(nodeIdent, "FlowCapableNode identifier can not be null!");
 143         // In single node cluster, node should be in local cache before we get any flow/group/meter
 144         // data change event from data store. So first check should pass.
 145         // In case of 3-node cluster, when shard leader changes, clustering will send blob of data
 146         // present in operational data store and config data store. So ideally local node cache
 147         // should get populated. But to handle a scenario where flow request comes before the blob
 148         // of config/operational data gets processes, it won't find node in local cache and it will
 149         // skip the flow/group/meter operational. This requires an addition check, where it reads
 150         // node from operational data store and if it's present it calls flowNodeConnected to explicitly
 151         // trigger the event of new node connected.
 152         return provider.isNodeOwner(nodeIdent);
 153     }
 154 }