Compare commits
5 Commits
Author | SHA1 | Date | |
---|---|---|---|
16df5b6b7e | |||
eb97f7b0a3 | |||
5acad1096f | |||
76312a0e56 | |||
9c4c7f5690 |
4
.github/workflows/build.yml
vendored
4
.github/workflows/build.yml
vendored
@ -16,6 +16,9 @@ jobs:
|
|||||||
- name: checkout repository
|
- name: checkout repository
|
||||||
uses: actions/checkout@v4
|
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
|
- name: install golang
|
||||||
uses: actions/setup-go@v4
|
uses: actions/setup-go@v4
|
||||||
with:
|
with:
|
||||||
@ -52,5 +55,4 @@ jobs:
|
|||||||
dist/esgo2dump_${{ github.ref_name }}_windows_amd64.exe
|
dist/esgo2dump_${{ github.ref_name }}_windows_amd64.exe
|
||||||
dist/esgo2dump_${{ github.ref_name }}_windows_arm64.exe
|
dist/esgo2dump_${{ github.ref_name }}_windows_arm64.exe
|
||||||
dist/esgo2dump_${{ github.ref_name }}_darwin_amd64
|
dist/esgo2dump_${{ github.ref_name }}_darwin_amd64
|
||||||
dist/esgo2dump_${{ github.ref_name }}_darwin_amd64
|
|
||||||
dist/esgo2dump_${{ github.ref_name }}_darwin_arm64
|
dist/esgo2dump_${{ github.ref_name }}_darwin_arm64
|
@ -33,10 +33,13 @@ esgo2dump --input=http://127.0.0.1:9200/some_index --output=./data.json --query_
|
|||||||
f_query string
|
f_query string
|
||||||
|
|
||||||
f_query_file string
|
f_query_file string
|
||||||
|
|
||||||
|
f_version bool
|
||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
rootCommand.Flags().BoolVar(&opt.Debug, "debug", false, "")
|
rootCommand.Flags().BoolVar(&opt.Debug, "debug", false, "")
|
||||||
|
rootCommand.Flags().BoolVarP(&f_version, "version", "v", false, "print esgo2dump version")
|
||||||
rootCommand.Flags().IntVar(&opt.Timeout, "timeout", 30, "max timeout seconds per operation with limit")
|
rootCommand.Flags().IntVar(&opt.Timeout, "timeout", 30, "max timeout seconds per operation with limit")
|
||||||
|
|
||||||
rootCommand.Flags().StringVarP(&f_input, "input", "i", "", "*required: input file or es url (example :data.json / http://127.0.0.1:9200/my_index)")
|
rootCommand.Flags().StringVarP(&f_input, "input", "i", "", "*required: input file or es url (example :data.json / http://127.0.0.1:9200/my_index)")
|
||||||
|
@ -50,6 +50,11 @@ func run(cmd *cobra.Command, args []string) error {
|
|||||||
logrus.SetLevel(logrus.DebugLevel)
|
logrus.SetLevel(logrus.DebugLevel)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if f_version {
|
||||||
|
logrus.Infof("esgo2dump (Version: %s)", opt.Version)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
if err = check(cmd); err != nil {
|
if err = check(cmd); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -142,6 +147,7 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
scanner := bufio.NewScanner(qf)
|
scanner := bufio.NewScanner(qf)
|
||||||
|
scanner.Buffer(make([]byte, 1*1024*1024), 5*1024*1024)
|
||||||
lineCount := 1
|
lineCount := 1
|
||||||
for scanner.Scan() {
|
for scanner.Scan() {
|
||||||
line := scanner.Text()
|
line := scanner.Text()
|
||||||
@ -190,6 +196,10 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
|
|||||||
|
|
||||||
if len(lines) == 0 {
|
if len(lines) == 0 {
|
||||||
input.ResetOffset()
|
input.ResetOffset()
|
||||||
|
if query != nil {
|
||||||
|
bs, _ := json.Marshal(query)
|
||||||
|
logrus.Infof("Dump: query_file query=%s read done!!!", string(bs))
|
||||||
|
}
|
||||||
continue Loop
|
continue Loop
|
||||||
}
|
}
|
||||||
|
|
||||||
|
3
internal/opt/version.go
Normal file
3
internal/opt/version.go
Normal file
@ -0,0 +1,3 @@
|
|||||||
|
package opt
|
||||||
|
|
||||||
|
const Version = "v0.1.2"
|
@ -128,7 +128,26 @@ func (c *client) Close() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *client) ResetOffset() {
|
func (c *client) ResetOffset() {
|
||||||
c.scrollId = ""
|
defer func() {
|
||||||
|
c.scrollId = ""
|
||||||
|
}()
|
||||||
|
|
||||||
|
bs, _ := json.Marshal(map[string]string{
|
||||||
|
"scroll_id": c.scrollId,
|
||||||
|
})
|
||||||
|
|
||||||
|
rr, err := c.c.ClearScroll(
|
||||||
|
c.c.ClearScroll.WithContext(util.Timeout(3)),
|
||||||
|
c.c.ClearScroll.WithBody(bytes.NewReader(bs)),
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
logrus.Warnf("ResetOffset: clear scroll id=%s err=%v", c.scrollId, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if rr.StatusCode != 200 {
|
||||||
|
logrus.Warnf("ResetOffset: clear scroll id=%s msg=%s", c.scrollId, rr.String())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (int, error) {
|
func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (int, error) {
|
||||||
var (
|
var (
|
||||||
@ -195,7 +214,7 @@ func (c *client) ReadData(ctx context.Context, i int, query map[string]any) ([]*
|
|||||||
c.c.Search.WithIndex(c.index),
|
c.c.Search.WithIndex(c.index),
|
||||||
c.c.Search.WithSize(i),
|
c.c.Search.WithSize(i),
|
||||||
c.c.Search.WithFrom(0),
|
c.c.Search.WithFrom(0),
|
||||||
c.c.Search.WithScroll(time.Duration(opt.ScrollDurationSeconds) * time.Second),
|
c.c.Search.WithScroll(time.Duration(opt.Timeout*2) * time.Second),
|
||||||
}
|
}
|
||||||
|
|
||||||
if query != nil && len(query) > 0 {
|
if query != nil && len(query) > 0 {
|
||||||
@ -223,7 +242,7 @@ func (c *client) ReadData(ctx context.Context, i int, query map[string]any) ([]*
|
|||||||
|
|
||||||
if resp, err = c.c.Scroll(
|
if resp, err = c.c.Scroll(
|
||||||
c.c.Scroll.WithScrollID(c.scrollId),
|
c.c.Scroll.WithScrollID(c.scrollId),
|
||||||
c.c.Scroll.WithScroll(time.Duration(opt.ScrollDurationSeconds)*time.Second),
|
c.c.Scroll.WithScroll(time.Duration(opt.Timeout*2)*time.Second),
|
||||||
); err != nil {
|
); err != nil {
|
||||||
return result.Hits.Hits, nil
|
return result.Hits.Hits, nil
|
||||||
}
|
}
|
||||||
|
Reference in New Issue
Block a user