natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1189, enhance qos 

Add property `qos.enable` to control telnet open/close
This commit is contained in:
qinliujie 2018-01-18 15:17:32 +08:00 committed by ken.lj
parent 6f7a146faf
commit 9e8738b5ad
17 changed files with 205 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java
View File

@ -609,6 +609,12 @@ public class Constants {
public static final String REGISTER_IP_KEY = "register.ip";
public static final String QOS_ENABLE = "qos.enable";
public static final String QOS_PORT = "qos.port";
public static final String ACCEPT_FOREIGN_IP = "qos.accept.foreign.ip";
/*
* private Constants(){ }
*/

3
dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/ConcurrentHashSet.java
View File

@ -21,6 +21,7 @@ import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class ConcurrentHashSet<E> extends AbstractSet<E> implements Set<E>, java.io.Serializable {
@ -28,7 +29,7 @@ public class ConcurrentHashSet<E> extends AbstractSet<E> implements Set<E>, java
private static final Object PRESENT = new Object();
private final ConcurrentHashMap<E, Object> map;
private final ConcurrentMap<E, Object> map;
public ConcurrentHashSet() {
map = new ConcurrentHashMap<E, Object>();

33
dubbo-config/dubbo-config-api/src/main/java/com/alibaba/dubbo/config/ApplicationConfig.java
View File

@ -71,6 +71,12 @@ public class ApplicationConfig extends AbstractConfig {
// directory for saving thread dump
private String dumpDirectory;
private Boolean qosEnable;
private Integer qosPort;
private Boolean qosAcceptForeignIp;
// customized parameters
private Map<String, String> parameters;
@ -210,6 +216,33 @@ public class ApplicationConfig extends AbstractConfig {
this.dumpDirectory = dumpDirectory;
}
@Parameter(key = "qos.enable")
public Boolean getQosEnable() {
return qosEnable;
}
public void setQosEnable(Boolean qosEnable) {
this.qosEnable = qosEnable;
}
@Parameter(key = "qos.port")
public Integer getQosPort() {
return qosPort;
}
public void setQosPort(Integer qosPort) {
this.qosPort = qosPort;
}
@Parameter(key = "qos.accept.foreign.ip")
public Boolean getQosAcceptForeignIp() {
return qosAcceptForeignIp;
}
public void setQosAcceptForeignIp(Boolean qosAcceptForeignIp) {
this.qosAcceptForeignIp = qosAcceptForeignIp;
}
public Map<String, String> getParameters() {
return parameters;
}

2
dubbo-demo/dubbo-demo-consumer/src/main/resources/dubbo.properties
View File

@ -1 +1 @@
dubbo.qos.port=33333
dubbo.application.qos.port=33333

2
dubbo-demo/dubbo-demo-provider/src/main/resources/dubbo.properties
View File

@ -1 +1 @@
dubbo.qos.port=22222
dubbo.application.qos.port=22222

4
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/command/impl/Quit.java
View File

@ -19,12 +19,12 @@ package com.alibaba.dubbo.qos.command.impl;
import com.alibaba.dubbo.qos.command.BaseCommand;
import com.alibaba.dubbo.qos.command.CommandContext;
import com.alibaba.dubbo.qos.command.annotation.Cmd;
import com.alibaba.dubbo.qos.common.Constants;
import com.alibaba.dubbo.qos.common.QosConstants;
@Cmd(name = "quit",summary = "quit telnet console")
public class Quit implements BaseCommand {
@Override
public String execute(CommandContext commandContext, String[] args) {
return Constants.CLOSE;
return QosConstants.CLOSE;
}
}

7
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/common/Constants.java → dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/common/QosConstants.java
View File

@ -16,14 +16,11 @@
*/
package com.alibaba.dubbo.qos.common;
public interface Constants {
public interface QosConstants {
int DEFAULT_PORT = 22222;
// system property for specifying qos port
String QOS_PORT = "dubbo.qos.port";
String BR_STR = "\r\n";
String CLOSE = "close!";
// system property for whether to accept foreign IP to connect or not
String ACCEPT_FOREIGN_IP = "dubbo.qos.accept.foreign.ip";
}

95
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/protocol/QosProtocolWrapper.java Normal file
View File

@ -0,0 +1,95 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 com.alibaba.dubbo.qos.protocol;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.qos.server.Server;
import com.alibaba.dubbo.rpc.Exporter;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.Protocol;
import com.alibaba.dubbo.rpc.RpcException;
import java.util.concurrent.atomic.AtomicBoolean;
import static com.alibaba.dubbo.common.Constants.ACCEPT_FOREIGN_IP;
import static com.alibaba.dubbo.common.Constants.QOS_ENABLE;
import static com.alibaba.dubbo.common.Constants.QOS_PORT;
public class QosProtocolWrapper implements Protocol {
private static AtomicBoolean hasStarted = new AtomicBoolean(false);
private Protocol protocol;
public QosProtocolWrapper(Protocol protocol) {
if (protocol == null) {
throw new IllegalArgumentException("protocol == null");
}
this.protocol = protocol;
}
@Override
public int getDefaultPort() {
return protocol.getDefaultPort();
}
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
startQosServer(invoker.getUrl());
return protocol.export(invoker);
}
return protocol.export(invoker);
}
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
startQosServer(url);
return protocol.refer(type, url);
}
return protocol.refer(type, url);
}
@Override
public void destroy() {
protocol.destroy();
}
private void startQosServer(URL url) {
if (!hasStarted.compareAndSet(false, true)) {
return;
}
try {
boolean qosEnable = Boolean.parseBoolean(url.getParameter(QOS_ENABLE,"true"));
if (!qosEnable) {
return;
}
int port = Integer.parseInt(url.getParameter(QOS_PORT,"22222"));
boolean acceptForeignIp = Boolean.parseBoolean(url.getParameter(ACCEPT_FOREIGN_IP,"true"));
Server server = com.alibaba.dubbo.qos.server.Server.getInstance();
server.setPort(port);
server.setAcceptForeignIp(acceptForeignIp);
server.start();
} catch (Throwable throwable) {
//throw new RpcException("fail to start qos server", throwable);
}
}
}

2
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/server/DubboLogo.java
View File

@ -18,7 +18,7 @@ package com.alibaba.dubbo.qos.server;
public class DubboLogo {
public static String dubbo =
" ████████▄ ███ █▄ ▀█████████▄ ▀█████████▄ ▄██████▄ \n" +
" ████████▄ ███ █▄ ▀█████████▄ ▀█████████▄ ▄██████▄ \n" +
" ███ ▀███ ███ ███ ███ ███ ███ ███ ███ ███ \n" +
" ███ ███ ███ ███ ███ ███ ███ ███ ███ ███ \n" +
" ███ ███ ███ ███ ▄███▄▄▄██▀ ▄███▄▄▄██▀ ███ ███ \n" +

30
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/server/Server.java
View File

@ -18,8 +18,6 @@ package com.alibaba.dubbo.qos.server;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.ConfigUtils;
import com.alibaba.dubbo.qos.common.Constants;
import com.alibaba.dubbo.qos.server.handler.QosProcessHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
@ -49,11 +47,9 @@ public class Server {
return INSTANCE;
}
private int port = Integer.parseInt(ConfigUtils.getProperty(Constants.QOS_PORT, Constants.DEFAULT_PORT + ""));
private int port;
public int getPort() {
return port;
}
private boolean acceptForeignIp = true;
private EventLoopGroup boss;
@ -74,6 +70,10 @@ public class Server {
this.welcome = welcome;
}
public int getPort() {
return port;
}
/**
* start server, bind port
*/
@ -92,7 +92,7 @@ public class Server {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new QosProcessHandler(welcome));
ch.pipeline().addLast(new QosProcessHandler(welcome, acceptForeignIp));
}
});
try {
@ -116,4 +116,20 @@ public class Server {
worker.shutdownGracefully();
}
}
public void setPort(int port) {
this.port = port;
}
public boolean isAcceptForeignIp() {
return acceptForeignIp;
}
public void setAcceptForeignIp(boolean acceptForeignIp) {
this.acceptForeignIp = acceptForeignIp;
}
public String getWelcome() {
return welcome;
}
}

