fix: write error stack bug

.gitignore

@@ -1,11 +1,7 @@
 .idea
 .vscode
 .DS_Store
-*data.json
-*mapping.json
-*setting.json
-*output.json
-*test.json
+*.json
 *.txt
 dist
 xtest

internal/cmd/cmd.go

@@ -57,6 +57,7 @@ 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

@@ -60,10 +60,14 @@ func run(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
+	log.Debug("init: new input io success!")
+
 	if ioo, err = newIO(opt.Cfg.Args.Output, interfaces.IOOutput, es_oversion); err != nil {
 		return err
 	}
 
+	log.Debug("init: new output io success!")
+
 	defer func() {
 		_ = ioi.Close()
 		_ = ioo.Close()
@@ -186,10 +190,10 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
 		wg   = sync.WaitGroup{}
 	)
 
-	go func() {
 	wg.Add(1)
+	go func() {
 		if err = output.WriteData(ctx, wch); err != nil {
-			e2ch <- err
+			log.Fatal("Dump: write data err: %s", err.Error())
 		}
 
 		wg.Done()
@@ -210,18 +214,26 @@ Loop:
 			case <-ctx.Done():
 				return ctx.Err()
 			case err, ok = <-ech:
-				if err != nil {
-					return err
-				}
-
+				if !ok {
+					log.Debug("pipe: read io closed")
 					continue Loop
-			case err, _ = <-e2ch:
+				}
+				log.Debug("pipe: got err from read io, err = %s", err.Error())
+				return err
+			case err, ok = <-e2ch:
+				if !ok {
+					log.Debug("pipe: write io closed")
+					continue Loop
+				}
+				log.Debug("pipe: got err from write io, err = %s", err.Error())
 				return err
 			case docs, ok = <-dch:
 				if !ok || len(docs) == 0 {
 					continue Loop
 				}
 
+				log.Debug("pipe: got %d docs from read io", len(docs))
+
 				wch <- docs
 			}
 		}
@@ -229,6 +241,7 @@ Loop:
 
 	close(wch)
 
+	log.Debug("pipe: wait for all io closed")
 	wg.Wait()
 
 	return nil
@@ -268,7 +281,7 @@ func newIO(source string, ioType interfaces.IO, esv string) (interfaces.DumpIO,
 
 	switch esv {
 	case "7":
-		return xes.NewClient(iurl, ioType)
+		return xes.NewClient(source, ioType)
 	case "6":
 		return xes.NewClientV6(iurl, ioType)
 	case "8":

internal/opt/opt.go

@@ -17,6 +17,7 @@ type args struct {
 type config struct {
 	Debug       bool `json:"-"`
 	Dev         bool `json:"-"`
+	DisablePing bool `json:"-"`
 	Args        args `json:"-"`
 }
 

internal/xes/xes7.go

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

internal/xfile/xfile.go

@@ -119,6 +119,7 @@ func (c *client) ReadData(ctx context.Context, size int, _ map[string]any, _ []s
 		dch       = make(chan []*model.ESSource)
 		ech       = make(chan error)
 		ready     = make(chan bool)
+		total     = 0
 	)
 
 	go func(ctx context.Context) {
@@ -144,6 +145,7 @@ func (c *client) ReadData(ctx context.Context, size int, _ map[string]any, _ []s
 
 				list = append(list, item)
 				count++
+				total++
 
 				if count >= size {
 					dch <- list
@@ -162,6 +164,8 @@ func (c *client) ReadData(ctx context.Context, size int, _ map[string]any, _ []s
 		if err = c.scanner.Err(); err != nil {
 			ech <- err
 		}
+
+		log.Debug("read: read file succeed! total=%d", total)
 	}(ctx)
 
 	<-ready

xes/es6/read.go

@@ -32,7 +32,6 @@ func ReadData(ctx context.Context, client *elastic.Client, index string, size, m
 
 		defer func() {
 			close(dataCh)
-			close(errCh)
 
 			if scrollId != "" {
 				bs, _ := json.Marshal(map[string]string{

xes/es7/client.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 	"time"
 
@@ -16,37 +17,58 @@ import (
 	"github.com/samber/lo"
 )
 
-func NewClient(ctx context.Context, url *url.URL) (*elastic.Client, error) {
+// Deprecated. use uri query: http://<username>:<password>@example.com:port?ping=false&...
+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
-		urlUsername string
-		urlPassword string
+		username string
+		password string
 		client   *elastic.Client
-		errCh       = make(chan error)
-		cliCh       = make(chan *elastic.Client)
-		endpoints   = lo.Map(
-			strings.Split(url.Host, ","),
+		ins      *url.URL
+	)
+
+	if ins, err = url.Parse(uri); err != nil {
+		return nil, err
+	}
+
+	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", url.Scheme, item)
+			return fmt.Sprintf("%s://%s", ins.Scheme, item)
 		},
 	)
-	)
 
-	if url.User != nil {
-		urlUsername = url.User.Username()
-		if p, ok := url.User.Password(); ok {
-			urlPassword = p
-		}
+	if ins.User != nil {
+		username = ins.User.Username()
+		password, _ = ins.User.Password()
 	}
 
-	ncFunc := func(endpoints []string, username, password string) {
-		var (
-			err      error
-			cli      *elastic.Client
-			infoResp *esapi.Response
-		)
+	query := ins.Query()
 
-		if cli, err = elastic.NewClient(
+	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,
@@ -59,35 +81,25 @@ func NewClient(ctx context.Context, url *url.URL) (*elastic.Client, error) {
 				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 				DialContext:     (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
 			},
+			DiscoverNodesOnStart: cfg2.Sniff,
 		},
 	); 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,7 +1,6 @@
 package es7
 
 import (
-	"net/url"
 	"testing"
 
 	"github.com/loveuer/esgo2dump/internal/tool"
@@ -9,9 +8,8 @@ import (
 
 func TestNewClient(t *testing.T) {
 	uri := "http://es1.dev:9200,es2.dev:9200"
-	ins, _ := url.Parse(uri)
 
-	c, err := NewClient(tool.Timeout(5), ins)
+	c, err := NewClient(tool.Timeout(5), uri)
 	if err != nil {
 		t.Fatal(err.Error())
 	}

xes/es7/write.go

@@ -35,11 +35,28 @@ func WriteData(ctx context.Context, client *elastic.Client, index string, docsCh
 			count := 0
 
 			if indexer, err = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
+				NumWorkers:    0,
+				FlushBytes:    0,
+				FlushInterval: 0,
 				Client:        client,
+				Decoder:       nil,
+				OnError: func(ctx context.Context, err error) {
+					log.Error("es7.writer: on error log, err = %s", err.Error())
+				},
 				Index:               index,
 				ErrorTrace:          true,
-				OnError: func(ctx context.Context, err error) {
-				},
+				FilterPath:          []string{},
+				Header:              map[string][]string{},
+				Human:               false,
+				Pipeline:            "",
+				Pretty:              false,
+				Refresh:             "",
+				Routing:             "",
+				Source:              []string{},
+				SourceExcludes:      []string{},
+				SourceIncludes:      []string{},
+				Timeout:             0,
+				WaitForActiveShards: "",
 			}); err != nil {
 				return err
 			}