From 24564489b8bbca83b111b34baeffc7a73ce88006 Mon Sep 17 00:00:00 2001 From: loveuer Date: Fri, 21 Jun 2024 15:52:34 +0800 Subject: [PATCH] feat: support es sort --- internal/cmd/cmd.go | 2 ++ internal/cmd/run.go | 2 +- internal/interfaces/dumpio.go | 2 +- internal/xes/xes6.go | 4 ++-- internal/xes/xes7.go | 4 ++-- internal/xfile/xfile.go | 2 +- xes/es6/read.go | 13 ++++++++++++- xes/es7/read.go | 16 +++++++++++++++- 8 files changed, 36 insertions(+), 9 deletions(-) diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go index a506e37..f636a19 100644 --- a/internal/cmd/cmd.go +++ b/internal/cmd/cmd.go @@ -37,6 +37,7 @@ esgo2dump --input=http://127.0.0.1:9200/some_index --output=./data.json --query_ f_limit int f_type string f_source string + f_sort string f_query string f_query_file string @@ -57,6 +58,7 @@ func init() { rootCommand.Flags().StringVar(&es_oversion, "o-version", "7", "output(es) version") rootCommand.Flags().StringVarP(&f_type, "type", "t", "data", "data/mapping/setting") rootCommand.Flags().StringVarP(&f_source, "source", "s", "", "query source, use ';' to separate") + rootCommand.Flags().StringVar(&f_sort, "sort", "", "sort, : format, for example: time:desc or name:asc") rootCommand.Flags().StringVarP(&f_query, "query", "q", "", `query dsl, example: {"bool":{"must":[{"term":{"name":{"value":"some_name"}}}],"must_not":[{"range":{"age":{"gte":18,"lt":60}}}]}}`) rootCommand.Flags().StringVar(&f_query_file, "query_file", "", `query json file (will execute line by line)`) rootCommand.Flags().IntVarP(&f_limit, "limit", "l", 100, "") diff --git a/internal/cmd/run.go b/internal/cmd/run.go index 0e8fc60..e19e436 100644 --- a/internal/cmd/run.go +++ b/internal/cmd/run.go @@ -209,7 +209,7 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error { Loop: for _, query := range queries { - dch, ech = input.ReadData(ctx, f_limit, query, sources) + dch, ech = input.ReadData(ctx, f_limit, query, sources, []string{f_sort}) for { select { diff --git a/internal/interfaces/dumpio.go b/internal/interfaces/dumpio.go index 3dda023..ff6e6b7 100644 --- a/internal/interfaces/dumpio.go +++ b/internal/interfaces/dumpio.go @@ -6,7 +6,7 @@ import ( ) type DumpIO interface { - ReadData(ctx context.Context, size int, query map[string]any, includeFields []string) (<-chan []*model.ESSource, <-chan error) + ReadData(ctx context.Context, size int, query map[string]any, includeFields []string, sort []string) (<-chan []*model.ESSource, <-chan error) WriteData(ctx context.Context, docsCh <-chan []*model.ESSource) error ReadMapping(context.Context) (map[string]any, error) diff --git a/internal/xes/xes6.go b/internal/xes/xes6.go index 6d1cd2c..57081a0 100644 --- a/internal/xes/xes6.go +++ b/internal/xes/xes6.go @@ -135,8 +135,8 @@ func (c *clientv6) Close() error { return nil } -func (c *clientv6) ReadData(ctx context.Context, size int, query map[string]any, source []string) (<-chan []*model.ESSource, <-chan error) { - dch, ech := es6.ReadData(ctx, c.client, c.index, size, 0, query, source) +func (c *clientv6) ReadData(ctx context.Context, size int, query map[string]any, source []string, sort []string) (<-chan []*model.ESSource, <-chan error) { + dch, ech := es6.ReadData(ctx, c.client, c.index, size, 0, query, source, sort) return dch, ech } diff --git a/internal/xes/xes7.go b/internal/xes/xes7.go index 3c0501f..6496db8 100644 --- a/internal/xes/xes7.go +++ b/internal/xes/xes7.go @@ -125,8 +125,8 @@ func (c *client) Close() error { // return count, nil //} -func (c *client) ReadData(ctx context.Context, size int, query map[string]any, source []string) (<-chan []*model.ESSource, <-chan error) { - dch, ech := es7.ReadData(ctx, c.client, c.index, size, 0, query, source) +func (c *client) ReadData(ctx context.Context, size int, query map[string]any, source []string, sort []string) (<-chan []*model.ESSource, <-chan error) { + dch, ech := es7.ReadData(ctx, c.client, c.index, size, 0, query, source, sort) return dch, ech } diff --git a/internal/xfile/xfile.go b/internal/xfile/xfile.go index a431581..bfb251b 100644 --- a/internal/xfile/xfile.go +++ b/internal/xfile/xfile.go @@ -110,7 +110,7 @@ func (c *client) IsFile() bool { return true } -func (c *client) ReadData(ctx context.Context, size int, _ map[string]any, _ []string) (<-chan []*model.ESSource, <-chan error) { +func (c *client) ReadData(ctx context.Context, size int, _ map[string]any, _ []string, _ []string) (<-chan []*model.ESSource, <-chan error) { var ( err error count = 0 diff --git a/xes/es6/read.go b/xes/es6/read.go index 32af1b4..fe1f809 100644 --- a/xes/es6/read.go +++ b/xes/es6/read.go @@ -10,10 +10,11 @@ import ( "github.com/loveuer/esgo2dump/internal/util" "github.com/loveuer/esgo2dump/log" "github.com/loveuer/esgo2dump/model" + "github.com/samber/lo" "time" ) -func ReadData(ctx context.Context, client *elastic.Client, index string, size, max int, query map[string]any, source []string) (<-chan []*model.ESSource, <-chan error) { +func ReadData(ctx context.Context, client *elastic.Client, index string, size, max int, query map[string]any, source []string, sort []string) (<-chan []*model.ESSource, <-chan error) { var ( dataCh = make(chan []*model.ESSource) errCh = make(chan error) @@ -71,6 +72,16 @@ func ReadData(ctx context.Context, client *elastic.Client, index string, size, m qs = append(qs, client.Search.WithSourceIncludes(source...)) } + if len(sort) > 0 { + sorts := lo.Filter(sort, func(item string, index int) bool { + return item != "" + }) + + if len(sorts) > 0 { + qs = append(qs, client.Search.WithSort(sorts...)) + } + } + if query != nil && len(query) > 0 { queryBs, _ := json.Marshal(map[string]any{"query": query}) qs = append(qs, client.Search.WithBody(bytes.NewReader(queryBs))) diff --git a/xes/es7/read.go b/xes/es7/read.go index 2905063..4710e89 100644 --- a/xes/es7/read.go +++ b/xes/es7/read.go @@ -10,10 +10,14 @@ import ( "github.com/loveuer/esgo2dump/internal/util" "github.com/loveuer/esgo2dump/log" "github.com/loveuer/esgo2dump/model" + "github.com/samber/lo" "time" ) -func ReadData(ctx context.Context, client *elastic.Client, index string, size, max int, query map[string]any, source []string) (<-chan []*model.ESSource, <-chan error) { +// ReadData es7 read data +// @param[source]: a list of include fields to extract and return from the _source field. +// @param[sort]: a list of : pairs. +func ReadData(ctx context.Context, client *elastic.Client, index string, size, max int, query map[string]any, source []string, sort []string) (<-chan []*model.ESSource, <-chan error) { var ( dataCh = make(chan []*model.ESSource) errCh = make(chan error) @@ -71,6 +75,16 @@ func ReadData(ctx context.Context, client *elastic.Client, index string, size, m qs = append(qs, client.Search.WithSourceIncludes(source...)) } + if len(sort) > 0 { + sorts := lo.Filter(sort, func(item string, index int) bool { + return item != "" + }) + + if len(sorts) > 0 { + qs = append(qs, client.Search.WithSort(sorts...)) + } + } + if query != nil && len(query) > 0 { queryBs, _ := json.Marshal(map[string]any{"query": query}) qs = append(qs, client.Search.WithBody(bytes.NewReader(queryBs)))