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

fix: update log statistics response, remove sorting by time

Browse Source 
Signed-off-by: huanggze <loganhuang@yunify.com>
This commit is contained in:
huanggze
2019-06-04 13:53:56 +08:00
committed by zryfish
parent 96e32cc003
commit 6f941dce91
1 changed files with 9 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
pkg/simple/client/elasticsearch/esclient.go
View File

@@ -137,12 +137,9 @@ type ContainerHighLightField struct {
type EmptyField struct { type EmptyField struct {
} }
// The aggs object holds two aggregations to be computed by Elasticsearch // StatisticsAggs, the struct for `aggs` of type Request, holds a cardinality aggregation for distinct container counting
// ContainterAgg is a cardinality aggregation to calculate the count of distinct containers
// StartTimeAgg is a top hits aggregation to retrieve the first record
type StatisticsAggs struct { type StatisticsAggs struct {
ContainerAgg ContainerAgg `json:"containers"` ContainerAgg ContainerAgg `json:"containers"`
StartTimeAgg StartTimeAgg `json:"starttime"`
} }
type ContainerAgg struct { type ContainerAgg struct {
@@ -153,15 +150,6 @@ type AggField struct {
Field string `json:"field"` Field string `json:"field"`
} }
type StartTimeAgg struct {
TopHits TopHits `json:"top_hits"`
}
type TopHits struct {
Sort []Sort `json:"sort"`
Size int `json:"size"`
}
type HistogramAggs struct { type HistogramAggs struct {
HistogramAgg HistogramAgg `json:"histogram"` HistogramAgg HistogramAgg `json:"histogram"`
} }
@@ -255,8 +243,7 @@ func createQueryRequest(param QueryParameters) (int, []byte, error) {
if param.Operation == "statistics" { if param.Operation == "statistics" {
operation = OperationStatistics operation = OperationStatistics
containerAgg := AggField{"kubernetes.docker_id.keyword"} containerAgg := AggField{"kubernetes.docker_id.keyword"}
startTimeAgg := TopHits{[]Sort{{Order{"asc"}}}, 1} statisticAggs := StatisticsAggs{ContainerAgg{containerAgg}}
statisticAggs := StatisticsAggs{ContainerAgg{containerAgg}, StartTimeAgg{startTimeAgg}}
request.Aggs = statisticAggs request.Aggs = statisticAggs
request.Size = 0 request.Size = 0
} else if param.Operation == "histogram" { } else if param.Operation == "histogram" {
@@ -361,20 +348,15 @@ type ReadResult struct {
Records []LogRecord `json:"records,omitempty"` Records []LogRecord `json:"records,omitempty"`
} }
// The aggregations object represents the return from an aggregation (see StatisticsAggs type) // StatisticsResponseAggregations, the struct for `aggregations` of type Reponse, holds return results from the aggregation StatisticsAggs
type StatisticsResponseAggregations struct { type StatisticsResponseAggregations struct {
ContainerCount ContainerCount `json:"containers"` ContainerCount ContainerCount `json:"containers"`
StartTime StartTimeTopHit `json:"starttime"`
} }
type ContainerCount struct { type ContainerCount struct {
Value int64 `json:"value"` Value int64 `json:"value"`
} }
type StartTimeTopHit struct {
Hits Hits `json:"hits"`
}
type HistogramAggregations struct { type HistogramAggregations struct {
HistogramAggregation HistogramAggregation `json:"histogram"` HistogramAggregation HistogramAggregation `json:"histogram"`
} }
@@ -396,7 +378,6 @@ type HistogramRecord struct {
type StatisticsResult struct { type StatisticsResult struct {
Containers int64 `json:"containers"` Containers int64 `json:"containers"`
Logs int64 `json:"logs"` Logs int64 `json:"logs"`
StartTime int64 `json:"starttime"`
} }
type HistogramResult struct { type HistogramResult struct {
@@ -460,13 +441,14 @@ func parseQueryResult(operation int, param QueryParameters, body []byte, query [
if response.Status != 0 { if response.Status != 0 {
//Elastic error, eg, es_rejected_execute_exception //Elastic error, eg, es_rejected_execute_exception
queryResult.Status = response.Status queryResult.Status = response.Status
glog.Errorln("The query failed with no response")
return &queryResult return &queryResult
} }
if response.Shards.Successful != response.Shards.Total { if response.Shards.Successful != response.Shards.Total {
//Elastic some shards error //Elastic some shards error
queryResult.Status = http.StatusInternalServerError glog.Warningf("Not all shards succeed, successful shards: %d, skipped shards: %d, failed shards: %d",
return &queryResult response.Shards.Successful, response.Shards.Skipped, response.Shards.Failed)
} }
switch operation { switch operation {
@@ -496,8 +478,7 @@ func parseQueryResult(operation int, param QueryParameters, body []byte, query [
queryResult.Status = http.StatusInternalServerError queryResult.Status = http.StatusInternalServerError
return &queryResult return &queryResult
} }
queryResult.Statistics = &StatisticsResult{Containers: statisticsResponse.ContainerCount.Value, queryResult.Statistics = &StatisticsResult{Containers: statisticsResponse.ContainerCount.Value, Logs: response.Hits.Total}
Logs: statisticsResponse.StartTime.Hits.Total, StartTime: statisticsResponse.StartTime.Hits.Hits[0].Sort[0]}
case OperationHistogram: case OperationHistogram:
var histogramResult HistogramResult var histogramResult HistogramResult
@@ -583,6 +564,7 @@ func Query(param QueryParameters) *QueryResult {
} }
url := fmt.Sprintf("http://%s:%s/%s*/_search", es.Host, es.Port, es.Index) url := fmt.Sprintf("http://%s:%s/%s*/_search", es.Host, es.Port, es.Index)
request, err := http.NewRequest("GET", url, bytes.NewBuffer(query)) request, err := http.NewRequest("GET", url, bytes.NewBuffer(query))
if err != nil { if err != nil {
glog.Errorln(err) glog.Errorln(err)