Bug 5558 - Closing session after a large RIB is sent kills BGPCEP
[bgpcep.git] / bgp / rib-impl / src / main / java / org / opendaylight / protocol / bgp / rib / impl / AbstractBGPSessionNegotiator.java
index 96bf3af781ea82f5765074798f506555d0ba4c3b..4f00c9af2ef3b2747af82e77eae545884998e540 100644 (file)
@@ -22,11 +22,9 @@ import org.opendaylight.protocol.bgp.parser.BGPDocumentedException;
 import org.opendaylight.protocol.bgp.parser.BGPError;
 import org.opendaylight.protocol.bgp.rib.impl.spi.BGPPeerRegistry;
 import org.opendaylight.protocol.bgp.rib.impl.spi.BGPSessionPreferences;
-import org.opendaylight.protocol.bgp.rib.impl.spi.BGPSessionValidator;
 import org.opendaylight.protocol.bgp.rib.spi.BGPSessionListener;
 import org.opendaylight.protocol.bgp.rib.spi.SessionNegotiator;
 import org.opendaylight.protocol.util.Values;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.AsNumber;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.IpAddress;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.Ipv4Address;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.message.rev130919.Keepalive;
@@ -75,7 +73,6 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
 
     private static final Logger LOG = LoggerFactory.getLogger(AbstractBGPSessionNegotiator.class);
     private final BGPPeerRegistry registry;
-    private final BGPSessionValidator sessionValidator;
     private final Promise<BGPSessionImpl> promise;
     private final Channel channel;
     @GuardedBy("this")
