fix: write error stack bug

.gitignore

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

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 {
+				if !ok {
-					return err
+					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

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,10 +17,16 @@ import (
 	"github.com/samber/lo"
 )
 
+// 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:
@@ -55,6 +62,12 @@ func NewClient(ctx context.Context, uri string, configs ...Config) (*elastic.Cli
 		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,
@@ -68,12 +81,15 @@ func NewClient(ctx context.Context, uri string, configs ...Config) (*elastic.Cli
 				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 				DialContext:     (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
 			},
+			DiscoverNodesOnStart: cfg2.Sniff,
 		},
 	); err != nil {
 		return nil, err
 	}
 
-	if !cfg.DisablePing {
+	// 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

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
 			}