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