bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/AdjRibInWriter.java

   1 /*
   2  * Copyright (c) 2015 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.protocol.bgp.rib.impl;
   9
  10 import static java.util.Objects.requireNonNull;
  11 import static org.opendaylight.protocol.bgp.rib.spi.RIBNodeIdentifiers.ADJRIBIN_NID;
  12 import static org.opendaylight.protocol.bgp.rib.spi.RIBNodeIdentifiers.ADJRIBOUT_NID;
  13 import static org.opendaylight.protocol.bgp.rib.spi.RIBNodeIdentifiers.ATTRIBUTES_NID;
  14 import static org.opendaylight.protocol.bgp.rib.spi.RIBNodeIdentifiers.EFFRIBIN_NID;
  15 import static org.opendaylight.protocol.bgp.rib.spi.RIBNodeIdentifiers.TABLES_NID;
  16 import static org.opendaylight.protocol.bgp.rib.spi.RIBNodeIdentifiers.UPTODATE_NID;
  17
  18 import com.google.common.annotations.VisibleForTesting;
  19 import com.google.common.collect.ImmutableMap;
  20 import com.google.common.collect.ImmutableMap.Builder;
  21 import com.google.common.util.concurrent.FluentFuture;
  22 import com.google.common.util.concurrent.FutureCallback;
  23 import com.google.common.util.concurrent.MoreExecutors;
  24 import java.util.Collection;
  25 import java.util.Collections;
  26 import java.util.HashMap;
  27 import java.util.List;
  28 import java.util.Locale;
  29 import java.util.Map;
  30 import java.util.Map.Entry;
  31 import java.util.Optional;
  32 import java.util.Set;
  33 import java.util.concurrent.CountDownLatch;
  34 import java.util.concurrent.ExecutionException;
  35 import java.util.stream.Collectors;
  36 import org.checkerframework.checker.lock.qual.GuardedBy;
  37 import org.eclipse.jdt.annotation.NonNull;
  38 import org.eclipse.jdt.annotation.Nullable;
  39 import org.opendaylight.mdsal.common.api.CommitInfo;
  40 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  41 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
  42 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
  43 import org.opendaylight.protocol.bgp.rib.impl.ApplicationPeer.RegisterAppPeerListener;
  44 import org.opendaylight.protocol.bgp.rib.impl.spi.PeerTransactionChain;
  45 import org.opendaylight.protocol.bgp.rib.impl.spi.RIBSupportContext;
  46 import org.opendaylight.protocol.bgp.rib.impl.spi.RIBSupportContextRegistry;
  47 import org.opendaylight.protocol.bgp.rib.spi.IdentifierUtils;
  48 import org.opendaylight.protocol.bgp.rib.spi.PeerRoleUtil;
  49 import org.opendaylight.protocol.bgp.rib.spi.RIBNormalizedNodes;
  50 import org.opendaylight.protocol.bgp.rib.spi.RIBQNames;
  51 import org.opendaylight.protocol.bgp.rib.spi.RibSupportUtils;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.SendReceive;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.reach.MpReachNlri;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.unreach.MpUnreachNlri;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.PeerId;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.PeerRole;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.Peer;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.peer.SupportedTables;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.rib.TablesKey;
  60 import org.opendaylight.yangtools.yang.common.QName;
  61 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  62 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.InstanceIdentifierBuilder;
  63 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  64 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  65 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  66 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  67 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  68 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  69 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
  70 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  71 import org.slf4j.Logger;
  72 import org.slf4j.LoggerFactory;
  73
  74 /**
  75  * Writer of Adjacency-RIB-In for a single peer. An instance of this object is attached to each {@link BGPPeer} and
  76  * {@link ApplicationPeer}. This class is NOT thread-safe.
  77  */
  78 final class AdjRibInWriter {
  79
  80     private static final Logger LOG = LoggerFactory.getLogger(AdjRibInWriter.class);
  81
  82     private static final QName PEER_ROLE_QNAME = QName.create(Peer.QNAME, "peer-role").intern();
  83     private static final NodeIdentifier PEER_ID = NodeIdentifier.create(RIBQNames.PEER_ID_QNAME);
  84     private static final NodeIdentifier PEER_ROLE = NodeIdentifier.create(PEER_ROLE_QNAME);
  85     private static final NodeIdentifier PEER_TABLES = NodeIdentifier.create(SupportedTables.QNAME);
  86     private static final QName SEND_RECEIVE = QName.create(SupportedTables.QNAME, "send-receive").intern();
  87
  88     // FIXME: is there a utility method to construct this?
  89     private static final MapNode EMPTY_TABLES = ImmutableNodes.mapNodeBuilder(TABLES_NID).build();
  90     private static final ContainerNode EMPTY_ADJRIBIN = Builders.containerBuilder()
  91             .withNodeIdentifier(ADJRIBIN_NID).addChild(EMPTY_TABLES).build();
  92     private static final ContainerNode EMPTY_EFFRIBIN = Builders.containerBuilder()
  93             .withNodeIdentifier(EFFRIBIN_NID).addChild(EMPTY_TABLES).build();
  94     private static final ContainerNode EMPTY_ADJRIBOUT = Builders.containerBuilder()
  95             .withNodeIdentifier(ADJRIBOUT_NID).addChild(EMPTY_TABLES).build();
  96
  97     private final Map<TablesKey, TableContext> tables;
  98     private final YangInstanceIdentifier ribPath;
  99     private final PeerTransactionChain chain;
 100     private final PeerRole role;
 101     @GuardedBy("this")
 102     private final Map<TablesKey, Collection<NodeIdentifierWithPredicates>> staleRoutesRegistry = new HashMap<>();
 103     @GuardedBy("this")
 104     private FluentFuture<? extends CommitInfo> submitted;
 105
 106     private AdjRibInWriter(final YangInstanceIdentifier ribPath, final PeerTransactionChain chain, final PeerRole role,
 107             final Map<TablesKey, TableContext> tables) {
 108         this.ribPath = requireNonNull(ribPath);
 109         this.chain = requireNonNull(chain);
 110         this.tables = requireNonNull(tables);
 111         this.role = requireNonNull(role);
 112     }
 113
 114     /**
 115      * Create a new writer using a transaction chain.
 116      *
 117      * @param role                peer's role
 118      * @param chain               transaction chain  @return A fresh writer instance
 119      */
 120     static AdjRibInWriter create(final @NonNull YangInstanceIdentifier ribId, final @NonNull PeerRole role,
 121             final @NonNull PeerTransactionChain chain) {
 122         return new AdjRibInWriter(ribId, chain, role, Collections.emptyMap());
 123     }
 124
 125     /**
 126      * Transform this writer to a new writer, which is in charge of specified tables.
 127      * Empty tables are created for new entries and old tables are deleted. Once this
 128      * method returns, the old instance must not be reasonably used.
 129      *
 130      * @param newPeerId         new peer BGP identifier
 131      * @param peerPath          path of the peer in the datastore
 132      * @param registry          RIB extension registry
 133      * @param tableTypes        New tables, must not be null
 134      * @param addPathTablesType supported add path tables
 135      * @return New writer
 136      */
 137     AdjRibInWriter transform(final PeerId newPeerId, final YangInstanceIdentifier peerPath,
 138             final RIBSupportContextRegistry registry,
 139             final Set<TablesKey> tableTypes, final Map<TablesKey, SendReceive> addPathTablesType) {
 140         return transform(newPeerId, peerPath, registry, tableTypes, addPathTablesType, null);
 141     }
 142
 143     AdjRibInWriter transform(final PeerId newPeerId, final YangInstanceIdentifier peerPath,
 144             final RIBSupportContextRegistry registry, final Set<TablesKey> tableTypes,
 145             final Map<TablesKey, SendReceive> addPathTablesType,
 146             final @Nullable RegisterAppPeerListener registerAppPeerListener) {
 147         final DOMDataTreeWriteTransaction tx = chain.getDomChain().newWriteOnlyTransaction();
 148
 149         createEmptyPeerStructure(newPeerId, peerPath, tx);
 150         final ImmutableMap<TablesKey, TableContext> tb = createNewTableInstances(peerPath, registry, tableTypes,
 151                 addPathTablesType, tx);
 152
 153         tx.commit().addCallback(new FutureCallback<CommitInfo>() {
 154             @Override
 155             public void onSuccess(final CommitInfo result) {
 156                 if (registerAppPeerListener != null) {
 157                     LOG.trace("Application Peer Listener registered");
 158                     registerAppPeerListener.register();
 159                 }
 160             }
 161
 162             @Override
 163             public void onFailure(final Throwable throwable) {
 164                 if (registerAppPeerListener != null) {
 165                     LOG.error("Failed to create Empty Structure, Application Peer Listener won't be registered",
 166                             throwable);
 167                 } else {
 168                     LOG.error("Failed to create Empty Structure", throwable);
 169                 }
 170             }
 171         }, MoreExecutors.directExecutor());
 172         return new AdjRibInWriter(ribPath, chain, role, tb);
 173     }
 174
 175     /**
 176      * Create new table instances, potentially creating their empty entries.
 177      */
 178     private static ImmutableMap<TablesKey, TableContext> createNewTableInstances(
 179             final YangInstanceIdentifier newPeerPath, final RIBSupportContextRegistry registry,
 180             final Set<TablesKey> tableTypes, final Map<TablesKey, SendReceive> addPathTablesType,
 181             final DOMDataTreeWriteTransaction tx) {
 182
 183         final Builder<TablesKey, TableContext> tb = ImmutableMap.builder();
 184         for (final TablesKey tableKey : tableTypes) {
 185             final RIBSupportContext rs = registry.getRIBSupportContext(tableKey);
 186             // TODO: Use returned value once Instance Identifier builder allows for it.
 187             final NodeIdentifierWithPredicates instanceIdentifierKey = RibSupportUtils.toYangTablesKey(tableKey);
 188             if (rs == null) {
 189                 LOG.warn("No support for table type {}, skipping it", tableKey);
 190                 continue;
 191             }
 192             installAdjRibsOutTables(newPeerPath, rs, instanceIdentifierKey, tableKey,
 193                     addPathTablesType.get(tableKey), tx);
 194             installAdjRibInTables(newPeerPath, tableKey, rs, instanceIdentifierKey, tx, tb);
 195         }
 196         return tb.build();
 197     }
 198
 199     private static void installAdjRibInTables(final YangInstanceIdentifier newPeerPath, final TablesKey tableKey,
 200             final RIBSupportContext rs, final NodeIdentifierWithPredicates instanceIdentifierKey,
 201             final DOMDataTreeWriteTransaction tx, final Builder<TablesKey, TableContext> tb) {
 202         // We will use table keys very often, make sure they are optimized
 203         final InstanceIdentifierBuilder idb = YangInstanceIdentifier.builder(newPeerPath
 204                 .node(EMPTY_ADJRIBIN.getIdentifier()).node(TABLES_NID)).node(instanceIdentifierKey);
 205
 206         final TableContext ctx = new TableContext(rs, idb.build());
 207         ctx.createEmptyTableStructure(tx);
 208
 209         tx.merge(LogicalDatastoreType.OPERATIONAL, ctx.getTableId().node(ATTRIBUTES_NID).node(UPTODATE_NID),
 210             RIBNormalizedNodes.ATTRIBUTES_UPTODATE_FALSE);
 211         LOG.debug("Created table instance {}", ctx.getTableId());
 212         tb.put(tableKey, ctx);
 213     }
 214
 215     private static void installAdjRibsOutTables(final YangInstanceIdentifier newPeerPath, final RIBSupportContext rs,
 216             final NodeIdentifierWithPredicates instanceIdentifierKey, final TablesKey tableKey,
 217             final SendReceive sendReceive, final DOMDataTreeWriteTransaction tx) {
 218         final NodeIdentifierWithPredicates supTablesKey = RibSupportUtils.toYangKey(SupportedTables.QNAME, tableKey);
 219         final var tt = Builders.mapEntryBuilder().withNodeIdentifier(supTablesKey);
 220         for (final Entry<QName, Object> e : supTablesKey.entrySet()) {
 221             tt.withChild(ImmutableNodes.leafNode(e.getKey(), e.getValue()));
 222         }
 223         if (sendReceive != null) {
 224             tt.withChild(ImmutableNodes.leafNode(SEND_RECEIVE, sendReceive.toString().toLowerCase(Locale.ENGLISH)));
 225         }
 226         tx.put(LogicalDatastoreType.OPERATIONAL, newPeerPath.node(PEER_TABLES).node(supTablesKey), tt.build());
 227         rs.createEmptyTableStructure(tx, newPeerPath.node(EMPTY_ADJRIBOUT.getIdentifier())
 228                 .node(TABLES_NID).node(instanceIdentifierKey));
 229     }
 230
 231     private void createEmptyPeerStructure(final PeerId newPeerId,
 232             final YangInstanceIdentifier peerPath, final DOMDataTreeWriteTransaction tx) {
 233         final NodeIdentifierWithPredicates peerKey = IdentifierUtils.domPeerId(newPeerId);
 234
 235         tx.put(LogicalDatastoreType.OPERATIONAL, peerPath, peerSkeleton(peerKey, newPeerId.getValue()));
 236         LOG.debug("New peer {} structure installed.", peerPath);
 237     }
 238
 239     @VisibleForTesting
 240     MapEntryNode peerSkeleton(final NodeIdentifierWithPredicates peerKey, final String peerId) {
 241         return Builders.mapEntryBuilder()
 242             .withNodeIdentifier(peerKey)
 243             .withChild(ImmutableNodes.leafNode(PEER_ID, peerId))
 244             .withChild(ImmutableNodes.leafNode(PEER_ROLE, PeerRoleUtil.roleForString(role)))
 245             .withChild(Builders.mapBuilder().withNodeIdentifier(PEER_TABLES).build())
 246             .withChild(EMPTY_ADJRIBIN)
 247             .withChild(EMPTY_EFFRIBIN)
 248             .withChild(EMPTY_ADJRIBOUT)
 249             .build();
 250     }
 251
 252     void markTableUptodate(final TablesKey tableTypes) {
 253         final DOMDataTreeWriteTransaction tx = chain.getDomChain().newWriteOnlyTransaction();
 254         final TableContext ctx = tables.get(tableTypes);
 255         tx.merge(LogicalDatastoreType.OPERATIONAL, ctx.getTableId().node(ATTRIBUTES_NID).node(UPTODATE_NID),
 256             RIBNormalizedNodes.ATTRIBUTES_UPTODATE_TRUE);
 257         tx.commit().addCallback(new FutureCallback<CommitInfo>() {
 258             @Override
 259             public void onSuccess(final CommitInfo result) {
 260                 LOG.trace("Write Attributes uptodate, succeed");
 261             }
 262
 263             @Override
 264             public void onFailure(final Throwable throwable) {
 265                 LOG.error("Write Attributes uptodate failed", throwable);
 266             }
 267         }, MoreExecutors.directExecutor());
 268     }
 269
 270     void updateRoutes(final MpReachNlri nlri, final org.opendaylight.yang.gen.v1.urn.opendaylight.params
 271             .xml.ns.yang.bgp.message.rev200120.path.attributes.Attributes attributes) {
 272         final TablesKey key = new TablesKey(nlri.getAfi(), nlri.getSafi());
 273         final TableContext ctx = tables.get(key);
 274         if (ctx == null) {
 275             LOG.debug("No table for {}, not accepting NLRI {}", key, nlri);
 276             return;
 277         }
 278
 279         final DOMDataTreeWriteTransaction tx = chain.getDomChain().newWriteOnlyTransaction();
 280         final Collection<NodeIdentifierWithPredicates> routeKeys = ctx.writeRoutes(tx, nlri, attributes);
 281         final Collection<NodeIdentifierWithPredicates> staleRoutes = staleRoutesRegistry.get(key);
 282         if (staleRoutes != null) {
 283             staleRoutes.removeAll(routeKeys);
 284         }
 285         LOG.trace("Write routes {}", nlri);
 286         final FluentFuture<? extends CommitInfo> future = tx.commit();
 287         submitted = future;
 288         future.addCallback(new FutureCallback<CommitInfo>() {
 289             @Override
 290             public void onSuccess(final CommitInfo result) {
 291                 LOG.trace("Write routes {}, succeed", nlri);
 292             }
 293
 294             @Override
 295             public void onFailure(final Throwable throwable) {
 296                 LOG.error("Write routes failed", throwable);
 297             }
 298         }, MoreExecutors.directExecutor());
 299     }
 300
 301     void removeRoutes(final MpUnreachNlri nlri) {
 302         final TablesKey key = new TablesKey(nlri.getAfi(), nlri.getSafi());
 303         final TableContext ctx = tables.get(key);
 304         if (ctx == null) {
 305             LOG.debug("No table for {}, not accepting NLRI {}", key, nlri);
 306             return;
 307         }
 308         LOG.trace("Removing routes {}", nlri);
 309         final DOMDataTreeWriteTransaction tx = chain.getDomChain().newWriteOnlyTransaction();
 310         ctx.removeRoutes(tx, nlri);
 311         final FluentFuture<? extends CommitInfo> future = tx.commit();
 312         submitted = future;
 313         future.addCallback(new FutureCallback<CommitInfo>() {
 314             @Override
 315             public void onSuccess(final CommitInfo result) {
 316                 LOG.trace("Removing routes {}, succeed", nlri);
 317             }
 318
 319             @Override
 320             public void onFailure(final Throwable throwable) {
 321                 LOG.error("Removing routes failed", throwable);
 322             }
 323         }, MoreExecutors.directExecutor());
 324     }
 325
 326     void releaseChain() {
 327         if (submitted != null) {
 328             try {
 329                 submitted.get();
 330             } catch (final InterruptedException | ExecutionException throwable) {
 331                 LOG.error("Write routes failed", throwable);
 332             }
 333         }
 334     }
 335
 336     void storeStaleRoutes(final Set<TablesKey> gracefulTables) {
 337         final CountDownLatch latch = new CountDownLatch(gracefulTables.size());
 338
 339         try (DOMDataTreeReadTransaction tx = chain.getDomChain().newReadOnlyTransaction()) {
 340             for (TablesKey tablesKey : gracefulTables) {
 341                 final TableContext ctx = tables.get(tablesKey);
 342                 if (ctx == null) {
 343                     LOG.warn("Missing table for address family {}", tablesKey);
 344                     latch.countDown();
 345                     continue;
 346                 }
 347
 348                 tx.read(LogicalDatastoreType.OPERATIONAL, ctx.routesPath()).addCallback(
 349                     new FutureCallback<Optional<NormalizedNode>>() {
 350                         @Override
 351                         public void onSuccess(final Optional<NormalizedNode> routesOptional) {
 352                             try {
 353                                 if (routesOptional.isPresent()) {
 354                                     synchronized (staleRoutesRegistry) {
 355                                         final MapNode routesNode = (MapNode) routesOptional.get();
 356                                         final List<NodeIdentifierWithPredicates> routes = routesNode.body().stream()
 357                                                 .map(MapEntryNode::getIdentifier)
 358                                                 .collect(Collectors.toList());
 359                                         if (!routes.isEmpty()) {
 360                                             staleRoutesRegistry.put(tablesKey, routes);
 361                                         }
 362                                     }
 363                                 }
 364                             } finally {
 365                                 latch.countDown();
 366                             }
 367                         }
 368
 369                         @Override
 370                         public void onFailure(final Throwable throwable) {
 371                             LOG.warn("Failed to store stale routes for table {}", tablesKey, throwable);
 372                             latch.countDown();
 373                         }
 374                     }, MoreExecutors.directExecutor());
 375             }
 376         }
 377
 378         try {
 379             latch.await();
 380         } catch (InterruptedException e) {
 381             LOG.warn("Interrupted while waiting to store stale routes with {} tasks of {} to finish", latch.getCount(),
 382                 gracefulTables, e);
 383         }
 384     }
 385
 386     void removeStaleRoutes(final TablesKey tableKey) {
 387         final TableContext ctx = tables.get(tableKey);
 388         if (ctx == null) {
 389             LOG.debug("No table for {}, not removing any stale routes", tableKey);
 390             return;
 391         }
 392         final Collection<NodeIdentifierWithPredicates> routeKeys = staleRoutesRegistry.get(tableKey);
 393         if (routeKeys == null || routeKeys.isEmpty()) {
 394             LOG.debug("No stale routes present in table {}", tableKey);
 395             return;
 396         }
 397         LOG.trace("Removing routes {}", routeKeys);
 398         final DOMDataTreeWriteTransaction tx = chain.getDomChain().newWriteOnlyTransaction();
 399         routeKeys.forEach(routeKey -> {
 400             tx.delete(LogicalDatastoreType.OPERATIONAL, ctx.routePath(routeKey));
 401         });
 402         final FluentFuture<? extends CommitInfo> future = tx.commit();
 403         submitted = future;
 404         future.addCallback(new FutureCallback<CommitInfo>() {
 405             @Override
 406             public void onSuccess(final CommitInfo result) {
 407                 LOG.trace("Removing routes {}, succeed", routeKeys);
 408                 synchronized (staleRoutesRegistry) {
 409                     staleRoutesRegistry.remove(tableKey);
 410                 }
 411             }
 412
 413             @Override
 414             public void onFailure(final Throwable throwable) {
 415                 LOG.warn("Removing routes {}, failed", routeKeys, throwable);
 416             }
 417         }, MoreExecutors.directExecutor());
 418     }
 419
 420     FluentFuture<? extends CommitInfo> clearTables(final Set<TablesKey> tablesToClear) {
 421         if (tablesToClear == null || tablesToClear.isEmpty()) {
 422             return CommitInfo.emptyFluentFuture();
 423         }
 424
 425         final DOMDataTreeWriteTransaction wtx = chain.getDomChain().newWriteOnlyTransaction();
 426         tablesToClear.forEach(tableKey -> {
 427             final TableContext ctx = tables.get(tableKey);
 428             wtx.delete(LogicalDatastoreType.OPERATIONAL, ctx.routesPath().getParent());
 429         });
 430         return wtx.commit();
 431     }
 432 }