13
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/server/handler/LocalHostPermitHandler.java
View File

@ -17,8 +17,7 @@
package com.alibaba.dubbo.qos.server.handler;
import com.alibaba.dubbo.common.utils.ConfigUtils;
import com.alibaba.dubbo.qos.common.Constants;
import com.alibaba.dubbo.qos.common.QosConstants;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@ -31,14 +30,18 @@ import java.net.InetSocketAddress;
public class LocalHostPermitHandler extends ChannelHandlerAdapter {
// true means to accept foreign IP
private static boolean acceptForeignIp = Boolean.valueOf(ConfigUtils.getProperty(Constants.ACCEPT_FOREIGN_IP, "true"));
private boolean acceptForeignIp;
public LocalHostPermitHandler(boolean acceptForeignIp) {
this.acceptForeignIp = acceptForeignIp;
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (!acceptForeignIp) {
if (!((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().isLoopbackAddress()) {
ByteBuf cb = Unpooled.wrappedBuffer((Constants.BR_STR + "Foreign Ip Not Permitted."
+ Constants.BR_STR).getBytes());
ByteBuf cb = Unpooled.wrappedBuffer((QosConstants.BR_STR + "Foreign Ip Not Permitted."
+ QosConstants.BR_STR).getBytes());
ctx.writeAndFlush(cb).addListener(ChannelFutureListener.CLOSE);
}
}

28
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/server/handler/QosProcessHandler.java
View File

@ -34,21 +34,24 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
public class QosProcessHandler extends ByteToMessageDecoder {
private ScheduledFuture<?> welcomeFuture;
private String welcome;
// true means to accept foreign IP
private boolean acceptForeignIp;
public static String prompt = "dubbo>";
public QosProcessHandler(String welcome){
public QosProcessHandler(String welcome, boolean acceptForeignIp) {
this.welcome = welcome;
this.acceptForeignIp = acceptForeignIp;
}
@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
welcomeFuture = ctx.executor().schedule(new Runnable() {
@Override
public void run() {
if (welcome != null) {
@ -56,21 +59,21 @@ public class QosProcessHandler extends ByteToMessageDecoder {
ctx.writeAndFlush(Unpooled.wrappedBuffer(prompt.getBytes()));
}
}
}, 500, TimeUnit.MILLISECONDS);
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if (in.readableBytes() < 1) {
return;
}
// read one byte to guess protocol
final int magic = in.getByte(in.readerIndex());
ChannelPipeline p = ctx.pipeline();
p.addLast(new LocalHostPermitHandler());
p.addLast(new LocalHostPermitHandler(acceptForeignIp));
if (isHttp(magic)) {
// no welcome output for http protocol
if (welcomeFuture != null && welcomeFuture.isCancellable()) {
@ -84,11 +87,12 @@ public class QosProcessHandler extends ByteToMessageDecoder {
p.addLast(new LineBasedFrameDecoder(2048));
p.addLast(new StringDecoder(CharsetUtil.UTF_8));
p.addLast(new StringEncoder(CharsetUtil.UTF_8));
p.addLast(new IdleStateHandler(0,0,5 * 60));
p.addLast(new IdleStateHandler(0, 0, 5 * 60));
p.addLast(new TelnetProcessHandler());
p.remove(this);
}
}
// G for GET, and P for POST
private static boolean isHttp(int magic) {
return magic == 'G' || magic == 'P';

10
dubbo-plugin/dubbo-qos/src/main/java/com/alibaba/dubbo/qos/server/handler/TelnetProcessHandler.java
View File

@ -24,7 +24,7 @@ import com.alibaba.dubbo.qos.command.CommandExecutor;
import com.alibaba.dubbo.qos.command.DefaultCommandExecutor;
import com.alibaba.dubbo.qos.command.NoSuchCommandException;
import com.alibaba.dubbo.qos.command.decoder.TelnetCommandDecoder;
import com.alibaba.dubbo.qos.common.Constants;
import com.alibaba.dubbo.qos.common.QosConstants;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
@ -50,18 +50,18 @@ public class TelnetProcessHandler extends SimpleChannelInboundHandler<String> {
try {
String result = commandExecutor.execute(commandContext);
if (StringUtils.equals(Constants.CLOSE, result)) {
if (StringUtils.equals(QosConstants.CLOSE, result)) {
ctx.writeAndFlush(getByeLabel()).addListener(ChannelFutureListener.CLOSE);
} else {
ctx.writeAndFlush(result + Constants.BR_STR + QosProcessHandler.prompt);
ctx.writeAndFlush(result + QosConstants.BR_STR + QosProcessHandler.prompt);
}
} catch (NoSuchCommandException ex) {
ctx.writeAndFlush(msg + " :no such command");
ctx.writeAndFlush(Constants.BR_STR + QosProcessHandler.prompt);
ctx.writeAndFlush(QosConstants.BR_STR + QosProcessHandler.prompt);
log.error("can not found command " + commandContext, ex);
} catch (Exception ex) {
ctx.writeAndFlush(msg + " :fail to execute commandContext by " + ex.getMessage());
ctx.writeAndFlush(Constants.BR_STR + QosProcessHandler.prompt);
ctx.writeAndFlush(QosConstants.BR_STR + QosProcessHandler.prompt);
log.error("execute commandContext got exception " + commandContext, ex);
}
}

1
dubbo-plugin/dubbo-qos/src/main/resources/META-INF/dubbo/internal/com.alibaba.dubbo.rpc.Protocol Normal file
View File

@ -0,0 +1 @@
qos=com.alibaba.dubbo.qos.protocol.QosProtocolWrapper

9
dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryProtocol.java
View File

@ -43,6 +43,10 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import static com.alibaba.dubbo.common.Constants.ACCEPT_FOREIGN_IP;
import static com.alibaba.dubbo.common.Constants.QOS_ENABLE;
import static com.alibaba.dubbo.common.Constants.QOS_PORT;
/**
* RegistryProtocol
*
@ -237,7 +241,10 @@ public class RegistryProtocol implements Protocol {
final URL registedProviderUrl = providerUrl.removeParameters(getFilteredKeys(providerUrl))
.removeParameter(Constants.MONITOR_KEY)
.removeParameter(Constants.BIND_IP_KEY)
.removeParameter(Constants.BIND_PORT_KEY);
.removeParameter(Constants.BIND_PORT_KEY)
.removeParameter(QOS_ENABLE)
.removeParameter(QOS_PORT)
.removeParameter(ACCEPT_FOREIGN_IP);
return registedProviderUrl;
}

2
dubbo-remoting/dubbo-remoting-api/src/main/java/com/alibaba/dubbo/remoting/transport/ChannelHandlerDelegate.java
View File

@ -19,5 +19,5 @@ package com.alibaba.dubbo.remoting.transport;
import com.alibaba.dubbo.remoting.ChannelHandler;
public interface ChannelHandlerDelegate extends ChannelHandler {
public ChannelHandler getHandler();
ChannelHandler getHandler();
}

12
dubbo-rpc/dubbo-rpc-api/src/main/java/com/alibaba/dubbo/rpc/protocol/ProtocolListenerWrapper.java
View File

@ -35,18 +35,6 @@ import java.util.Collections;
* ListenerProtocol
*/
public class ProtocolListenerWrapper implements Protocol {
static {
try {
Class serverClass = Protocol.class.getClassLoader().loadClass("com.alibaba.dubbo.qos.server.Server");
Method serverGetInstanceMethod = serverClass.getMethod("getInstance");
Object serverInstance = serverGetInstanceMethod.invoke(null);
Method startMethod = serverClass.getMethod("start");
startMethod.invoke(serverInstance);
}catch (Throwable throwable){
}
}
private final Protocol protocol;