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

   1 /*
   2  * Copyright (c) 2014 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.ADJRIBOUT_NID;
  12 import static org.opendaylight.protocol.bgp.rib.spi.RIBNodeIdentifiers.TABLES_NID;
  13
  14 import com.google.common.base.MoreObjects;
  15 import com.google.common.base.Stopwatch;
  16 import com.google.common.cache.CacheBuilder;
  17 import com.google.common.cache.CacheLoader;
  18 import com.google.common.cache.LoadingCache;
  19 import com.google.common.collect.ImmutableClassToInstanceMap;
  20 import com.google.common.collect.ImmutableMap;
  21 import com.google.common.collect.ImmutableSet;
  22 import com.google.common.collect.Sets;
  23 import com.google.common.util.concurrent.FluentFuture;
  24 import com.google.common.util.concurrent.FutureCallback;
  25 import com.google.common.util.concurrent.Futures;
  26 import com.google.common.util.concurrent.ListenableFuture;
  27 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  28 import java.util.ArrayList;
  29 import java.util.Collections;
  30 import java.util.HashMap;
  31 import java.util.HashSet;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.Objects;
  35 import java.util.Optional;
  36 import java.util.Set;
  37 import java.util.concurrent.TimeUnit;
  38 import java.util.stream.Collectors;
  39 import org.checkerframework.checker.lock.qual.GuardedBy;
  40 import org.checkerframework.checker.lock.qual.Holding;
  41 import org.eclipse.jdt.annotation.NonNull;
  42 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  43 import org.opendaylight.mdsal.common.api.CommitInfo;
  44 import org.opendaylight.protocol.bgp.openconfig.spi.BGPTableTypeRegistryConsumer;
  45 import org.opendaylight.protocol.bgp.parser.BGPDocumentedException;
  46 import org.opendaylight.protocol.bgp.parser.BGPError;
  47 import org.opendaylight.protocol.bgp.parser.impl.message.update.LocalPreferenceAttributeParser;
  48 import org.opendaylight.protocol.bgp.parser.spi.MessageUtil;
  49 import org.opendaylight.protocol.bgp.parser.spi.RevisedErrorHandlingSupport;
  50 import org.opendaylight.protocol.bgp.rib.impl.config.BgpPeer;
  51 import org.opendaylight.protocol.bgp.rib.impl.config.GracefulRestartUtil;
  52 import org.opendaylight.protocol.bgp.rib.impl.spi.BGPSessionPreferences;
  53 import org.opendaylight.protocol.bgp.rib.impl.spi.RIB;
  54 import org.opendaylight.protocol.bgp.rib.impl.state.BGPSessionStateProvider;
  55 import org.opendaylight.protocol.bgp.rib.spi.BGPSession;
  56 import org.opendaylight.protocol.bgp.rib.spi.BGPSessionListener;
  57 import org.opendaylight.protocol.bgp.rib.spi.BGPTerminationReason;
  58 import org.opendaylight.protocol.bgp.rib.spi.RIBSupport;
  59 import org.opendaylight.protocol.bgp.rib.spi.RouterIds;
  60 import org.opendaylight.protocol.bgp.rib.spi.state.BGPSessionState;
  61 import org.opendaylight.protocol.bgp.rib.spi.state.BGPTimersState;
  62 import org.opendaylight.protocol.bgp.rib.spi.state.BGPTransportState;
  63 import org.opendaylight.protocol.util.Ipv4Util;
  64 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.AsNumber;
  65 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddressNoZone;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.ipv4.prefixes.DestinationIpv4Builder;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.ipv4.prefixes.destination.ipv4.Ipv4Prefixes;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.ipv4.prefixes.destination.ipv4.Ipv4PrefixesBuilder;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.update.attributes.mp.reach.nlri.advertized.routes.destination.type.DestinationIpv4CaseBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.Update;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.open.message.BgpParameters;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.path.attributes.Attributes;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.path.attributes.AttributesBuilder;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.update.message.Nlri;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.BgpAddPathTableType;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.BgpTableType;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.RouteRefresh;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.SendReceive;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.reach.MpReachNlri;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.reach.MpReachNlriBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.reach.mp.reach.nlri.AdvertizedRoutesBuilder;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.unreach.MpUnreachNlri;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.unreach.MpUnreachNlriBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.unreach.mp.unreach.nlri.WithdrawnRoutesBuilder;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.mp.capabilities.GracefulRestartCapability;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.mp.capabilities.add.path.capability.AddressFamilies;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.ResetSession;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.RestartGracefully;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.RouteRefreshRequest;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.PeerRole;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.PeerKey;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.rib.TablesKey;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.ClusterIdentifier;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.Ipv4AddressFamily;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.RouteTarget;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.UnicastSubsequentAddressFamily;
  97 import org.opendaylight.yangtools.concepts.Registration;
  98 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  99 import org.opendaylight.yangtools.yang.binding.Notification;
 100 import org.opendaylight.yangtools.yang.binding.Rpc;
 101 import org.opendaylight.yangtools.yang.common.Empty;
 102 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 103 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
 104 import org.slf4j.Logger;
 105 import org.slf4j.LoggerFactory;
 106
 107 /**
 108  * Class representing a peer. We have a single instance for each peer, which provides translation from BGP events into
 109  * RIB actions.
 110  */
 111 public class BGPPeer extends AbstractPeer implements BGPSessionListener {
 112     private static final Logger LOG = LoggerFactory.getLogger(BGPPeer.class);
 113     private static final TablesKey IPV4_UCAST_TABLE_KEY =
 114         new TablesKey(Ipv4AddressFamily.VALUE, UnicastSubsequentAddressFamily.VALUE);
 115
 116     private final RIB rib;
 117
 118     // FIXME: Alright, this right here is a ton of state which has intertwined initialization and dependencies Split
 119     //        these out into separate behavior objects. This also has relationship with state in AbstractPeer -- which
 120     //        hints at an obvious layer of indirection. Yeah, yeah, we can always add one of those, but the point
 121     //        is that this class is a mutable meeting point, whereas the behaviour has captured invariants.
 122     private final LoadingCache<NodeIdentifierWithPredicates, YangInstanceIdentifier> tablesIId =
 123         CacheBuilder.newBuilder().build(new CacheLoader<NodeIdentifierWithPredicates, YangInstanceIdentifier>() {
 124             @Override
 125             public YangInstanceIdentifier load(final NodeIdentifierWithPredicates key) {
 126                 return peerRibOutIId.node(TABLES_NID).node(key).toOptimized();
 127             }
 128         });
 129
 130     private ImmutableSet<TablesKey> tables = ImmutableSet.of();
 131     private final Map<TablesKey, AdjRibOutListener> adjRibOutListenerSet = new HashMap<>();
 132     private final List<RouteTarget> rtMemberships = new ArrayList<>();
 133     private final RpcProviderService rpcRegistry;
 134     private final BGPTableTypeRegistryConsumer tableTypeRegistry;
 135     private final BgpPeer bgpPeer;
 136
 137     // FIXME: This should be a constant co-located with ApplicationPeer.peerId
 138     private YangInstanceIdentifier peerRibOutIId;
 139     @GuardedBy("this")
 140     private Registration trackerRegistration;
 141
 142     @GuardedBy("this")
 143     private BGPSession currentSession;
 144     @GuardedBy("this")
 145     private AdjRibInWriter ribWriter;
 146     @GuardedBy("this")
 147     private EffectiveRibInWriter effRibInWriter;
 148     private Registration rpcRegistration;
 149     private ImmutableMap<TablesKey, SendReceive> addPathTableMaps = ImmutableMap.of();
 150     // FIXME: This should be a constant co-located with ApplicationPeer.peerId
 151     private YangInstanceIdentifier peerPath;
 152     // FIXME: This is for supportsTable() -- a trivial behavior thing, where 'peer-down' type states always return false
 153     private boolean sessionUp;
 154     private boolean llgrSupport;
 155     private Stopwatch peerRestartStopwatch;
 156     private long currentSelectionDeferralTimerSeconds;
 157     private final List<TablesKey> missingEOT = new ArrayList<>();
 158
 159     public BGPPeer(
 160             final BGPTableTypeRegistryConsumer tableTypeRegistry,
 161             final IpAddressNoZone neighborAddress,
 162             final String peerGroupName,
 163             final RIB rib,
 164             final PeerRole role,
 165             final ClusterIdentifier clusterId,
 166             final AsNumber localAs,
 167             final RpcProviderService rpcRegistry,
 168             final Set<TablesKey> afiSafisAdvertized,
 169             final Set<TablesKey> afiSafisGracefulAdvertized,
 170             final Map<TablesKey, Integer> llGracefulTablesAdvertised,
 171             final BgpPeer bgpPeer) {
 172         super(rib, Ipv4Util.toStringIP(neighborAddress), peerGroupName, role, clusterId,
 173                 localAs, neighborAddress, afiSafisAdvertized, afiSafisGracefulAdvertized, llGracefulTablesAdvertised);
 174         this.tableTypeRegistry = requireNonNull(tableTypeRegistry);
 175         this.rib = requireNonNull(rib);
 176         this.rpcRegistry = rpcRegistry;
 177         this.bgpPeer = bgpPeer;
 178     }
 179
 180     private static Attributes nextHopToAttribute(final Attributes attrs, final MpReachNlri mpReach) {
 181         if (attrs.getCNextHop() == null && mpReach.getCNextHop() != null) {
 182             final AttributesBuilder attributesBuilder = new AttributesBuilder(attrs);
 183             attributesBuilder.setCNextHop(mpReach.getCNextHop());
 184             return attributesBuilder.build();
 185         }
 186         return attrs;
 187     }
 188
 189     /**
 190      * Creates MPReach for the prefixes to be handled in the same way as linkstate routes.
 191      *
 192      * @param message Update message containing prefixes in NLRI
 193      * @return MpReachNlri with prefixes from the nlri field
 194      */
 195     private static MpReachNlri prefixesToMpReach(final Update message) {
 196         final List<Ipv4Prefixes> prefixes = message.getNlri().stream()
 197                 .map(n -> new Ipv4PrefixesBuilder().setPrefix(n.getPrefix()).setPathId(n.getPathId()).build())
 198                 .collect(Collectors.toList());
 199         final MpReachNlriBuilder b = new MpReachNlriBuilder()
 200             .setAfi(Ipv4AddressFamily.VALUE)
 201             .setSafi(UnicastSubsequentAddressFamily.VALUE)
 202             .setAdvertizedRoutes(new AdvertizedRoutesBuilder()
 203                 .setDestinationType(new DestinationIpv4CaseBuilder()
 204                     .setDestinationIpv4(new DestinationIpv4Builder().setIpv4Prefixes(prefixes).build())
 205                     .build())
 206                 .build());
 207         if (message.getAttributes() != null) {
 208             b.setCNextHop(message.getAttributes().getCNextHop());
 209         }
 210         return b.build();
 211     }
 212
 213     /**
 214      * Create MPUnreach for the prefixes to be handled in the same way as linkstate routes.
 215      *
 216      * @param message            Update message containing withdrawn routes
 217      * @param isAnyNlriAnnounced isAnyNlriAnnounced
 218      * @return MpUnreachNlri with prefixes from the withdrawn routes field
 219      */
 220     private static MpUnreachNlri prefixesToMpUnreach(final Update message, final boolean isAnyNlriAnnounced) {
 221         final List<Ipv4Prefixes> prefixes = new ArrayList<>();
 222         message.getWithdrawnRoutes().forEach(w -> {
 223
 224             Optional<Nlri> nlriAnounced = Optional.empty();
 225             if (isAnyNlriAnnounced) {
 226                 nlriAnounced = message.getNlri().stream().filter(n -> Objects.equals(n.getPrefix(), w.getPrefix())
 227                         && Objects.equals(n.getPathId(), w.getPathId()))
 228                         .findAny();
 229             }
 230             if (!nlriAnounced.isPresent()) {
 231                 prefixes.add(new Ipv4PrefixesBuilder().setPrefix(w.getPrefix()).setPathId(w.getPathId()).build());
 232             }
 233         });
 234         return new MpUnreachNlriBuilder().setAfi(Ipv4AddressFamily.VALUE).setSafi(UnicastSubsequentAddressFamily.VALUE)
 235                 .setWithdrawnRoutes(new WithdrawnRoutesBuilder().setDestinationType(new org.opendaylight.yang.gen.v1
 236                         .urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.update.attributes.mp.unreach.nlri
 237                         .withdrawn.routes.destination.type.DestinationIpv4CaseBuilder().setDestinationIpv4(
 238                             new DestinationIpv4Builder().setIpv4Prefixes(prefixes).build()).build()).build()).build();
 239     }
 240
 241     private static ImmutableMap<TablesKey, SendReceive> mapTableTypesFamilies(
 242             final List<AddressFamilies> addPathTablesType) {
 243         return addPathTablesType.stream().collect(ImmutableMap.toImmutableMap(
 244             af -> new TablesKey(af.getAfi(), af.getSafi()), BgpAddPathTableType::getSendReceive));
 245     }
 246
 247     public synchronized void instantiateServiceInstance() {
 248         createDomChain();
 249         ribWriter = AdjRibInWriter.create(rib.getYangRibId(), getRole(), this);
 250         setActive(true);
 251     }
 252
 253     @Override
 254     public synchronized FluentFuture<? extends CommitInfo> close() {
 255         final FluentFuture<? extends CommitInfo> future = releaseConnection(true);
 256         closeDomChain();
 257         setActive(false);
 258         return future;
 259     }
 260
 261     @Override
 262     public void onMessage(final BGPSession session, final Notification<?> msg) throws BGPDocumentedException {
 263         if (msg instanceof Update update) {
 264             onUpdateMessage(update);
 265         } else if (msg instanceof RouteRefresh routeRefresh) {
 266             onRouteRefreshMessage(routeRefresh);
 267         } else {
 268             LOG.info("Ignoring unhandled message class {}", msg.getClass());
 269         }
 270     }
 271
 272     private void onRouteRefreshMessage(final RouteRefresh message) {
 273         final var rrAfi = message.getAfi();
 274         final var rrSafi = message.getSafi();
 275
 276         final TablesKey key = new TablesKey(rrAfi, rrSafi);
 277         synchronized (this) {
 278             final AdjRibOutListener listener = adjRibOutListenerSet.remove(key);
 279             if (listener != null) {
 280                 listener.close();
 281                 createAdjRibOutListener(key, listener.isMpSupported());
 282             } else {
 283                 LOG.info("Ignoring RouteRefresh message. Afi/Safi is not supported: {}, {}.", rrAfi, rrSafi);
 284             }
 285         }
 286     }
 287
 288     /**
 289      * Check for presence of well known mandatory attribute LOCAL_PREF in Update message.
 290      *
 291      * @param message Update message
 292      */
 293     private void checkMandatoryAttributesPresence(final Update message) throws BGPDocumentedException {
 294         if (MessageUtil.isAnyNlriPresent(message)) {
 295             final Attributes attrs = message.getAttributes();
 296             if (getRole() == PeerRole.Ibgp && (attrs == null || attrs.getLocalPref() == null)) {
 297                 throw new BGPDocumentedException(BGPError.MANDATORY_ATTR_MISSING_MSG + "LOCAL_PREF",
 298                         BGPError.WELL_KNOWN_ATTR_MISSING,
 299                         new byte[]{LocalPreferenceAttributeParser.TYPE});
 300             }
 301         }
 302     }
 303
 304     /**
 305      * Process Update message received.
 306      * Calls {@link #checkMandatoryAttributesPresence(Update)} to check for presence of mandatory attributes.
 307      *
 308      * @param message Update message
 309      */
 310     private synchronized void onUpdateMessage(final Update message) throws BGPDocumentedException {
 311         checkMandatoryAttributesPresence(message);
 312
 313         // update AdjRibs
 314         final Attributes attrs = message.getAttributes();
 315         MpReachNlri mpReach;
 316         final boolean isAnyNlriAnnounced = message.getNlri() != null;
 317         if (isAnyNlriAnnounced) {
 318             mpReach = prefixesToMpReach(message);
 319         } else {
 320             mpReach = MessageUtil.getMpReachNlri(attrs);
 321         }
 322         if (mpReach != null) {
 323             ribWriter.updateRoutes(mpReach, nextHopToAttribute(attrs, mpReach));
 324         }
 325         final MpUnreachNlri mpUnreach;
 326         if (message.getWithdrawnRoutes() != null) {
 327             mpUnreach = prefixesToMpUnreach(message, isAnyNlriAnnounced);
 328         } else {
 329             mpUnreach = MessageUtil.getMpUnreachNlri(attrs);
 330         }
 331         final boolean endOfRib = BgpPeerUtil.isEndOfRib(message);
 332         if (mpUnreach != null) {
 333             if (endOfRib) {
 334                 final TablesKey tablesKey = new TablesKey(mpUnreach.getAfi(), mpUnreach.getSafi());
 335                 ribWriter.removeStaleRoutes(tablesKey);
 336                 missingEOT.remove(tablesKey);
 337                 handleGracefulEndOfRib();
 338             } else {
 339                 ribWriter.removeRoutes(mpUnreach);
 340             }
 341         } else if (endOfRib) {
 342             ribWriter.removeStaleRoutes(IPV4_UCAST_TABLE_KEY);
 343             missingEOT.remove(IPV4_UCAST_TABLE_KEY);
 344             handleGracefulEndOfRib();
 345         }
 346     }
 347
 348     @Holding("this")
 349     private void handleGracefulEndOfRib() {
 350         if (isLocalRestarting()) {
 351             if (missingEOT.isEmpty()) {
 352                 createEffRibInWriter();
 353                 effRibInWriter.init();
 354                 registerPrefixesCounters(effRibInWriter, effRibInWriter);
 355                 for (final TablesKey key : getAfiSafisAdvertized()) {
 356                     createAdjRibOutListener(key, true);
 357                 }
 358                 setLocalRestartingState(false);
 359                 setGracefulPreferences(false, Collections.emptySet());
 360             }
 361         }
 362     }
 363
 364     @Override
 365     public synchronized void onSessionUp(final BGPSession session) {
 366         currentSession = session;
 367         sessionUp = true;
 368
 369         final var chain = rib.createPeerDOMChain();
 370         ribOutChain = chain;
 371         chain.addCallback(new FutureCallback<Empty>() {
 372             @Override
 373             public void onSuccess(final Empty result) {
 374                 LOG.debug("RibOut transaction chain {} successful.", chain);
 375             }
 376
 377             @Override
 378             public void onFailure(final Throwable cause) {
 379                 onRibOutChainFailed(cause);
 380             }
 381         });
 382
 383         if (currentSession instanceof BGPSessionStateProvider stateProvider) {
 384             stateProvider.registerMessagesCounter(this);
 385         }
 386         final GracefulRestartCapability advertisedGracefulRestartCapability =
 387                 session.getAdvertisedGracefulRestartCapability();
 388         final var advertisedTables = advertisedGracefulRestartCapability.getTables();
 389         final var advertisedLLTables = session.getAdvertisedLlGracefulRestartCapability().getTables();
 390
 391         final List<AddressFamilies> addPathTablesType = session.getAdvertisedAddPathTableTypes();
 392         final Set<BgpTableType> advertizedTableTypes = session.getAdvertisedTableTypes();
 393         LOG.info("Session with peer {} went up with tables {} and Add Path tables {}", getName(),
 394                 advertizedTableTypes, addPathTablesType);
 395         final Set<TablesKey> setTables = advertizedTableTypes.stream().map(t -> new TablesKey(t.getAfi(), t.getSafi()))
 396                 .collect(Collectors.toSet());
 397         tables = ImmutableSet.copyOf(setTables);
 398
 399         addPathTableMaps = mapTableTypesFamilies(addPathTablesType);
 400         final boolean restartingLocally = isLocalRestarting();
 401         if (!restartingLocally) {
 402             addBgp4Support();
 403         }
 404         if (!isRestartingGracefully()) {
 405             peerId = RouterIds.createPeerId(session.getBgpId());
 406
 407             final KeyedInstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib
 408                 .rev180329.bgp.rib.rib.Peer, PeerKey> peerIId = getInstanceIdentifier().child(org.opendaylight.yang.gen
 409                     .v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.Peer.class,
 410                     new PeerKey(peerId));
 411             peerPath = createPeerPath(peerId);
 412             peerRibOutIId = peerPath.node(ADJRIBOUT_NID);
 413             trackerRegistration = rib.getPeerTracker().registerPeer(this);
 414             createEffRibInWriter();
 415             registerPrefixesCounters(effRibInWriter, effRibInWriter);
 416
 417             effRibInWriter.init();
 418             ribWriter = ribWriter.transform(peerId, peerPath, rib.getRibSupportContext(),
 419                     tables, addPathTableMaps);
 420
 421             if (rpcRegistry != null) {
 422                 final var bgpPeerHandler = new BgpPeerRpc(this, session, tables);
 423                 rpcRegistration = rpcRegistry.registerRpcImplementations(
 424                     ImmutableClassToInstanceMap.<Rpc<?, ?>>builder()
 425                         .put(ResetSession.class, bgpPeerHandler::resetSession)
 426                         .put(RestartGracefully.class, bgpPeerHandler::restartGracefully)
 427                         .put(RouteRefreshRequest.class, bgpPeerHandler::routeRefreshRequest)
 428                         .build(),
 429                     ImmutableSet.of(rib.getInstanceIdentifier().child(
 430                         org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib
 431                         .Peer.class, new PeerKey(peerId))));
 432             }
 433         } else {
 434             final Set<TablesKey> forwardingTables;
 435             if (advertisedTables == null) {
 436                 forwardingTables = Collections.emptySet();
 437             } else {
 438                 forwardingTables = advertisedTables.values().stream()
 439                         .filter(table -> table.getAfiFlags() != null)
 440                         .filter(table -> table.getAfiFlags().getForwardingState())
 441                         .map(table -> new TablesKey(table.getAfi(), table.getSafi()))
 442                         .collect(Collectors.toSet());
 443             }
 444             ribWriter.clearTables(Sets.difference(tables, forwardingTables));
 445             if (restartingLocally) {
 446                 effRibInWriter.close();
 447                 peerRestartStopwatch = Stopwatch.createStarted();
 448                 handleSelectionReferralTimer();
 449                 missingEOT.addAll(tables);
 450             }
 451         }
 452         if (advertisedTables == null || advertisedTables.isEmpty()) {
 453             setAdvertizedGracefulRestartTableTypes(Collections.emptyList());
 454         } else {
 455             setAdvertizedGracefulRestartTableTypes(advertisedTables.values().stream()
 456                     .map(t -> new TablesKey(t.getAfi(), t.getSafi())).collect(Collectors.toList()));
 457         }
 458         setAfiSafiGracefulRestartState(advertisedGracefulRestartCapability.getRestartTime().toJava(), false,
 459             restartingLocally);
 460
 461         final Map<TablesKey, Integer> llTablesReceived;
 462         if (advertisedLLTables != null) {
 463             llTablesReceived = new HashMap<>();
 464             for (var table : advertisedLLTables.values()) {
 465                 llTablesReceived.put(new TablesKey(table.getAfi(), table.getSafi()),
 466                     table.getLongLivedStaleTime().getValue().intValue());
 467             }
 468         } else {
 469             llTablesReceived = Collections.emptyMap();
 470         }
 471         setAdvertizedLlGracefulRestartTableTypes(llTablesReceived);
 472
 473         if (!llTablesReceived.isEmpty()) {
 474             llgrSupport = true;
 475             // FIXME: propagate preserved tables
 476         } else {
 477             // FIXME: clear preserved tables
 478             llgrSupport = false;
 479         }
 480
 481         if (!restartingLocally) {
 482             if (!setTables.contains(IPV4_UCAST_TABLE_KEY)) {
 483                 createAdjRibOutListener(IPV4_UCAST_TABLE_KEY, false);
 484             }
 485             for (final TablesKey key : getAfiSafisAdvertized()) {
 486                 createAdjRibOutListener(key, true);
 487             }
 488         }
 489
 490         // SpotBugs does not grok Optional.ifPresent() and thinks we are using unsynchronized access
 491         final Optional<RevisedErrorHandlingSupport> errorHandling = bgpPeer.getErrorHandling();
 492         if (errorHandling.isPresent()) {
 493             currentSession.addDecoderConstraint(RevisedErrorHandlingSupport.class, errorHandling.orElseThrow());
 494         }
 495     }
 496
 497     private boolean isRestartingGracefully() {
 498         return isLocalRestarting() || isPeerRestarting();
 499     }
 500
 501     private synchronized void createEffRibInWriter() {
 502         final var chain = rib.createPeerDOMChain();
 503         chain.addCallback(this);
 504
 505         effRibInWriter = new EffectiveRibInWriter(this, rib, chain, peerPath, tables, tableTypeRegistry, rtMemberships,
 506             rtCache);
 507     }
 508
 509     //try to add a support for old-school BGP-4, if peer did not advertise IPv4-Unicast MP capability
 510     @Holding("this")
 511     private void addBgp4Support() {
 512         if (!tables.contains(IPV4_UCAST_TABLE_KEY)) {
 513             final HashSet<TablesKey> newSet = new HashSet<>(tables);
 514             newSet.add(IPV4_UCAST_TABLE_KEY);
 515             tables = ImmutableSet.copyOf(newSet);
 516         }
 517     }
 518
 519     @Holding("this")
 520     private void createAdjRibOutListener(final TablesKey key, final boolean mpSupport) {
 521         final RIBSupport<?, ?> ribSupport = rib.getRibSupportContext().getRIBSupport(key);
 522
 523         // not particularly nice
 524         if (ribSupport != null && currentSession instanceof BGPSessionImpl bgpSession) {
 525             final AdjRibOutListener adjRibOut = AdjRibOutListener.create(peerId, rib.getYangRibId(),
 526                 rib.getCodecsRegistry(), ribSupport, rib.getService(), bgpSession.getLimiter(), mpSupport);
 527             adjRibOutListenerSet.put(key, adjRibOut);
 528             registerPrefixesSentCounter(key, adjRibOut);
 529         }
 530     }
 531
 532     @Override
 533     public synchronized void onSessionDown(final BGPSession session, final Exception exc) {
 534         if (exc.getMessage().equals(BGPSessionImpl.END_OF_INPUT)) {
 535             LOG.info("Session with peer {} went down", getName());
 536         } else {
 537             LOG.info("Session with peer {} went down", getName(), exc);
 538         }
 539         releaseConnectionGracefully();
 540     }
 541
 542     @Override
 543     public synchronized void onSessionTerminated(final BGPSession session, final BGPTerminationReason cause) {
 544         LOG.info("Session with peer {} terminated: {}", getName(), cause);
 545         releaseConnectionGracefully();
 546     }
 547
 548     @Override
 549     public String toString() {
 550         return MoreObjects.toStringHelper(this).add("name", getName()).add("tables", tables).toString();
 551     }
 552
 553     @Override
 554     public synchronized FluentFuture<? extends CommitInfo> releaseConnection() {
 555         return releaseConnection(true);
 556     }
 557
 558     /**
 559      * On transaction chain failure, we don't want to wait for future.
 560      *
 561      * @param isWaitForSubmitted if true, wait for submitted future before closing binding chain. if false, don't wait.
 562      */
 563     @Holding("this")
 564     private @NonNull FluentFuture<? extends CommitInfo> releaseConnection(final boolean isWaitForSubmitted) {
 565         LOG.info("Closing session with peer");
 566         sessionUp = false;
 567         adjRibOutListenerSet.values().forEach(AdjRibOutListener::close);
 568         adjRibOutListenerSet.clear();
 569         final FluentFuture<? extends CommitInfo> future;
 570         // FIXME: this is a typical example of something which should be handled by a behavior into which we have
 571         //        transitioned way before this method is called. This really begs to be an abstract base class with
 572         //        a 'clearTables' or similar callout
 573         if (isRestartingGracefully()) {
 574             final Set<TablesKey> gracefulTables = getGracefulTables();
 575             ribWriter.storeStaleRoutes(gracefulTables);
 576             future = ribWriter.clearTables(Sets.difference(tables, gracefulTables));
 577             if (isPeerRestarting()) {
 578                 peerRestartStopwatch = Stopwatch.createStarted();
 579                 handleRestartTimer();
 580             }
 581         } else {
 582             future = terminateConnection();
 583         }
 584         releaseRibOutChain(isWaitForSubmitted);
 585
 586         closeSession();
 587         return future;
 588     }
 589
 590     @Holding("this")
 591     @SuppressWarnings("checkstyle:illegalCatch")
 592     private void closeSession() {
 593         if (currentSession != null) {
 594             try {
 595                 if (isRestartingGracefully()) {
 596                     currentSession.closeWithoutMessage();
 597                 } else {
 598                     currentSession.close();
 599                 }
 600             } catch (final Exception e) {
 601                 LOG.warn("Error closing session with peer", e);
 602             }
 603             currentSession = null;
 604         }
 605     }
 606
 607     private Set<TablesKey> getGracefulTables() {
 608         return tables.stream()
 609                 .filter(this::isGracefulRestartReceived)
 610                 .filter(this::isGracefulRestartAdvertized)
 611                 .collect(Collectors.toSet());
 612     }
 613
 614     private synchronized FluentFuture<? extends CommitInfo> terminateConnection() {
 615         if (trackerRegistration != null) {
 616             trackerRegistration.close();
 617             trackerRegistration = null;
 618         }
 619         if (rpcRegistration != null) {
 620             rpcRegistration.close();
 621         }
 622         ribWriter.releaseChain();
 623
 624         if (effRibInWriter != null) {
 625             effRibInWriter.close();
 626         }
 627         tables = ImmutableSet.of();
 628         addPathTableMaps = ImmutableMap.of();
 629         final var future = removePeer(peerPath);
 630         resetState();
 631
 632         return future;
 633     }
 634
 635     /**
 636      * If Graceful Restart Timer expires, remove all routes advertised by peer.
 637      */
 638     private synchronized void handleRestartTimer() {
 639         if (!isPeerRestarting()) {
 640             return;
 641         }
 642
 643         final long peerRestartTimeNanos = TimeUnit.SECONDS.toNanos(getPeerRestartTime());
 644         final long elapsedNanos = peerRestartStopwatch.elapsed(TimeUnit.NANOSECONDS);
 645         if (elapsedNanos >= peerRestartTimeNanos) {
 646             setAfiSafiGracefulRestartState(0, false, false);
 647             onSessionTerminated(currentSession, new BGPTerminationReason(BGPError.HOLD_TIMER_EXPIRED));
 648         }
 649
 650         currentSession.schedule(this::handleRestartTimer, peerRestartTimeNanos - elapsedNanos, TimeUnit.NANOSECONDS);
 651     }
 652
 653     private synchronized void handleSelectionReferralTimer() {
 654         if (!isLocalRestarting()) {
 655             return;
 656         }
 657
 658         final long referalTimerNanos = TimeUnit.SECONDS.toNanos(currentSelectionDeferralTimerSeconds);
 659         final long elapsedNanos = peerRestartStopwatch.elapsed(TimeUnit.NANOSECONDS);
 660         if (elapsedNanos >= referalTimerNanos) {
 661             missingEOT.clear();
 662             handleGracefulEndOfRib();
 663         }
 664         currentSession.schedule(this::handleSelectionReferralTimer, referalTimerNanos - elapsedNanos,
 665             TimeUnit.NANOSECONDS);
 666     }
 667
 668     @Holding("this")
 669     private void releaseConnectionGracefully() {
 670         if (getPeerRestartTime() > 0) {
 671             setRestartingState();
 672         }
 673         releaseConnection(true);
 674     }
 675
 676     @SuppressFBWarnings("IS2_INCONSISTENT_SYNC")
 677     @Override
 678     public SendReceive getSupportedAddPathTables(final TablesKey tableKey) {
 679         return addPathTableMaps.get(tableKey);
 680     }
 681
 682     @Override
 683     public boolean supportsTable(final TablesKey tableKey) {
 684         return sessionUp && getAfiSafisAdvertized().contains(tableKey) && tables.contains(tableKey);
 685     }
 686
 687     @Override
 688     public YangInstanceIdentifier getRibOutIId(final NodeIdentifierWithPredicates tablekey) {
 689         return tablesIId.getUnchecked(tablekey);
 690     }
 691
 692     @Override
 693     public synchronized void onFailure(final Throwable cause) {
 694         LOG.error("Transaction domChain failed.", cause);
 695         releaseConnection(true);
 696     }
 697
 698     private synchronized void onRibOutChainFailed(final Throwable cause) {
 699         LOG.error("RibOut transaction chain failed.", cause);
 700         releaseConnection(false);
 701     }
 702
 703     @Override
 704     public synchronized void markUptodate(final TablesKey tablesKey) {
 705         ribWriter.markTableUptodate(tablesKey);
 706     }
 707
 708     @Override
 709     public synchronized BGPSessionState getBGPSessionState() {
 710         if (currentSession instanceof BGPSessionStateProvider stateProvider) {
 711             return stateProvider.getBGPSessionState();
 712         }
 713         return null;
 714     }
 715
 716     @Override
 717     public synchronized BGPTimersState getBGPTimersState() {
 718         if (currentSession instanceof BGPSessionStateProvider stateProvider) {
 719             return stateProvider.getBGPTimersState();
 720         }
 721         return null;
 722     }
 723
 724     @Override
 725     public synchronized BGPTransportState getBGPTransportState() {
 726         if (currentSession instanceof BGPSessionStateProvider stateProvider) {
 727             return stateProvider.getBGPTransportState();
 728         }
 729         return null;
 730     }
 731
 732     @Override
 733     public List<RouteTarget> getMemberships() {
 734         return rtMemberships;
 735     }
 736
 737     @Override
 738     public synchronized ListenableFuture<?> restartGracefully(final long selectionDeferralTimerSeconds) {
 739         final Set<TablesKey> tablesToPreserve = getGracefulTables();
 740         if (tablesToPreserve == null || tablesToPreserve.isEmpty()) {
 741             LOG.info("Peer {} is not capable of graceful restart or have no matching graceful tables.", peerId);
 742             return Futures.immediateFailedFuture(new UnsupportedOperationException(
 743                     "Peer is not capable of graceful restart"));
 744         }
 745         setGracefulPreferences(true, tablesToPreserve);
 746         currentSelectionDeferralTimerSeconds = selectionDeferralTimerSeconds;
 747         setLocalRestartingState(true);
 748         return releaseConnection(true);
 749     }
 750
 751     @Override
 752     boolean supportsLLGR() {
 753         return llgrSupport;
 754     }
 755
 756     private synchronized void setGracefulPreferences(final boolean localRestarting,
 757                                                      final Set<TablesKey> preservedTables) {
 758         final Set<TablesKey> gracefulTables = tables.stream()
 759                 .filter(this::isGracefulRestartAdvertized)
 760                 .collect(Collectors.toSet());
 761         final BgpParameters bgpParameters = GracefulRestartUtil.getGracefulBgpParameters(
 762                 bgpPeer.getBgpFixedCapabilities(), gracefulTables, preservedTables,
 763                 bgpPeer.getGracefulRestartTimer(), localRestarting, Collections.emptySet());
 764         final BGPSessionPreferences oldPrefs = rib.getDispatcher().getBGPPeerRegistry()
 765                 .getPeerPreferences(getNeighborAddress());
 766         final BGPSessionPreferences newPrefs = new BGPSessionPreferences(
 767                 oldPrefs.getMyAs(),
 768                 oldPrefs.getHoldTime(),
 769                 oldPrefs.getBgpId(),
 770                 oldPrefs.getExpectedRemoteAs(),
 771                 Collections.singletonList(bgpParameters),
 772                 oldPrefs.getMd5Password());
 773         rib.getDispatcher().getBGPPeerRegistry()
 774                 .updatePeerPreferences(getNeighborAddress(), newPrefs);
 775     }
 776 }