@@ -85,34 +82,33 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
     private BGPSessionImpl session;
 
     public AbstractBGPSessionNegotiator(final Promise<BGPSessionImpl> promise, final Channel channel,
-            final BGPPeerRegistry registry, final BGPSessionValidator sessionValidator) {
+            final BGPPeerRegistry registry) {
         this.promise = Preconditions.checkNotNull(promise);
         this.channel = Preconditions.checkNotNull(channel);
         this.registry = registry;
-        this.sessionValidator = sessionValidator;
     }
 
     private synchronized void startNegotiation() {
-        Preconditions.checkState(this.state == State.IDLE);
+        // Open can be sent first either from ODL (IDLE) or from peer (OPEN_CONFIRM)
+        Preconditions.checkState(this.state == State.IDLE || this.state == State.OPEN_CONFIRM);
+        final IpAddress remoteIp = getRemoteIp();
 
         // Check if peer is configured in registry before retrieving preferences
-        if (!this.registry.isPeerConfigured(getRemoteIp())) {
+        if (!this.registry.isPeerConfigured(remoteIp)) {
             final BGPDocumentedException cause = new BGPDocumentedException(
-                    "BGP peer with ip: " + getRemoteIp()
-                    + " not configured, check configured peers in : "
-                    + this.registry, BGPError.CONNECTION_REJECTED);
+                String.format("BGP peer with ip: %s not configured, check configured peers in : %s", remoteIp, this.registry), BGPError.CONNECTION_REJECTED);
             negotiationFailed(cause);
             return;
         }
 
-        final BGPSessionPreferences preferences = getPreferences();
+        final BGPSessionPreferences preferences = this.registry.getPeerPreferences(remoteIp);
 
         int as = preferences.getMyAs().getValue().intValue();
         // Set as AS_TRANS if the value is bigger than 2B
         if (as > Values.UNSIGNED_SHORT_MAX_VALUE) {
             as = AS_TRANS;
         }
-        this.sendMessage(new OpenBuilder().setMyAsNumber(as).setHoldTimer(preferences.getHoldTime()).setBgpIdentifier(
+        sendMessage(new OpenBuilder().setMyAsNumber(as).setHoldTimer(preferences.getHoldTime()).setBgpIdentifier(
                 preferences.getBgpId()).setBgpParameters(preferences.getParams()).build());
         if (this.state != State.FINISHED) {
             this.state = State.OPEN_SENT;
@@ -130,10 +126,6 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
         }
     }
 
-    private BGPSessionPreferences getPreferences() {
-        return this.registry.getPeerPreferences(getRemoteIp());
-    }
-
     private IpAddress getRemoteIp() {
         return StrictBGPPeerRegistry.getIpAddress(this.channel.remoteAddress());
     }
@@ -143,8 +135,15 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
 
         switch (this.state) {
         case FINISHED:
+            sendMessage(buildErrorNotify(BGPError.FSM_ERROR));
+            return;
         case IDLE:
-            this.sendMessage(buildErrorNotify(BGPError.FSM_ERROR));
+            // to avoid race condition when Open message was sent by the peer before startNegotiation could be executed
+            if (msg instanceof Open) {
+                handleOpen((Open) msg);
+                return;
+            }
+            sendMessage(buildErrorNotify(BGPError.FSM_ERROR));
             return;
         case OPEN_CONFIRM:
             if (msg instanceof Keepalive) {
@@ -158,8 +157,7 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
             return;
         case OPEN_SENT:
             if (msg instanceof Open) {
-                final Open openObj = (Open) msg;
-                handleOpen(openObj);
+                handleOpen((Open) msg);
                 return;
             }
             break;
@@ -169,7 +167,7 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
 
         // Catch-all for unexpected message
         LOG.warn("Channel {} state {} unexpected message {}", this.channel, this.state, msg);
-        this.sendMessage(buildErrorNotify(BGPError.FSM_ERROR));
+        sendMessage(buildErrorNotify(BGPError.FSM_ERROR));
         negotiationFailed(new BGPDocumentedException("Unexpected message", BGPError.FSM_ERROR));
         this.state = State.FINISHED;
     }
@@ -187,18 +185,12 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
     }
 
     private void handleOpen(final Open openObj) {
+        final IpAddress remoteIp = getRemoteIp();
+        final BGPSessionPreferences preferences = this.registry.getPeerPreferences(remoteIp);
         try {
-            this.sessionValidator.validate(openObj, getPreferences());
-        } catch (final BGPDocumentedException e) {
-            negotiationFailed(e);
-            return;
-        }
-
-        try {
-            final BGPSessionListener peer = this.registry.getPeer(getRemoteIp(), getSourceId(openObj, getPreferences()),
-                    getDestinationId(openObj, getPreferences()), getAsNumber(openObj, getPreferences()));
-            this.sendMessage(new KeepaliveBuilder().build());
-            this.session = new BGPSessionImpl(peer, this.channel, openObj, getPreferences(), this.registry);
+            final BGPSessionListener peer = this.registry.getPeer(remoteIp, getSourceId(openObj, preferences), getDestinationId(openObj, preferences), openObj);
+            sendMessage(new KeepaliveBuilder().build());
+            this.session = new BGPSessionImpl(peer, this.channel, openObj, preferences, this.registry);
             this.state = State.OPEN_CONFIRM;
             LOG.debug("Channel {} moved to OpenConfirm state with remote proposal {}", this.channel, openObj);
         } catch (final BGPDocumentedException e) {
@@ -212,9 +204,11 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
         if (e instanceof BGPDocumentedException) {
             // although sendMessage() can also result in calling this method, it won't create a cycle. In case sendMessage() fails to
             // deliver the message, this method gets called with different exception (definitely not with BGPDocumentedException).
-            this.sendMessage(buildErrorNotify(((BGPDocumentedException)e).getError(), ((BGPDocumentedException) e).getData()));
+            sendMessage(buildErrorNotify(((BGPDocumentedException)e).getError(), ((BGPDocumentedException) e).getData()));
+        }
+        if (this.state == State.OPEN_CONFIRM) {
+            this.registry.removePeerSession(getRemoteIp());
         }
-        this.registry.removePeerSession(getRemoteIp());
         negotiationFailedCloseChannel(e);
         this.state = State.FINISHED;
     }
@@ -233,35 +227,28 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
      */
     protected abstract Ipv4Address getSourceId(final Open openMsg, final BGPSessionPreferences preferences);
 
-    /**
-     * @param openMsg Open message received from remote BGP speaker
-     * @param preferences Local BGP speaker preferences
-     * @return AS Number of device that initiate connection
-     */
-    protected abstract AsNumber getAsNumber(final Open openMsg, final BGPSessionPreferences preferences);
-
     public synchronized State getState() {
         return this.state;
     }
 
-    private final void negotiationSuccessful(BGPSessionImpl session) {
+    private void negotiationSuccessful(final BGPSessionImpl session) {
         LOG.debug("Negotiation on channel {} successful with session {}", this.channel, session);
-        channel.pipeline().replace(this, "session", session);
-        promise.setSuccess(session);
+        this.channel.pipeline().replace(this, "session", session);
+        this.promise.setSuccess(session);
     }
 
-    private void negotiationFailedCloseChannel(Throwable cause) {
+    private void negotiationFailedCloseChannel(final Throwable cause) {
         LOG.debug("Negotiation on channel {} failed", this.channel, cause);
-        channel.close();
-        promise.setFailure(cause);
+        this.channel.close();
+        this.promise.setFailure(cause);
     }
 
-    private final void sendMessage(final Notification msg) {
-        channel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
+    private void sendMessage(final Notification msg) {
+        this.channel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
             @Override
-            public void operationComplete(ChannelFuture f) {
+            public void operationComplete(final ChannelFuture f) {
                 if (!f.isSuccess()) {
-                    LOG.info("Failed to send message {}", msg, f.cause());
+                    LOG.warn("Failed to send message {}", msg, f.cause());
                     negotiationFailedCloseChannel(f.cause());
                 } else {
                     LOG.trace("Message {} sent to socket", msg);
@@ -272,11 +259,11 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
     }
 
     @Override
-    public final void channelActive(ChannelHandlerContext ctx) {
+    public final void channelActive(final ChannelHandlerContext ctx) {
         LOG.debug("Starting session negotiation on channel {}", this.channel);
 
         try {
-            this.startNegotiation();
+            startNegotiation();
         } catch (final Exception e) {
             LOG.warn("Unexpected negotiation failure", e);
             negotiationFailedCloseChannel(e);
@@ -285,12 +272,12 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
     }
 
     @Override
-    public final void channelRead(ChannelHandlerContext ctx, Object msg) {
+    public final void channelRead(final ChannelHandlerContext ctx, final Object msg) {
         LOG.debug("Negotiation read invoked on channel {}", this.channel);
 
         try {
             handleMessage((Notification) msg);
-        } catch (Exception e) {
+        } catch (final Exception e) {
             LOG.debug("Unexpected error while handling negotiation message {}", msg, e);
             negotiationFailedCloseChannel(e);
         }
@@ -298,7 +285,7 @@ public abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandler
     }
 
     @Override
-    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+    public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
         LOG.info("Unexpected error during negotiation", cause);
         negotiationFailedCloseChannel(cause);
     }