홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
YuanyouZhang
Committed by Gerrit Code Review
ebe1261e 1 parent 41ad728f

[ONOS-2393]--OVSDB- The implementation of the netty controller. 

Change-Id: Ib24f5e6eb242a977ff6a1debdc22dcc09ff205c9
Inline Side-by-side
Showing 7 changed files with 579 additions and 52 deletions
<?xml version="1.0"?>
<!--
~ Copyright 2015 Open Networking Laboratory
~
~ Licensed under the Apache License, Version 2.0 (the "License");
~ you may not use this file except in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<project
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
......
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2015 Open Networking Laboratory
~
~ Licensed under the Apache License, Version 2.0 (the "License");
~ you may not use this file except in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.onosproject</groupId>
<artifactId>onos-ovsdb</artifactId>
<version>1.3.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>onos-ovsdb-ctl</artifactId>
<packaging>bundle</packaging>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.felix</groupId>
<artifactId>org.apache.felix.scr.annotations</artifactId>
</dependency>
<dependency>
<groupId>org.osgi</groupId>
<artifactId>org.osgi.compendium</artifactId>
</dependency>
<dependency>
<groupId>org.onosproject</groupId>
<artifactId>onos-ovsdb-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.onosproject</groupId>
<artifactId>onos-ovsdb-rfc</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-scr-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.ovsdb.controller.impl;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.onosproject.ovsdb.controller.driver.OvsdbProviderService;
/**
* The listener class. Handles when the node disconnect.
*/
public class ChannelConnectionListener implements ChannelFutureListener {
private final OvsdbProviderService providerService;
/**
* Constructor from a OvsdbProviderService providerService.
*
* @param providerService the providerService to use
*/
public ChannelConnectionListener(OvsdbProviderService providerService) {
this.providerService = providerService;
}
@Override
public void operationComplete(ChannelFuture arg0) {
providerService.nodeRemoved();
}
}
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.ovsdb.controller.impl;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.ServerChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.onlab.packet.IpAddress;
import org.onosproject.ovsdb.controller.OvsdbConstant;
import org.onosproject.ovsdb.controller.OvsdbNodeId;
import org.onosproject.ovsdb.controller.driver.DefaultOvsdbClient;
import org.onosproject.ovsdb.controller.driver.OvsdbAgent;
import org.onosproject.ovsdb.controller.driver.OvsdbProviderService;
import org.onosproject.ovsdb.rfc.jsonrpc.Callback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* The main controller class. Handles all setup and network listeners -
* Distributed ovsdbClient.
*/
public class Controller {
protected static final Logger log = LoggerFactory
.getLogger(Controller.class);
private int ovsdbPort = OvsdbConstant.OVSDBPORT;
private OvsdbAgent agent;
private Callback monitorCallback;
private final ExecutorService executorService = Executors
.newFixedThreadPool(10);
private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
private Class<? extends ServerChannel> serverChannelClass;
/**
* Initialization.
*/
private void initEventLoopGroup() {
bossGroup = new NioEventLoopGroup();
workerGroup = new NioEventLoopGroup();
serverChannelClass = NioServerSocketChannel.class;
}
/**
* Accepts incoming connections.
*/
private void startAcceptingConnections() throws InterruptedException {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(serverChannelClass)
.childHandler(new OnosCommunicationChannelInitializer());
b.option(ChannelOption.SO_BACKLOG, 128);
b.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
b.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);
b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
b.childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture cf = b.bind(ovsdbPort).sync();
cf.channel().closeFuture().sync();
}
/**
* Tells controller that we're ready to accept ovsdb node loop.
*/
public void run() throws InterruptedException {
initEventLoopGroup();
startAcceptingConnections();
}
/**
* Adds channel pipiline to handle a new connected node.
*/
private class OnosCommunicationChannelInitializer
extends ChannelInitializer<SocketChannel> {
protected void initChannel(SocketChannel channel) throws Exception {
log.info("New channel created");
channel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
channel.pipeline().addLast(new MessageDecoder());
handleNewNodeConnection(channel);
}
}
/**
* Handles the new connection of a node.
*
* @param channel the channel to use.
*/
private void handleNewNodeConnection(final Channel channel) {
executorService.execute(new Runnable() {
@Override
public void run() {
log.info("Handle new node connection");
IpAddress ipAddress = IpAddress
.valueOf(((InetSocketAddress) channel.remoteAddress())
.getAddress().getHostAddress());
long port = ((InetSocketAddress) channel.remoteAddress())
.getPort();
log.info("Get connection from ip address {} : {}",
ipAddress.toString(), port);
OvsdbNodeId nodeId = new OvsdbNodeId(ipAddress, port);
OvsdbProviderService ovsdbProviderService = getNodeInstance(nodeId,
agent,
monitorCallback,
channel);
ovsdbProviderService.setConnection(true);
OvsdbJsonRpcHandler ovsdbJsonRpcHandler = new OvsdbJsonRpcHandler(
nodeId);
ovsdbJsonRpcHandler
.setOvsdbProviderService(ovsdbProviderService);
channel.pipeline().addLast(ovsdbJsonRpcHandler);
ovsdbProviderService.nodeAdded();
ChannelFuture closeFuture = channel.closeFuture();
closeFuture
.addListener(new ChannelConnectionListener(
ovsdbProviderService));
}
});
}
/**
* Gets an ovsdb client instance.
*
* @param nodeId data ovsdb node id
* @param agent OvsdbAgent
* @param monitorCallback Callback
* @param channel Channel
* @return OvsdbProviderService instance
*/
protected OvsdbProviderService getNodeInstance(OvsdbNodeId nodeId,
OvsdbAgent agent,
Callback monitorCallback,
Channel channel) {
OvsdbProviderService ovsdbProviderService = new DefaultOvsdbClient(
nodeId);
ovsdbProviderService.setAgent(agent);
ovsdbProviderService.setCallback(monitorCallback);
ovsdbProviderService.setChannel(channel);
return ovsdbProviderService;
}
/**
* Starts controller.
*
* @param agent OvsdbAgent
* @param monitorCallback Callback
*/
public void start(OvsdbAgent agent, Callback monitorCallback) {
this.agent = agent;
this.monitorCallback = monitorCallback;
try {
this.run();
} catch (InterruptedException e) {
log.warn("Interrupted while waiting to start");
Thread.currentThread().interrupt();
}
}
/**
* Stops controller.
*
*/
public void stop() {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.ovsdb.controller.impl;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.util.List;
import org.onosproject.ovsdb.rfc.jsonrpc.JsonReadContext;
import org.onosproject.ovsdb.rfc.utils.JsonRpcReaderUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Decoder for inbound messages.
*/
public class MessageDecoder extends ByteToMessageDecoder {
private final Logger log = LoggerFactory.getLogger(MessageDecoder.class);
private final JsonReadContext context = new JsonReadContext();
/**
* Default constructor.
*/
public MessageDecoder() {
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buf,
List<Object> out) throws Exception {
log.debug("Message decoder");
JsonRpcReaderUtil.readToJsonNode(buf, out, context);
}
@Override
public void exceptionCaught(ChannelHandlerContext context, Throwable cause) {
log.error("Exception inside channel handling pipeline.", cause);
context.close();
}
}
/*
* Copyright 2015 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.ovsdb.controller.impl;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.onosproject.ovsdb.controller.OvsdbNodeId;
import org.onosproject.ovsdb.controller.driver.OvsdbProviderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.base.Strings;
/**
* Channel handler deals with the node connection and dispatches
* ovsdb messages to the appropriate locations.
*/
public final class OvsdbJsonRpcHandler extends ChannelInboundHandlerAdapter {
protected static final Logger log = LoggerFactory
.getLogger(OvsdbJsonRpcHandler.class);
private OvsdbNodeId ovsdbNodeId;
private OvsdbProviderService ovsdbProviderService;
/**
* Constructor from a OvsdbNodeId ovsdbNodeId.
*
* @param ovsdbNodeId the ovsdbNodeId to use
*/
public OvsdbJsonRpcHandler(OvsdbNodeId ovsdbNodeId) {
super();
this.ovsdbNodeId = ovsdbNodeId;
}
/**
* Gets the ovsdbProviderService instance.
*
* @return the instance of the ovsdbProviderService
*/
public OvsdbProviderService getOvsdbProviderService() {
return ovsdbProviderService;
}
/**
* Sets the ovsdbProviderService instance.
*
* @param ovsdbNodeDriver the ovsdbNodeDriver to use
*/
public void setOvsdbProviderService(OvsdbProviderService ovsdbNodeDriver) {
this.ovsdbProviderService = ovsdbNodeDriver;
}
/**
* Gets the OvsdbNodeId instance.
*
* @return the instance of the OvsdbNodeId
*/
public OvsdbNodeId getNodeId() {
return ovsdbNodeId;
}
/**
* Sets the ovsdb node id.
*
* @param ovsdbNodeId the ovsdbNodeId to use
*/
public void setNodeId(OvsdbNodeId ovsdbNodeId) {
this.ovsdbNodeId = ovsdbNodeId;
}
/**
* Processes an JsonNode message received on the channel.
*
* @param jsonNode The OvsdbJsonRpcHandler that received the message
*/
private void processOvsdbMessage(JsonNode jsonNode) {
log.info("Handle ovsdb message");
if (jsonNode.has("result")) {
log.debug("Handle ovsdb result");
ovsdbProviderService.processResult(jsonNode);
} else if (jsonNode.hasNonNull("method")) {
log.debug("Handle ovsdb request");
if (jsonNode.has("id")
&& !Strings.isNullOrEmpty(jsonNode.get("id").asText())) {
ovsdbProviderService.processRequest(jsonNode);
}
}
return;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
log.debug("Receive message from ovsdb");
if (msg instanceof JsonNode) {
JsonNode jsonNode = (JsonNode) msg;
processOvsdbMessage(jsonNode);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext context, Throwable cause) {
log.error("Exception inside channel handling pipeline.", cause);
context.close();
}
}
<?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.onosproject</groupId>
<artifactId>onos</artifactId>
<version>1.3.0-SNAPSHOT</version>
</parent>
<artifactId>onos-ovsdb</artifactId>
<name>onos-ovsdb</name>
<packaging>pom</packaging>
<description>ONOS OVSDB southbound plugin</description>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.onosproject</groupId>
<artifactId>onlab-misc</artifactId>
</dependency>
<dependency>
<groupId>org.onosproject</groupId>
<artifactId>onlab-junit</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-buffer</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
</plugin>
</plugins>
</build>
<modules>
<module>api</module>
<module>rfc</module>
</modules>
</project>
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2015 Open Networking Laboratory
~
~ Licensed under the Apache License, Version 2.0 (the "License");
~ you may not use this file except in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.onosproject</groupId>
<artifactId>onos</artifactId>
<version>1.3.0-SNAPSHOT</version>
</parent>
<artifactId>onos-ovsdb</artifactId>
<name>onos-ovsdb</name>
<packaging>pom</packaging>
<description>ONOS OVSDB southbound plugin</description>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.onosproject</groupId>
<artifactId>onlab-misc</artifactId>
</dependency>
<dependency>
<groupId>org.onosproject</groupId>
<artifactId>onlab-junit</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-buffer</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
</plugin>
</plugins>
</build>
<modules>
<module>api</module>
<module>rfc</module>
<module>ctl</module>
</modules>
</project>
\ No newline at end of file
......