feat: put out es bulk err

This commit is contained in:
loveuer 2024-03-26 18:10:19 +08:00
parent f990923dd8
commit f75e31ffbb
2 changed files with 13 additions and 1 deletions

View File

@ -28,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:

View File

@ -111,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,
@ -134,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
} }
@ -144,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