loveuer/esgo2dump
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: color log, upx release; format: debug log

Browse Source
This commit is contained in:
loveuer
2024-05-10 21:40:02 +08:00
parent 63b9abbc76
commit 44a125a524
11 changed files with 173 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
internal/cmd/run.go
View File

@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/loveuer/esgo2dump/internal/log"
"net/url"
"os"
"strings"
@ -52,12 +53,12 @@ func run(cmd *cobra.Command, args []string) error {
if opt.Debug {
logrus.SetLevel(logrus.DebugLevel)
logrus.SetReportCaller(true)
logrus.SetFormatter(&logrus.TextFormatter{})
logrus.SetFormatter(&logrus.JSONFormatter{})
}
if f_version {
logrus.Infof("esgo2dump (Version: %s)", opt.Version)
return nil
fmt.Printf("esgo2dump (Version: %s)\n", opt.Version)
os.Exit(0)
}
if err = check(cmd); err != nil {
@ -91,7 +92,7 @@ func run(cmd *cobra.Command, args []string) error {
return err
}
logrus.Info("Dump: write data succeed!!!")
log.Info("Dump: write data succeed!!!")
return nil
case "mapping":
@ -104,7 +105,7 @@ func run(cmd *cobra.Command, args []string) error {
return err
}
logrus.Info("Dump: write mapping succeed!!!")
log.Info("Dump: write mapping succeed!!!")
return nil
case "setting":
@ -117,7 +118,7 @@ func run(cmd *cobra.Command, args []string) error {
return err
}
logrus.Info("Dump: write setting succeed!!!")
log.Info("Dump: write setting succeed!!!")
return nil
default:
@ -208,13 +209,16 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
return
}
logrus.Debugf("executeData: input read_data got lines=%d", len(lines))
logrus.
WithField("action", "input read data got lines").
WithField("lines", len(lines)).
Debug()
if len(lines) == 0 {
input.ResetOffset()
if query != nil {
bs, _ := json.Marshal(query)
logrus.Infof("Dump: query_file query=%s read done!!!", string(bs))
log.Info("Dump: query_file query=%s read done!!!", string(bs))
}
continue Loop
}
@ -250,15 +254,18 @@ func executeData(ctx context.Context, input, output interfaces.DumpIO) error {
return err
}
logrus.Debugf("executeData: output write_data succeed lines=%d", succeed)
logrus.
WithField("action", "output wrote data lines").
WithField("lines", succeed).
Debug()
if succeed != len(docs) {
return fmt.Errorf("cmd.run: got lines=%d, only succeed=%d", len(docs), succeed)
return fmt.Errorf("output got lines=%d, only succeed=%d", len(docs), succeed)
}
total += succeed
logrus.Infof("Dump: succeed=%d total=%d docs succeed!!!", succeed, total)
log.Info("Dump: succeed=%d total=%d docs succeed!!!", succeed, total)
}
}
}
@ -271,39 +278,61 @@ func newIO(source string, ioType interfaces.IO, esv string) (interfaces.DumpIO,
qm = make(map[string]any)
)
logrus.Debugf("newIO.%s: source string=%s", ioType.Code(), source)
logrus.
WithField("action", "new_io").
WithField("type", ioType.Code()).
WithField("source", source).
WithField("es_version", esv).
Debug()
if iurl, err = url.Parse(source); err != nil {
logrus.Debugf("newIO.%s: url parse source err=%v", ioType.Code(), err)
logrus.
WithField("action", "new_io url parse error").
WithField("type", ioType.Code()).
WithField("source", source).
WithField("err", err).
Debug()
goto ClientByFile
}
if !(iurl.Scheme == "http" || iurl.Scheme == "https") {
logrus.Debugf("newIO.%s: url scheme=%s invalid", ioType.Code(), iurl.Scheme)
logrus.
WithField("action", "new_io url scheme error").
WithField("type", ioType.Code()).
WithField("source", source).
WithField("scheme", iurl.Scheme).
Debug()
goto ClientByFile
}
if iurl.Host == "" {
logrus.Debugf("newIO.%s: url host empty", ioType.Code())
logrus.
WithField("action", "new_io url host empty").
WithField("type", ioType.Code()).
WithField("source", source).
Debug()
goto ClientByFile
}
if ioType == interfaces.IOInput && f_query != "" {
if err = json.Unmarshal([]byte(f_query), &qm); err != nil {
logrus.Debugf("newIO.%s: query=%s invalid to map[string]any", ioType.Code(), f_query)
logrus.
WithField("action", "new_io query string invalid").
WithField("type", ioType.Code()).
WithField("source", source).
WithField("query", f_query).
Debug()
return nil, fmt.Errorf("invalid query err=%v", err)
}
}
logrus.Debugf("newIO.%s: source as url=%+v version=%s", ioType.Code(), *iurl, esv)
switch esv {
case "7":
return xes.NewClient(iurl, ioType)
case "6":
return xes.NewClientV6(iurl, ioType)
case "8":
return nil, errors.New("es version 8 comming soon")
return nil, errors.New("es version 8 coming soon")
default:
return nil, fmt.Errorf("unknown es version=%s", esv)
}

