diff --git a/pkg/monitor/dbinit/metric_dbinit.go b/pkg/monitor/dbinit/metric_dbinit.go index 57d08dbb08..e12faa1736 100644 --- a/pkg/monitor/dbinit/metric_dbinit.go +++ b/pkg/monitor/dbinit/metric_dbinit.go @@ -30,7 +30,7 @@ func RegistryMetricCreateInput(name, displayName, resType, database string, scor metricInitInputMap = make(map[string]monitor.MetricCreateInput) } if _, ok := metricInitInputMap[name]; ok { - log.Errorf("inputMeasurementName:%s has exist", name) + log.Fatalf("inputMeasurementName: %q has already existed.", name) return } metricInitInputMap[name] = monitor.MetricCreateInput{ @@ -205,9 +205,11 @@ func init() { }) // rds_conn - RegistryMetricCreateInput("rds_conn", "Rds connect", monitor.METRIC_RES_TYPE_RDS, + RegistryMetricCreateInput("rds_conn", "Rds connection", monitor.METRIC_RES_TYPE_RDS, monitor.METRIC_DATABASE_TELE, 5, []monitor.MetricFieldCreateInput{ newMetricFieldCreateInput("used_percent", "Connection usage", monitor.METRIC_UNIT_PERCENT, 1), + newMetricFieldCreateInput("active_count", "active connection count", monitor.METRIC_UNIT_COUNT, 2), + newMetricFieldCreateInput("failed_count", "failed connection count", monitor.METRIC_UNIT_COUNT, 3), }) // rds_cpu @@ -235,13 +237,6 @@ func init() { newMetricFieldCreateInput("used_percent", "Percentage of used disks", monitor.METRIC_UNIT_PERCENT, 1), }) - // rds_conn - RegistryMetricCreateInput("rds_conn", "Rds connection", monitor.METRIC_RES_TYPE_RDS, - monitor.METRIC_DATABASE_TELE, 5, []monitor.MetricFieldCreateInput{ - newMetricFieldCreateInput("active_count", "active connection count", monitor.METRIC_UNIT_COUNT, 1), - newMetricFieldCreateInput("failed_count", "failed connection count", monitor.METRIC_UNIT_COUNT, 2), - }) - // dcs_cpu RegistryMetricCreateInput("dcs_cpu", "Redis CPU usage", monitor.METRIC_RES_TYPE_REDIS, monitor.METRIC_DATABASE_TELE, 1, []monitor.MetricFieldCreateInput{ diff --git a/pkg/monitor/models/alertresource.go b/pkg/monitor/models/alertresource.go index 9ac929ce2f..85e3ccc5bf 100644 --- a/pkg/monitor/models/alertresource.go +++ b/pkg/monitor/models/alertresource.go @@ -486,7 +486,7 @@ func (manager *SAlertResourceManager) GetAdminRoleUsers(ctx context.Context, use session := auth.GetAdminSession(ctx, "") rid, err := identity.RolesV3.GetId(session, "admin", jsonutils.NewDict()) if err != nil { - errors.Errorf("get role id error:%v", err) + errors.Errorf("get role admin id error: %v", err) return } query.Add(jsonutils.NewString(rid), "role", "id") @@ -494,13 +494,13 @@ func (manager *SAlertResourceManager) GetAdminRoleUsers(ctx context.Context, use query.Set("offset", jsonutils.NewInt(int64(offset))) result, err := identity.RoleAssignments.List(session, query) if err != nil { - errors.Errorf("get admin role list error:%v", err) + errors.Errorf("get admin role list by query: %s, error: %v", query, err) return } for _, roleAssign := range result.Data { userId, err := roleAssign.GetString("user", "id") if err != nil { - log.Errorf("roleAssign:%v", roleAssign) + log.Errorf("get user.id from roleAssign %s: %v", roleAssign, err) continue } //_, err = .NotifyReceiver.GetById(session, userId, jsonutils.NewDict()) diff --git a/pkg/monitor/models/datasource.go b/pkg/monitor/models/datasource.go index 8d5ee04de9..d71ecd61d0 100644 --- a/pkg/monitor/models/datasource.go +++ b/pkg/monitor/models/datasource.go @@ -24,6 +24,8 @@ import ( "strings" "time" + "github.com/influxdata/promql/v2/pkg/labels" + "github.com/zexi/influxql-to-metricsql/converter/translator" "golang.org/x/sync/errgroup" "yunion.io/x/jsonutils" @@ -45,19 +47,16 @@ import ( "yunion.io/x/onecloud/pkg/util/stringutils2" ) +const ( + VICTORIA_METRICS_DB_TAG_KEY = "db" + VICTORIA_METRICS_DB_TAG_VAL_TELEGRAF = "telegraf" +) + var ( DataSourceManager *SDataSourceManager compile = regexp.MustCompile(`\w{8}(-\w{4}){3}-\w{12}`) ) -const ( - DefaultDataSource = "default" -) - -const ( - ErrDataSourceDefaultNotFound = errors.Error("Default data source not found") -) - func init() { DataSourceManager = &SDataSourceManager{ SStandaloneResourceBaseManager: db.NewStandaloneResourceBaseManager( @@ -170,12 +169,12 @@ func (self *SDataSourceManager) getMeasurementQueryInfluxdb(query jsonutils.JSON return } -func (self *SDataSourceManager) GetMeasurementsWithDescriptionInfos(query jsonutils.JSONObject, measurementFilter string, tagFilter *monitor.MetricQueryTag) (jsonutils.JSONObject, error) { +func (self *SDataSourceManager) GetMeasurementsWithDescriptionInfos(query jsonutils.JSONObject, tagFilter *monitor.MetricQueryTag) (jsonutils.JSONObject, error) { ret := jsonutils.NewDict() rtnMeasurements := make([]monitor.InfluxMeasurement, 0) - measurements, err := MetricMeasurementManager.getInfluxdbMeasurements() + measurements, err := MetricMeasurementManager.getMeasurementsFromDB() if err != nil { - return jsonutils.JSONNull, errors.Wrap(err, "getInfluxdbMeasurements") + return jsonutils.JSONNull, errors.Wrap(err, "getMeasurementsFromDB") } filterMeasurements, err := self.filterMeasurementsByTime(measurements, query, tagFilter) if err != nil { @@ -315,18 +314,13 @@ func (self *SDataSourceManager) getMetricDescriptions(influxdbMeasurements []mon return } -type influxdbQueryChan struct { - queryRtnChan chan monitor.InfluxMeasurement - count int -} - func (self *SDataSourceManager) filterMeasurementsByTime( measurements []monitor.InfluxMeasurement, query jsonutils.JSONObject, tagFilter *monitor.MetricQueryTag) ([]monitor.InfluxMeasurement, error) { timeF, err := self.getFromAndToFromParam(query) if err != nil { return nil, err } - filterMeasurements, err := self.getFilterMeasurementsAsync(timeF.From, timeF.To, measurements, tagFilter) + filterMeasurements, err := self.getFilterMeasurementsParallel(timeF.From, timeF.To, measurements, tagFilter) if err != nil { return nil, err } @@ -361,51 +355,65 @@ func (self *SDataSourceManager) getFromAndToFromParam(query jsonutils.JSONObject return timeF, nil } -func (self *SDataSourceManager) getFilterMeasurementsAsync(from, to string, +func (self *SDataSourceManager) getFilterMeasurementsParallel(from, to string, measurements []monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) ([]monitor.InfluxMeasurement, error) { - filterMeasurements := make([]monitor.InfluxMeasurement, 0) - queryChan := new(influxdbQueryChan) - queryChan.queryRtnChan = make(chan monitor.InfluxMeasurement, len(measurements)) - queryChan.count = len(measurements) + filterMeasurements := make([]monitor.InfluxMeasurement, len(measurements)) ctx, cancel := context.WithTimeout(context.Background(), time.Second*60) defer cancel() measurementQueryGroup, _ := errgroup.WithContext(ctx) - for i, _ := range measurements { - tmp := measurements[i] + for i := range measurements { + index := i + tmp := measurements[index] measurementQueryGroup.Go(func() error { - return self.getFilterMeasurement(queryChan, from, to, tmp, tagFilter) + errCh := make(chan error) + go func() { + ret, err := self.getFilterMeasurement(from, to, tmp, tagFilter) + if err != nil { + errCh <- errors.Wrapf(err, "getFilterMeasurement %d", index) + return + } + filterMeasurements[index] = *ret + errCh <- nil + }() + + for { + select { + case <-ctx.Done(): + return errors.Wrap(ctx.Err(), "filter measurement from TSDB") + case err := <-errCh: + if err != nil { + return err + } + return nil + } + } }) } - measurementQueryGroup.Go(func() error { - for i := 0; i < queryChan.count; i++ { - select { - case filterMeasurement := <-queryChan.queryRtnChan: - if len(filterMeasurement.Measurement) != 0 { - filterMeasurements = append(filterMeasurements, filterMeasurement) - } - case <-ctx.Done(): - return fmt.Errorf("filter measurement time out") - } + if err := measurementQueryGroup.Wait(); err != nil { + return nil, errors.Wrap(err, "measuremetnQueryGroup.Wait()") + } + ret := make([]monitor.InfluxMeasurement, 0) + for _, fm := range filterMeasurements { + if len(fm.Measurement) != 0 { + tmp := fm + ret = append(ret, tmp) } - return nil - }) - err := measurementQueryGroup.Wait() - return filterMeasurements, err + } + return ret, nil } -func (self *SDataSourceManager) getFilterMeasurement(queryChan *influxdbQueryChan, from, to string, measurement monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) error { +func (self *SDataSourceManager) getFilterMeasurement(from, to string, measurement monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) (*monitor.InfluxMeasurement, error) { dds, _ := datasource.GetDefaultSource("") ep, err := datasource.GetDefaultQueryEndpoint() if err != nil { - return errors.Wrap(err, "GetDefaultQueryEndpoint") + return nil, errors.Wrap(err, "GetDefaultQueryEndpoint") } retMs, err := ep.FilterMeasurement(context.Background(), dds, from, to, &measurement, tagFilter) if err != nil { - return errors.Wrap(err, "Get endpoint filtered measurement") + return nil, errors.Wrap(err, "Get endpoint filtered measurement") } - queryChan.queryRtnChan <- *retMs - return nil + return retMs, nil } func renderTimeFilter(from, to string) string { @@ -441,11 +449,6 @@ func (self *SDataSourceManager) GetMetricMeasurement(userCred mcclient.TokenCred if len(from) == 0 { return jsonutils.JSONNull, merrors.NewArgIsEmptyErr("from") } - dataSource, err := datasource.GetDefaultSource("") - if err != nil { - return jsonutils.JSONNull, errors.Wrap(err, "s.GetDefaultSource") - } - timeF, err := self.getFromAndToFromParam(query) if err != nil { return nil, errors.Wrap(err, "getFromAndToFromParam") @@ -453,51 +456,12 @@ func (self *SDataSourceManager) GetMetricMeasurement(userCred mcclient.TokenCred skipCheckSeries := jsonutils.QueryBoolean(query, "skip_check_series", false) - db := influxdb.NewInfluxdb(dataSource.Url) - db.SetDatabase(database) - output := new(monitor.InfluxMeasurement) output.Measurement = measurement output.Database = database output.TagValue = make(map[string][]string, 0) - // for _, val := range monitor.METRIC_ATTRI { - // if err := getAttributesOnMeasurement(database, val, output, db); err != nil { - // return jsonutils.JSONNull, errors.Wrap(err, "getAttributesOnMeasurement error") - // } - // } output.FieldKey = []string{field} - //err = getTagValue(database, output, db) - - // tagValChan := influxdbTagValueChan{ - // rtnChan: make(chan map[string][]string, len(output.FieldKey)), - // count: len(output.FieldKey), - // //count: 1, - // } - - //** ctx, cancel := context.WithTimeout(context.Background(), time.Second*30) - // tagValGroup, _ := errgroup.WithContext(ctx) - // defer cancel() - // tagValGroup.Go(func() error { - // return self.filterTagValue(*output, timeF, db, &tagValChan, tagFilter) - // }) - // tagValGroup.Go(func() error { - // for i := 0; i < tagValChan.count; i++ { - // select { - // case tagVal := <-tagValChan.rtnChan: - // if len(tagVal) != 0 { - // tagValUnion(output, tagVal) - // } - // case <-ctx.Done(): - // return fmt.Errorf("filter Union TagValue time out") - // } - // } - // return nil - // }) - // err = tagValGroup.Wait() - // if err != nil { - // return jsonutils.JSONNull, errors.Wrap(err, "getTagValue error") - //** } if err := getTagValues(userCred, output, timeF, tagFilter, skipCheckSeries); err != nil { return jsonutils.JSONNull, errors.Wrap(err, "getTagValues error") } @@ -515,13 +479,22 @@ func (self *SDataSourceManager) filterRtnTags(output *monitor.InfluxMeasurement) break } } - for _, tag := range []string{"source", "status", hostconsts.TELEGRAF_TAG_KEY_HOST_TYPE, - hostconsts.TELEGRAF_TAG_KEY_RES_TYPE, "is_vm", "os_type", hostconsts.TELEGRAF_TAG_KEY_PLATFORM, - hostconsts.TELEGRAF_TAG_KEY_HYPERVISOR, "domain_name", "region", "ips", "vip", "vip_eip", "eip", "eip_mode"} { + for _, tag := range []string{ + "source", "status", hostconsts.TELEGRAF_TAG_KEY_HOST_TYPE, hostconsts.TELEGRAF_TAG_KEY_RES_TYPE, + "is_vm", "os_type", hostconsts.TELEGRAF_TAG_KEY_PLATFORM, hostconsts.TELEGRAF_TAG_KEY_HYPERVISOR, + "domain_name", "region", "ips", "vip", "vip_eip", "eip", "eip_mode", + labels.MetricName, translator.UNION_RESULT_NAME, + } { if _, ok := output.TagValue[tag]; ok { delete(output.TagValue, tag) } } + // hide VictoriaMetrics telegraf db tag + if val, ok := output.TagValue[VICTORIA_METRICS_DB_TAG_KEY]; ok { + if len(val) == 1 && val[0] == VICTORIA_METRICS_DB_TAG_VAL_TELEGRAF { + delete(output.TagValue, VICTORIA_METRICS_DB_TAG_KEY) + } + } repTag := make([]string, 0) for tag, _ := range output.TagValue { @@ -726,8 +699,9 @@ func getTagValues(userCred mcclient.TokenCredential, output *monitor.InfluxMeasu } q := monitor.MetricQueryInput{ - From: timeF.From, - To: timeF.To, + From: timeF.From, + To: timeF.To, + Interval: "5m", MetricQuery: []*monitor.AlertQuery{ aq, }, diff --git a/pkg/monitor/models/metric.go b/pkg/monitor/models/metric.go index 968c9e6ace..ba2a9bb5c9 100644 --- a/pkg/monitor/models/metric.go +++ b/pkg/monitor/models/metric.go @@ -374,20 +374,20 @@ func (manager *SMetricMeasurementManager) getMeasurement(query *sqlchemy.SQuery) return measurements, nil } -func (manager *SMetricMeasurementManager) getInfluxdbMeasurements() (influxdbMeasurements []monitor.InfluxMeasurement, err error) { - metric, err := manager.getMeasurement(manager.Query()) +func (manager *SMetricMeasurementManager) getMeasurementsFromDB() ([]monitor.InfluxMeasurement, error) { + ms, err := manager.getMeasurement(manager.Query()) if err != nil { - return + return nil, errors.Wrap(err, "getMeasurement") } - for i, _ := range metric { - influxdbMeasurements = append(influxdbMeasurements, monitor.InfluxMeasurement{ - Database: metric[i].Database, - Measurement: metric[i].Name, - ResType: metric[i].ResType, - }) + ret := make([]monitor.InfluxMeasurement, len(ms)) + for i := range ms { + ret[i] = monitor.InfluxMeasurement{ + Database: ms[i].Database, + Measurement: ms[i].Name, + ResType: ms[i].ResType, + } } - return - + return ret, nil } func (measurement *SMetricMeasurement) getFieldsQuery() *sqlchemy.SQuery { @@ -416,7 +416,6 @@ func (manager *SMetricMeasurementManager) Init() error { } func (man *SMetricMeasurementManager) Run(ctx context.Context) error { - err := man.initJsonMetricInfo(ctx) if err != nil { return errors.Wrap(err, "init metric json error") diff --git a/pkg/monitor/models/unifiedmonitor.go b/pkg/monitor/models/unifiedmonitor.go index 2a02549e48..f92e6b18f9 100644 --- a/pkg/monitor/models/unifiedmonitor.go +++ b/pkg/monitor/models/unifiedmonitor.go @@ -22,6 +22,9 @@ import ( "strings" "time" + "github.com/influxdata/promql/v2/pkg/labels" + "github.com/zexi/influxql-to-metricsql/converter/translator" + "yunion.io/x/jsonutils" "yunion.io/x/log" "yunion.io/x/pkg/errors" @@ -79,7 +82,7 @@ func (self *SUnifiedMonitorManager) GetPropertyMeasurements(ctx context.Context, if err != nil { return nil, errors.Wrap(err, "getTagFilterByRequestQuery") } - return DataSourceManager.GetMeasurementsWithDescriptionInfos(query, "", filter) + return DataSourceManager.GetMeasurementsWithDescriptionInfos(query, filter) } func getTagFilterByRequestQuery(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (*monitor.MetricQueryTag, error) { @@ -507,12 +510,21 @@ func fillSerieTags(series *monitor.TimeSeriesSlice) { break } } - for _, tag := range []string{"source", "status", hostconsts.TELEGRAF_TAG_KEY_HOST_TYPE, - hostconsts.TELEGRAF_TAG_KEY_RES_TYPE, "cpu", "is_vm", "os_type", "domain_name", "region"} { + for _, tag := range []string{ + "source", "status", hostconsts.TELEGRAF_TAG_KEY_HOST_TYPE, + hostconsts.TELEGRAF_TAG_KEY_RES_TYPE, "cpu", + "is_vm", "os_type", "domain_name", "region", + labels.MetricName, translator.UNION_RESULT_NAME, + } { if _, ok := serie.Tags[tag]; ok { delete(serie.Tags, tag) } } + if val, ok := serie.Tags[VICTORIA_METRICS_DB_TAG_KEY]; ok { + if val == VICTORIA_METRICS_DB_TAG_VAL_TELEGRAF { + delete(serie.Tags, VICTORIA_METRICS_DB_TAG_KEY) + } + } (*series)[i] = serie } } diff --git a/pkg/monitor/tsdb/driver/victoriametrics/vm.go b/pkg/monitor/tsdb/driver/victoriametrics/vm.go index 265e53930f..70cb4cfd62 100644 --- a/pkg/monitor/tsdb/driver/victoriametrics/vm.go +++ b/pkg/monitor/tsdb/driver/victoriametrics/vm.go @@ -101,7 +101,7 @@ func queryRange(ctx context.Context, ds *tsdb.DataSource, tr *influxql.TimeRange return nil, errors.Wrap(err, "GetHttpClient of data source") } vmTr := NewTimeRangeByInfluxTimeRange(tr) - if interval <= 0 { + if interval <= 0 || interval < 1*time.Minute { interval = time.Minute * 5 } return cli.QueryRange(ctx, httpCli, promQL, interval, vmTr, false) @@ -252,6 +252,7 @@ func parsePointValue(value interface{}) interface{} { func (vm *vmAdapter) FilterMeasurement(ctx context.Context, ds *tsdb.DataSource, from, to string, ms *monitor.InfluxMeasurement, tagFilter *monitor.MetricQueryTag) (*monitor.InfluxMeasurement, error) { retMs := new(monitor.InfluxMeasurement) q := mod.NewAlertQuery(ms.Database, ms.Measurement).From(from).To(to) + q.Interval("5m") q.Selects().Select("*").LAST() if tagFilter != nil { q.Where().AddTag(tagFilter)