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