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