[PIP28][PROXY] Pulsar Proxy Gateway Improvement #3915

foreversunyao · 2019-03-27T12:02:19Z

Motivation

https://github.com/apache/pulsar/wiki/PIP-28%3A-Pulsar-Proxy-Gateway-Improvement

Modifications

added a new handler ParserProxyHandler.java to parse requests independently and output

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): no
The public API: no
The schema: don't know
The default values of configurations: yes
The wire protocol: no
The rest endpoints: no
The admin cli options: no
Anything that affects deployment: performance

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? need a doc
Still working on this feature, will make a document after it's almost done

foreversunyao · 2019-03-27T12:03:20Z

Work in progress I am still working on this feature, no need to review currently.

sijie · 2019-03-29T05:19:27Z

@foreversunyao I prepended (WIP) in the caption. feel free to remove it when it is ready for reviews.

foreversunyao · 2019-03-29T11:14:31Z

@foreversunyao I prepended (WIP) in the caption. feel free to remove it when it is ready for reviews.

sure, still testing if it can cover all scenarios

foreversunyao · 2019-04-09T03:16:16Z

run cpp tests

foreversunyao · 2019-04-09T14:52:06Z

@sijie , should be ready to review, though there are two minor differences with original PIP-28.

frameDecoder: when enable full parsing feature, code will keep frameDecoder handler in pipeline(could not be as fast as tcp direct proxy and parsing buffer separately)
lag: it was time cost of proxy processing, but after second thought showing lag(producer lag could be latency and consumer lag) could be more reasonable for pub-sub system

foreversunyao · 2019-04-10T00:45:27Z

run java8 tests

foreversunyao · 2019-04-10T11:22:01Z

run java8 tests

foreversunyao · 2019-04-11T12:05:36Z

run java8 tests

foreversunyao · 2019-04-11T12:06:09Z

tested it locally, and passed

[INFO] -------------------------------------------------------
[INFO]  T E S T S
[INFO] -------------------------------------------------------
[INFO] Running org.apache.pulsar.proxy.server.ProxyLookupThrottlingTest
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 6.585 s - in org.apache.pulsar.proxy.server.ProxyLookupThrottlingTest
[INFO] 
[INFO] Results:
[INFO] 
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
[INFO] 
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS

sijie

overall looks good. left some comments, most of them are around coding styles (i.e. spaces).

also it would be good to have some unit tests or integration tests for this feature. or at least make sure the proxy is working well when proxyLogLevel > 0 as when proxyLogLevel == 0.

sijie · 2019-04-13T21:37:15Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+
+    private Channel channel;
+    //inbound
+    protected static final String frontendConn = "frontendconn";


nit: use upper case for CONSTANTS.

Suggested change

protected static final String frontendConn = "frontendconn";

protected static final String FRONTEND_CONN = "frontendconn";

sijie · 2019-04-13T21:37:35Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+    //inbound
+    protected static final String frontendConn = "frontendconn";
+    //outbound
+    protected static final String backendConn = "backendconn";


Suggested change

protected static final String backendConn = "backendconn";

protected static final String BACKEND_CONN = "backendconn";

sijie · 2019-04-13T21:42:46Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java

@@ -148,6 +148,15 @@
        doc = "Path for the file used to determine the rotation status for the proxy instance"
            + " when responding to service discovery health checks"
    )
