-
Notifications
You must be signed in to change notification settings - Fork 4
/
elastic.go
115 lines (95 loc) · 2.92 KB
/
elastic.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
package logharbour
import (
"context"
"errors"
"fmt"
"log"
"strings"
"github.com/elastic/go-elasticsearch/v8"
"github.com/elastic/go-elasticsearch/v8/esapi"
)
// ElasticsearchWriter defines methods for Elasticsearch writer
type ElasticsearchWriter interface {
Write(index string, documentID string, body string) error
}
type ElasticsearchClient struct {
client *elasticsearch.Client
}
// NewElasticsearchClient creates a new Elasticsearch client with the given configuration
func NewElasticsearchClient(cfg elasticsearch.Config) (*ElasticsearchClient, error) {
esClient, err := elasticsearch.NewClient(cfg)
if err != nil {
return nil, err
}
return &ElasticsearchClient{client: esClient}, nil
}
// Write sends a document to Elasticsearch. It implements ElasticsearchWriter.
func (ec *ElasticsearchClient) Write(index string, documentID string, body string) error {
req := esapi.IndexRequest{
Index: index,
DocumentID: documentID,
Body: strings.NewReader(body),
}
res, err := req.Do(context.Background(), ec.client)
if err != nil {
return err
}
defer res.Body.Close()
if res.IsError() {
log.Printf("Error response from Elasticsearch: %s", res.String())
return errors.New(res.String())
}
return nil
}
func (ec *ElasticsearchClient) CreateIndex(indexName, mapping string) error {
res, err := ec.client.Indices.Create(indexName, ec.client.Indices.Create.WithBody(strings.NewReader(mapping)))
if err != nil {
return err
}
if res.IsError() {
return fmt.Errorf("error creating index: %s", res.String())
}
return nil
}
func (ec *ElasticsearchClient) IndexExists(indexName string) (bool, error) {
res, err := ec.client.Indices.Exists([]string{indexName})
if err != nil {
return false, err
}
return res.StatusCode == 200, nil
}
// Write sends a document to Elasticsearch with retry logic.
// func (ec *ElasticsearchClient) Write(index string, documentID string, body string) error {
// var maxAttempts = 5
// var initialBackoff = 1 * time.Second
// operation := func() error {
// req := esapi.IndexRequest{
// Index: index,
// DocumentID: documentID,
// Body: strings.NewReader(body),
// }
// res, err := req.Do(context.Background(), ec.client)
// if err != nil {
// return err
// }
// defer res.Body.Close()
// if res.IsError() {
// log.Printf("Error response from Elasticsearch: %s", res.String())
// return errors.New(res.String())
// }
// return nil
// }
// for attempt := 1; attempt <= maxAttempts; attempt++ {
// err := operation()
// if err == nil {
// return nil // Success
// }
// if attempt == maxAttempts {
// return fmt.Errorf("after %d attempts, last error: %s", attempt, err)
// }
// wait := initialBackoff * time.Duration(1<<(attempt-1)) // Exponential backoff
// log.Printf("Attempt %d failed, retrying in %v: %v", attempt, wait, err)
// time.Sleep(wait)
// }
// return fmt.Errorf("reached max attempts without success")
// }