From dab4bf91696474e6e287562802d4accf1c6a8f2c Mon Sep 17 00:00:00 2001 From: Jian Qiu Date: Fri, 6 Jun 2025 18:12:04 +0800 Subject: [PATCH] fix: http perf stats monitor (#22684) Co-authored-by: Qiu Jian --- pkg/apis/const.go | 25 + pkg/cloudmon/misc/httpstats.go | 437 ++++++++++++++++++ pkg/cloudmon/misc/httpstats_test.go | 215 +++++++++ pkg/cloudmon/misc/system.go | 332 ++++++------- pkg/monitor/dbinit/measurements/service.go | 6 + .../dbinit/measurements/service_http_code.go | 144 +++++- 6 files changed, 975 insertions(+), 184 deletions(-) create mode 100644 pkg/cloudmon/misc/httpstats.go create mode 100644 pkg/cloudmon/misc/httpstats_test.go diff --git a/pkg/apis/const.go b/pkg/apis/const.go index 5de084c2ed..54a5a2727c 100644 --- a/pkg/apis/const.go +++ b/pkg/apis/const.go @@ -90,6 +90,7 @@ var ( SERVICE_TYPE_CLOUDEVENT, SERVICE_TYPE_ANSIBLE, SERVICE_TYPE_INFLUXDB, + SERVICE_TYPE_VICTORIA_METRICS, SERVICE_TYPE_APIMAP, SERVICE_TYPE_LOG, "autoupdate", @@ -112,6 +113,30 @@ var ( SERVICE_TYPE_ETCD, "itsm", SERVICE_TYPE_NTP, + "kafka", + } + + EXTERNAL_SERVICES = []string{ + SERVICE_TYPE_OFFLINE_CLOUDMETA, + SERVICE_TYPE_CLOUDMETA, + SERVICE_TYPE_SCHEDULER, + SERVICE_TYPE_VNCPROXY, + SERVICE_TYPE_ETCD, + SERVICE_TYPE_INFLUXDB, + SERVICE_TYPE_INFLUXDB, + SERVICE_TYPE_VICTORIA_METRICS, + SERVICE_TYPE_LOG, + "s3gateway", + "common", + "websocket", + "echarts-ssr", + "cloudwatcher", + "cloudnet", + "repo", + SERVICE_TYPE_ETCD, + "itsm", + SERVICE_TYPE_NTP, + "kafka", } ) diff --git a/pkg/cloudmon/misc/httpstats.go b/pkg/cloudmon/misc/httpstats.go new file mode 100644 index 0000000000..572286d735 --- /dev/null +++ b/pkg/cloudmon/misc/httpstats.go @@ -0,0 +1,437 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package misc + +import ( + "fmt" + "time" + + "yunion.io/x/log" + + "yunion.io/x/onecloud/pkg/cloudmon/options" + "yunion.io/x/onecloud/pkg/util/influxdb" +) + +type SHttpStats struct { + HttpCode2xx float64 `json:"duration.2XX"` + HttpCode4xx float64 `json:"duration.4XX"` + HttpCode5xx float64 `json:"duration.5XX"` + HitHttpCode2xx int64 `json:"hit.2XX"` + HitHttpCode4xx int64 `json:"hit.4XX"` + HitHttpCode5xx int64 `json:"hit.5XX"` + + Method string `json:"method"` + Name string `json:"name"` + Path string `json:"path"` +} + +type sApiHttpStats struct { + SHttpStats + + Paths []SHttpStats `json:"paths"` +} + +func (apiStats *sApiHttpStats) convertSnapshot(now time.Time) *sHttpStatsSnapshot { + snapshot := &sHttpStatsSnapshot{ + snapshotAt: now, + stats: map[string]*SHttpStats{}, + } + apiStats.SHttpStats.Method = "any" + apiStats.SHttpStats.Path = "any" + apiStats.SHttpStats.Name = "any" + snapshot.stats[getStatsKey(apiStats.SHttpStats.Method, apiStats.SHttpStats.Path, apiStats.SHttpStats.Name)] = &apiStats.SHttpStats + for i := range apiStats.Paths { + pathStats := apiStats.Paths[i] + snapshot.stats[getStatsKey(pathStats.Method, pathStats.Path, pathStats.Name)] = &pathStats + } + return snapshot +} + +type sHttpStatsExt struct { + Duration2xx float64 `json:"duration_2xx"` + Duration4xx float64 `json:"duration_4xx"` + Duration5xx float64 `json:"duration_5xx"` + Hit2xx int64 `json:"hit_2xx"` + Hit4xx int64 `json:"hit_4xx"` + Hit5xx int64 `json:"hit_5xx"` + + Duration2xxDiff float64 `json:"duration_2xx_diff"` + Duration4xxDiff float64 `json:"duration_4xx_diff"` + Duration5xxDiff float64 `json:"duration_5xx_diff"` + Hit2xxDiff int64 `json:"hit_2xx_diff"` + Hit4xxDiff int64 `json:"hit_4xx_diff"` + Hit5xxDiff int64 `json:"hit_5xx_diff"` + + Method string `json:"method"` + Path string `json:"path"` + Name string `json:"name"` + + HasDiff bool `json:"has_diff"` +} + +func (v sHttpStatsExt) DelayMs2xx() float64 { + if v.Hit2xxDiff > 0 { + return v.Duration2xxDiff / float64(v.Hit2xxDiff) + } + return -1 +} + +func (v sHttpStatsExt) Qps2xx(interval time.Duration) float64 { + if interval > 0 { + return float64(v.Hit2xxDiff) / interval.Seconds() + } + return -1 +} + +func (v sHttpStatsExt) DelayMs4xx() float64 { + if v.Hit4xxDiff > 0 { + return v.Duration4xxDiff / float64(v.Hit4xxDiff) + } + return -1 +} + +func (v sHttpStatsExt) Qps4xx(interval time.Duration) float64 { + if interval > 0 { + return float64(v.Hit4xxDiff) / interval.Seconds() + } + return -1 +} + +func (v sHttpStatsExt) DelayMs5xx() float64 { + if v.Hit5xxDiff > 0 { + return v.Duration5xxDiff / float64(v.Hit5xxDiff) + } + return -1 +} + +func (v sHttpStatsExt) Qps5xx(interval time.Duration) float64 { + if interval > 0 { + return float64(v.Hit5xxDiff) / interval.Seconds() + } + return -1 +} + +func (v sHttpStatsExt) Qps(interval time.Duration) float64 { + if interval > 0 { + return float64(v.Hit2xxDiff+v.Hit4xxDiff+v.Hit5xxDiff) / interval.Seconds() + } + return -1 +} + +func (v sHttpStatsExt) DelayMs() float64 { + if v.HitDiff() > 0 { + return v.DurationMsDiff() / float64(v.HitDiff()) + } + return -1 +} + +func (v sHttpStatsExt) DurationMs() float64 { + return v.Duration2xx + v.Duration4xx + v.Duration5xx +} + +func (v sHttpStatsExt) Hit() int64 { + return v.Hit2xx + v.Hit4xx + v.Hit5xx +} + +func (v sHttpStatsExt) DurationMsDiff() float64 { + return v.Duration2xxDiff + v.Duration4xxDiff + v.Duration5xxDiff +} + +func (v sHttpStatsExt) HitDiff() int64 { + return v.Hit2xxDiff + v.Hit4xxDiff + v.Hit5xxDiff +} + +func (v sHttpStatsExt) Percent2xx() float64 { + if v.DurationMsDiff() > 0 { + return v.Duration2xxDiff * 100 / v.DurationMsDiff() + } + return -1 +} + +func (v sHttpStatsExt) PercentHit2xx() float64 { + if v.HitDiff() > 0 { + return float64(v.Hit2xxDiff) * 100 / float64(v.HitDiff()) + } + return -1 +} + +func (v sHttpStatsExt) Percent4xx() float64 { + if v.DurationMsDiff() > 0 { + return v.Duration4xxDiff * 100 / v.DurationMsDiff() + } + return -1 +} + +func (v sHttpStatsExt) PercentHit4xx() float64 { + if v.HitDiff() > 0 { + return float64(v.Hit4xxDiff) * 100 / float64(v.HitDiff()) + } + return -1 +} + +func (v sHttpStatsExt) Percent5xx() float64 { + if v.DurationMsDiff() > 0 { + return v.Duration5xxDiff * 100 / v.DurationMsDiff() + } + return -1 +} + +func (v sHttpStatsExt) PercentHit5xx() float64 { + if v.HitDiff() > 0 { + return float64(v.Hit5xxDiff) * 100 / float64(v.HitDiff()) + } + return -1 +} + +type sHttpStatsSnapshot struct { + snapshotAt time.Time + stats map[string]*SHttpStats +} + +type sHttpStatsDiff struct { + snapshotAt time.Time + interval time.Duration + stats map[string]*sHttpStatsExt +} + +func calculateHttpStatsDiff(prevSnap *sHttpStatsSnapshot, nowSnap *sHttpStatsSnapshot) *sHttpStatsDiff { + diff := sHttpStatsDiff{ + snapshotAt: nowSnap.snapshotAt, + stats: map[string]*sHttpStatsExt{}, + } + if prevSnap != nil { + rootKey := getStatsKey("any", "any", "any") + prevRoot := prevSnap.stats[rootKey] + nowRoot := nowSnap.stats[rootKey] + + if prevRoot.HttpCode2xx > nowRoot.HttpCode2xx || prevRoot.HttpCode4xx > nowRoot.HttpCode4xx || prevRoot.HttpCode5xx > nowRoot.HttpCode5xx || + prevRoot.HitHttpCode2xx > nowRoot.HitHttpCode2xx || prevRoot.HitHttpCode4xx > nowRoot.HitHttpCode4xx || prevRoot.HitHttpCode5xx > nowRoot.HitHttpCode5xx { + // detect a reset, skip this round + prevSnap = nil + } + } + if prevSnap != nil { + diff.interval = nowSnap.snapshotAt.Sub(prevSnap.snapshotAt) + } else { + intvMin := options.Options.CollectServiceMetricIntervalMinute + if intvMin <= 0 { + intvMin = 1 + } + diff.interval = time.Duration(intvMin) * time.Minute + } + + for k := range nowSnap.stats { + v := nowSnap.stats[k] + diffStats := sHttpStatsExt{ + Duration2xx: v.HttpCode2xx, + Duration4xx: v.HttpCode4xx, + Duration5xx: v.HttpCode5xx, + Hit2xx: v.HitHttpCode2xx, + Hit4xx: v.HitHttpCode4xx, + Hit5xx: v.HitHttpCode5xx, + + Method: v.Method, + Path: v.Path, + Name: v.Name, + } + if prevSnap != nil { + if prevStats, ok := prevSnap.stats[k]; ok { + diffStats.HasDiff = true + diffStats.Duration2xxDiff = v.HttpCode2xx - prevStats.HttpCode2xx + diffStats.Duration4xxDiff = v.HttpCode4xx - prevStats.HttpCode4xx + diffStats.Duration5xxDiff = v.HttpCode5xx - prevStats.HttpCode5xx + diffStats.Hit2xxDiff = v.HitHttpCode2xx - prevStats.HitHttpCode2xx + diffStats.Hit4xxDiff = v.HitHttpCode4xx - prevStats.HitHttpCode4xx + diffStats.Hit5xxDiff = v.HitHttpCode5xx - prevStats.HitHttpCode5xx + } + } + + diff.stats[k] = &diffStats + } + return &diff +} + +var ( + httpStatsSnapshot = map[string]*sHttpStatsSnapshot{} +) + +func getStatsKey(method, path, name string) string { + return fmt.Sprintf("%s.%s.%s", method, path, name) +} + +func getSnapshotKey(serviceName, url string) string { + return fmt.Sprintf("%s.%s", serviceName, url) +} + +func updateHttpStatsSnapshot(serviceName string, url string, now time.Time, apiStats sApiHttpStats, serviceType, regionId, version string) []influxdb.SMetricData { + snapshot := apiStats.convertSnapshot(now) + + snapshotKey := getSnapshotKey(serviceName, url) + var vdiffStats *sHttpStatsDiff + if prevSnap, ok := httpStatsSnapshot[snapshotKey]; ok { + vdiffStats = calculateHttpStatsDiff(prevSnap, snapshot) + } else { + // no prev records, just add + vdiffStats = calculateHttpStatsDiff(nil, snapshot) + } + httpStatsSnapshot[snapshotKey] = snapshot + + metrics := vdiffStats.metrics(serviceName, serviceType, regionId, version) + + log.Debugf("updateHttpStatsSnapshot %s %s snapshotAt: %s diffAt: %s intval: %f metrics: %d", serviceName, url, snapshot.snapshotAt, vdiffStats.snapshotAt, vdiffStats.interval.Seconds(), len(metrics)) + + return metrics +} + +func appendMetric(metrics []influxdb.SKeyValue, key string, v float64) []influxdb.SKeyValue { + if v >= 0 { + metrics = append(metrics, influxdb.SKeyValue{ + Key: key, + Value: fmt.Sprintf("%f", v), + }) + } + return metrics +} + +func (diff *sHttpStatsDiff) metrics(service, serviceType, regionId, version string) []influxdb.SMetricData { + metrics := make([]influxdb.SMetricData, 0) + genTags := func(v *sHttpStatsExt) []influxdb.SKeyValue { + return []influxdb.SKeyValue{ + { + Key: "service", + Value: service, + }, + { + Key: "service_type", + Value: serviceType, + }, + { + Key: "region", + Value: regionId, + }, + { + Key: "version", + Value: version, + }, + { + Key: "method", + Value: v.Method, + }, + { + Key: "path", + Value: v.Path, + }, + { + Key: "name", + Value: v.Name, + }, + { + Key: "interval_secs", + Value: fmt.Sprintf("%f", diff.interval.Seconds()), + }, + } + } + for k := range diff.stats { + v := diff.stats[k] + + // ignore stats with no hit + if v.Hit() <= 0 { + continue + } + // ignore stats with no diff + if v.HasDiff && v.HitDiff() <= 0 { + continue + } + + metric := influxdb.SMetricData{ + Name: METIRCY_TYPE_HTTP_REQUST, + Timestamp: diff.snapshotAt, + Tags: genTags(v), + Metrics: []influxdb.SKeyValue{ + { + Key: "duration_ms_any", + Value: fmt.Sprintf("%f", v.DurationMs()), + }, + { + Key: "hit_any", + Value: fmt.Sprintf("%d", v.Hit()), + }, + }, + } + if v.HasDiff { + metric.Metrics = appendMetric(metric.Metrics, "dura_ms_delta_any", v.DurationMsDiff()) + metric.Metrics = appendMetric(metric.Metrics, "hit_delta_any", float64(v.HitDiff())) + metric.Metrics = appendMetric(metric.Metrics, "delay_ms_any", v.DelayMs()) + metric.Metrics = appendMetric(metric.Metrics, "qps_any", v.Qps(diff.interval)) + } + metric.Metrics = append(metric.Metrics, []influxdb.SKeyValue{ + { + Key: "duration_ms_2xx", + Value: fmt.Sprintf("%f", v.Duration2xx), + }, + { + Key: "hit_2xx", + Value: fmt.Sprintf("%d", v.Hit2xx), + }, + }...) + if v.HasDiff { + metric.Metrics = appendMetric(metric.Metrics, "dura_ms_delta_2xx", v.Duration2xxDiff) + metric.Metrics = appendMetric(metric.Metrics, "hit_delta_2xx", float64(v.Hit2xxDiff)) + metric.Metrics = appendMetric(metric.Metrics, "delay_ms_2xx", v.DelayMs2xx()) + metric.Metrics = appendMetric(metric.Metrics, "percent_hit_2xx", v.PercentHit2xx()) + metric.Metrics = appendMetric(metric.Metrics, "percent_duration_2xx", v.Percent2xx()) + metric.Metrics = appendMetric(metric.Metrics, "qps_2xx", v.Qps2xx(diff.interval)) + } + metric.Metrics = append(metric.Metrics, []influxdb.SKeyValue{ + { + Key: "duration_ms_4xx", + Value: fmt.Sprintf("%f", v.Duration4xx), + }, + { + Key: "hit_4xx", + Value: fmt.Sprintf("%d", v.Hit4xx), + }, + }...) + if v.HasDiff { + metric.Metrics = appendMetric(metric.Metrics, "dura_ms_delta_4xx", v.Duration4xxDiff) + metric.Metrics = appendMetric(metric.Metrics, "hit_delta_4xx", float64(v.Hit4xxDiff)) + metric.Metrics = appendMetric(metric.Metrics, "delay_ms_4xx", v.DelayMs4xx()) + metric.Metrics = appendMetric(metric.Metrics, "percent_hit_4xx", v.PercentHit4xx()) + metric.Metrics = appendMetric(metric.Metrics, "percent_duration_4xx", v.Percent4xx()) + metric.Metrics = appendMetric(metric.Metrics, "qps_4xx", v.Qps4xx(diff.interval)) + } + metric.Metrics = append(metric.Metrics, []influxdb.SKeyValue{ + { + Key: "duration_ms_5xx", + Value: fmt.Sprintf("%f", v.Duration5xx), + }, + { + Key: "hit_5xx", + Value: fmt.Sprintf("%d", v.Hit5xx), + }, + }...) + if v.HasDiff { + metric.Metrics = appendMetric(metric.Metrics, "dura_ms_delta_5xx", v.Duration5xxDiff) + metric.Metrics = appendMetric(metric.Metrics, "hit_delta_5xx", float64(v.Hit5xxDiff)) + metric.Metrics = appendMetric(metric.Metrics, "delay_ms_5xx", v.DelayMs5xx()) + metric.Metrics = appendMetric(metric.Metrics, "percent_hit_5xx", v.PercentHit5xx()) + metric.Metrics = appendMetric(metric.Metrics, "percent_duration_5xx", v.Percent5xx()) + metric.Metrics = appendMetric(metric.Metrics, "qps_5xx", v.Qps5xx(diff.interval)) + } + + metrics = append(metrics, metric) + } + return metrics +} diff --git a/pkg/cloudmon/misc/httpstats_test.go b/pkg/cloudmon/misc/httpstats_test.go new file mode 100644 index 0000000000..364591405a --- /dev/null +++ b/pkg/cloudmon/misc/httpstats_test.go @@ -0,0 +1,215 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package misc + +import ( + "testing" + "time" + + "yunion.io/x/jsonutils" + "yunion.io/x/pkg/util/timeutils" +) + +func TestUnmarshalHttpStats(t *testing.T) { + input := `{"duration.2XX":4532188.710192006, "duration.4XX":196870.22898800002, "duration.5XX":32787.14909299999, "hit.2XX":88678, "hit.4XX":1748, "hit.5XX":179, "paths":[{"duration.2XX":150, "duration.4XX":250, "duration.5XX":310, "hit.2XX":1500, "hit.4XX":2500, "hit.5XX":3500, "method":"GET", "path":"/servers", "name":"list_servers"}, {"duration.2XX":150, "duration.4XX":250, "duration.5XX":350, "hit.2XX":1500, "hit.4XX":2500, "hit.5XX":3500, "method":"POST", "path":"/servers", "name":"create_servers"}]}` + inputJson, err := jsonutils.ParseString(input) + if err != nil { + t.Fatal(err) + } + t.Logf("json: %s", inputJson.PrettyString()) + var stats sApiHttpStats + err = inputJson.Unmarshal(&stats) + if err != nil { + t.Fatal(err) + } + if stats.HttpCode2xx != 4532188.710192006 { + t.Fatalf("duration.2XX: %f", stats.HttpCode2xx) + } + if stats.HttpCode4xx != 196870.22898800002 { + t.Fatalf("duration.4XX: %f", stats.HttpCode4xx) + } + if stats.HttpCode5xx != 32787.14909299999 { + t.Fatalf("duration.5XX: %f", stats.HttpCode5xx) + } + if stats.HitHttpCode2xx != 88678 { + t.Fatalf("hit.2XX: %d", stats.HitHttpCode2xx) + } + if stats.HitHttpCode4xx != 1748 { + t.Fatalf("hit.4XX: %d", stats.HitHttpCode4xx) + } + if stats.HitHttpCode5xx != 179 { + t.Fatalf("hit.5XX: %d", stats.HitHttpCode5xx) + } +} + +func TestHttpStats(t *testing.T) { + cases := []struct { + prevTime time.Time + nowTime time.Time + prevStats *sApiHttpStats + nowStats sApiHttpStats + }{ + { + nowTime: func() time.Time { + tm, _ := timeutils.ParseTimeStr("2025-06-01 00:01:00") + return tm + }(), + nowStats: sApiHttpStats{ + SHttpStats: SHttpStats{ + HttpCode2xx: 400, + HttpCode4xx: 700, + HttpCode5xx: 1000, + HitHttpCode2xx: 4000, + HitHttpCode4xx: 7000, + HitHttpCode5xx: 10000, + }, + Paths: []SHttpStats{ + { + HttpCode2xx: 150, + HttpCode4xx: 250, + HttpCode5xx: 310, + HitHttpCode2xx: 1500, + HitHttpCode4xx: 2500, + HitHttpCode5xx: 3500, + Method: "GET", + Path: "/servers", + Name: "list_servers", + }, + { + HttpCode2xx: 150, + HttpCode4xx: 250, + HttpCode5xx: 350, + HitHttpCode2xx: 1500, + HitHttpCode4xx: 2500, + HitHttpCode5xx: 3500, + Method: "POST", + Path: "/servers", + Name: "create_servers", + }, + { + HttpCode2xx: 100, + HttpCode4xx: 200, + HttpCode5xx: 300, + HitHttpCode2xx: 1000, + HitHttpCode4xx: 2000, + HitHttpCode5xx: 3000, + Method: "PUT", + Path: "/servers/*", + Name: "update_servers", + }, + }, + }, + }, + { + prevTime: func() time.Time { + tm, _ := timeutils.ParseTimeStr("2025-06-01 00:00:00") + return tm + }(), + nowTime: func() time.Time { + tm, _ := timeutils.ParseTimeStr("2025-06-01 00:01:00") + return tm + }(), + prevStats: &sApiHttpStats{ + SHttpStats: SHttpStats{ + HttpCode2xx: 200, + HttpCode4xx: 400, + HttpCode5xx: 600, + HitHttpCode2xx: 2000, + HitHttpCode4xx: 4000, + HitHttpCode5xx: 6000, + }, + Paths: []SHttpStats{ + { + HttpCode2xx: 100, + HttpCode4xx: 200, + HttpCode5xx: 300, + HitHttpCode2xx: 1000, + HitHttpCode4xx: 2000, + HitHttpCode5xx: 3000, + Method: "GET", + Path: "/servers", + Name: "list_servers", + }, + { + HttpCode2xx: 100, + HttpCode4xx: 200, + HttpCode5xx: 300, + HitHttpCode2xx: 1000, + HitHttpCode4xx: 2000, + HitHttpCode5xx: 3000, + Method: "POST", + Path: "/servers", + Name: "create_servers", + }, + }, + }, + nowStats: sApiHttpStats{ + SHttpStats: SHttpStats{ + HttpCode2xx: 400, + HttpCode4xx: 700, + HttpCode5xx: 1000, + HitHttpCode2xx: 4000, + HitHttpCode4xx: 7000, + HitHttpCode5xx: 10000, + }, + Paths: []SHttpStats{ + { + HttpCode2xx: 150, + HttpCode4xx: 250, + HttpCode5xx: 310, + HitHttpCode2xx: 1500, + HitHttpCode4xx: 2500, + HitHttpCode5xx: 3500, + Method: "GET", + Path: "/servers", + Name: "list_servers", + }, + { + HttpCode2xx: 150, + HttpCode4xx: 250, + HttpCode5xx: 350, + HitHttpCode2xx: 1500, + HitHttpCode4xx: 2500, + HitHttpCode5xx: 3500, + Method: "POST", + Path: "/servers", + Name: "create_servers", + }, + { + HttpCode2xx: 100, + HttpCode4xx: 200, + HttpCode5xx: 300, + HitHttpCode2xx: 1000, + HitHttpCode4xx: 2000, + HitHttpCode5xx: 3000, + Method: "PUT", + Path: "/servers/*", + Name: "update_servers", + }, + }, + }, + }, + } + for _, c := range cases { + var prev *sHttpStatsSnapshot + if c.prevStats != nil { + prev = c.prevStats.convertSnapshot(c.prevTime) + } + curr := c.nowStats.convertSnapshot(c.nowTime) + diff := calculateHttpStatsDiff(prev, curr) + metrics := diff.metrics("test", "test", "test", "test") + t.Log(jsonutils.Marshal(metrics).PrettyString()) + } +} diff --git a/pkg/cloudmon/misc/system.go b/pkg/cloudmon/misc/system.go index b51aef96f5..b82b6bc1aa 100644 --- a/pkg/cloudmon/misc/system.go +++ b/pkg/cloudmon/misc/system.go @@ -19,6 +19,9 @@ import ( "fmt" "io" "net/http" + "net/url" + "regexp" + "strings" "time" "yunion.io/x/jsonutils" @@ -26,7 +29,6 @@ import ( "yunion.io/x/pkg/errors" "yunion.io/x/pkg/gotypes" "yunion.io/x/pkg/util/httputils" - "yunion.io/x/pkg/util/version" "yunion.io/x/pkg/utils" "yunion.io/x/onecloud/pkg/apis" @@ -38,6 +40,7 @@ import ( "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/mcclient/modules/compute" "yunion.io/x/onecloud/pkg/mcclient/modules/identity" + baseoptions "yunion.io/x/onecloud/pkg/mcclient/options" "yunion.io/x/onecloud/pkg/util/influxdb" ) @@ -50,19 +53,73 @@ const ( METIRCY_TYPE_PROCESS = "process" ) -func getEndpoints(ctx context.Context, s *mcclient.ClientSession) ([]api.EndpointDetails, error) { - resp, err := identity.EndpointsV3.List(s, jsonutils.Marshal(map[string]string{ - "scope": "system", - "enable": "true", - "details": "true", - "interface": "internal", - "limit": "50", - })) - if err != nil { - return nil, errors.Wrapf(err, "Endpoints.List") +func getEndpoints(s *mcclient.ClientSession) ([]api.EndpointDetails, error) { + ret := make([]api.EndpointDetails, 0) + params := baseoptions.BaseListOptions{} + limit := 1024 + params.Limit = &limit + boolTrue := true + params.Details = &boolTrue + params.Scope = "system" + params.Filter = []string{ + "interface.equals(internal)", + "enabled.equals(1)", } - ret := []api.EndpointDetails{} - return ret, jsonutils.Update(&ret, resp.Data) + + for { + offset := len(ret) + params.Offset = &offset + resp, err := identity.EndpointsV3.List(s, jsonutils.Marshal(params)) + if err != nil { + return nil, errors.Wrapf(err, "Endpoints.List") + } + for i := range resp.Data { + endpoint := api.EndpointDetails{} + err := resp.Data[i].Unmarshal(&endpoint) + if err != nil { + return nil, errors.Wrapf(err, "Unmarshal") + } + ret = append(ret, endpoint) + } + if len(ret) >= resp.Total { + break + } + } + return ret, nil +} + +func getHosts(s *mcclient.ClientSession) ([]compute_api.HostDetails, error) { + params := compute_api.HostListInput{} + boolFalse := false + limit := 100 + params.Limit = &limit + params.Brand = []string{compute_api.CLOUD_PROVIDER_ONECLOUD} + params.Scope = "system" + params.Status = []string{compute_api.HOST_STATUS_RUNNING} + params.HostStatus = []string{compute_api.HOST_ONLINE} + params.Details = &boolFalse + + hosts := []compute_api.HostDetails{} + for { + offset := len(hosts) + params.Offset = &offset + resp, err := compute.Hosts.List(s, jsonutils.Marshal(params)) + if err != nil { + return nil, errors.Wrapf(err, "Hosts.List") + } + for i := range resp.Data { + host := compute_api.HostDetails{} + err := resp.Data[i].Unmarshal(&host) + if err != nil { + return nil, errors.Wrapf(err, "Unmarshal") + } + hosts = append(hosts, host) + } + if len(hosts) >= resp.Total { + break + } + } + return hosts, nil } func CollectServiceMetrics(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) { @@ -77,18 +134,13 @@ func CollectServiceMetrics(ctx context.Context, userCred mcclient.TokenCredentia tk := auth.AdminCredential().GetTokenString() err = func() error { - endpoints, err := getEndpoints(ctx, s) + endpoints, err := getEndpoints(s) if err != nil { return errors.Wrapf(err, "getEndpoints") } metrics := []influxdb.SMetricData{} for _, ep := range endpoints { - if utils.IsInStringArray(ep.ServiceType, apis.NO_RESOURCE_SERVICES) || utils.IsInStringArray(ep.ServiceType, []string{ - apis.SERVICE_TYPE_IMAGE, - apis.SERVICE_TYPE_MONITOR, - apis.SERVICE_TYPE_VICTORIA_METRICS, - "k8s", - }) { + if utils.IsInStringArray(ep.ServiceType, apis.EXTERNAL_SERVICES) { continue } url := httputils.JoinPath(ep.Url, "version") @@ -114,52 +166,40 @@ func CollectServiceMetrics(ctx context.Context, userCred mcclient.TokenCredentia } metrics = append(metrics, part...) } - return influxdb.SendMetrics(urls, SYSTEM_METRIC_DATABASE, metrics, false) + if len(metrics) > 0 { + err := influxdb.SendMetrics(urls, SYSTEM_METRIC_DATABASE, metrics, false) + if err != nil { + return errors.Wrapf(err, "SendMetrics") + } + } + return nil }() if err != nil { log.Errorf("collect service metric error: %v", err) } - params := compute_api.HostListInput{} - limit := 20 - params.Limit = &limit - params.Brand = []string{compute_api.CLOUD_PROVIDER_ONECLOUD} - params.Scope = "system" - params.Status = []string{compute_api.HOST_STATUS_RUNNING} - details := false - params.Details = &details - hosts := []compute_api.HostDetails{} - for { - offset := len(hosts) - params.Offset = &offset - resp, err := compute.Hosts.List(s, jsonutils.Marshal(params)) + + { + hosts, err := getHosts(s) if err != nil { - return + log.Errorf("get hosts error: %v", err) } - part := []compute_api.HostDetails{} - err = jsonutils.Update(&part, resp.Data) - if err != nil { - return + metrics := []influxdb.SMetricData{} + for _, host := range hosts { + part := collectHostMetrics(ctx, host, tk) + metrics = append(metrics, part...) } - hosts = append(hosts, part...) - if len(hosts) >= resp.Total { - break + if len(metrics) > 0 { + err := influxdb.SendMetrics(urls, SYSTEM_METRIC_DATABASE, metrics, false) + if err != nil { + log.Errorf("send host metrics error: %v", err) + } } } - metrics := []influxdb.SMetricData{} - for _, host := range hosts { - service := fmt.Sprintf("host-%s", host.Name) - part, err := collectWorkerMetrics(ctx, host.ManagerUri, service, version.GetShortString(), tk) - if err != nil { - log.Errorf("collect host %s metric error: %v", service, err) - continue - } - metrics = append(metrics, part...) - } - influxdb.SendMetrics(urls, SYSTEM_METRIC_DATABASE, metrics, false) } -func collectStatsMetrics(ctx context.Context, ep api.EndpointDetails, version, token string) ([]influxdb.SMetricData, error) { - statsUrl := httputils.JoinPath(ep.Url, "stats") +func collectApiStatsMetrics(ctx context.Context, serviceName string, serviceType string, regionId string, url string, version, token string) ([]influxdb.SMetricData, error) { + log.Debugf("collectApiStatsMetrics %s %s %s %s %s", serviceName, serviceType, regionId, url, version) + statsUrl := httputils.JoinPath(baseUrlF(url), "stats") hdr := http.Header{} hdr.Set("X-Auth-Token", token) _, ret, err := httputils.JSONRequest( @@ -175,131 +215,19 @@ func collectStatsMetrics(ctx context.Context, ep api.EndpointDetails, version, t return []influxdb.SMetricData{}, nil } - stats := struct { - HttpCode2xx float64 `json:"duration.2XX"` - HttpCode4xx float64 `json:"duration.4XX"` - HttpCode5xx float64 `json:"duration.5XX"` - HitHttpCode2xx int `json:"hit.2XX"` - HitHttpCode4xx int `json:"hit.4XX"` - HitHttpCode5xx int `json:"hit.5XX"` - Paths []struct { - HttpCode2xx float64 `json:"duration.2XX"` - HttpCode4xx float64 `json:"duration.4XX"` - HttpCode5xx float64 `json:"duration.5XX"` - HitHttpCode2xx int `json:"hit.2XX"` - HitHttpCode4xx int `json:"hit.4XX"` - HitHttpCode5xx int `json:"hit.5XX"` - Method string - Name string - Path string - } `json:"paths"` - }{} + stats := sApiHttpStats{} err = ret.Unmarshal(&stats) if err != nil { return nil, errors.Wrapf(err, "Unmarshal") } - result := []influxdb.SMetricData{} - metric := influxdb.SMetricData{ - Name: METIRCY_TYPE_HTTP_REQUST, - Timestamp: time.Now(), - Tags: []influxdb.SKeyValue{ - { - Key: "version", - Value: version, - }, - { - Key: "service", - Value: ep.ServiceName, - }, - }, - Metrics: []influxdb.SKeyValue{ - { - Key: "duration.2xx", - Value: fmt.Sprintf("%.2f", stats.HttpCode2xx), - }, - { - Key: "duration.4xx", - Value: fmt.Sprintf("%.2f", stats.HttpCode4xx), - }, - { - Key: "duration.5xx", - Value: fmt.Sprintf("%.2f", stats.HttpCode5xx), - }, - { - Key: "hit.2xx", - Value: fmt.Sprintf("%d", stats.HitHttpCode2xx), - }, - { - Key: "hit.4xx", - Value: fmt.Sprintf("%d", stats.HitHttpCode4xx), - }, - { - Key: "hit.5xx", - Value: fmt.Sprintf("%d", stats.HitHttpCode5xx), - }, - }, - } - result = append(result, metric) - for _, path := range stats.Paths { - metric = influxdb.SMetricData{ - Name: METIRCY_TYPE_HTTP_REQUST, - Timestamp: time.Now(), - Tags: []influxdb.SKeyValue{ - { - Key: "version", - Value: version, - }, - { - Key: "service", - Value: ep.ServiceName, - }, - { - Key: "method", - Value: path.Method, - }, - { - Key: "path", - Value: path.Path, - }, - { - Key: "url", - Value: path.Name, - }, - }, - Metrics: []influxdb.SKeyValue{ - { - Key: "duration.2xx", - Value: fmt.Sprintf("%.2f", path.HttpCode2xx), - }, - { - Key: "duration.4xx", - Value: fmt.Sprintf("%.2f", path.HttpCode4xx), - }, - { - Key: "duration.5xx", - Value: fmt.Sprintf("%.2f", path.HttpCode5xx), - }, - { - Key: "hit.2xx", - Value: fmt.Sprintf("%d", path.HitHttpCode2xx), - }, - { - Key: "hit.4xx", - Value: fmt.Sprintf("%d", path.HitHttpCode4xx), - }, - { - Key: "hit.5xx", - Value: fmt.Sprintf("%d", path.HitHttpCode5xx), - }, - }, - } - result = append(result, metric) - } - return result, nil + + metrics := updateHttpStatsSnapshot(serviceName, url, time.Now(), stats, serviceType, regionId, version) + + return metrics, nil } -func collectWorkerMetrics(ctx context.Context, url, service, version, token string) ([]influxdb.SMetricData, error) { - statsUrl := httputils.JoinPath(url, "worker_stats") +func collectWorkerMetrics(ctx context.Context, url, service, serviceType, regionId, version, token string) ([]influxdb.SMetricData, error) { + statsUrl := httputils.JoinPath(baseUrlF(url), "worker_stats") hdr := http.Header{} hdr.Set("X-Auth-Token", token) _, ret, err := httputils.JSONRequest( @@ -344,6 +272,14 @@ func collectWorkerMetrics(ctx context.Context, url, service, version, token stri Key: "service", Value: service, }, + { + Key: "service_type", + Value: serviceType, + }, + { + Key: "region", + Value: regionId, + }, { Key: "worker_name", Value: worker.Name, @@ -366,6 +302,14 @@ func collectWorkerMetrics(ctx context.Context, url, service, version, token stri Key: "queue_cnt", Value: fmt.Sprintf("%d", worker.QueueCnt), }, + { + Key: "total_workload", + Value: fmt.Sprintf("%d", worker.ActiveWorkerCnt+worker.QueueCnt+worker.DetachWorkerCnt), + }, + { + Key: "active_workload", + Value: fmt.Sprintf("%d", worker.ActiveWorkerCnt+worker.DetachWorkerCnt), + }, }, } result = append(result, metric) @@ -375,7 +319,7 @@ func collectWorkerMetrics(ctx context.Context, url, service, version, token stri } func collectDatabaseMetrics(ctx context.Context, ep api.EndpointDetails, version, token string) ([]influxdb.SMetricData, error) { - statsUrl := httputils.JoinPath(ep.Url, "db_stats") + statsUrl := httputils.JoinPath(baseUrlF(ep.Url), "db_stats") hdr := http.Header{} hdr.Set("X-Auth-Token", token) _, ret, err := httputils.JSONRequest( @@ -462,7 +406,7 @@ func collectDatabaseMetrics(ctx context.Context, ep api.EndpointDetails, version } func collectProcessMetrics(ctx context.Context, ep api.EndpointDetails, version, token string) ([]influxdb.SMetricData, error) { - statsUrl := httputils.JoinPath(ep.Url, "process_stats") + statsUrl := httputils.JoinPath(baseUrlF(ep.Url), "process_stats") hdr := http.Header{} hdr.Set("X-Auth-Token", token) _, ret, err := httputils.JSONRequest( @@ -518,14 +462,28 @@ func collectProcessMetrics(ctx context.Context, ep api.EndpointDetails, version, return []influxdb.SMetricData{metric}, nil } +func baseUrlF(baseurl string) string { + obj, _ := url.Parse(baseurl) + lastSlashPos := strings.LastIndex(obj.Path, "/") + if lastSlashPos >= 0 { + lastSeg := obj.Path[lastSlashPos+1:] + verReg := regexp.MustCompile(`^v\d+`) + if verReg.MatchString(lastSeg) { + obj.Path = obj.Path[:lastSlashPos] + } + } + ret := obj.String() + return ret +} + func collectServiceMetrics(ctx context.Context, ep api.EndpointDetails, version, token string) ([]influxdb.SMetricData, error) { ret, errs := []influxdb.SMetricData{}, []error{} - stats, err := collectStatsMetrics(ctx, ep, version, token) + stats, err := collectApiStatsMetrics(ctx, ep.ServiceName, ep.ServiceType, ep.RegionId, ep.Url, version, token) if err != nil { errs = append(errs, err) } ret = append(ret, stats...) - worker, err := collectWorkerMetrics(ctx, ep.Url, ep.ServiceName, version, token) + worker, err := collectWorkerMetrics(ctx, ep.Url, ep.ServiceName, ep.ServiceType, ep.RegionId, version, token) if err != nil { errs = append(errs, err) } @@ -542,3 +500,21 @@ func collectServiceMetrics(ctx context.Context, ep api.EndpointDetails, version, ret = append(ret, process...) return ret, errors.NewAggregate(errs) } + +func collectHostMetrics(ctx context.Context, host compute_api.HostDetails, token string) []influxdb.SMetricData { + metrics := []influxdb.SMetricData{} + service := fmt.Sprintf("host-%s", host.Name) + part, err := collectWorkerMetrics(ctx, host.ManagerUri, service, "host", host.Region, host.Version, token) + if err != nil { + log.Errorf("collect host %s metric error: %v", service, err) + } else { + metrics = append(metrics, part...) + } + part, err = collectApiStatsMetrics(ctx, service, "host", host.Region, host.ManagerUri, host.Version, token) + if err != nil { + log.Errorf("collect host %s metric error: %v", service, err) + } else { + metrics = append(metrics, part...) + } + return metrics +} diff --git a/pkg/monitor/dbinit/measurements/service.go b/pkg/monitor/dbinit/measurements/service.go index cc44e11aad..647f43f11e 100644 --- a/pkg/monitor/dbinit/measurements/service.go +++ b/pkg/monitor/dbinit/measurements/service.go @@ -36,6 +36,12 @@ var worker = SMeasurement{ { "queue_cnt", "Worker Queue Count", monitor.METRIC_UNIT_NULL, }, + { + "total_workload", "Total workload", monitor.METRIC_UNIT_NULL, + }, + { + "active_workload", "Active workload", monitor.METRIC_UNIT_NULL, + }, }, } diff --git a/pkg/monitor/dbinit/measurements/service_http_code.go b/pkg/monitor/dbinit/measurements/service_http_code.go index 5b59a29530..4ade4483cf 100644 --- a/pkg/monitor/dbinit/measurements/service_http_code.go +++ b/pkg/monitor/dbinit/measurements/service_http_code.go @@ -25,22 +25,154 @@ var serviceHttpCode = SMeasurement{ }, Metrics: []SMetric{ { - "duration.2xx", "http code 2xxx duration", monitor.METRIC_UNIT_NULL, + Name: "duration_ms_any", + DisplayName: "Accumulated request duration in milliseconds", + Unit: monitor.METRIC_UNIT_MS, }, { - "duration.4xx", "http code 4xxx duration", monitor.METRIC_UNIT_NULL, + Name: "dura_ms_delta_any", + DisplayName: "Accumulated request duration in milliseconds in last interval", + Unit: monitor.METRIC_UNIT_MS, }, { - "duration.5xx", "http code 5xxx duration", monitor.METRIC_UNIT_NULL, + Name: "hit_any", + DisplayName: "Accumulated request count", + Unit: monitor.METRIC_UNIT_NULL, }, { - "hit.2xx", "http code 2xxx hit", monitor.METRIC_UNIT_NULL, + Name: "hit_delta_any", + DisplayName: "Accumulated request count in last interval", + Unit: monitor.METRIC_UNIT_NULL, }, { - "hit.4xx", "http code 4xxx hit", monitor.METRIC_UNIT_NULL, + Name: "delay_ms_any", + DisplayName: "Average request delay in miilliseconds", + Unit: monitor.METRIC_UNIT_MS, }, { - "hit.5xx", "http code 5xxx hit", monitor.METRIC_UNIT_NULL, + Name: "qps_any", + DisplayName: "Averatge request per second", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "duration_ms_2xx", + DisplayName: "Accumulated request duration in milliseconds for 2xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "dura_ms_delta_2xx", + DisplayName: "Accumulated request duration in milliseconds in last interval for 2xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "hit_2xx", + DisplayName: "Accumulated request count for 2xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "hit_delta_2xx", + DisplayName: "Accumulated request count in last interval for 2xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "delay_ms_2xx", + DisplayName: "Average request delay in miilliseconds for 2xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "percent_hit_2xx", + DisplayName: "Request hit weight in percentage for 2xx http code", + Unit: monitor.METRIC_UNIT_PERCENT, + }, + { + Name: "percent_duration_2xx", + DisplayName: "Request duration weight in percentage for 2xx http code", + Unit: monitor.METRIC_UNIT_PERCENT, + }, + { + Name: "qps_2xx", + DisplayName: "Averatge request per second for 2xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "duration_ms_4xx", + DisplayName: "Accumulated request duration in milliseconds for 4xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "dura_ms_delta_4xx", + DisplayName: "Accumulated request duration in milliseconds in last interval for 4xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "hit_4xx", + DisplayName: "Accumulated request count for 4xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "hit_delta_4xx", + DisplayName: "Accumulated request count in last interval for 4xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "delay_ms_4xx", + DisplayName: "Average request delay in miilliseconds for 4xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "percent_hit_4xx", + DisplayName: "Request hit weight in percentage for 4xx http code", + Unit: monitor.METRIC_UNIT_PERCENT, + }, + { + Name: "percent_duration_4xx", + DisplayName: "Request duration weight in percentage for 4xx http code", + Unit: monitor.METRIC_UNIT_PERCENT, + }, + { + Name: "qps_4xx", + DisplayName: "Averatge request per second for 4xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "duration_ms_5xx", + DisplayName: "Accumulated request duration in milliseconds for 5xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "dura_ms_delta_5xx", + DisplayName: "Accumulated request duration in milliseconds in last interval for 5xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "hit_5xx", + DisplayName: "Accumulated request count for 5xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "hit_delta_5xx", + DisplayName: "Accumulated request count in last interval for 5xx http code", + Unit: monitor.METRIC_UNIT_NULL, + }, + { + Name: "delay_ms_5xx", + DisplayName: "Average request delay in miilliseconds for 5xx http code", + Unit: monitor.METRIC_UNIT_MS, + }, + { + Name: "percent_hit_5xx", + DisplayName: "Request hit weight in percentage for 5xx http code", + Unit: monitor.METRIC_UNIT_PERCENT, + }, + { + Name: "percent_duration_5xx", + DisplayName: "Request duration weight in percentage for 5xx http code", + Unit: monitor.METRIC_UNIT_PERCENT, + }, + { + Name: "qps_5xx", + DisplayName: "Averatge request per second for 5xx http code", + Unit: monitor.METRIC_UNIT_NULL, }, }, }