2
internal/interfaces/dumpio.go
View File

@ -3,7 +3,7 @@ package interfaces
import "context"
type DumpIO interface {
ReadData(context.Context, int, map[string]any, []string) ([]*ESSource, error)
ReadData(ctx context.Context, size int, query map[string]any, includeFields []string) ([]*ESSource, error)
WriteData(ctx context.Context, docs []*ESSource) (int, error)
ResetOffset()

43
internal/log/log.go Normal file
View File

@ -0,0 +1,43 @@
package log
import (
"bytes"
"fmt"
"github.com/fatih/color"
"sync"
"time"
)
var (
red = color.New(color.FgRed)
green = color.New(color.FgGreen)
yellow = color.New(color.FgYellow)
locker = &sync.Mutex{}
timeFormat = "06/01/02T15:04:05"
)
func Info(msg string, data ...any) {
buf := &bytes.Buffer{}
_, _ = green.Fprint(buf, "Info ")
_, _ = fmt.Fprintf(buf, "| %s | ", time.Now().Format(timeFormat))
_, _ = fmt.Fprintf(buf, msg, data...)
fmt.Println(buf.String())
}
func Warn(msg string, data ...any) {
buf := &bytes.Buffer{}
_, _ = yellow.Fprint(buf, "Warn ")
_, _ = fmt.Fprintf(buf, "| %s | ", time.Now().Format(timeFormat))
_, _ = fmt.Fprintf(buf, msg, data...)
fmt.Println(buf.String())
}
func Error(msg string, data ...any) {
buf := &bytes.Buffer{}
_, _ = red.Fprint(buf, "Error ")
_, _ = fmt.Fprintf(buf, "| %s | ", time.Now().Format(timeFormat))
_, _ = fmt.Fprintf(buf, msg, data...)
fmt.Println(buf.String())
}

2
internal/opt/version.go
View File

@ -1,3 +1,3 @@
package opt
const Version = "v0.1.2"
const Version = "v0.2.1"

38
internal/xes/xes.go
View File

@ -6,6 +6,7 @@ import (
"crypto/tls"
"encoding/json"
"fmt"
"github.com/loveuer/esgo2dump/internal/log"
"net"
"net/http"
"net/url"
@ -39,7 +40,13 @@ func NewClient(url *url.URL, iot interfaces.IO) (interfaces.DumpIO, error) {
}
}
logrus.Debugf("xes.NewClient: endpoint=%s index=%s (username=%s password=%s)", address, urlIndex, urlUsername, urlPassword)
logrus.
WithField("action", "new es client v7").
WithField("endpoint", address).
WithField("index", urlIndex).
WithField("username", urlUsername).
WithField("password", urlPassword).
Debug()
if urlIndex == "" {
return nil, fmt.Errorf("please specify index name: (like => http://127.0.0.1:9200/my_index)")
@ -67,20 +74,30 @@ func NewClient(url *url.URL, iot interfaces.IO) (interfaces.DumpIO, error) {
},
},
); err != nil {
logrus.Debugf("xes.NewClient: elastic new client with endpont=%s err=%v", endpoints, err)
logrus.
WithField("action", "new es client v7 error").
WithField("endpoints", endpoints).
WithField("err", err).
Debug()
errCh <- err
return
}
if infoResp, err = cli.Info(); err != nil {
logrus.Debugf("xes.NewClient: ping err=%v", err)
logrus.
WithField("action", "es client v7 ping error").
WithField("err", err).
Debug()
errCh <- err
return
}
if infoResp.StatusCode != 200 {
err = fmt.Errorf("info xes status=%d", infoResp.StatusCode)
logrus.Debugf("xes.NewClient: status err=%v", err)
logrus.
WithField("action", "es client v7 ping status error").
WithField("status", infoResp.StatusCode).
Debug()
errCh <- err
return
}
@ -141,12 +158,12 @@ func (c *client) ResetOffset() {
c.c.ClearScroll.WithBody(bytes.NewReader(bs)),
)
if err != nil {
logrus.Warnf("ResetOffset: clear scroll id=%s err=%v", c.scrollId, err)
log.Warn("ResetOffset: clear scroll id=%s err=%v", c.scrollId, err)
return
}
if rr.StatusCode != 200 {
logrus.Warnf("ResetOffset: clear scroll id=%s msg=%s", c.scrollId, rr.String())
log.Warn("ResetOffset: clear scroll id=%s msg=%s", c.scrollId, rr.String())
}
}
func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (int, error) {
@ -157,9 +174,12 @@ func (c *client) WriteData(ctx context.Context, docs []*interfaces.ESSource) (in
be error
)
if indexer, err = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
Client: c.c,
Index: c.index,
Refresh: "",
Client: c.c,
Index: c.index,
ErrorTrace: true,
OnError: func(ctx context.Context, err error) {
},
}); err != nil {
return 0, err
}

