feat: add resctrl qos collector

Signed-off-by: Rouzip <1226015390@qq.com>

pkg/koordlet/metriccache/metric_resources.go

@@ -72,4 +72,7 @@ var (
 	HostAppCPUUsageMetric                 = defaultMetricFactory.New(HostAppCPUUsage).withPropertySchema(MetricPropertyHostAppName)
 	HostAppMemoryUsageMetric              = defaultMetricFactory.New(HostAppMemoryUsage).withPropertySchema(MetricPropertyHostAppName)
 	HostAppMemoryUsageWithPageCacheMetric = defaultMetricFactory.New(HostAppMemoryWithPageCacheUsage).withPropertySchema(MetricPropertyHostAppName)
+
+	// Resctrl
+	QosResctrl = defaultMetricFactory.New(ResctrlQos).withPropertySchema(MetricPropertyNodeQos, MetricPropertyResctrlType, MetricPropertyResctrlCacheId, MetricPropertyResctrlMbType)
 )

pkg/koordlet/metriccache/metric_types.go

@@ -18,6 +18,7 @@ package metriccache
 
 import (
 	"fmt"
+	"strconv"
 	"time"
 )
 
@@ -73,6 +74,9 @@ const (
 	// CPI
 	ContainerMetricCPI MetricKind = "container_cpi"
 
+	// Resctrl
+	ResctrlQos MetricKind = "qos_resctrl_resource"
+
 	// PSI
 	ContainerMetricPSI                 MetricKind = "container_psi"
 	ContainerMetricPSICPUFullSupported MetricKind = "container_psi_cpu_full_supported"
@@ -102,6 +106,12 @@ const (
 
 	MetricPropertyCPIResource MetricProperty = "cpi_resource"
 
+	MetricPropertyNodeQos MetricProperty = "node_qos"
+
+	MetricPropertyResctrlType    MetricProperty = "resctrl_type"
+	MetricPropertyResctrlCacheId MetricProperty = "cache_id"
+	MetricPropertyResctrlMbType  MetricProperty = "resctrl_mb_type"
+
 	MetricPropertyPSIResource  MetricProperty = "psi_resource"
 	MetricPropertyPSIPrecision MetricProperty = "psi_precision"
 	MetricPropertyPSIDegree    MetricProperty = "psi_degree"
@@ -141,6 +151,7 @@ var MetricPropertiesFunc = struct {
 	GPU                 func(string, string) map[MetricProperty]string
 	PSICPUFullSupported func(string, string) map[MetricProperty]string
 	ContainerCPI        func(string, string, string) map[MetricProperty]string
+	QosResctrl          func(string, int, string, string) map[MetricProperty]string
 	PodPSI              func(string, string, string, string) map[MetricProperty]string
 	ContainerPSI        func(string, string, string, string, string) map[MetricProperty]string
 	PodGPU              func(string, string, string) map[MetricProperty]string
@@ -160,6 +171,14 @@ var MetricPropertiesFunc = struct {
 	PSICPUFullSupported: func(podUID, containerID string) map[MetricProperty]string {
 		return map[MetricProperty]string{MetricPropertyPodUID: podUID, MetricPropertyContainerID: containerID}
 	},
+	QosResctrl: func(qos string, cacheid int, resctrlType string, resctrlMbType string) map[MetricProperty]string {
+		return map[MetricProperty]string{
+			MetricPropertyResctrlType:    resctrlType,
+			MetricPropertyResctrlCacheId: strconv.Itoa(cacheid),
+			MetricPropertyResctrlMbType:  resctrlMbType,
+			MetricPropertyNodeQos:        qos,
+		}
+	},
 	ContainerCPI: func(podUID, containerID, cpiResource string) map[MetricProperty]string {
 		return map[MetricProperty]string{MetricPropertyPodUID: podUID, MetricPropertyContainerID: containerID, MetricPropertyCPIResource: cpiResource}
 	},

pkg/koordlet/metrics/external_metrics.go

@@ -35,4 +35,5 @@ func init() {
 	ExternalMustRegister(ResourceSummaryCollectors...)
 	ExternalMustRegister(CPICollectors...)
 	ExternalMustRegister(PSICollectors...)
+	ExternalMustRegister(ResctrlCollectors...)
 }

pkg/koordlet/metrics/resctrl.go

@@ -0,0 +1,55 @@
+/*
+Copyright 2022 The Koordinator Authors.
+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 metrics
+
+import (
+	"strconv"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	ResourceTypeLLC = "llc"
+	ResourceTypeMB  = "mb"
+
+	ResctrlResourceType = "resctrl_resource_type"
+	ResctrlCacheId      = "resctrl_cacheid"
+	ResctrlQos          = "resctrl_qos"
+	ResctrlMbType       = "resctrl_mb_type"
+)
+
+var (
+	QosResctrl = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Subsystem: KoordletSubsystem,
+		Name:      "qos_resctrl",
+		Help:      "qos resctrl collected by koordlet",
+	}, []string{NodeKey, ResctrlResourceType, ResctrlCacheId, ResctrlQos, ResctrlMbType})
+
+	ResctrlCollectors = []prometheus.Collector{
+		QosResctrl,
+	}
+)
+
+func ResetQosResctrl() {
+	QosResctrl.Reset()
+}
+
+func RecordQosResctrl(resourceType string, cacheId int, qos, mbType string, value uint64) {
+	labels := genNodeLabels()
+	labels[ResctrlResourceType] = resourceType
+	labels[ResctrlCacheId] = strconv.Itoa(cacheId)
+	labels[ResctrlQos] = qos
+	labels[ResctrlMbType] = mbType
+	QosResctrl.With(labels).Set(float64(value))
+}

pkg/koordlet/metricsadvisor/collectors/resctrl/resctrl_collector.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2022 The Koordinator Authors.
+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 resctrl
+
+import (
+	"time"
+
+	"go.uber.org/atomic"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
+
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/metriccache"
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/metrics"
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/metricsadvisor/framework"
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/qosmanager/plugins/resctrl"
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/resourceexecutor"
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/statesinformer"
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/util/system"
+)
+
+const (
+	CollectorName = "resctrlCollector"
+)
+
+type resctrlCollector struct {
+	collectInterval      time.Duration
+	started              *atomic.Bool
+	metricCache          metriccache.MetricCache
+	statesInformer       statesinformer.StatesInformer
+	resctrlReader        resourceexecutor.ResctrlReader
+	resctrlCollectorGate bool
+}
+
+func New(opt *framework.Options) framework.Collector {
+	return &resctrlCollector{
+		collectInterval: opt.Config.ResctrlCollectorInterval,
+		statesInformer:  opt.StatesInformer,
+		metricCache:     opt.MetricCache,
+		resctrlReader:   opt.ResctrlReader,
+		started:         atomic.NewBool(false),
+	}
+}
+
+func (r *resctrlCollector) Enabled() bool {
+	return r.resctrlCollectorGate
+}
+
+func (r *resctrlCollector) Setup(c *framework.Context) {}
+
+func (r *resctrlCollector) Started() bool {
+	return r.started.Load()
+}
+
+func (r *resctrlCollector) Run(stopCh <-chan struct{}) {
+	go wait.Until(r.collectQoSResctrlStat, r.collectInterval, stopCh)
+}
+
+func (r *resctrlCollector) collectQoSResctrlStat() {
+	err := system.CheckResctrlSchemataValid()
+	if err != nil {
+		// use resctrl/schemata to check mount state TODO: use another method to check
+		klog.Warningf("cannot find resctrl file system schemata, error: %v", err)
+		return
+	}
+	klog.V(6).Info("start collect QoS resctrl Stat")
+	resctrlMetrics := make([]metriccache.MetricSample, 0)
+	collectTime := time.Now()
+	for _, qos := range []string{
+		resctrl.LSRResctrlGroup,
+		resctrl.LSResctrlGroup,
+		resctrl.BEResctrlGroup} {
+		l3Map, err := r.resctrlReader.ReadResctrlL3Stat(qos)
+		if err != nil {
+			klog.Warningf("collect QoS %s resctrl llc data error: %v", qos, err)
+			continue
+		}
+		for cacheId, value := range l3Map {
+			metrics.RecordQosResctrl(metrics.ResourceTypeLLC, int(cacheId), qos, "", value)
+			llcSample, err := metriccache.QosResctrl.GenerateSample(metriccache.MetricPropertiesFunc.QosResctrl(qos, int(cacheId), metrics.ResourceTypeLLC, ""), collectTime, float64(value))
+			if err != nil {
+				klog.Warningf("generate QoS %s resctrl llc sample error: %v", qos, err)
+			}
+			resctrlMetrics = append(resctrlMetrics, llcSample)
+		}
+		mbMap, err := r.resctrlReader.ReadResctrlMBStat(qos)
+		if err != nil {
+			klog.V(4).Infof("collect QoS %s resctrl mb data error: %v", qos, err)
+		}
+		for cacheId, value := range mbMap {
+			for mbType, mbValue := range value {
+				metrics.RecordQosResctrl(metrics.ResourceTypeMB, int(cacheId), qos, mbType, mbValue)
+				mbSample, err := metriccache.QosResctrl.GenerateSample(metriccache.MetricPropertiesFunc.QosResctrl(qos, int(cacheId), metrics.ResourceTypeMB, mbType), collectTime, float64(mbValue))
+				if err != nil {
+					klog.Warningf("generate QoS %s resctrl mb sample error: %v", qos, err)
+				}
+				resctrlMetrics = append(resctrlMetrics, mbSample)
+			}
+		}
+	}
+
+	// save QoS resctrl data to tsdb
+	r.saveMetric(resctrlMetrics)
+
+	r.started.Store(true)
+	klog.V(6).Infof("collect resctrl data at %s", time.Now())
+}
+
+func (r *resctrlCollector) saveMetric(samples []metriccache.MetricSample) error {
+	if len(samples) == 0 {
+		return nil
+	}
+
+	appender := r.metricCache.Appender()
+	if err := appender.Append(samples); err != nil {
+		klog.ErrorS(err, "Append container metrics error")
+		return err
+	}
+
+	if err := appender.Commit(); err != nil {
+		klog.ErrorS(err, "Commit container metrics failed")
+		return err
+	}
+
+	return nil
+}

pkg/koordlet/metricsadvisor/framework/config.go

@@ -35,7 +35,9 @@ type Config struct {
 	PSICollectorInterval             time.Duration
 	CPICollectorTimeWindow           time.Duration
 	ColdPageCollectorInterval        time.Duration
+	ResctrlCollectorInterval         time.Duration
 	EnablePageCacheCollector         bool
+	EnableResctrlCollector           bool
 }
 
 func NewDefaultConfig() *Config {
@@ -48,7 +50,9 @@ func NewDefaultConfig() *Config {
 		PSICollectorInterval:             10 * time.Second,
 		CPICollectorTimeWindow:           10 * time.Second,
 		ColdPageCollectorInterval:        5 * time.Second,
+		ResctrlCollectorInterval:         1 * time.Second,
 		EnablePageCacheCollector:         false,
+		EnableResctrlCollector:           false,
 	}
 }
 
@@ -62,4 +66,6 @@ func (c *Config) InitFlags(fs *flag.FlagSet) {
 	fs.DurationVar(&c.CPICollectorTimeWindow, "collect-cpi-timewindow", c.CPICollectorTimeWindow, "Collect cpi time window. Non-zero values should contain a corresponding time unit (e.g. 1s, 2m, 3h).")
 	fs.DurationVar(&c.ColdPageCollectorInterval, "coldpage-collector-interval", c.ColdPageCollectorInterval, "Collect cold page interval. Non-zero values should contain a corresponding time unit (e.g. 1s, 2m, 3h).")
 	fs.BoolVar(&c.EnablePageCacheCollector, "enable-pagecache-collector", c.EnablePageCacheCollector, "Enable cache collector of node, pods and containers")
+	fs.BoolVar(&c.EnableResctrlCollector, "enable-resctrl-collector", c.EnableResctrlCollector, "Enable cache collector of node, pods and containers")
+	fs.DurationVar(&c.ResctrlCollectorInterval, "resctrl-collector-interval", c.ResctrlCollectorInterval, "Collect cpi time window. Non-zero values should contain a corresponding time unit (e.g. 1s, 2m, 3h).")
 }

pkg/koordlet/metricsadvisor/framework/config_test.go

@@ -34,6 +34,7 @@ func Test_NewDefaultConfig(t *testing.T) {
 		PSICollectorInterval:             10 * time.Second,
 		CPICollectorTimeWindow:           10 * time.Second,
 		ColdPageCollectorInterval:        5 * time.Second,
+		ResctrlCollectorInterval:         1 * time.Second,
 		EnablePageCacheCollector:         false,
 	}
 	defaultConfig := NewDefaultConfig()
@@ -51,6 +52,7 @@ func Test_InitFlags(t *testing.T) {
 		"--psi-collector-interval=5s",
 		"--collect-cpi-timewindow=15s",
 		"--coldpage-collector-interval=15s",
+		"--resctrl-collector-interval=90s",
 	}
 	fs := flag.NewFlagSet(cmdArgs[0], flag.ExitOnError)
 
@@ -63,6 +65,7 @@ func Test_InitFlags(t *testing.T) {
 		PSICollectorInterval             time.Duration
 		CPICollectorTimeWindow           time.Duration
 		ColdPageCollectorInterval        time.Duration
+		ResctrlCollectorInterval         time.Duration
 	}
 	type args struct {
 		fs *flag.FlagSet
@@ -83,6 +86,7 @@ func Test_InitFlags(t *testing.T) {
 				PSICollectorInterval:             5 * time.Second,
 				CPICollectorTimeWindow:           15 * time.Second,
 				ColdPageCollectorInterval:        15 * time.Second,
+				ResctrlCollectorInterval:         90 * time.Second,
 			},
 			args: args{fs: fs},
 		},
@@ -98,6 +102,7 @@ func Test_InitFlags(t *testing.T) {
 				PSICollectorInterval:             tt.fields.PSICollectorInterval,
 				CPICollectorTimeWindow:           tt.fields.CPICollectorTimeWindow,
 				ColdPageCollectorInterval:        tt.fields.ColdPageCollectorInterval,
+				ResctrlCollectorInterval:         tt.fields.ResctrlCollectorInterval,
 			}
 			c := NewDefaultConfig()
 			c.InitFlags(tt.args.fs)

pkg/koordlet/metricsadvisor/framework/options.go

@@ -27,5 +27,6 @@ type Options struct {
 	StatesInformer statesinformer.StatesInformer
 	MetricCache    metriccache.MetricCache
 	CgroupReader   resourceexecutor.CgroupReader
+	ResctrlReader  resourceexecutor.ResctrlReader
 	PodFilters     map[string]PodFilter
 }

pkg/koordlet/metricsadvisor/metrics_advisor.go

@@ -45,6 +45,7 @@ func NewMetricAdvisor(cfg *framework.Config, statesInformer statesinformer.State
 		MetricCache:    metricCache,
 		CgroupReader:   resourceexecutor.NewCgroupReader(),
 		PodFilters:     podFilters,
+		ResctrlReader:  resourceexecutor.NewResctrlReader(),
 	}
 	ctx := &framework.Context{
 		DeviceCollectors: make(map[string]framework.DeviceCollector, len(devicePlugins)),