diff --git a/pkg/monitor/models/commonalert.go b/pkg/monitor/models/commonalert.go index ac5467952f..75f8763c70 100644 --- a/pkg/monitor/models/commonalert.go +++ b/pkg/monitor/models/commonalert.go @@ -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 diff --git a/pkg/monitor/models/datasource.go b/pkg/monitor/models/datasource.go index 43bb0ed6e7..f25b85352b 100644 --- a/pkg/monitor/models/datasource.go +++ b/pkg/monitor/models/datasource.go @@ -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 diff --git a/pkg/monitor/models/unifiedmonitor.go b/pkg/monitor/models/unifiedmonitor.go index 5f38eea3ba..189fce668d 100644 --- a/pkg/monitor/models/unifiedmonitor.go +++ b/pkg/monitor/models/unifiedmonitor.go @@ -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) { diff --git a/pkg/monitor/tsdb/driver/influxdb/influxdb.go b/pkg/monitor/tsdb/driver/influxdb/influxdb.go index b5dfcddeea..25b9ff188f 100644 --- a/pkg/monitor/tsdb/driver/influxdb/influxdb.go +++ b/pkg/monitor/tsdb/driver/influxdb/influxdb.go @@ -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 +} diff --git a/pkg/monitor/tsdb/driver/influxdb/response_parser.go b/pkg/monitor/tsdb/driver/influxdb/response_parser.go index 0701afb699..2e14b301e7 100644 --- a/pkg/monitor/tsdb/driver/influxdb/response_parser.go +++ b/pkg/monitor/tsdb/driver/influxdb/response_parser.go @@ -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 { diff --git a/pkg/monitor/tsdb/driver/victoriametrics/vm.go b/pkg/monitor/tsdb/driver/victoriametrics/vm.go index 1b3788e2aa..a8385c3f8b 100644 --- a/pkg/monitor/tsdb/driver/victoriametrics/vm.go +++ b/pkg/monitor/tsdb/driver/victoriametrics/vm.go @@ -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 +} diff --git a/pkg/monitor/tsdb/query_endpoint.go b/pkg/monitor/tsdb/query_endpoint.go index 80c5af0994..e91c0e7ce8 100644 --- a/pkg/monitor/tsdb/query_endpoint.go +++ b/pkg/monitor/tsdb/query_endpoint.go @@ -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