31
internal/xes/xes6.go
View File

@ -6,6 +6,7 @@ import (
"crypto/tls"
"encoding/json"
"fmt"
"github.com/loveuer/esgo2dump/internal/log"
"io"
"net"
"net/http"
@ -40,7 +41,13 @@ func NewClientV6(url *url.URL, iot interfaces.IO) (interfaces.DumpIO, error) {
}
}
logrus.Debugf("xes.NewClient: endpoint=%s index=%s (username=%s password=%s)", address, urlIndex, urlUsername, urlPassword)
logrus.
WithField("action", "new es client v6").
WithField("endpoint", address).
WithField("index", urlIndex).
WithField("username", urlUsername).
WithField("password", urlPassword).
Debug()
if urlIndex == "" {
return nil, fmt.Errorf("please specify index name: (like => http://127.0.0.1:9200/my_index)")
@ -68,20 +75,30 @@ func NewClientV6(url *url.URL, iot interfaces.IO) (interfaces.DumpIO, error) {
},
},
); err != nil {
logrus.Debugf("xes.NewClient: elastic new client with endpont=%s err=%v", endpoints, err)
logrus.
WithField("action", "new es client v6 error").
WithField("endpoints", endpoints).
WithField("err", err).
Debug()
errCh <- err
return
}
if infoResp, err = cli.Info(); err != nil {
logrus.Debugf("xes.NewClient: ping err=%v", err)
logrus.
WithField("action", "es client v6 ping error").
WithField("err", err).
Debug()
errCh <- err
return
}
if infoResp.StatusCode != 200 {
err = fmt.Errorf("info xes status=%d", infoResp.StatusCode)
logrus.Debugf("xes.NewClient: status err=%v", err)
logrus.
WithField("action", "es client v6 ping status error").
WithField("status", infoResp.StatusCode).
Debug()
errCh <- err
return
}
@ -142,12 +159,12 @@ func (c *clientv6) ResetOffset() {
c.c.ClearScroll.WithBody(bytes.NewReader(bs)),
)
if err != nil {
logrus.Warnf("ResetOffset: clear scroll id=%s err=%v", c.scrollId, err)
log.Warn("ResetOffset: clear scroll id=%s err=%v", c.scrollId, err)
return
}
if rr.StatusCode != 200 {
logrus.Warnf("ResetOffset: clear scroll id=%s msg=%s", c.scrollId, rr.String())
log.Warn("ResetOffset: clear scroll id=%s msg=%s", c.scrollId, rr.String())
}
}
func (c *clientv6) WriteData(ctx context.Context, docs []*interfaces.ESSource) (int, error) {
@ -160,8 +177,8 @@ func (c *clientv6) WriteData(ctx context.Context, docs []*interfaces.ESSource) (
if indexer, err = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
Client: c.c,
Index: c.index,
Refresh: "",
DocumentType: "_doc",
ErrorTrace: true,
}); err != nil {
return 0, err
}