Compare commits

..

No commits in common. "486ffba68223c28beed671056824b408366794f4" and "91ddffe752e89d0cc042e59c194633fa759d126f" have entirely different histories.

7 changed files with 22 additions and 43 deletions

2
go.mod
View File

@ -1,4 +1,4 @@
module github.com/loveuer/esgo2dump module esgo2dump
go 1.18 go 1.18

View File

@ -2,8 +2,7 @@ package cmd
import ( import (
"context" "context"
"esgo2dump/internal/opt"
"github.com/loveuer/esgo2dump/internal/opt"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )

View File

@ -4,17 +4,16 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"esgo2dump/internal/interfaces"
"esgo2dump/internal/opt"
"esgo2dump/internal/xes"
"esgo2dump/internal/xfile"
"fmt" "fmt"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"io" "io"
"net/url" "net/url"
"os" "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 { func run(cmd *cobra.Command, args []string) error {
@ -28,10 +27,6 @@ func run(cmd *cobra.Command, args []string) error {
logrus.SetLevel(logrus.DebugLevel) logrus.SetLevel(logrus.DebugLevel)
} }
if f_limit == 0 || f_limit > 10000 {
return fmt.Errorf("invalid limit(1 - 10000)")
}
switch f_type { switch f_type {
case "data", "mapping", "setting": case "data", "mapping", "setting":
default: default:
@ -90,7 +85,6 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
err error err error
lines []*interfaces.ESSource lines []*interfaces.ESSource
succeed int succeed int
total int
) )
for { for {
@ -115,9 +109,7 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
return fmt.Errorf("cmd.run: got lines=%d, only succeed=%d", len(lines), succeed) return fmt.Errorf("cmd.run: got lines=%d, only succeed=%d", len(lines), succeed)
} }
total += succeed logrus.Infof("Dump: %d docs succeed!!!", succeed)
logrus.Infof("Dump: succeed=%d total=%d docs succeed!!!", succeed, total)
} }
} }

View File

@ -5,20 +5,19 @@ import (
"context" "context"
"crypto/tls" "crypto/tls"
"encoding/json" "encoding/json"
"esgo2dump/internal/interfaces"
"esgo2dump/internal/opt"
"esgo2dump/internal/util"
"fmt" "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"
"net/http" "net/http"
"net/url" "net/url"
"strings" "strings"
"time" "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) { 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 err error
indexer esutil.BulkIndexer indexer esutil.BulkIndexer
count int count int
be error
) )
if indexer, err = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{ if indexer, err = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
Client: c.c, Client: c.c,
@ -135,9 +133,6 @@ func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (in
Index: c.index, Index: c.index,
DocumentID: doc.DocId, DocumentID: doc.DocId,
Body: bytes.NewReader(bs), Body: bytes.NewReader(bs),
OnFailure: func(ctx context.Context, item esutil.BulkIndexerItem, item2 esutil.BulkIndexerResponseItem, bulkErr error) {
be = bulkErr
},
}); err != nil { }); err != nil {
return 0, err return 0, err
} }
@ -148,13 +143,9 @@ func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (in
return 0, err return 0, err
} }
if be != nil {
return 0, be
}
stats := indexer.Stats() stats := indexer.Stats()
if stats.NumFailed > 0 { 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 return count, nil

View File

@ -1,10 +1,9 @@
package xes package xes
import ( import (
"testing" "esgo2dump/internal/util"
elastic "github.com/elastic/go-elasticsearch/v7" elastic "github.com/elastic/go-elasticsearch/v7"
"github.com/loveuer/esgo2dump/internal/util" "testing"
) )
func TestGetESMapping(t *testing.T) { func TestGetESMapping(t *testing.T) {

View File

@ -4,11 +4,10 @@ import (
"bufio" "bufio"
"context" "context"
"encoding/json" "encoding/json"
"esgo2dump/internal/interfaces"
"github.com/sirupsen/logrus"
"io" "io"
"os" "os"
"github.com/loveuer/esgo2dump/internal/interfaces"
"github.com/sirupsen/logrus"
) )
type client struct { type client struct {

View File

@ -2,11 +2,10 @@ package main
import ( import (
"context" "context"
"esgo2dump/internal/cmd"
"os/signal" "os/signal"
"syscall" "syscall"
"github.com/loveuer/esgo2dump/internal/cmd"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
) )