pcep/topology/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 static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.collect.Iterables;
  13 import com.google.common.collect.Maps;
  14 import com.google.common.util.concurrent.FluentFuture;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import com.google.common.util.concurrent.MoreExecutors;
  18 import io.netty.util.concurrent.FutureListener;
  19 import java.net.InetAddress;
  20 import java.util.ArrayList;
  21 import java.util.Collection;
  22 import java.util.HashMap;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Map.Entry;
  26 import java.util.Objects;
  27 import java.util.Optional;
  28 import java.util.Timer;
  29 import java.util.TimerTask;
  30 import java.util.concurrent.ConcurrentHashMap;
  31 import java.util.concurrent.TimeUnit;
  32 import java.util.concurrent.atomic.AtomicBoolean;
  33 import org.checkerframework.checker.lock.qual.GuardedBy;
  34 import org.checkerframework.checker.lock.qual.Holding;
  35 import org.eclipse.jdt.annotation.NonNull;
  36 import org.eclipse.jdt.annotation.Nullable;
  37 import org.opendaylight.bgpcep.pcep.topology.provider.session.stats.SessionStateImpl;
  38 import org.opendaylight.bgpcep.pcep.topology.provider.session.stats.TopologySessionStats;
  39 import org.opendaylight.mdsal.binding.api.WriteTransaction;
  40 import org.opendaylight.mdsal.common.api.CommitInfo;
  41 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  42 import org.opendaylight.protocol.pcep.PCEPCloseTermination;
  43 import org.opendaylight.protocol.pcep.PCEPSession;
  44 import org.opendaylight.protocol.pcep.PCEPTerminationReason;
  45 import org.opendaylight.protocol.pcep.TerminationReason;
  46 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IetfInetUtil;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.initiated.rev200720.Stateful1;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.LspObject;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.Path1;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.PlspId;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.SrpIdNumber;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.StatefulTlv1Builder;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.Tlvs1;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.lsp.object.Lsp;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.stateful.capability.tlv.Stateful;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.Message;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.Object;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.open.object.open.Tlvs;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.LspId;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.Node1;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.Node1Builder;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.OperationResult;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.PccSyncState;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TearDownSessionInput;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.lsp.metadata.Metadata;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.PathComputationClient;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.PathComputationClientBuilder;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLsp;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLspBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLspKey;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.StatefulTlvBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.reported.lsp.Path;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.reported.lsp.PathKey;
  74 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  75 import org.opendaylight.yangtools.yang.binding.DataObject;
  76 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  77 import org.opendaylight.yangtools.yang.common.RpcResult;
  78 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  79 import org.slf4j.Logger;
  80 import org.slf4j.LoggerFactory;
  81
  82 /**
  83  * Base class for PCEP topology providers. It handles the common tasks involved in managing a PCEP server (PCE)
  84  * endpoint, and exposing a network topology based on it. It needs to be subclassed to form a fully functional block,
  85  * where the subclass provides handling of incoming messages.
  86  */
  87 public abstract class AbstractTopologySessionListener implements TopologySessionListener, TopologySessionStats {
  88     private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologySessionListener.class);
  89
  90     private final AtomicBoolean statefulCapability = new AtomicBoolean(false);
  91     private final AtomicBoolean lspUpdateCapability = new AtomicBoolean(false);
  92     private final AtomicBoolean initiationCapability = new AtomicBoolean(false);
  93
  94     @GuardedBy("this")
  95     final Map<PlspId, String> lsps = new HashMap<>();
  96     @GuardedBy("this")
  97     SessionStateImpl listenerState;
  98
  99     @GuardedBy("this")
 100     private final Map<SrpIdNumber, PCEPRequest> requests = new HashMap<>();
 101     @GuardedBy("this")
 102     private final Map<String, ReportedLsp> lspData = new ConcurrentHashMap<>();
 103     private final ServerSessionManager serverSessionManager;
 104     private InstanceIdentifier<PathComputationClient> pccIdentifier;
 105     @GuardedBy("this")
 106     private TopologyNodeState nodeState;
 107     private final AtomicBoolean synced = new AtomicBoolean(false);
 108     @GuardedBy("this")
 109     private PCEPSession session;
 110     @GuardedBy("this")
 111     private SyncOptimization syncOptimization;
 112     @GuardedBy("this")
 113     private boolean triggeredResyncInProcess;
 114
 115     AbstractTopologySessionListener(final ServerSessionManager serverSessionManager) {
 116         this.serverSessionManager = requireNonNull(serverSessionManager);
 117     }
 118
 119     @Override
 120     public final void onSessionUp(final PCEPSession psession) {
 121         synchronized (serverSessionManager) {
 122             synchronized (this) {
 123                 /*
 124                  * The session went up. Look up the router in Inventory model, create it if it
 125                  * is not there (marking that fact for later deletion), and mark it as
 126                  * synchronizing. Also create it in the topology model, with empty LSP list.
 127                  */
 128                 final InetAddress peerAddress = psession.getRemoteAddress();
 129
 130                 syncOptimization = new SyncOptimization(psession);
 131                 final boolean haveLspDbVersion = syncOptimization.isDbVersionPresent();
 132
 133                 final TopologyNodeState state =
 134                         serverSessionManager.takeNodeState(peerAddress, this, haveLspDbVersion);
 135
 136                 // takeNodeState(..) may fail when the server session manager is being restarted
 137                 // due to configuration change
 138                 if (state == null) {
 139                     LOG.error("Unable to fetch topology node state for PCEP session. Closing session {}", psession);
 140                     psession.close(TerminationReason.UNKNOWN);
 141                     onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN));
 142                     return;
 143                 }
 144
 145                 if (session != null || nodeState != null) {
 146                     LOG.error("PCEP session is already up with {}. Closing session {}", peerAddress, psession);
 147                     psession.close(TerminationReason.UNKNOWN);
 148                     onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN));
 149                     return;
 150                 }
 151                 session = psession;
 152                 nodeState = state;
 153
 154                 LOG.trace("Peer {} resolved to topology node {}", peerAddress, state.getNodeId());
 155
 156                 // Our augmentation in the topology node
 157                 final PathComputationClientBuilder pccBuilder = new PathComputationClientBuilder()
 158                     .setIpAddress(IetfInetUtil.INSTANCE.ipAddressNoZoneFor(peerAddress));
 159
 160                 // Let subclass fill the details
 161                 updateStatefulCapabilities(pccBuilder, peerAddress, psession.getRemoteTlvs());
 162
 163                 synced.set(isSynchronized());
 164
 165                 final InstanceIdentifier<Node1> topologyAugment = state.getNodeId().augmentation(Node1.class);
 166                 pccIdentifier = topologyAugment.child(PathComputationClient.class);
 167
 168                 if (haveLspDbVersion) {
 169                     final Node initialNodeState = state.getInitialNodeState();
 170                     if (initialNodeState != null) {
 171                         loadLspData(initialNodeState, lspData, lsps, isIncrementalSynchro());
 172                         pccBuilder.setReportedLsp(
 173                             initialNodeState.augmentation(Node1.class).getPathComputationClient().getReportedLsp());
 174                     }
 175                 }
 176                 state.storeNode(topologyAugment,
 177                         new Node1Builder().setPathComputationClient(pccBuilder.build()).build(), psession);
 178
 179                 listenerState = new SessionStateImpl(this, psession);
 180                 serverSessionManager.bind(state.getNodeId(), listenerState);
 181                 LOG.info("Session with {} attached to topology node {}", peerAddress, state.getNodeId());
 182             }
 183         }
 184     }
 185
 186     @Holding("this")
 187     private void updateStatefulCapabilities(final PathComputationClientBuilder pccBuilder,
 188             final InetAddress peerAddress, final @Nullable Tlvs remoteTlvs) {
 189         if (remoteTlvs != null) {
 190             final Tlvs1 statefulTlvs = remoteTlvs.augmentation(Tlvs1.class);
 191             if (statefulTlvs != null) {
 192                 final Stateful stateful = statefulTlvs.getStateful();
 193                 if (stateful != null) {
 194                     statefulCapability.set(true);
 195                     final var updateCap = stateful.getLspUpdateCapability();
 196                     if (updateCap != null) {
 197                         lspUpdateCapability.set(updateCap);
 198                     }
 199                     final Stateful1 stateful1 = stateful.augmentation(Stateful1.class);
 200                     if (stateful1 != null) {
 201                         final var initiation = stateful1.getInitiation();
 202                         if (initiation != null) {
 203                             initiationCapability.set(initiation);
 204                         }
 205                     }
 206
 207                     pccBuilder.setReportedLsp(Map.of());
 208                     if (isSynchronized()) {
 209                         pccBuilder.setStateSync(PccSyncState.Synchronized);
 210                     } else if (isTriggeredInitialSynchro()) {
 211                         pccBuilder.setStateSync(PccSyncState.TriggeredInitialSync);
 212                     } else if (isIncrementalSynchro()) {
 213                         pccBuilder.setStateSync(PccSyncState.IncrementalSync);
 214                     } else {
 215                         pccBuilder.setStateSync(PccSyncState.InitialResync);
 216                     }
 217                     pccBuilder.setStatefulTlv(new StatefulTlvBuilder()
 218                         .addAugmentation(new StatefulTlv1Builder(statefulTlvs).build())
 219                         .build());
 220                     return;
 221                 }
 222             }
 223         }
 224         LOG.debug("Peer {} does not advertise stateful TLV", peerAddress);
 225     }
 226
 227     synchronized void updatePccState(final PccSyncState pccSyncState) {
 228         if (nodeState == null) {
 229             LOG.info("Server Session Manager is closed.");
 230             session.close(TerminationReason.UNKNOWN);
 231             return;
 232         }
 233         final MessageContext ctx = new MessageContext(nodeState.getChain().newWriteOnlyTransaction());
 234         updatePccNode(ctx, new PathComputationClientBuilder().setStateSync(pccSyncState).build());
 235         if (pccSyncState != PccSyncState.Synchronized) {
 236             synced.set(false);
 237             triggeredResyncInProcess = true;
 238         }
 239         // All set, commit the modifications
 240         ctx.trans.commit().addCallback(new FutureCallback<CommitInfo>() {
 241             @Override
 242             public void onSuccess(final CommitInfo result) {
 243                 LOG.trace("Pcc Internal state for session {} updated successfully", session);
 244             }
 245
 246             @Override
 247             public void onFailure(final Throwable cause) {
 248                 LOG.error("Failed to update Pcc internal state for session {}", session, cause);
 249                 session.close(TerminationReason.UNKNOWN);
 250             }
 251         }, MoreExecutors.directExecutor());
 252     }
 253
 254     synchronized boolean isTriggeredSyncInProcess() {
 255         return triggeredResyncInProcess;
 256     }
 257
 258     /**
 259      * Tear down the given PCEP session. It's OK to call this method even after the session
 260      * is already down. It always clear up the current session status.
 261      */
 262     @SuppressWarnings("checkstyle:IllegalCatch")
 263     private void tearDown(final PCEPSession psession) {
 264         requireNonNull(psession);
 265         synchronized (serverSessionManager) {
 266             synchronized (this) {
 267                 serverSessionManager.releaseNodeState(nodeState, psession.getRemoteAddress(), isLspDbPersisted());
 268                 clearNodeState();
 269
 270                 try {
 271                     if (session != null) {
 272                         session.close();
 273                     }
 274                     psession.close();
 275                 } catch (final Exception e) {
 276                     LOG.error("Session {} cannot be closed.", psession, e);
 277                 }
 278                 session = null;
 279                 listenerState = null;
 280                 syncOptimization = null;
 281
 282                 // Clear all requests we know about
 283                 for (final Entry<SrpIdNumber, PCEPRequest> e : requests.entrySet()) {
 284                     // FIXME: exhaustive when we have JDK17+
 285                     switch (e.getValue().cancel()) {
 286                         case DONE:
 287                             // Done is done, nothing to do
 288                             LOG.trace("Request {} was done when session went down.", e.getKey());
 289                             break;
 290                         case UNACKED:
 291                             // Peer has not acked: results in failure
 292                             LOG.info("Request {} was incomplete when session went down, failing the instruction",
 293                                     e.getKey());
 294                             break;
 295                         case UNSENT:
 296                             // Peer has not been sent to the peer: results in cancellation
 297                             LOG.debug("Request {} was not sent when session went down, cancelling the instruction",
 298                                     e.getKey());
 299                             break;
 300                         default:
 301                             break;
 302                     }
 303                 }
 304                 requests.clear();
 305             }
 306         }
 307     }
 308
 309     @Override
 310     public final void onSessionDown(final PCEPSession psession, final Exception exception) {
 311         LOG.warn("Session {} went down unexpectedly", psession, exception);
 312         tearDown(psession);
 313     }
 314
 315     @Override
 316     public final void onSessionTerminated(final PCEPSession psession, final PCEPTerminationReason reason) {
 317         LOG.info("Session {} terminated by peer with reason {}", psession, reason);
 318         tearDown(psession);
 319     }
 320
 321     @Override
 322     public final synchronized void onMessage(final PCEPSession psession, final Message message) {
 323         if (nodeState == null) {
 324             LOG.warn("Topology node state is null. Unhandled message {} on session {}", message, psession);
 325             psession.close(TerminationReason.UNKNOWN);
 326             return;
 327         }
 328         final MessageContext ctx = new MessageContext(nodeState.getChain().newWriteOnlyTransaction());
 329
 330         if (onMessage(ctx, message)) {
 331             LOG.warn("Unhandled message {} on session {}", message, psession);
 332             //cancel not supported, submit empty transaction
 333             ctx.trans.commit().addCallback(new FutureCallback<CommitInfo>() {
 334                 @Override
 335                 public void onSuccess(final CommitInfo result) {
 336                     LOG.trace("Successful commit");
 337                 }
 338
 339                 @Override
 340                 public void onFailure(final Throwable trw) {
 341                     LOG.error("Failed commit", trw);
 342                 }
 343             }, MoreExecutors.directExecutor());
 344             return;
 345         }
 346
 347         ctx.trans.commit().addCallback(new FutureCallback<CommitInfo>() {
 348             @Override
 349             public void onSuccess(final CommitInfo result) {
 350                 LOG.trace("Internal state for session {} updated successfully", psession);
 351                 ctx.notifyRequests();
 352
 353             }
 354
 355             @Override
 356             public void onFailure(final Throwable throwable) {
 357                 LOG.error("Failed to update internal state for session {}, closing it", psession, throwable);
 358                 ctx.notifyRequests();
 359                 psession.close(TerminationReason.UNKNOWN);
 360             }
 361         }, MoreExecutors.directExecutor());
 362     }
 363
 364     /**
 365      * Perform revision-specific message processing when a message arrives.
 366      *
 367      * @param ctx     Message processing context
 368      * @param message Protocol message
 369      * @return True if the message type is not handle.
 370      */
 371     protected abstract boolean onMessage(MessageContext ctx, Message message);
 372
 373     @Override
 374     public void close() {
 375         synchronized (serverSessionManager) {
 376             synchronized (this) {
 377                 clearNodeState();
 378                 if (session != null) {
 379                     LOG.info("Closing session {}", session);
 380                     session.close(TerminationReason.UNKNOWN);
 381                 }
 382             }
 383         }
 384     }
 385
 386     @Holding({"this.serverSessionManager", "this"})
 387     private void clearNodeState() {
 388         if (nodeState != null) {
 389             serverSessionManager.unbind(nodeState.getNodeId());
 390             LOG.debug("Clear Node state: {}", nodeState.getNodeId());
 391             nodeState = null;
 392         }
 393     }
 394
 395     final synchronized PCEPRequest removeRequest(final SrpIdNumber id) {
 396         final PCEPRequest ret = requests.remove(id);
 397         if (ret != null && listenerState != null) {
 398             listenerState.processRequestStats(ret.getElapsedMillis());
 399         }
 400         LOG.trace("Removed request {} object {}", id, ret);
 401         return ret;
 402     }
 403
 404     final synchronized ListenableFuture<OperationResult> sendMessage(final Message message, final SrpIdNumber requestId,
 405             final Metadata metadata) {
 406         final var sendFuture = session.sendMessage(message);
 407         listenerState.updateStatefulSentMsg(message);
 408         final PCEPRequest req = new PCEPRequest(metadata);
 409         requests.put(requestId, req);
 410         final short rpcTimeout = serverSessionManager.getRpcTimeout();
 411         LOG.trace("RPC response timeout value is {} seconds", rpcTimeout);
 412         if (rpcTimeout > 0) {
 413             setupTimeoutHandler(requestId, req, rpcTimeout);
 414         }
 415
 416         sendFuture.addListener((FutureListener<Void>) future -> {
 417             if (!future.isSuccess()) {
 418                 synchronized (AbstractTopologySessionListener.this) {
 419                     requests.remove(requestId);
 420                 }
 421                 req.cancel();
 422                 LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause());
 423             } else {
 424                 req.markUnacked();
 425                 LOG.trace("Request {} sent to peer (object {})", requestId, req);
 426             }
 427         });
 428
 429         return req.getFuture();
 430     }
 431
 432     private void setupTimeoutHandler(final SrpIdNumber requestId, final PCEPRequest req, final short timeout) {
 433         final Timer timer = req.getTimer();
 434         timer.schedule(new TimerTask() {
 435             @Override
 436             public void run() {
 437                 synchronized (AbstractTopologySessionListener.this) {
 438                     requests.remove(requestId);
 439                 }
 440                 req.cancel();
 441                 LOG.info("Request {} timed-out waiting for response", requestId);
 442             }
 443         }, TimeUnit.SECONDS.toMillis(timeout));
 444         LOG.trace("Set up response timeout handler for request {}", requestId);
 445     }
 446
 447     /**
 448      * Update an LSP in the data store.
 449      *
 450      * @param ctx       Message context
 451      * @param id        Revision-specific LSP identifier
 452      * @param lspName   LSP name
 453      * @param rlb       Reported LSP builder
 454      * @param solicited True if the update was solicited
 455      * @param remove    True if this is an LSP path removal
 456      */
 457     protected final synchronized void updateLsp(final MessageContext ctx, final PlspId id, final String lspName,
 458             final ReportedLspBuilder rlb, final boolean solicited, final boolean remove) {
 459
 460         final String name;
 461         if (lspName == null) {
 462             name = lsps.get(id);
 463             if (name == null) {
 464                 LOG.error("PLSPID {} seen for the first time, not reporting the LSP", id);
 465                 return;
 466             }
 467         } else {
 468             name = lspName;
 469         }
 470
 471         LOG.debug("Saved LSP {} with name {}", id, name);
 472         lsps.put(id, name);
 473
 474         final ReportedLsp previous = lspData.get(name);
 475         // if no previous report about the lsp exist, just proceed
 476         if (previous != null) {
 477             final Map<PathKey, Path> updatedPaths = makeBeforeBreak(rlb, previous, name, remove);
 478             // if all paths or the last path were deleted, delete whole tunnel
 479             if (updatedPaths.isEmpty()) {
 480                 LOG.debug("All paths were removed, removing LSP with {}.", id);
 481                 removeLsp(ctx, id);
 482                 return;
 483             }
 484             rlb.setPath(updatedPaths);
 485         }
 486         rlb.withKey(new ReportedLspKey(name));
 487         rlb.setName(name);
 488
 489         // If this is an unsolicited update. We need to make sure we retain the metadata already present
 490         if (solicited) {
 491             nodeState.setLspMetadata(name, rlb.getMetadata());
 492         } else {
 493             rlb.setMetadata(nodeState.getLspMetadata(name));
 494         }
 495
 496         final ReportedLsp rl = rlb.build();
 497         ctx.trans.put(LogicalDatastoreType.OPERATIONAL, pccIdentifier.child(ReportedLsp.class, rlb.key()), rl);
 498         LOG.debug("LSP {} updated to MD-SAL", name);
 499
 500         lspData.put(name, rl);
 501     }
 502
 503     private static Map<PathKey, Path> makeBeforeBreak(final ReportedLspBuilder rlb, final ReportedLsp previous,
 504             final String name, final boolean remove) {
 505         // just one path should be reported
 506         final Path path = Iterables.getOnlyElement(rlb.getPath().values());
 507         final var reportedLspId = path.getLspId();
 508         final List<Path> updatedPaths;
 509         //lspId = 0 and remove = false -> tunnel is down, still exists but no path is signaled
 510         //remove existing tunnel's paths now, as explicit path remove will not come
 511         if (!remove && reportedLspId.getValue().toJava() == 0) {
 512             updatedPaths = new ArrayList<>();
 513             LOG.debug("Remove previous paths {} to this lsp name {}", previous.getPath(), name);
 514         } else {
 515             // check previous report for existing paths
 516             final Collection<Path> prev = previous.nonnullPath().values();
 517             updatedPaths = new ArrayList<>(prev);
 518             LOG.debug("Found previous paths {} to this lsp name {}", updatedPaths, name);
 519             for (final Path prevPath : prev) {
 520                 //we found reported path in previous reports
 521                 if (prevPath.getLspId().getValue().toJava() == 0 || prevPath.getLspId().equals(reportedLspId)) {
 522                     LOG.debug("Match on lsp-id {}", prevPath.getLspId().getValue());
 523                     // path that was reported previously and does have the same lsp-id, path will be updated
 524                     final boolean r = updatedPaths.remove(prevPath);
 525                     LOG.trace("Request removed? {}", r);
 526                 }
 527             }
 528         }
 529         // if the path does not exist in previous report, add it to path list, it's a new ERO
 530         // only one path will be added
 531         //lspId is 0 means confirmation message that shouldn't be added (because we have no means of deleting it later)
 532         LOG.trace("Adding new path {} to {}", path, updatedPaths);
 533         updatedPaths.add(path);
 534         if (remove) {
 535             if (reportedLspId.getValue().toJava() == 0) {
 536                 // if lsp-id also 0, remove all paths
 537                 LOG.debug("Removing all paths.");
 538                 updatedPaths.clear();
 539             } else {
 540                 // path is marked to be removed
 541                 LOG.debug("Removing path {} from {}", path, updatedPaths);
 542                 final boolean r = updatedPaths.remove(path);
 543                 LOG.trace("Request removed? {}", r);
 544             }
 545         }
 546         LOG.debug("Setting new paths {} to lsp {}", updatedPaths, name);
 547         return Maps.uniqueIndex(updatedPaths, Path::key);
 548     }
 549
 550     /**
 551      * Indicate that the peer has completed state synchronization.
 552      *
 553      * @param ctx Message context
 554      */
 555     protected final synchronized void stateSynchronizationAchieved(final MessageContext ctx) {
 556         if (synced.getAndSet(true)) {
 557             LOG.debug("State synchronization achieved while synchronizing, not updating state");
 558             return;
 559         }
 560
 561         triggeredResyncInProcess = false;
 562         updatePccNode(ctx, new PathComputationClientBuilder().setStateSync(PccSyncState.Synchronized).build());
 563
 564         // The node has completed synchronization, cleanup metadata no longer reported back
 565         nodeState.cleanupExcept(lsps.values());
 566         LOG.debug("Session {} achieved synchronized state", session);
 567     }
 568
 569     protected final synchronized void updatePccNode(final MessageContext ctx, final PathComputationClient pcc) {
 570         ctx.trans.merge(LogicalDatastoreType.OPERATIONAL, pccIdentifier, pcc);
 571     }
 572
 573     protected final @NonNull InstanceIdentifier<ReportedLsp> lspIdentifier(final String name) {
 574         return pccIdentifier.child(ReportedLsp.class, new ReportedLspKey(name));
 575     }
 576
 577     /**
 578      * Remove LSP from the database.
 579      *
 580      * @param ctx Message Context
 581      * @param id  Revision-specific LSP identifier
 582      */
 583     protected final synchronized void removeLsp(final MessageContext ctx, final PlspId id) {
 584         final String name = lsps.remove(id);
 585         LOG.debug("LSP {} removed", name);
 586         ctx.trans.delete(LogicalDatastoreType.OPERATIONAL, lspIdentifier(name));
 587         lspData.remove(name);
 588     }
 589
 590     @Holding("this")
 591     final String lookupLspName(final PlspId id) {
 592         return lsps.get(requireNonNull(id, "ID parameter null."));
 593     }
 594
 595     /**
 596      * Reads operational data on this node. Doesn't attempt to read the data,
 597      * if the node does not exist. In this case returns null.
 598      *
 599      * @param id InstanceIdentifier of the node
 600      * @return null if the node does not exists, or operational data
 601      */
 602     final synchronized <T extends DataObject> FluentFuture<Optional<T>> readOperationalData(
 603             final InstanceIdentifier<T> id) {
 604         return nodeState == null ? null : nodeState.readOperationalData(id);
 605     }
 606
 607     protected abstract Object validateReportedLsp(Optional<ReportedLsp> rep, LspId input);
 608
 609     protected abstract void loadLspData(Node node, Map<String, ReportedLsp> lspData, Map<PlspId, String> lsps,
 610             boolean incrementalSynchro);
 611
 612     final boolean isLspDbPersisted() {
 613         return syncOptimization != null && syncOptimization.isSyncAvoidanceEnabled();
 614     }
 615
 616     /**
 617      * Is Incremental synchronization if LSP-DB-VERSION are included,
 618      * LSP-DB-VERSION TLV values doesnt match, and  LSP-SYNC-CAPABILITY is enabled.
 619      */
 620     final synchronized boolean isIncrementalSynchro() {
 621         return syncOptimization != null && syncOptimization.isSyncAvoidanceEnabled()
 622                 && syncOptimization.isDeltaSyncEnabled();
 623     }
 624
 625     final synchronized boolean isTriggeredInitialSynchro() {
 626         return syncOptimization != null && syncOptimization.isTriggeredInitSyncEnabled();
 627     }
 628
 629     final synchronized boolean isTriggeredReSyncEnabled() {
 630         return syncOptimization != null && syncOptimization.isTriggeredReSyncEnabled();
 631     }
 632
 633     protected final synchronized boolean isSynchronized() {
 634         return syncOptimization != null && syncOptimization.doesLspDbMatch();
 635     }
 636
 637     @Override
 638     public final int getDelegatedLspsCount() {
 639         return Math.toIntExact(lspData.values().stream()
 640             .map(ReportedLsp::getPath).filter(pathList -> pathList != null && !pathList.isEmpty())
 641             // pick the first path, as delegate status should be same in each path
 642             .map(pathList -> pathList.values().iterator().next())
 643             .map(path -> path.augmentation(Path1.class)).filter(Objects::nonNull)
 644             .map(LspObject::getLsp).filter(Objects::nonNull)
 645             .filter(Lsp::getDelegate)
 646             .count());
 647     }
 648
 649     @Override
 650     public final boolean isSessionSynchronized() {
 651         return synced.get();
 652     }
 653
 654     @Override
 655     public final boolean isInitiationCapability() {
 656         return initiationCapability.get();
 657     }
 658
 659     @Override
 660     public final boolean isStatefulCapability() {
 661         return statefulCapability.get();
 662     }
 663
 664     @Override
 665     public final boolean isLspUpdateCapability() {
 666         return lspUpdateCapability.get();
 667     }
 668
 669
 670     @Override
 671     public synchronized ListenableFuture<RpcResult<Void>> tearDownSession(final TearDownSessionInput input) {
 672         close();
 673         return RpcResultBuilder.<Void>success().buildFuture();
 674     }
 675
 676     static final class MessageContext {
 677         private final Collection<PCEPRequest> requests = new ArrayList<>();
 678         private final WriteTransaction trans;
 679
 680         private MessageContext(final WriteTransaction trans) {
 681             this.trans = requireNonNull(trans);
 682         }
 683
 684         void resolveRequest(final PCEPRequest req) {
 685             requests.add(req);
 686         }
 687
 688         private void notifyRequests() {
 689             for (final PCEPRequest r : requests) {
 690                 r.finish(OperationResults.SUCCESS);
 691             }
 692         }
 693     }
 694 }