Rework topology instantiation to split the individual concerns.
PCEPTopologyTracker is responsible for finding out when a particular
topology becomes interesting to the underlying code (and when becomes
uninteresting).
For each such topology, a PCEPTopologySingleton is instantiated. It then
handles allocation of the underlying InstructionScheduler as well as
bootstrapping a cluster-wide singleton PCEPTopologyInstance to handle
that topology's configuration.
PCEPTopologyInstance tracks configuration datastore changes for a single
topology. For each update it constructs an immutable
PCEPTopologyConfiguration and forwards it to its PCEPTopologyProvider.
PCEPTopologyProvider handles updates of the configuration in the most
graceful manner possible, sequencing possible Netty Channel restart, but
otherwise just updaing the low-level componennts
JIRA: BGPCEP-893
Change-Id: I56c47b89ab87cf33a6be1016fa132b6c1aa98cbb
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>
<groupId>io.netty</groupId>
<artifactId>netty-transport</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-transport-native-epoll</artifactId>
+ <classifier>linux-x86_64</classifier>
+ </dependency>
<dependency>
<groupId>org.checkerframework</groupId>
<artifactId>checker-qual</artifactId>
import java.net.InetSocketAddress;
import org.eclipse.jdt.annotation.NonNull;
-import org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyConfiguration;
import org.opendaylight.protocol.concepts.KeyMapping;
import org.opendaylight.protocol.pcep.PCEPDispatcherDependencies;
import org.opendaylight.protocol.pcep.PCEPPeerProposal;
final class PCEPStatefulPeerProposal implements PCEPPeerProposal {
private static final Logger LOG = LoggerFactory.getLogger(PCEPStatefulPeerProposal.class);
- private final DataBroker dataBroker;
private final InstanceIdentifier<Topology> topologyId;
- private final SpeakerIdMapping speakerIds;
+ private final DataBroker dataBroker;
+
+ private volatile SpeakerIdMapping speakerIds;
PCEPStatefulPeerProposal(final DataBroker dataBroker, final InstanceIdentifier<Topology> topologyId,
final SpeakerIdMapping speakerIds) {
this.dataBroker = requireNonNull(dataBroker);
this.topologyId = requireNonNull(topologyId);
- // FIXME: BGPCEP-989: once we have DTCL, we certainly should be able to maintain this mapping as well
+ // FIXME: BGPCEP-983: we should acquire this through DTCL
+ this.speakerIds = requireNonNull(speakerIds);
+ }
+
+ void setSpeakerIds(final SpeakerIdMapping speakerIds) {
this.speakerIds = requireNonNull(speakerIds);
}
if (speakerId == null && !result.isPresent()) {
return;
}
- final Tlvs3Builder syncBuilder = new Tlvs3Builder();
+ final Tlvs3Builder syncBuilder = new Tlvs3Builder();
if (result.isPresent()) {
syncBuilder.setLspDbVersion(result.get());
}
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.bgpcep.pcep.topology.provider.config;
+package org.opendaylight.bgpcep.pcep.topology.provider;
import static java.util.Objects.requireNonNull;
import java.util.Map;
import org.eclipse.jdt.annotation.NonNull;
import org.eclipse.jdt.annotation.Nullable;
-import org.opendaylight.bgpcep.pcep.topology.provider.SpeakerIdMapping;
import org.opendaylight.protocol.concepts.KeyMapping;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IetfInetUtil;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddressNoZone;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.PortNumber;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.config.rev200120.pcep.config.SessionConfig;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.config.rev181109.PcepNodeConfig;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.config.rev181109.PcepTopologyTypeConfig;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.sync.optimizations.config.rev181109.PcepNodeSyncConfig;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
import org.opendaylight.yangtools.concepts.Immutable;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
-public final class PCEPTopologyConfiguration implements Immutable {
+final class PCEPTopologyConfiguration implements Immutable {
+ private final @NonNull SpeakerIdMapping speakerIds;
private final @NonNull InetSocketAddress address;
private final @NonNull KeyMapping keys;
private final short rpcTimeout;
- private final @NonNull SpeakerIdMapping speakerIds;
- private final @NonNull KeyedInstanceIdentifier<Topology, TopologyKey> topology;
-
- public PCEPTopologyConfiguration(final @NonNull Topology topology) {
- this.topology = InstanceIdentifier.create(NetworkTopology.class).child(Topology.class, topology.key());
- final SessionConfig config = topology.augmentation(PcepTopologyTypeConfig.class).getSessionConfig();
- address = getInetSocketAddress(config.getListenAddress(), config.getListenPort());
- keys = constructKeys(topology.getNode());
- speakerIds = contructSpeakersId(topology.getNode());
- rpcTimeout = config.getRpcTimeout();
+ PCEPTopologyConfiguration(final @NonNull InetSocketAddress address, final short rpcTimeout,
+ final @NonNull KeyMapping keys, final @NonNull SpeakerIdMapping speakerIds) {
+ this.address = requireNonNull(address);
+ this.keys = requireNonNull(keys);
+ this.speakerIds = requireNonNull(speakerIds);
+ this.rpcTimeout = rpcTimeout;
}
- public @NonNull TopologyId getTopologyId() {
- return topology.getKey().getTopologyId();
- }
+ static @Nullable PCEPTopologyConfiguration of(final @NonNull Topology topology) {
+ // FIXME: this should live in the pcep topology type's presence container and be mandatory
+ final var pcepConfig = topology.augmentation(PcepTopologyTypeConfig.class);
+ if (pcepConfig == null) {
+ return null;
+ }
+ final var sessionConfig = pcepConfig.getSessionConfig();
+ if (sessionConfig == null) {
+ return null;
+ }
- public @NonNull KeyedInstanceIdentifier<Topology, TopologyKey> getTopology() {
- return topology;
+ return new PCEPTopologyConfiguration(
+ getInetSocketAddress(sessionConfig.getListenAddress(), sessionConfig.getListenPort()),
+ sessionConfig.getRpcTimeout(), constructKeys(topology.getNode()), contructSpeakersId(topology.getNode()));
}
- public short getRpcTimeout() {
+ short getRpcTimeout() {
return rpcTimeout;
}
- public @NonNull InetSocketAddress getAddress() {
+ @NonNull InetSocketAddress getAddress() {
return address;
}
- public @NonNull KeyMapping getKeys() {
+ @NonNull KeyMapping getKeys() {
return keys;
}
- public @NonNull SpeakerIdMapping getSpeakerIds() {
+ @NonNull SpeakerIdMapping getSpeakerIds() {
return speakerIds;
}
--- /dev/null
+/*
+ * Copyright (c) 2021 PANTHEON.tech, s.r.o. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.bgpcep.pcep.topology.provider;
+
+import static com.google.common.base.Verify.verifyNotNull;
+import static java.util.Objects.requireNonNull;
+
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.ListenableFuture;
+import java.util.Collection;
+import java.util.Dictionary;
+import java.util.Hashtable;
+import org.checkerframework.checker.lock.qual.GuardedBy;
+import org.eclipse.jdt.annotation.NonNull;
+import org.gaul.modernizer_maven_annotations.SuppressModernizer;
+import org.opendaylight.bgpcep.programming.spi.InstructionScheduler;
+import org.opendaylight.bgpcep.topology.DefaultTopologyReference;
+import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
+import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
+import org.opendaylight.mdsal.binding.api.DataTreeModification;
+import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
+import org.opendaylight.yangtools.concepts.Registration;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceRegistration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class tracks the configuration content of a particular topology instance and propagates updates towards its
+ * associated {@link PCEPTopologyProvider}.
+ */
+final class PCEPTopologyInstance implements ClusteredDataTreeChangeListener<Topology> {
+ private static final Logger LOG = LoggerFactory.getLogger(PCEPTopologyInstance.class);
+
+ private final @NonNull TopologyKey topology;
+
+ @GuardedBy("this")
+ private PCEPTopologyProvider provider;
+ @GuardedBy("this")
+ private Registration reg;
+ @GuardedBy("this")
+ private ServiceRegistration<?> osgiReg;
+
+ PCEPTopologyInstance(final TopologyKey topology, final PCEPTopologyProviderDependencies dependencies,
+ final InstructionScheduler scheduler, final BundleContext bundleContext) {
+ this.topology = requireNonNull(topology);
+
+ final var instanceIdentifier = InstanceIdentifier.create(NetworkTopology.class).child(Topology.class, topology);
+
+ provider = new PCEPTopologyProvider(instanceIdentifier, dependencies, scheduler);
+
+ // FIXME: BGPCEP-960: this should not be necessary
+ osgiReg = bundleContext.registerService(DefaultTopologyReference.class.getName(), provider, props());
+
+ reg = dependencies.getDataBroker().registerDataTreeChangeListener(
+ DataTreeIdentifier.create(LogicalDatastoreType.CONFIGURATION, instanceIdentifier), this);
+ LOG.info("Topology instance for {} initialized", topologyId());
+ }
+
+ synchronized ListenableFuture<?> terminate() {
+ verifyNotNull(reg, "Topology %s instance %s already terminating", topologyId(), this);
+ reg.close();
+ reg = null;
+
+ osgiReg.unregister();
+ osgiReg = null;
+
+ final var ret = provider.stop();
+ provider = null;
+ return ret;
+ }
+
+ @Override
+ public synchronized void onDataTreeChanged(final Collection<DataTreeModification<Topology>> changes) {
+ if (reg == null) {
+ // We have been shut down, do not process any more updates
+ return;
+ }
+
+ // We are only interested in the after-image
+ final var content = Iterables.getLast(changes).getRootNode().getDataAfter();
+ if (content != null) {
+ LOG.trace("Updating topology {} configuration to {}", topologyId(), content);
+ provider.updateConfiguration(PCEPTopologyConfiguration.of(content));
+ } else {
+ LOG.info("Topology {} configuration disappeared, ignoring update in anticipation of shutdown",
+ topologyId());
+ }
+ }
+
+ private String topologyId() {
+ return TopologyUtils.friendlyId(topology);
+ }
+
+ @SuppressModernizer
+ private Dictionary<String, String> props() {
+ final Dictionary<String, String> properties = new Hashtable<>(2);
+ properties.put(PCEPTopologyProvider.class.getName(), topologyId());
+ return properties;
+ }
+}
*/
package org.opendaylight.bgpcep.pcep.topology.provider;
-import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Verify.verify;
import static java.util.Objects.requireNonNull;
-import com.google.common.util.concurrent.FluentFuture;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import java.util.List;
+import io.netty.channel.epoll.EpollChannelOption;
+import java.util.Arrays;
+import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyConfiguration;
-import org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyProviderDependencies;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.checkerframework.checker.lock.qual.GuardedBy;
+import org.checkerframework.checker.lock.qual.Holding;
+import org.eclipse.jdt.annotation.NonNull;
+import org.eclipse.jdt.annotation.Nullable;
import org.opendaylight.bgpcep.programming.spi.InstructionScheduler;
import org.opendaylight.bgpcep.topology.DefaultTopologyReference;
import org.opendaylight.mdsal.binding.api.RpcProviderService;
-import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.protocol.pcep.PCEPCapability;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.programming.rev181109.NetworkTopologyPcepProgrammingService;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.NetworkTopologyPcepService;
-import org.opendaylight.yangtools.concepts.ObjectRegistration;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
+import org.opendaylight.yangtools.concepts.Registration;
+import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.Empty;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public final class PCEPTopologyProvider extends DefaultTopologyReference {
- private final ServerSessionManager manager;
- private final PCEPTopologyProviderDependencies dependenciesProvider;
- private final PCEPTopologyConfiguration configDependencies;
+final class PCEPTopologyProvider extends DefaultTopologyReference {
+ private static final Logger LOG = LoggerFactory.getLogger(PCEPTopologyProvider.class);
+
+ private final KeyedInstanceIdentifier<Topology, TopologyKey> instanceIdentifier;
+ private final PCEPTopologyProviderDependencies dependencies;
private final InstructionScheduler scheduler;
- private ObjectRegistration<NetworkTopologyPcepProgrammingService> network;
- private ObjectRegistration<NetworkTopologyPcepService> element;
+ // High-level state bits: currently running asynchronous operation, current configuration and the next configuration
+ // to apply after the async operation completes
+ @GuardedBy("this")
+ private ListenableFuture<?> asyncOperation;
+ @GuardedBy("this")
+ private PCEPTopologyConfiguration currentConfig;
+ @GuardedBy("this")
+ private Optional<PCEPTopologyConfiguration> nextConfig;
+
+ // Future indicating shutdown in progress
+ @GuardedBy("this")
+ private SettableFuture<Empty> stopFuture;
+
+ // Low-level state bits
+ @GuardedBy("this")
+ private ServerSessionManager manager;
+ @GuardedBy("this")
+ private PCEPStatefulPeerProposal proposal;
+ @GuardedBy("this")
private Channel channel;
+ @GuardedBy("this")
+ private Registration networkReg;
+ @GuardedBy("this")
+ private Registration elementReg;
- private PCEPTopologyProvider(
- final PCEPTopologyConfiguration configDependencies,
- final PCEPTopologyProviderDependencies dependenciesProvider,
- final ServerSessionManager manager, final InstructionScheduler scheduler) {
- super(configDependencies.getTopology());
- this.dependenciesProvider = requireNonNull(dependenciesProvider);
- this.configDependencies = configDependencies;
- this.manager = requireNonNull(manager);
+ PCEPTopologyProvider(final KeyedInstanceIdentifier<Topology, TopologyKey> instanceIdentifier,
+ final PCEPTopologyProviderDependencies dependencies, final InstructionScheduler scheduler) {
+ super(instanceIdentifier);
+ this.instanceIdentifier = requireNonNull(instanceIdentifier);
+ this.dependencies = requireNonNull(dependencies);
this.scheduler = requireNonNull(scheduler);
}
- public static PCEPTopologyProvider create(final PCEPTopologyProviderDependencies dependenciesProvider,
- final InstructionScheduler scheduler, final PCEPTopologyConfiguration configDependencies) {
- final List<PCEPCapability> capabilities = dependenciesProvider.getPCEPDispatcher()
- .getPCEPSessionNegotiatorFactory().getPCEPSessionProposalFactory().getCapabilities();
- if (capabilities.stream().filter(PCEPCapability::isStateful).findAny().isEmpty()) {
- throw new IllegalStateException(
- "Stateful capability not defined, aborting PCEP Topology Provider instantiation");
+ synchronized ListenableFuture<?> stop() {
+ if (stopFuture != null) {
+ // Already stopping, just return the future
+ return stopFuture;
+ }
+
+ stopFuture = SettableFuture.create();
+ applyConfiguration(null);
+ if (asyncOperation == null) {
+ stopFuture.set(Empty.getInstance());
+ }
+ return stopFuture;
+ }
+
+ synchronized void updateConfiguration(final @Nullable PCEPTopologyConfiguration newConfiguration) {
+ // FIXME: BGPCEP-960: this check should be a one-time thing in PCEPTopologyTracker startup once we have OSGi DS
+ final var effectiveConfig = dependencies.getPCEPDispatcher().getPCEPSessionNegotiatorFactory()
+ .getPCEPSessionProposalFactory().getCapabilities().stream()
+ .anyMatch(PCEPCapability::isStateful) ? newConfiguration : null;
+
+ applyConfiguration(effectiveConfig);
+ }
+
+ @Holding("this")
+ private void applyConfiguration(final @Nullable PCEPTopologyConfiguration newConfiguration) {
+ if (asyncOperation != null) {
+ LOG.debug("Topology Provider {} is undergoing reconfiguration, delaying reconfiguration", topologyId());
+ nextConfig = Optional.ofNullable(newConfiguration);
+ } else {
+ doApplyConfiguration(newConfiguration);
}
+ }
+
+ @Holding("this")
+ private void doApplyConfiguration(final @Nullable PCEPTopologyConfiguration newConfiguration) {
+ LOG.debug("Topology Provider {} applying configuration {}", topologyId(), newConfiguration);
+
+ // Perform obvious enable/disable operations
+ if (newConfiguration == null) {
+ if (currentConfig != null) {
+ LOG.info("Topology Provider {} lost configuration, disabling it", topologyId());
+ disable();
+ }
+ return;
+ }
+ if (currentConfig == null) {
+ LOG.info("Topology Provider {} received configuration, enabling it", topologyId());
+ enable(newConfiguration);
+ return;
+ }
+
+ // We need to perform a complete restart if the listen address changes
+ final var currentAddress = currentConfig.getAddress();
+ final var newAddress = newConfiguration.getAddress();
+ if (!currentAddress.equals(newAddress)) {
+ LOG.info("Topology Provider {} listen address changed from {} to {}, restarting", topologyId(),
+ currentAddress, newAddress);
+ applyConfiguration(null);
+ applyConfiguration(newConfiguration);
+ return;
+ }
+
+ // TCP-MD5 configuration is propagated from the server channel to individual channels. For any node that has
+ // changed this configuration we need to tear down any existing session.
+ final var currentKeys = currentConfig.getKeys().asMap();
+ final var newKeys = newConfiguration.getKeys().asMap();
+ final var outdatedNodes = Stream.concat(currentKeys.keySet().stream(), newKeys.keySet().stream())
+ .distinct()
+ .filter(nodeId -> !Arrays.equals(currentKeys.get(nodeId), newKeys.get(nodeId)))
+ .collect(Collectors.toUnmodifiableList());
+
+ proposal.setSpeakerIds(newConfiguration.getSpeakerIds());
+ manager.setRpcTimeout(newConfiguration.getRpcTimeout());
+ if (!outdatedNodes.isEmpty()) {
+ LOG.info("Topology Provider {} updating {} TCP-MD5 keys", topologyId(), outdatedNodes.size());
+ if (channel.config().setOption(EpollChannelOption.TCP_MD5SIG, newKeys)) {
+ manager.tearDownSessions(outdatedNodes);
+ } else {
+ LOG.warn("Topology Provider {} failed to update TCP-MD5 keys", topologyId());
+ }
+ }
+
+ currentConfig = newConfiguration;
+ LOG.info("Topology Provider {} configuration updated", topologyId());
+ }
+
+ @Holding("this")
+ private void enable(final PCEPTopologyConfiguration newConfiguration) {
+ // Assert we are performing an asynchronous operation
+ final var future = startOperation();
+ currentConfig = newConfiguration;
- return new PCEPTopologyProvider(configDependencies, dependenciesProvider,
- new ServerSessionManager(dependenciesProvider, configDependencies), scheduler);
+ // First start the manager
+ manager = new ServerSessionManager(instanceIdentifier, dependencies, newConfiguration.getRpcTimeout());
+ final var managerStart = manager.start();
+ managerStart.addListener(() -> enableChannel(future, Futures.getUnchecked(managerStart)),
+ MoreExecutors.directExecutor());
}
- public void instantiateServiceInstance() throws ExecutionException, InterruptedException {
- final RpcProviderService rpcRegistry = dependenciesProvider.getRpcProviderRegistry();
+ private synchronized void enableChannel(final SettableFuture<Empty> future, final Boolean managerSuccess) {
+ if (!managerSuccess) {
+ manager = null;
+ currentConfig = null;
+ finishOperation(future);
+ return;
+ }
- element = requireNonNull(rpcRegistry.registerRpcImplementation(NetworkTopologyPcepService.class,
- new TopologyRPCs(manager), Set.of(configDependencies.getTopology())));
+ proposal = new PCEPStatefulPeerProposal(dependencies.getDataBroker(), instanceIdentifier,
+ currentConfig.getSpeakerIds());
- network = requireNonNull(rpcRegistry.registerRpcImplementation(NetworkTopologyPcepProgrammingService.class,
- new TopologyProgramming(scheduler, manager), Set.of(configDependencies.getTopology())));
+ LOG.info("PCEP Topology Provider {} starting server channel", topologyId());
+ final var channelFuture = dependencies.getPCEPDispatcher().createServer(
+ new PCEPDispatcherDependenciesImpl(manager, proposal, currentConfig));
+ channelFuture.addListener(ignored -> enableRPCs(future, channelFuture));
+ }
- manager.instantiateServiceInstance();
- final ChannelFuture channelFuture = dependenciesProvider.getPCEPDispatcher()
- .createServer(manager.getPCEPDispatcherDependencies());
- channelFuture.get();
+ private synchronized void enableRPCs(final SettableFuture<Empty> future, final ChannelFuture channelFuture) {
+ final var channelFailure = channelFuture.cause();
+ if (channelFailure != null) {
+ LOG.error("Topology Provider {} failed to initialize server channel", topologyId(), channelFailure);
+ disableManager(future);
+ return;
+ }
channel = channelFuture.channel();
+
+ // Register RPCs
+ final RpcProviderService rpcRegistry = dependencies.getRpcProviderRegistry();
+ elementReg = rpcRegistry.registerRpcImplementation(NetworkTopologyPcepService.class,
+ new TopologyRPCs(manager), Set.of(instanceIdentifier));
+ networkReg = rpcRegistry.registerRpcImplementation(NetworkTopologyPcepProgrammingService.class,
+ new TopologyProgramming(scheduler, manager), Set.of(instanceIdentifier));
+
+ // We are now completely initialized
+ LOG.info("PCEP Topology Provider {} enabled", topologyId());
+ finishOperation(future);
+ }
+
+ @Holding("this")
+ private void disable() {
+ // Unregister RPCs
+ if (networkReg != null) {
+ networkReg.close();
+ networkReg = null;
+ }
+ if (elementReg != null) {
+ elementReg.close();
+ elementReg = null;
+ }
+
+ // Assert we are performing an asynchronous operation
+ final var future = startOperation();
+
+ // Disable channel
+ final var channelFuture = channel.close();
+ channel = null;
+ channelFuture.addListener(ignored -> disableManager(future));
+ }
+
+ @Holding("this")
+ private void disableManager(final SettableFuture<Empty> future) {
+ proposal = null;
+ final var managerStop = manager.stop();
+ manager = null;
+ managerStop.addListener(() -> finishStopManager(future), MoreExecutors.directExecutor());
}
- public FluentFuture<? extends CommitInfo> closeServiceInstance() {
- //FIXME return also channelClose once ListenableFuture implements wildcard
- channel.close().addListener((ChannelFutureListener) future ->
- checkArgument(future.isSuccess(), "Channel failed to close: %s", future.cause()));
+ private synchronized void finishStopManager(final SettableFuture<Empty> future) {
+ // We are now completely shut down
+ currentConfig = null;
+ finishOperation(future);
+ }
- if (network != null) {
- network.close();
- network = null;
+ @Holding("this")
+ private SettableFuture<Empty> startOperation() {
+ verify(asyncOperation == null, "Operation %s has not finished yet", asyncOperation);
+ final var future = SettableFuture.<Empty>create();
+ asyncOperation = future;
+ return future;
+ }
+
+ @Holding("this")
+ private void finishOperation(final SettableFuture<Empty> future) {
+ asyncOperation = null;
+ future.set(Empty.getInstance());
+
+ // Process next configuration change if there is one
+ if (nextConfig != null) {
+ final var config = nextConfig.orElse(null);
+ nextConfig = null;
+ doApplyConfiguration(config);
+ return;
}
- if (element != null) {
- element.close();
- element = null;
+
+ // Check if we are shutting down
+ if (stopFuture != null) {
+ stopFuture.set(Empty.getInstance());
}
- return manager.closeServiceInstance();
+ }
+
+ private @NonNull String topologyId() {
+ return TopologyUtils.friendlyId(instanceIdentifier);
}
}
\ No newline at end of file
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.bgpcep.pcep.topology.provider.config;
+package org.opendaylight.bgpcep.pcep.topology.provider;
-import com.google.common.annotations.Beta;
import org.eclipse.jdt.annotation.NonNullByDefault;
import org.opendaylight.bgpcep.pcep.server.PceServerProvider;
import org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry;
/**
* Provides required dependencies for PCEPTopologyProviderProvider instantiation.
*/
-@Beta
@NonNullByDefault
-public interface PCEPTopologyProviderDependencies {
+interface PCEPTopologyProviderDependencies {
/**
* PCEP Dispatcher.
*
--- /dev/null
+/*
+ * Copyright (c) 2021 PANTHEON.tech, s.r.o. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.bgpcep.pcep.topology.provider;
+
+import static com.google.common.base.Verify.verify;
+import static com.google.common.base.Verify.verifyNotNull;
+import static java.util.Objects.requireNonNull;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
+import java.util.concurrent.ExecutionException;
+import org.checkerframework.checker.lock.qual.GuardedBy;
+import org.checkerframework.checker.lock.qual.Holding;
+import org.eclipse.jdt.annotation.NonNull;
+import org.opendaylight.bgpcep.programming.spi.InstructionScheduler;
+import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
+import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
+import org.opendaylight.yangtools.concepts.Registration;
+import org.opendaylight.yangtools.yang.common.Empty;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Glue between asynchronous lifecycle driven by {@link PCEPTopologyTracker} and asynchronous instantiation of a
+ * {@link ClusterSingletonService} for a particular topology.
+ */
+final class PCEPTopologySingleton {
+ // Common class for possible states
+ private abstract static class State {
+ // Nothing here
+ }
+
+ // The state for alive-and-kickin'. We are registered for cluster-wide instantiation
+ private final class Active extends State implements ClusterSingletonService {
+ private final InstructionScheduler scheduler;
+
+ @GuardedBy("this")
+ private SettableFuture<Empty> closeFuture;
+ @GuardedBy("this")
+ private PCEPTopologyInstance instance;
+ @GuardedBy("this")
+ private Registration reg;
+
+ Active() {
+ scheduler = tracker.instructionSchedulerFactory.createInstructionScheduler(
+ topology.getTopologyId().getValue());
+ reg = tracker.singletonService.registerClusterSingletonService(this);
+ }
+
+ @Override
+ public ServiceGroupIdentifier getIdentifier() {
+ return scheduler.getIdentifier();
+ }
+
+ @Override
+ public synchronized void instantiateServiceInstance() {
+ if (reg == null) {
+ LOG.trace("Topology {} instance {} is closed, instantiation skipped", topologyId(),
+ PCEPTopologySingleton.this);
+ return;
+ }
+
+ LOG.trace("Topology {} instance {} instantiating", topologyId(), PCEPTopologySingleton.this);
+ instance = new PCEPTopologyInstance(topology, tracker, scheduler, tracker.bundleContext);
+ }
+
+ @Override
+ public ListenableFuture<?> closeServiceInstance() {
+ // First adjust our state under the lock...
+ final SettableFuture<Empty> close;
+ final ListenableFuture<?> ret;
+
+ synchronized (this) {
+ LOG.trace("Topology {} instance {} closing", topologyId(), PCEPTopologySingleton.this);
+ if (closeFuture == null) {
+ close = closeFuture = SettableFuture.create();
+ } else {
+ close = closeFuture;
+ }
+
+ ret = verifyNotNull(instance).terminate();
+ instance = null;
+ }
+
+ // ... and then add completion callback. Order of operations is significant, as we want to update our state
+ // before we give anybody a chance to react to closeFuture.
+ ret.addListener(() -> {
+ LOG.trace("Topology {} instance {} completing close", topologyId(), PCEPTopologySingleton.this);
+ synchronized (this) {
+ closeFuture = null;
+ }
+ LOG.trace("Topology {} instance {} closed", topologyId(), PCEPTopologySingleton.this);
+ close.set(Empty.getInstance());
+ }, MoreExecutors.directExecutor());
+ return close;
+ }
+
+ ListenableFuture<?> terminate() {
+ // Acquire the service termination future and shut down the scheduler once it completes
+ final var future = lockedTerminate();
+ future.addListener(scheduler::close, MoreExecutors.directExecutor());
+ return future;
+ }
+
+ // This part of the shutdown procedure needs to synchronize with instantiation and closure
+ private synchronized ListenableFuture<?> lockedTerminate() {
+ verifyNotNull(reg, "Topology %s instance %s already terminating", topologyId(), PCEPTopologySingleton.this);
+
+ final ListenableFuture<?> ret;
+ if (closeFuture == null) {
+ // Service is not being closed, we need to create a future...
+ ret = closeFuture = SettableFuture.create();
+ if (instance == null) {
+ // ... and there is no instance, hence we need to also compete it immediate
+ closeFuture.set(Empty.getInstance());
+ }
+ } else {
+ // Service close is already going on, reuse that future
+ ret = closeFuture;
+ }
+
+ // Close the registration, potentially triggering closeServiceInstance(), which may even run on this thread
+ reg.close();
+ reg = null;
+ return ret;
+ }
+ }
+
+ // Instance terminating, we never get out of this state
+ private static final class Terminating extends State {
+ final @NonNull ListenableFuture<?> future;
+
+ Terminating(final ListenableFuture<?> future) {
+ this.future = requireNonNull(future);
+ }
+ }
+
+ // Instance is waiting for a previous incarnation to finish terminating
+ private static final class Waiting extends State {
+ final @NonNull ListenableFuture<?> future;
+
+ Waiting(final ListenableFuture<?> future) {
+ this.future = requireNonNull(future);
+ }
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(PCEPTopologySingleton.class);
+
+ private final PCEPTopologyTracker tracker;
+ private final TopologyKey topology;
+
+ @GuardedBy("this")
+ private State state;
+ @GuardedBy("this")
+ private SettableFuture<?> cleanupFuture;
+
+ private PCEPTopologySingleton(final PCEPTopologySingleton previous, final ListenableFuture<?> future) {
+ tracker = previous.tracker;
+ topology = previous.topology;
+ state = new Waiting(future);
+ future.addListener(this::becomeActive, MoreExecutors.directExecutor());
+ }
+
+ PCEPTopologySingleton(final PCEPTopologyTracker tracker, final TopologyKey topology) {
+ this.tracker = requireNonNull(tracker);
+ this.topology = requireNonNull(topology);
+ state = new Active();
+ }
+
+ synchronized void destroy() {
+ if (state instanceof Active) {
+ LOG.trace("Starting destruction of topology {} instance {}", topologyId(), this);
+ becomeTerminating(((Active) state).terminate());
+ } else if (state instanceof Waiting) {
+ LOG.trace("Topology {} instance {} destroyed while waiting", topologyId(), this);
+ becomeTerminating(((Waiting) state).future);
+ } else {
+ verify(state instanceof Terminating, "Unexpected state %s", state);
+ LOG.trace("Topology {} instance {} is already being destroyed", topologyId(), this);
+ }
+ }
+
+ @NonNull PCEPTopologySingleton resurrect() {
+ return new PCEPTopologySingleton(this, acquireCleanup());
+ }
+
+ void awaitCleanup() {
+ try {
+ acquireCleanup().get();
+ } catch (InterruptedException e) {
+ LOG.info("Interrupted while waiting for topology {} cleanup", topologyId(), e);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ LOG.error("Topology {} cleanup failed", topologyId(), e);
+ }
+ }
+
+ private synchronized @NonNull ListenableFuture<?> acquireCleanup() {
+ return verifyTerminating().future;
+ }
+
+ private synchronized void becomeActive() {
+ if (state instanceof Waiting) {
+ LOG.trace("Topology {} instance {} becoming active", topologyId(), this);
+ state = new Active();
+ } else {
+ verifyTerminating();
+ LOG.trace("Skipping activation of terminated topology {} instance {}", topologyId(), this);
+ }
+ }
+
+ @Holding("this")
+ private void becomeTerminating(final ListenableFuture<?> future) {
+ state = new Terminating(future);
+ future.addListener(() -> tracker.finishDestroy(topology, this), MoreExecutors.directExecutor());
+ }
+
+ @Holding("this")
+ private Terminating verifyTerminating() {
+ verify(state instanceof Terminating, "Unexpected topology %s instance %s state %s", topologyId(), this, state);
+ return (Terminating) state;
+ }
+
+ private String topologyId() {
+ return TopologyUtils.friendlyId(topology);
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2021 PANTHEON.tech, s.r.o. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.bgpcep.pcep.topology.provider;
+
+import static com.google.common.base.Verify.verifyNotNull;
+import static java.util.Objects.requireNonNull;
+import static org.opendaylight.bgpcep.pcep.topology.provider.TopologyUtils.friendlyId;
+
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.checkerframework.checker.lock.qual.GuardedBy;
+import org.eclipse.jdt.annotation.NonNull;
+import org.opendaylight.bgpcep.pcep.server.PceServerProvider;
+import org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry;
+import org.opendaylight.bgpcep.programming.spi.InstructionSchedulerFactory;
+import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
+import org.opendaylight.mdsal.binding.api.DataBroker;
+import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
+import org.opendaylight.mdsal.binding.api.DataTreeModification;
+import org.opendaylight.mdsal.binding.api.RpcProviderService;
+import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
+import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
+import org.opendaylight.protocol.pcep.PCEPDispatcher;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TopologyTypes1;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.topology.pcep.type.TopologyPcep;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypes;
+import org.opendaylight.yangtools.concepts.Registration;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.osgi.framework.BundleContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Primary entrypoint into this component. Once an instance of this class is instantiated, it will subscribe to
+ * changes to the configuration datastore. There it filters only topologies which have {@link TopologyPcep} type and for
+ * each one of those instantiates a cluster-wide singleton to handle lifecycle of services attached to that topology.
+ */
+public final class PCEPTopologyTracker
+ implements PCEPTopologyProviderDependencies, ClusteredDataTreeChangeListener<TopologyPcep>, AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(PCEPTopologyTracker.class);
+
+ // Services we are using
+ final @NonNull InstructionSchedulerFactory instructionSchedulerFactory;
+ final @NonNull ClusterSingletonServiceProvider singletonService;
+ // FIXME: BGPCEP-960: this should not be needed
+ final @NonNull BundleContext bundleContext;
+ private final @NonNull TopologySessionStatsRegistry stateRegistry;
+ private final @NonNull RpcProviderService rpcProviderRegistry;
+ private final @NonNull PceServerProvider pceServerProvider;
+ private final @NonNull PCEPDispatcher pcepDispatcher;
+ private final @NonNull DataBroker dataBroker;
+
+ // We are reusing our monitor as the universal lock. We have to account for three distinct threads competing for
+ // our state:
+ // 1) the typical DTCL callback thread invoking onDataTreeChanged()
+ // 2) instance cleanup thread invoking finishDestroy()
+ // 3) framework shutdown thread invoking close()
+ //
+ // We need to track not only instances which are deemed alive by the class, but also all instances for which cleanup
+ // has not finished yet, so close() can properly wait for cleanup to finish.
+ //
+ // Since close() will terminate the DTCL subscription, the synchronization between 1) and 3) is rather trivial.
+ //
+ // The interaction between DTCL and cleanup is tricky. DTCL can report rapid create/destroy/create events and
+ // cleanup is asynchronous and when the dust settles we need to end up in the corrected overall state (created or
+ // destroyed).
+ //
+ // In order to achieve that without risking deadlocks, instances are tracked using a concurrent map and each
+ // 'create' edge allocates a new PCEPTopologyInstance object.
+ private final ConcurrentMap<TopologyKey, PCEPTopologySingleton> instances = new ConcurrentHashMap<>();
+ @GuardedBy("this")
+ private Registration reg;
+
+ public PCEPTopologyTracker(final DataBroker dataBroker, final ClusterSingletonServiceProvider singletonService,
+ final RpcProviderService rpcProviderRegistry, final PCEPDispatcher pcepDispatcher,
+ final InstructionSchedulerFactory instructionSchedulerFactory,
+ final TopologySessionStatsRegistry stateRegistry, final PceServerProvider pceServerProvider,
+ // FIXME: we should not be needing this OSGi dependency
+ final BundleContext bundleContext) {
+ this.dataBroker = requireNonNull(dataBroker);
+ this.singletonService = requireNonNull(singletonService);
+ this.rpcProviderRegistry = requireNonNull(rpcProviderRegistry);
+ this.pcepDispatcher = requireNonNull(pcepDispatcher);
+ this.instructionSchedulerFactory = requireNonNull(instructionSchedulerFactory);
+ this.stateRegistry = requireNonNull(stateRegistry);
+ this.pceServerProvider = requireNonNull(pceServerProvider);
+ this.bundleContext = requireNonNull(bundleContext);
+
+ reg = dataBroker.registerDataTreeChangeListener(DataTreeIdentifier.create(LogicalDatastoreType.CONFIGURATION,
+ InstanceIdentifier.builder(NetworkTopology.class).child(Topology.class).child(TopologyTypes.class)
+ .augmentation(TopologyTypes1.class).child(TopologyPcep.class).build()), this);
+ LOG.info("PCEP Topology tracker initialized");
+ }
+
+ @Override
+ public PCEPDispatcher getPCEPDispatcher() {
+ return pcepDispatcher;
+ }
+
+ @Override
+ public RpcProviderService getRpcProviderRegistry() {
+ return rpcProviderRegistry;
+ }
+
+ @Override
+ public DataBroker getDataBroker() {
+ return dataBroker;
+ }
+
+ @Override
+ public TopologySessionStatsRegistry getStateRegistry() {
+ return stateRegistry;
+ }
+
+ @Override
+ public PceServerProvider getPceServerProvider() {
+ return pceServerProvider;
+ }
+
+ @Override
+ public synchronized void close() {
+ if (reg == null) {
+ // Already closed, bail out
+ return;
+ }
+
+ LOG.info("PCEP Topology tracker shutting down");
+ reg.close();
+ reg = null;
+
+ // First pass: destroy all tracked instances
+ instances.values().forEach(PCEPTopologySingleton::destroy);
+ // Second pass: wait for cleanup
+ instances.values().forEach(PCEPTopologySingleton::awaitCleanup);
+ LOG.info("PCEP Topology tracker shut down");
+ }
+
+ @Override
+ public synchronized void onDataTreeChanged(final Collection<DataTreeModification<TopologyPcep>> changes) {
+ if (reg == null) {
+ // Registration has been terminated, do not process any changes
+ return;
+ }
+
+ for (var change : changes) {
+ final var root = change.getRootNode();
+ switch (root.getModificationType()) {
+ case WRITE:
+ // We only care if the topology has been newly introduced, not when its details have changed
+ if (root.getDataBefore() == null) {
+ createInstance(extractTopologyKey(change));
+ }
+ break;
+ case DELETE:
+ destroyInstance(extractTopologyKey(change));
+ break;
+ default:
+ // No-op
+ }
+ }
+ }
+
+ private void createInstance(final @NonNull TopologyKey topology) {
+ final var existing = instances.remove(topology);
+ final PCEPTopologySingleton instance;
+ if (existing == null) {
+ LOG.info("Creating topology instance for {}", friendlyId(topology));
+ instance = new PCEPTopologySingleton(this, topology);
+ } else {
+ LOG.info("Resurrecting topology instance for {}", friendlyId(topology));
+ instance = existing.resurrect();
+ }
+ instances.put(topology, instance);
+ }
+
+ private void destroyInstance(final @NonNull TopologyKey topology) {
+ final var existing = instances.get(topology);
+ if (existing != null) {
+ LOG.info("Destroying topology instance for {}", friendlyId(topology));
+ existing.destroy();
+ } else {
+ LOG.warn("Attempted to destroy non-existent topology instance for {}", friendlyId(topology));
+ }
+ }
+
+ void finishDestroy(final TopologyKey topology, final PCEPTopologySingleton instance) {
+ if (instances.remove(topology, instance)) {
+ LOG.info("Destroyed topology instance of {}", friendlyId(topology));
+ }
+ }
+
+ private static @NonNull TopologyKey extractTopologyKey(final DataTreeModification<?> change) {
+ final var path = change.getRootPath().getRootIdentifier();
+ return verifyNotNull(path.firstKeyOf(Topology.class), "No topology key in %s", path);
+ }
+}
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
import java.net.InetAddress;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.checkerframework.checker.lock.qual.GuardedBy;
import org.eclipse.jdt.annotation.NonNull;
-import org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyConfiguration;
-import org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyProviderDependencies;
import org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry;
import org.opendaylight.mdsal.binding.api.WriteTransaction;
import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
-import org.opendaylight.protocol.pcep.PCEPDispatcherDependencies;
import org.opendaylight.protocol.pcep.PCEPSession;
import org.opendaylight.protocol.pcep.PCEPSessionListenerFactory;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.stats.rev171113.PcepSessionState;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.OperationResult;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.RemoveLspArgs;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TearDownSessionInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TearDownSessionInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TopologyTypes1Builder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TriggerSyncArgs;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.UpdateLspArgs;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.topology.pcep.type.TopologyPcepBuilder;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypesBuilder;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
import org.opendaylight.yangtools.yang.common.RpcError;
import org.opendaylight.yangtools.yang.common.RpcResult;
private static final Logger LOG = LoggerFactory.getLogger(ServerSessionManager.class);
private static final long DEFAULT_HOLD_STATE_NANOS = TimeUnit.MINUTES.toNanos(5);
+ private final @NonNull KeyedInstanceIdentifier<Topology, TopologyKey> topology;
+ private final @NonNull PCEPTopologyProviderDependencies dependencies;
+
@VisibleForTesting
final AtomicBoolean isClosed = new AtomicBoolean(false);
@GuardedBy("this")
private final Map<NodeId, TopologySessionListener> nodes = new HashMap<>();
@GuardedBy("this")
private final Map<NodeId, TopologyNodeState> state = new HashMap<>();
- private final InstanceIdentifier<Topology> topology;
- private final short rpcTimeout;
- private final PCEPTopologyProviderDependencies dependencies;
- private final PCEPDispatcherDependencies pcepDispatcherDependencies;
-
- ServerSessionManager(
- final PCEPTopologyProviderDependencies dependencies,
- final PCEPTopologyConfiguration configuration) {
+
+ private volatile short rpcTimeout;
+
+ ServerSessionManager(final KeyedInstanceIdentifier<Topology, TopologyKey> instanceIdentifier,
+ final PCEPTopologyProviderDependencies dependencies, final short rpcTimeout) {
this.dependencies = requireNonNull(dependencies);
- topology = requireNonNull(configuration.getTopology());
- rpcTimeout = configuration.getRpcTimeout();
- pcepDispatcherDependencies = new PCEPDispatcherDependenciesImpl(this,
- new PCEPStatefulPeerProposal(dependencies.getDataBroker(), topology, configuration.getSpeakerIds()),
- configuration);
+ topology = requireNonNull(instanceIdentifier);
+ this.rpcTimeout = rpcTimeout;
}
- /**
- * Create Base Topology.
- */
- final synchronized void instantiateServiceInstance() {
- final TopologyKey key = InstanceIdentifier.keyOf(topology);
- final TopologyId topologyId = key.getTopologyId();
- final WriteTransaction tx = dependencies.getDataBroker().newWriteOnlyTransaction();
- tx.mergeParentStructurePut(LogicalDatastoreType.OPERATIONAL, topology, new TopologyBuilder()
- .withKey(key)
- .setTopologyId(topologyId).setTopologyTypes(new TopologyTypesBuilder()
- .addAugmentation(new TopologyTypes1Builder()
- .setTopologyPcep(new TopologyPcepBuilder().build())
- .build())
+ // Initialize the operational view of the topology.
+ final ListenableFuture<Boolean> start() {
+ LOG.info("Creating PCEP Topology {}", topologyId());
+
+ final var tx = dependencies.getDataBroker().newWriteOnlyTransaction();
+ tx.put(LogicalDatastoreType.OPERATIONAL, topology, new TopologyBuilder()
+ .withKey(topology.getKey())
+ .setTopologyTypes(new TopologyTypesBuilder()
+ .addAugmentation(new TopologyTypes1Builder().setTopologyPcep(new TopologyPcepBuilder().build()).build())
.build())
.build());
- try {
- tx.commit().get();
- LOG.info("PCEP Topology {} created successfully.", topologyId.getValue());
- ServerSessionManager.this.isClosed.set(false);
- } catch (final ExecutionException | InterruptedException throwable) {
- LOG.error("Failed to create PCEP Topology {}.", topologyId.getValue(), throwable);
- ServerSessionManager.this.isClosed.set(true);
+
+ final var future = SettableFuture.<Boolean>create();
+ final var txFuture = tx.commit();
+ txFuture.addCallback(new FutureCallback<CommitInfo>() {
+ @Override
+ public void onSuccess(final CommitInfo result) {
+ LOG.info("PCEP Topology {} created successfully.", topologyId());
+ isClosed.set(false);
+ future.set(Boolean.TRUE);
+ }
+
+ @Override
+ public void onFailure(final Throwable failure) {
+ LOG.error("Failed to create PCEP Topology {}.", topologyId(), failure);
+ isClosed.set(true);
+ future.set(Boolean.FALSE);
+ }
+ }, MoreExecutors.directExecutor());
+
+ return future;
+ }
+
+ final synchronized FluentFuture<? extends CommitInfo> stop() {
+ if (isClosed.getAndSet(true)) {
+ LOG.error("Session Manager has already been closed.");
+ return CommitInfo.emptyFluentFuture();
+ }
+ for (final TopologySessionListener node : nodes.values()) {
+ node.close();
+ }
+ nodes.clear();
+ for (final TopologyNodeState topologyNodeState : state.values()) {
+ topologyNodeState.close();
}
+ state.clear();
+
+ final WriteTransaction t = dependencies.getDataBroker().newWriteOnlyTransaction();
+ t.delete(LogicalDatastoreType.OPERATIONAL, topology);
+ final FluentFuture<? extends CommitInfo> future = t.commit();
+ future.addCallback(new FutureCallback<CommitInfo>() {
+ @Override
+ public void onSuccess(final CommitInfo result) {
+ LOG.debug("Topology {} removed", topology);
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ LOG.warn("Failed to remove Topology {}", topology, throwable);
+ }
+ }, MoreExecutors.directExecutor());
+ return future;
}
final synchronized void releaseNodeState(final TopologyNodeState nodeState, final PCEPSession session,
.buildFuture();
}
- final synchronized FluentFuture<? extends CommitInfo> closeServiceInstance() {
- if (isClosed.getAndSet(true)) {
- LOG.error("Session Manager has already been closed.");
- return CommitInfo.emptyFluentFuture();
- }
- for (final TopologySessionListener node : nodes.values()) {
- node.close();
- }
- nodes.clear();
- for (final TopologyNodeState topologyNodeState : state.values()) {
- topologyNodeState.close();
- }
- state.clear();
-
- final WriteTransaction t = dependencies.getDataBroker().newWriteOnlyTransaction();
- t.delete(LogicalDatastoreType.OPERATIONAL, topology);
- final FluentFuture<? extends CommitInfo> future = t.commit();
- future.addCallback(new FutureCallback<CommitInfo>() {
- @Override
- public void onSuccess(final CommitInfo result) {
- LOG.debug("Topology {} removed", topology);
- }
+ final short getRpcTimeout() {
+ return rpcTimeout;
+ }
- @Override
- public void onFailure(final Throwable throwable) {
- LOG.warn("Failed to remove Topology {}", topology, throwable);
- }
- }, MoreExecutors.directExecutor());
- return future;
+ final void setRpcTimeout(final short rpcTimeout) {
+ this.rpcTimeout = rpcTimeout;
}
- final short getRpcTimeout() {
- return rpcTimeout;
+ final void tearDownSessions(final List<InetAddress> outdatedNodes) {
+ for (var address : outdatedNodes) {
+ tearDownSession(new TearDownSessionInputBuilder().setNode(createNodeId(address)).build());
+ }
}
@Override
dependencies.getStateRegistry().unbind(nodeId);
}
- final PCEPDispatcherDependencies getPCEPDispatcherDependencies() {
- return pcepDispatcherDependencies;
- }
-
final PCEPTopologyProviderDependencies getPCEPTopologyProviderDependencies() {
return dependencies;
}
static @NonNull NodeId createNodeId(final InetAddress addr) {
return new NodeId("pcc://" + addr.getHostAddress());
}
+
+ private @NonNull String topologyId() {
+ return TopologyUtils.friendlyId(topology);
+ }
}
\ No newline at end of file
import org.eclipse.jdt.annotation.Nullable;
import org.opendaylight.yangtools.concepts.Immutable;
-public final class SpeakerIdMapping implements Immutable {
+final class SpeakerIdMapping implements Immutable {
private static final @NonNull SpeakerIdMapping EMPTY = new SpeakerIdMapping(ImmutableMap.of());
private final ImmutableMap<InetAddress, byte[]> map;
this.map = ImmutableMap.copyOf(map);
}
- public static @NonNull SpeakerIdMapping of() {
+ static @NonNull SpeakerIdMapping of() {
return EMPTY;
}
- public static @NonNull SpeakerIdMapping copyOf(final Map<InetAddress, byte[]> map) {
+ static @NonNull SpeakerIdMapping copyOf(final Map<InetAddress, byte[]> map) {
return map.isEmpty() ? of()
// Defensive: disconnect byte[]s from caller
: new SpeakerIdMapping(Maps.transformValues(map, byte[]::clone));
}
- public byte @Nullable [] speakerIdForAddress(final InetAddress address) {
+ byte @Nullable [] speakerIdForAddress(final InetAddress address) {
final byte[] found = map.get(address);
// Defensive: do not leak byte[]
return found == null ? null : found.clone();
--- /dev/null
+/*
+ * Copyright (c) 2021 PANTHEON.tech, s.r.o. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.bgpcep.pcep.topology.provider;
+
+import org.eclipse.jdt.annotation.NonNull;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
+import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
+
+/**
+ * Utilities for dealing with various PCEP topology-related constructs.
+ */
+final class TopologyUtils {
+ private TopologyUtils() {
+ // Hidden on purpose
+ }
+
+ static @NonNull String friendlyId(final KeyedInstanceIdentifier<Topology, TopologyKey> identifier) {
+ return friendlyId(identifier.getKey());
+ }
+
+ static @NonNull String friendlyId(final TopologyKey key) {
+ return friendlyId(key.getTopologyId());
+ }
+
+ static @NonNull String friendlyId(final TopologyId id) {
+ return id.getValue();
+ }
+}
+++ /dev/null
-/*
- * Copyright (c) 2017 Pantheon Technologies s.r.o. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.bgpcep.pcep.topology.provider.config;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import org.checkerframework.checker.lock.qual.GuardedBy;
-import org.checkerframework.checker.lock.qual.Holding;
-import org.eclipse.jdt.annotation.Nullable;
-import org.opendaylight.bgpcep.pcep.server.PceServerProvider;
-import org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry;
-import org.opendaylight.bgpcep.programming.spi.InstructionSchedulerFactory;
-import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
-import org.opendaylight.mdsal.binding.api.DataBroker;
-import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
-import org.opendaylight.mdsal.binding.api.DataTreeModification;
-import org.opendaylight.mdsal.binding.api.RpcProviderService;
-import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
-import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
-import org.opendaylight.protocol.pcep.PCEPDispatcher;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TopologyTypes1;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypes;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.osgi.framework.BundleContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public final class PCEPTopologyDeployerImpl implements ClusteredDataTreeChangeListener<Topology>, AutoCloseable {
- private static final Logger LOG = LoggerFactory.getLogger(PCEPTopologyDeployerImpl.class);
- private static final long TIMEOUT_NS = TimeUnit.SECONDS.toNanos(5);
-
- private final InstructionSchedulerFactory instructionSchedulerFactory;
- private final ClusterSingletonServiceProvider singletonService;
- private final PCEPDispatcher pcepDispatcher;
- private final DataBroker dataBroker;
- private final RpcProviderService rpcProviderRegistry;
- private final TopologySessionStatsRegistry stateRegistry;
- private final PceServerProvider pceServerProvider;
- private final BundleContext bundleContext;
-
- @GuardedBy("this")
- private final Map<TopologyId, PCEPTopologyProviderBean> pcepTopologyServices = new HashMap<>();
- @GuardedBy("this")
- private ListenerRegistration<PCEPTopologyDeployerImpl> listenerRegistration;
-
- public PCEPTopologyDeployerImpl(final DataBroker dataBroker, final ClusterSingletonServiceProvider singletonService,
- final RpcProviderService rpcProviderRegistry, final PCEPDispatcher pcepDispatcher,
- final InstructionSchedulerFactory instructionSchedulerFactory,
- final TopologySessionStatsRegistry stateRegistry, final PceServerProvider pceServerProvider,
- // FIXME: we should not be needing this OSGi dependency
- final BundleContext bundleContext) {
- this.dataBroker = requireNonNull(dataBroker);
- this.singletonService = requireNonNull(singletonService);
- this.rpcProviderRegistry = requireNonNull(rpcProviderRegistry);
- this.pcepDispatcher = requireNonNull(pcepDispatcher);
- this.instructionSchedulerFactory = requireNonNull(instructionSchedulerFactory);
- this.stateRegistry = requireNonNull(stateRegistry);
- this.pceServerProvider = requireNonNull(pceServerProvider);
- this.bundleContext = requireNonNull(bundleContext);
-
- LOG.info("PCEP Topology Deployer initialized");
- listenerRegistration = dataBroker.registerDataTreeChangeListener(
- DataTreeIdentifier.create(LogicalDatastoreType.CONFIGURATION,
- InstanceIdentifier.builder(NetworkTopology.class).child(Topology.class).build()), this);
- }
-
- @Override
- public synchronized void onDataTreeChanged(final Collection<DataTreeModification<Topology>> changes) {
- for (var change : changes) {
- final var topo = change.getRootNode();
- switch (topo.getModificationType()) {
- case SUBTREE_MODIFIED:
- case WRITE:
- // FIXME: BGPCEP-983: propagate all modifications
- updateTopologyProvider(topo.getDataAfter());
- break;
- case DELETE:
- removeTopologyProvider(topo.getDataBefore());
- break;
- default:
- throw new IllegalStateException("Unhandled modification type " + topo.getModificationType());
- }
- }
- }
-
- @Holding("this")
- private void updateTopologyProvider(final Topology topology) {
- if (!filterPcepTopologies(topology.getTopologyTypes())) {
- return;
- }
- final TopologyId topologyId = topology.getTopologyId();
- LOG.info("Updating Topology {}", topologyId);
- final PCEPTopologyProviderBean previous = pcepTopologyServices.remove(topologyId);
- closeTopology(previous, topologyId);
- createTopologyProvider(topology);
- }
-
- @Holding("this")
- private void createTopologyProvider(final Topology topology) {
- if (!filterPcepTopologies(topology.getTopologyTypes())) {
- return;
- }
- final TopologyId topologyId = topology.getTopologyId();
- if (pcepTopologyServices.containsKey(topologyId)) {
- LOG.warn("Topology Provider {} already exist. New instance won't be created", topologyId);
- return;
- }
- LOG.info("Creating Topology {}", topologyId);
- LOG.trace("Topology {}.", topology);
-
- final PCEPTopologyProviderBean pcepTopologyProviderBean = new PCEPTopologyProviderBean(
- dataBroker, pcepDispatcher, rpcProviderRegistry, stateRegistry, pceServerProvider);
- pcepTopologyServices.put(topologyId, pcepTopologyProviderBean);
-
- pcepTopologyProviderBean.start(instructionSchedulerFactory, singletonService,
- new PCEPTopologyConfiguration(topology), bundleContext);
- }
-
- @Holding("this")
- private synchronized void removeTopologyProvider(final Topology topology) {
- if (!filterPcepTopologies(topology.getTopologyTypes())) {
- return;
- }
- final TopologyId topologyId = topology.getTopologyId();
- closeTopology(pcepTopologyServices.remove(topologyId), topologyId);
- }
-
- @Override
- public synchronized void close() {
- LOG.info("PCEP Topology Deployer closing");
- if (listenerRegistration != null) {
- listenerRegistration.close();
- listenerRegistration = null;
- }
- for (Map.Entry<TopologyId, PCEPTopologyProviderBean> entry : pcepTopologyServices.entrySet()) {
- closeTopology(entry.getValue(), entry.getKey());
- }
- pcepTopologyServices.clear();
- LOG.info("PCEP Topology Deployer closed");
- }
-
- private static void closeTopology(final PCEPTopologyProviderBean topology, final TopologyId topologyId) {
- if (topology == null) {
- return;
- }
- LOG.info("Removing Topology {}", topologyId);
- try {
- topology.closeServiceInstance().get(TIMEOUT_NS, TimeUnit.NANOSECONDS);
- } catch (final InterruptedException | TimeoutException | ExecutionException e) {
- LOG.error("Topology {} instance failed to close service instance", topologyId, e);
- }
- topology.close();
- }
-
- private static boolean filterPcepTopologies(final @Nullable TopologyTypes topologyTypes) {
- if (topologyTypes == null) {
- return false;
- }
- final TopologyTypes1 aug = topologyTypes.augmentation(TopologyTypes1.class);
- return aug != null && aug.getTopologyPcep() != null;
- }
-}
\ No newline at end of file
+++ /dev/null
-/*
- * Copyright (c) 2017 Pantheon Technologies s.r.o. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.bgpcep.pcep.topology.provider.config;
-
-import static com.google.common.base.Preconditions.checkState;
-import static java.util.Objects.requireNonNull;
-
-import com.google.common.util.concurrent.FluentFuture;
-import java.util.List;
-import org.checkerframework.checker.lock.qual.GuardedBy;
-import org.opendaylight.bgpcep.pcep.server.PceServerProvider;
-import org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry;
-import org.opendaylight.bgpcep.programming.spi.InstructionSchedulerFactory;
-import org.opendaylight.mdsal.binding.api.DataBroker;
-import org.opendaylight.mdsal.binding.api.RpcProviderService;
-import org.opendaylight.mdsal.common.api.CommitInfo;
-import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
-import org.opendaylight.protocol.pcep.PCEPCapability;
-import org.opendaylight.protocol.pcep.PCEPDispatcher;
-import org.osgi.framework.BundleContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-final class PCEPTopologyProviderBean implements PCEPTopologyProviderDependencies, AutoCloseable {
- private static final Logger LOG = LoggerFactory.getLogger(PCEPTopologyProviderBean.class);
-
- private final PCEPDispatcher pcepDispatcher;
- private final DataBroker dataBroker;
- private final RpcProviderService rpcProviderRegistry;
- private final TopologySessionStatsRegistry stateRegistry;
- private final PceServerProvider pceServerProvider;
- @GuardedBy("this")
- private PCEPTopologyProviderSingleton pcepTopoProviderCSS;
-
- PCEPTopologyProviderBean(
- final DataBroker dataBroker,
- final PCEPDispatcher pcepDispatcher,
- final RpcProviderService rpcProviderRegistry,
- final TopologySessionStatsRegistry stateRegistry,
- final PceServerProvider pceServerProvider) {
- this.pcepDispatcher = requireNonNull(pcepDispatcher);
- this.dataBroker = requireNonNull(dataBroker);
- this.rpcProviderRegistry = requireNonNull(rpcProviderRegistry);
- this.stateRegistry = requireNonNull(stateRegistry);
- this.pceServerProvider = requireNonNull(pceServerProvider);
-
- // FIXME: this check should happen before we attempt anything
- final List<PCEPCapability> capabilities = pcepDispatcher.getPCEPSessionNegotiatorFactory()
- .getPCEPSessionProposalFactory().getCapabilities();
- if (!capabilities.stream().anyMatch(PCEPCapability::isStateful)) {
- throw new IllegalStateException(
- "Stateful capability not defined, aborting PCEP Topology Deployer instantiation");
- }
- }
-
- synchronized FluentFuture<? extends CommitInfo> closeServiceInstance() {
- if (pcepTopoProviderCSS != null) {
- return pcepTopoProviderCSS.closeServiceInstance();
- }
- return CommitInfo.emptyFluentFuture();
- }
-
- @Override
- public synchronized void close() {
- if (pcepTopoProviderCSS != null) {
- pcepTopoProviderCSS.close();
- pcepTopoProviderCSS = null;
- }
- }
-
- @SuppressWarnings("checkstyle:IllegalCatch")
- synchronized void start(final InstructionSchedulerFactory instructionSchedulerFactory,
- final ClusterSingletonServiceProvider cssp, final PCEPTopologyConfiguration configDependencies,
- final BundleContext bundleContext) {
- checkState(pcepTopoProviderCSS == null, "Previous instance %s was not closed.", this);
- try {
- pcepTopoProviderCSS = new PCEPTopologyProviderSingleton(configDependencies, this,
- instructionSchedulerFactory, cssp, bundleContext);
- } catch (final Exception e) {
- LOG.debug("Failed to create PCEPTopologyProvider {}", configDependencies.getTopologyId().getValue(), e);
- }
- }
-
- @Override
- public PCEPDispatcher getPCEPDispatcher() {
- return pcepDispatcher;
- }
-
- @Override
- public RpcProviderService getRpcProviderRegistry() {
- return rpcProviderRegistry;
- }
-
- @Override
- public DataBroker getDataBroker() {
- return dataBroker;
- }
-
- @Override
- public TopologySessionStatsRegistry getStateRegistry() {
- return stateRegistry;
- }
-
- @Override
- public PceServerProvider getPceServerProvider() {
- return pceServerProvider;
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2017 Pantheon Technologies s.r.o. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.bgpcep.pcep.topology.provider.config;
-
-import com.google.common.util.concurrent.FluentFuture;
-import java.util.Dictionary;
-import java.util.Hashtable;
-import org.checkerframework.checker.lock.qual.GuardedBy;
-import org.gaul.modernizer_maven_annotations.SuppressModernizer;
-import org.opendaylight.bgpcep.pcep.topology.provider.PCEPTopologyProvider;
-import org.opendaylight.bgpcep.programming.spi.InstructionScheduler;
-import org.opendaylight.bgpcep.programming.spi.InstructionSchedulerFactory;
-import org.opendaylight.bgpcep.topology.DefaultTopologyReference;
-import org.opendaylight.mdsal.common.api.CommitInfo;
-import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
-import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
-import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
-import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.ServiceRegistration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-final class PCEPTopologyProviderSingleton implements ClusterSingletonService, AutoCloseable {
- private static final Logger LOG = LoggerFactory.getLogger(PCEPTopologyProviderSingleton.class);
-
- private final PCEPTopologyProvider pcepTopoProvider;
- private final InstructionScheduler scheduler;
-
- private ServiceRegistration<?> serviceRegistration;
- private ClusterSingletonServiceRegistration cssRegistration;
-
- @GuardedBy("this")
- private boolean serviceInstantiated;
-
- PCEPTopologyProviderSingleton(final PCEPTopologyConfiguration configDependencies,
- final PCEPTopologyProviderDependencies dependenciesProvider,
- final InstructionSchedulerFactory instructionSchedulerFactory, final ClusterSingletonServiceProvider cssp,
- // FIXME: this should not be needed
- final BundleContext bundleContext) {
- scheduler = instructionSchedulerFactory.createInstructionScheduler(
- configDependencies.getTopologyId().getValue());
-
- // FIXME: this should only be created once we are up
- pcepTopoProvider = PCEPTopologyProvider.create(dependenciesProvider, scheduler, configDependencies);
-
- // FIXME: this should only be registered once we are up
- serviceRegistration = bundleContext.registerService(DefaultTopologyReference.class.getName(),
- pcepTopoProvider, props(configDependencies));
- LOG.info("PCEP Topology Provider service {} registered", getIdentifier().getName());
- cssRegistration = cssp.registerClusterSingletonService(this);
- }
-
- @Override
- @SuppressWarnings("checkstyle:IllegalCatch")
- public synchronized void instantiateServiceInstance() {
- LOG.info("PCEP Topology Provider Singleton Service {} instantiated", getIdentifier().getName());
- try {
- pcepTopoProvider.instantiateServiceInstance();
- } catch (final Exception e) {
- LOG.error("Failed to instantiate PCEP Topology provider", e);
- }
- serviceInstantiated = true;
- }
-
- @Override
- public synchronized FluentFuture<? extends CommitInfo> closeServiceInstance() {
- LOG.info("Close PCEP Topology Provider Singleton Service {}", getIdentifier().getName());
- if (serviceInstantiated) {
- serviceInstantiated = false;
- return pcepTopoProvider.closeServiceInstance();
- }
- return CommitInfo.emptyFluentFuture();
- }
-
- @Override
- public ServiceGroupIdentifier getIdentifier() {
- return scheduler.getIdentifier();
- }
-
- @Override
- public synchronized void close() {
- if (cssRegistration != null) {
- cssRegistration.close();
- cssRegistration = null;
- }
- if (serviceRegistration != null) {
- serviceRegistration.unregister();
- serviceRegistration = null;
- }
- scheduler.close();
- }
-
- @SuppressModernizer
- private static Dictionary<String, String> props(final PCEPTopologyConfiguration configDependencies) {
- final Dictionary<String, String> properties = new Hashtable<>();
- properties.put(PCEPTopologyProvider.class.getName(), configDependencies.getTopologyId().getValue());
- return properties;
- }
-}
\ No newline at end of file
interface="org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry"/>
<reference id="pceServerProvider" interface="org.opendaylight.bgpcep.pcep.server.PceServerProvider"/>
- <bean id="pcepTopologyDeployer"
- class="org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyDeployerImpl"
+ <bean id="pcepTopologyTracker"
+ class="org.opendaylight.bgpcep.pcep.topology.provider.PCEPTopologyTracker"
destroy-method="close">
<argument ref="dataBroker"/>
<argument ref="clusterSingletonServiceProvider"/>
package org.opendaylight.bgpcep.pcep.topology.provider;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
-import static org.opendaylight.protocol.util.CheckTestUtil.checkEquals;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
-import org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyConfiguration;
-import org.opendaylight.bgpcep.pcep.topology.provider.config.PCEPTopologyProviderDependencies;
import org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry;
import org.opendaylight.mdsal.binding.dom.adapter.test.AbstractConcurrentDataBrokerTest;
import org.opendaylight.protocol.pcep.impl.DefaultPCEPSessionNegotiator;
import org.opendaylight.protocol.pcep.impl.PCEPSessionImpl;
import org.opendaylight.protocol.util.InetSocketAddressUtil;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddressNoZone;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpPrefix;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4AddressNoZone;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4Prefix;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.PortNumber;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.config.rev200120.pcep.config.SessionConfigBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.explicit.route.object.Ero;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.explicit.route.object.EroBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.explicit.route.object.ero.Subobject;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.rsvp.rev150820.basic.explicit.route.subobjects.subobject.type.IpPrefixCase;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.rsvp.rev150820.basic.explicit.route.subobjects.subobject.type.IpPrefixCaseBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.rsvp.rev150820.basic.explicit.route.subobjects.subobject.type.ip.prefix._case.IpPrefixBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.config.rev181109.PcepTopologyTypeConfigBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.NetworkTopologyPcepService;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.Node1;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.PathComputationClient;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
-import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
import org.opendaylight.yangtools.yang.binding.Notification;
-import org.opendaylight.yangtools.yang.common.Uint16;
import org.opendaylight.yangtools.yang.common.Uint8;
public abstract class AbstractPCEPSessionTest extends AbstractConcurrentDataBrokerTest {
- static final short DEAD_TIMER = 30;
- static final short KEEP_ALIVE = 10;
static final short RPC_TIMEOUT = 4;
private static final TopologyKey TEST_TOPOLOGY_ID = new TopologyKey(new TopologyId("testtopo"));
- static final InstanceIdentifier<Topology> TOPO_IID = InstanceIdentifier.builder(NetworkTopology.class)
- .child(Topology.class, TEST_TOPOLOGY_ID).build();
+ static final KeyedInstanceIdentifier<Topology, TopologyKey> TOPO_IID =
+ InstanceIdentifier.create(NetworkTopology.class).child(Topology.class, TEST_TOPOLOGY_ID);
private static final String IPV4_MASK = "/32";
final String testAddress = InetSocketAddressUtil.getRandomLoopbackIpAddress();
final NodeId nodeId = new NodeId("pcc://" + testAddress);
doReturn(statsRegistry).when(topologyDependencies).getStateRegistry();
doReturn(null).when(topologyDependencies).getPceServerProvider();
- final PCEPTopologyConfiguration configDep = new PCEPTopologyConfiguration(new TopologyBuilder()
- .withKey(TEST_TOPOLOGY_ID)
- .addAugmentation(new PcepTopologyTypeConfigBuilder()
- .setSessionConfig(new SessionConfigBuilder()
- .setListenAddress(new IpAddressNoZone(new Ipv4AddressNoZone(testAddress)))
- .setListenPort(new PortNumber(Uint16.valueOf(4189)))
- .setRpcTimeout(RPC_TIMEOUT)
- .build())
- .build())
- .build());
- manager = customizeSessionManager(new ServerSessionManager(topologyDependencies, configDep));
+ manager = customizeSessionManager(new ServerSessionManager(TOPO_IID, topologyDependencies, RPC_TIMEOUT));
startSessionManager();
neg = new DefaultPCEPSessionNegotiator(promise, clientListener, manager.getSessionListener(), (short) 1, 5,
localPrefs);
}
void startSessionManager() throws Exception {
- manager.instantiateServiceInstance();
- checkEquals(() -> assertFalse(manager.isClosed.get()));
+ assertTrue(manager.start().get());
+ assertFalse(manager.isClosed.get());
}
void stopSessionManager() {
- manager.closeServiceInstance();
+ manager.stop();
}
@After
public class PCEPTopologySessionListenerTest extends AbstractPCEPSessionTest {
private final String tunnelName = "pcc_" + testAddress + "_tunnel_0";
+ private static final short DEAD_TIMER = 30;
+ private static final short KEEP_ALIVE = 10;
private PCEPTopologySessionListener listener;
Code Review / bgpcep.git / commitdiff