90 lines
2 KiB
Go
90 lines
2 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/elastic/go-elasticsearch/v8"
|
|
)
|
|
|
|
type ElasticsearchExporterV2 struct {
|
|
client *elasticsearch.Client
|
|
config ElasticsearchConfig
|
|
}
|
|
|
|
func NewElasticsearchExporterV2(config ElasticsearchConfig) (*ElasticsearchExporterV2, error) {
|
|
client, err := NewElasticsearchClient(config)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create Elasticsearch client: %w", err)
|
|
}
|
|
|
|
return &ElasticsearchExporterV2{
|
|
client: client,
|
|
config: config,
|
|
}, nil
|
|
}
|
|
|
|
func (e *ElasticsearchExporterV2) Export(ctx context.Context, entries []LogEntry) error {
|
|
if len(entries) == 0 {
|
|
return nil
|
|
}
|
|
|
|
var body strings.Builder
|
|
for _, entry := range entries {
|
|
indexName := e.config.Index
|
|
|
|
indexLine := fmt.Sprintf(`{"index":{"_index":"%s"}}`, indexName)
|
|
body.WriteString(indexLine)
|
|
body.WriteString("\n")
|
|
|
|
data, err := json.Marshal(entry)
|
|
if err != nil {
|
|
slog.Error("error marshalling JSON", "error", err)
|
|
continue
|
|
}
|
|
body.WriteString(string(data))
|
|
body.WriteString("\n")
|
|
}
|
|
|
|
timeout := time.Duration(e.config.Timeout) * time.Second
|
|
ctx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
res, err := e.client.Bulk(
|
|
strings.NewReader(body.String()),
|
|
e.client.Bulk.WithContext(ctx),
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("bulk request error: %w", err)
|
|
}
|
|
defer res.Body.Close()
|
|
|
|
if res.IsError() {
|
|
return fmt.Errorf("bulk request failed: %s", res.String())
|
|
}
|
|
|
|
slog.Debug("Batch successfully exported to Elasticsearch", "count", len(entries))
|
|
return nil
|
|
}
|
|
|
|
func (e *ElasticsearchExporterV2) HealthCheck(ctx context.Context) error {
|
|
timeout := time.Duration(e.config.Timeout) * time.Second
|
|
ctx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
res, err := e.client.Info(e.client.Info.WithContext(ctx))
|
|
if err != nil {
|
|
return fmt.Errorf("health check failed: %w", err)
|
|
}
|
|
defer res.Body.Close()
|
|
|
|
if res.IsError() {
|
|
return fmt.Errorf("health check failed: %s", res.String())
|
|
}
|
|
|
|
return nil
|
|
}
|