mirror of
https://github.com/paralus/paralus.git
synced 2026-05-13 20:06:47 +00:00
85 lines
2.4 KiB
Go
85 lines
2.4 KiB
Go
package service
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
|
|
v6Client "github.com/elastic/go-elasticsearch"
|
|
)
|
|
|
|
type ElasticSearchQuery struct {
|
|
url string
|
|
indexPattern string
|
|
logPrefix string
|
|
esClient *v6Client.Client
|
|
}
|
|
|
|
func NewElasticSearchQuery(url string, indexPattern string, logPrefix string) (*ElasticSearchQuery, error) {
|
|
cfg := v6Client.Config{
|
|
Addresses: []string{
|
|
url,
|
|
},
|
|
}
|
|
esClient, err := v6Client.NewClient(cfg)
|
|
if err != nil {
|
|
_log.Errorw("NewElasticSearchQuery: Not able to get New Elastic Search Client")
|
|
return nil, err
|
|
}
|
|
// res, err := esClient.Info()
|
|
// if err != nil {
|
|
// _log.Errorw("NewElasticSearchQuery: Error in getting ES client Info")
|
|
// return nil, err
|
|
// }
|
|
// _log.Infow(logPrefix+":Connected to elastic search ", "cluster", res, "index", indexPattern)
|
|
esQuery := &ElasticSearchQuery{
|
|
url: url,
|
|
indexPattern: indexPattern,
|
|
logPrefix: logPrefix,
|
|
esClient: esClient,
|
|
}
|
|
return esQuery, nil
|
|
}
|
|
|
|
//Handle Fires the search query
|
|
func (q *ElasticSearchQuery) Handle(msg bytes.Buffer) (map[string]interface{}, error) {
|
|
_log.Debugw("Searching elastic search: ", "index", q.indexPattern, "url", q.url, "q", q)
|
|
res, err := q.esClient.Search(
|
|
q.esClient.Search.WithContext(context.Background()),
|
|
q.esClient.Search.WithIndex(q.indexPattern),
|
|
q.esClient.Search.WithBody(&msg),
|
|
q.esClient.Search.WithTrackTotalHits(true),
|
|
)
|
|
|
|
if err != nil {
|
|
_log.Errorw("Error getting response:", "err", err)
|
|
return nil, err
|
|
}
|
|
|
|
defer res.Body.Close()
|
|
|
|
if res.IsError() {
|
|
_log.Warnw(q.logPrefix+" Error in search request ", "request", msg.String(), "response", res.String())
|
|
var e map[string]interface{}
|
|
if err := json.NewDecoder(res.Body).Decode(&e); err != nil {
|
|
_log.Errorw("Error parsing the response body", "err", err, "request", msg.String(), "response", res.String())
|
|
return nil, err
|
|
} else {
|
|
if e["error"].(map[string]interface{})["type"] == "index_not_found_exception" {
|
|
_log.Warnw(q.logPrefix+"Skipping this query as its a new setup", "request", msg.String(), "response", res.String())
|
|
return nil, nil
|
|
} else {
|
|
_log.Errorw("Error received from ES", "request", msg.String(), "response", res.String())
|
|
return nil, errors.New(res.Status() + " received from ES")
|
|
}
|
|
}
|
|
}
|
|
var r map[string]interface{}
|
|
if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
|
|
_log.Errorw("Error parsing the response body", "err", err)
|
|
return nil, err
|
|
}
|
|
return r, nil
|
|
}
|