Refactor Elasticsearch client error handling and introduce bulk index action marshaling

- Updated error handling in the Elasticsearch client to utilize a new `readErrorBody` function for better diagnostics when pinging and searching.
- Refactored the `flushBatch` method to marshal bulk index actions using a dedicated `marshalBulkIndexAction` function, improving code clarity and error handling.
- Enhanced the overall structure of the Elasticsearch client for improved maintainability and readability.

This update enhances the robustness of the Elasticsearch client, ensuring more informative error messages and cleaner code organization.
This commit is contained in:
Mazyar
2026-06-30 13:03:39 +03:30
parent 241e3b5f2d
commit 92c6c7d99a
+27 -8
View File
@@ -86,8 +86,7 @@ func (c *client) ping(ctx context.Context) error {
defer res.Body.Close() defer res.Body.Close()
if res.IsError() { if res.IsError() {
body, _ := io.ReadAll(res.Body) return fmt.Errorf("elasticsearch ping failed: %s", readErrorBody(res.Body))
return fmt.Errorf("elasticsearch ping failed: %s", string(body))
} }
return nil return nil
@@ -126,8 +125,7 @@ func (c *client) Search(ctx context.Context, filter SearchFilter) (*SearchResult
defer res.Body.Close() defer res.Body.Close()
if res.IsError() { if res.IsError() {
body, _ := io.ReadAll(res.Body) return nil, fmt.Errorf("elasticsearch search error: %s", readErrorBody(res.Body))
return nil, fmt.Errorf("elasticsearch search error: %s", string(body))
} }
return decodeSearchResponse(res.Body) return decodeSearchResponse(res.Body)
@@ -191,8 +189,11 @@ func (c *client) flushBatch(batch []AuditDocument) error {
var buf bytes.Buffer var buf bytes.Buffer
for _, doc := range batch { for _, doc := range batch {
indexName := indexNameFor(c.config.IndexPrefix, doc.At) indexName := indexNameFor(c.config.IndexPrefix, doc.At)
meta := fmt.Sprintf(`{"index":{"_index":"%s"}}`, indexName) meta, err := marshalBulkIndexAction(indexName)
buf.WriteString(meta) if err != nil {
return fmt.Errorf("marshal bulk index action: %w", err)
}
buf.Write(meta)
buf.WriteByte('\n') buf.WriteByte('\n')
body, err := json.Marshal(doc) body, err := json.Marshal(doc)
@@ -213,13 +214,31 @@ func (c *client) flushBatch(batch []AuditDocument) error {
defer res.Body.Close() defer res.Body.Close()
if res.IsError() { if res.IsError() {
responseBody, _ := io.ReadAll(res.Body) return fmt.Errorf("elasticsearch bulk error: %s", readErrorBody(res.Body))
return fmt.Errorf("elasticsearch bulk error: %s", string(responseBody))
} }
return nil return nil
} }
type bulkIndexAction struct {
Index struct {
Index string `json:"_index"`
} `json:"index"`
}
func marshalBulkIndexAction(indexName string) ([]byte, error) {
action := bulkIndexAction{}
action.Index.Index = indexName
return json.Marshal(action)
}
// readErrorBody best-effort reads an Elasticsearch error response for diagnostics.
// Read errors are intentionally ignored because the HTTP call already failed.
func readErrorBody(body io.Reader) string {
responseBody, _ := io.ReadAll(body)
return string(responseBody)
}
func indexNameFor(prefix string, at int64) string { func indexNameFor(prefix string, at int64) string {
t := time.Unix(at, 0).UTC() t := time.Unix(at, 0).UTC()
return fmt.Sprintf("%s-%04d.%02d", prefix, t.Year(), int(t.Month())) return fmt.Sprintf("%s-%04d.%02d", prefix, t.Year(), int(t.Month()))