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.ImmutableMap;
  20 import com.google.common.collect.ImmutableSet;
  21 import com.google.common.collect.Sets;
  22 import com.google.common.util.concurrent.FluentFuture;
  23 import com.google.common.util.concurrent.FutureCallback;
  24 import com.google.common.util.concurrent.Futures;
  25 import com.google.common.util.concurrent.ListenableFuture;
  26 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  27 import java.util.ArrayList;
  28 import java.util.Collections;
  29 import java.util.HashMap;
  30 import java.util.HashSet;
  31 import java.util.List;
  32 import java.util.Map;
  33 import java.util.Objects;
  34 import java.util.Optional;
  35 import java.util.Set;
  36 import java.util.concurrent.TimeUnit;
  37 import java.util.stream.Collectors;
  38 import org.checkerframework.checker.lock.qual.GuardedBy;
  39 import org.checkerframework.checker.lock.qual.Holding;
  40 import org.eclipse.jdt.annotation.NonNull;
  41 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  42 import org.opendaylight.mdsal.common.api.CommitInfo;
  43 import org.opendaylight.protocol.bgp.openconfig.spi.BGPTableTypeRegistryConsumer;
  44 import org.opendaylight.protocol.bgp.parser.BGPDocumentedException;
  45 import org.opendaylight.protocol.bgp.parser.BGPError;
  46 import org.opendaylight.protocol.bgp.parser.impl.message.update.LocalPreferenceAttributeParser;
  47 import org.opendaylight.protocol.bgp.parser.spi.MessageUtil;
  48 import org.opendaylight.protocol.bgp.parser.spi.RevisedErrorHandlingSupport;
  49 import org.opendaylight.protocol.bgp.rib.impl.config.BgpPeer;
  50 import org.opendaylight.protocol.bgp.rib.impl.config.GracefulRestartUtil;
  51 import org.opendaylight.protocol.bgp.rib.impl.spi.BGPSessionPreferences;
  52 import org.opendaylight.protocol.bgp.rib.impl.spi.RIB;
  53 import org.opendaylight.protocol.bgp.rib.impl.state.BGPSessionStateProvider;
  54 import org.opendaylight.protocol.bgp.rib.spi.BGPSession;
  55 import org.opendaylight.protocol.bgp.rib.spi.BGPSessionListener;
  56 import org.opendaylight.protocol.bgp.rib.spi.BGPTerminationReason;
  57 import org.opendaylight.protocol.bgp.rib.spi.RIBSupport;
  58 import org.opendaylight.protocol.bgp.rib.spi.RouterIds;
  59 import org.opendaylight.protocol.bgp.rib.spi.state.BGPSessionState;
  60 import org.opendaylight.protocol.bgp.rib.spi.state.BGPTimersState;
  61 import org.opendaylight.protocol.bgp.rib.spi.state.BGPTransportState;
  62 import org.opendaylight.protocol.util.Ipv4Util;
  63 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.AsNumber;
  64 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddressNoZone;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.ipv4.prefixes.DestinationIpv4Builder;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.ipv4.prefixes.destination.ipv4.Ipv4Prefixes;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.ipv4.prefixes.destination.ipv4.Ipv4PrefixesBuilder;
  68 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;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.Update;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.open.message.BgpParameters;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.path.attributes.Attributes;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.path.attributes.AttributesBuilder;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev200120.update.message.Nlri;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.BgpAddPathTableType;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.BgpTableType;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.RouteRefresh;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.SendReceive;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.reach.MpReachNlri;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.reach.MpReachNlriBuilder;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.reach.mp.reach.nlri.AdvertizedRoutesBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.unreach.MpUnreachNlri;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.unreach.MpUnreachNlriBuilder;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.attributes.unreach.mp.unreach.nlri.WithdrawnRoutesBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.mp.capabilities.GracefulRestartCapability;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.mp.capabilities.add.path.capability.AddressFamilies;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.ResetSession;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.RestartGracefully;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.RouteRefreshRequest;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.PeerRole;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.PeerKey;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.rib.TablesKey;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.ClusterIdentifier;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.Ipv4AddressFamily;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.RouteTarget;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev200120.UnicastSubsequentAddressFamily;
  96 import org.opendaylight.yangtools.concepts.Registration;
  97 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  98 import org.opendaylight.yangtools.yang.binding.Notification;
  99 import org.opendaylight.yangtools.yang.common.Empty;
 100 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 101 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
 102 import org.slf4j.Logger;
 103 import org.slf4j.LoggerFactory;
 104
 105 /**
 106  * Class representing a peer. We have a single instance for each peer, which provides translation from BGP events into
 107  * RIB actions.
 108  */
 109 public class BGPPeer extends AbstractPeer implements BGPSessionListener {
 110     private static final Logger LOG = LoggerFactory.getLogger(BGPPeer.class);
 111     private static final TablesKey IPV4_UCAST_TABLE_KEY =
 112         new TablesKey(Ipv4AddressFamily.VALUE, UnicastSubsequentAddressFamily.VALUE);
 113
 114     private final RIB rib;
 115
 116     // FIXME: Alright, this right here is a ton of state which has intertwined initialization and dependencies Split
 117     //        these out into separate behavior objects. This also has relationship with state in AbstractPeer -- which
 118     //        hints at an obvious layer of indirection. Yeah, yeah, we can always add one of those, but the point
 119     //        is that this class is a mutable meeting point, whereas the behaviour has captured invariants.
 120     private final LoadingCache<NodeIdentifierWithPredicates, YangInstanceIdentifier> tablesIId =
 121         CacheBuilder.newBuilder().build(new CacheLoader<NodeIdentifierWithPredicates, YangInstanceIdentifier>() {
 122             @Override
 123             public YangInstanceIdentifier load(final NodeIdentifierWithPredicates key) {
 124                 return peerRibOutIId.node(TABLES_NID).node(key).toOptimized();
 125             }
 126         });
 127
 128     private ImmutableSet<TablesKey> tables = ImmutableSet.of();
 129     private final Map<TablesKey, AdjRibOutListener> adjRibOutListenerSet = new HashMap<>();
 130     private final List<RouteTarget> rtMemberships = new ArrayList<>();
 131     private final RpcProviderService rpcRegistry;
 132     private final BGPTableTypeRegistryConsumer tableTypeRegistry;
 133     private final BgpPeer bgpPeer;
 134
 135     // FIXME: This should be a constant co-located with ApplicationPeer.peerId
 136     private YangInstanceIdentifier peerRibOutIId;
 137     @GuardedBy("this")
 138     private Registration trackerRegistration;
 139
 140     @GuardedBy("this")
 141     private BGPSession currentSession;
 142     @GuardedBy("this")
 143     private AdjRibInWriter ribWriter;
 144     @GuardedBy("this")
 145     private EffectiveRibInWriter effRibInWriter;
 146     private Registration rpcRegistration;
 147     private ImmutableMap<TablesKey, SendReceive> addPathTableMaps = ImmutableMap.of();
 148     // FIXME: This should be a constant co-located with ApplicationPeer.peerId
 149     private YangInstanceIdentifier peerPath;
 150     // FIXME: This is for supportsTable() -- a trivial behavior thing, where 'peer-down' type states always return false
 151     private boolean sessionUp;
 152     private boolean llgrSupport;
 153     private Stopwatch peerRestartStopwatch;
 154     private long currentSelectionDeferralTimerSeconds;
 155     private final List<TablesKey> missingEOT = new ArrayList<>();
 156
 157     public BGPPeer(
 158             final BGPTableTypeRegistryConsumer tableTypeRegistry,
 159             final IpAddressNoZone neighborAddress,
 160             final String peerGroupName,
 161             final RIB rib,
 162             final PeerRole role,
 163             final ClusterIdentifier clusterId,
 164             final AsNumber localAs,
 165             final RpcProviderService rpcRegistry,
 166             final Set<TablesKey> afiSafisAdvertized,
 167             final Set<TablesKey> afiSafisGracefulAdvertized,
 168             final Map<TablesKey, Integer> llGracefulTablesAdvertised,
 169             final BgpPeer bgpPeer) {
 170         super(rib, Ipv4Util.toStringIP(neighborAddress), peerGroupName, role, clusterId, localAs, neighborAddress,
 171             afiSafisAdvertized, afiSafisGracefulAdvertized, llGracefulTablesAdvertised);
 172         this.tableTypeRegistry = requireNonNull(tableTypeRegistry);
 173         this.rib = requireNonNull(rib);
 174         this.rpcRegistry = rpcRegistry;
 175         this.bgpPeer = bgpPeer;
 176
 177         createDomChain();
 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(List.of(
 424                     (ResetSession) bgpPeerHandler::resetSession,
 425                     (RestartGracefully) bgpPeerHandler::restartGracefully,
 426                     (RouteRefreshRequest) bgpPeerHandler::routeRefreshRequest), ImmutableSet.of(
 427                         rib.getInstanceIdentifier().child(
 428                             org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib
 429                             .rib.Peer.class, new PeerKey(peerId))));
 430             }
 431         } else {
 432             final Set<TablesKey> forwardingTables;
 433             if (advertisedTables == null) {
 434                 forwardingTables = Collections.emptySet();
 435             } else {
 436                 forwardingTables = advertisedTables.values().stream()
 437                         .filter(table -> table.getAfiFlags() != null)
 438                         .filter(table -> table.getAfiFlags().getForwardingState())
 439                         .map(table -> new TablesKey(table.getAfi(), table.getSafi()))
 440                         .collect(Collectors.toSet());
 441             }
 442             ribWriter.clearTables(Sets.difference(tables, forwardingTables));
 443             if (restartingLocally) {
 444                 effRibInWriter.close();
 445                 peerRestartStopwatch = Stopwatch.createStarted();
 446                 handleSelectionReferralTimer();
 447                 missingEOT.addAll(tables);
 448             }
 449         }
 450         if (advertisedTables == null || advertisedTables.isEmpty()) {
 451             setAdvertizedGracefulRestartTableTypes(Collections.emptyList());
 452         } else {
 453             setAdvertizedGracefulRestartTableTypes(advertisedTables.values().stream()
 454                     .map(t -> new TablesKey(t.getAfi(), t.getSafi())).collect(Collectors.toList()));
 455         }
 456         setAfiSafiGracefulRestartState(advertisedGracefulRestartCapability.getRestartTime().toJava(), false,
 457             restartingLocally);
 458
 459         final Map<TablesKey, Integer> llTablesReceived;
 460         if (advertisedLLTables != null) {
 461             llTablesReceived = new HashMap<>();
 462             for (var table : advertisedLLTables.values()) {
 463                 llTablesReceived.put(new TablesKey(table.getAfi(), table.getSafi()),
 464                     table.getLongLivedStaleTime().getValue().intValue());
 465             }
 466         } else {
 467             llTablesReceived = Collections.emptyMap();
 468         }
 469         setAdvertizedLlGracefulRestartTableTypes(llTablesReceived);
 470
 471         if (!llTablesReceived.isEmpty()) {
 472             llgrSupport = true;
 473             // FIXME: propagate preserved tables
 474         } else {
 475             // FIXME: clear preserved tables
 476             llgrSupport = false;
 477         }
 478
 479         if (!restartingLocally) {
 480             if (!setTables.contains(IPV4_UCAST_TABLE_KEY)) {
 481                 createAdjRibOutListener(IPV4_UCAST_TABLE_KEY, false);
 482             }
 483             for (final TablesKey key : getAfiSafisAdvertized()) {
 484                 createAdjRibOutListener(key, true);
 485             }
 486         }
 487
 488         // SpotBugs does not grok Optional.ifPresent() and thinks we are using unsynchronized access
 489         final Optional<RevisedErrorHandlingSupport> errorHandling = bgpPeer.getErrorHandling();
 490         if (errorHandling.isPresent()) {
 491             currentSession.addDecoderConstraint(RevisedErrorHandlingSupport.class, errorHandling.orElseThrow());
 492         }
 493     }
 494
 495     private boolean isRestartingGracefully() {
 496         return isLocalRestarting() || isPeerRestarting();
 497     }
 498
 499     private synchronized void createEffRibInWriter() {
 500         final var chain = rib.createPeerDOMChain();
 501         chain.addCallback(this);
 502
 503         effRibInWriter = new EffectiveRibInWriter(this, rib, chain, peerPath, tables, tableTypeRegistry, rtMemberships,
 504             rtCache);
 505     }
 506
 507     //try to add a support for old-school BGP-4, if peer did not advertise IPv4-Unicast MP capability
 508     @Holding("this")
 509     private void addBgp4Support() {
 510         if (!tables.contains(IPV4_UCAST_TABLE_KEY)) {
 511             final HashSet<TablesKey> newSet = new HashSet<>(tables);
 512             newSet.add(IPV4_UCAST_TABLE_KEY);
 513             tables = ImmutableSet.copyOf(newSet);
 514         }
 515     }
 516
 517     @Holding("this")
 518     private void createAdjRibOutListener(final TablesKey key, final boolean mpSupport) {
 519         final RIBSupport<?, ?> ribSupport = rib.getRibSupportContext().getRIBSupport(key);
 520
 521         // not particularly nice
 522         if (ribSupport != null && currentSession instanceof BGPSessionImpl bgpSession) {
 523             final AdjRibOutListener adjRibOut = AdjRibOutListener.create(peerId, rib.getYangRibId(),
 524                 rib.getCodecsRegistry(), ribSupport, rib.getService(), bgpSession.getLimiter(), mpSupport);
 525             adjRibOutListenerSet.put(key, adjRibOut);
 526             registerPrefixesSentCounter(key, adjRibOut);
 527         }
 528     }
 529
 530     @Override
 531     public synchronized void onSessionDown(final BGPSession session, final Exception exc) {
 532         if (exc.getMessage().equals(BGPSessionImpl.END_OF_INPUT)) {
 533             LOG.info("Session with peer {} went down", getName());
 534         } else {
 535             LOG.info("Session with peer {} went down", getName(), exc);
 536         }
 537         releaseConnectionGracefully();
 538     }
 539
 540     @Override
 541     public synchronized void onSessionTerminated(final BGPSession session, final BGPTerminationReason cause) {
 542         LOG.info("Session with peer {} terminated: {}", getName(), cause);
 543         releaseConnectionGracefully();
 544     }
 545
 546     @Override
 547     public String toString() {
 548         return MoreObjects.toStringHelper(this).add("name", getName()).add("tables", tables).toString();
 549     }
 550
 551     @Override
 552     public synchronized FluentFuture<? extends CommitInfo> releaseConnection() {
 553         return releaseConnection(true);
 554     }
 555
 556     /**
 557      * On transaction chain failure, we don't want to wait for future.
 558      *
 559      * @param isWaitForSubmitted if true, wait for submitted future before closing binding chain. if false, don't wait.
 560      */
 561     @Holding("this")
 562     private @NonNull FluentFuture<? extends CommitInfo> releaseConnection(final boolean isWaitForSubmitted) {
 563         LOG.info("Closing session with peer");
 564         sessionUp = false;
 565         adjRibOutListenerSet.values().forEach(AdjRibOutListener::close);
 566         adjRibOutListenerSet.clear();
 567         final FluentFuture<? extends CommitInfo> future;
 568         // FIXME: this is a typical example of something which should be handled by a behavior into which we have
 569         //        transitioned way before this method is called. This really begs to be an abstract base class with
 570         //        a 'clearTables' or similar callout
 571         if (isRestartingGracefully()) {
 572             final Set<TablesKey> gracefulTables = getGracefulTables();
 573             ribWriter.storeStaleRoutes(gracefulTables);
 574             future = ribWriter.clearTables(Sets.difference(tables, gracefulTables));
 575             if (isPeerRestarting()) {
 576                 peerRestartStopwatch = Stopwatch.createStarted();
 577                 handleRestartTimer();
 578             }
 579         } else {
 580             future = terminateConnection();
 581         }
 582         releaseRibOutChain(isWaitForSubmitted);
 583
 584         closeSession();
 585         return future;
 586     }
 587
 588     @Holding("this")
 589     @SuppressWarnings("checkstyle:illegalCatch")
 590     private void closeSession() {
 591         if (currentSession != null) {
 592             try {
 593                 if (isRestartingGracefully()) {
 594                     currentSession.closeWithoutMessage();
 595                 } else {
 596                     currentSession.close();
 597                 }
 598             } catch (final Exception e) {
 599                 LOG.warn("Error closing session with peer", e);
 600             }
 601             currentSession = null;
 602         }
 603     }
 604
 605     private Set<TablesKey> getGracefulTables() {
 606         return tables.stream()
 607                 .filter(this::isGracefulRestartReceived)
 608                 .filter(this::isGracefulRestartAdvertized)
 609                 .collect(Collectors.toSet());
 610     }
 611
 612     private synchronized FluentFuture<? extends CommitInfo> terminateConnection() {
 613         if (trackerRegistration != null) {
 614             trackerRegistration.close();
 615             trackerRegistration = null;
 616         }
 617         if (rpcRegistration != null) {
 618             rpcRegistration.close();
 619         }
 620         ribWriter.releaseChain();
 621
 622         if (effRibInWriter != null) {
 623             effRibInWriter.close();
 624         }
 625         tables = ImmutableSet.of();
 626         addPathTableMaps = ImmutableMap.of();
 627         final var future = removePeer(peerPath);
 628         resetState();
 629
 630         return future;
 631     }
 632
 633     /**
 634      * If Graceful Restart Timer expires, remove all routes advertised by peer.
 635      */
 636     private synchronized void handleRestartTimer() {
 637         if (!isPeerRestarting()) {
 638             return;
 639         }
 640
 641         final long peerRestartTimeNanos = TimeUnit.SECONDS.toNanos(getPeerRestartTime());
 642         final long elapsedNanos = peerRestartStopwatch.elapsed(TimeUnit.NANOSECONDS);
 643         if (elapsedNanos >= peerRestartTimeNanos) {
 644             setAfiSafiGracefulRestartState(0, false, false);
 645             onSessionTerminated(currentSession, new BGPTerminationReason(BGPError.HOLD_TIMER_EXPIRED));
 646         }
 647
 648         currentSession.schedule(this::handleRestartTimer, peerRestartTimeNanos - elapsedNanos, TimeUnit.NANOSECONDS);
 649     }
 650
 651     private synchronized void handleSelectionReferralTimer() {
 652         if (!isLocalRestarting()) {
 653             return;
 654         }
 655
 656         final long referalTimerNanos = TimeUnit.SECONDS.toNanos(currentSelectionDeferralTimerSeconds);
 657         final long elapsedNanos = peerRestartStopwatch.elapsed(TimeUnit.NANOSECONDS);
 658         if (elapsedNanos >= referalTimerNanos) {
 659             missingEOT.clear();
 660             handleGracefulEndOfRib();
 661         }
 662         currentSession.schedule(this::handleSelectionReferralTimer, referalTimerNanos - elapsedNanos,
 663             TimeUnit.NANOSECONDS);
 664     }
 665
 666     @Holding("this")
 667     private void releaseConnectionGracefully() {
 668         if (getPeerRestartTime() > 0) {
 669             setRestartingState();
 670         }
 671         releaseConnection(true);
 672     }
 673
 674     @SuppressFBWarnings("IS2_INCONSISTENT_SYNC")
 675     @Override
 676     public SendReceive getSupportedAddPathTables(final TablesKey tableKey) {
 677         return addPathTableMaps.get(tableKey);
 678     }
 679
 680     @Override
 681     public boolean supportsTable(final TablesKey tableKey) {
 682         return sessionUp && getAfiSafisAdvertized().contains(tableKey) && tables.contains(tableKey);
 683     }
 684
 685     @Override
 686     public YangInstanceIdentifier getRibOutIId(final NodeIdentifierWithPredicates tablekey) {
 687         return tablesIId.getUnchecked(tablekey);
 688     }
 689
 690     @Override
 691     public synchronized void onFailure(final Throwable cause) {
 692         LOG.error("Transaction domChain failed.", cause);
 693         releaseConnection(true);
 694     }
 695
 696     private synchronized void onRibOutChainFailed(final Throwable cause) {
 697         LOG.error("RibOut transaction chain failed.", cause);
 698         releaseConnection(false);
 699     }
 700
 701     @Override
 702     public synchronized void markUptodate(final TablesKey tablesKey) {
 703         ribWriter.markTableUptodate(tablesKey);
 704     }
 705
 706     @Override
 707     public synchronized BGPSessionState getBGPSessionState() {
 708         if (currentSession instanceof BGPSessionStateProvider stateProvider) {
 709             return stateProvider.getBGPSessionState();
 710         }
 711         return null;
 712     }
 713
 714     @Override
 715     public synchronized BGPTimersState getBGPTimersState() {
 716         if (currentSession instanceof BGPSessionStateProvider stateProvider) {
 717             return stateProvider.getBGPTimersState();
 718         }
 719         return null;
 720     }
 721
 722     @Override
 723     public synchronized BGPTransportState getBGPTransportState() {
 724         if (currentSession instanceof BGPSessionStateProvider stateProvider) {
 725             return stateProvider.getBGPTransportState();
 726         }
 727         return null;
 728     }
 729
 730     @Override
 731     public List<RouteTarget> getMemberships() {
 732         return rtMemberships;
 733     }
 734
 735     @Override
 736     public synchronized ListenableFuture<?> restartGracefully(final long selectionDeferralTimerSeconds) {
 737         final Set<TablesKey> tablesToPreserve = getGracefulTables();
 738         if (tablesToPreserve == null || tablesToPreserve.isEmpty()) {
 739             LOG.info("Peer {} is not capable of graceful restart or have no matching graceful tables.", peerId);
 740             return Futures.immediateFailedFuture(new UnsupportedOperationException(
 741                     "Peer is not capable of graceful restart"));
 742         }
 743         setGracefulPreferences(true, tablesToPreserve);
 744         currentSelectionDeferralTimerSeconds = selectionDeferralTimerSeconds;
 745         setLocalRestartingState(true);
 746         return releaseConnection(true);
 747     }
 748
 749     @Override
 750     boolean supportsLLGR() {
 751         return llgrSupport;
 752     }
 753
 754     private synchronized void setGracefulPreferences(final boolean localRestarting,
 755                                                      final Set<TablesKey> preservedTables) {
 756         final Set<TablesKey> gracefulTables = tables.stream()
 757                 .filter(this::isGracefulRestartAdvertized)
 758                 .collect(Collectors.toSet());
 759         final BgpParameters bgpParameters = GracefulRestartUtil.getGracefulBgpParameters(
 760                 bgpPeer.getBgpFixedCapabilities(), gracefulTables, preservedTables,
 761                 bgpPeer.getGracefulRestartTimer(), localRestarting, Collections.emptySet());
 762         final BGPSessionPreferences oldPrefs = rib.getDispatcher().getBGPPeerRegistry()
 763                 .getPeerPreferences(getNeighborAddress());
 764         final BGPSessionPreferences newPrefs = new BGPSessionPreferences(
 765                 oldPrefs.getMyAs(),
 766                 oldPrefs.getHoldTime(),
 767                 oldPrefs.getBgpId(),
 768                 oldPrefs.getExpectedRemoteAs(),
 769                 Collections.singletonList(bgpParameters),
 770                 oldPrefs.getMd5Password());
 771         rib.getDispatcher().getBGPPeerRegistry()
 772                 .updatePeerPreferences(getNeighborAddress(), newPrefs);
 773     }
 774 }