Compare commits
4 Commits
91ddffe752
...
486ffba682
Author | SHA1 | Date | |
---|---|---|---|
|
486ffba682 | ||
|
f75e31ffbb | ||
|
f990923dd8 | ||
|
ff7aa194aa |
2
go.mod
2
go.mod
@ -1,4 +1,4 @@
|
|||||||
module esgo2dump
|
module github.com/loveuer/esgo2dump
|
||||||
|
|
||||||
go 1.18
|
go 1.18
|
||||||
|
|
||||||
|
@ -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"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -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
|
||||||
|
@ -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) {
|
||||||
|
@ -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 {
|
||||||
|
Loading…
x
Reference in New Issue
Block a user