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

add qos protocol to pu server (#10395) 

* add qos to pu server

* add qos enable flag set

* null pointer Exception of mock url

* revoke pu actions to connection establishment

* telnet detector adjust

* frameworkModel already exists

* comment code style

* qos comments fix

* add telnet ayt detect

* change func name to lower camel case

* frameworkModel inject in Constructor

* frameworkModel inject in Constructor
This commit is contained in:
TrueAbc 2022-08-15 09:47:39 +08:00 committed by GitHub
parent e528bca7e5
commit 6d85688cc4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 266 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/protocol/QosProtocolWrapper.java
View File

@ -21,7 +21,9 @@ import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.qos.common.QosConstants;
import org.apache.dubbo.qos.pu.QosWireProtocol;
import org.apache.dubbo.qos.server.Server;
import org.apache.dubbo.remoting.api.WireProtocol;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
@ -96,6 +98,10 @@ public class QosProtocolWrapper implements Protocol, ScopeModelAware {
}
boolean qosEnable = url.getParameter(QOS_ENABLE, true);
WireProtocol qosWireProtocol = frameworkModel.getExtensionLoader(WireProtocol.class).getExtension("qos");
if(qosWireProtocol != null) {
((QosWireProtocol) qosWireProtocol).setQosEnable(qosEnable);
}
if (!qosEnable) {
logger.info("qos won't be started because it is disabled. " +
"Please check dubbo.application.qos.enable is configured either in system property, " +

56
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/QosDetector.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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 org.apache.dubbo.qos.pu;
import org.apache.dubbo.remoting.api.ProtocolDetector;
import org.apache.dubbo.remoting.buffer.ChannelBuffer;
import org.apache.dubbo.rpc.model.FrameworkModel;
public class QosDetector implements ProtocolDetector {
private final QosHTTP1Detector qosHTTP1Detector = new QosHTTP1Detector();
private final TelnetDetector telnetDetector;
private boolean QosEnableFlag = true;
public void setQosEnableFlag(boolean qosEnableFlag) {
QosEnableFlag = qosEnableFlag;
}
public QosDetector(FrameworkModel frameworkModel) {
this.telnetDetector = new TelnetDetector(frameworkModel);
}
@Override
public Result detect(ChannelBuffer in) {
if(!QosEnableFlag) {
return Result.UNRECOGNIZED;
}
Result h1Res = qosHTTP1Detector.detect(in);
if(h1Res.equals(Result.RECOGNIZED)) {
return h1Res;
}
Result telRes = telnetDetector.detect(in);
if(telRes.equals(Result.RECOGNIZED)) {
return telRes;
}
if(h1Res.equals(Result.NEED_MORE_DATA) || telRes.equals(Result.NEED_MORE_DATA)) {
return Result.NEED_MORE_DATA;
}
return Result.UNRECOGNIZED;
}
}

39
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/QosHTTP1Detector.java Normal file
View File

@ -0,0 +1,39 @@
/*
* 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 org.apache.dubbo.qos.pu;
import org.apache.dubbo.remoting.api.ProtocolDetector;
import org.apache.dubbo.remoting.buffer.ChannelBuffer;
public class QosHTTP1Detector implements ProtocolDetector {
private static boolean isHttp(int magic) {
return magic == 'G' || magic == 'P';
}
@Override
public Result detect(ChannelBuffer in) {
if (in.readableBytes() < 2) {
return Result.NEED_MORE_DATA;
}
final int magic = in.getByte(in.readerIndex());
// h2 starts with "PR"
if (isHttp(magic) && in.getByte(in.readerIndex()+1) != 'R' ){
return Result.RECOGNIZED;
}
return Result.UNRECOGNIZED;
}
}

63
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/QosWireProtocol.java Normal file
View File

@ -0,0 +1,63 @@
/*
* 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 org.apache.dubbo.qos.pu;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.qos.server.DubboLogo;
import org.apache.dubbo.qos.server.handler.QosProcessHandler;
import org.apache.dubbo.remoting.ChannelHandler;
import org.apache.dubbo.remoting.api.AbstractWireProtocol;
import org.apache.dubbo.remoting.api.pu.ChannelHandlerPretender;
import org.apache.dubbo.remoting.api.pu.ChannelOperator;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ScopeModelAware;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.ssl.SslContext;
import java.util.ArrayList;
import java.util.List;
@Activate
public class QosWireProtocol extends AbstractWireProtocol implements ScopeModelAware {
public QosWireProtocol(FrameworkModel frameworkModel) {
super(new QosDetector(frameworkModel));
}
public void setQosEnable(boolean flag) {
((QosDetector)this.detector()).setQosEnableFlag(flag);
}
@Override
public void configServerProtocolHandler(URL url, ChannelOperator operator) {
// add qosProcess handler
QosProcessHandler handler = new QosProcessHandler(url.getOrDefaultFrameworkModel(),
DubboLogo.DUBBO, false);
List<ChannelHandler> handlers = new ArrayList<>();
handlers.add(new ChannelHandlerPretender(handler));
operator.configChannelHandler(handlers);
}
@Override
public void configClientPipeline(URL url, ChannelPipeline pipeline, SslContext sslContext) {
}
}

95
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/pu/TelnetDetector.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 org.apache.dubbo.qos.pu;
import org.apache.dubbo.qos.command.BaseCommand;
import org.apache.dubbo.qos.command.CommandContext;
import org.apache.dubbo.qos.command.decoder.TelnetCommandDecoder;
import org.apache.dubbo.remoting.api.ProtocolDetector;
import org.apache.dubbo.remoting.buffer.ChannelBuffer;
import org.apache.dubbo.remoting.buffer.ChannelBuffers;
import org.apache.dubbo.remoting.buffer.HeapChannelBuffer;
import org.apache.dubbo.rpc.model.FrameworkModel;
import io.netty.util.CharsetUtil;
import static java.lang.Math.min;
public class TelnetDetector implements ProtocolDetector {
private FrameworkModel frameworkModel;
private final int MaxSize = 2048;
private final ChannelBuffer AytPreface = new HeapChannelBuffer(new byte[]{(byte) 0xff, (byte) 0xf6});
public TelnetDetector(FrameworkModel frameworkModel) {
this.frameworkModel = frameworkModel;
}
@Override
public Result detect(ChannelBuffer in) {
if (in.readableBytes() >= MaxSize) {
return Result.UNRECOGNIZED;
}
Result resCommand = commandDetect(in);
if (resCommand.equals(Result.RECOGNIZED)){
return resCommand;
}
Result resAyt = telnetAytDetect(in);
if (resAyt.equals(Result.RECOGNIZED)) {
return resAyt;
}
if (resAyt.equals(Result.UNRECOGNIZED) && resCommand.equals(Result.UNRECOGNIZED)) {
return Result.UNRECOGNIZED;
}
return Result.NEED_MORE_DATA;
}
private Result commandDetect(ChannelBuffer in) {
// detect if remote channel send a qos command to server
ChannelBuffer back = in.copy();
byte[] backBytes = new byte[back.readableBytes()];
back.getBytes(back.readerIndex(), backBytes);
String s = new String(backBytes, CharsetUtil.UTF_8);
// trim /r/n to let parser work for input
s = s.trim();
CommandContext commandContext = TelnetCommandDecoder.decode(s);
if(frameworkModel.getExtensionLoader(BaseCommand.class).hasExtension(commandContext.getCommandName())){
return Result.RECOGNIZED;
}
return Result.UNRECOGNIZED;
}
private Result telnetAytDetect(ChannelBuffer in) {
// detect if remote channel send a telnet ayt command to server
int prefaceLen = AytPreface.readableBytes();
int bytesRead = min(in.readableBytes(), prefaceLen);
if(bytesRead == 0 || !ChannelBuffers.prefixEquals(in, AytPreface, bytesRead)) {
return Result.UNRECOGNIZED;
}
if(bytesRead == prefaceLen) {
// we need to consume preface because it's not a qos command
// consume and remember to mark, pu server handler reset reader index
in.readBytes(AytPreface.readableBytes());
in.markReaderIndex();
return Result.RECOGNIZED;
}
return Result.NEED_MORE_DATA;
}
}

1
dubbo-plugin/dubbo-qos/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.api.WireProtocol Normal file
View File

@ -0,0 +1 @@
qos=org.apache.dubbo.qos.pu.QosWireProtocol

6
dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java
View File

@ -77,7 +77,8 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
// Will use the first five bytes to detect a protocol.
if (in.readableBytes() < 5) {
// size of telnet command ls is 2 bytes
if (in.readableBytes() < 2) {
return;
}
@ -127,7 +128,8 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
}
private boolean isSsl(ByteBuf buf) {
if (detectSsl) {
// at least 5 bytes to determine if data is encrypted
if (detectSsl && buf.readableBytes() >= 5) {
return SslHandler.isEncrypted(buf);
}
return false;

2
dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServerHandler.java
View File

@ -95,6 +95,8 @@ public class NettyServerHandler extends ChannelDuplexHandler {
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
handler.received(channel, msg);
// trigger qos handler
ctx.fireChannelRead(msg);
}