You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "gunli (via GitHub)" <gi...@apache.org> on 2023/05/29 07:50:45 UTC
[GitHub] [inlong] gunli opened a new pull request, #8111: [INLONG-4961][DataProxy] add dataproxy-sdk-golang
gunli opened a new pull request, #8111:
URL: https://github.com/apache/inlong/pull/8111
### [INLONG-4961][DataProxy] add dataproxy-sdk-golang
- Fixes #4961
### Motivation
Add a golang version SDK of InLong Data Proxy
### Modifications
- Add a discoverer to do service discovery;
- Add a conn pool to keep the connections to the server;
- Add a Client interface to handle request;
- Add a worker to do the data send/receive work;
- Add other util packages/functions to help the whole send/receive jobs;
### Verifying this change
*(Please pick either of the following options)*
- [ ] This change is a trivial rework/code cleanup without any test coverage.
- [ ] This change is already covered by existing tests, such as:
*(please describe tests)*
- [x] This change added tests and can be verified as follows:
- Prepare a groupID and some streamIDs in the server side and define the log schema of the streamIDs;
- Write a test client to send messages to the server, example codes can be found in: dataproxy/example_test.go;
- Verify how many message was sent and received.
### Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (not documented)
- I will create a follow-up issue for adding the documentation later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang commented on a diff in pull request #8111: [INLONG-4961][DataProxy] add dataproxy-sdk-golang
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang commented on code in PR #8111:
URL: https://github.com/apache/inlong/pull/8111#discussion_r1209640138
##########
inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/README.md:
##########
@@ -0,0 +1,197 @@
+# dataproxy-sdk-golang
Review Comment:
This line is needless.
##########
inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/README.md:
##########
@@ -0,0 +1,197 @@
+# dataproxy-sdk-golang
+
+## Overview
+
+dataproxy-sdk-golang is the golang version of InLong data proxy client SDK.
+
Review Comment:
Extra blank line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] gunli commented on a diff in pull request #8111: [INLONG-4961][DataProxy] Add dataproxy-sdk-golang
Posted by "gunli (via GitHub)" <gi...@apache.org>.
gunli commented on code in PR #8111:
URL: https://github.com/apache/inlong/pull/8111#discussion_r1209652770
##########
inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/README.md:
##########
@@ -0,0 +1,197 @@
+# dataproxy-sdk-golang
+
+## Overview
+
+dataproxy-sdk-golang is the golang version of InLong data proxy client SDK.
+
Review Comment:
fixed now
##########
inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/README.md:
##########
@@ -0,0 +1,197 @@
+# dataproxy-sdk-golang
+
+## Overview
+
+dataproxy-sdk-golang is the golang version of InLong data proxy client SDK.
+
+
+
+## Features
+
+- Service discovery;
+- Connection pool, buffer pool, byte pool;
+- Backoff retry;
+- Concurrently batch send;
+- Send synchronously;
+- Send asynchronously;
+- Close gracefully;
+- Hookable debug log;
+- Heartbeat;
+- Metrics;
+- Snappy compress;
+- Additional column;
+- Server offline re-balance;
+
+
+## Usage
+
+### example
+
+refer: cli/main.go
+
+``` go
+package main
+
+import (
+ "context"
+ "errors"
+ "flag"
+ "fmt"
+ "log"
+ "strings"
+ "time"
+
+ "go.uber.org/atomic"
+
+ "github.com/apache/inlong/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy"
+)
+
+var (
+ set string
+ url string
+ groupID string
+ streamID string
+ payload string
+ count int
+ addCols mapFlag
+ async bool
+ succeed atomic.Int32
+ failed atomic.Int32
+)
+
+type mapFlag map[string]string
+
+func (f mapFlag) String() string {
+ return fmt.Sprintf("%v", map[string]string(f))
+}
+
+func (f mapFlag) Set(value string) error {
+ split := strings.SplitN(value, "=", 2)
+ if len(split) < 2 {
+ return errors.New("invalid map flag")
+ }
+
+ f[split[0]] = split[1]
+ return nil
+}
+
+func main() {
+ addCols = make(map[string]string)
+ flag.StringVar(&set, "set", "SH_IEG", "dataproxy set")
+ flag.StringVar(&url, "url", dataproxy.DefaultURL, "dataproxy URL")
+ flag.StringVar(&groupID, "group-id", "b_ieg_tglogv3_test", "dataproxy group ID")
+ flag.StringVar(&streamID, "stream-id", "GameSvrState", "dataproxy stream ID")
+ flag.StringVar(&payload, "payload", "GameSvrState|GameSvrId-Test|2023-01-11 10:08:30|127.0.0.1|1", "message payload")
+ flag.IntVar(&count, "count", 10, "send count")
+ flag.Var(&addCols, "col", "add columns, for example: -col k1=v1 -col k2=v2")
+ flag.BoolVar(&async, "async", false, "send asynchronously")
+ flag.Parse()
+
+ var err error
+ client, err := dataproxy.NewClient(
+ dataproxy.WithSet(set),
+ dataproxy.WithGroupID(groupID),
+ dataproxy.WithURL(url),
+ dataproxy.WithMetricsName("clit"),
+ dataproxy.WithAddColumns(addCols),
+ )
+
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ msg := dataproxy.Message{GroupID: groupID, StreamID: streamID, Payload: []byte(payload)}
+ for i := 0; i < count; i++ {
+ if !async {
+ err = client.Send(context.Background(), msg)
+ if err != nil {
+ fmt.Println(err)
+ }
+ } else {
+ client.SendAsync(context.Background(), msg, onResult)
+ }
+ }
+
+ if async {
+ wait()
+ }
+}
+
+func onResult(msg dataproxy.Message, err error) {
+ if err != nil {
+ fmt.Println("error message, streamID = " + msg.StreamID + ", Payload = " + string(msg.Payload))
+ failed.Add(1)
+ } else {
+ succeed.Add(1)
+ }
+}
+
+func wait() {
+ for {
+ if int(succeed.Load()+failed.Load()) >= count {
+ fmt.Println("succeed:", succeed.Load())
+ fmt.Println("failed:", failed.Load())
+ return
+ }
+ time.Sleep(1 * time.Second)
+ }
+}
+
+```
+
+### Options
+
+refer: dataproxy/options.go
+
+``` go
+// Options is the data proxy go client configs
Review Comment:
fixed now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang commented on pull request #8111: [INLONG-4961][DataProxy] Add dataproxy-sdk-golang
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang commented on PR #8111:
URL: https://github.com/apache/inlong/pull/8111#issuecomment-1568214278
Please rebase from the master to fix the failed UT after https://github.com/apache/inlong/pull/8128 is merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #8111: [INLONG-4961][DataProxy] Add dataproxy-sdk-golang
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang merged PR #8111:
URL: https://github.com/apache/inlong/pull/8111
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang commented on a diff in pull request #8111: [INLONG-4961][DataProxy] add dataproxy-sdk-golang
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang commented on code in PR #8111:
URL: https://github.com/apache/inlong/pull/8111#discussion_r1209641026
##########
inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/README.md:
##########
@@ -0,0 +1,197 @@
+# dataproxy-sdk-golang
+
+## Overview
+
+dataproxy-sdk-golang is the golang version of InLong data proxy client SDK.
+
+
+
+## Features
+
+- Service discovery;
+- Connection pool, buffer pool, byte pool;
+- Backoff retry;
+- Concurrently batch send;
+- Send synchronously;
+- Send asynchronously;
+- Close gracefully;
+- Hookable debug log;
+- Heartbeat;
+- Metrics;
+- Snappy compress;
+- Additional column;
+- Server offline re-balance;
+
+
+## Usage
+
+### example
+
+refer: cli/main.go
+
+``` go
+package main
+
+import (
+ "context"
+ "errors"
+ "flag"
+ "fmt"
+ "log"
+ "strings"
+ "time"
+
+ "go.uber.org/atomic"
+
+ "github.com/apache/inlong/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy"
+)
+
+var (
+ set string
+ url string
+ groupID string
+ streamID string
+ payload string
+ count int
+ addCols mapFlag
+ async bool
+ succeed atomic.Int32
+ failed atomic.Int32
+)
+
+type mapFlag map[string]string
+
+func (f mapFlag) String() string {
+ return fmt.Sprintf("%v", map[string]string(f))
+}
+
+func (f mapFlag) Set(value string) error {
+ split := strings.SplitN(value, "=", 2)
+ if len(split) < 2 {
+ return errors.New("invalid map flag")
+ }
+
+ f[split[0]] = split[1]
+ return nil
+}
+
+func main() {
+ addCols = make(map[string]string)
+ flag.StringVar(&set, "set", "SH_IEG", "dataproxy set")
+ flag.StringVar(&url, "url", dataproxy.DefaultURL, "dataproxy URL")
+ flag.StringVar(&groupID, "group-id", "b_ieg_tglogv3_test", "dataproxy group ID")
+ flag.StringVar(&streamID, "stream-id", "GameSvrState", "dataproxy stream ID")
+ flag.StringVar(&payload, "payload", "GameSvrState|GameSvrId-Test|2023-01-11 10:08:30|127.0.0.1|1", "message payload")
+ flag.IntVar(&count, "count", 10, "send count")
+ flag.Var(&addCols, "col", "add columns, for example: -col k1=v1 -col k2=v2")
+ flag.BoolVar(&async, "async", false, "send asynchronously")
+ flag.Parse()
+
+ var err error
+ client, err := dataproxy.NewClient(
+ dataproxy.WithSet(set),
+ dataproxy.WithGroupID(groupID),
+ dataproxy.WithURL(url),
+ dataproxy.WithMetricsName("clit"),
+ dataproxy.WithAddColumns(addCols),
+ )
+
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ msg := dataproxy.Message{GroupID: groupID, StreamID: streamID, Payload: []byte(payload)}
+ for i := 0; i < count; i++ {
+ if !async {
+ err = client.Send(context.Background(), msg)
+ if err != nil {
+ fmt.Println(err)
+ }
+ } else {
+ client.SendAsync(context.Background(), msg, onResult)
+ }
+ }
+
+ if async {
+ wait()
+ }
+}
+
+func onResult(msg dataproxy.Message, err error) {
+ if err != nil {
+ fmt.Println("error message, streamID = " + msg.StreamID + ", Payload = " + string(msg.Payload))
+ failed.Add(1)
+ } else {
+ succeed.Add(1)
+ }
+}
+
+func wait() {
+ for {
+ if int(succeed.Load()+failed.Load()) >= count {
+ fmt.Println("succeed:", succeed.Load())
+ fmt.Println("failed:", failed.Load())
+ return
+ }
+ time.Sleep(1 * time.Second)
+ }
+}
+
+```
+
+### Options
+
+refer: dataproxy/options.go
+
+``` go
+// Options is the data proxy go client configs
Review Comment:
data proxy
->
DataProxy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org