admin/kubesphere
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

significantly improve log search performance

Browse Source 
Signed-off-by: huanggze <loganhuang@yunify.com>
This commit is contained in:
huanggze
2020-07-14 19:18:05 +08:00
parent 0f63b5ba5f
commit 6a5738d66a
7 changed files with 107 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
pkg/simple/client/logging/elasticsearch/elasticsearch.go
View File

@@ -10,6 +10,7 @@ import (
"kubesphere.io/kubesphere/pkg/simple/client/logging/elasticsearch/versions/v5"
"kubesphere.io/kubesphere/pkg/simple/client/logging/elasticsearch/versions/v6"
"kubesphere.io/kubesphere/pkg/simple/client/logging/elasticsearch/versions/v7"
"kubesphere.io/kubesphere/pkg/utils/esutil"
"kubesphere.io/kubesphere/pkg/utils/stringutils"
"strings"
"sync"
@@ -33,7 +34,7 @@ type Elasticsearch struct {
// versioned es client interface
type client interface {
Search(body []byte, scroll bool) ([]byte, error)
Search(indices string, body []byte, scroll bool) ([]byte, error)
Scroll(id string) ([]byte, error)
ClearScroll(id string)
GetTotalHitCount(v interface{}) int64
@@ -147,7 +148,7 @@ func (es *Elasticsearch) GetCurrentStats(sf logging.SearchFilter) (logging.Stati
return logging.Statistics{}, err
}
b, err := es.c.Search(body, true)
b, err := es.c.Search(esutil.ResolveIndexNames(es.index, sf.Starttime, sf.Endtime), body, true)
if err != nil {
return logging.Statistics{}, err
}
@@ -180,7 +181,7 @@ func (es *Elasticsearch) CountLogsByInterval(sf logging.SearchFilter, interval s
return logging.Histogram{}, err
}
b, err := es.c.Search(body, false)
b, err := es.c.Search(esutil.ResolveIndexNames(es.index, sf.Starttime, sf.Endtime), body, false)
if err != nil {
return logging.Histogram{}, err
}
@@ -219,7 +220,7 @@ func (es *Elasticsearch) SearchLogs(sf logging.SearchFilter, f, s int64, o strin
return logging.Logs{}, err
}
b, err := es.c.Search(body, false)
b, err := es.c.Search(esutil.ResolveIndexNames(es.index, sf.Starttime, sf.Endtime), body, false)
if err != nil {
return logging.Logs{}, err
}
@@ -264,7 +265,7 @@ func (es *Elasticsearch) ExportLogs(sf logging.SearchFilter, w io.Writer) error
return err
}
b, err := es.c.Search(body, true)
b, err := es.c.Search(esutil.ResolveIndexNames(es.index, sf.Starttime, sf.Endtime), body, true)
defer es.ClearScroll(id)
if err != nil {
return err

11
pkg/simple/client/logging/elasticsearch/elasticsearch_test.go
View File

@@ -224,17 +224,14 @@ func mockElasticsearchService(pattern, fakeResp string, fakeCode int) *httptest.
}
func newElasticsearchClient(srv *httptest.Server, version string) *Elasticsearch {
var es *Elasticsearch
es := &Elasticsearch{index: "ks-logstash-log"}
switch version {
case ElasticV5:
client, _ := v5.New(srv.URL, "ks-logstash-log")
es = &Elasticsearch{c: client}
es.c, _ = v5.New(srv.URL, "ks-logstash-log")
case ElasticV6:
client, _ := v6.New(srv.URL, "ks-logstash-log")
es = &Elasticsearch{c: client}
es.c, _ = v6.New(srv.URL, "ks-logstash-log")
case ElasticV7:
client, _ := v7.New(srv.URL, "ks-logstash-log")
es = &Elasticsearch{c: client}
es.c, _ = v7.New(srv.URL, "ks-logstash-log")
}
return es
}

5
pkg/simple/client/logging/elasticsearch/versions/v5/v5.go
View File

@@ -24,10 +24,11 @@ func New(address string, index string) (*Elastic, error) {
return &Elastic{client: client, index: index}, err
}
func (e *Elastic) Search(body []byte, scroll bool) ([]byte, error) {
func (e *Elastic) Search(indices string, body []byte, scroll bool) ([]byte, error) {
opts := []func(*esapi.SearchRequest){
e.client.Search.WithContext(context.Background()),
e.client.Search.WithIndex(fmt.Sprintf("%s*", e.index)),
e.client.Search.WithIndex(indices),
e.client.Search.WithIgnoreUnavailable(true),
e.client.Search.WithBody(bytes.NewBuffer(body)),
}
if scroll {

5
pkg/simple/client/logging/elasticsearch/versions/v6/v6.go
View File

@@ -24,10 +24,11 @@ func New(address string, index string) (*Elastic, error) {
return &Elastic{Client: client, index: index}, err
}
func (e *Elastic) Search(body []byte, scroll bool) ([]byte, error) {
func (e *Elastic) Search(indices string, body []byte, scroll bool) ([]byte, error) {
opts := []func(*esapi.SearchRequest){
e.Client.Search.WithContext(context.Background()),
e.Client.Search.WithIndex(fmt.Sprintf("%s*", e.index)),
e.Client.Search.WithIndex(indices),
e.Client.Search.WithIgnoreUnavailable(true),
e.Client.Search.WithBody(bytes.NewBuffer(body)),
}
if scroll {

5
pkg/simple/client/logging/elasticsearch/versions/v7/v7.go
View File

@@ -24,10 +24,11 @@ func New(address string, index string) (*Elastic, error) {
return &Elastic{client: client, index: index}, err
}
func (e *Elastic) Search(body []byte, scroll bool) ([]byte, error) {
func (e *Elastic) Search(indices string, body []byte, scroll bool) ([]byte, error) {
opts := []func(*esapi.SearchRequest){
e.client.Search.WithContext(context.Background()),
e.client.Search.WithIndex(fmt.Sprintf("%s*", e.index)),
e.client.Search.WithIndex(indices),
e.client.Search.WithIgnoreUnavailable(true),
e.client.Search.WithBody(bytes.NewBuffer(body)),
}
if scroll {