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
  12 import com.google.common.base.MoreObjects;
  13 import com.google.common.base.Objects;
  14 import com.google.common.base.Stopwatch;
  15 import com.google.common.cache.CacheBuilder;
  16 import com.google.common.cache.CacheLoader;
  17 import com.google.common.cache.LoadingCache;
  18 import com.google.common.collect.ImmutableMap;
  19 import com.google.common.collect.ImmutableSet;
  20 import com.google.common.collect.Sets;
  21 import com.google.common.net.InetAddresses;
  22 import com.google.common.util.concurrent.FluentFuture;
  23 import com.google.common.util.concurrent.Futures;
  24 import com.google.common.util.concurrent.ListenableFuture;
  25 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  26 import java.util.ArrayList;
  27 import java.util.Collections;
  28 import java.util.HashMap;
  29 import java.util.HashSet;
  30 import java.util.List;
  31 import java.util.Map;
  32 import java.util.Optional;
  33 import java.util.Set;
  34 import java.util.concurrent.ScheduledThreadPoolExecutor;
  35 import java.util.concurrent.TimeUnit;
  36 import java.util.stream.Collectors;
  37 import javax.annotation.concurrent.GuardedBy;
  38 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  39 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
  40 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
  41 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  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.rib.impl.config.BgpPeer;
  49 import org.opendaylight.protocol.bgp.rib.impl.config.GracefulRestartUtil;
  50 import org.opendaylight.protocol.bgp.rib.impl.spi.BGPSessionPreferences;
  51 import org.opendaylight.protocol.bgp.rib.impl.spi.RIB;
  52 import org.opendaylight.protocol.bgp.rib.impl.state.BGPSessionStateProvider;
  53 import org.opendaylight.protocol.bgp.rib.spi.BGPSession;
  54 import org.opendaylight.protocol.bgp.rib.spi.BGPSessionListener;
  55 import org.opendaylight.protocol.bgp.rib.spi.BGPTerminationReason;
  56 import org.opendaylight.protocol.bgp.rib.spi.RIBSupport;
  57 import org.opendaylight.protocol.bgp.rib.spi.RouterIds;
  58 import org.opendaylight.protocol.bgp.rib.spi.state.BGPSessionState;
  59 import org.opendaylight.protocol.bgp.rib.spi.state.BGPTimersState;
  60 import org.opendaylight.protocol.bgp.rib.spi.state.BGPTransportState;
  61 import org.opendaylight.protocol.concepts.AbstractRegistration;
  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.IpAddress;
  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.rev180329.Update;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev180329.open.message.BgpParameters;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev180329.path.attributes.Attributes;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev180329.path.attributes.AttributesBuilder;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev180329.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.mp.capabilities.GracefulRestartCapability;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.mp.capabilities.add.path.capability.AddressFamilies;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.update.attributes.MpReachNlri;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.update.attributes.MpReachNlriBuilder;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.update.attributes.MpUnreachNlri;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.update.attributes.MpUnreachNlriBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.update.attributes.mp.reach.nlri.AdvertizedRoutesBuilder;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.update.attributes.mp.unreach.nlri.WithdrawnRoutesBuilder;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.BgpPeerRpcService;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.PeerContext;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.PeerRole;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.PeerKey;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.peer.AdjRibOut;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.rib.Tables;
  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.rev180329.AddressFamily;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.ClusterIdentifier;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.Ipv4AddressFamily;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.RouteTarget;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.SubsequentAddressFamily;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.UnicastSubsequentAddressFamily;
  99 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 100 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 101 import org.opendaylight.yangtools.yang.binding.Notification;
 102 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 103 import org.slf4j.Logger;
 104 import org.slf4j.LoggerFactory;
 105
 106 /**
 107  * Class representing a peer. We have a single instance for each peer, which provides translation from BGP events into
 108  * RIB actions.
 109  */
 110 public class BGPPeer extends AbstractPeer implements BGPSessionListener {
 111     private static final Logger LOG = LoggerFactory.getLogger(BGPPeer.class);
 112     private static final TablesKey IPV4_UCAST_TABLE_KEY = new TablesKey(Ipv4AddressFamily.class, UnicastSubsequentAddressFamily.class);
 113
 114     private Set<TablesKey> tables = Collections.emptySet();
 115     private final RIB rib;
 116     private final Map<TablesKey, AdjRibOutListener> adjRibOutListenerSet = new HashMap<>();
 117     private final List<RouteTarget> rtMemberships = new ArrayList<>();
 118     private final RpcProviderRegistry rpcRegistry;
 119     private final BGPTableTypeRegistryConsumer tableTypeRegistry;
 120     private final BgpPeer bgpPeer;
 121     private InstanceIdentifier<AdjRibOut> peerRibOutIId;
 122     private KeyedInstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib
 123             .rev180329.bgp.rib.rib.Peer, PeerKey> peerIId;
 124     @GuardedBy("this")
 125     private AbstractRegistration trackerRegistration;
 126     private final LoadingCache<TablesKey, KeyedInstanceIdentifier<Tables, TablesKey>> tablesIId
 127             = CacheBuilder.newBuilder()
 128             .build(new CacheLoader<TablesKey, KeyedInstanceIdentifier<Tables, TablesKey>>() {
 129                 @Override
 130                 public KeyedInstanceIdentifier<Tables, TablesKey> load(final TablesKey tablesKey) {
 131                     return BGPPeer.this.peerRibOutIId.child(Tables.class, tablesKey);
 132                 }
 133             });
 134
 135     @GuardedBy("this")
 136     private BGPSession session;
 137     @GuardedBy("this")
 138     private AdjRibInWriter ribWriter;
 139     @GuardedBy("this")
 140     private EffectiveRibInWriter effRibInWriter;
 141     private RoutedRpcRegistration<BgpPeerRpcService> rpcRegistration;
 142     private Map<TablesKey, SendReceive> addPathTableMaps = Collections.emptyMap();
 143     private YangInstanceIdentifier peerPath;
 144     private boolean sessionUp;
 145     private Stopwatch peerRestartStopwatch;
 146     private long selectionDeferralTimerSeconds;
 147     private final List<TablesKey> missingEOT = new ArrayList<>();
 148
 149     public BGPPeer(
 150             final BGPTableTypeRegistryConsumer tableTypeRegistry,
 151             final IpAddress neighborAddress,
 152             final String peerGroupName,
 153             final RIB rib,
 154             final PeerRole role,
 155             final ClusterIdentifier clusterId,
 156             final AsNumber localAs,
 157             final RpcProviderRegistry rpcRegistry,
 158             final Set<TablesKey> afiSafisAdvertized,
 159             final Set<TablesKey> afiSafisGracefulAdvertized,
 160             final BgpPeer bgpPeer) {
 161         super(rib, Ipv4Util.toStringIP(neighborAddress), peerGroupName, role, clusterId,
 162                 localAs, neighborAddress, afiSafisAdvertized, afiSafisGracefulAdvertized);
 163         this.tableTypeRegistry = requireNonNull(tableTypeRegistry);
 164         this.rib = requireNonNull(rib);
 165         this.rpcRegistry = rpcRegistry;
 166         this.bgpPeer = bgpPeer;
 167     }
 168
 169     BGPPeer(
 170             final BGPTableTypeRegistryConsumer tableTypeRegistry,
 171             final IpAddress neighborAddress,
 172             final RIB rib,
 173             final PeerRole role,
 174             final RpcProviderRegistry rpcRegistry,
 175             final Set<TablesKey> afiSafisAdvertized,
 176             final Set<TablesKey> afiSafisGracefulAdvertized) {
 177         this(tableTypeRegistry, neighborAddress, null, rib, role, null, null, rpcRegistry,
 178                 afiSafisAdvertized, afiSafisGracefulAdvertized, null);
 179     }
 180
 181
 182     private static Attributes nextHopToAttribute(final Attributes attrs, final MpReachNlri mpReach) {
 183         if (attrs.getCNextHop() == null && mpReach.getCNextHop() != null) {
 184             final AttributesBuilder attributesBuilder = new AttributesBuilder(attrs);
 185             attributesBuilder.setCNextHop(mpReach.getCNextHop());
 186             return attributesBuilder.build();
 187         }
 188         return attrs;
 189     }
 190
 191     /**
 192      * Creates MPReach for the prefixes to be handled in the same way as linkstate routes.
 193      *
 194      * @param message Update message containing prefixes in NLRI
 195      * @return MpReachNlri with prefixes from the nlri field
 196      */
 197     private static MpReachNlri prefixesToMpReach(final Update message) {
 198         final List<Ipv4Prefixes> prefixes = message.getNlri().stream()
 199                 .map(n -> new Ipv4PrefixesBuilder().setPrefix(n.getPrefix()).setPathId(n.getPathId()).build())
 200                 .collect(Collectors.toList());
 201         final MpReachNlriBuilder b = new MpReachNlriBuilder().setAfi(Ipv4AddressFamily.class).setSafi(
 202                 UnicastSubsequentAddressFamily.class).setAdvertizedRoutes(
 203                 new AdvertizedRoutesBuilder().setDestinationType(
 204                         new DestinationIpv4CaseBuilder().setDestinationIpv4(
 205                                 new DestinationIpv4Builder().setIpv4Prefixes(prefixes).build()).build()).build());
 206         if (message.getAttributes() != null) {
 207             b.setCNextHop(message.getAttributes().getCNextHop());
 208         }
 209         return b.build();
 210     }
 211
 212     /**
 213      * Create MPUnreach for the prefixes to be handled in the same way as linkstate routes.
 214      *
 215      * @param message            Update message containing withdrawn routes
 216      * @param isAnyNlriAnnounced isAnyNlriAnnounced
 217      * @return MpUnreachNlri with prefixes from the withdrawn routes field
 218      */
 219     private static MpUnreachNlri prefixesToMpUnreach(final Update message, final boolean isAnyNlriAnnounced) {
 220         final List<Ipv4Prefixes> prefixes = new ArrayList<>();
 221         message.getWithdrawnRoutes().forEach(w -> {
 222
 223             Optional<Nlri> nlriAnounced = Optional.empty();
 224             if (isAnyNlriAnnounced) {
 225                 nlriAnounced = message.getNlri().stream().filter(n -> Objects.equal(n.getPrefix(), w.getPrefix())
 226                         && Objects.equal(n.getPathId(), w.getPathId()))
 227                         .findAny();
 228             }
 229             if (!nlriAnounced.isPresent()) {
 230                 prefixes.add(new Ipv4PrefixesBuilder().setPrefix(w.getPrefix()).setPathId(w.getPathId()).build());
 231             }
 232         });
 233         return new MpUnreachNlriBuilder().setAfi(Ipv4AddressFamily.class).setSafi(UnicastSubsequentAddressFamily.class)
 234                 .setWithdrawnRoutes(new WithdrawnRoutesBuilder().setDestinationType(new org.opendaylight.yang.gen.v1
 235                         .urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.update.attributes.mp.unreach.nlri
 236                         .withdrawn.routes.destination.type.DestinationIpv4CaseBuilder().setDestinationIpv4(
 237                         new DestinationIpv4Builder().setIpv4Prefixes(prefixes).build()).build()).build()).build();
 238     }
 239
 240     private static Map<TablesKey, SendReceive> mapTableTypesFamilies(final List<AddressFamilies> addPathTablesType) {
 241         return ImmutableMap.copyOf(addPathTablesType.stream().collect(Collectors.toMap(af -> new TablesKey(af.getAfi(),
 242                         af.getSafi()), BgpAddPathTableType::getSendReceive)));
 243     }
 244
 245     public synchronized void instantiateServiceInstance() {
 246         this.ribWriter = AdjRibInWriter.create(this.rib.getYangRibId(), this.peerRole, this);
 247         setActive(true);
 248     }
 249
 250     @Override
 251     public synchronized FluentFuture<? extends CommitInfo> close() {
 252         final FluentFuture<? extends CommitInfo> future = releaseConnection();
 253         closeDomChain();
 254         setActive(false);
 255         return future;
 256     }
 257
 258     @Override
 259     public void onMessage(final BGPSession session, final Notification msg) throws BGPDocumentedException {
 260         if (msg instanceof Update) {
 261             onUpdateMessage((Update) msg);
 262         } else if (msg instanceof RouteRefresh) {
 263             onRouteRefreshMessage((RouteRefresh) msg);
 264         } else {
 265             LOG.info("Ignoring unhandled message class {}", msg.getClass());
 266         }
 267     }
 268
 269     private void onRouteRefreshMessage(final RouteRefresh message) {
 270         final Class<? extends AddressFamily> rrAfi = message.getAfi();
 271         final Class<? extends SubsequentAddressFamily> rrSafi = message.getSafi();
 272
 273         final TablesKey key = new TablesKey(rrAfi, rrSafi);
 274         final AdjRibOutListener listener = this.adjRibOutListenerSet.get(key);
 275         if (listener != null) {
 276             listener.close();
 277             this.adjRibOutListenerSet.remove(key);
 278             createAdjRibOutListener(key, listener.isMpSupported());
 279         } else {
 280             LOG.info("Ignoring RouteRefresh message. Afi/Safi is not supported: {}, {}.", rrAfi, rrSafi);
 281         }
 282     }
 283
 284     /**
 285      * Check for presence of well known mandatory attribute LOCAL_PREF in Update message.
 286      *
 287      * @param message Update message
 288      */
 289     private void checkMandatoryAttributesPresence(final Update message) throws BGPDocumentedException {
 290         if (MessageUtil.isAnyNlriPresent(message)) {
 291             final Attributes attrs = message.getAttributes();
 292             if (this.peerRole == PeerRole.Ibgp && (attrs == null || attrs.getLocalPref() == null)) {
 293                 throw new BGPDocumentedException(BGPError.MANDATORY_ATTR_MISSING_MSG + "LOCAL_PREF",
 294                         BGPError.WELL_KNOWN_ATTR_MISSING,
 295                         new byte[]{LocalPreferenceAttributeParser.TYPE});
 296             }
 297         }
 298     }
 299
 300     /**
 301      * Process Update message received.
 302      * Calls {@link #checkMandatoryAttributesPresence(Update)} to check for presence of mandatory attributes.
 303      *
 304      * @param message Update message
 305      */
 306     private synchronized void onUpdateMessage(final Update message) throws BGPDocumentedException {
 307         checkMandatoryAttributesPresence(message);
 308
 309         // update AdjRibs
 310         final Attributes attrs = message.getAttributes();
 311         MpReachNlri mpReach;
 312         final boolean isAnyNlriAnnounced = message.getNlri() != null;
 313         if (isAnyNlriAnnounced) {
 314             mpReach = prefixesToMpReach(message);
 315         } else {
 316             mpReach = MessageUtil.getMpReachNlri(attrs);
 317         }
 318         if (mpReach != null) {
 319             this.ribWriter.updateRoutes(mpReach, nextHopToAttribute(attrs, mpReach));
 320         }
 321         MpUnreachNlri mpUnreach;
 322         if (message.getWithdrawnRoutes() != null) {
 323             mpUnreach = prefixesToMpUnreach(message, isAnyNlriAnnounced);
 324         } else {
 325             mpUnreach = MessageUtil.getMpUnreachNlri(attrs);
 326         }
 327         final boolean endOfRib = BgpPeerUtil.isEndOfRib(message);
 328         if (mpUnreach != null) {
 329             if (endOfRib) {
 330                 final TablesKey tablesKey = new TablesKey(mpUnreach.getAfi(), mpUnreach.getSafi());
 331                 this.ribWriter.removeStaleRoutes(tablesKey);
 332                 this.missingEOT.remove(tablesKey);
 333                 handleGracefulEndOfRib();
 334             } else {
 335                 this.ribWriter.removeRoutes(mpUnreach);
 336             }
 337         } else if (endOfRib) {
 338             this.ribWriter.removeStaleRoutes(IPV4_UCAST_TABLE_KEY);
 339             this.missingEOT.remove(IPV4_UCAST_TABLE_KEY);
 340             handleGracefulEndOfRib();
 341         }
 342     }
 343
 344     private synchronized void handleGracefulEndOfRib() {
 345         if (isLocalRestarting()) {
 346             if (this.missingEOT.isEmpty()) {
 347                 createEffRibInWriter();
 348                 this.effRibInWriter.init();
 349                 registerPrefixesCounters(this.effRibInWriter, this.effRibInWriter);
 350                 for (final TablesKey key : this.tables) {
 351                     createAdjRibOutListener(key, true);
 352                 }
 353                 setLocalRestartingState(false);
 354                 setGracefulPreferences(false, Collections.emptySet());
 355             }
 356         }
 357     }
 358
 359     @Override
 360     public synchronized void onSessionUp(final BGPSession session) {
 361         this.session = session;
 362         this.sessionUp = true;
 363         this.bindingChain = this.rib.createPeerChain(this);
 364         if (this.session instanceof BGPSessionStateProvider) {
 365             ((BGPSessionStateProvider) this.session).registerMessagesCounter(this);
 366         }
 367         final GracefulRestartCapability advertisedGracefulRestartCapability =
 368                 session.getAdvertisedGracefulRestartCapability();
 369         final List<org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.multiprotocol.rev180329.mp.capabilities.graceful.restart.capability.Tables> advertisedTables =
 370                 advertisedGracefulRestartCapability.getTables();
 371         final List<AddressFamilies> addPathTablesType = session.getAdvertisedAddPathTableTypes();
 372         final Set<BgpTableType> advertizedTableTypes = session.getAdvertisedTableTypes();
 373         LOG.info("Session with peer {} went up with tables {} and Add Path tables {}", this.name,
 374                 advertizedTableTypes, addPathTablesType);
 375         final Set<TablesKey> setTables = advertizedTableTypes.stream().map(t -> new TablesKey(t.getAfi(), t.getSafi()))
 376                 .collect(Collectors.toSet());
 377         this.tables = ImmutableSet.copyOf(setTables);
 378         this.addPathTableMaps = mapTableTypesFamilies(addPathTablesType);
 379         final boolean restartingLocally = isLocalRestarting();
 380
 381         if (!isRestartingGracefully()) {
 382             this.rawIdentifier = InetAddresses.forString(session.getBgpId().getValue()).getAddress();
 383             this.peerId = RouterIds.createPeerId(session.getBgpId());
 384             this.peerIId = getInstanceIdentifier().child(org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns
 385                             .yang.bgp.rib.rev180329.bgp.rib.rib.Peer.class, new PeerKey(this.peerId));
 386             this.peerPath = createPeerPath();
 387             this.peerRibOutIId = peerIId.child(AdjRibOut.class);
 388             this.trackerRegistration = this.rib.getPeerTracker().registerPeer(this);
 389             createEffRibInWriter();
 390             registerPrefixesCounters(this.effRibInWriter, this.effRibInWriter);
 391
 392             this.effRibInWriter.init();
 393             this.ribWriter = this.ribWriter.transform(this.peerId, this.peerPath, this.rib.getRibSupportContext(),
 394                     this.tables, this.addPathTableMaps);
 395
 396             if (this.rpcRegistry != null) {
 397                 this.rpcRegistration = this.rpcRegistry.addRoutedRpcImplementation(BgpPeerRpcService.class,
 398                         new BgpPeerRpc(this, session, this.tables));
 399                 final KeyedInstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib
 400                         .rev180329.bgp.rib.rib.Peer, PeerKey> path = this.rib.getInstanceIdentifier()
 401                         .child(org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib
 402                                 .rib.Peer.class, new PeerKey(this.peerId));
 403                 this.rpcRegistration.registerPath(PeerContext.class, path);
 404             }
 405         } else {
 406             final Set<TablesKey> forwardingTables;
 407             if (advertisedTables == null) {
 408                 forwardingTables = Collections.emptySet();
 409             } else {
 410                 forwardingTables = advertisedTables.stream()
 411                         .filter(table -> table.getAfiFlags() != null)
 412                         .filter(table -> table.getAfiFlags().isForwardingState())
 413                         .map(table -> new TablesKey(table.getAfi(), table.getSafi()))
 414                         .collect(Collectors.toSet());
 415             }
 416             this.ribWriter.clearTables(Sets.difference(this.tables, forwardingTables));
 417             if (restartingLocally) {
 418                 this.peerRestartStopwatch = Stopwatch.createStarted();
 419                 handleSelectionReferralTimer();
 420                 this.missingEOT.addAll(this.tables);
 421             }
 422         }
 423         if (advertisedTables == null ||
 424                 advertisedTables.isEmpty()) {
 425             setAdvertizedGracefulRestartTableTypes(Collections.emptyList());
 426         } else {
 427             setAdvertizedGracefulRestartTableTypes(advertisedTables.stream()
 428                     .map(t -> new TablesKey(t.getAfi(), t.getSafi())).collect(Collectors.toList()));
 429         }
 430         final int restartTime = advertisedGracefulRestartCapability.getRestartTime();
 431         setAfiSafiGracefulRestartState(restartTime, false, restartingLocally);
 432         if (!restartingLocally) {
 433             addBgp4Support();
 434             for (final TablesKey key : this.tables) {
 435                 createAdjRibOutListener(key, true);
 436             }
 437         }
 438     }
 439
 440     private boolean isRestartingGracefully() {
 441         return isLocalRestarting() || isPeerRestarting();
 442     }
 443
 444     private synchronized void createEffRibInWriter() {
 445         this.effRibInWriter = new EffectiveRibInWriter(this, this.rib,
 446                 this.rib.createPeerChain(this),
 447                 this.peerIId, this.tables, this.tableTypeRegistry,
 448                 this.rtMemberships,
 449                 this.rtCache);
 450     }
 451
 452     //try to add a support for old-school BGP-4, if peer did not advertise IPv4-Unicast MP capability
 453     private synchronized void addBgp4Support() {
 454         if (!this.tables.contains(IPV4_UCAST_TABLE_KEY)) {
 455             final HashSet<TablesKey> newSet = new HashSet<>(this.tables);
 456             newSet.add(IPV4_UCAST_TABLE_KEY);
 457             this.tables = ImmutableSet.copyOf(newSet);
 458             createAdjRibOutListener(IPV4_UCAST_TABLE_KEY, false);
 459         }
 460     }
 461
 462     private synchronized void createAdjRibOutListener(final TablesKey key,
 463                                                       final boolean mpSupport) {
 464         final RIBSupport<?, ?, ?, ?> ribSupport = this.rib.getRibSupportContext().getRIBSupport(key);
 465
 466         // not particularly nice
 467         if (ribSupport != null && this.session instanceof BGPSessionImpl) {
 468             final ChannelOutputLimiter limiter = ((BGPSessionImpl) this.session).getLimiter();
 469             final AdjRibOutListener adjRibOut = AdjRibOutListener.create(this.peerId, key,
 470                     this.rib.getYangRibId(), this.rib.getCodecsRegistry(), ribSupport,
 471                     this.rib.getService(), limiter, mpSupport);
 472             this.adjRibOutListenerSet.put(key, adjRibOut);
 473             registerPrefixesSentCounter(key, adjRibOut);
 474         }
 475     }
 476
 477     @Override
 478     public synchronized void onSessionDown(final BGPSession session, final Exception e) {
 479         if (e.getMessage().equals(BGPSessionImpl.END_OF_INPUT)) {
 480             LOG.info("Session with peer {} went down", this.name);
 481         } else {
 482             LOG.info("Session with peer {} went down", this.name, e);
 483         }
 484         releaseConnectionGracefully();
 485     }
 486
 487     @Override
 488     public synchronized void onSessionTerminated(final BGPSession session, final BGPTerminationReason cause) {
 489         LOG.info("Session with peer {} terminated: {}", this.name, cause);
 490         releaseConnectionGracefully();
 491     }
 492
 493     @Override
 494     public String toString() {
 495         return MoreObjects.toStringHelper(this)
 496                 .add("name", this.name)
 497                 .add("tables", this.tables).toString();
 498         }
 499
 500     @Override
 501     public synchronized FluentFuture<? extends CommitInfo> releaseConnection() {
 502         LOG.info("Closing session with peer");
 503         this.sessionUp = false;
 504         this.adjRibOutListenerSet.values().forEach(AdjRibOutListener::close);
 505         this.adjRibOutListenerSet.clear();
 506         final FluentFuture<? extends CommitInfo> future;
 507         if (!isRestartingGracefully()) {
 508             future = terminateConnection();
 509         } else {
 510             if (isLocalRestarting()){
 511                 this.effRibInWriter.close();
 512             }
 513             final Set<TablesKey> gracefulTables = getGracefulTables();
 514             this.ribWriter.storeStaleRoutes(gracefulTables);
 515             future = this.ribWriter.clearTables(Sets.difference(this.tables, gracefulTables));
 516             if (isPeerRestarting()) {
 517                 this.peerRestartStopwatch = Stopwatch.createStarted();
 518                 handleRestartTimer();
 519             }
 520         }
 521         releaseBindingChain();
 522
 523         if (this.session != null) {
 524             try {
 525                 if (isRestartingGracefully()) {
 526                     this.session.closeWithoutMessage();
 527                 } else {
 528                     this.session.close();
 529                 }
 530             } catch (final Exception e) {
 531                 LOG.warn("Error closing session with peer", e);
 532             }
 533             this.session = null;
 534         }
 535         return future;
 536     }
 537
 538     private Set<TablesKey> getGracefulTables() {
 539         return this.tables.stream()
 540                 .filter(this::isGracefulRestartReceived)
 541                 .filter(this::isGracefulRestartAdvertized)
 542                 .collect(Collectors.toSet());
 543     }
 544
 545     private synchronized FluentFuture<? extends CommitInfo> terminateConnection() {
 546         final FluentFuture<? extends CommitInfo> future;
 547         if (this.trackerRegistration != null) {
 548             this.trackerRegistration.close();
 549             this.trackerRegistration = null;
 550         }
 551         if (this.rpcRegistration != null) {
 552             this.rpcRegistration.close();
 553         }
 554         this.ribWriter.releaseChain();
 555
 556         if (this.effRibInWriter != null) {
 557             this.effRibInWriter.close();
 558         }
 559         this.tables = Collections.emptySet();
 560         this.addPathTableMaps = Collections.emptyMap();
 561         future = removePeer(this.peerPath);
 562         resetState();
 563
 564         return future;
 565     }
 566
 567     /**
 568      * If Graceful Restart Timer expires, remove all routes advertised by peer.
 569      */
 570     private synchronized void handleRestartTimer() {
 571         if (!isPeerRestarting()) {
 572             return;
 573         }
 574
 575         final long peerRestartTimeNanos = TimeUnit.SECONDS.toNanos(getPeerRestartTime());
 576         final long elapsedNanos = this.peerRestartStopwatch.elapsed(TimeUnit.NANOSECONDS);
 577         if (elapsedNanos >= peerRestartTimeNanos) {
 578             setAfiSafiGracefulRestartState(0, false, false);
 579             onSessionTerminated(this.session, new BGPTerminationReason(BGPError.HOLD_TIMER_EXPIRED));
 580         }
 581         new ScheduledThreadPoolExecutor(1)
 582                 .schedule(this::handleRestartTimer, peerRestartTimeNanos - elapsedNanos, TimeUnit.NANOSECONDS);
 583     }
 584
 585     private synchronized void handleSelectionReferralTimer() {
 586         if (!isLocalRestarting()) {
 587             return;
 588         }
 589
 590         final long referalTimerNanos = TimeUnit.SECONDS.toNanos(this.selectionDeferralTimerSeconds);
 591         final long elapsedNanos = this.peerRestartStopwatch.elapsed(TimeUnit.NANOSECONDS);
 592         if (elapsedNanos >= referalTimerNanos) {
 593             this.missingEOT.clear();
 594             handleGracefulEndOfRib();
 595         }
 596         new ScheduledThreadPoolExecutor(1)
 597                 .schedule(this::handleSelectionReferralTimer, referalTimerNanos - elapsedNanos, TimeUnit.NANOSECONDS);
 598     }
 599
 600     private void releaseConnectionGracefully() {
 601         if (getPeerRestartTime() > 0) {
 602             setRestartingState();
 603         }
 604         releaseConnection();
 605     }
 606
 607     @SuppressFBWarnings("IS2_INCONSISTENT_SYNC")
 608     @Override
 609     public SendReceive getSupportedAddPathTables(final TablesKey tableKey) {
 610         return this.addPathTableMaps.get(tableKey);
 611     }
 612
 613     @Override
 614     public boolean supportsTable(final TablesKey tableKey) {
 615         return this.tables.contains(tableKey) && this.sessionUp;
 616     }
 617
 618     @Override
 619     public KeyedInstanceIdentifier<Tables, TablesKey> getRibOutIId(final TablesKey tablesKey) {
 620         return this.tablesIId.getUnchecked(tablesKey);
 621     }
 622
 623     @Override
 624     public synchronized void onTransactionChainFailed(final TransactionChain<?, ?> chain,
 625             final AsyncTransaction<?, ?> transaction, final Throwable cause) {
 626         LOG.error("Transaction domChain failed.", cause);
 627         releaseConnection();
 628     }
 629
 630     @Override
 631     public synchronized void markUptodate(final TablesKey tablesKey) {
 632         this.ribWriter.markTableUptodate(tablesKey);
 633     }
 634
 635     @Override
 636     public synchronized BGPSessionState getBGPSessionState() {
 637         if (this.session instanceof BGPSessionStateProvider) {
 638             return ((BGPSessionStateProvider) this.session).getBGPSessionState();
 639         }
 640         return null;
 641     }
 642
 643     @Override
 644     public synchronized BGPTimersState getBGPTimersState() {
 645         if (this.session instanceof BGPSessionStateProvider) {
 646             return ((BGPSessionStateProvider) this.session).getBGPTimersState();
 647         }
 648         return null;
 649     }
 650
 651     @Override
 652     public synchronized BGPTransportState getBGPTransportState() {
 653         if (this.session instanceof BGPSessionStateProvider) {
 654             return ((BGPSessionStateProvider) this.session).getBGPTransportState();
 655         }
 656         return null;
 657     }
 658
 659     @Override
 660     public List<RouteTarget> getMemberships() {
 661         return this.rtMemberships;
 662     }
 663
 664     @Override
 665     public synchronized ListenableFuture<?> restartGracefully(final long selectionDeferralTimerSeconds) {
 666         final Set<TablesKey> tablesToPreserve = getGracefulTables();
 667         if (tablesToPreserve == null || tablesToPreserve.isEmpty()) {
 668             LOG.info("Peer {} is not capable of graceful restart or have no matching graceful tables.", this.peerId);
 669             return Futures.immediateFailedFuture(new UnsupportedOperationException(
 670                     "Peer is not capable of graceful restart"));
 671         }
 672         setGracefulPreferences(true, tablesToPreserve);
 673         this.selectionDeferralTimerSeconds = selectionDeferralTimerSeconds;
 674         setLocalRestartingState(true);
 675         return releaseConnection();
 676     }
 677
 678     private synchronized void setGracefulPreferences(final boolean localRestarting,
 679                                                      final Set<TablesKey> preservedTables) {
 680         final Set<TablesKey> gracefulTables = this.tables.stream()
 681                 .filter(this::isGracefulRestartAdvertized)
 682                 .collect(Collectors.toSet());
 683         final BgpParameters bgpParameters = GracefulRestartUtil.getGracefulBgpParameters(
 684                 this.bgpPeer.getBgpFixedCapabilities(), gracefulTables, preservedTables,
 685                 this.bgpPeer.getGracefulRestartTimer(), localRestarting);
 686         final BGPSessionPreferences oldPrefs = this.rib.getDispatcher().getBGPPeerRegistry()
 687                 .getPeerPreferences(getNeighborAddress());
 688         final BGPSessionPreferences newPrefs = new BGPSessionPreferences(
 689                 oldPrefs.getMyAs(),
 690                 oldPrefs.getHoldTime(),
 691                 oldPrefs.getBgpId(),
 692                 oldPrefs.getExpectedRemoteAs(),
 693                 Collections.singletonList(bgpParameters),
 694                 oldPrefs.getMd5Password());
 695         this.rib.getDispatcher().getBGPPeerRegistry()
 696                 .updatePeerPreferences(getNeighborAddress(), newPrefs);
 697     }
 698 }