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

fix(monitor): fetch measurements from victoriametrics

Browse Source
This commit is contained in:
Zexi Li
2023-12-13 18:49:40 +08:00
parent a5f2194afc
commit aef50a49a6
6 changed files with 102 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/monitor/dbinit/metric_dbinit.go
+4 -9
View File
@@ -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{
pkg/monitor/models/alertresource.go
+3 -3
View File
@@ -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())
pkg/monitor/models/datasource.go
+67 -93
View File
@@ -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,
},
pkg/monitor/models/metric.go
+11 -12
View File
@@ -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")
pkg/monitor/models/unifiedmonitor.go
+15 -3
View File
@@ -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
}
}
pkg/monitor/tsdb/driver/victoriametrics/vm.go
+2 -1
View File
@@ -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)