Compare commits

..

4 Commits

Author SHA1 Message Date
loveuer
486ffba682 update: more msg 2024-03-26 18:26:10 +08:00
loveuer
f75e31ffbb feat: put out es bulk err 2024-03-26 18:10:19 +08:00
loveuer
f990923dd8 Merge tag 'v0.0.3'
v0.0.3 beta
2024-03-26 17:59:30 +08:00
loveuer
ff7aa194aa update: add write mapping,setting info msg, go module name 2024-03-26 17:31:36 +08:00
7 changed files with 43 additions and 22 deletions

2
go.mod
View File

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

View File

@ -2,7 +2,8 @@ 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,16 +4,17 @@ 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 {
@ -27,6 +28,10 @@ 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:
@ -85,6 +90,7 @@ 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 {
@ -109,7 +115,9 @@ 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)
} }
logrus.Infof("Dump: %d docs succeed!!!", succeed) total += succeed
logrus.Infof("Dump: succeed=%d total=%d docs succeed!!!", succeed, total)
} }
} }

View File

@ -5,19 +5,20 @@ 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) {
@ -110,6 +111,7 @@ 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,
@ -133,6 +135,9 @@ 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
} }
@ -143,9 +148,13 @@ 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=%d", stats.NumFailed) return count, fmt.Errorf("write to xes failed_count=%d bulk_count=%d", stats.NumFailed, count)
} }
return count, nil return count, nil

View File

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

View File

@ -4,10 +4,11 @@ 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,10 +2,11 @@ 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"
) )