+    private Integer proxyLogLevel = 0;
+
+    @FieldContext(


I think the @FieldContext was added to the wrong place. it should be for proxyLogLevel, and the one for proxyLogLevel should be for statusFilePath.

sijie · 2019-04-13T21:43:32Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+        this.connType=type;
+    }
+
+    private void logging (Channel conn,PulsarApi.BaseCommand.Type cmdtype,String info,List<RawMessage> messages) throws Exception{


Suggested change

private void logging (Channel conn,PulsarApi.BaseCommand.Type cmdtype,String info,List<RawMessage> messages) throws Exception{

private void logging(Channel conn, PulsarApi.BaseCommand.Type cmdtype, String info, List<RawMessage> messages) throws Exception {

sijie · 2019-04-13T21:43:42Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+
+    private void logging (Channel conn,PulsarApi.BaseCommand.Type cmdtype,String info,List<RawMessage> messages) throws Exception{
+
+        if (messages !=null){


Suggested change

if (messages !=null){

if (messages !=null) {

sijie · 2019-04-13T21:45:03Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+        if (messages !=null){
+            // lag
+            for (int i=0;i <messages.size();i++){
+                info = info + "["+(System.currentTimeMillis() - messages.get(i).getPublishTime())+"] "+new String(ByteBufUtil.getBytes((messages.get(i)).getData()),"UTF8");


Suggested change

info = info + "["+(System.currentTimeMillis() - messages.get(i).getPublishTime())+"] "+new String(ByteBufUtil.getBytes((messages.get(i)).getData()),"UTF8");

info = info + "["+ (System.currentTimeMillis() - messages.get(i).getPublishTime()) + "] " + new String(ByteBufUtil.getBytes((messages.get(i)).getData()), "UTF8");

sijie · 2019-04-13T21:46:29Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+
+        switch (this.connType){
+            case ParserProxyHandler.frontendConn:
+                log.info("conn:{} cmd:{} msg:{}","["+ conn.remoteAddress().toString()+""+conn.localAddress()+channel.localAddress()+channel.remoteAddress()+"]",cmdtype,info);


I see conn.localAddress() is used twice in this logging statement. is that done intentionally?

Suggested change

log.info("conn:{} cmd:{} msg:{}","["+ conn.remoteAddress().toString()+""+conn.localAddress()+channel.localAddress()+channel.remoteAddress()+"]",cmdtype,info);

log.info("conn:{} cmd:{} msg:{}", "[" + conn.remoteAddress().toString() + conn.localAddress() + channel.localAddress() + channel.remoteAddress() +"]", cmdtype,info);

good catch, actually I was trying to add the feautre(whitelist for client as well as broker side), this need get broker ip/port from inbound channel, but right now it would be in next PR. I will update this.

sijie · 2019-04-13T21:46:51Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+            case ParserProxyHandler.frontendConn:
+                log.info("conn:{} cmd:{} msg:{}","["+ conn.remoteAddress().toString()+""+conn.localAddress()+channel.localAddress()+channel.remoteAddress()+"]",cmdtype,info);
+                break;
+            case ParserProxyHandler.backendConn:


same comments as above

sijie · 2019-04-13T21:47:08Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+        ByteBuf buffer = (ByteBuf)(msg);
+
+        try {
+            //


remove this line

sijie · 2019-04-13T21:47:27Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+
+            log.error("{},{},{}",e.getMessage(),e.getStackTrace(),e.getCause());
+
+        }finally {


Suggested change

}finally {

} finally {

foreversunyao · 2019-04-15T06:21:40Z

@sijie , thanks for reviewing, will update soon.

foreversunyao · 2019-04-15T14:17:18Z

@sijie , I updated the code according to what you mentioned, and also added a unitest ProxyParserTest.java

sijie

LGTM +1

jiazhai · 2019-04-18T06:35:40Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java

+import java.util.Map;
+
+
+public class ParserProxyHandler extends ChannelInboundHandlerAdapter {


Is there any intention to extend the functionality for this class?
Seems currently it is mainly for proxy logging handling. How about rename this class name to make it clearer?

thanks for reviewing.

yes, I would add some whitelist feature on client ip/broker ip ,topic name and msg content.

parsing protocol should be the first step for other future funtions

jiazhai

+1, thanks for the work. left a minor comments.

foreversunyao force-pushed the pip28-v2 branch 3 times, most recently from 547e23c to 9fe61ce Compare March 28, 2019 14:25

sijie changed the title ~~[PIP28][PROXY] Pulsar Proxy Gateway Improvement~~ (WIP) [PIP28][PROXY] Pulsar Proxy Gateway Improvement Mar 29, 2019

foreversunyao force-pushed the pip28-v2 branch 2 times, most recently from 2e65a09 to e562a50 Compare April 8, 2019 12:27

foreversunyao force-pushed the pip28-v2 branch 2 times, most recently from cb6e924 to 67aff21 Compare April 9, 2019 13:24

foreversunyao changed the title ~~(WIP) [PIP28][PROXY] Pulsar Proxy Gateway Improvement~~ [PIP28][PROXY] Pulsar Proxy Gateway Improvement Apr 9, 2019

foreversunyao force-pushed the pip28-v2 branch 4 times, most recently from 53a0c06 to 345fc3a Compare April 9, 2019 14:44

sijie reviewed Apr 13, 2019

View reviewed changes

foreversunyao force-pushed the pip28-v2 branch 3 times, most recently from 5b8c218 to a7d2ced Compare April 15, 2019 14:13

pip28-v2 add unit test

3ad6093

foreversunyao force-pushed the pip28-v2 branch from a7d2ced to 3ad6093 Compare April 15, 2019 14:16

sijie approved these changes Apr 16, 2019

View reviewed changes

sijie assigned foreversunyao Apr 16, 2019

sijie added area/proxy type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages type/feature The PR added a new feature or issue requested a new feature labels Apr 16, 2019

sijie added this to the 2.4.0 milestone Apr 16, 2019

sijie requested review from merlimat and jiazhai April 16, 2019 08:34

jiazhai reviewed Apr 18, 2019

View reviewed changes

jiazhai approved these changes Apr 18, 2019

View reviewed changes

sijie merged commit 95f07f8 into apache:master May 8, 2019

foreversunyao deleted the pip28-v2 branch May 14, 2019 03:00

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[PIP28][PROXY] Pulsar Proxy Gateway Improvement #3915

[PIP28][PROXY] Pulsar Proxy Gateway Improvement #3915

foreversunyao commented Mar 27, 2019

foreversunyao commented Mar 27, 2019

sijie commented Mar 29, 2019

foreversunyao commented Mar 29, 2019

foreversunyao commented Apr 9, 2019

foreversunyao commented Apr 9, 2019

foreversunyao commented Apr 10, 2019

foreversunyao commented Apr 10, 2019

foreversunyao commented Apr 11, 2019

foreversunyao commented Apr 11, 2019 •

edited

Loading

sijie left a comment

sijie Apr 13, 2019

sijie Apr 13, 2019

sijie Apr 13, 2019

sijie Apr 13, 2019

sijie Apr 13, 2019

sijie Apr 13, 2019

sijie Apr 13, 2019

foreversunyao Apr 15, 2019

sijie Apr 13, 2019

sijie Apr 13, 2019

sijie Apr 13, 2019

foreversunyao commented Apr 15, 2019

foreversunyao commented Apr 15, 2019

sijie left a comment

jiazhai Apr 18, 2019

foreversunyao Apr 19, 2019

foreversunyao Apr 19, 2019

jiazhai left a comment

	protected static final String frontendConn = "frontendconn";
	protected static final String FRONTEND_CONN = "frontendconn";

	protected static final String backendConn = "backendconn";
	protected static final String BACKEND_CONN = "backendconn";

	private void logging (Channel conn,PulsarApi.BaseCommand.Type cmdtype,String info,List<RawMessage> messages) throws Exception{
	private void logging(Channel conn, PulsarApi.BaseCommand.Type cmdtype, String info, List<RawMessage> messages) throws Exception {


		private void logging (Channel conn,PulsarApi.BaseCommand.Type cmdtype,String info,List<RawMessage> messages) throws Exception{

		if (messages !=null){

	info = info + "["+(System.currentTimeMillis() - messages.get(i).getPublishTime())+"] "+new String(ByteBufUtil.getBytes((messages.get(i)).getData()),"UTF8");
	info = info + "["+ (System.currentTimeMillis() - messages.get(i).getPublishTime()) + "] " + new String(ByteBufUtil.getBytes((messages.get(i)).getData()), "UTF8");

	log.info("conn:{} cmd:{} msg:{}","["+ conn.remoteAddress().toString()+""+conn.localAddress()+channel.localAddress()+channel.remoteAddress()+"]",cmdtype,info);
	log.info("conn:{} cmd:{} msg:{}", "[" + conn.remoteAddress().toString() + conn.localAddress() + channel.localAddress() + channel.remoteAddress() +"]", cmdtype,info);


		log.error("{},{},{}",e.getMessage(),e.getStackTrace(),e.getCause());

		}finally {

		import java.util.Map;


		public class ParserProxyHandler extends ChannelInboundHandlerAdapter {

[PIP28][PROXY] Pulsar Proxy Gateway Improvement #3915

[PIP28][PROXY] Pulsar Proxy Gateway Improvement #3915

Conversation

foreversunyao commented Mar 27, 2019

Motivation

Modifications

Does this pull request potentially affect one of the following parts:

Documentation

foreversunyao commented Mar 27, 2019

sijie commented Mar 29, 2019

foreversunyao commented Mar 29, 2019

foreversunyao commented Apr 9, 2019

foreversunyao commented Apr 9, 2019

foreversunyao commented Apr 10, 2019

foreversunyao commented Apr 10, 2019

foreversunyao commented Apr 11, 2019

foreversunyao commented Apr 11, 2019 • edited Loading

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

foreversunyao commented Apr 15, 2019

foreversunyao commented Apr 15, 2019

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiazhai left a comment

Choose a reason for hiding this comment

foreversunyao commented Apr 11, 2019 •

edited

Loading