fix: size 0 bug

fix: huge index can't sort _id, back use scroll_id
refac: some files arch

.github/workflows/build.yml

@@ -16,6 +16,9 @@ jobs:
     - name: checkout repository
       uses: actions/checkout@v4
 
+    - name: fill version
+      run: sed -i -E "s/v[0-9]+.[0-9]+.[0-9]+/${{ github.ref_name }}/g" internal/opt/version.go
+
     - name: install golang
       uses: actions/setup-go@v4
       with:

internal/cmd/cmd.go

@@ -57,7 +57,6 @@ esgo2dump --input=http://127.0.0.1:9200/some_index --output=./data.json --query_
 func init() {
 	rootCommand.PersistentFlags().BoolVar(&opt.Cfg.Debug, "debug", false, "")
 	rootCommand.PersistentFlags().BoolVar(&opt.Cfg.Dev, "dev", false, "")
-	rootCommand.PersistentFlags().BoolVar(&opt.Cfg.DisablePing, "disable-ping", false, "")
 	rootCommand.PersistentFlags().BoolVarP(&opt.Cfg.Args.Version, "version", "v", false, "print esgo2dump version")
 
 	rootCommand.Flags().IntVar(&opt.Cfg.Args.Timeout, "timeout", 30, "max timeout seconds per operation with limit")

internal/cmd/run.go

@@ -268,7 +268,7 @@ func newIO(source string, ioType interfaces.IO, esv string) (interfaces.DumpIO,
 
 	switch esv {
 	case "7":
-		return xes.NewClient(source, ioType)
+		return xes.NewClient(iurl, ioType)
 	case "6":
 		return xes.NewClientV6(iurl, ioType)
 	case "8":

internal/opt/opt.go

@@ -15,10 +15,9 @@ type args struct {
 }
 
 type config struct {
-	Debug       bool `json:"-"`
+	Debug bool `json:"-"`
-	Dev         bool `json:"-"`
+	Dev   bool `json:"-"`
-	DisablePing bool `json:"-"`
+	Args  args `json:"-"`
-	Args        args `json:"-"`
 }
 
 var Cfg = &config{}

internal/xes/xes7.go

@@ -32,27 +32,22 @@ func (c *client) WriteData(ctx context.Context, docsCh <-chan []*model.ESSource)
 	return es7.WriteData(ctx, c.client, c.index, docsCh, c)
 }
 
-func NewClient(uri string, iot interfaces.IO) (interfaces.DumpIO, error) {
+func NewClient(url *url.URL, iot interfaces.IO) (interfaces.DumpIO, error) {
 	var (
-		cli   *elastic.Client
+		urlIndex = strings.TrimPrefix(url.Path, "/")
-		err   error
+		cli      *elastic.Client
-		ins   *url.URL
+		err      error
-		index string
 	)
 
-	if ins, err = url.Parse(uri); err != nil {
+	if urlIndex == "" {
-		return nil, err
-	}
-
-	if index = strings.TrimSpace(strings.TrimPrefix(ins.Path, "/")); index == "" {
 		return nil, fmt.Errorf("please specify index name: (like => http://127.0.0.1:9200/my_index)")
 	}
 
-	if cli, err = es7.NewClient(context.TODO(), uri, es7.Config{DisablePing: opt.Cfg.DisablePing}); err != nil {
+	if cli, err = es7.NewClient(context.TODO(), url); err != nil {
 		return nil, err
 	}
 
-	return &client{client: cli, iot: iot, index: index}, nil
+	return &client{client: cli, iot: iot, index: urlIndex}, nil
 }
 
 func (c *client) checkResponse(r *esapi.Response) error {

xes/es7/client.go

@@ -7,7 +7,6 @@ import (
 	"net"
 	"net/http"
 	"net/url"
-	"strconv"
 	"strings"
 	"time"
 
@@ -17,89 +16,78 @@ import (
 	"github.com/samber/lo"
 )
 
-// Deprecated. use uri query: http://<username>:<password>@example.com:port?ping=false&...
+func NewClient(ctx context.Context, url *url.URL) (*elastic.Client, error) {
-type Config struct {
-	DisablePing bool
-}
-
-type UriConfig struct {
-	Ping  bool `json:"ping"`
-	Sniff bool `json:"sniff"`
-}
-
-// NewClient
-// new esv7 client
-// uri example:
-//   - http://127.0.0.1:9200
-//   - https://<username>:<password>@node1.dev:9200,node2.dev:19200,node3.dev:29200
-func NewClient(ctx context.Context, uri string, configs ...Config) (*elastic.Client, error) {
 	var (
-		err      error
+		err         error
-		username string
+		urlUsername string
-		password string
+		urlPassword string
-		client   *elastic.Client
+		client      *elastic.Client
-		ins      *url.URL
+		errCh       = make(chan error)
-	)
+		cliCh       = make(chan *elastic.Client)
-
+		endpoints   = lo.Map(
-	if ins, err = url.Parse(uri); err != nil {
+			strings.Split(url.Host, ","),
-		return nil, err
+			func(item string, index int) string {
-	}
+				return fmt.Sprintf("%s://%s", url.Scheme, item)
-
-	cfg := Config{}
-	if len(configs) > 0 {
-		cfg = configs[0]
-	}
-
-	endpoints := lo.Map(
-		strings.Split(ins.Host, ","),
-		func(item string, index int) string {
-			return fmt.Sprintf("%s://%s", ins.Scheme, item)
-		},
-	)
-
-	if ins.User != nil {
-		username = ins.User.Username()
-		password, _ = ins.User.Password()
-	}
-
-	query := ins.Query()
-
-	cfg2 := &UriConfig{}
-	cfg2.Ping, _ = strconv.ParseBool(query.Get("ping"))
-	cfg2.Sniff, _ = strconv.ParseBool(query.Get("sniff"))
-
-	if client, err = elastic.NewClient(
-		elastic.Config{
-			Addresses:     endpoints,
-			Username:      username,
-			Password:      password,
-			CACert:        nil,
-			RetryOnStatus: []int{429},
-			MaxRetries:    3,
-			RetryBackoff:  nil,
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-				DialContext:     (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
 			},
-			DiscoverNodesOnStart: cfg2.Sniff,
+		)
-		},
+	)
-	); err != nil {
+
+	if url.User != nil {
+		urlUsername = url.User.Username()
+		if p, ok := url.User.Password(); ok {
+			urlPassword = p
+		}
+	}
+
+	ncFunc := func(endpoints []string, username, password string) {
+		var (
+			err      error
+			cli      *elastic.Client
+			infoResp *esapi.Response
+		)
+
+		if cli, err = elastic.NewClient(
+			elastic.Config{
+				Addresses:     endpoints,
+				Username:      username,
+				Password:      password,
+				CACert:        nil,
+				RetryOnStatus: []int{429},
+				MaxRetries:    3,
+				RetryBackoff:  nil,
+				Transport: &http.Transport{
+					TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+					DialContext:     (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
+				},
+			},
+		); err != nil {
+			errCh <- err
+			return
+		}
+
+		if infoResp, err = cli.Info(); err != nil {
+			errCh <- err
+			return
+		}
+
+		if infoResp.StatusCode != 200 {
+			err = fmt.Errorf("info es7 status=%d", infoResp.StatusCode)
+			errCh <- err
+			return
+		}
+
+		cliCh <- cli
+	}
+
+	go ncFunc(endpoints, urlUsername, urlPassword)
+	timeout := tool.TimeoutCtx(ctx, 10)
+
+	select {
+	case <-timeout.Done():
+		return nil, fmt.Errorf("dial es=%v err=%v", endpoints, context.DeadlineExceeded)
+	case client = <-cliCh:
+		return client, nil
+	case err = <-errCh:
 		return nil, err
 	}
-
-	// Deprecated.
-	cfg.DisablePing = cfg.DisablePing || cfg2.Ping
-	if cfg.DisablePing {
-		var res *esapi.Response
-		if res, err = client.Ping(client.Ping.WithContext(tool.TimeoutCtx(ctx, 5))); err != nil {
-			return nil, err
-		}
-
-		if res.StatusCode != 200 {
-			err = fmt.Errorf("ping es server response: %s", res.String())
-			return nil, err
-		}
-	}
-
-	return client, nil
 }

xes/es7/client_test.go

@@ -1,6 +1,7 @@
 package es7
 
 import (
+	"net/url"
 	"testing"
 
 	"github.com/loveuer/esgo2dump/internal/tool"
@@ -8,8 +9,9 @@ import (
 
 func TestNewClient(t *testing.T) {
 	uri := "http://es1.dev:9200,es2.dev:9200"
+	ins, _ := url.Parse(uri)
 
-	c, err := NewClient(tool.Timeout(5), uri)
+	c, err := NewClient(tool.Timeout(5), ins)
 	if err != nil {
 		t.Fatal(err.Error())
 	}