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