Compare commits
No commits in common. "f75e31ffbb983e90cda6239aade54ef93365b593" and "887f450cf8022273936cfeb63d0aa518f742dc7b" have entirely different histories.
f75e31ffbb
...
887f450cf8
@ -2,8 +2,7 @@ package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/loveuer/esgo2dump/internal/opt"
|
||||
"esgo2dump/internal/opt"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
|
@ -4,17 +4,16 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"esgo2dump/internal/interfaces"
|
||||
"esgo2dump/internal/opt"
|
||||
"esgo2dump/internal/xes"
|
||||
"esgo2dump/internal/xfile"
|
||||
"fmt"
|
||||
"github.com/sirupsen/logrus"
|
||||
"github.com/spf13/cobra"
|
||||
"io"
|
||||
"net/url"
|
||||
"os"
|
||||
|
||||
"github.com/loveuer/esgo2dump/internal/interfaces"
|
||||
"github.com/loveuer/esgo2dump/internal/opt"
|
||||
"github.com/loveuer/esgo2dump/internal/xes"
|
||||
"github.com/loveuer/esgo2dump/internal/xfile"
|
||||
"github.com/sirupsen/logrus"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
func run(cmd *cobra.Command, args []string) error {
|
||||
@ -28,10 +27,6 @@ func run(cmd *cobra.Command, args []string) error {
|
||||
logrus.SetLevel(logrus.DebugLevel)
|
||||
}
|
||||
|
||||
if f_limit == 0 || f_limit > 10000 {
|
||||
return fmt.Errorf("invalid limit(1 - 10000)")
|
||||
}
|
||||
|
||||
switch f_type {
|
||||
case "data", "mapping", "setting":
|
||||
default:
|
||||
@ -60,26 +55,14 @@ func run(cmd *cobra.Command, args []string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = ioo.WriteMapping(cmd.Context(), mapping); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
logrus.Info("Dump: write mapping succeed!!!")
|
||||
|
||||
return nil
|
||||
return ioo.WriteMapping(cmd.Context(), mapping)
|
||||
case "setting":
|
||||
var setting map[string]any
|
||||
if setting, err = ioi.ReadSetting(cmd.Context()); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = ioo.WriteSetting(cmd.Context(), setting); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
logrus.Info("Dump: write setting succeed!!!")
|
||||
|
||||
return nil
|
||||
return ioo.WriteSetting(cmd.Context(), setting)
|
||||
default:
|
||||
return fmt.Errorf("unknown type=%s", f_type)
|
||||
}
|
||||
@ -115,6 +98,7 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
|
||||
}
|
||||
|
||||
logrus.Infof("Dump: %d docs succeed!!!", succeed)
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -5,20 +5,19 @@ import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"esgo2dump/internal/interfaces"
|
||||
"esgo2dump/internal/opt"
|
||||
"esgo2dump/internal/util"
|
||||
"fmt"
|
||||
elastic "github.com/elastic/go-elasticsearch/v7"
|
||||
"github.com/elastic/go-elasticsearch/v7/esapi"
|
||||
"github.com/elastic/go-elasticsearch/v7/esutil"
|
||||
"github.com/sirupsen/logrus"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
elastic "github.com/elastic/go-elasticsearch/v7"
|
||||
"github.com/elastic/go-elasticsearch/v7/esapi"
|
||||
"github.com/elastic/go-elasticsearch/v7/esutil"
|
||||
"github.com/loveuer/esgo2dump/internal/interfaces"
|
||||
"github.com/loveuer/esgo2dump/internal/opt"
|
||||
"github.com/loveuer/esgo2dump/internal/util"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
func NewClient(url *url.URL, iot interfaces.IO, qm map[string]any) (interfaces.DumpIO, error) {
|
||||
@ -111,7 +110,6 @@ func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (in
|
||||
err error
|
||||
indexer esutil.BulkIndexer
|
||||
count int
|
||||
be error
|
||||
)
|
||||
if indexer, err = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
|
||||
Client: c.c,
|
||||
@ -135,9 +133,6 @@ func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (in
|
||||
Index: c.index,
|
||||
DocumentID: doc.DocId,
|
||||
Body: bytes.NewReader(bs),
|
||||
OnFailure: func(ctx context.Context, item esutil.BulkIndexerItem, item2 esutil.BulkIndexerResponseItem, bulkErr error) {
|
||||
be = bulkErr
|
||||
},
|
||||
}); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
@ -148,13 +143,9 @@ func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (in
|
||||
return 0, err
|
||||
}
|
||||
|
||||
if be != nil {
|
||||
return 0, be
|
||||
}
|
||||
|
||||
stats := indexer.Stats()
|
||||
if stats.NumFailed > 0 {
|
||||
return count, fmt.Errorf("write to xes failed_count=%d bulk_count=%d", stats.NumFailed, count)
|
||||
return count, fmt.Errorf("write to xes failed=%d", stats.NumFailed)
|
||||
}
|
||||
|
||||
return count, nil
|
||||
|
@ -1,10 +1,9 @@
|
||||
package xes
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"esgo2dump/internal/util"
|
||||
elastic "github.com/elastic/go-elasticsearch/v7"
|
||||
"github.com/loveuer/esgo2dump/internal/util"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestGetESMapping(t *testing.T) {
|
||||
|
@ -4,11 +4,10 @@ import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"esgo2dump/internal/interfaces"
|
||||
"github.com/sirupsen/logrus"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
"github.com/loveuer/esgo2dump/internal/interfaces"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
type client struct {
|
||||
|
Loading…
x
Reference in New Issue
Block a user