sxp-mapper/src/main/java/org/opendaylight/groupbasedpolicy/sxp/mapper/impl/listen/MasterDatabaseBindingListenerImpl.java

   1 /*
   2  * Copyright (c) 2016 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.groupbasedpolicy.sxp.mapper.impl.listen;
   9
  10 import java.util.Collection;
  11
  12 import javax.annotation.Nonnull;
  13
  14 import org.opendaylight.controller.md.sal.binding.api.ClusteredDataTreeChangeListener;
  15 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  16 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
  17 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
  18 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  19 import org.opendaylight.groupbasedpolicy.sxp.mapper.api.DSAsyncDao;
  20 import org.opendaylight.groupbasedpolicy.sxp.mapper.api.MasterDatabaseBindingListener;
  21 import org.opendaylight.groupbasedpolicy.sxp.mapper.api.SimpleCachedDao;
  22 import org.opendaylight.groupbasedpolicy.sxp.mapper.api.SxpMapperReactor;
  23 import org.opendaylight.groupbasedpolicy.sxp.mapper.impl.util.EPTemplateUtil;
  24 import org.opendaylight.groupbasedpolicy.sxp.mapper.impl.util.L3EPServiceUtil;
  25 import org.opendaylight.groupbasedpolicy.sxp.mapper.impl.util.SxpListenerUtil;
  26 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpPrefix;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.groupbasedpolicy.sxp.mapper.model.rev160302.sxp.mapper.EndpointForwardingTemplateBySubnet;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.groupbasedpolicy.sxp.mapper.model.rev160302.sxp.mapper.EndpointPolicyTemplateBySgt;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.database.rev160308.Sgt;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.database.rev160308.master.database.fields.MasterDatabaseBinding;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.SxpNodeIdentity;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.network.topology.topology.node.SxpDomains;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.network.topology.topology.node.sxp.domains.SxpDomain;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.sxp.databases.fields.MasterDatabase;
  35 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  36 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  37 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  38 import org.opendaylight.yangtools.yang.common.RpcError;
  39 import org.opendaylight.yangtools.yang.common.RpcResult;
  40 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 import com.google.common.base.Optional;
  45 import com.google.common.base.Preconditions;
  46 import com.google.common.util.concurrent.AsyncFunction;
  47 import com.google.common.util.concurrent.FutureCallback;
  48 import com.google.common.util.concurrent.Futures;
  49 import com.google.common.util.concurrent.ListenableFuture;
  50
  51 /**
  52  * listens to sxp master database and propagates change events for further processing
  53  */
  54 public class MasterDatabaseBindingListenerImpl implements MasterDatabaseBindingListener {
  55
  56     private static final Logger LOG = LoggerFactory.getLogger(MasterDatabaseBindingListenerImpl.class);
  57
  58     private static final FutureCallback<RpcResult<Void>> RPC_POLICY_RESULT_FUTURE_CALLBACK =
  59             L3EPServiceUtil.createFailureLoggingCallback("failed to read epPolicyTemplate");
  60
  61     public static final FutureCallback<RpcResult<Void>> RPC_FW_RESULT_FUTURE_CALLBACK =
  62             L3EPServiceUtil.createFailureLoggingCallback("failed to read epForwardingTemplate");
  63
  64     private final SxpMapperReactor sxpMapperReactor;
  65     private final SimpleCachedDao<IpPrefix, MasterDatabaseBinding> masterDBBindingDaoCached;
  66     private final DSAsyncDao<Sgt, EndpointPolicyTemplateBySgt> epPolicyTemplateDao;
  67     private final DSAsyncDao<IpPrefix, EndpointForwardingTemplateBySubnet> epForwardingTemplateDao;
  68
  69     private final ListenerRegistration<? extends ClusteredDataTreeChangeListener> listenerRegistration;
  70     private final InstanceIdentifier<MasterDatabaseBinding> sxpDbPath;
  71
  72     public MasterDatabaseBindingListenerImpl(final DataBroker dataBroker,
  73                                              final SxpMapperReactor sxpMapperReactor,
  74                                              final SimpleCachedDao<IpPrefix, MasterDatabaseBinding> masterDBBindingDaoCached,
  75                                              final DSAsyncDao<Sgt, EndpointPolicyTemplateBySgt> epPolicyTemplateDao,
  76                                              final DSAsyncDao<IpPrefix, EndpointForwardingTemplateBySubnet> epForwardingTemplateDao) {
  77         this.sxpMapperReactor = Preconditions.checkNotNull(sxpMapperReactor);
  78         this.masterDBBindingDaoCached = Preconditions.checkNotNull(masterDBBindingDaoCached);
  79         this.epPolicyTemplateDao = Preconditions.checkNotNull(epPolicyTemplateDao);
  80         this.epForwardingTemplateDao = Preconditions.checkNotNull(epForwardingTemplateDao);
  81         sxpDbPath = MasterDatabaseBindingListener.SXP_TOPOLOGY_PATH
  82                 .child(Node.class)
  83                 .augmentation(SxpNodeIdentity.class)
  84                 .child(SxpDomains.class)
  85                 .child(SxpDomain.class)
  86                 .child(MasterDatabase.class)
  87                 .child(MasterDatabaseBinding.class);
  88
  89         final DataTreeIdentifier<MasterDatabaseBinding> dataTreeIdentifier = new DataTreeIdentifier<>(
  90                 LogicalDatastoreType.OPERATIONAL, sxpDbPath);
  91         listenerRegistration = dataBroker.registerDataTreeChangeListener(dataTreeIdentifier, this);
  92         LOG.debug("started listening to {}", sxpDbPath);
  93     }
  94
  95     @Override
  96     public void onDataTreeChanged(@Nonnull final Collection<DataTreeModification<MasterDatabaseBinding>> collection) {
  97         for (DataTreeModification<MasterDatabaseBinding> change : collection) {
  98             LOG.trace("received modification: {} -> {}", change.getRootPath(), change.getRootNode().getModificationType());
  99             // update cached dao
 100             final MasterDatabaseBinding sxpMasterDBItem = change.getRootNode().getDataAfter();
 101             if (sxpMasterDBItem == null) {
 102                 //TODO: cover sgt-ip mapping removal
 103                 LOG.debug("ip-sgt mapping was removed - NOOP: {}",
 104                         change.getRootPath().getRootIdentifier().firstKeyOf(MasterDatabaseBinding.class));
 105             } else {
 106                 final IpPrefix ipPrefixKey = sxpMasterDBItem.getIpPrefix();
 107                 SxpListenerUtil.updateCachedDao(masterDBBindingDaoCached, ipPrefixKey, change);
 108                 processWithEPTemplates(sxpMasterDBItem);
 109             }
 110         }
 111     }
 112
 113     private void processWithEPTemplates(final MasterDatabaseBinding sxpMasterDBItem) {
 114         final ListenableFuture<Optional<EndpointPolicyTemplateBySgt>> epPolicyTemplateFuture =
 115                 epPolicyTemplateDao.read(sxpMasterDBItem.getSecurityGroupTag());
 116
 117         final ListenableFuture<Optional<EndpointForwardingTemplateBySubnet>> epForwardingTemplateFuture =
 118                 epForwardingTemplateDao.read(sxpMasterDBItem.getIpPrefix());
 119
 120         final ListenableFuture<EPTemplateUtil.OptionalMutablePair<EndpointPolicyTemplateBySgt, EndpointForwardingTemplateBySubnet>> compositeRead
 121                 = EPTemplateUtil.compositeRead(epPolicyTemplateFuture, epForwardingTemplateFuture);
 122
 123         final ListenableFuture<RpcResult<Void>> rpcResult = Futures.transform(compositeRead,
 124                 new AsyncFunction<EPTemplateUtil.OptionalMutablePair<EndpointPolicyTemplateBySgt, EndpointForwardingTemplateBySubnet>, RpcResult<Void>>() {
 125                     @Override
 126                     public ListenableFuture<RpcResult<Void>> apply(final EPTemplateUtil.OptionalMutablePair<EndpointPolicyTemplateBySgt, EndpointForwardingTemplateBySubnet> input) throws Exception {
 127                         final ListenableFuture<RpcResult<Void>> result;
 128                         if (input == null) {
 129                             LOG.debug("no ep*Templates available for sgt/ip-prefix: {}/{}",
 130                                     sxpMasterDBItem.getSecurityGroupTag(),
 131                                     sxpMasterDBItem.getIpPrefix());
 132                             result = RpcResultBuilder.<Void>failed()
 133                                     .withError(RpcError.ErrorType.APPLICATION,
 134                                             "no ep-templates available for" + sxpMasterDBItem)
 135                                     .buildFuture();
 136
 137                         } else if (!input.getLeft().isPresent()) {
 138                             LOG.debug("no epPolicyTemplate available for sgt: {}", sxpMasterDBItem.getSecurityGroupTag());
 139                             result = RpcResultBuilder.<Void>failed()
 140                                     .withError(RpcError.ErrorType.APPLICATION,
 141                                             "no epPolicyTemplate available for " + sxpMasterDBItem)
 142                                     .buildFuture();
 143                         } else if (!input.getRight().isPresent()) {
 144                             LOG.debug("no epForwardingTemplate available for ip-prefix: {}",
 145                                     sxpMasterDBItem.getIpPrefix());
 146                             result = RpcResultBuilder.<Void>failed()
 147                                     .withError(RpcError.ErrorType.APPLICATION,
 148                                             "no epForwardingTemplate available for " + sxpMasterDBItem)
 149                                     .buildFuture();
 150                         } else {
 151                             LOG.trace("processing sxpMasterDB event and epPolicyTemplate for sgt/ip-prefix: {}/{}",
 152                                     sxpMasterDBItem.getSecurityGroupTag(),
 153                                     sxpMasterDBItem.getIpPrefix());
 154                             result = sxpMapperReactor.processTemplatesAndSxpMasterDB(input.getLeft().get(),
 155                                     input.getRight().get(), sxpMasterDBItem);
 156                 }
 157                 return result;
 158             }
 159         });
 160
 161         Futures.addCallback(rpcResult, RPC_POLICY_RESULT_FUTURE_CALLBACK);
 162     }
 163
 164     @Override
 165     public void close() throws Exception {
 166         LOG.debug("closing listener registration to {}", sxpDbPath);
 167         listenerRegistration.close();
 168     }
 169 }