opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-19 10:46:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(monitor): raw_name of influxdb query (#21659)

Browse Source
This commit is contained in:
Zexi Li
2024-11-22 15:52:39 +08:00
committed by GitHub
parent 467ac29e13
commit 5246ba11ac
7 changed files with 103 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/monitor/models/commonalert.go
+1 -1
View File
@@ -300,7 +300,7 @@ func (man *SCommonAlertManager) ValidateMetricQuery(metricRequest *monitor.Commo
To: metricRequest.To,
Interval: metricRequest.Interval,
}
setDefaultValue(q.AlertQuery, &metriInputQuery, scope, ownerId, true)
setDefaultValue(q.AlertQuery, &metriInputQuery, scope, ownerId)
err := UnifiedMonitorManager.ValidateInputQuery(q.AlertQuery, &metriInputQuery)
if err != nil {
return err
pkg/monitor/models/datasource.go
+35 -26
View File
@@ -43,6 +43,7 @@ import (
"yunion.io/x/onecloud/pkg/mcclient/auth"
"yunion.io/x/onecloud/pkg/monitor/datasource"
merrors "yunion.io/x/onecloud/pkg/monitor/errors"
"yunion.io/x/onecloud/pkg/monitor/tsdb"
"yunion.io/x/onecloud/pkg/monitor/validators"
"yunion.io/x/onecloud/pkg/util/influxdb"
"yunion.io/x/onecloud/pkg/util/stringutils2"
@@ -99,7 +100,7 @@ func (m *SDataSourceManager) GetSource(id string) (*SDataSource, error) {
return ret.(*SDataSource), nil
}
func (self *SDataSourceManager) GetDatabases() (jsonutils.JSONObject, error) {
func (m *SDataSourceManager) GetDatabases() (jsonutils.JSONObject, error) {
ret := jsonutils.NewDict()
dataSource, err := datasource.GetDefaultSource("")
if err != nil {
@@ -115,11 +116,11 @@ func (self *SDataSourceManager) GetDatabases() (jsonutils.JSONObject, error) {
return ret, nil
}
func (self *SDataSourceManager) GetMeasurements(query jsonutils.JSONObject,
func (m *SDataSourceManager) GetMeasurements(query jsonutils.JSONObject,
measurementFilter, tagFilter string) (jsonutils.JSONObject,
error) {
ret := jsonutils.NewDict()
measurements, err := self.getMeasurementQueryInfluxdb(query, measurementFilter, tagFilter)
measurements, err := m.getMeasurementQueryInfluxdb(query, measurementFilter, tagFilter)
if err != nil {
return jsonutils.JSONNull, err
}
@@ -127,7 +128,7 @@ func (self *SDataSourceManager) GetMeasurements(query jsonutils.JSONObject,
return ret, nil
}
func (self *SDataSourceManager) getMeasurementQueryInfluxdb(query jsonutils.JSONObject,
func (m *SDataSourceManager) getMeasurementQueryInfluxdb(query jsonutils.JSONObject,
measurementFilter, tagFilter string) (rtnMeasurements []monitor.InfluxMeasurement, err error) {
database, _ := query.GetString("database")
if database == "" {
@@ -170,18 +171,18 @@ func (self *SDataSourceManager) getMeasurementQueryInfluxdb(query jsonutils.JSON
return
}
func (self *SDataSourceManager) GetMeasurementsWithDescriptionInfos(query jsonutils.JSONObject, tagFilter *monitor.MetricQueryTag) (jsonutils.JSONObject, error) {
func (m *SDataSourceManager) GetMeasurementsWithDescriptionInfos(query jsonutils.JSONObject, tagFilter *monitor.MetricQueryTag) (jsonutils.JSONObject, error) {
ret := jsonutils.NewDict()
rtnMeasurements := make([]monitor.InfluxMeasurement, 0)
measurements, err := MetricMeasurementManager.getMeasurementsFromDB()
if err != nil {
return jsonutils.JSONNull, errors.Wrap(err, "getMeasurementsFromDB")
}
filterMeasurements, err := self.filterMeasurementsByTime(measurements, query, tagFilter)
filterMeasurements, err := m.filterMeasurementsByTime(measurements, query, tagFilter)
if err != nil {
return jsonutils.JSONNull, errors.Wrap(err, "filterMeasurementsByTime error")
}
filterMeasurements = self.getMetricDescriptions(filterMeasurements)
filterMeasurements = m.getMetricDescriptions(filterMeasurements)
if len(filterMeasurements) != 0 {
rtnMeasurements = append(rtnMeasurements, filterMeasurements...)
}
@@ -212,7 +213,7 @@ func (self *SDataSourceManager) GetMeasurementsWithDescriptionInfos(query jsonut
return ret, nil
}
func (self *SDataSourceManager) GetMeasurementsWithOutTimeFilter(query jsonutils.JSONObject,
func (m *SDataSourceManager) GetMeasurementsWithOutTimeFilter(query jsonutils.JSONObject,
measurementFilter, tagFilter string) (jsonutils.JSONObject,
error) {
ret := jsonutils.NewDict()
@@ -257,7 +258,7 @@ func (self *SDataSourceManager) GetMeasurementsWithOutTimeFilter(query jsonutils
return ret, nil
}
func (self *SDataSourceManager) getMetricDescriptions(influxdbMeasurements []monitor.InfluxMeasurement) (
func (m *SDataSourceManager) getMetricDescriptions(influxdbMeasurements []monitor.InfluxMeasurement) (
descMeasurements []monitor.InfluxMeasurement) {
userCred := auth.AdminCredential()
listInput := new(monitor.MetricListInput)
@@ -315,13 +316,13 @@ func (self *SDataSourceManager) getMetricDescriptions(influxdbMeasurements []mon
return
}
func (self *SDataSourceManager) filterMeasurementsByTime(
func (m *SDataSourceManager) filterMeasurementsByTime(
measurements []monitor.InfluxMeasurement, query jsonutils.JSONObject, tagFilter *monitor.MetricQueryTag) ([]monitor.InfluxMeasurement, error) {
timeF, err := self.getFromAndToFromParam(query)
timeF, err := m.getFromAndToFromParam(query)
if err != nil {
return nil, err
}
filterMeasurements, err := self.getFilterMeasurementsParallel(timeF.From, timeF.To, measurements, tagFilter)
filterMeasurements, err := m.getFilterMeasurementsParallel(timeF.From, timeF.To, measurements, tagFilter)
if err != nil {
return nil, err
}
@@ -333,7 +334,7 @@ type timeFilter struct {
To string
}
func (self *SDataSourceManager) getFromAndToFromParam(query jsonutils.JSONObject) (timeFilter, error) {
func (m *SDataSourceManager) getFromAndToFromParam(query jsonutils.JSONObject) (timeFilter, error) {
timeF := timeFilter{}
from, _ := query.GetString("from")
if len(from) == 0 {
@@ -356,7 +357,7 @@ func (self *SDataSourceManager) getFromAndToFromParam(query jsonutils.JSONObject
return timeF, nil
}
func (self *SDataSourceManager) getFilterMeasurementsParallel(from, to string,
func (m *SDataSourceManager) getFilterMeasurementsParallel(from, to string,
measurements []monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) ([]monitor.InfluxMeasurement, error) {
filterMeasurements := make([]monitor.InfluxMeasurement, len(measurements))
ctx, cancel := context.WithTimeout(context.Background(), time.Second*60)
@@ -369,7 +370,7 @@ func (self *SDataSourceManager) getFilterMeasurementsParallel(from, to string,
measurementQueryGroup.Go(func() error {
errCh := make(chan error)
go func() {
ret, err := self.getFilterMeasurement(from, to, tmp, tagFilter)
ret, err := m.getFilterMeasurement(from, to, tmp, tagFilter)
if err != nil {
errCh <- errors.Wrapf(err, "getFilterMeasurement %d", index)
return
@@ -404,12 +405,20 @@ func (self *SDataSourceManager) getFilterMeasurementsParallel(from, to string,
return ret, nil
}
func (self *SDataSourceManager) getFilterMeasurement(from, to string, measurement monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) (*monitor.InfluxMeasurement, error) {
dds, _ := datasource.GetDefaultSource("")
func (m *SDataSourceManager) GetTSDBDriver() (tsdb.TsdbQueryEndpoint, error) {
ep, err := datasource.GetDefaultQueryEndpoint()
if err != nil {
return nil, errors.Wrap(err, "GetDefaultQueryEndpoint")
}
return ep, nil
}
func (m *SDataSourceManager) getFilterMeasurement(from, to string, measurement monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) (*monitor.InfluxMeasurement, error) {
dds, _ := datasource.GetDefaultSource("")
ep, err := m.GetTSDBDriver()
if err != nil {
return nil, errors.Wrap(err, "GetDefaultQueryEndpoint")
}
retMs, err := ep.FilterMeasurement(context.Background(), dds, from, to, &measurement, tagFilter)
if err != nil {
return nil, errors.Wrap(err, "Get endpoint filtered measurement")
@@ -433,7 +442,7 @@ func renderTimeFilter(from, to string) string {
}
func (self *SDataSourceManager) GetMetricMeasurement(userCred mcclient.TokenCredential, query jsonutils.JSONObject, tagFilter *monitor.MetricQueryTag) (jsonutils.JSONObject, error) {
func (m *SDataSourceManager) GetMetricMeasurement(userCred mcclient.TokenCredential, query jsonutils.JSONObject, tagFilter *monitor.MetricQueryTag) (jsonutils.JSONObject, error) {
database, _ := query.GetString("database")
if database == "" {
return jsonutils.JSONNull, merrors.NewArgIsEmptyErr("database")
@@ -450,7 +459,7 @@ func (self *SDataSourceManager) GetMetricMeasurement(userCred mcclient.TokenCred
if len(from) == 0 {
return jsonutils.JSONNull, merrors.NewArgIsEmptyErr("from")
}
timeF, err := self.getFromAndToFromParam(query)
timeF, err := m.getFromAndToFromParam(query)
if err != nil {
return nil, errors.Wrap(err, "getFromAndToFromParam")
}
@@ -467,12 +476,12 @@ func (self *SDataSourceManager) GetMetricMeasurement(userCred mcclient.TokenCred
return jsonutils.JSONNull, errors.Wrap(err, "getTagValues error")
}
self.filterRtnTags(output)
m.filterRtnTags(output)
return jsonutils.Marshal(output), nil
}
func (self *SDataSourceManager) filterRtnTags(output *monitor.InfluxMeasurement) {
func (m *SDataSourceManager) filterRtnTags(output *monitor.InfluxMeasurement) {
for _, tag := range []string{hostconsts.TELEGRAF_TAG_KEY_BRAND, hostconsts.TELEGRAF_TAG_KEY_PLATFORM,
hostconsts.TELEGRAF_TAG_KEY_HYPERVISOR} {
if val, ok := output.TagValue[tag]; ok {
@@ -504,7 +513,7 @@ func (self *SDataSourceManager) filterRtnTags(output *monitor.InfluxMeasurement)
output.TagKey = repTag
}
func (self *SDataSourceManager) filterTagValue(measurement monitor.InfluxMeasurement, timeF timeFilter,
func (m *SDataSourceManager) filterTagValue(measurement monitor.InfluxMeasurement, timeF timeFilter,
db *influxdb.SInfluxdb, tagValChan *influxdbTagValueChan, tagFilter string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Second*15)
tagValGroup2, _ := errgroup.WithContext(ctx)
@@ -515,7 +524,7 @@ func (self *SDataSourceManager) filterTagValue(measurement monitor.InfluxMeasure
for i, _ := range measurement.TagKey {
tmpkey := measurement.TagKey[i]
tagValGroup2.Go(func() error {
return self.getFilterMeasurementTagValue(&tagValChan2, timeF.From, timeF.To, measurement.FieldKey[0],
return m.getFilterMeasurementTagValue(&tagValChan2, timeF.From, timeF.To, measurement.FieldKey[0],
tmpkey, measurement, db, tagFilter)
})
}
@@ -580,7 +589,7 @@ type InfluxdbSubscription struct {
Url string
}
func (self *SDataSourceManager) AddSubscription(subscription InfluxdbSubscription) error {
func (m *SDataSourceManager) AddSubscription(subscription InfluxdbSubscription) error {
query := fmt.Sprintf("CREATE SUBSCRIPTION %s ON %s.%s DESTINATIONS ALL %s",
jsonutils.NewString(subscription.SubName).String(),
@@ -609,7 +618,7 @@ func (self *SDataSourceManager) AddSubscription(subscription InfluxdbSubscriptio
return nil
}
func (self *SDataSourceManager) DropSubscription(subscription InfluxdbSubscription) error {
func (m *SDataSourceManager) DropSubscription(subscription InfluxdbSubscription) error {
query := fmt.Sprintf("DROP SUBSCRIPTION %s ON %s.%s", jsonutils.NewString(subscription.SubName).String(),
jsonutils.NewString(subscription.DataBase).String(),
jsonutils.NewString(subscription.Rc).String(),
@@ -791,7 +800,7 @@ type influxdbTagValueChan struct {
count int
}
func (self *SDataSourceManager) getFilterMeasurementTagValue(tagValueChan *influxdbTagValueChan, from string,
func (m *SDataSourceManager) getFilterMeasurementTagValue(tagValueChan *influxdbTagValueChan, from string,
to string, field string, tagKey string,
measurement monitor.InfluxMeasurement, db *influxdb.SInfluxdb, tagFilter string) error {
var buffer bytes.Buffer
pkg/monitor/models/unifiedmonitor.go
+8 -28
View File
@@ -222,7 +222,7 @@ func (self *SUnifiedMonitorManager) PerformQuery(ctx context.Context, userCred m
if ownId == nil {
ownId = userCred
}
setDefaultValue(q, inputQuery, scope, ownId, false)
setDefaultValue(q, inputQuery, scope, ownId)
if err := self.ValidateInputQuery(q, inputQuery); err != nil {
return nil, errors.Wrapf(err, "ValidateInputQuery")
}
@@ -386,17 +386,14 @@ func (self *SUnifiedMonitorManager) ValidateInputQuery(query *monitor.AlertQuery
func setDefaultValue(
query *monitor.AlertQuery,
inputQuery *monitor.MetricQueryInput,
scope string, ownerId mcclient.IIdentityProvider,
isAlert bool) {
scope string, ownerId mcclient.IIdentityProvider) {
query.From = inputQuery.From
query.To = inputQuery.To
query.Model.Interval = inputQuery.Interval
metricMeasurement, _ := MetricMeasurementManager.GetCache().Get(query.Model.Measurement)
if isAlert {
checkQueryGroupBy(query, inputQuery)
}
checkQueryGroupBy(query, inputQuery)
if len(inputQuery.Interval) != 0 {
query.Model.GroupBy = append(query.Model.GroupBy,
@@ -432,18 +429,9 @@ func setDefaultValue(
query.Model.Database = database
}
for i, sel := range query.Model.Selects {
if len(sel) > 1 {
continue
}
if isAlert {
sel = append(sel, monitor.MetricQueryPart{
Type: "mean",
Params: []string{},
})
}
query.Model.Selects[i] = sel
}
drv, _ := DataSourceManager.GetTSDBDriver()
query = drv.FillSelect(query)
var projectId, domainId string
switch rbacscope.TRbacScope(scope) {
case rbacscope.ScopeProject:
@@ -503,16 +491,8 @@ func checkQueryGroupBy(query *monitor.AlertQuery, inputQuery *monitor.MetricQuer
if metricMeasurement != nil {
tagId = monitor.GetMeasurementTagIdKeyByResType(metricMeasurement.ResType)
}
if len(tagId) == 0 || (len(inputQuery.Slimit) != 0 && len(inputQuery.Soffset) != 0) {
tagId = "*"
}
if tagId != "" {
query.Model.GroupBy = append(query.Model.GroupBy,
monitor.MetricQueryPart{
Type: "field",
Params: []string{tagId},
})
}
drv, _ := DataSourceManager.GetTSDBDriver()
query = drv.FillGroupBy(query, inputQuery, tagId)
}
func fillSerieTags(series *monitor.TimeSeriesSlice) {
pkg/monitor/tsdb/driver/influxdb/influxdb.go
+28
View File
@@ -245,3 +245,31 @@ func (e *InfluxdbExecutor) FilterMeasurement(
return retMs, nil
}
func (e *InfluxdbExecutor) FillSelect(query *monitor.AlertQuery) *monitor.AlertQuery {
for i, sel := range query.Model.Selects {
if len(sel) > 1 {
continue
}
sel = append(sel, monitor.MetricQueryPart{
Type: "mean",
Params: []string{},
})
query.Model.Selects[i] = sel
}
return query
}
func (e *InfluxdbExecutor) FillGroupBy(query *monitor.AlertQuery, inputQuery *monitor.MetricQueryInput, tagId string) *monitor.AlertQuery {
if len(tagId) == 0 || (len(inputQuery.Slimit) != 0 && len(inputQuery.Soffset) != 0) {
tagId = "*"
}
if tagId != "" {
query.Model.GroupBy = append(query.Model.GroupBy,
monitor.MetricQueryPart{
Type: "field",
Params: []string{tagId},
})
}
return query
}
pkg/monitor/tsdb/driver/influxdb/response_parser.go
+21 -3
View File
@@ -23,6 +23,7 @@ import (
"yunion.io/x/log"
"yunion.io/x/pkg/errors"
"yunion.io/x/pkg/util/sets"
"yunion.io/x/onecloud/pkg/apis/monitor"
"yunion.io/x/onecloud/pkg/monitor/tsdb"
@@ -76,6 +77,23 @@ func (rp *ResponseParser) transformRows(rows []Row, queryResult *tsdb.QueryResul
func (rp *ResponseParser) transformRowsV2(rows []Row, queryResult *tsdb.QueryResult, query *Query) monitor.TimeSeriesSlice {
var result monitor.TimeSeriesSlice
// 添加值不同的 tag key
diffTagKeys := sets.NewString()
if len(rows) > 1 {
row0 := rows[0]
restRows := rows[1:]
for tagKey, tagVal := range row0.Tags {
for _, rr := range restRows {
resultTagVal := rr.Tags[tagKey]
if tagVal != resultTagVal {
diffTagKeys.Insert(tagKey)
break
}
}
}
}
for idx, row := range rows {
col := ""
columns := make([]string, 0)
@@ -106,21 +124,21 @@ func (rp *ResponseParser) transformRowsV2(rows []Row, queryResult *tsdb.QueryRes
tags[key] = val_
}
name := rp.formatSerieName(row, col, query)
ts := tsdb.NewTimeSeries(name, formatRawName(idx, name, query, tags), columns, points, tags)
ts := tsdb.NewTimeSeries(name, formatRawName(idx, name, query, tags, diffTagKeys), columns, points, tags)
result = append(result, ts)
}
return result
}
func formatRawName(idx int, name string, query *Query, tags map[string]string) string {
func formatRawName(idx int, name string, query *Query, tags map[string]string, diffTagKeys sets.String) string {
groupByTags := []string{}
for _, group := range query.GroupBy {
if group.Type == "tag" {
groupByTags = append(groupByTags, group.Params[0])
}
}
return tsdb.FormatRawName(idx, name, groupByTags, tags, nil)
return tsdb.FormatRawName(idx, name, groupByTags, tags, diffTagKeys)
}
func (rp *ResponseParser) transformRowToTable(row Row, table *tsdb.Table) *tsdb.Table {
pkg/monitor/tsdb/driver/victoriametrics/vm.go
+8
View File
@@ -304,3 +304,11 @@ func (vm *vmAdapter) FilterMeasurement(ctx context.Context, ds *tsdb.DataSource,
}
return retMs, nil
}
func (vm *vmAdapter) FillSelect(query *monitor.AlertQuery) *monitor.AlertQuery {
return query
}
func (vm *vmAdapter) FillGroupBy(query *monitor.AlertQuery, inputQuery *monitor.MetricQueryInput, tagId string) *monitor.AlertQuery {
return query
}
pkg/monitor/tsdb/query_endpoint.go
+2
View File
@@ -25,6 +25,8 @@ import (
type TsdbQueryEndpoint interface {
Query(ctx context.Context, ds *DataSource, query *TsdbQuery) (*Response, error)
FilterMeasurement(ctx context.Context, ds *DataSource, from, to string, ms *monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) (*monitor.InfluxMeasurement, error)
FillSelect(query *monitor.AlertQuery) *monitor.AlertQuery
FillGroupBy(query *monitor.AlertQuery, inputQuery *monitor.MetricQueryInput, tagId string) *monitor.AlertQuery
}
var registry map[string]GetTsdbQueryEndpointFn