<classifier>features</classifier>
<type>xml</type>
</dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>features-netconf</artifactId>
+ <classifier>features</classifier>
+ <type>xml</type>
+ </dependency>
<dependency>
<groupId>org.opendaylight.controller</groupId>
<artifactId>features-config-persister</artifactId>
<repository>mvn:org.opendaylight.controller/features-config/${config.version}/xml/features</repository>
<repository>mvn:org.opendaylight.controller/features-config-persister/${config.version}/xml/features</repository>
<repository>mvn:org.opendaylight.controller/features-config-netty/${config.version}/xml/features</repository>
+ <repository>mvn:org.opendaylight.controller/features-netconf/${netconf.version}/xml/features</repository>
<repository>mvn:org.opendaylight.controller/features-akka/${commons.opendaylight.version}/xml/features</repository>
<feature name='odl-mdsal-all' version='${project.version}' description="OpenDaylight :: MDSAL :: All">
<feature version='${project.version}'>odl-mdsal-broker</feature>
<bundle>mvn:org.opendaylight.controller/sal-common-impl/${mdsal.version}</bundle>
<bundle>mvn:org.opendaylight.controller/sal-common-util/${mdsal.version}</bundle>
</feature>
+
+ <!-- TODO move to netconf features, however there are some weird dependencies on features-config-persister all over that cause cyclic dependencies-->
+ <feature name='odl-netconf-mdsal' version='${project.version}' description="OpenDaylight :: Netconf :: All">
+ <feature version='${config.version}'>odl-config-all</feature>
+ <feature version='${netconf.version}'>odl-netconf-all</feature>
+ <bundle>mvn:org.opendaylight.controller/netconf-ssh/${netconf.version}</bundle>
+ <feature version='${mdsal.version}'>odl-mdsal-broker</feature>
+ <bundle>mvn:org.opendaylight.controller/mdsal-netconf-connector/${netconf.version}</bundle>
+ <!-- TODO 01-netconf.xml file requires netconf-config-dispatcher to be present and its part of netconf-connector features. Clean Up-->
+ <bundle>mvn:org.opendaylight.controller/netconf-config-dispatcher/${config.version}</bundle>
+ <configfile finalname='${config.configfile.directory}/${config.netconf.client.configfile}'>mvn:org.opendaylight.controller/netconf-config/${netconf.version}/xml/config</configfile>
+ <configfile finalname='${config.configfile.directory}/${config.netconf.mdsal.configfile}'>mvn:org.opendaylight.controller/netconf-mdsal-config/${netconf.version}/xml/config</configfile>
+ </feature>
+
<feature name='odl-mdsal-broker' version='${project.version}' description="OpenDaylight :: MDSAL :: Broker">
<feature version='${yangtools.version}'>odl-yangtools-common</feature>
<feature version='${yangtools.version}'>odl-yangtools-binding</feature>
<groupId>org.opendaylight.controller</groupId>
<artifactId>netconf-config-dispatcher</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>mdsal-netconf-connector</artifactId>
+ </dependency>
<dependency>
<groupId>org.opendaylight.controller</groupId>
<artifactId>netconf-tcp</artifactId>
-->
<feature version='${project.version}'>odl-netconf-connector</feature>
<feature version='${project.version}'>odl-netconf-connector-ssh</feature>
+
+
</feature>
<!--
Necessary TODO: Define your features. It is useful to list then in order of dependency. So if A depends on B, list A first.
<groupId>org.opendaylight.controller</groupId>
<artifactId>netconf-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>netconf-config</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>netconf-mdsal-config</artifactId>
+ </dependency>
<dependency>
<groupId>org.opendaylight.controller</groupId>
<artifactId>netconf-auth</artifactId>
<config.xsql.configfile>04-xsql.xml</config.xsql.configfile>
<config.netconf.client.configfile>01-netconf.xml</config.netconf.client.configfile>
<config.toaster.configfile>03-toaster-sample.xml</config.toaster.configfile>
+ <config.netconf.mdsal.configfile>08-mdsal-netconf.xml</config.netconf.mdsal.configfile>
<config.restconf.configfile>10-rest-connector.xml</config.restconf.configfile>
<config.netconf.connector.configfile>99-netconf-connector.xml</config.netconf.connector.configfile>
<configuration.implementation.version>0.5.0-SNAPSHOT</configuration.implementation.version>
/**
*
*/
+@Deprecated
public final class NeverReconnectStrategyFactoryModule extends org.opendaylight.controller.config.yang.protocol.framework.AbstractNeverReconnectStrategyFactoryModule
{
/**
*
*/
+@Deprecated
public class NeverReconnectStrategyFactoryModuleFactory extends org.opendaylight.controller.config.yang.protocol.framework.AbstractNeverReconnectStrategyFactoryModuleFactory
{
/**
*
*/
+@Deprecated
public final class ReconnectImmediatelyStrategyFactoryModule extends org.opendaylight.controller.config.yang.protocol.framework.AbstractReconnectImmediatelyStrategyFactoryModule
{
/**
*
*/
+@Deprecated
public class ReconnectImmediatelyStrategyFactoryModuleFactory extends org.opendaylight.controller.config.yang.protocol.framework.AbstractReconnectImmediatelyStrategyFactoryModuleFactory
{
/**
*
*/
+@Deprecated
public final class TimedReconnectStrategyFactoryModule extends org.opendaylight.controller.config.yang.protocol.framework.AbstractTimedReconnectStrategyFactoryModule
{
/**
*
*/
+@Deprecated
public class TimedReconnectStrategyFactoryModuleFactory extends org.opendaylight.controller.config.yang.protocol.framework.AbstractTimedReconnectStrategyFactoryModuleFactory
{
* Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the
* start method that will handle sockets in different thread.
*/
+@Deprecated
public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> implements Closeable {
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Deprecated
public abstract class AbstractProtocolSession<M> extends SimpleChannelInboundHandler<Object> implements ProtocolSession<M> {
private static final Logger LOG = LoggerFactory.getLogger(AbstractProtocolSession.class);
* @param <M> Protocol message type
* @param <S> Protocol session type, has to extend ProtocolSession<M>
*/
+@Deprecated
public abstract class AbstractSessionNegotiator<M, S extends AbstractProtocolSession<?>> extends ChannelInboundHandlerAdapter implements SessionNegotiator<S> {
private final Logger LOG = LoggerFactory.getLogger(AbstractSessionNegotiator.class);
private final Promise<S> promise;
* Utility ReconnectStrategy singleton, which will cause the reconnect process
* to always fail.
*/
+@Deprecated
@ThreadSafe
public final class NeverReconnectStrategy implements ReconnectStrategy {
private final EventExecutor executor;
*
* This interface should be implemented by a final class representing a protocol specific session.
*/
+@Deprecated
public interface ProtocolSession<T> extends Closeable {
@Override
void close();
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Deprecated
@ThreadSafe
final class ProtocolSessionPromise<S extends ProtocolSession<?>> extends DefaultPromise<S> {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolSessionPromise.class);
* Utility ReconnectStrategy singleton, which will cause the reconnect process
* to immediately schedule a reconnection attempt.
*/
+@Deprecated
@ThreadSafe
public final class ReconnectImmediatelyStrategy implements ReconnectStrategy {
private static final Logger LOG = LoggerFactory.getLogger(ReconnectImmediatelyStrategy.class);
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Deprecated
final class ReconnectPromise<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> extends DefaultPromise<Void> {
private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
* not attempt any more connection attempts and should abort the reconnection
* process.
*/
+@Deprecated
public interface ReconnectStrategy {
/**
* Query the strategy for the connect timeout.
* primarily useful for allowing injection of a specific type of strategy for
* on-demand use, pretty much like you would use a ThreadFactory.
*/
+@Deprecated
public interface ReconnectStrategyFactory {
/**
* Create a new ReconnectStrategy.
* implemented by a protocol specific abstract class, that is extended by
* a final class that implements the methods.
*/
+@Deprecated
public interface SessionListener<M, S extends ProtocolSession<?>, T extends TerminationReason> extends EventListener {
/**
* Fired when the session was established successfully.
* implemented by a protocol specific abstract class, that is extended by
* a final class that implements the methods.
*/
+@Deprecated
public interface SessionListenerFactory<T extends SessionListener<?, ?, ?>> {
/**
* Returns one session listener
*
* @param <T> Protocol session type.
*/
+@Deprecated
public interface SessionNegotiator<T extends ProtocolSession<?>> extends ChannelInboundHandler {
}
*
* @param <S> session type
*/
+@Deprecated
public interface SessionNegotiatorFactory<M, S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> {
/**
* Create a new negotiator attached to a channel, which will notify
/**
* Marker interface for grouping session termination cause.
*/
+@Deprecated
public interface TerminationReason {
/**
*
* Both these caps can be combined, with the strategy giving up as soon as the first one is reached.
*/
+@Deprecated
@ThreadSafe
public final class TimedReconnectStrategy implements ReconnectStrategy {
private static final Logger LOG = LoggerFactory.getLogger(TimedReconnectStrategy.class);
}
protected int adjustedIndex(long logEntryIndex) {
- if(snapshotIndex < 0){
+ if (snapshotIndex < 0) {
return (int) logEntryIndex;
}
return (int) (logEntryIndex - (snapshotIndex + 1));
return journal.size();
}
+ @Override
+ public int dataSize() {
+ return dataSize;
+ }
+
@Override
public boolean isPresent(long logEntryIndex) {
if (logEntryIndex > lastIndex()) {
previousSnapshotIndex = -1;
previousSnapshotTerm = -1;
dataSize = 0;
+ // need to recalc the datasize based on the entries left after precommit.
+ for(ReplicatedLogEntry logEntry : journal) {
+ dataSize += logEntry.size();
+ }
+
}
@Override
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
-import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
import org.opendaylight.controller.cluster.raft.behaviors.AbstractRaftActorBehavior;
import org.opendaylight.controller.cluster.raft.behaviors.Follower;
import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
-import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
-import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
handleCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot());
} else {
- if (!(message instanceof AppendEntriesMessages.AppendEntries)
- && !(message instanceof AppendEntriesReply) && !(message instanceof SendHeartBeat)) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: onReceiveCommand: message: {}", persistenceId(), message.getClass());
- }
- }
-
RaftActorBehavior oldBehavior = currentBehavior;
currentBehavior = currentBehavior.handleMessage(getSender(), message);
/**
* This method is called during recovery to reconstruct the state of the actor.
*
- * @param snapshot A snapshot of the state of the actor
+ * @param snapshotBytes A snapshot of the state of the actor
*/
protected abstract void applyRecoverySnapshot(byte[] snapshotBytes);
LOG.info("{}: Persisting of snapshot done:{}", persistenceId(), sn.getLogMessage());
- //be greedy and remove entries from in-mem journal which are in the snapshot
- // and update snapshotIndex and snapshotTerm without waiting for the success,
+ long dataThreshold = Runtime.getRuntime().totalMemory() *
+ getRaftActorContext().getConfigParams().getSnapshotDataThresholdPercentage() / 100;
+ if (context.getReplicatedLog().dataSize() > dataThreshold) {
+ // if memory is less, clear the log based on lastApplied.
+ // this could/should only happen if one of the followers is down
+ // as normally we keep removing from the log when its replicated to all.
+ context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getLastAppliedIndex(),
+ captureSnapshot.getLastAppliedTerm());
- context.getReplicatedLog().snapshotPreCommit(
- captureSnapshot.getLastAppliedIndex(),
- captureSnapshot.getLastAppliedTerm());
+ } else {
+ // clear the log based on replicatedToAllIndex
+ context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getReplicatedToAllIndex(),
+ captureSnapshot.getReplicatedToAllTerm());
+ }
+ getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " +
"and term:{}", persistenceId(), captureSnapshot.getLastAppliedIndex(),
// FIXME: Maybe this should be done after the command is saved
journal.subList(adjustedIndex , journal.size()).clear();
- persistence().persist(new DeleteEntries(adjustedIndex), new Procedure<DeleteEntries>(){
+ persistence().persist(new DeleteEntries(adjustedIndex), new Procedure<DeleteEntries>() {
- @Override public void apply(DeleteEntries param)
- throws Exception {
+ @Override
+ public void apply(DeleteEntries param)
+ throws Exception {
//FIXME : Doing nothing for now
dataSize = 0;
- for(ReplicatedLogEntry entry : journal){
+ for (ReplicatedLogEntry entry : journal) {
dataSize += entry.size();
}
}
appendAndPersist(replicatedLogEntry, null);
}
- @Override
- public int dataSize() {
- return dataSize;
- }
-
public void appendAndPersist(
final ReplicatedLogEntry replicatedLogEntry,
final Procedure<ReplicatedLogEntry> callback) {
long dataSizeForCheck = dataSize;
dataSizeSinceLastSnapshot += logEntrySize;
- long journalSize = lastIndex()+1;
+ long journalSize = lastIndex() + 1;
if(!hasFollowers()) {
// When we do not have followers we do not maintain an in-memory log
}
// send a CaptureSnapshot to self to make the expensive operation async.
- getSelf().tell(new CaptureSnapshot(
- lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm),
+ long replicatedToAllIndex = getCurrentBehavior().getReplicatedToAllIndex();
+ ReplicatedLogEntry replicatedToAllEntry = context.getReplicatedLog().get(replicatedToAllIndex);
+ getSelf().tell(new CaptureSnapshot(lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm,
+ (replicatedToAllEntry != null ? replicatedToAllEntry.getIndex() : -1),
+ (replicatedToAllEntry != null ? replicatedToAllEntry.getTerm() : -1)),
null);
context.setSnapshotCaptureInitiated(true);
}
- if(callback != null){
+ if (callback != null){
callback.apply(replicatedLogEntry);
}
}
* sets snapshot term
* @param snapshotTerm
*/
- public void setSnapshotTerm(long snapshotTerm);
+ void setSnapshotTerm(long snapshotTerm);
/**
* Clears the journal entries with startIndex(inclusive) and endIndex (exclusive)
* @param startIndex
* @param endIndex
*/
- public void clear(int startIndex, int endIndex);
+ void clear(int startIndex, int endIndex);
/**
* Handles all the bookkeeping in order to perform a rollback in the
* @param snapshotCapturedIndex
* @param snapshotCapturedTerm
*/
- public void snapshotPreCommit(long snapshotCapturedIndex, long snapshotCapturedTerm);
+ void snapshotPreCommit(long snapshotCapturedIndex, long snapshotCapturedTerm);
/**
* Sets the Replicated log to state after snapshot success.
*/
- public void snapshotCommit();
+ void snapshotCommit();
/**
* Restores the replicated log to a state in the event of a save snapshot failure
*/
- public void snapshotRollback();
+ void snapshotRollback();
/**
* Size of the data in the log (in bytes)
*/
- public int dataSize();
+ int dataSize();
+
}
private long lastIndex;
private long lastTerm;
private boolean installSnapshotInitiated;
+ private long replicatedToAllIndex;
+ private long replicatedToAllTerm;
public CaptureSnapshot(long lastIndex, long lastTerm,
- long lastAppliedIndex, long lastAppliedTerm) {
- this(lastIndex, lastTerm, lastAppliedIndex, lastAppliedTerm, false);
+ long lastAppliedIndex, long lastAppliedTerm, long replicatedToAllIndex, long replicatedToAllTerm) {
+ this(lastIndex, lastTerm, lastAppliedIndex, lastAppliedTerm, replicatedToAllIndex , replicatedToAllTerm, false);
}
public CaptureSnapshot(long lastIndex, long lastTerm,long lastAppliedIndex,
- long lastAppliedTerm, boolean installSnapshotInitiated) {
+ long lastAppliedTerm, long replicatedToAllIndex, long replicatedToAllTerm, boolean installSnapshotInitiated) {
this.lastIndex = lastIndex;
this.lastTerm = lastTerm;
this.lastAppliedIndex = lastAppliedIndex;
this.lastAppliedTerm = lastAppliedTerm;
this.installSnapshotInitiated = installSnapshotInitiated;
+ this.replicatedToAllIndex = replicatedToAllIndex;
+ this.replicatedToAllTerm = replicatedToAllTerm;
}
public long getLastAppliedIndex() {
public boolean isInstallSnapshotInitiated() {
return installSnapshotInitiated;
}
+
+ public long getReplicatedToAllIndex() {
+ return replicatedToAllIndex;
+ }
+
+ public long getReplicatedToAllTerm() {
+ return replicatedToAllTerm;
+ }
}
private Optional<ByteString> snapshot;
- private long replicatedToAllIndex = -1;
-
public AbstractLeader(RaftActorContext context) {
- super(context);
+ super(context, RaftState.Leader);
final Builder<String, FollowerLogInformation> ftlBuilder = ImmutableMap.builder();
for (String followerId : context.getPeerAddresses().keySet()) {
leaderId = context.getId();
- LOG.debug("{}: Election: Leader has following peers: {}", context.getId(), getFollowerIds());
+ LOG.debug("{}: Election: Leader has following peers: {}", logName(), getFollowerIds());
minReplicationCount = getMajorityVoteCount(getFollowerIds().size());
// Upon election: send initial empty AppendEntries RPCs
// (heartbeat) to each server; repeat during idle periods to
// prevent election timeouts (§5.2)
- sendAppendEntries(0);
+ sendAppendEntries(0, false);
}
/**
return followerToLog.keySet();
}
- private Optional<ByteString> getSnapshot() {
- return snapshot;
- }
-
@VisibleForTesting
void setSnapshot(Optional<ByteString> snapshot) {
this.snapshot = snapshot;
protected RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
- }
+ LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
return this;
}
protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
AppendEntriesReply appendEntriesReply) {
- if(! appendEntriesReply.isSuccess()) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntriesReply: {}", context.getId(), appendEntriesReply);
- }
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply);
+ } else if(LOG.isDebugEnabled() && !appendEntriesReply.isSuccess()) {
+ LOG.debug("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply);
}
// Update the FollowerLogInformation
followerToLog.get(followerId);
if(followerLogInformation == null){
- LOG.error("{}: handleAppendEntriesReply - unknown follower {}", context.getId(), followerId);
+ LOG.error("{}: handleAppendEntriesReply - unknown follower {}", logName(), followerId);
return this;
}
// Apply the change to the state machine
if (context.getCommitIndex() > context.getLastApplied()) {
+ LOG.debug("{}: handleAppendEntriesReply: applying to log - commitIndex: {}, lastAppliedIndex: {}",
+ logName(), context.getCommitIndex(), context.getLastApplied());
+
applyLogToStateMachine(context.getCommitIndex());
}
}
//Send the next log entry immediately, if possible, no need to wait for heartbeat to trigger that event
- sendUpdatesToFollower(followerId, followerLogInformation, false);
+ sendUpdatesToFollower(followerId, followerLogInformation, false, false);
return this;
}
private void purgeInMemoryLog() {
- //find the lowest index across followers which has been replicated to all. -1 if there are no followers.
+ //find the lowest index across followers which has been replicated to all.
+ // lastApplied if there are no followers, so that we keep clearing the log for single-node
// we would delete the in-mem log from that index on, in-order to minimize mem usage
// we would also share this info thru AE with the followers so that they can delete their log entries as well.
- long minReplicatedToAllIndex = followerToLog.isEmpty() ? -1 : Long.MAX_VALUE;
+ long minReplicatedToAllIndex = followerToLog.isEmpty() ? context.getLastApplied() : Long.MAX_VALUE;
for (FollowerLogInformation info : followerToLog.values()) {
minReplicatedToAllIndex = Math.min(minReplicatedToAllIndex, info.getMatchIndex());
}
- replicatedToAllIndex = fakeSnapshot(minReplicatedToAllIndex, replicatedToAllIndex);
+ super.performSnapshotWithoutCapture(minReplicatedToAllIndex);
}
@Override
return this;
}
- @Override
- public RaftState state() {
- return RaftState.Leader;
- }
-
@Override
public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
Preconditions.checkNotNull(sender, "sender should not be null");
// set currentTerm = T, convert to follower (§5.1)
// This applies to all RPC messages and responses
if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
- LOG.debug("{}: Term {} in \"{}\" message is greater than leader's term {}", context.getId(),
- rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
+ LOG.debug("{}: Term {} in \"{}\" message is greater than leader's term {} - switching to Follower",
+ logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
}
private void handleInstallSnapshotReply(InstallSnapshotReply reply) {
+ LOG.debug("{}: handleInstallSnapshotReply: {}", logName(), reply);
+
String followerId = reply.getFollowerId();
FollowerToSnapshot followerToSnapshot = mapFollowerToSnapshot.get(followerId);
if (followerToSnapshot == null) {
LOG.error("{}: FollowerId {} in InstallSnapshotReply not known to Leader",
- context.getId(), followerId);
+ logName(), followerId);
return;
}
//this was the last chunk reply
if(LOG.isDebugEnabled()) {
LOG.debug("{}: InstallSnapshotReply received, " +
- "last chunk received, Chunk:{}. Follower:{} Setting nextIndex:{}",
- context.getId(), reply.getChunkIndex(), followerId,
+ "last chunk received, Chunk: {}. Follower: {} Setting nextIndex: {}",
+ logName(), reply.getChunkIndex(), followerId,
context.getReplicatedLog().getSnapshotIndex() + 1
);
}
context.getReplicatedLog().getSnapshotIndex() + 1);
mapFollowerToSnapshot.remove(followerId);
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: followerToLog.get(followerId).getNextIndex()=" +
- context.getId(), followerToLog.get(followerId).getNextIndex());
- }
+ LOG.debug("{}: follower: {}, matchIndex set to {}, nextIndex set to {}",
+ logName(), followerId, followerLogInformation.getMatchIndex(),
+ followerLogInformation.getNextIndex());
if (mapFollowerToSnapshot.isEmpty()) {
// once there are no pending followers receiving snapshots
}
} else {
LOG.info("{}: InstallSnapshotReply received sending snapshot chunk failed, Will retry, Chunk: {}",
- context.getId(), reply.getChunkIndex());
+ logName(), reply.getChunkIndex());
followerToSnapshot.markSendStatus(false);
}
} else {
LOG.error("{}: Chunk index {} in InstallSnapshotReply from follower {} does not match expected index {}",
- context.getId(), reply.getChunkIndex(), followerId,
+ logName(), reply.getChunkIndex(), followerId,
followerToSnapshot.getChunkIndex());
if(reply.getChunkIndex() == INVALID_CHUNK_INDEX){
private void replicate(Replicate replicate) {
long logIndex = replicate.getReplicatedLogEntry().getIndex();
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Replicate message {}", context.getId(), logIndex);
- }
+ LOG.debug("{}: Replicate message: identifier: {}, logIndex: {}", logName(),
+ replicate.getIdentifier(), logIndex);
// Create a tracker entry we will use this later to notify the
// client actor
context.setCommitIndex(logIndex);
applyLogToStateMachine(logIndex);
} else {
- sendAppendEntries(0);
+ sendAppendEntries(0, false);
}
}
- private void sendAppendEntries(long timeSinceLastActivityInterval) {
+ private void sendAppendEntries(long timeSinceLastActivityInterval, boolean isHeartbeat) {
// Send an AppendEntries to all followers
for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
final String followerId = e.getKey();
// This checks helps not to send a repeat message to the follower
if(!followerLogInformation.isFollowerActive() ||
followerLogInformation.timeSinceLastActivity() >= timeSinceLastActivityInterval) {
- sendUpdatesToFollower(followerId, followerLogInformation, true);
+ sendUpdatesToFollower(followerId, followerLogInformation, true, isHeartbeat);
}
}
}
*/
private void sendUpdatesToFollower(String followerId, FollowerLogInformation followerLogInformation,
- boolean sendHeartbeat) {
+ boolean sendHeartbeat, boolean isHeartbeat) {
ActorSelection followerActor = context.getPeerActorSelection(followerId);
if (followerActor != null) {
} else {
long leaderLastIndex = context.getReplicatedLog().lastIndex();
long leaderSnapShotIndex = context.getReplicatedLog().getSnapshotIndex();
- if (isFollowerActive &&
- context.getReplicatedLog().isPresent(followerNextIndex)) {
+
+ if(!isHeartbeat || LOG.isTraceEnabled()) {
+ LOG.debug("{}: Checking sendAppendEntries for follower {}, leaderLastIndex: {}, leaderSnapShotIndex: {}",
+ logName(), followerId, leaderLastIndex, leaderSnapShotIndex);
+ }
+
+ if (isFollowerActive && context.getReplicatedLog().isPresent(followerNextIndex)) {
+
+ LOG.debug("{}: sendAppendEntries: {} is present for follower {}", logName(),
+ followerNextIndex, followerId);
+
// FIXME : Sending one entry at a time
final List<ReplicatedLogEntry> entries = context.getReplicatedLog().getFrom(followerNextIndex, 1);
sendAppendEntriesToFollower(followerActor, followerNextIndex, entries, followerId);
} else if (isFollowerActive && followerNextIndex >= 0 &&
- leaderLastIndex >= followerNextIndex) {
+ leaderLastIndex > followerNextIndex && !context.isSnapshotCaptureInitiated()) {
// if the followers next index is not present in the leaders log, and
// if the follower is just not starting and if leader's index is more than followers index
// then snapshot should be sent
if (LOG.isDebugEnabled()) {
- LOG.debug("InitiateInstallSnapshot to follower:{}," +
- "follower-nextIndex:{}, leader-snapshot-index:{}, " +
- "leader-last-index:{}", followerId,
- followerNextIndex, leaderSnapShotIndex, leaderLastIndex
- );
+ LOG.debug(String.format("%s: InitiateInstallSnapshot to follower: %s," +
+ "follower-nextIndex: %d, leader-snapshot-index: %d, " +
+ "leader-last-index: %d", logName(), followerId,
+ followerNextIndex, leaderSnapShotIndex, leaderLastIndex));
}
// Send heartbeat to follower whenever install snapshot is initiated.
AppendEntries appendEntries = new AppendEntries(currentTerm(), context.getId(),
prevLogIndex(followerNextIndex),
prevLogTerm(followerNextIndex), entries,
- context.getCommitIndex(), replicatedToAllIndex);
+ context.getCommitIndex(), super.getReplicatedToAllIndex());
- if(!entries.isEmpty()) {
- LOG.debug("{}: Sending AppendEntries to follower {}: {}", context.getId(), followerId,
+ if(!entries.isEmpty() || LOG.isTraceEnabled()) {
+ LOG.debug("{}: Sending AppendEntries to follower {}: {}", logName(), followerId,
appendEntries);
}
}
/**
- * /**
* Install Snapshot works as follows
* 1. Leader initiates the capture snapshot by sending a CaptureSnapshot message to actor
* 2. RaftActor on receipt of the CaptureSnapshotReply (from Shard), stores the received snapshot in the replicated log
* @param followerNextIndex
*/
private void initiateCaptureSnapshot(String followerId, long followerNextIndex) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: initiateCaptureSnapshot, followers {}", context.getId(), followerToLog.keySet());
- }
-
if (!context.getReplicatedLog().isPresent(followerNextIndex) &&
context.getReplicatedLog().isInSnapshot(followerNextIndex)) {
} else if (!context.isSnapshotCaptureInitiated()) {
- LOG.info("{}: Initiating Snapshot Capture to Install Snapshot, Leader:{}", context.getId(), getLeaderId());
+ LOG.info("{}: Initiating Snapshot Capture to Install Snapshot, Leader:{}", logName(), getLeaderId());
ReplicatedLogEntry lastAppliedEntry = context.getReplicatedLog().get(context.getLastApplied());
long lastAppliedIndex = -1;
long lastAppliedTerm = -1;
if (lastAppliedEntry != null) {
lastAppliedIndex = lastAppliedEntry.getIndex();
lastAppliedTerm = lastAppliedEntry.getTerm();
- } else if (context.getReplicatedLog().getSnapshotIndex() > -1) {
+ } else if (context.getReplicatedLog().getSnapshotIndex() > -1) {
lastAppliedIndex = context.getReplicatedLog().getSnapshotIndex();
lastAppliedTerm = context.getReplicatedLog().getSnapshotTerm();
}
boolean isInstallSnapshotInitiated = true;
- actor().tell(new CaptureSnapshot(lastIndex(), lastTerm(),
- lastAppliedIndex, lastAppliedTerm, isInstallSnapshotInitiated),
- actor());
+ long replicatedToAllIndex = super.getReplicatedToAllIndex();
+ ReplicatedLogEntry replicatedToAllEntry = context.getReplicatedLog().get(replicatedToAllIndex);
+ actor().tell(new CaptureSnapshot(lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm,
+ (replicatedToAllEntry != null ? replicatedToAllEntry.getIndex() : -1),
+ (replicatedToAllEntry != null ? replicatedToAllEntry.getTerm() : -1),
+ isInstallSnapshotInitiated), actor());
context.setSnapshotCaptureInitiated(true);
}
}
private void sendInstallSnapshot() {
+ LOG.debug("{}: sendInstallSnapshot", logName());
for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
ActorSelection followerActor = context.getPeerActorSelection(e.getKey());
context.getReplicatedLog().getSnapshotIndex(),
context.getReplicatedLog().getSnapshotTerm(),
nextSnapshotChunk,
- followerToSnapshot.incrementChunkIndex(),
- followerToSnapshot.getTotalChunks(),
+ followerToSnapshot.incrementChunkIndex(),
+ followerToSnapshot.getTotalChunks(),
Optional.of(followerToSnapshot.getLastChunkHashCode())
).toSerializable(),
actor()
);
LOG.info("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}",
- context.getId(), followerActor.path(),
+ logName(), followerActor.path(),
followerToSnapshot.getChunkIndex(),
followerToSnapshot.getTotalChunks());
}
} catch (IOException e) {
- LOG.error("{}: InstallSnapshot failed for Leader.", context.getId(), e);
+ LOG.error("{}: InstallSnapshot failed for Leader.", logName(), e);
}
}
mapFollowerToSnapshot.put(followerId, followerToSnapshot);
}
ByteString nextChunk = followerToSnapshot.getNextChunk();
- if (LOG.isDebugEnabled()) {
- LOG.debug("{}: Leader's snapshot nextChunk size:{}", context.getId(), nextChunk.size());
- }
+
+ LOG.debug("{}: next snapshot chunk size for follower {}: {}", logName(), followerId, nextChunk.size());
+
return nextChunk;
}
private void sendHeartBeat() {
if (!followerToLog.isEmpty()) {
- sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toMillis());
+ LOG.trace("{}: Sending heartbeat", logName());
+ sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toMillis(), true);
}
}
((size % context.getConfigParams().getSnapshotChunkSize()) > 0 ? 1 : 0);
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Snapshot {} bytes, total chunks to send:{}",
- context.getId(), size, totalChunks);
+ logName(), size, totalChunks);
}
replyReceivedForOffset = -1;
chunkIndex = AbstractLeader.FIRST_CHUNK_INDEX;
}
}
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Next chunk: length={}, offset={},size={}", context.getId(),
+
+ LOG.debug("{}: Next chunk: length={}, offset={},size={}", logName(),
snapshotLength, start, size);
- }
+
ByteString substring = getSnapshotBytes().substring(start, start + size);
nextChunkHashCode = substring.hashCode();
return substring;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
+import org.opendaylight.controller.cluster.raft.RaftState;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.SerializationUtils;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
*/
protected String leaderId = null;
+ private long replicatedToAllIndex = -1;
- protected AbstractRaftActorBehavior(RaftActorContext context) {
+ private final String logName;
+
+ private final RaftState state;
+
+ protected AbstractRaftActorBehavior(RaftActorContext context, RaftState state) {
this.context = context;
+ this.state = state;
this.LOG = context.getLogger();
+
+ logName = String.format("%s (%s)", context.getId(), state);
+ }
+
+ @Override
+ public RaftState state() {
+ return state;
+ }
+
+ public String logName() {
+ return logName;
+ }
+
+ @Override
+ public void setReplicatedToAllIndex(long replicatedToAllIndex) {
+ this.replicatedToAllIndex = replicatedToAllIndex;
+ }
+
+ @Override
+ public long getReplicatedToAllIndex() {
+ return replicatedToAllIndex;
}
/**
if (appendEntries.getTerm() < currentTerm()) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Cannot append entries because sender term {} is less than {}",
- context.getId(), appendEntries.getTerm(), currentTerm());
+ logName(), appendEntries.getTerm(), currentTerm());
}
sender.tell(
* @param requestVote
* @return
*/
- protected RaftActorBehavior requestVote(ActorRef sender,
- RequestVote requestVote) {
+ protected RaftActorBehavior requestVote(ActorRef sender, RequestVote requestVote) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Received {}", context.getId(), requestVote);
- }
+ LOG.debug("{}: In requestVote: {}", logName(), requestVote);
boolean grantVote = false;
}
}
- sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor());
+ RequestVoteReply reply = new RequestVoteReply(currentTerm(), grantVote);
+
+ LOG.debug("{}: requestVote returning: {}", logName(), reply);
+
+ sender.tell(reply, actor());
return this;
}
// around as the rest wont be present either
LOG.warn(
"{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}",
- context.getId(), i, i, index);
+ logName(), i, i, index);
break;
}
}
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Setting last applied to {}", context.getId(), newLastApplied);
+ LOG.debug("{}: Setting last applied to {}", logName(), newLastApplied);
}
context.setLastApplied(newLastApplied);
}
protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) {
- LOG.info("{} :- Switching from behavior {} to {}", context.getId(), this.state(), behavior.state());
+ LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), behavior.state());
try {
close();
} catch (Exception e) {
- LOG.error("{}: Failed to close behavior : {}", context.getId(), this.state(), e);
+ LOG.error("{}: Failed to close behavior : {}", logName(), this.state(), e);
}
return behavior;
}
- protected long fakeSnapshot(final long minReplicatedToAllIndex, final long currentReplicatedIndex) {
+ /**
+ * Performs a snapshot with no capture on the replicated log.
+ * It clears the log from the supplied index or last-applied-1 which ever is minimum.
+ *
+ * @param snapshotCapturedIndex
+ */
+ protected void performSnapshotWithoutCapture(final long snapshotCapturedIndex) {
// we would want to keep the lastApplied as its used while capturing snapshots
- long tempMin = Math.min(minReplicatedToAllIndex,
- (context.getLastApplied() > -1 ? context.getLastApplied() - 1 : -1));
+ long lastApplied = context.getLastApplied();
+ long tempMin = Math.min(snapshotCapturedIndex, (lastApplied > -1 ? lastApplied - 1 : -1));
if (tempMin > -1 && context.getReplicatedLog().isPresent(tempMin)) {
- context.getReplicatedLog().snapshotPreCommit(tempMin, context.getTermInformation().getCurrentTerm());
+ LOG.debug("{}: fakeSnapshot purging log to {} for term {}", logName(), tempMin,
+ context.getTermInformation().getCurrentTerm());
+
+ //use the term of the temp-min, since we check for isPresent, entry will not be null
+ ReplicatedLogEntry entry = context.getReplicatedLog().get(tempMin);
+ context.getReplicatedLog().snapshotPreCommit(tempMin, entry.getTerm());
context.getReplicatedLog().snapshotCommit();
- return tempMin;
+ setReplicatedToAllIndex(tempMin);
}
- return currentReplicatedIndex;
}
+
}
private final Set<String> peers;
public Candidate(RaftActorContext context) {
- super(context);
+ super(context, RaftState.Candidate);
peers = context.getPeerAddresses().keySet();
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Election: Candidate has following peers: {}", context.getId(), peers);
+ LOG.debug("{}: Election: Candidate has following peers: {}", logName(), peers);
}
votesRequired = getMajorityVoteCount(peers.size());
AppendEntries appendEntries) {
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
+ LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
}
return this;
}
@Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
- RequestVoteReply requestVoteReply) {
+ RequestVoteReply requestVoteReply) {
+
+ LOG.debug("{}: handleRequestVoteReply: {}, current voteCount: {}", logName(), requestVoteReply,
+ voteCount);
if (requestVoteReply.isVoteGranted()) {
voteCount++;
return this;
}
- @Override public RaftState state() {
- return RaftState.Candidate;
- }
-
@Override
public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
RaftRPC rpc = (RaftRPC) message;
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: RaftRPC message received {} my term is {}", context.getId(), rpc,
+ LOG.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc,
context.getTermInformation().getCurrentTerm());
}
}
if (message instanceof ElectionTimeout) {
+ LOG.debug("{}: Received ElectionTimeout", logName());
+
if (votesRequired == 0) {
// If there are no peers then we should be a Leader
// We wait for the election timeout to occur before declare
// Increment the election term and vote for self
long currentTerm = context.getTermInformation().getCurrentTerm();
- context.getTermInformation().updateAndPersist(currentTerm + 1,
- context.getId());
+ long newTerm = currentTerm + 1;
+ context.getTermInformation().updateAndPersist(newTerm, context.getId());
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Starting new term {}", context.getId(), (currentTerm + 1));
- }
+ LOG.debug("{}: Starting new term {}", logName(), newTerm);
// Request for a vote
// TODO: Retry request for vote if replies do not arrive in a reasonable
for (String peerId : peers) {
ActorSelection peerActor = context.getPeerActorSelection(peerId);
if(peerActor != null) {
- peerActor.tell(new RequestVote(
+ RequestVote requestVote = new RequestVote(
context.getTermInformation().getCurrentTerm(),
context.getId(),
context.getReplicatedLog().lastIndex(),
- context.getReplicatedLog().lastTerm()),
- context.getActor()
- );
- }
- }
+ context.getReplicatedLog().lastTerm());
+ LOG.debug("{}: Sending {} to peer {}", logName(), requestVote, peerId);
+ peerActor.tell(requestVote, context.getActor());
+ }
+ }
}
@Override public void close() throws Exception {
private SnapshotTracker snapshotTracker = null;
public Follower(RaftActorContext context) {
- super(context);
+ super(context, RaftState.Follower);
scheduleElection(electionDuration());
}
@Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries) {
- if(appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
- }
+ int numLogEntries = appendEntries.getEntries() != null ? appendEntries.getEntries().size() : 0;
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("{}: handleAppendEntries: {}", logName(), appendEntries);
+ } else if(LOG.isDebugEnabled() && numLogEntries > 0) {
+ LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
}
// TODO : Refactor this method into a bunch of smaller methods
boolean outOfSync = true;
// First check if the logs are in sync or not
- if (lastIndex() == -1
- && appendEntries.getPrevLogIndex() != -1) {
+ long lastIndex = lastIndex();
+ if (lastIndex == -1 && appendEntries.getPrevLogIndex() != -1) {
// The follower's log is out of sync because the leader does have
// an entry at prevLogIndex and this follower has no entries in
// it's log.
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: The followers log is empty and the senders prevLogIndex is {}",
- context.getId(), appendEntries.getPrevLogIndex());
- }
-
- } else if (lastIndex() > -1
- && appendEntries.getPrevLogIndex() != -1
- && !prevEntryPresent) {
+ LOG.debug("{}: The followers log is empty and the senders prevLogIndex is {}",
+ logName(), appendEntries.getPrevLogIndex());
+ } else if (lastIndex > -1 && appendEntries.getPrevLogIndex() != -1 && !prevEntryPresent) {
// The follower's log is out of sync because the Leader's
// prevLogIndex entry was not found in it's log
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it",
- context.getId(), appendEntries.getPrevLogIndex());
- }
-
- } else if (lastIndex() > -1
- && prevEntryPresent
- && prevLogTerm != appendEntries.getPrevLogTerm()) {
+ LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it",
+ logName(), appendEntries.getPrevLogIndex());
+ } else if (lastIndex > -1 && prevEntryPresent && prevLogTerm != appendEntries.getPrevLogTerm()) {
// The follower's log is out of sync because the Leader's
// prevLogIndex entry does exist in the follower's log but it has
// a different term in it
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "{}: Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}"
- , context.getId(), prevLogTerm
- , appendEntries.getPrevLogTerm());
- }
+ LOG.debug(
+ "{}: Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}",
+ logName(), prevLogTerm, appendEntries.getPrevLogTerm());
} else {
outOfSync = false;
}
if (outOfSync) {
// We found that the log was out of sync so just send a negative
// reply and return
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Follower ({}) is out-of-sync, " +
- "so sending negative reply, lastIndex():{}, lastTerm():{}",
- context.getId(), context.getId(), lastIndex(), lastTerm()
- );
- }
- sender.tell(
- new AppendEntriesReply(context.getId(), currentTerm(), false,
- lastIndex(), lastTerm()), actor()
- );
+
+ LOG.debug("{}: Follower is out-of-sync, so sending negative reply, lastIndex: {}, lastTerm: {}",
+ logName(), lastIndex, lastTerm());
+
+ sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex,
+ lastTerm()), actor());
return this;
}
- if (appendEntries.getEntries() != null
- && appendEntries.getEntries().size() > 0) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Number of entries to be appended = {}", context.getId(),
+ if (appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) {
+
+ LOG.debug("{}: Number of entries to be appended = {}", logName(),
appendEntries.getEntries().size());
- }
// 3. If an existing entry conflicts with a new one (same index
// but different terms), delete the existing entry and all that
break;
}
- if (newEntry.getTerm() == matchEntry
- .getTerm()) {
+ if (newEntry.getTerm() == matchEntry.getTerm()) {
continue;
}
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Removing entries from log starting at {}", context.getId(),
+ LOG.debug("{}: Removing entries from log starting at {}", logName(),
matchEntry.getIndex());
- }
// Entries do not match so remove all subsequent entries
- context.getReplicatedLog()
- .removeFromAndPersist(matchEntry.getIndex());
+ context.getReplicatedLog().removeFromAndPersist(matchEntry.getIndex());
break;
}
}
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: After cleanup entries to be added from = {}", context.getId(),
- (addEntriesFrom + lastIndex()));
- }
+ lastIndex = lastIndex();
+ LOG.debug("{}: After cleanup entries to be added from = {}", logName(),
+ (addEntriesFrom + lastIndex));
// 4. Append any new entries not already in the log
- for (int i = addEntriesFrom;
- i < appendEntries.getEntries().size(); i++) {
+ for (int i = addEntriesFrom; i < appendEntries.getEntries().size(); i++) {
+ ReplicatedLogEntry entry = appendEntries.getEntries().get(i);
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Append entry to log {}", context.getId(),
- appendEntries.getEntries().get(i).getData());
- }
- context.getReplicatedLog().appendAndPersist(appendEntries.getEntries().get(i));
- }
+ LOG.debug("{}: Append entry to log {}", logName(), entry.getData());
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Log size is now {}", context.getId(), context.getReplicatedLog().size());
+ context.getReplicatedLog().appendAndPersist(entry);
}
- }
+ LOG.debug("{}: Log size is now {}", logName(), context.getReplicatedLog().size());
+ }
// 5. If leaderCommit > commitIndex, set commitIndex =
// min(leaderCommit, index of last new entry)
+ lastIndex = lastIndex();
long prevCommitIndex = context.getCommitIndex();
- context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(),
- context.getReplicatedLog().lastIndex()));
+ context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(), lastIndex));
if (prevCommitIndex != context.getCommitIndex()) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Commit index set to {}", context.getId(), context.getCommitIndex());
- }
+ LOG.debug("{}: Commit index set to {}", logName(), context.getCommitIndex());
}
// If commitIndex > lastApplied: increment lastApplied, apply
// log[lastApplied] to state machine (§5.3)
// check if there are any entries to be applied. last-applied can be equal to last-index
if (appendEntries.getLeaderCommit() > context.getLastApplied() &&
- context.getLastApplied() < lastIndex()) {
+ context.getLastApplied() < lastIndex) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: applyLogToStateMachine, " +
- "appendEntries.getLeaderCommit():{}," +
- "context.getLastApplied():{}, lastIndex():{}", context.getId(),
- appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex()
- );
+ "appendEntries.getLeaderCommit(): {}," +
+ "context.getLastApplied(): {}, lastIndex(): {}", logName(),
+ appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex);
}
applyLogToStateMachine(appendEntries.getLeaderCommit());
}
- sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true,
- lastIndex(), lastTerm()), actor());
+ AppendEntriesReply reply = new AppendEntriesReply(context.getId(), currentTerm(), true,
+ lastIndex, lastTerm());
+
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("{}: handleAppendEntries returning : {}", logName(), reply);
+ } else if(LOG.isDebugEnabled() && numLogEntries > 0) {
+ LOG.debug("{}: handleAppendEntries returning : {}", logName(), reply);
+ }
+
+ sender.tell(reply, actor());
if (!context.isSnapshotCaptureInitiated()) {
- fakeSnapshot(appendEntries.getReplicatedToAllIndex(), appendEntries.getReplicatedToAllIndex());
+ super.performSnapshotWithoutCapture(appendEntries.getReplicatedToAllIndex());
}
return this;
return this;
}
- @Override public RaftState state() {
- return RaftState.Follower;
- }
-
@Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
Object message = fromSerializableMessage(originalMessage);
// set currentTerm = T, convert to follower (§5.1)
// This applies to all RPC messages and responses
if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
+ LOG.debug("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term",
+ logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
+
context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
}
}
if (message instanceof ElectionTimeout) {
+ LOG.debug("{}: Received ElectionTimeout - switching to Candidate", logName());
return switchBehavior(new Candidate(context));
} else if (message instanceof InstallSnapshot) {
private void handleInstallSnapshot(ActorRef sender, InstallSnapshot installSnapshot) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: InstallSnapshot received by follower " +
- "datasize:{} , Chunk:{}/{}", context.getId(), installSnapshot.getData().size(),
- installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks()
- );
- }
+
+ LOG.debug("{}: InstallSnapshot received from leader {}, datasize: {} , Chunk: {}/{}",
+ logName(), installSnapshot.getLeaderId(), installSnapshot.getData().size(),
+ installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks());
if(snapshotTracker == null){
snapshotTracker = new SnapshotTracker(LOG, installSnapshot.getTotalChunks());
}
- sender.tell(new InstallSnapshotReply(
- currentTerm(), context.getId(), installSnapshot.getChunkIndex(),
- true), actor());
+ InstallSnapshotReply reply = new InstallSnapshotReply(
+ currentTerm(), context.getId(), installSnapshot.getChunkIndex(), true);
+
+ LOG.debug("{}: handleInstallSnapshot returning: {}", logName(), reply);
+
+ sender.tell(reply, actor());
} catch (SnapshotTracker.InvalidChunkException e) {
+ LOG.debug("{}: Exception in InstallSnapshot of follower", logName(), e);
sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
-1, false), actor());
snapshotTracker = null;
} catch (Exception e){
- LOG.error("{}: Exception in InstallSnapshot of follower", context.getId(), e);
+ LOG.error("{}: Exception in InstallSnapshot of follower", logName(), e);
+
//send reply with success as false. The chunk will be sent again on failure
sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
installSnapshot.getChunkIndex(), false), actor());
ByteString getSnapshotChunksCollected(){
return snapshotTracker != null ? snapshotTracker.getCollectedChunks() : ByteString.EMPTY;
}
-
-
}
* @return
*/
String getLeaderId();
+
+ /**
+ * setting the index of the log entry which is replicated to all nodes
+ * @param replicatedToAllIndex
+ */
+ void setReplicatedToAllIndex(long replicatedToAllIndex);
+
+ /**
+ * getting the index of the log entry which is replicated to all nodes
+ * @return
+ */
+ long getReplicatedToAllIndex();
}
return replicatedToAllIndex;
}
+
@Override
public String toString() {
- final StringBuilder sb =
- new StringBuilder("AppendEntries{");
- sb.append("term=").append(getTerm());
- sb.append("leaderId='").append(leaderId).append('\'');
- sb.append(", prevLogIndex=").append(prevLogIndex);
- sb.append(", prevLogTerm=").append(prevLogTerm);
- sb.append(", entries=").append(entries);
- sb.append(", leaderCommit=").append(leaderCommit);
- sb.append(", replicatedToAllIndex=").append(replicatedToAllIndex);
- sb.append('}');
- return sb.toString();
+ StringBuilder builder = new StringBuilder();
+ builder.append("AppendEntries [term=").append(term).append(", leaderId=").append(leaderId)
+ .append(", prevLogIndex=").append(prevLogIndex).append(", prevLogTerm=").append(prevLogTerm)
+ .append(", entries=").append(entries).append(", leaderCommit=").append(leaderCommit)
+ .append(", replicatedToAllIndex=").append(replicatedToAllIndex).append("]");
+ return builder.toString();
}
public <T extends Object> Object toSerializable() {
return followerId;
}
- @Override public String toString() {
- final StringBuilder sb =
- new StringBuilder("AppendEntriesReply{");
- sb.append("term=").append(term);
- sb.append(", success=").append(success);
- sb.append(", logLastIndex=").append(logLastIndex);
- sb.append(", logLastTerm=").append(logLastTerm);
- sb.append(", followerId='").append(followerId).append('\'');
- sb.append('}');
- return sb.toString();
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("AppendEntriesReply [term=").append(term).append(", success=").append(success)
+ .append(", logLastIndex=").append(logLastIndex).append(", logLastTerm=").append(logLastTerm)
+ .append(", followerId=").append(followerId).append("]");
+ return builder.toString();
}
}
return installSnapshot;
}
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("InstallSnapshot [term=").append(term).append(", leaderId=").append(leaderId)
+ .append(", lastIncludedIndex=").append(lastIncludedIndex).append(", lastIncludedTerm=")
+ .append(lastIncludedTerm).append(", data=").append(data).append(", chunkIndex=").append(chunkIndex)
+ .append(", totalChunks=").append(totalChunks).append(", lastChunkHashCode=").append(lastChunkHashCode)
+ .append("]");
+ return builder.toString();
+ }
}
public boolean isSuccess() {
return success;
}
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("InstallSnapshotReply [term=").append(term).append(", followerId=").append(followerId)
+ .append(", chunkIndex=").append(chunkIndex).append(", success=").append(success).append("]");
+ return builder.toString();
+ }
}
this.lastLogTerm = lastLogTerm;
}
- @Override public String toString() {
- final StringBuilder sb =
- new StringBuilder("RequestVote{");
- sb.append("term='").append(getTerm()).append('\'');
- sb.append("candidateId='").append(candidateId).append('\'');
- sb.append(", lastLogIndex=").append(lastLogIndex);
- sb.append(", lastLogTerm=").append(lastLogTerm);
- sb.append('}');
- return sb.toString();
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("RequestVote [term=").append(term).append(", candidateId=").append(candidateId)
+ .append(", lastLogIndex=").append(lastLogIndex).append(", lastLogTerm=").append(lastLogTerm)
+ .append("]");
+ return builder.toString();
}
}
public boolean isVoteGranted() {
return voteGranted;
}
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("RequestVoteReply [term=").append(term).append(", voteGranted=").append(voteGranted).append("]");
+ return builder.toString();
+ }
}
import org.junit.Test;
import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockReplicatedLogEntry;
+
/**
*
*/
@Test
public void testSnapshotPreCommit() {
+ //add 4 more entries
replicatedLogImpl.append(new MockReplicatedLogEntry(2, 4, new MockPayload("E")));
replicatedLogImpl.append(new MockReplicatedLogEntry(2, 5, new MockPayload("F")));
replicatedLogImpl.append(new MockReplicatedLogEntry(3, 6, new MockPayload("G")));
replicatedLogImpl.append(new MockReplicatedLogEntry(3, 7, new MockPayload("H")));
+ //sending negative values should not cause any changes
+ replicatedLogImpl.snapshotPreCommit(-1, -1);
+ assertEquals(8, replicatedLogImpl.size());
+ assertEquals(-1, replicatedLogImpl.getSnapshotIndex());
+
replicatedLogImpl.snapshotPreCommit(4, 3);
assertEquals(3, replicatedLogImpl.size());
assertEquals(4, replicatedLogImpl.getSnapshotIndex());
assertEquals(0, replicatedLogImpl.size());
assertEquals(7, replicatedLogImpl.getSnapshotIndex());
+ }
+
+ @Test
+ public void testIsPresent() {
+ assertTrue(replicatedLogImpl.isPresent(0));
+ assertTrue(replicatedLogImpl.isPresent(1));
+ assertTrue(replicatedLogImpl.isPresent(2));
+ assertTrue(replicatedLogImpl.isPresent(3));
+
+ replicatedLogImpl.append(new MockReplicatedLogEntry(2, 4, new MockPayload("D")));
+ replicatedLogImpl.snapshotPreCommit(3, 2); //snapshot on 3
+ replicatedLogImpl.snapshotCommit();
+
+ assertFalse(replicatedLogImpl.isPresent(0));
+ assertFalse(replicatedLogImpl.isPresent(1));
+ assertFalse(replicatedLogImpl.isPresent(2));
+ assertFalse(replicatedLogImpl.isPresent(3));
+ assertTrue(replicatedLogImpl.isPresent(4));
+
+ replicatedLogImpl.snapshotPreCommit(4, 2); //snapshot on 4
+ replicatedLogImpl.snapshotCommit();
+ assertFalse(replicatedLogImpl.isPresent(4));
+ replicatedLogImpl.append(new MockReplicatedLogEntry(2, 5, new MockPayload("D")));
+ assertTrue(replicatedLogImpl.isPresent(5));
}
// create a snapshot for test
new MockRaftActorContext.MockPayload("C"),
new MockRaftActorContext.MockPayload("D")));
- mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1));
+ mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1,-1, 1, -1, 1));
RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
mockRaftActor.setCurrentBehavior(new Follower(raftActorContext));
- mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, 2, 1));
+ long replicatedToAllIndex = 1;
+ mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, 2, 1, replicatedToAllIndex, 1));
verify(mockRaftActor.delegate).createSnapshot();
verify(dataPersistenceProvider).deleteMessages(100);
- assertEquals(2, mockRaftActor.getReplicatedLog().size());
+ assertEquals(3, mockRaftActor.getReplicatedLog().size());
+ assertEquals(1, mockRaftActor.getCurrentBehavior().getReplicatedToAllIndex());
+ assertNotNull(mockRaftActor.getReplicatedLog().get(2));
assertNotNull(mockRaftActor.getReplicatedLog().get(3));
assertNotNull(mockRaftActor.getReplicatedLog().get(4));
// Index 2 will not be in the log because it was removed due to snapshotting
- assertNull(mockRaftActor.getReplicatedLog().get(2));
+ assertNull(mockRaftActor.getReplicatedLog().get(1));
+ assertNull(mockRaftActor.getReplicatedLog().get(0));
}
};
mockRaftActor.setCurrentBehavior(new Leader(raftActorContext));
- mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1));
+ mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1, -1, 1));
mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
assertEquals(8, leaderActor.getReplicatedLog().size());
- leaderActor.onReceiveCommand(new CaptureSnapshot(6, 1, 4, 1));
+ leaderActor.onReceiveCommand(new CaptureSnapshot(6, 1, 4, 1, 4, 1));
+
leaderActor.getRaftActorContext().setSnapshotCaptureInitiated(true);
verify(leaderActor.delegate).createSnapshot();
followerActor.waitForInitializeBehaviorComplete();
- // create 8 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot
+
Follower follower = new Follower(followerActor.getRaftActorContext());
followerActor.setCurrentBehavior(follower);
assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
+ // create 6 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot
MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
followerActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(0, 6, 1).build());
- // log as indices 0-5
+ // log has indices 0-5
assertEquals(6, followerActor.getReplicatedLog().size());
//snapshot on 4
- followerActor.onReceiveCommand(new CaptureSnapshot(5, 1, 4, 1));
+ followerActor.onReceiveCommand(new CaptureSnapshot(5, 1, 4, 1, 4, 1));
+
followerActor.getRaftActorContext().setSnapshotCaptureInitiated(true);
verify(followerActor.delegate).createSnapshot();
followerActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
assertFalse(followerActor.getRaftActorContext().isSnapshotCaptureInitiated());
- // capture snapshot reply should remove the snapshotted entries only
+ // capture snapshot reply should remove the snapshotted entries only till replicatedToAllIndex
assertEquals(3, followerActor.getReplicatedLog().size()); //indexes 5,6,7 left in the log
assertEquals(7, followerActor.getReplicatedLog().lastIndex());
// create 5 entries in the log
MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
leaderActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(5, 10, 1).build());
+
//set the snapshot index to 4 , 0 to 4 are snapshotted
leaderActor.getRaftActorContext().getReplicatedLog().setSnapshotIndex(4);
+ //setting replicatedToAllIndex = 9, for the log to clear
+ leader.setReplicatedToAllIndex(9);
assertEquals(5, leaderActor.getReplicatedLog().size());
+ assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1));
assertEquals(5, leaderActor.getReplicatedLog().size());
+ assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
// set the 2nd follower nextIndex to 1 which has been snapshotted
leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 0, 1));
assertEquals(5, leaderActor.getReplicatedLog().size());
+ assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
// simulate a real snapshot
leaderActor.onReceiveCommand(new SendHeartBeat());
leaderActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
assertFalse(leaderActor.getRaftActorContext().isSnapshotCaptureInitiated());
- assertEquals("Real snapshot didn't clear the log till lastApplied", 0, leaderActor.getReplicatedLog().size());
+ assertEquals("Real snapshot didn't clear the log till replicatedToAllIndex", 0, leaderActor.getReplicatedLog().size());
//reply from a slow follower after should not raise errors
leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1));
package org.opendaylight.controller.cluster.raft.behaviors;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
+import java.util.ArrayList;
+import java.util.List;
import org.junit.Test;
import org.opendaylight.controller.cluster.raft.AbstractActorTest;
import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest {
private final ActorRef behaviorActor = getSystem().actorOf(Props.create(
}
@Test
- public void testFakeSnapshots() {
+ public void testPerformSnapshot() {
MockRaftActorContext context = new MockRaftActorContext("test", getSystem(), behaviorActor);
- AbstractRaftActorBehavior behavior = new Leader(context);
- context.getTermInformation().update(1, "leader");
+ AbstractRaftActorBehavior abstractBehavior = (AbstractRaftActorBehavior) createBehavior(context);
+ if (abstractBehavior instanceof Candidate) {
+ return;
+ }
- //entry with 1 index=0 entry with replicatedToAllIndex = 0, does not do anything, returns the
+ context.getTermInformation().update(1, "test");
+
+ //log has 1 entry with replicatedToAllIndex = 0, does not do anything, returns the
context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build());
context.setLastApplied(0);
- assertEquals(-1, behavior.fakeSnapshot(0, -1));
+ abstractBehavior.performSnapshotWithoutCapture(0);
+ assertEquals(-1, abstractBehavior.getReplicatedToAllIndex());
assertEquals(1, context.getReplicatedLog().size());
//2 entries, lastApplied still 0, no purging.
- context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0,2,1).build());
+ context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build());
context.setLastApplied(0);
- assertEquals(-1, behavior.fakeSnapshot(0, -1));
+ abstractBehavior.performSnapshotWithoutCapture(0);
+ assertEquals(-1, abstractBehavior.getReplicatedToAllIndex());
assertEquals(2, context.getReplicatedLog().size());
//2 entries, lastApplied still 0, no purging.
- context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0,2,1).build());
+ context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build());
context.setLastApplied(1);
- assertEquals(0, behavior.fakeSnapshot(0, -1));
+ abstractBehavior.performSnapshotWithoutCapture(0);
+ assertEquals(0, abstractBehavior.getReplicatedToAllIndex());
assertEquals(1, context.getReplicatedLog().size());
//5 entries, lastApplied =2 and replicatedIndex = 3, but since we want to keep the lastapplied, indices 0 and 1 will only get purged
- context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0,5,1).build());
+ context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 5, 1).build());
context.setLastApplied(2);
- assertEquals(1, behavior.fakeSnapshot(3, 1));
+ abstractBehavior.performSnapshotWithoutCapture(3);
+ assertEquals(1, abstractBehavior.getReplicatedToAllIndex());
assertEquals(3, context.getReplicatedLog().size());
-
+ // scenario where Last applied > Replicated to all index (becoz of a slow follower)
+ context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
+ context.setLastApplied(2);
+ abstractBehavior.performSnapshotWithoutCapture(1);
+ assertEquals(1, abstractBehavior.getReplicatedToAllIndex());
+ assertEquals(1, context.getReplicatedLog().size());
}
+
protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(
ActorRef actorRef, RaftRPC rpc) {
protected final Logger LOG = LoggerFactory.getLogger(getClass());
public AbstractUntypedPersistentActor() {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Actor created {}", getSelf());
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Actor created {}", getSelf());
}
getContext().
system().
@Override public void onReceiveCommand(Object message) throws Exception {
final String messageType = message.getClass().getSimpleName();
- if(LOG.isDebugEnabled()) {
- LOG.debug("Received message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Received message {}", messageType);
}
handleCommand(message);
- if(LOG.isDebugEnabled()) {
- LOG.debug("Done handling message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Done handling message {}", messageType);
}
}
@Override public void onReceiveRecover(Object message) throws Exception {
final String messageType = message.getClass().getSimpleName();
- if(LOG.isDebugEnabled()) {
- LOG.debug("Received message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Received message {}", messageType);
}
handleRecover(message);
- if(LOG.isDebugEnabled()) {
- LOG.debug("Done handling message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Done handling message {}", messageType);
}
}
@Override
public void onReceiveCommand(final Object message) throws Exception {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: onReceiveCommand: Received message {} from {}", persistenceId(), message, getSender());
- }
-
if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
handleCreateTransaction(message);
} else if(message instanceof ForwardedReadyTransaction) {
sender().tell((returnSerialized ? readDataReply.toSerializable(clientTxVersion): readDataReply), self());
} catch (Exception e) {
- LOG.error(String.format("Unexpected error reading path %s", path), e);
+ LOG.debug(String.format("Unexpected error reading path %s", path), e);
shardStats.incrementFailedReadTransactionsCount();
sender().tell(new akka.actor.Status.Failure(e), self());
}
NormalizedNode<?,?> expectedRoot = readStore(shard, YangInstanceIdentifier.builder().build());
- CaptureSnapshot capture = new CaptureSnapshot(-1, -1, -1, -1);
+ CaptureSnapshot capture = new CaptureSnapshot(-1, -1, -1, -1, -1, -1);
shard.tell(capture, getRef());
assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
*/
package org.opendaylight.controller.md.sal.dom.api;
+import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import java.util.Objects;
import javax.annotation.Nonnull;
@Override
public final String toString() {
- return com.google.common.base.Objects.toStringHelper(this).omitNullValues().add("type", type).add("contextReference", getContextReference()).toString();
+ return MoreObjects.toStringHelper(this).omitNullValues().add("type", type).add("contextReference", getContextReference()).toString();
}
}
package org.opendaylight.controller.sal.core.api;
import java.util.concurrent.Future;
-
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.opendaylight.yangtools.yang.data.api.CompositeNode;
+/**
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcService} instead.
+ */
+@Deprecated
public interface RpcConsumptionRegistry {
/**
* Sends an RPC to other components registered to the broker.
*/
package org.opendaylight.controller.sal.core.api;
+import com.google.common.util.concurrent.ListenableFuture;
import java.util.Set;
-
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.opendaylight.yangtools.yang.data.api.CompositeNode;
-import com.google.common.util.concurrent.ListenableFuture;
-
/**
* {@link Provider}'s implementation of an RPC.
*
* {@link RpcResult}
* <li> {@link Broker} returns the {@link RpcResult} to {@link Consumer}
* </ol>
+ *
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementation} instead.
*/
+@Deprecated
public interface RpcImplementation extends Provider.ProviderFunctionality {
/**
/**
* Exception reported when no RPC implementation is found in the system.
+ *
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationNotAvailableException} instead.
*/
+@Deprecated
public class RpcImplementationUnavailableException extends RuntimeException {
private static final long serialVersionUID = 1L;
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+/**
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcProviderService} and {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcService} instead.
+ */
+@Deprecated
public interface RpcProvisionRegistry extends RpcImplementation, BrokerService, RouteChangePublisher<RpcRoutingContext, YangInstanceIdentifier>, DOMService {
/**
package org.opendaylight.controller.sal.core.api;
import java.util.EventListener;
-
import org.opendaylight.yangtools.yang.common.QName;
+/**
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcAvailabilityListener} instead.
+ */
+@Deprecated
public interface RpcRegistrationListener extends EventListener {
public void onRpcImplementationAdded(QName name);
*/
package org.opendaylight.controller.md.sal.dom.broker.impl;
-import com.google.common.base.Objects;
-import com.google.common.base.Objects.ToStringHelper;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.MoreObjects.ToStringHelper;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.FutureCallback;
@Override
public String toString() {
- return addToStringAttributes(Objects.toStringHelper(this)).toString();
+ return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
}
protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.japi.Creator;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
private final Configuration configuration;
private final String followerId;
private final Logger LOG = LoggerFactory.getLogger(DummyShard.class);
+ private long lastMessageIndex = -1;
+ private long lastMessageSize = 0;
public DummyShard(Configuration configuration, String followerId) {
this.configuration = configuration;
}
protected void handleAppendEntries(AppendEntries req) throws InterruptedException {
- LOG.info("{} - Received AppendEntries message : leader term, index, size = {}, {}, {}", followerId, req.getTerm(),req.getLeaderCommit(), req.getEntries().size());
+
+ LOG.info("{} - Received AppendEntries message : leader term = {}, index = {}, prevLogIndex = {}, size = {}",
+ followerId, req.getTerm(),req.getLeaderCommit(), req.getPrevLogIndex(), req.getEntries().size());
+
+ if(lastMessageIndex == req.getLeaderCommit() && req.getEntries().size() > 0 && lastMessageSize > 0){
+ LOG.error("{} - Duplicate message with leaderCommit = {} prevLogIndex = {} received", followerId, req.getLeaderCommit(), req.getPrevLogIndex());
+ }
+
+ lastMessageIndex = req.getLeaderCommit();
+ lastMessageSize = req.getEntries().size();
+
long lastIndex = req.getLeaderCommit();
- if (req.getEntries().size() > 0)
- lastIndex = req.getEntries().get(0).getIndex();
+ if (req.getEntries().size() > 0) {
+ for(ReplicatedLogEntry entry : req.getEntries()) {
+ lastIndex = entry.getIndex();
+ }
+ }
- if (configuration.shouldCauseTrouble()) {
+ if (configuration.shouldCauseTrouble() && req.getEntries().size() > 0) {
boolean ignore = false;
if (configuration.shouldDropReplies()) {
--- /dev/null
+org.slf4j.simpleLogger.showDateTime=true
+org.slf4j.simpleLogger.dateTimeFormat=hh:mm:ss,S a
+org.slf4j.simpleLogger.logFile=System.out
+org.slf4j.simpleLogger.showShortLogName=true
+org.slf4j.simpleLogger.levelInBrackets=true
+org.slf4j.simpleLogger.defaultLogLevel=info
\ No newline at end of file
*/
package org.opendaylight.controller.md.sal.dom.store.impl;
-import com.google.common.base.Objects;
-import com.google.common.base.Objects.ToStringHelper;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.MoreObjects.ToStringHelper;
import com.google.common.base.Preconditions;
-
import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
import org.slf4j.Logger;
@Override
public final String toString() {
- return addToStringAttributes(Objects.toStringHelper(this)).toString();
+ return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
}
/**
package org.opendaylight.controller.md.sal.dom.store.impl;
import static com.google.common.base.Preconditions.checkState;
-import com.google.common.base.Objects.ToStringHelper;
+import com.google.common.base.MoreObjects.ToStringHelper;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import java.util.Dictionary;
import java.util.Hashtable;
+import org.opendaylight.controller.netconf.api.util.NetconfConstants;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
import org.opendaylight.yangtools.yang.model.api.SchemaContextProvider;
import org.osgi.framework.BundleActivator;
NetconfOperationServiceFactoryImpl factory = new NetconfOperationServiceFactoryImpl(yangStoreService);
LOG.debug("Registering into OSGi");
Dictionary<String, String> properties = new Hashtable<>();
- properties.put("name", "config-netconf-connector");
+ properties.put(NetconfConstants.SERVICE_NAME, NetconfConstants.CONFIG_NETCONF_CONNECTOR);
osgiRegistration = context.registerService(NetconfOperationServiceFactory.class, factory, properties);
}
}
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>netconf-subsystem</artifactId>
+ <version>0.3.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>mdsal-netconf-connector</artifactId>
+ <packaging>bundle</packaging>
+ <name>${project.artifactId}</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>netconf-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>netconf-mapping-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>netconf-util</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-data-impl</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>sal-core-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>commons.logback_settings</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>mockito-configuration</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>sal-core-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>config-util</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-data-operations</artifactId>
+ <version>0.7.0-SNAPSHOT</version>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <configuration>
+ <instructions>
+ <Import-Package>*</Import-Package>
+ </instructions>
+ </configuration>
+ </plugin>
+ <!--FIXME extract yang plugin definition into parent-->
+ <plugin>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>config</id>
+ <goals>
+ <goal>generate-sources</goal>
+ </goals>
+ <configuration>
+ <codeGenerators>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator</codeGeneratorClass>
+ <outputBaseDir>${jmxGeneratorPath}</outputBaseDir>
+ <additionalConfiguration>
+ <namespaceToPackage1>urn:opendaylight:params:xml:ns:yang:controller==org.opendaylight.controller.config.yang</namespaceToPackage1>
+ </additionalConfiguration>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.maven.sal.api.gen.plugin.CodeGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>${salGeneratorPath}</outputBaseDir>
+ </generator>
+ </codeGenerators>
+ <inspectDependencies>true</inspectDependencies>
+ </configuration>
+ </execution>
+ </executions>
+ <dependencies>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>yang-jmx-generator-plugin</artifactId>
+ <version>${config.version}</version>
+ </dependency>
+ </dependencies>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.config.yang.netconf.mdsal.mapper;
+
+import org.opendaylight.controller.netconf.mdsal.connector.MdsalNetconfOperationServiceFactory;
+
+public class NetconfMdsalMapperModule extends org.opendaylight.controller.config.yang.netconf.mdsal.mapper.AbstractNetconfMdsalMapperModule {
+ public NetconfMdsalMapperModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+ super(identifier, dependencyResolver);
+ }
+
+ public NetconfMdsalMapperModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.netconf.mdsal.mapper.NetconfMdsalMapperModule oldModule, java.lang.AutoCloseable oldInstance) {
+ super(identifier, dependencyResolver, oldModule, oldInstance);
+ }
+
+ @Override
+ public void customValidation() {
+ // add custom validation form module attributes here.
+ }
+
+ @Override
+ public java.lang.AutoCloseable createInstance() {
+ return new MdsalNetconfOperationServiceFactory(getRootSchemaServiceDependency(), getDomBrokerDependency());
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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
+ */
+
+/*
+* Generated file
+*
+* Generated from: yang module name: netconf-mdsal-mapper yang module local name: netconf-mdsal-mapper
+* Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
+* Generated at: Wed Jan 14 14:58:42 CET 2015
+*
+* Do not modify this file unless it is present under src/main directory
+*/
+package org.opendaylight.controller.config.yang.netconf.mdsal.mapper;
+public class NetconfMdsalMapperModuleFactory extends org.opendaylight.controller.config.yang.netconf.mdsal.mapper.AbstractNetconfMdsalMapperModuleFactory {
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector;
+
+import com.google.common.base.Preconditions;
+import java.util.concurrent.atomic.AtomicReference;
+import org.opendaylight.controller.sal.core.api.model.SchemaService;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
+
+public class CurrentSchemaContext implements SchemaContextListener, AutoCloseable {
+ final AtomicReference<SchemaContext> currentContext = new AtomicReference<SchemaContext>();
+ private final ListenerRegistration<SchemaContextListener> schemaContextListenerListenerRegistration;
+
+ public SchemaContext getCurrentContext() {
+ Preconditions.checkState(currentContext.get() != null, "Current context not received");
+ return currentContext.get();
+ }
+
+ public CurrentSchemaContext(final SchemaService schemaService) {
+ schemaContextListenerListenerRegistration = schemaService.registerSchemaContextListener(this);
+ }
+
+ @Override
+ public void onGlobalContextUpdated(final SchemaContext schemaContext) {
+ currentContext.set(schemaContext);
+ }
+
+ @Override
+ public void close() throws Exception {
+ schemaContextListenerListenerRegistration.close();
+ currentContext.set(null);
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector;
+
+import com.google.common.base.Optional;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
+import org.opendaylight.controller.netconf.mapping.api.Capability;
+import org.opendaylight.controller.netconf.mapping.api.NetconfOperation;
+import org.opendaylight.controller.netconf.mapping.api.NetconfOperationService;
+import org.opendaylight.yangtools.yang.common.SimpleDateFormatUtil;
+import org.opendaylight.yangtools.yang.model.api.Module;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MdsalNetconfOperationService implements NetconfOperationService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MdsalNetconfOperationService.class);
+
+ private final CurrentSchemaContext schemaContext;
+ private final String netconfSessionIdForReporting;
+ private final OperationProvider operationProvider;
+
+ public MdsalNetconfOperationService(final CurrentSchemaContext schemaContext, final String netconfSessionIdForReporting,
+ final DOMDataBroker dataBroker) {
+ this.schemaContext = schemaContext;
+ // TODO schema contexts are different in data broker and the one we receive here ... the one received here should be updated same way as broker is
+ this.netconfSessionIdForReporting = netconfSessionIdForReporting;
+ this.operationProvider = new OperationProvider(netconfSessionIdForReporting, schemaContext, dataBroker);
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ // TODO does this get called dynamically ?
+ @Override
+ public Set<Capability> getCapabilities() {
+ final Set<Capability> capabilities = new HashSet<>();
+ // [RFC6241] 8.3. Candidate Configuration Capability
+ capabilities.add(new BasicCapability("urn:ietf:params:netconf:capability:candidate:1.0"));
+
+ final SchemaContext currentContext = schemaContext.getCurrentContext();
+ final Set<Module> modules = currentContext.getModules();
+ for (final Module module : modules) {
+ if(currentContext.getModuleSource(module).isPresent()) {
+ capabilities.add(new YangStoreCapability(module, currentContext.getModuleSource(module).get()));
+ } else {
+ LOG.warn("Missing source for module {}. This module will not be available from netconf server for session {}",
+ module, netconfSessionIdForReporting);
+ }
+ }
+
+ return capabilities;
+ }
+
+ @Override
+ public Set<NetconfOperation> getNetconfOperations() {
+ return operationProvider.getOperations();
+ }
+
+ // TODO reuse from netconf impl
+ private static class BasicCapability implements Capability {
+
+ private final String capability;
+
+ private BasicCapability(final String capability) {
+ this.capability = capability;
+ }
+
+ @Override
+ public String getCapabilityUri() {
+ return capability;
+ }
+
+ @Override
+ public Optional<String> getModuleNamespace() {
+ return Optional.absent();
+ }
+
+ @Override
+ public Optional<String> getModuleName() {
+ return Optional.absent();
+ }
+
+ @Override
+ public Optional<String> getRevision() {
+ return Optional.absent();
+ }
+
+ @Override
+ public Optional<String> getCapabilitySchema() {
+ return Optional.absent();
+ }
+
+ @Override
+ public Collection<String> getLocation() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public String toString() {
+ return capability;
+ }
+ }
+
+ private static final class YangStoreCapability extends BasicCapability {
+
+ private final String content;
+ private final String revision;
+ private final String moduleName;
+ private final String moduleNamespace;
+
+ public YangStoreCapability(final Module module, final String moduleContent) {
+ super(toCapabilityURI(module));
+ this.content = moduleContent;
+ this.moduleName = module.getName();
+ this.moduleNamespace = module.getNamespace().toString();
+ this.revision = SimpleDateFormatUtil.getRevisionFormat().format(module.getRevision());
+ }
+
+ @Override
+ public Optional<String> getCapabilitySchema() {
+ return Optional.of(content);
+ }
+
+ private static String toCapabilityURI(final Module module) {
+ return String.valueOf(module.getNamespace()) + "?module="
+ + module.getName() + "&revision=" + SimpleDateFormatUtil.getRevisionFormat().format(module.getRevision());
+ }
+
+ @Override
+ public Optional<String> getModuleName() {
+ return Optional.of(moduleName);
+ }
+
+ @Override
+ public Optional<String> getModuleNamespace() {
+ return Optional.of(moduleNamespace);
+ }
+
+ @Override
+ public Optional<String> getRevision() {
+ return Optional.of(revision);
+ }
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector;
+
+import com.google.common.base.Preconditions;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
+import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
+import org.opendaylight.controller.sal.core.api.model.SchemaService;
+
+public class MdsalNetconfOperationServiceFactory implements NetconfOperationServiceFactory, AutoCloseable {
+
+ private final DOMDataBroker dataBroker;
+ private final CurrentSchemaContext currentSchemaContext;
+
+ public MdsalNetconfOperationServiceFactory(final SchemaService schemaService, final DOMDataBroker domDataBroker) {
+ this.currentSchemaContext = new CurrentSchemaContext(Preconditions.checkNotNull(schemaService));
+ this.dataBroker = Preconditions.checkNotNull(domDataBroker);
+ }
+
+ @Override
+ public MdsalNetconfOperationService createService(final String netconfSessionIdForReporting) {
+ return new MdsalNetconfOperationService(currentSchemaContext, netconfSessionIdForReporting, dataBroker);
+ }
+
+ @Override
+ public void close() throws Exception {
+ currentSchemaContext.close();
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector;
+
+import com.google.common.collect.Sets;
+import java.util.Set;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
+import org.opendaylight.controller.netconf.mapping.api.NetconfOperation;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.Commit;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.DiscardChanges;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.EditConfig;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.Lock;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.Unlock;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.get.Get;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.get.GetConfig;
+
+final class OperationProvider {
+
+ private final String netconfSessionIdForReporting;
+ private final CurrentSchemaContext schemaContext;
+ private final DOMDataBroker dataBroker;
+ private final TransactionProvider transactionProvider;
+
+ public OperationProvider(final String netconfSessionIdForReporting, final CurrentSchemaContext schemaContext, final DOMDataBroker dataBroker) {
+ this.netconfSessionIdForReporting = netconfSessionIdForReporting;
+ this.schemaContext = schemaContext;
+ this.dataBroker = dataBroker;
+ this.transactionProvider = new TransactionProvider(dataBroker, netconfSessionIdForReporting);
+
+ }
+
+ Set<NetconfOperation> getOperations() {
+ return Sets.<NetconfOperation>newHashSet(
+ new Commit(netconfSessionIdForReporting, transactionProvider),
+ new DiscardChanges(netconfSessionIdForReporting, transactionProvider),
+ new EditConfig(netconfSessionIdForReporting, schemaContext, transactionProvider),
+ new Get(netconfSessionIdForReporting, schemaContext, transactionProvider),
+ new GetConfig(netconfSessionIdForReporting, schemaContext, transactionProvider),
+ new Lock(netconfSessionIdForReporting),
+ new Unlock(netconfSessionIdForReporting)
+ );
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector;
+
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.CheckedFuture;
+import java.util.ArrayList;
+import java.util.List;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorSeverity;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorTag;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+//TODO make a global TransactionProvider for all Netconf sessions instead of each session having one.
+public class TransactionProvider implements AutoCloseable{
+
+ private static final Logger LOG = LoggerFactory.getLogger(TransactionProvider.class);
+
+ private final DOMDataBroker dataBroker;
+
+ private DOMDataReadWriteTransaction candidateTransaction = null;
+ private DOMDataReadWriteTransaction runningTransaction = null;
+ private final List<DOMDataReadWriteTransaction> allOpenReadWriteTransactions = new ArrayList<>();
+
+ private final String netconfSessionIdForReporting;
+
+ private static final String NO_TRANSACTION_FOUND_FOR_SESSION = "No candidateTransaction found for session ";
+
+
+ public TransactionProvider(DOMDataBroker dataBroker, String netconfSessionIdForReporting) {
+ this.dataBroker = dataBroker;
+ this.netconfSessionIdForReporting = netconfSessionIdForReporting;
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ for (DOMDataReadWriteTransaction rwt : allOpenReadWriteTransactions) {
+ rwt.cancel();
+ }
+
+ allOpenReadWriteTransactions.clear();
+ }
+
+ public synchronized Optional<DOMDataReadWriteTransaction> getCandidateTransaction() {
+ if (candidateTransaction == null) {
+ return Optional.absent();
+ }
+
+ return Optional.of(candidateTransaction);
+ }
+
+ public synchronized DOMDataReadWriteTransaction getOrCreateTransaction() {
+ if (getCandidateTransaction().isPresent()) {
+ return getCandidateTransaction().get();
+ }
+
+ candidateTransaction = dataBroker.newReadWriteTransaction();
+ allOpenReadWriteTransactions.add(candidateTransaction);
+ return candidateTransaction;
+ }
+
+ public synchronized boolean commitTransaction() throws NetconfDocumentedException {
+ if (!getCandidateTransaction().isPresent()) {
+ throw new NetconfDocumentedException(NO_TRANSACTION_FOUND_FOR_SESSION + netconfSessionIdForReporting,
+ ErrorType.application, ErrorTag.operation_failed, ErrorSeverity.error);
+ }
+
+ CheckedFuture<Void, TransactionCommitFailedException> future = candidateTransaction.submit();
+ try {
+ future.checkedGet();
+ } catch (TransactionCommitFailedException e) {
+ LOG.debug("Transaction {} failed on", candidateTransaction, e);
+ throw new NetconfDocumentedException("Transaction commit failed on " + e.getMessage() + " " + netconfSessionIdForReporting,
+ ErrorType.application, ErrorTag.operation_failed, ErrorSeverity.error);
+ }
+ allOpenReadWriteTransactions.remove(candidateTransaction);
+ candidateTransaction = null;
+
+ return true;
+ }
+
+ public synchronized void abortTransaction() {
+ LOG.debug("Aborting current candidateTransaction");
+ Optional<DOMDataReadWriteTransaction> otx = getCandidateTransaction();
+ Preconditions.checkState(otx.isPresent(), NO_TRANSACTION_FOUND_FOR_SESSION + netconfSessionIdForReporting);
+ candidateTransaction.cancel();
+ allOpenReadWriteTransactions.remove(candidateTransaction);
+ candidateTransaction = null;
+ }
+
+ public synchronized DOMDataReadWriteTransaction createRunningTransaction() {
+ runningTransaction = dataBroker.newReadWriteTransaction();
+ allOpenReadWriteTransactions.add(runningTransaction);
+ return runningTransaction;
+ }
+
+ public synchronized boolean commitRunningTransaction(DOMDataReadWriteTransaction tx) throws NetconfDocumentedException {
+ allOpenReadWriteTransactions.remove(tx);
+
+ CheckedFuture<Void, TransactionCommitFailedException> future = tx.submit();
+ try {
+ future.checkedGet();
+ } catch (TransactionCommitFailedException e) {
+ LOG.debug("Transaction {} failed on", tx, e);
+ throw new NetconfDocumentedException("Transaction commit failed on " + e.getMessage() + " " + netconfSessionIdForReporting,
+ ErrorType.application, ErrorTag.operation_failed, ErrorSeverity.error);
+ }
+
+ return true;
+ }
+
+ public synchronized void abortRunningTransaction(DOMDataReadWriteTransaction tx) {
+ LOG.debug("Aborting current running Transaction");
+ Preconditions.checkState(runningTransaction != null, NO_TRANSACTION_FOUND_FOR_SESSION + netconfSessionIdForReporting);
+ tx.cancel();
+ allOpenReadWriteTransactions.remove(tx);
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops;
+
+import com.google.common.base.Optional;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
+import org.opendaylight.controller.netconf.mdsal.connector.TransactionProvider;
+import org.opendaylight.controller.netconf.util.mapping.AbstractLastNetconfOperation;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.controller.netconf.util.xml.XmlUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+public class Commit extends AbstractLastNetconfOperation{
+
+ private static final Logger LOG = LoggerFactory.getLogger(Commit.class);
+
+ private static final String OPERATION_NAME = "commit";
+ private final TransactionProvider transactionProvider;
+
+ public Commit(final String netconfSessionIdForReporting, final TransactionProvider transactionProvider) {
+ super(netconfSessionIdForReporting);
+ this.transactionProvider = transactionProvider;
+
+ }
+
+ @Override
+ protected Element handleWithNoSubsequentOperations(final Document document, final XmlElement operationElement) throws NetconfDocumentedException {
+
+ boolean commitStatus = transactionProvider.commitTransaction();
+ LOG.trace("Transaction commited succesfuly", commitStatus);
+
+ return XmlUtil.createElement(document, XmlNetconfConstants.OK, Optional.<String>absent());
+ }
+
+ @Override
+ protected String getOperationName() {
+ return OPERATION_NAME;
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops;
+
+public enum Datastore {
+ candidate, running
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops;
+
+import com.google.common.base.Optional;
+import java.util.HashMap;
+import java.util.Map;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorSeverity;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorTag;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorType;
+import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
+import org.opendaylight.controller.netconf.mdsal.connector.TransactionProvider;
+import org.opendaylight.controller.netconf.util.mapping.AbstractLastNetconfOperation;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.controller.netconf.util.xml.XmlUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+public class DiscardChanges extends AbstractLastNetconfOperation{
+
+ private static final Logger LOG = LoggerFactory.getLogger(DiscardChanges.class);
+
+ private static final String OPERATION_NAME = "discard-changes";
+
+ private final TransactionProvider transactionProvider;
+
+ public DiscardChanges(final String netconfSessionIdForReporting, final TransactionProvider transactionProvider) {
+ super(netconfSessionIdForReporting);
+ this.transactionProvider = transactionProvider;
+ }
+
+ @Override
+ protected Element handleWithNoSubsequentOperations(final Document document, final XmlElement operationElement) throws NetconfDocumentedException {
+ operationElement.getOnlyChildElement(OPERATION_NAME);
+
+ try {
+ transactionProvider.abortTransaction();
+ } catch (IllegalStateException e) {
+ LOG.warn("Abort failed ", e);
+ final Map<String, String> errorInfo = new HashMap<>();
+ errorInfo
+ .put(ErrorTag.operation_failed.name(),
+ "Operation failed. Use 'get-config' or 'edit-config' before triggering 'discard-changes' operation");
+ throw new NetconfDocumentedException(e.getMessage(), e, ErrorType.application, ErrorTag.operation_failed,
+ ErrorSeverity.error, errorInfo);
+ }
+ return XmlUtil.createElement(document, XmlNetconfConstants.OK, Optional.<String>absent());
+ }
+
+ @Override
+ protected String getOperationName() {
+ return OPERATION_NAME;
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.CheckedFuture;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Collections;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorSeverity;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorTag;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorType;
+import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
+import org.opendaylight.controller.netconf.mdsal.connector.CurrentSchemaContext;
+import org.opendaylight.controller.netconf.mdsal.connector.TransactionProvider;
+import org.opendaylight.controller.netconf.util.mapping.AbstractLastNetconfOperation;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.controller.netconf.util.xml.XmlUtil;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
+import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.transform.dom.DomUtils;
+import org.opendaylight.yangtools.yang.data.impl.schema.transform.dom.parser.DomToNormalizedNodeParserFactory;
+import org.opendaylight.yangtools.yang.data.operations.DataModificationException;
+import org.opendaylight.yangtools.yang.data.operations.DataModificationException.DataExistsException;
+import org.opendaylight.yangtools.yang.data.operations.DataModificationException.DataMissingException;
+import org.opendaylight.yangtools.yang.data.operations.DataOperations;
+import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
+import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
+import org.opendaylight.yangtools.yang.model.api.ListSchemaNode;
+import org.opendaylight.yangtools.yang.model.api.Module;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+public class EditConfig extends AbstractLastNetconfOperation {
+
+ private static final Logger LOG = LoggerFactory.getLogger(EditConfig.class);
+
+ private static final String OPERATION_NAME = "edit-config";
+ private static final String CONFIG_KEY = "config";
+
+ private final CurrentSchemaContext schemaContext;
+ private final TransactionProvider transactionProvider;
+
+ public EditConfig(final String netconfSessionIdForReporting, final CurrentSchemaContext schemaContext, final TransactionProvider transactionProvider) {
+ super(netconfSessionIdForReporting);
+ this.schemaContext = schemaContext;
+ this.transactionProvider = transactionProvider;
+ }
+
+ @Override
+ protected Element handleWithNoSubsequentOperations(final Document document, final XmlElement operationElement) throws NetconfDocumentedException {
+ final XmlElement configElement = getConfigElement(operationElement);
+
+ for (XmlElement element : configElement.getChildElements()) {
+ final String ns = element.getNamespace();
+ final DataSchemaNode schemaNode = getSchemaNodeFromNamespace(ns, element).get();
+ YangInstanceIdentifier ident = YangInstanceIdentifier.of(schemaNode.getQName());
+
+ final NormalizedNode storedNode = readStoredNode(LogicalDatastoreType.CONFIGURATION, ident);
+ try {
+ final Optional<NormalizedNode<?,?>> newNode = modifyNode(schemaNode, element, storedNode);
+ final DOMDataReadWriteTransaction rwTx = transactionProvider.getOrCreateTransaction();
+ if (newNode.isPresent()) {
+ rwTx.put(LogicalDatastoreType.CONFIGURATION, ident, newNode.get());
+ } else {
+ rwTx.delete(LogicalDatastoreType.CONFIGURATION, ident);
+ }
+ } catch (final DataModificationException e) {
+ if (e instanceof DataExistsException) {
+ throw new NetconfDocumentedException(e.getMessage(), e, ErrorType.protocol, ErrorTag.data_exists, ErrorSeverity.error);
+ } else if (e instanceof DataMissingException) {
+ throw new NetconfDocumentedException(e.getMessage(), e, ErrorType.protocol, ErrorTag.data_missing, ErrorSeverity.error);
+ } else {
+ //should never happen, since in edit-config only the 2 previous cases can happen
+ throw new NetconfDocumentedException(e.getMessage(), e, ErrorType.protocol, ErrorTag.operation_failed, ErrorSeverity.error);
+ }
+ }
+ }
+
+ return XmlUtil.createElement(document, XmlNetconfConstants.OK, Optional.<String>absent());
+ }
+
+ private NormalizedNode readStoredNode(final LogicalDatastoreType logicalDatastoreType, final YangInstanceIdentifier path) throws NetconfDocumentedException{
+ final DOMDataReadWriteTransaction rwTx = transactionProvider.getOrCreateTransaction();
+ final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readFuture = rwTx.read(logicalDatastoreType, path);
+ try {
+ if (readFuture.checkedGet().isPresent()) {
+ final NormalizedNode node = readFuture.checkedGet().get();
+ return node;
+ } else {
+ LOG.warn("Unable to read node : {} from {} datastore", path, logicalDatastoreType);
+ }
+ } catch (final ReadFailedException e) {
+ //only log this since DataOperations.modify will handle throwing an exception or writing the node.
+ LOG.warn("Unable to read stored data: {}", path, e);
+ }
+
+ //we can return null here since DataOperations.modify handles null as input
+ return null;
+ }
+
+ private Optional<DataSchemaNode> getSchemaNodeFromNamespace(final String namespace, final XmlElement element){
+ Optional<DataSchemaNode> dataSchemaNode = Optional.absent();
+ try {
+ //returns module with newest revision since findModuleByNamespace returns a set of modules and we only need the newest one
+ final Module module = schemaContext.getCurrentContext().findModuleByNamespaceAndRevision(new URI(namespace), null);
+ dataSchemaNode = Optional.of(module.getDataChildByName(element.getName()));
+ } catch (URISyntaxException e) {
+ LOG.debug("Unable to create URI for namespace : {}", namespace);
+ }
+
+ return dataSchemaNode;
+ }
+
+ private Optional<NormalizedNode<?, ?>> modifyNode(final DataSchemaNode schemaNode, final XmlElement element, final NormalizedNode storedNode) throws DataModificationException{
+ if (schemaNode instanceof ContainerSchemaNode) {
+ final ContainerNode modifiedNode =
+ DomToNormalizedNodeParserFactory
+ .getInstance(DomUtils.defaultValueCodecProvider())
+ .getContainerNodeParser()
+ .parse(Collections.singletonList(element.getDomElement()), (ContainerSchemaNode) schemaNode);
+
+ final Optional<ContainerNode> oNode = DataOperations.modify((ContainerSchemaNode) schemaNode, (ContainerNode) storedNode, modifiedNode);
+ if (!oNode.isPresent()) {
+ return Optional.absent();
+ }
+
+ final NormalizedNode<?,?> node = oNode.get();
+ return Optional.<NormalizedNode<?,?>>of(node);
+ } else if (schemaNode instanceof ListSchemaNode) {
+ final MapNode modifiedNode =
+ DomToNormalizedNodeParserFactory
+ .getInstance(DomUtils.defaultValueCodecProvider())
+ .getMapNodeParser()
+ .parse(Collections.singletonList(element.getDomElement()), (ListSchemaNode) schemaNode);
+
+ final Optional<MapNode> oNode = DataOperations.modify((ListSchemaNode) schemaNode, (MapNode) storedNode, modifiedNode);
+ if (!oNode.isPresent()) {
+ return Optional.absent();
+ }
+
+ final NormalizedNode<?, ?> node = oNode.get();
+ return Optional.<NormalizedNode<?,?>>of(node);
+ } else {
+ //this should never happen since edit-config on any other node type should not be possible nor makes sense
+ LOG.debug("DataNode from module is not ContainerSchemaNode nor ListSchemaNode, aborting..");
+ return Optional.absent();
+ }
+
+ }
+
+ private XmlElement getConfigElement(final XmlElement operationElement) throws NetconfDocumentedException{
+ final Optional<XmlElement> configChildNode = operationElement.getOnlyChildElementOptionally(CONFIG_KEY);
+ if (!configChildNode.isPresent()) {
+ throw new NetconfDocumentedException("Can't get child element with name: " + CONFIG_KEY,
+ ErrorType.application,
+ ErrorTag.unknown_element,
+ ErrorSeverity.error);
+ }
+
+ return configChildNode.get();
+ }
+
+ @Override
+ protected String getOperationName() {
+ return OPERATION_NAME;
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops;
+
+import com.google.common.base.Optional;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
+import org.opendaylight.controller.netconf.util.exception.MissingNameSpaceException;
+import org.opendaylight.controller.netconf.util.exception.UnexpectedNamespaceException;
+import org.opendaylight.controller.netconf.util.mapping.AbstractLastNetconfOperation;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.controller.netconf.util.xml.XmlUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+public class Lock extends AbstractLastNetconfOperation{
+
+ private static final Logger LOG = LoggerFactory.getLogger(Lock.class);
+
+ private static final String OPERATION_NAME = "lock";
+ private static final String TARGET_KEY = "target";
+
+ public Lock(final String netconfSessionIdForReporting) {
+ super(netconfSessionIdForReporting);
+ }
+
+ @Override
+ protected Element handleWithNoSubsequentOperations(final Document document, final XmlElement operationElement) throws NetconfDocumentedException {
+ final Datastore targetDatastore = extractTargetParameter(operationElement);
+ if (targetDatastore == Datastore.candidate) {
+ LOG.debug("Locking candidate datastore on session: {}", getNetconfSessionIdForReporting());
+ return XmlUtil.createElement(document, XmlNetconfConstants.OK, Optional.<String>absent());
+ }
+
+ throw new NetconfDocumentedException("Unable to lock " + targetDatastore + " datastore", NetconfDocumentedException.ErrorType.application,
+ NetconfDocumentedException.ErrorTag.operation_not_supported, NetconfDocumentedException.ErrorSeverity.error);
+ }
+
+ static Datastore extractTargetParameter(final XmlElement operationElement) throws NetconfDocumentedException {
+ final XmlElement targetChildNode;
+ try {
+ final XmlElement targetElement = operationElement.getOnlyChildElementWithSameNamespace(TARGET_KEY);
+ targetChildNode = targetElement.getOnlyChildElementWithSameNamespace();
+ } catch (final MissingNameSpaceException | UnexpectedNamespaceException e) {
+ LOG.trace("Can't get only child element with same namespace", e);
+ throw NetconfDocumentedException.wrap(e);
+ }
+
+ return Datastore.valueOf(targetChildNode.getName());
+ }
+
+ @Override
+ protected String getOperationName() {
+ return OPERATION_NAME;
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops;
+
+import com.google.common.base.Optional;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
+import org.opendaylight.controller.netconf.util.mapping.AbstractLastNetconfOperation;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.controller.netconf.util.xml.XmlUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+public class Unlock extends AbstractLastNetconfOperation{
+
+ private static final Logger LOG = LoggerFactory.getLogger(Unlock.class);
+
+ private static final String OPERATION_NAME = "unlock";
+ private static final String TARGET_KEY = "target";
+
+ public Unlock(final String netconfSessionIdForReporting) {
+ super(netconfSessionIdForReporting);
+ }
+
+ @Override
+ protected Element handleWithNoSubsequentOperations(final Document document, final XmlElement operationElement) throws NetconfDocumentedException {
+ final Datastore targetDatastore = Lock.extractTargetParameter(operationElement);
+ if (targetDatastore == Datastore.candidate) {
+ LOG.debug("Unlocking candidate datastore on session: {}", getNetconfSessionIdForReporting());
+ return XmlUtil.createElement(document, XmlNetconfConstants.OK, Optional.<String>absent());
+ }
+
+ throw new NetconfDocumentedException("Unable to unlock " + targetDatastore + " datastore", NetconfDocumentedException.ErrorType.application,
+ NetconfDocumentedException.ErrorTag.operation_not_supported, NetconfDocumentedException.ErrorSeverity.error);
+ }
+
+ @Override
+ protected String getOperationName() {
+ return OPERATION_NAME;
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops.get;
+
+import com.google.common.base.Function;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Iterables;
+import java.io.IOException;
+import javax.xml.stream.XMLOutputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamWriter;
+import javax.xml.transform.dom.DOMResult;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
+import org.opendaylight.controller.netconf.mdsal.connector.CurrentSchemaContext;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.Datastore;
+import org.opendaylight.controller.netconf.util.mapping.AbstractLastNetconfOperation;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
+import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter;
+import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeWriter;
+import org.opendaylight.yangtools.yang.data.impl.codec.xml.XMLStreamNormalizedNodeStreamWriter;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+import org.w3c.dom.Document;
+import org.w3c.dom.Node;
+
+public abstract class AbstractGet extends AbstractLastNetconfOperation {
+
+ protected static final YangInstanceIdentifier ROOT = YangInstanceIdentifier.builder().build();
+
+ protected final CurrentSchemaContext schemaContext;
+
+
+ public AbstractGet(final String netconfSessionIdForReporting, final CurrentSchemaContext schemaContext) {
+ super(netconfSessionIdForReporting);
+ this.schemaContext = schemaContext;
+ }
+
+ private static final XMLOutputFactory XML_OUTPUT_FACTORY;
+
+ static {
+ XML_OUTPUT_FACTORY = XMLOutputFactory.newFactory();
+ XML_OUTPUT_FACTORY.setProperty(XMLOutputFactory.IS_REPAIRING_NAMESPACES, true);
+ }
+
+ protected Node transformNormalizedNode(final Document document, final NormalizedNode<?, ?> data, final YangInstanceIdentifier dataRoot) {
+// boolean isDataRoot = true;
+
+ final DOMResult result = new DOMResult(document.createElement(XmlNetconfConstants.DATA_KEY));
+
+ final XMLStreamWriter xmlWriter = getXmlStreamWriter(result);
+
+ final NormalizedNodeStreamWriter nnStreamWriter = XMLStreamNormalizedNodeStreamWriter.create(xmlWriter,
+ schemaContext.getCurrentContext(), getSchemaPath(dataRoot));
+
+ final NormalizedNodeWriter nnWriter = NormalizedNodeWriter.forStreamWriter(nnStreamWriter);
+
+// if (isDataRoot) {
+ writeRootElement(xmlWriter, nnWriter, (ContainerNode) data);
+// } else {
+// if (data instanceof MapEntryNode) {
+// // Restconf allows returning one list item. We need to wrap it
+// // in map node in order to serialize it properly
+// data = ImmutableNodes.mapNodeBuilder(data.getNodeType()).addChild((MapEntryNode) data).build();
+// }
+// nnWriter.write(data);
+// nnWriter.flush();
+// }
+ return result.getNode();
+ }
+
+ private XMLStreamWriter getXmlStreamWriter(final DOMResult result) {
+ try {
+ return XML_OUTPUT_FACTORY.createXMLStreamWriter(result);
+ } catch (final XMLStreamException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static final Function<PathArgument, QName> PATH_ARG_TO_QNAME = new Function<YangInstanceIdentifier.PathArgument, QName>() {
+ @Override
+ public QName apply(final YangInstanceIdentifier.PathArgument input) {
+ return input.getNodeType();
+ }
+ };
+
+ private SchemaPath getSchemaPath(final YangInstanceIdentifier dataRoot) {
+ return SchemaPath.create(Iterables.transform(dataRoot.getPathArguments(), PATH_ARG_TO_QNAME), dataRoot.equals(ROOT));
+ }
+
+ // TODO this code is located in Restconf already
+ private void writeRootElement(final XMLStreamWriter xmlWriter, final NormalizedNodeWriter nnWriter, final ContainerNode data) {
+ try {
+ final QName name = SchemaContext.NAME;
+ for (final DataContainerChild<? extends PathArgument, ?> child : data.getValue()) {
+ nnWriter.write(child);
+ }
+ nnWriter.flush();
+ xmlWriter.flush();
+ } catch (XMLStreamException | IOException e) {
+ Throwables.propagate(e);
+ }
+ }
+
+ protected static final class GetConfigExecution {
+ private final Datastore datastore;
+
+ public GetConfigExecution(final Datastore datastore) {
+ this.datastore = datastore;
+ }
+
+ public Datastore getDatastore() {
+ return datastore;
+ }
+
+ static GetConfigExecution fromXml(final XmlElement xml, final String operationName) throws NetconfDocumentedException {
+ try {
+ validateInputRpc(xml, operationName);
+ } catch (final NetconfDocumentedException e) {
+ throw new NetconfDocumentedException("Incorrect RPC: " + e.getMessage(), e.getErrorType(), e.getErrorTag(), e.getErrorSeverity(), e.getErrorInfo());
+ }
+
+ final Datastore sourceDatastore;
+ try {
+ sourceDatastore = parseSource(xml);
+ } catch (final NetconfDocumentedException e) {
+ throw new NetconfDocumentedException("Get-config source attribute error: " + e.getMessage(), e.getErrorType(), e.getErrorTag(), e.getErrorSeverity(), e.getErrorInfo());
+ }
+
+ // Add filter
+
+ return new GetConfigExecution(sourceDatastore);
+ }
+
+ private static Datastore parseSource(final XmlElement xml) throws NetconfDocumentedException {
+ final Datastore sourceDatastore;
+ final XmlElement sourceElement = xml.getOnlyChildElement(XmlNetconfConstants.SOURCE_KEY,
+ XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0);
+
+ final String sourceParsed = sourceElement.getOnlyChildElement().getName();
+ sourceDatastore = Datastore.valueOf(sourceParsed);
+ return sourceDatastore;
+ }
+
+ private static void validateInputRpc(final XmlElement xml, String operationName) throws NetconfDocumentedException{
+ xml.checkName(operationName);
+ xml.checkNamespace(XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0);
+ }
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops.get;
+
+import com.google.common.base.Optional;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorSeverity;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorTag;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorType;
+import org.opendaylight.controller.netconf.mdsal.connector.CurrentSchemaContext;
+import org.opendaylight.controller.netconf.mdsal.connector.TransactionProvider;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.Datastore;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+public class Get extends AbstractGet {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Get.class);
+
+ private static final String OPERATION_NAME = "get";
+
+ private final TransactionProvider transactionProvider;
+
+ public Get(final String netconfSessionIdForReporting, final CurrentSchemaContext schemaContext, final TransactionProvider transactionProvider) {
+ super(netconfSessionIdForReporting, schemaContext);
+ this.transactionProvider = transactionProvider;
+ }
+
+ @Override
+ protected Element handleWithNoSubsequentOperations(Document document, XmlElement operationElement) throws NetconfDocumentedException {
+ GetConfigExecution getConfigExecution = null;
+ try {
+ getConfigExecution = GetConfigExecution.fromXml(operationElement, OPERATION_NAME);
+
+ } catch (final NetconfDocumentedException e) {
+ LOG.warn("Get request processing failed on session: {}", getNetconfSessionIdForReporting(), e);
+ throw e;
+ }
+
+ final YangInstanceIdentifier dataRoot = ROOT;
+ DOMDataReadWriteTransaction rwTx = getTransaction(getConfigExecution.getDatastore());
+ try {
+ final Optional<NormalizedNode<?, ?>> normalizedNodeOptional = rwTx.read(LogicalDatastoreType.OPERATIONAL, dataRoot).checkedGet();
+ if (getConfigExecution.getDatastore() == Datastore.running) {
+ transactionProvider.abortRunningTransaction(rwTx);
+ rwTx = null;
+ }
+ return (Element) transformNormalizedNode(document, normalizedNodeOptional.get(), dataRoot);
+ } catch (ReadFailedException e) {
+ LOG.warn("Unable to read data: {}", dataRoot, e);
+ throw new IllegalStateException("Unable to read data " + dataRoot, e);
+ }
+ }
+
+ private DOMDataReadWriteTransaction getTransaction(Datastore datastore) throws NetconfDocumentedException{
+ if (datastore == Datastore.candidate) {
+ return transactionProvider.getOrCreateTransaction();
+ } else if (datastore == Datastore.running) {
+ return transactionProvider.createRunningTransaction();
+ }
+ throw new NetconfDocumentedException("Incorrect Datastore: ", ErrorType.protocol, ErrorTag.bad_element, ErrorSeverity.error);
+ }
+
+ @Override
+ protected String getOperationName() {
+ return OPERATION_NAME;
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.mdsal.connector.ops.get;
+
+import com.google.common.base.Optional;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorSeverity;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorTag;
+import org.opendaylight.controller.netconf.api.NetconfDocumentedException.ErrorType;
+import org.opendaylight.controller.netconf.mdsal.connector.CurrentSchemaContext;
+import org.opendaylight.controller.netconf.mdsal.connector.TransactionProvider;
+import org.opendaylight.controller.netconf.mdsal.connector.ops.Datastore;
+import org.opendaylight.controller.netconf.util.xml.XmlElement;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
+
+public class GetConfig extends AbstractGet {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GetConfig.class);
+
+ private static final String OPERATION_NAME = "get-config";
+
+ private final TransactionProvider transactionProvider;
+
+ public GetConfig(final String netconfSessionIdForReporting, final CurrentSchemaContext schemaContext, final TransactionProvider transactionProvider) {
+ super(netconfSessionIdForReporting, schemaContext);
+ this.transactionProvider = transactionProvider;
+ }
+
+ @Override
+ protected Element handleWithNoSubsequentOperations(Document document, XmlElement operationElement) throws NetconfDocumentedException {
+ GetConfigExecution getConfigExecution = null;
+ try {
+ getConfigExecution = GetConfigExecution.fromXml(operationElement, OPERATION_NAME);
+
+ } catch (final NetconfDocumentedException e) {
+ LOG.warn("Get request processing failed on session: {}", getNetconfSessionIdForReporting(), e);
+ throw e;
+ }
+
+ final YangInstanceIdentifier dataRoot = ROOT;
+ DOMDataReadWriteTransaction rwTx = getTransaction(getConfigExecution.getDatastore());
+ try {
+ final Optional<NormalizedNode<?, ?>> normalizedNodeOptional = rwTx.read(LogicalDatastoreType.CONFIGURATION, dataRoot).checkedGet();
+ if (getConfigExecution.getDatastore() == Datastore.running) {
+ transactionProvider.abortRunningTransaction(rwTx);
+ rwTx = null;
+ }
+ return (Element) transformNormalizedNode(document, normalizedNodeOptional.get(), dataRoot);
+ } catch (ReadFailedException e) {
+ LOG.warn("Unable to read data: {}", dataRoot, e);
+ throw new IllegalStateException("Unable to read data " + dataRoot, e);
+ }
+ }
+
+ private DOMDataReadWriteTransaction getTransaction(Datastore datastore) throws NetconfDocumentedException{
+ if (datastore == Datastore.candidate) {
+ return transactionProvider.getOrCreateTransaction();
+ } else if (datastore == Datastore.running) {
+ return transactionProvider.createRunningTransaction();
+ }
+ throw new NetconfDocumentedException("Incorrect Datastore: ", ErrorType.protocol, ErrorTag.bad_element, ErrorSeverity.error);
+ }
+
+ @Override
+ protected String getOperationName() {
+ return OPERATION_NAME;
+ }
+
+}
--- /dev/null
+module netconf-mdsal-mapper {
+ yang-version 1;
+ namespace "urn:opendaylight:params:xml:ns:yang:controller:netconf:mdsal:mapper";
+ prefix "nmm";
+
+ import netconf-northbound-mapper { prefix nnm; revision-date 2015-01-14; }
+ import opendaylight-md-sal-dom { prefix md-sal-dom; revision-date 2013-10-28; }
+ import config { prefix config; revision-date 2013-04-05; }
+
+ organization "Cisco Systems, Inc.";
+
+ description
+ "This module contains the base YANG definitions for
+ an MD-SAL mapper implementation";
+
+ revision "2015-01-14" {
+ description
+ "Initial revision.";
+ }
+
+ identity netconf-mdsal-mapper {
+ base config:module-type;
+ config:provided-service nnm:netconf-northbound-mapper;
+ }
+
+ augment "/config:modules/config:module/config:configuration" {
+ case netconf-mdsal-mapper {
+ when "/config:modules/config:module/config:type = 'netconf-mdsal-mapper'";
+
+ container root-schema-service {
+ uses config:service-ref {
+ refine type {
+ mandatory false;
+ config:required-identity md-sal-dom:schema-service;
+ }
+ }
+ }
+
+ container dom-broker {
+ uses config:service-ref {
+ refine type {
+ mandatory false;
+ config:required-identity md-sal-dom:dom-async-data-broker;
+ }
+ }
+ }
+ }
+ }
+
+}
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
- <configuration>
- <instructions>
- <Export-Package>org.opendaylight.controller.netconf.api,
- org.opendaylight.controller.netconf.api.jmx,
- org.opendaylight.controller.netconf.api.xml,
- org.opendaylight.controller.netconf.api.monitoring,</Export-Package>
- </instructions>
- </configuration>
</plugin>
+ <plugin>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>config</id>
+ <goals>
+ <goal>generate-sources</goal>
+ </goals>
+ <configuration>
+ <codeGenerators>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator</codeGeneratorClass>
+ <outputBaseDir>${jmxGeneratorPath}</outputBaseDir>
+ <additionalConfiguration>
+ <namespaceToPackage1>urn:opendaylight:params:xml:ns:yang:controller==org.opendaylight.controller.config.yang</namespaceToPackage1>
+ </additionalConfiguration>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.maven.sal.api.gen.plugin.CodeGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>${salGeneratorPath}</outputBaseDir>
+ </generator>
+ </codeGenerators>
+ <inspectDependencies>true</inspectDependencies>
+ </configuration>
+ </execution>
+ </executions>
+ <dependencies>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>yang-jmx-generator-plugin</artifactId>
+ <version>${config.version}</version>
+ </dependency>
+ </dependencies>
+ </plugin>
</plugins>
</build>
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.api;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.local.LocalAddress;
+import java.net.InetSocketAddress;
+
+public interface NetconfServerDispatcher {
+
+ ChannelFuture createServer(InetSocketAddress address);
+
+ ChannelFuture createLocalServer(LocalAddress address);
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.api.util;
+
+public final class NetconfConstants {
+ /*
+ * TODO define marker interface in mapping-api that the serviceFactories in cofing subsystem
+ * will implement so we can check for services with instanceof instead of constants
+ */
+ public static final String SERVICE_NAME = "name";
+ public static final String CONFIG_NETCONF_CONNECTOR = "config-netconf-connector";
+ public static final String NETCONF_MONITORING = "ietf-netconf-monitoring";
+}
--- /dev/null
+module netconf-northbound {
+ yang-version 1;
+ namespace "urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound";
+ prefix "nn";
+
+ import config { prefix config; revision-date 2013-04-05; }
+
+ description
+ "This module contains the base YANG definitions for
+ netconf northbound server API";
+
+ revision "2015-01-14" {
+ description
+ "Initial revision.";
+ }
+
+ identity netconf-server-dispatcher {
+ base "config:service-type";
+ config:java-class "org.opendaylight.controller.netconf.api.NetconfServerDispatcher";
+ }
+
+}
\ No newline at end of file
<artifactId>netconf-connector-config</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>netconf-mdsal-config</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>mdsal-netconf-connector</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>netconf-impl</artifactId>
<name>global-netconf-processing-executor-threadfactory</name>
</threadFactory>
</module>
+
+ <module>
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl:scheduled">prefix:threadpool-scheduled</type>
+ <name>global-netconf-ssh-scheduled-executor</name>
+ <max-thread-count xmlns="urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl:scheduled">8</max-thread-count>
+
+ <threadFactory xmlns="urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl:scheduled">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:threadpool">prefix:threadfactory</type>
+ <name>global-netconf-processing-executor-threadfactory</name>
+ </threadFactory>
+ </module>
</modules>
<services xmlns="urn:opendaylight:params:xml:ns:yang:controller:config">
<name>global-netconf-processing-executor</name>
<provider>/modules/module[type='threadpool-flexible'][name='global-netconf-processing-executor']</provider>
</instance>
+ <instance>
+ <name>global-netconf-ssh-scheduled-executor</name>
+ <provider>/modules/module[type='threadpool-scheduled'][name='global-netconf-ssh-scheduled-executor']</provider>
+ </instance>
</service>
</services>
<capability>urn:opendaylight:params:xml:ns:yang:controller:config:netconf:client:dispatcher?module=odl-netconfig-client-cfg&revision=2014-04-08</capability>
<capability>urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl?module=threadpool-impl&revision=2013-04-05</capability>
<capability>urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl:flexible?module=threadpool-impl-flexible&revision=2013-12-01</capability>
+ <capability>urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl:scheduled?module=threadpool-impl-scheduled&revision=2013-12-01</capability>
</required-capabilities>
</snapshot>
<name>global-netconf-processing-executor</name>
</processing-executor>
</module>
- </modules>
+ </modules>
</data>
</configuration>
<required-capabilities>
<configuration>
<instructions>
<Bundle-Activator>org.opendaylight.controller.netconf.impl.osgi.NetconfImplActivator</Bundle-Activator>
+ <Export-Package>org.opendaylight.controller.netconf.impl.*</Export-Package>
</instructions>
</configuration>
</plugin>
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>config</id>
+ <goals>
+ <goal>generate-sources</goal>
+ </goals>
+ <configuration>
+ <codeGenerators>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator</codeGeneratorClass>
+ <outputBaseDir>${jmxGeneratorPath}</outputBaseDir>
+ <additionalConfiguration>
+ <namespaceToPackage1>urn:opendaylight:params:xml:ns:yang:controller==org.opendaylight.controller.config.yang</namespaceToPackage1>
+ </additionalConfiguration>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.maven.sal.api.gen.plugin.CodeGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>${salGeneratorPath}</outputBaseDir>
+ </generator>
+ </codeGenerators>
+ <inspectDependencies>true</inspectDependencies>
+ </configuration>
+ </execution>
+ </executions>
+ <dependencies>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>yang-jmx-generator-plugin</artifactId>
+ <version>${config.version}</version>
+ </dependency>
+ </dependencies>
+ </plugin>
</plugins>
</build>
--- /dev/null
+package org.opendaylight.controller.config.yang.config.netconf.northbound.impl;
+
+import org.opendaylight.controller.config.api.JmxAttributeValidationException;
+import org.opendaylight.controller.netconf.impl.CommitNotifier;
+import org.opendaylight.controller.netconf.impl.NetconfServerDispatcherImpl;
+import org.opendaylight.controller.netconf.impl.NetconfServerSessionNegotiatorFactory;
+import org.opendaylight.controller.netconf.impl.SessionIdProvider;
+import org.opendaylight.controller.netconf.impl.osgi.NetconfMonitoringServiceImpl;
+import org.opendaylight.controller.netconf.impl.osgi.NetconfOperationServiceFactoryListenerImpl;
+import org.opendaylight.controller.netconf.impl.osgi.SessionMonitoringService;
+import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
+
+public class NetconfServerDispatcherModule extends org.opendaylight.controller.config.yang.config.netconf.northbound.impl.AbstractNetconfServerDispatcherModule {
+ public NetconfServerDispatcherModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+ super(identifier, dependencyResolver);
+ }
+
+ public NetconfServerDispatcherModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.config.netconf.northbound.impl.NetconfServerDispatcherModule oldModule, java.lang.AutoCloseable oldInstance) {
+ super(identifier, dependencyResolver, oldModule, oldInstance);
+ }
+
+ @Override
+ public void customValidation() {
+ JmxAttributeValidationException.checkCondition(getConnectionTimeoutMillis() > 0, "Invalid connection timeout", connectionTimeoutMillisJmxAttribute);
+ }
+
+ @Override
+ public java.lang.AutoCloseable createInstance() {
+
+ final NetconfOperationServiceFactoryListenerImpl aggregatedOpProvider = getAggregatedOpProvider();
+ final SessionMonitoringService monitoringService = startMonitoringService(aggregatedOpProvider);
+ final NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
+ getTimerDependency(), aggregatedOpProvider, new SessionIdProvider(), getConnectionTimeoutMillis(), CommitNotifier.NoopCommitNotifier.getInstance(), monitoringService);
+ final NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcherImpl.ServerChannelInitializer(
+ serverNegotiatorFactory);
+
+ return new NetconfServerDispatcherImpl(serverChannelInitializer, getBossThreadGroupDependency(), getWorkerThreadGroupDependency()) {
+
+ @Override
+ public void close() {
+ // NOOP, close should not be present here, the deprecated method closes injected evet loop groups
+ }
+ };
+
+ }
+
+ private NetconfMonitoringServiceImpl startMonitoringService(final NetconfOperationServiceFactoryListenerImpl netconfOperationProvider) {
+ return new NetconfMonitoringServiceImpl(netconfOperationProvider);
+ }
+
+ private NetconfOperationServiceFactoryListenerImpl getAggregatedOpProvider() {
+ final NetconfOperationServiceFactoryListenerImpl netconfOperationProvider = new NetconfOperationServiceFactoryListenerImpl();
+ for (final NetconfOperationServiceFactory netconfOperationServiceFactory : getMappersDependency()) {
+ netconfOperationProvider.onAddNetconfOperationServiceFactory(netconfOperationServiceFactory);
+ }
+ return netconfOperationProvider;
+ }
+
+
+}
--- /dev/null
+/*
+* Generated file
+*
+* Generated from: yang module name: netconf-northbound-impl yang module local name: netconf-server-dispatcher-impl
+* Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
+* Generated at: Thu Feb 12 11:32:29 CET 2015
+*
+* Do not modify this file unless it is present under src/main directory
+*/
+package org.opendaylight.controller.config.yang.config.netconf.northbound.impl;
+public class NetconfServerDispatcherModuleFactory extends org.opendaylight.controller.config.yang.config.netconf.northbound.impl.AbstractNetconfServerDispatcherModuleFactory {
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. 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.controller.netconf.impl;
+
+import java.util.Set;
+import org.w3c.dom.Element;
+
+public interface CommitNotifier {
+ void sendCommitNotification(String message, Element cfgSnapshot, Set<String> capabilities);
+
+ public static final class NoopCommitNotifier implements CommitNotifier {
+
+ private static final CommitNotifier INSTANCE = new NoopCommitNotifier();
+
+ private NoopCommitNotifier() {}
+
+ public static CommitNotifier getInstance() {
+ return INSTANCE;
+ }
+
+ @Override
+ public void sendCommitNotification(final String message, final Element cfgSnapshot, final Set<String> capabilities) {
+ // NOOP
+ }
+ }
+}
import org.w3c.dom.Element;
public class DefaultCommitNotificationProducer extends NotificationBroadcasterSupport implements
- DefaultCommitOperationMXBean, AutoCloseable {
+ DefaultCommitOperationMXBean, AutoCloseable, CommitNotifier {
private static final Logger LOG = LoggerFactory.getLogger(DefaultCommitNotificationProducer.class);
}
}
+ @Override
public void sendCommitNotification(String message, Element cfgSnapshot, Set<String> capabilities) {
CommitJMXNotification notif = NetconfJMXNotification.afterCommit(this, message, cfgSnapshot, capabilities);
LOG.debug("Notification about commit {} sent", notif);
package org.opendaylight.controller.netconf.impl;
-import com.google.common.annotations.VisibleForTesting;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.concurrent.Promise;
import java.net.InetSocketAddress;
+import org.opendaylight.controller.netconf.api.NetconfServerDispatcher;
import org.opendaylight.controller.netconf.impl.util.DeserializerExceptionHandler;
import org.opendaylight.controller.netconf.nettyutil.AbstractChannelInitializer;
import org.opendaylight.protocol.framework.AbstractDispatcher;
-public class NetconfServerDispatcher extends AbstractDispatcher<NetconfServerSession, NetconfServerSessionListener> {
+public class NetconfServerDispatcherImpl extends AbstractDispatcher<NetconfServerSession, NetconfServerSessionListener> implements NetconfServerDispatcher {
private final ServerChannelInitializer initializer;
- public NetconfServerDispatcher(ServerChannelInitializer serverChannelInitializer, EventLoopGroup bossGroup,
- EventLoopGroup workerGroup) {
+ public NetconfServerDispatcherImpl(ServerChannelInitializer serverChannelInitializer, EventLoopGroup bossGroup,
+ EventLoopGroup workerGroup) {
super(bossGroup, workerGroup);
this.initializer = serverChannelInitializer;
}
- @VisibleForTesting
+ @Override
public ChannelFuture createServer(InetSocketAddress address) {
-
return super.createServer(address, new PipelineInitializer<NetconfServerSession>() {
@Override
public void initializeChannel(final SocketChannel ch, final Promise<NetconfServerSession> promise) {
});
}
+ @Override
public ChannelFuture createLocalServer(LocalAddress address) {
return super.createServer(address, LocalServerChannel.class, new ChannelPipelineInitializer<LocalChannel, NetconfServerSession>() {
@Override
public class NetconfServerSessionListenerFactory implements SessionListenerFactory<NetconfServerSessionListener> {
- private final DefaultCommitNotificationProducer commitNotifier;
+ private final CommitNotifier commitNotifier;
private final SessionMonitoringService monitor;
private final NetconfOperationServiceSnapshot netconfOperationServiceSnapshot;
private final CapabilityProvider capabilityProvider;
- public NetconfServerSessionListenerFactory(final DefaultCommitNotificationProducer commitNotifier,
+ public NetconfServerSessionListenerFactory(final CommitNotifier commitNotifier,
final SessionMonitoringService monitor,
final NetconfOperationServiceSnapshot netconfOperationServiceSnapshot,
final CapabilityProvider capabilityProvider) {
private final SessionIdProvider idProvider;
private final NetconfOperationProvider netconfOperationProvider;
private final long connectionTimeoutMillis;
- private final DefaultCommitNotificationProducer commitNotificationProducer;
+ private final CommitNotifier commitNotificationProducer;
private final SessionMonitoringService monitoringService;
private static final Logger LOG = LoggerFactory.getLogger(NetconfServerSessionNegotiatorFactory.class);
private final Set<String> baseCapabilities;
// TODO too many params, refactor
public NetconfServerSessionNegotiatorFactory(Timer timer, NetconfOperationProvider netconfOperationProvider,
SessionIdProvider idProvider, long connectionTimeoutMillis,
- DefaultCommitNotificationProducer commitNot,
+ CommitNotifier commitNot,
SessionMonitoringService monitoringService) {
this(timer, netconfOperationProvider, idProvider, connectionTimeoutMillis, commitNot, monitoringService, DEFAULT_BASE_CAPABILITIES);
}
// TODO too many params, refactor
public NetconfServerSessionNegotiatorFactory(Timer timer, NetconfOperationProvider netconfOperationProvider,
SessionIdProvider idProvider, long connectionTimeoutMillis,
- DefaultCommitNotificationProducer commitNot,
+ CommitNotifier commitNot,
SessionMonitoringService monitoringService, Set<String> baseCapabilities) {
this.timer = timer;
this.netconfOperationProvider = netconfOperationProvider;
import java.io.InputStream;
import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
-import org.opendaylight.controller.netconf.impl.DefaultCommitNotificationProducer;
+import org.opendaylight.controller.netconf.impl.CommitNotifier;
import org.opendaylight.controller.netconf.impl.mapping.CapabilityProvider;
import org.opendaylight.controller.netconf.impl.osgi.NetconfOperationRouter;
import org.opendaylight.controller.netconf.mapping.api.HandlingPriority;
private static final String NOTIFY_ATTR = "notify";
- private final DefaultCommitNotificationProducer notificationProducer;
+ private final CommitNotifier notificationProducer;
private final CapabilityProvider cap;
private final NetconfOperationRouter operationRouter;
- public DefaultCommit(DefaultCommitNotificationProducer notifier, CapabilityProvider cap,
+ public DefaultCommit(CommitNotifier notifier, CapabilityProvider cap,
String netconfSessionIdForReporting, NetconfOperationRouter netconfOperationRouter) {
super(netconfSessionIdForReporting);
this.notificationProducer = notifier;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.netconf.api.monitoring.NetconfMonitoringService;
import org.opendaylight.controller.netconf.impl.DefaultCommitNotificationProducer;
-import org.opendaylight.controller.netconf.impl.NetconfServerDispatcher;
+import org.opendaylight.controller.netconf.impl.NetconfServerDispatcherImpl;
import org.opendaylight.controller.netconf.impl.NetconfServerSessionNegotiatorFactory;
import org.opendaylight.controller.netconf.impl.SessionIdProvider;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider;
eventLoopGroup = new NioEventLoopGroup();
- NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(
+ NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcherImpl.ServerChannelInitializer(
serverNegotiatorFactory);
- NetconfServerDispatcher dispatch = new NetconfServerDispatcher(serverChannelInitializer, eventLoopGroup, eventLoopGroup);
+ NetconfServerDispatcherImpl dispatch = new NetconfServerDispatcherImpl(serverChannelInitializer, eventLoopGroup, eventLoopGroup);
LocalAddress address = NetconfConfigUtil.getNetconfLocalAddress();
LOG.trace("Starting local netconf server at {}", address);
import java.util.Set;
import java.util.TreeMap;
import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
-import org.opendaylight.controller.netconf.impl.DefaultCommitNotificationProducer;
+import org.opendaylight.controller.netconf.impl.CommitNotifier;
import org.opendaylight.controller.netconf.impl.NetconfServerSession;
import org.opendaylight.controller.netconf.impl.mapping.CapabilityProvider;
import org.opendaylight.controller.netconf.impl.mapping.operations.DefaultCloseSession;
private final Collection<NetconfOperation> allNetconfOperations;
public NetconfOperationRouterImpl(final NetconfOperationServiceSnapshot netconfOperationServiceSnapshot, final CapabilityProvider capabilityProvider,
- final DefaultCommitNotificationProducer commitNotifier) {
+ final CommitNotifier commitNotifier) {
this.netconfOperationServiceSnapshot = Preconditions.checkNotNull(netconfOperationServiceSnapshot);
final String sessionId = netconfOperationServiceSnapshot.getNetconfSessionIdForReporting();
*/
package org.opendaylight.controller.netconf.impl.osgi;
+import org.opendaylight.controller.netconf.api.util.NetconfConstants;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceReference;
@Override
public NetconfOperationServiceFactory addingService(ServiceReference<NetconfOperationServiceFactory> reference) {
- NetconfOperationServiceFactory netconfOperationServiceFactory = super.addingService(reference);
- factoriesListener.onAddNetconfOperationServiceFactory(netconfOperationServiceFactory);
- return netconfOperationServiceFactory;
+ Object property = reference.getProperty(NetconfConstants.SERVICE_NAME);
+ if (property != null
+ && (property.equals(NetconfConstants.CONFIG_NETCONF_CONNECTOR)
+ || property.equals(NetconfConstants.NETCONF_MONITORING))) {
+ NetconfOperationServiceFactory netconfOperationServiceFactory = super.addingService(reference);
+ factoriesListener.onAddNetconfOperationServiceFactory(netconfOperationServiceFactory);
+ return netconfOperationServiceFactory;
+ }
+
+ return null;
}
@Override
public void removedService(ServiceReference<NetconfOperationServiceFactory> reference,
NetconfOperationServiceFactory netconfOperationServiceFactory) {
- factoriesListener.onRemoveNetconfOperationServiceFactory(netconfOperationServiceFactory);
+ if (netconfOperationServiceFactory != null)
+ factoriesListener.onRemoveNetconfOperationServiceFactory(netconfOperationServiceFactory);
}
}
--- /dev/null
+// vi: set smarttab et sw=4 tabstop=4:
+module netconf-northbound-impl {
+
+ yang-version 1;
+ namespace "urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl";
+ prefix "cfg-net-s-i";
+
+ import config { prefix config; revision-date 2013-04-05; }
+ import netconf-northbound-mapper { prefix nnm; revision-date 2015-01-14; }
+ import netconf-northbound { prefix nn; revision-date 2015-01-14; }
+ import netty {prefix netty; }
+
+ description
+ "This module contains the base YANG definitions for
+ netconf-server-dispatcher implementation.
+
+ Copyright (c)2013 Cisco Systems, Inc. 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";
+
+ revision "2015-01-12" {
+ description
+ "Initial revision.";
+ }
+
+ identity netconf-server-dispatcher-impl {
+ base config:module-type;
+ config:provided-service nn:netconf-server-dispatcher;
+ config:java-name-prefix NetconfServerDispatcher;
+ }
+
+ augment "/config:modules/config:module/config:configuration" {
+ case netconf-server-dispatcher-impl {
+ when "/config:modules/config:module/config:type = 'netconf-server-dispatcher-impl'";
+
+ leaf connection-timeout-millis {
+ description "Specifies timeout in milliseconds after which connection must be established.";
+ type uint32;
+ default 20000;
+ }
+
+ container boss-thread-group {
+ uses config:service-ref {
+ refine type {
+ config:required-identity netty:netty-threadgroup;
+ }
+ }
+ }
+
+ container worker-thread-group {
+ uses config:service-ref {
+ refine type {
+ config:required-identity netty:netty-threadgroup;
+ }
+ }
+ }
+
+ list mappers {
+ uses config:service-ref {
+ refine type {
+ mandatory true;
+ config:required-identity nnm:netconf-northbound-mapper;
+ }
+ }
+ }
+
+ container timer {
+ uses config:service-ref {
+ refine type {
+ config:required-identity netty:netty-timer;
+ }
+ }
+ }
+ }
+ }
+
+}
\ No newline at end of file
commitNot = new DefaultCommitNotificationProducer(ManagementFactory.getPlatformMBeanServer());
- NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(serverNegotiatorFactory);
- final NetconfServerDispatcher dispatch = new NetconfServerDispatcher(serverChannelInitializer, nettyGroup, nettyGroup);
+ NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcherImpl.ServerChannelInitializer(serverNegotiatorFactory);
+ final NetconfServerDispatcherImpl dispatch = new NetconfServerDispatcherImpl(serverChannelInitializer, nettyGroup, nettyGroup);
ChannelFuture s = dispatch.createServer(netconfAddress);
s.await();
public class NetconfDispatcherImplTest {
private EventLoopGroup nettyGroup;
- private NetconfServerDispatcher dispatch;
+ private NetconfServerDispatcherImpl dispatch;
private DefaultCommitNotificationProducer commitNot;
private HashedWheelTimer hashedWheelTimer;
NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
hashedWheelTimer, factoriesListener, idProvider, 5000, commitNot, ConcurrentClientsTest.createMockedMonitoringService());
- NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(serverNegotiatorFactory);
+ NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcherImpl.ServerChannelInitializer(serverNegotiatorFactory);
- dispatch = new NetconfServerDispatcher(
+ dispatch = new NetconfServerDispatcherImpl(
serverChannelInitializer, nettyGroup, nettyGroup);
}
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
+import org.opendaylight.controller.netconf.api.util.NetconfConstants;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Filter;
doNothing().when(listener).onRemoveNetconfOperationServiceFactory(any(NetconfOperationServiceFactory.class));
doReturn(filter).when(context).createFilter(anyString());
doReturn("").when(reference).toString();
+ doReturn(NetconfConstants.CONFIG_NETCONF_CONNECTOR).when(reference).getProperty(NetconfConstants.SERVICE_NAME);
doReturn(factory).when(context).getService(any(ServiceReference.class));
doReturn("").when(factory).toString();
doNothing().when(listener).onAddNetconfOperationServiceFactory(any(NetconfOperationServiceFactory.class));
import org.opendaylight.controller.netconf.confignetconfconnector.osgi.NetconfOperationServiceFactoryImpl;
import org.opendaylight.controller.netconf.confignetconfconnector.osgi.YangStoreService;
import org.opendaylight.controller.netconf.impl.DefaultCommitNotificationProducer;
-import org.opendaylight.controller.netconf.impl.NetconfServerDispatcher;
+import org.opendaylight.controller.netconf.impl.NetconfServerDispatcherImpl;
import org.opendaylight.controller.netconf.impl.NetconfServerSessionNegotiatorFactory;
import org.opendaylight.controller.netconf.impl.SessionIdProvider;
import org.opendaylight.controller.netconf.impl.osgi.NetconfMonitoringServiceImpl;
}
private Channel startNetconfTcpServer(final NetconfOperationServiceFactoryListenerImpl factoriesListener) throws Exception {
- final NetconfServerDispatcher dispatch = createDispatcher(factoriesListener, getNetconfMonitoringService(), getNotificationProducer());
+ final NetconfServerDispatcherImpl dispatch = createDispatcher(factoriesListener, getNetconfMonitoringService(), getNotificationProducer());
final ChannelFuture s;
if(getTcpServerAddress() instanceof LocalAddress) {
return yangDependencies;
}
- protected NetconfServerDispatcher createDispatcher(
+ protected NetconfServerDispatcherImpl createDispatcher(
final NetconfOperationServiceFactoryListenerImpl factoriesListener, final SessionMonitoringService sessionMonitoringService,
final DefaultCommitNotificationProducer commitNotifier) {
final SessionIdProvider idProvider = new SessionIdProvider();
final NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
hashedWheelTimer, factoriesListener, idProvider, SERVER_CONNECTION_TIMEOUT_MILLIS, commitNotifier, sessionMonitoringService);
- final NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(
+ final NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcherImpl.ServerChannelInitializer(
serverNegotiatorFactory);
- return new NetconfServerDispatcher(serverChannelInitializer, nettyThreadgroup, nettyThreadgroup);
+ return new NetconfServerDispatcherImpl(serverChannelInitializer, nettyThreadgroup, nettyThreadgroup);
}
protected HashedWheelTimer getHashedWheelTimer() {
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
- <configuration>
- <instructions>
- <Export-Package>org.opendaylight.controller.netconf.mapping.api,</Export-Package>
- </instructions>
- </configuration>
</plugin>
+ <plugin>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>config</id>
+ <goals>
+ <goal>generate-sources</goal>
+ </goals>
+ <configuration>
+ <codeGenerators>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator</codeGeneratorClass>
+ <outputBaseDir>${jmxGeneratorPath}</outputBaseDir>
+ <additionalConfiguration>
+ <namespaceToPackage1>urn:opendaylight:params:xml:ns:yang:controller==org.opendaylight.controller.config.yang</namespaceToPackage1>
+ </additionalConfiguration>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.maven.sal.api.gen.plugin.CodeGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>${salGeneratorPath}</outputBaseDir>
+ </generator>
+ </codeGenerators>
+ <inspectDependencies>true</inspectDependencies>
+ </configuration>
+ </execution>
+ </executions>
+ <dependencies>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>yang-jmx-generator-plugin</artifactId>
+ <version>${config.version}</version>
+ </dependency>
+ </dependencies>
+ </plugin>
</plugins>
</build>
</project>
--- /dev/null
+module netconf-northbound-mapper {
+ yang-version 1;
+ namespace "urn:opendaylight:params:xml:ns:yang:controller:netconf:north:mapper";
+ prefix "nnm";
+
+ import config { prefix config; revision-date 2013-04-05; }
+
+ description
+ "This module contains the base YANG definitions for
+ mapping services plugged into a netconf northbound server";
+
+ revision "2015-01-14" {
+ description
+ "Initial revision.";
+ }
+
+ identity netconf-northbound-mapper {
+ base "config:service-type";
+ config:java-class "org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory";
+ }
+
+}
\ No newline at end of file
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Copyright (c) 2015 Cisco Systems, Inc. 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
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>netconf-subsystem</artifactId>
+ <version>0.3.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>netconf-mdsal-config</artifactId>
+ <description>Configuration files for netconf for mdsal</description>
+ <packaging>jar</packaging>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-artifacts</id>
+ <goals>
+ <goal>attach-artifact</goal>
+ </goals>
+ <phase>package</phase>
+ <configuration>
+ <artifacts>
+ <artifact>
+ <file>${project.build.directory}/classes/initial/08-netconf-mdsal.xml</file>
+ <type>xml</type>
+ <classifier>config</classifier>
+ </artifact>
+ </artifacts>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Copyright (c) 2015 Cisco Systems, Inc. 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
+ -->
+
+<!-- vi: set et smarttab sw=4 tabstop=4: -->
+
+<snapshot>
+ <configuration>
+ <data xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
+ <modules xmlns="urn:opendaylight:params:xml:ns:yang:controller:config">
+
+ <module>
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netconf:mdsal:mapper">prefix:netconf-mdsal-mapper</type>
+ <name>netconf-mdsal-mapper</name>
+ <root-schema-service>
+ <type xmlns:dom="urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom">dom:schema-service</type>
+ <name>yang-schema-service</name>
+ </root-schema-service>
+ <dom-broker xmlns="urn:opendaylight:params:xml:ns:yang:controller:netconf:mdsal:mapper">
+ <type xmlns:dom="urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom">dom:dom-async-data-broker</type>
+ <name>inmemory-data-broker</name>
+ </dom-broker>
+ </module>
+
+ <module>
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl">prefix:netconf-server-dispatcher-impl</type>
+ <name>netconf-mdsal-server-dispatcher</name>
+ <mappers xmlns="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl">
+ <type xmlns:dom="urn:opendaylight:params:xml:ns:yang:controller:netconf:north:mapper">dom:netconf-northbound-mapper</type>
+ <name>netconf-mdsal-mapper</name>
+ </mappers>
+ <boss-thread-group xmlns="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netty">prefix:netty-threadgroup</type>
+ <name>global-boss-group</name>
+ </boss-thread-group>
+ <worker-thread-group xmlns="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netty">prefix:netty-threadgroup</type>
+ <name>global-worker-group</name>
+ </worker-thread-group>
+ <timer xmlns="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netty">prefix:netty-timer</type>
+ <name>global-timer</name>
+ </timer>
+ </module>
+
+ <module>
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh">prefix:netconf-northbound-ssh</type>
+ <name>netconf-mdsal-ssh-server</name>
+
+ <event-executor xmlns="urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netty">prefix:netty-event-executor</type>
+ <name>global-event-executor</name>
+ </event-executor>
+ <worker-thread-group xmlns="urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netty">prefix:netty-threadgroup</type>
+ <name>global-worker-group</name>
+ </worker-thread-group>
+ <processing-executor xmlns="urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:threadpool">prefix:threadpool</type>
+ <name>global-netconf-ssh-scheduled-executor</name>
+ </processing-executor>
+ <dispatcher xmlns="urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound">prefix:netconf-server-dispatcher</type>
+ <name>netconf-mdsal-server-dispatcher</name>
+ </dispatcher>
+
+ <username>admin</username>
+ <password>admin</password>
+ </module>
+
+ </modules>
+
+ <services xmlns="urn:opendaylight:params:xml:ns:yang:controller:config">
+ <service>
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netconf:north:mapper">prefix:netconf-northbound-mapper</type>
+ <instance>
+ <name>netconf-mdsal-mapper</name>
+ <provider>/modules/module[type='netconf-mdsal-mapper'][name='netconf-mdsal-mapper']</provider>
+ </instance>
+ </service>
+ <service>
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound">prefix:netconf-server-dispatcher</type>
+ <instance>
+ <name>netconf-mdsal-server-dispatcher</name>
+ <provider>/modules/module[type='netconf-server-dispatcher-impl'][name='netconf-mdsal-server-dispatcher']</provider>
+ </instance>
+ </service>
+ </services>
+
+ </data>
+ </configuration>
+ <required-capabilities>
+ <capability>urn:opendaylight:params:xml:ns:yang:controller:netconf:mdsal:mapper?module=netconf-mdsal-mapper&revision=2015-01-14</capability>
+ <capability>urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh?module=netconf-northbound-ssh&revision=2015-01-14</capability>
+ <capability>urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl?module=netconf-northbound-impl&revision=2015-01-12</capability>
+ <capability>urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl:scheduled?module=threadpool-impl-scheduled&revision=2013-12-01</capability>
+ </required-capabilities>
+</snapshot>
package org.opendaylight.controller.netconf.monitoring.osgi;
import com.google.common.base.Preconditions;
+import java.util.Dictionary;
import java.util.Hashtable;
import org.opendaylight.controller.netconf.api.monitoring.NetconfMonitoringService;
+import org.opendaylight.controller.netconf.api.util.NetconfConstants;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceReference;
final NetconfOperationServiceFactory factory = new NetconfMonitoringActivator.NetconfMonitoringOperationServiceFactory(
operationService);
- reg = context.registerService(NetconfOperationServiceFactory.class, factory, new Hashtable<String, Object>());
+ Dictionary<String, String> properties = new Hashtable<>();
+ properties.put(NetconfConstants.SERVICE_NAME, NetconfConstants.NETCONF_MONITORING);
+ reg = context.registerService(NetconfOperationServiceFactory.class, factory, properties);
return netconfMonitoringService;
}
import com.google.common.collect.Sets;
import java.util.Collection;
import java.util.Collections;
+import java.util.Dictionary;
import java.util.Hashtable;
import java.util.Set;
+import org.opendaylight.controller.netconf.api.util.NetconfConstants;
import org.opendaylight.controller.netconf.mapping.api.Capability;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperation;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationService;
}
};
- operationaServiceRegistration = context.registerService(NetconfOperationServiceFactory.class, netconfOperationServiceFactory, new Hashtable<String, Object>());
+ Dictionary<String, String> properties = new Hashtable<>();
+ properties.put(NetconfConstants.SERVICE_NAME, NetconfConstants.NETCONF_MONITORING);
+ operationaServiceRegistration = context.registerService(NetconfOperationServiceFactory.class, netconfOperationServiceFactory, properties);
}
<groupId>${project.groupId}</groupId>
<artifactId>netconf-util</artifactId>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>netconf-impl</artifactId>
+ </dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>threadpool-config-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>netty-config-api</artifactId>
+ <version>0.3.0-SNAPSHOT</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.opendaylight.yangtools.model</groupId>
+ <artifactId>ietf-inet-types</artifactId>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>config</id>
+ <goals>
+ <goal>generate-sources</goal>
+ </goals>
+ <configuration>
+ <codeGenerators>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator</codeGeneratorClass>
+ <outputBaseDir>${jmxGeneratorPath}</outputBaseDir>
+ <additionalConfiguration>
+ <namespaceToPackage1>urn:opendaylight:params:xml:ns:yang:controller==org.opendaylight.controller.config.yang</namespaceToPackage1>
+ </additionalConfiguration>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.maven.sal.api.gen.plugin.CodeGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>${salGeneratorPath}</outputBaseDir>
+ </generator>
+ </codeGenerators>
+ <inspectDependencies>true</inspectDependencies>
+ </configuration>
+ </execution>
+ </executions>
+ <dependencies>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>yang-jmx-generator-plugin</artifactId>
+ <version>${config.version}</version>
+ </dependency>
+ </dependencies>
+ </plugin>
</plugins>
</build>
--- /dev/null
+package org.opendaylight.controller.config.yang.netconf.northbound.ssh;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.local.LocalAddress;
+import io.netty.util.concurrent.GenericFutureListener;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.concurrent.Executors;
+import org.apache.sshd.server.PasswordAuthenticator;
+import org.apache.sshd.server.keyprovider.PEMGeneratorHostKeyProvider;
+import org.apache.sshd.server.session.ServerSession;
+import org.opendaylight.controller.netconf.api.NetconfServerDispatcher;
+import org.opendaylight.controller.netconf.ssh.SshProxyServer;
+import org.opendaylight.controller.netconf.ssh.SshProxyServerConfigurationBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NetconfNorthboundSshModule extends org.opendaylight.controller.config.yang.netconf.northbound.ssh.AbstractNetconfNorthboundSshModule {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NetconfNorthboundSshModule.class);
+
+ public NetconfNorthboundSshModule(final org.opendaylight.controller.config.api.ModuleIdentifier identifier, final org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+ super(identifier, dependencyResolver);
+ }
+
+ public NetconfNorthboundSshModule(final org.opendaylight.controller.config.api.ModuleIdentifier identifier, final org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, final org.opendaylight.controller.config.yang.netconf.northbound.ssh.NetconfNorthboundSshModule oldModule, final java.lang.AutoCloseable oldInstance) {
+ super(identifier, dependencyResolver, oldModule, oldInstance);
+ }
+
+ @Override
+ public void customValidation() {
+ // add custom validation form module attributes here.
+ }
+
+ @Override
+ public java.lang.AutoCloseable createInstance() {
+ final NetconfServerDispatcher dispatch = getDispatcherDependency();
+
+ final LocalAddress localAddress = new LocalAddress(getPort().toString());
+ final ChannelFuture localServer = dispatch.createLocalServer(localAddress);
+
+ final SshProxyServer sshProxyServer = new SshProxyServer(Executors.newScheduledThreadPool(1), getWorkerThreadGroupDependency(), getEventExecutorDependency());
+
+ final InetSocketAddress bindingAddress = getInetAddress();
+ final SshProxyServerConfigurationBuilder sshProxyServerConfigurationBuilder = new SshProxyServerConfigurationBuilder();
+ sshProxyServerConfigurationBuilder.setBindingAddress(bindingAddress);
+ sshProxyServerConfigurationBuilder.setLocalAddress(localAddress);
+ sshProxyServerConfigurationBuilder.setAuthenticator(new UserAuthenticator(getUsername(), getPassword()));
+ sshProxyServerConfigurationBuilder.setIdleTimeout(Integer.MAX_VALUE);
+ sshProxyServerConfigurationBuilder.setKeyPairProvider(new PEMGeneratorHostKeyProvider());
+
+ localServer.addListener(new GenericFutureListener<ChannelFuture>() {
+
+ @Override
+ public void operationComplete(final ChannelFuture future) {
+ if(future.isDone() && !future.isCancelled()) {
+ try {
+ sshProxyServer.bind(sshProxyServerConfigurationBuilder.createSshProxyServerConfiguration());
+ LOG.info("Netconf SSH endpoint started successfully at {}", bindingAddress);
+ } catch (final IOException e) {
+ throw new RuntimeException("Unable to start SSH netconf server", e);
+ }
+ } else {
+ LOG.warn("Unable to start SSH netconf server at {}", bindingAddress, future.cause());
+ throw new RuntimeException("Unable to start SSH netconf server", future.cause());
+ }
+ }
+ });
+
+ return new NetconfServerCloseable(localServer, sshProxyServer);
+ }
+
+ private InetSocketAddress getInetAddress() {
+ try {
+ final InetAddress inetAd = InetAddress.getByName(getBindingAddress().getIpv4Address() == null ? getBindingAddress().getIpv6Address().getValue() : getBindingAddress().getIpv4Address().getValue());
+ return new InetSocketAddress(inetAd, getPort().getValue());
+ } catch (final UnknownHostException e) {
+ throw new IllegalArgumentException("Unable to bind netconf endpoint to address " + getBindingAddress(), e);
+ }
+ }
+
+ private static final class NetconfServerCloseable implements AutoCloseable {
+ private final ChannelFuture localServer;
+ private final SshProxyServer sshProxyServer;
+
+ public NetconfServerCloseable(final ChannelFuture localServer, final SshProxyServer sshProxyServer) {
+ this.localServer = localServer;
+ this.sshProxyServer = sshProxyServer;
+ }
+
+ @Override
+ public void close() throws Exception {
+ sshProxyServer.close();
+
+ if(localServer.isDone()) {
+ localServer.channel().close();
+ } else {
+ localServer.cancel(true);
+ }
+ }
+ }
+
+
+ private static final class UserAuthenticator implements PasswordAuthenticator {
+
+ private final String username;
+ private final String password;
+
+ public UserAuthenticator(final String username, final String password) {
+ this.username = username;
+ this.password = password;
+ }
+
+ @Override
+ public boolean authenticate(final String username, final String password, final ServerSession session) {
+ // FIXME use aaa stuff here instead
+ return this.username.equals(username) && this.password.equals(password);
+ }
+ }
+}
--- /dev/null
+/*
+* Generated file
+*
+* Generated from: yang module name: netconf-northbound-ssh yang module local name: netconf-northbound-ssh
+* Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
+* Generated at: Mon Feb 09 14:09:07 CET 2015
+*
+* Do not modify this file unless it is present under src/main directory
+*/
+package org.opendaylight.controller.config.yang.netconf.northbound.ssh;
+public class NetconfNorthboundSshModuleFactory extends org.opendaylight.controller.config.yang.netconf.northbound.ssh.AbstractNetconfNorthboundSshModuleFactory {
+
+}
--- /dev/null
+module netconf-northbound-ssh {
+ yang-version 1;
+ namespace "urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh";
+ prefix "nni";
+
+ import netconf-northbound-mapper { prefix nnm; revision-date 2015-01-14; }
+ import netconf-northbound { prefix nn; revision-date 2015-01-14; }
+ import config { prefix config; revision-date 2013-04-05; }
+ import threadpool {prefix th;}
+ import netty {prefix netty;}
+ import ietf-inet-types { prefix inet; revision-date 2010-09-24; }
+
+ organization "Cisco Systems, Inc.";
+
+ description
+ "This module contains the base YANG definitions for
+ a default implementation of netconf northbound server";
+
+ revision "2015-01-14" {
+ description
+ "Initial revision.";
+ }
+
+ identity netconf-northbound-ssh {
+ base config:module-type;
+ config:java-name-prefix NetconfNorthboundSsh;
+ }
+
+ augment "/config:modules/config:module/config:configuration" {
+ case netconf-northbound-ssh {
+ when "/config:modules/config:module/config:type = 'netconf-northbound-ssh'";
+
+ leaf port {
+ type inet:port-number;
+ default 2830;
+ }
+
+ leaf binding-address {
+ type inet:ip-address;
+ default "0.0.0.0";
+ }
+
+ container processing-executor {
+ uses config:service-ref {
+ refine type {
+ mandatory true;
+ config:required-identity th:scheduled-threadpool;
+ }
+ }
+
+ description "Required by the mina-ssh library used in SSH endpoint";
+ }
+
+ container event-executor {
+ uses config:service-ref {
+ refine type {
+ mandatory true;
+ config:required-identity netty:netty-event-executor;
+ }
+ }
+ }
+
+ container worker-thread-group {
+ uses config:service-ref {
+ refine type {
+ config:required-identity netty:netty-threadgroup;
+ }
+ }
+ }
+
+ container dispatcher {
+ uses config:service-ref {
+ refine type {
+ config:required-identity nn:netconf-server-dispatcher;
+ }
+ }
+ }
+
+ // FIXME use auth provider from aaa instead
+ leaf username {
+ description "Specifies username credential";
+ type string;
+ }
+
+ leaf password {
+ description "Specifies password credential";
+ type string;
+ }
+
+
+ }
+ }
+
+}
\ No newline at end of file
import org.opendaylight.controller.netconf.api.monitoring.NetconfManagementSession;
import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants;
import org.opendaylight.controller.netconf.impl.DefaultCommitNotificationProducer;
-import org.opendaylight.controller.netconf.impl.NetconfServerDispatcher;
+import org.opendaylight.controller.netconf.impl.NetconfServerDispatcherImpl;
import org.opendaylight.controller.netconf.impl.NetconfServerSessionNegotiatorFactory;
import org.opendaylight.controller.netconf.impl.SessionIdProvider;
import org.opendaylight.controller.netconf.impl.osgi.NetconfMonitoringServiceImpl;
this.nioExecutor = nioExecutor;
}
- private NetconfServerDispatcher createDispatcher(final Map<ModuleBuilder, String> moduleBuilders, final boolean exi, final int generateConfigsTimeout, final Optional<File> notificationsFile) {
+ private NetconfServerDispatcherImpl createDispatcher(final Map<ModuleBuilder, String> moduleBuilders, final boolean exi, final int generateConfigsTimeout, final Optional<File> notificationsFile) {
final Set<Capability> capabilities = Sets.newHashSet(Collections2.transform(moduleBuilders.keySet(), new Function<ModuleBuilder, Capability>() {
@Override
final NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
hashedWheelTimer, simulatedOperationProvider, idProvider, generateConfigsTimeout, commitNotifier, new LoggingMonitoringService(), serverCapabilities);
- final NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(
+ final NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcherImpl.ServerChannelInitializer(
serverNegotiatorFactory);
- return new NetconfServerDispatcher(serverChannelInitializer, nettyThreadgroup, nettyThreadgroup);
+ return new NetconfServerDispatcherImpl(serverChannelInitializer, nettyThreadgroup, nettyThreadgroup);
}
private Map<ModuleBuilder, String> toModuleBuilders(final Map<SourceIdentifier, Map.Entry<ASTSchemaSource, YangTextSchemaSource>> sources) {
final Map<ModuleBuilder, String> moduleBuilders = parseSchemasToModuleBuilders(params);
- final NetconfServerDispatcher dispatcher = createDispatcher(moduleBuilders, params.exi, params.generateConfigsTimeout, Optional.fromNullable(params.notificationFile));
+ final NetconfServerDispatcherImpl dispatcher = createDispatcher(moduleBuilders, params.exi, params.generateConfigsTimeout, Optional.fromNullable(params.notificationFile));
int currentPort = params.startingPort;
<module>netconf-config</module>
<module>netconf-impl</module>
<module>config-netconf-connector</module>
+ <module>mdsal-netconf-connector</module>
<module>netconf-util</module>
<module>netconf-netty-util</module>
<module>config-persister-impl</module>
<module>ietf-netconf-notifications</module>
<module>ietf-netconf-monitoring-extension</module>
<module>netconf-connector-config</module>
+ <module>netconf-mdsal-config</module>
<module>netconf-auth</module>
<module>netconf-usermanager</module>
<module>netconf-testtool</module>
Code Review / controller.git / commitdiff