sxp-integration/sxp-ep-provider/src/main/java/org/opendaylight/groupbasedpolicy/sxp/ep/provider/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
   9 package org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.listen;
  10
  11 import com.google.common.base.Optional;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.util.concurrent.AsyncFunction;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import java.util.Collection;
  18 import javax.annotation.Nonnull;
  19 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  20 import org.opendaylight.controller.md.sal.binding.api.ClusteredDataTreeChangeListener;
  21 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
  22 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
  23 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  24 import org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.DSAsyncDao;
  25 import org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.MasterDatabaseBindingListener;
  26 import org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.SimpleCachedDao;
  27 import org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.SxpMapperReactor;
  28 import org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.util.EPTemplateUtil;
  29 import org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.util.L3EPServiceUtil;
  30 import org.opendaylight.groupbasedpolicy.sxp.ep.provider.impl.util.SxpListenerUtil;
  31 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpPrefix;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.groupbasedpolicy.sxp.integration.sxp.ep.provider.model.rev160302.sxp.ep.mapper.EndpointForwardingTemplateBySubnet;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.groupbasedpolicy.sxp.integration.sxp.ep.provider.model.rev160302.sxp.ep.mapper.EndpointPolicyTemplateBySgt;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.database.rev160308.Sgt;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.database.rev160308.master.database.fields.MasterDatabaseBinding;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.SxpNodeIdentity;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.network.topology.topology.node.SxpDomains;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.network.topology.topology.node.sxp.domains.SxpDomain;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.sxp.node.rev160308.sxp.databases.fields.MasterDatabase;
  40 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  41 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  42 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  43 import org.opendaylight.yangtools.yang.common.RpcError;
  44 import org.opendaylight.yangtools.yang.common.RpcResult;
  45 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  46 import org.slf4j.Logger;
  47 import org.slf4j.LoggerFactory;
  48
  49 /**
  50  * listens to sxp master database and propagates change events for further processing
  51  */
  52 public class MasterDatabaseBindingListenerImpl implements MasterDatabaseBindingListener {
  53
  54     private static final Logger LOG = LoggerFactory.getLogger(MasterDatabaseBindingListenerImpl.class);
  55
  56     private static final FutureCallback<RpcResult<Void>> RPC_POLICY_RESULT_FUTURE_CALLBACK =
  57             L3EPServiceUtil.createFailureLoggingCallback("failed to read epPolicyTemplate");
  58
  59     public static final FutureCallback<RpcResult<Void>> RPC_FW_RESULT_FUTURE_CALLBACK =
  60             L3EPServiceUtil.createFailureLoggingCallback("failed to read epForwardingTemplate");
  61
  62     private final SxpMapperReactor sxpMapperReactor;
  63     private final SimpleCachedDao<IpPrefix, MasterDatabaseBinding> masterDBBindingDaoCached;
  64     private final DSAsyncDao<Sgt, EndpointPolicyTemplateBySgt> epPolicyTemplateDao;
  65     private final DSAsyncDao<IpPrefix, EndpointForwardingTemplateBySubnet> epForwardingTemplateDao;
  66
  67     private final ListenerRegistration<? extends ClusteredDataTreeChangeListener> listenerRegistration;
  68     private final InstanceIdentifier<MasterDatabaseBinding> sxpDbPath;
  69
  70     public MasterDatabaseBindingListenerImpl(final DataBroker dataBroker,
  71                                              final SxpMapperReactor sxpMapperReactor,
  72                                              final SimpleCachedDao<IpPrefix, MasterDatabaseBinding> masterDBBindingDaoCached,
  73                                              final DSAsyncDao<Sgt, EndpointPolicyTemplateBySgt> epPolicyTemplateDao,
  74                                              final DSAsyncDao<IpPrefix, EndpointForwardingTemplateBySubnet> epForwardingTemplateDao) {
  75         this.sxpMapperReactor = Preconditions.checkNotNull(sxpMapperReactor);
  76         this.masterDBBindingDaoCached = Preconditions.checkNotNull(masterDBBindingDaoCached);
  77         this.epPolicyTemplateDao = Preconditions.checkNotNull(epPolicyTemplateDao);
  78         this.epForwardingTemplateDao = Preconditions.checkNotNull(epForwardingTemplateDao);
  79
  80         //TODO: get exact sxp-node path from config (pointing to ise listener).. start listening later - when config appears
  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 }