pcep/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java

   1 /*
   2  * Copyright (c) 2013 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.bgpcep.pcep.topology.provider;
   9
  10 import com.google.common.base.Optional;
  11 import com.google.common.base.Preconditions;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.Futures;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15
  16 import io.netty.util.concurrent.FutureListener;
  17
  18 import java.net.InetAddress;
  19 import java.util.ArrayList;
  20 import java.util.Collection;
  21 import java.util.HashMap;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.Map.Entry;
  25 import java.util.concurrent.ExecutionException;
  26
  27 import javax.annotation.concurrent.GuardedBy;
  28
  29 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  30 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  31 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  32 import org.opendaylight.protocol.pcep.PCEPSession;
  33 import org.opendaylight.protocol.pcep.PCEPSessionListener;
  34 import org.opendaylight.protocol.pcep.PCEPTerminationReason;
  35 import org.opendaylight.protocol.pcep.TerminationReason;
  36 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.IpAddressBuilder;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.MessageHeader;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.ProtocolVersion;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.rsvp.rev130820.LspId;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1Builder;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.OperationResult;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.PccSyncState;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.lsp.metadata.Metadata;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClient;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClientBuilder;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLsp;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLspBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLspKey;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.reported.lsp.Path;
  52 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  53 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  54 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  55 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
  56 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  57 import org.opendaylight.yangtools.yang.binding.DataContainer;
  58 import org.opendaylight.yangtools.yang.binding.DataObject;
  59 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  60 import org.slf4j.Logger;
  61 import org.slf4j.LoggerFactory;
  62
  63 /**
  64  * Base class for PCEP topology providers. It handles the common tasks involved in managing a PCEP server (PCE)
  65  * endpoint, and exposing a network topology based on it. It needs to be subclassed to form a fully functional block,
  66  * where the subclass provides handling of incoming messages.
  67  *
  68  * @param <S> identifier type of requests
  69  * @param <L> identifier type for LSPs
  70  */
  71 public abstract class AbstractTopologySessionListener<S, L> implements PCEPSessionListener, TopologySessionListener {
  72     protected static final class MessageContext {
  73         private final Collection<PCEPRequest> requests = new ArrayList<>();
  74         private final WriteTransaction trans;
  75
  76         private MessageContext(final WriteTransaction trans) {
  77             this.trans = Preconditions.checkNotNull(trans);
  78         }
  79
  80         void resolveRequest(final PCEPRequest req) {
  81             requests.add(req);
  82         }
  83
  84         private void notifyRequests() {
  85             for (PCEPRequest r : requests) {
  86                 r.done(OperationResults.SUCCESS);
  87             }
  88         }
  89     }
  90
  91     protected static final MessageHeader MESSAGE_HEADER = new MessageHeader() {
  92         private final ProtocolVersion version = new ProtocolVersion((short) 1);
  93
  94         @Override
  95         public Class<? extends DataContainer> getImplementedInterface() {
  96             return MessageHeader.class;
  97         }
  98
  99         @Override
 100         public ProtocolVersion getVersion() {
 101             return this.version;
 102         }
 103     };
 104     private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologySessionListener.class);
 105
 106     private final Map<S, PCEPRequest> requests = new HashMap<>();
 107     private final Map<String, ReportedLsp> lspData = new HashMap<>();
 108     private final Map<L, String> lsps = new HashMap<>();
 109     private final ServerSessionManager serverSessionManager;
 110     private InstanceIdentifier<Node> topologyNode;
 111     private InstanceIdentifier<Node1> topologyAugment;
 112     private InstanceIdentifier<PathComputationClient> pccIdentifier;
 113     private TopologyNodeState nodeState;
 114     private boolean ownsTopology = false;
 115     private boolean synced = false;
 116     private PCEPSession session;
 117
 118     protected AbstractTopologySessionListener(final ServerSessionManager serverSessionManager) {
 119         this.serverSessionManager = Preconditions.checkNotNull(serverSessionManager);
 120     }
 121
 122     private static String createNodeId(final InetAddress addr) {
 123         return "pcc://" + addr.getHostAddress();
 124     }
 125
 126     private Node topologyNode(final ReadWriteTransaction trans, final InetAddress address) {
 127         final String pccId = createNodeId(address);
 128
 129         // FIXME: Futures.transform...
 130         try {
 131             Optional<Topology> topoMaybe = trans.read(LogicalDatastoreType.OPERATIONAL, this.serverSessionManager.getTopology()).get();
 132             Preconditions.checkState(topoMaybe.isPresent(), "Failed to find topology.");
 133             final Topology topo = topoMaybe.get();
 134             for (final Node n : topo.getNode()) {
 135                 LOG.debug("Matching topology node {} to id {}", n, pccId);
 136                 if (n.getNodeId().getValue().equals(pccId)) {
 137                     this.topologyNode = this.serverSessionManager.getTopology().child(Node.class, n.getKey());
 138                     LOG.debug("Reusing topology node {} for id {} at {}", n, pccId, this.topologyNode);
 139                     return n;
 140                 }
 141             }
 142         } catch (InterruptedException | ExecutionException e) {
 143             throw new IllegalStateException("Failed to ensure topology presence.", e);
 144         }
 145
 146         /*
 147          * We failed to find a matching node. Let's create a dynamic one
 148          * and note that we are the owner (so we clean it up afterwards).
 149          */
 150         final NodeId id = new NodeId(pccId);
 151         final NodeKey nk = new NodeKey(id);
 152         final InstanceIdentifier<Node> nti = this.serverSessionManager.getTopology().child(Node.class, nk);
 153
 154         final Node ret = new NodeBuilder().setKey(nk).setNodeId(id).build();
 155
 156         trans.put(LogicalDatastoreType.OPERATIONAL, nti, ret);
 157         LOG.debug("Created topology node {} for id {} at {}", ret, pccId, nti);
 158         this.ownsTopology = true;
 159         this.topologyNode = nti;
 160         return ret;
 161     }
 162
 163     @Override
 164     public final synchronized void onSessionUp(final PCEPSession session) {
 165         /*
 166          * The session went up. Look up the router in Inventory model,
 167          * create it if it is not there (marking that fact for later
 168          * deletion), and mark it as synchronizing. Also create it in
 169          * the topology model, with empty LSP list.
 170          */
 171         final InetAddress peerAddress = session.getRemoteAddress();
 172         final ReadWriteTransaction trans = this.serverSessionManager.rwTransaction();
 173
 174         final Node topoNode = topologyNode(trans, peerAddress);
 175         LOG.trace("Peer {} resolved to topology node {}", peerAddress, topoNode);
 176
 177         // Our augmentation in the topology node
 178         this.synced = false;
 179
 180         final PathComputationClientBuilder pccBuilder;
 181         pccBuilder = new PathComputationClientBuilder();
 182         pccBuilder.setIpAddress(IpAddressBuilder.getDefaultInstance(peerAddress.getHostAddress()));
 183
 184         onSessionUp(session, pccBuilder);
 185
 186         final Node1 ta = new Node1Builder().setPathComputationClient(pccBuilder.build()).build();
 187
 188         this.topologyAugment = this.topologyNode.augmentation(Node1.class);
 189         this.pccIdentifier = topologyAugment.child(PathComputationClient.class);
 190
 191         trans.put(LogicalDatastoreType.OPERATIONAL, this.topologyAugment, ta);
 192         LOG.trace("Peer data {} set to {}", this.topologyAugment, ta);
 193
 194         // All set, commit the modifications
 195         Futures.addCallback(trans.submit(), new FutureCallback<Void>() {
 196             @Override
 197             public void onSuccess(final Void result) {
 198                 LOG.trace("Internal state for session {} updated successfully", session);
 199             }
 200
 201             @Override
 202             public void onFailure(final Throwable t) {
 203                 LOG.error("Failed to update internal state for session {}, terminating it", session, t);
 204                 session.close(TerminationReason.Unknown);
 205             }
 206         });
 207
 208         this.nodeState = this.serverSessionManager.takeNodeState(topoNode.getNodeId(), this);
 209         this.session = session;
 210         LOG.info("Session with {} attached to topology node {}", session.getRemoteAddress(), topoNode.getNodeId());
 211     }
 212
 213     @GuardedBy("this")
 214     private void tearDown(final PCEPSession session) {
 215         this.serverSessionManager.releaseNodeState(this.nodeState);
 216         this.nodeState = null;
 217         this.session = null;
 218
 219         // The session went down. Undo all the Topology changes we have done.
 220         final WriteTransaction trans = this.serverSessionManager.beginTransaction();
 221         trans.delete(LogicalDatastoreType.OPERATIONAL, this.topologyAugment);
 222         if (this.ownsTopology) {
 223             trans.delete(LogicalDatastoreType.OPERATIONAL, this.topologyNode);
 224         }
 225
 226         Futures.addCallback(trans.submit(), new FutureCallback<Void>() {
 227             @Override
 228             public void onSuccess(final Void result) {
 229                 LOG.trace("Internal state for session {} cleaned up successfully", session);
 230             }
 231
 232             @Override
 233             public void onFailure(final Throwable t) {
 234                 LOG.error("Failed to cleanup internal state for session {}", session, t);
 235             }
 236         });
 237
 238         // Clear all requests we know about
 239         for (final Entry<S, PCEPRequest> e : this.requests.entrySet()) {
 240             final PCEPRequest r = e.getValue();
 241             switch (r.getState()) {
 242             case DONE:
 243                 // Done is done, nothing to do
 244                 break;
 245             case UNACKED:
 246                 // Peer has not acked: results in failure
 247                 LOG.info("Request {} was incomplete when session went down, failing the instruction", e.getKey());
 248                 r.done(OperationResults.NOACK);
 249                 break;
 250             case UNSENT:
 251                 // Peer has not been sent to the peer: results in cancellation
 252                 LOG.debug("Request {} was not sent when session went down, cancelling the instruction", e.getKey());
 253                 r.done(OperationResults.UNSENT);
 254                 break;
 255             }
 256         }
 257         this.requests.clear();
 258     }
 259
 260     @Override
 261     public final synchronized void onSessionDown(final PCEPSession session, final Exception e) {
 262         LOG.warn("Session {} went down unexpectedly", session, e);
 263         tearDown(session);
 264     }
 265
 266     @Override
 267     public final synchronized void onSessionTerminated(final PCEPSession session, final PCEPTerminationReason reason) {
 268         LOG.info("Session {} terminated by peer with reason {}", session, reason);
 269         tearDown(session);
 270     }
 271
 272     @Override
 273     public final synchronized void onMessage(final PCEPSession session, final Message message) {
 274         final MessageContext ctx = new MessageContext(this.serverSessionManager.beginTransaction());
 275
 276         if (onMessage(ctx, message)) {
 277             LOG.info("Unhandled message {} on session {}", message, session);
 278             return;
 279         }
 280
 281         Futures.addCallback(ctx.trans.submit(), new FutureCallback<Void>() {
 282             @Override
 283             public void onSuccess(final Void result) {
 284                 LOG.trace("Internal state for session {} updated successfully", session);
 285                 ctx.notifyRequests();
 286             }
 287
 288             @Override
 289             public void onFailure(final Throwable t) {
 290                 LOG.error("Failed to update internal state for session {}, closing it", session, t);
 291                 ctx.notifyRequests();
 292                 session.close(TerminationReason.Unknown);
 293             }
 294         });
 295     }
 296
 297     @Override
 298     public void close() {
 299         if (this.session != null) {
 300             this.session.close(TerminationReason.Unknown);
 301         }
 302     }
 303
 304     protected final synchronized PCEPRequest removeRequest(final S id) {
 305         final PCEPRequest ret = this.requests.remove(id);
 306         LOG.trace("Removed request {} object {}", id, ret);
 307         return ret;
 308     }
 309
 310     protected final synchronized ListenableFuture<OperationResult> sendMessage(final Message message, final S requestId,
 311             final Metadata metadata) {
 312         final io.netty.util.concurrent.Future<Void> f = this.session.sendMessage(message);
 313         final PCEPRequest req = new PCEPRequest(metadata);
 314         this.requests.put(requestId, req);
 315
 316         f.addListener(new FutureListener<Void>() {
 317             @Override
 318             public void operationComplete(final io.netty.util.concurrent.Future<Void> future) {
 319                 if (!future.isSuccess()) {
 320                     synchronized (AbstractTopologySessionListener.this) {
 321                         requests.remove(requestId);
 322                     }
 323                     req.done(OperationResults.UNSENT);
 324                     LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause());
 325                 } else {
 326                     req.sent();
 327                     LOG.trace("Request {} sent to peer (object {})", requestId, req);
 328                 }
 329             }
 330         });
 331
 332         return req.getFuture();
 333     }
 334
 335     /**
 336      * Update an LSP in the data store
 337      *
 338      * @param ctx Message context
 339      * @param id Revision-specific LSP identifier
 340      * @param lspName LSP name
 341      * @param rlb Reported LSP builder
 342      * @param solicited True if the update was solicited
 343      * @param remove True if this is an LSP path removal
 344      */
 345     protected final synchronized void updateLsp(final MessageContext ctx, final L id, final String lspName,
 346             final ReportedLspBuilder rlb, final boolean solicited, final boolean remove) {
 347
 348         final String name;
 349         if (lspName == null) {
 350             name = this.lsps.get(id);
 351             if (name == null) {
 352                 LOG.error("PLSPID {} seen for the first time, not reporting the LSP", id);
 353                 return;
 354             }
 355         } else {
 356             name = lspName;
 357         }
 358
 359         LOG.debug("Saved LSP {} with name {}", id, name);
 360         this.lsps.put(id, name);
 361
 362         // just one path should be reported
 363         Preconditions.checkState(rlb.getPath().size() == 1);
 364         LspId reportedLspId = rlb.getPath().get(0).getLspId();
 365         // check previous report for existing paths
 366         ReportedLsp previous = this.lspData.get(name);
 367         // if no previous report about the lsp exist, just proceed
 368         if (previous != null) {
 369             List<Path> updatedPaths = new ArrayList<>(previous.getPath());
 370             LOG.debug("Found previous paths {} to this lsp name {}", updatedPaths, name);
 371             for (Path path : previous.getPath()) {
 372                 //we found reported path in previous reports
 373                 if (path.getLspId().getValue() == 0 || path.getLspId().equals(reportedLspId)) {
 374                     LOG.debug("Match on lsp-id {}", path.getLspId().getValue() );
 375                     // path that was reported previously and does have the same lsp-id, path will be updated
 376                     final boolean r = updatedPaths.remove(path);
 377                     LOG.trace("Request removed? {}", r);
 378                 }
 379             }
 380             // if the path does not exist in previous report, add it to path list, it's a new ERO
 381             // only one path will be added
 382             //lspId is 0 means confirmation message that shouldn't be added (because we have no means of deleting it later)
 383             LOG.trace("Adding new path {} to {}", rlb.getPath(), updatedPaths);
 384             updatedPaths.addAll(rlb.getPath());
 385             if (remove) {
 386                 if (reportedLspId.getValue() == 0) {
 387                     // if lsp-id also 0, remove all paths
 388                     LOG.debug("Removing all paths.");
 389                     updatedPaths.clear();
 390                 } else {
 391                     // path is marked to be removed
 392                     LOG.debug("Removing path {} from {}", rlb.getPath(), updatedPaths);
 393                     final boolean r = updatedPaths.removeAll(rlb.getPath());
 394                     LOG.trace("Request removed? {}", r);
 395                 }
 396             }
 397             // if all paths or the last path were deleted, delete whole tunnel
 398             if (updatedPaths.isEmpty()) {
 399                 LOG.debug("All paths were removed, removing LSP with {}.", id);
 400                 removeLsp(ctx, id);
 401                 return;
 402             }
 403             LOG.debug("Setting new paths {} to lsp {}", updatedPaths, name);
 404             rlb.setPath(updatedPaths);
 405         }
 406         Preconditions.checkState(name != null);
 407         rlb.setKey(new ReportedLspKey(name));
 408         rlb.setName(name);
 409
 410         // If this is an unsolicited update. We need to make sure we retain the metadata already present
 411         if (solicited) {
 412             this.nodeState.setLspMetadata(name, rlb.getMetadata());
 413         } else {
 414             rlb.setMetadata(this.nodeState.getLspMetadata(name));
 415         }
 416
 417         final ReportedLsp rl = rlb.build();
 418         ctx.trans.put(LogicalDatastoreType.OPERATIONAL, this.pccIdentifier.child(ReportedLsp.class, rlb.getKey()), rl);
 419         LOG.debug("LSP {} updated to MD-SAL", name);
 420
 421         this.lspData.put(name, rl);
 422     }
 423
 424     /**
 425      * Indicate that the peer has completed state synchronization.
 426      *
 427      * @param ctx Message context
 428      */
 429     protected final synchronized void stateSynchronizationAchieved(final MessageContext ctx) {
 430         if (this.synced) {
 431             LOG.debug("State synchronization achieved while synchronized, not updating state");
 432             return;
 433         }
 434
 435         // Update synchronization flag
 436         this.synced = true;
 437         ctx.trans.merge(LogicalDatastoreType.OPERATIONAL, this.pccIdentifier, new PathComputationClientBuilder().setStateSync(PccSyncState.Synchronized).build());
 438
 439         // The node has completed synchronization, cleanup metadata no longer reported back
 440         this.nodeState.cleanupExcept(this.lsps.values());
 441         LOG.debug("Session {} achieved synchronized state", this.session);
 442     }
 443
 444     protected final InstanceIdentifier<ReportedLsp> lspIdentifier(final String name) {
 445         return pccIdentifier.child(ReportedLsp.class, new ReportedLspKey(name));
 446     }
 447
 448     /**
 449      * Remove LSP from the database.
 450      *
 451      * @param ctx Message Context
 452      * @param id Revision-specific LSP identifier
 453      */
 454     protected final synchronized void removeLsp(final MessageContext ctx, final L id) {
 455         final String name = this.lsps.remove(id);
 456         LOG.debug("LSP {} removed", name);
 457         ctx.trans.delete(LogicalDatastoreType.OPERATIONAL, lspIdentifier(name));
 458         this.lspData.remove(name);
 459     }
 460
 461     protected abstract void onSessionUp(PCEPSession session, PathComputationClientBuilder pccBuilder);
 462
 463     /**
 464      * Perform revision-specific message processing when a message arrives.
 465      *
 466      * @param ctx Message processing context
 467      * @param message Protocol message
 468      * @return True if the message type is not handle.
 469      */
 470     protected abstract boolean onMessage(MessageContext ctx, Message message);
 471
 472     protected final String lookupLspName(final L id) {
 473         Preconditions.checkNotNull(id, "ID parameter null.");
 474         return this.lsps.get(id);
 475     }
 476
 477     protected final <T extends DataObject> ListenableFuture<Optional<T>> readOperationalData(final InstanceIdentifier<T> id) {
 478         return this.serverSessionManager.readOperationalData(id);
 479     }
 480 }