scottk/kubernetes
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/kubernetes.git synced 2025-11-04 04:08:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate custom collector for kubelet

Browse Source
This commit is contained in:
RainbowMango
2019-11-06 20:38:36 +08:00
parent 1016b8b58a
commit ee4394a306
9 changed files with 100 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
pkg/kubelet/metrics/collectors/log_metrics.go
View File

@@ -17,14 +17,13 @@ limitations under the License.
package collectors
import (
"github.com/prometheus/client_golang/prometheus"
"k8s.io/component-base/metrics"
"k8s.io/klog"
statsapi "k8s.io/kubernetes/pkg/kubelet/apis/stats/v1alpha1"
)
var (
descLogSize = prometheus.NewDesc(
descLogSize = metrics.NewDesc(
"kubelet_container_log_filesystem_used_bytes",
"Bytes used by the container's logs on the filesystem.",
[]string{
@@ -33,28 +32,35 @@ var (
"pod",
"container",
}, nil,
metrics.ALPHA,
"",
)
)
type logMetricsCollector struct {
metrics.BaseStableCollector
podStats func() ([]statsapi.PodStats, error)
}
// NewLogMetricsCollector implements the prometheus.Collector interface and
// Check if logMetricsCollector implements necessary interface
var _ metrics.StableCollector = &logMetricsCollector{}
// NewLogMetricsCollector implements the metrics.StableCollector interface and
// exposes metrics about container's log volume size.
func NewLogMetricsCollector(podStats func() ([]statsapi.PodStats, error)) prometheus.Collector {
func NewLogMetricsCollector(podStats func() ([]statsapi.PodStats, error)) metrics.StableCollector {
return &logMetricsCollector{
podStats: podStats,
}
}
// Describe implements the prometheus.Collector interface.
func (c *logMetricsCollector) Describe(ch chan<- *prometheus.Desc) {
// DescribeWithStability implements the metrics.StableCollector interface.
func (c *logMetricsCollector) DescribeWithStability(ch chan<- *metrics.Desc) {
ch <- descLogSize
}
// Collect implements the prometheus.Collector interface.
func (c *logMetricsCollector) Collect(ch chan<- prometheus.Metric) {
// CollectWithStability implements the metrics.StableCollector interface.
func (c *logMetricsCollector) CollectWithStability(ch chan<- metrics.Metric) {
podStats, err := c.podStats()
if err != nil {
klog.Errorf("failed to get pod stats: %v", err)
@@ -64,9 +70,9 @@ func (c *logMetricsCollector) Collect(ch chan<- prometheus.Metric) {
for _, ps := range podStats {
for _, c := range ps.Containers {
if c.Logs != nil && c.Logs.UsedBytes != nil {
ch <- prometheus.MustNewConstMetric(
ch <- metrics.NewLazyConstMetric(
descLogSize,
prometheus.GaugeValue,
metrics.GaugeValue,
float64(*c.Logs.UsedBytes),
ps.PodRef.UID,
ps.PodRef.Namespace,

12
pkg/kubelet/metrics/collectors/log_metrics_test.go
View File

@@ -25,18 +25,24 @@ import (
)
func TestNoMetricsCollected(t *testing.T) {
// Refresh Desc to share with different registry
descLogSize = descLogSize.GetRawDesc()
collector := &logMetricsCollector{
podStats: func() ([]statsapi.PodStats, error) {
return []statsapi.PodStats{}, nil
},
}
if err := testutil.CollectAndCompare(collector, strings.NewReader(""), ""); err != nil {
if err := testutil.CustomCollectAndCompare(collector, strings.NewReader(""), ""); err != nil {
t.Fatal(err)
}
}
func TestMetricsCollected(t *testing.T) {
// Refresh Desc to share with different registry
descLogSize = descLogSize.GetRawDesc()
size := uint64(18)
collector := &logMetricsCollector{
podStats: func() ([]statsapi.PodStats, error) {
@@ -60,8 +66,8 @@ func TestMetricsCollected(t *testing.T) {
},
}
err := testutil.CollectAndCompare(collector, strings.NewReader(`
# HELP kubelet_container_log_filesystem_used_bytes Bytes used by the container's logs on the filesystem.
err := testutil.CustomCollectAndCompare(collector, strings.NewReader(`
# HELP kubelet_container_log_filesystem_used_bytes [ALPHA] Bytes used by the container's logs on the filesystem.
# TYPE kubelet_container_log_filesystem_used_bytes gauge
kubelet_container_log_filesystem_used_bytes{container="containerName1",namespace="some-namespace",pod="podName1",uid="UID_some_id"} 18
`), "kubelet_container_log_filesystem_used_bytes")

62
pkg/kubelet/metrics/collectors/volume_stats.go
View File

@@ -17,58 +17,68 @@ limitations under the License.
package collectors
import (
"github.com/prometheus/client_golang/prometheus"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/klog"
"k8s.io/component-base/metrics"
stats "k8s.io/kubernetes/pkg/kubelet/apis/stats/v1alpha1"
"k8s.io/kubernetes/pkg/kubelet/metrics"
kubeletmetrics "k8s.io/kubernetes/pkg/kubelet/metrics"
serverstats "k8s.io/kubernetes/pkg/kubelet/server/stats"
)
var (
volumeStatsCapacityBytesDesc = prometheus.NewDesc(
prometheus.BuildFQName("", metrics.KubeletSubsystem, metrics.VolumeStatsCapacityBytesKey),
volumeStatsCapacityBytesDesc = metrics.NewDesc(
metrics.BuildFQName("", kubeletmetrics.KubeletSubsystem, kubeletmetrics.VolumeStatsCapacityBytesKey),
"Capacity in bytes of the volume",
[]string{"namespace", "persistentvolumeclaim"}, nil,
metrics.ALPHA, "",
)
volumeStatsAvailableBytesDesc = prometheus.NewDesc(
prometheus.BuildFQName("", metrics.KubeletSubsystem, metrics.VolumeStatsAvailableBytesKey),
volumeStatsAvailableBytesDesc = metrics.NewDesc(
metrics.BuildFQName("", kubeletmetrics.KubeletSubsystem, kubeletmetrics.VolumeStatsAvailableBytesKey),
"Number of available bytes in the volume",
[]string{"namespace", "persistentvolumeclaim"}, nil,
metrics.ALPHA, "",
)
volumeStatsUsedBytesDesc = prometheus.NewDesc(
prometheus.BuildFQName("", metrics.KubeletSubsystem, metrics.VolumeStatsUsedBytesKey),
volumeStatsUsedBytesDesc = metrics.NewDesc(
metrics.BuildFQName("", kubeletmetrics.KubeletSubsystem, kubeletmetrics.VolumeStatsUsedBytesKey),
"Number of used bytes in the volume",
[]string{"namespace", "persistentvolumeclaim"}, nil,
metrics.ALPHA, "",
)
volumeStatsInodesDesc = prometheus.NewDesc(
prometheus.BuildFQName("", metrics.KubeletSubsystem, metrics.VolumeStatsInodesKey),
volumeStatsInodesDesc = metrics.NewDesc(
metrics.BuildFQName("", kubeletmetrics.KubeletSubsystem, kubeletmetrics.VolumeStatsInodesKey),
"Maximum number of inodes in the volume",
[]string{"namespace", "persistentvolumeclaim"}, nil,
metrics.ALPHA, "",
)
volumeStatsInodesFreeDesc = prometheus.NewDesc(
prometheus.BuildFQName("", metrics.KubeletSubsystem, metrics.VolumeStatsInodesFreeKey),
volumeStatsInodesFreeDesc = metrics.NewDesc(
metrics.BuildFQName("", kubeletmetrics.KubeletSubsystem, kubeletmetrics.VolumeStatsInodesFreeKey),
"Number of free inodes in the volume",
[]string{"namespace", "persistentvolumeclaim"}, nil,
metrics.ALPHA, "",
)
volumeStatsInodesUsedDesc = prometheus.NewDesc(
prometheus.BuildFQName("", metrics.KubeletSubsystem, metrics.VolumeStatsInodesUsedKey),
volumeStatsInodesUsedDesc = metrics.NewDesc(
metrics.BuildFQName("", kubeletmetrics.KubeletSubsystem, kubeletmetrics.VolumeStatsInodesUsedKey),
"Number of used inodes in the volume",
[]string{"namespace", "persistentvolumeclaim"}, nil,
metrics.ALPHA, "",
)
)
type volumeStatsCollector struct {
metrics.BaseStableCollector
statsProvider serverstats.Provider
}
// NewVolumeStatsCollector creates a volume stats prometheus collector.
func NewVolumeStatsCollector(statsProvider serverstats.Provider) prometheus.Collector {
// Check if volumeStatsCollector implements necessary interface
var _ metrics.StableCollector = &volumeStatsCollector{}
// NewVolumeStatsCollector creates a volume stats metrics.StableCollector.
func NewVolumeStatsCollector(statsProvider serverstats.Provider) metrics.StableCollector {
return &volumeStatsCollector{statsProvider: statsProvider}
}
// Describe implements the prometheus.Collector interface.
func (collector *volumeStatsCollector) Describe(ch chan<- *prometheus.Desc) {
// DescribeWithStability implements the metrics.StableCollector interface.
func (collector *volumeStatsCollector) DescribeWithStability(ch chan<- *metrics.Desc) {
ch <- volumeStatsCapacityBytesDesc
ch <- volumeStatsAvailableBytesDesc
ch <- volumeStatsUsedBytesDesc
@@ -77,20 +87,16 @@ func (collector *volumeStatsCollector) Describe(ch chan<- *prometheus.Desc) {
ch <- volumeStatsInodesUsedDesc
}
// Collect implements the prometheus.Collector interface.
func (collector *volumeStatsCollector) Collect(ch chan<- prometheus.Metric) {
// CollectWithStability implements the metrics.StableCollector interface.
func (collector *volumeStatsCollector) CollectWithStability(ch chan<- metrics.Metric) {
podStats, err := collector.statsProvider.ListPodStats()
if err != nil {
return
}
addGauge := func(desc *prometheus.Desc, pvcRef *stats.PVCReference, v float64, lv ...string) {
addGauge := func(desc *metrics.Desc, pvcRef *stats.PVCReference, v float64, lv ...string) {
lv = append([]string{pvcRef.Namespace, pvcRef.Name}, lv...)
metric, err := prometheus.NewConstMetric(desc, prometheus.GaugeValue, v, lv...)
if err != nil {
klog.Warningf("Failed to generate metric: %v", err)
return
}
ch <- metric
ch <- metrics.NewLazyConstMetric(desc, metrics.GaugeValue, v, lv...)
}
allPVCs := sets.String{}
for _, podStat := range podStats {

14
pkg/kubelet/metrics/collectors/volume_stats_test.go
View File

@@ -34,17 +34,17 @@ func TestVolumeStatsCollector(t *testing.T) {
// Fixed metadata on type and help text. We prepend this to every expected
// output so we only have to modify a single place when doing adjustments.
const metadata = `
# HELP kubelet_volume_stats_available_bytes Number of available bytes in the volume
# HELP kubelet_volume_stats_available_bytes [ALPHA] Number of available bytes in the volume
# TYPE kubelet_volume_stats_available_bytes gauge
# HELP kubelet_volume_stats_capacity_bytes Capacity in bytes of the volume
# HELP kubelet_volume_stats_capacity_bytes [ALPHA] Capacity in bytes of the volume
# TYPE kubelet_volume_stats_capacity_bytes gauge
# HELP kubelet_volume_stats_inodes Maximum number of inodes in the volume
# HELP kubelet_volume_stats_inodes [ALPHA] Maximum number of inodes in the volume
# TYPE kubelet_volume_stats_inodes gauge
# HELP kubelet_volume_stats_inodes_free Number of free inodes in the volume
# HELP kubelet_volume_stats_inodes_free [ALPHA] Number of free inodes in the volume
# TYPE kubelet_volume_stats_inodes_free gauge
# HELP kubelet_volume_stats_inodes_used Number of used inodes in the volume
# HELP kubelet_volume_stats_inodes_used [ALPHA] Number of used inodes in the volume
# TYPE kubelet_volume_stats_inodes_used gauge
# HELP kubelet_volume_stats_used_bytes Number of used bytes in the volume
# HELP kubelet_volume_stats_used_bytes [ALPHA] Number of used bytes in the volume
# TYPE kubelet_volume_stats_used_bytes gauge
`
@@ -132,7 +132,7 @@ func TestVolumeStatsCollector(t *testing.T) {
mockStatsProvider := new(statstest.StatsProvider)
mockStatsProvider.On("ListPodStats").Return(podStats, nil)
mockStatsProvider.On("ListPodStatsAndUpdateCPUNanoCoreUsage").Return(podStats, nil)
if err := testutil.CollectAndCompare(&volumeStatsCollector{statsProvider: mockStatsProvider}, strings.NewReader(want), metrics...); err != nil {
if err := testutil.CustomCollectAndCompare(&volumeStatsCollector{statsProvider: mockStatsProvider}, strings.NewReader(want), metrics...); err != nil {
t.Errorf("unexpected collecting result:\n%s", err)
}
}

4
pkg/kubelet/metrics/metrics.go
View File

@@ -499,7 +499,7 @@ var (
var registerMetrics sync.Once
// Register registers all metrics.
func Register(containerCache kubecontainer.RuntimeCache, collectors ...metrics.Collector) {
func Register(containerCache kubecontainer.RuntimeCache, collectors ...metrics.StableCollector) {
// Register the metrics.
registerMetrics.Do(func() {
legacyregistry.MustRegister(NodeName)
@@ -540,7 +540,7 @@ func Register(containerCache kubecontainer.RuntimeCache, collectors ...metrics.C
legacyregistry.MustRegister(ConfigError)
}
for _, collector := range collectors {
legacyregistry.RawMustRegister(collector)
legacyregistry.CustomMustRegister(collector)
}
})
}

30
pkg/kubelet/pluginmanager/metrics/metrics.go
View File

@@ -19,8 +19,8 @@ package metrics
import (
"sync"
"github.com/prometheus/client_golang/prometheus"
"k8s.io/klog"
"k8s.io/component-base/metrics"
"k8s.io/component-base/metrics/legacyregistry"
"k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache"
)
@@ -32,11 +32,13 @@ const (
var (
registerMetrics sync.Once
totalPluginsDesc = prometheus.NewDesc(
totalPluginsDesc = metrics.NewDesc(
pluginManagerTotalPlugins,
"Number of plugins in Plugin Manager",
[]string{"socket_path", "state"},
nil,
metrics.ALPHA,
"",
)
)
@@ -55,35 +57,33 @@ func (pc pluginCount) add(state, pluginName string) {
// Register registers Plugin Manager metrics.
func Register(asw cache.ActualStateOfWorld, dsw cache.DesiredStateOfWorld) {
registerMetrics.Do(func() {
prometheus.MustRegister(&totalPluginsCollector{asw, dsw})
legacyregistry.CustomMustRegister(&totalPluginsCollector{asw: asw, dsw: dsw})
})
}
type totalPluginsCollector struct {
metrics.BaseStableCollector
asw cache.ActualStateOfWorld
dsw cache.DesiredStateOfWorld
}
var _ prometheus.Collector = &totalPluginsCollector{}
var _ metrics.StableCollector = &totalPluginsCollector{}
// Describe implements the prometheus.Collector interface.
func (c *totalPluginsCollector) Describe(ch chan<- *prometheus.Desc) {
// DescribeWithStability implements the metrics.StableCollector interface.
func (c *totalPluginsCollector) DescribeWithStability(ch chan<- *metrics.Desc) {
ch <- totalPluginsDesc
}
// Collect implements the prometheus.Collector interface.
func (c *totalPluginsCollector) Collect(ch chan<- prometheus.Metric) {
// CollectWithStability implements the metrics.StableCollector interface.
func (c *totalPluginsCollector) CollectWithStability(ch chan<- metrics.Metric) {
for stateName, pluginCount := range c.getPluginCount() {
for socketPath, count := range pluginCount {
metric, err := prometheus.NewConstMetric(totalPluginsDesc,
prometheus.GaugeValue,
ch <- metrics.NewLazyConstMetric(totalPluginsDesc,
metrics.GaugeValue,
float64(count),
socketPath,
stateName)
if err != nil {
klog.Warningf("Failed to create metric : %v", err)
}
ch <- metric
}
}
}

2
pkg/kubelet/pluginmanager/metrics/metrics_test.go
View File

@@ -42,7 +42,7 @@ func TestMetricCollection(t *testing.T) {
t.Fatalf("AddOrUpdatePlugin failed. Expected: <no error> Actual: <%v>", err)
}
metricCollector := &totalPluginsCollector{asw, dsw}
metricCollector := &totalPluginsCollector{asw: asw, dsw: dsw}
// Check if getPluginCount returns correct data
count := metricCollector.getPluginCount()

30
pkg/kubelet/volumemanager/metrics/metrics.go
View File

@@ -17,11 +17,10 @@ limitations under the License.
package metrics
import (
"k8s.io/component-base/metrics/legacyregistry"
"sync"
"github.com/prometheus/client_golang/prometheus"
"k8s.io/klog"
"k8s.io/component-base/metrics"
"k8s.io/component-base/metrics/legacyregistry"
"k8s.io/kubernetes/pkg/kubelet/volumemanager/cache"
"k8s.io/kubernetes/pkg/volume"
volumeutil "k8s.io/kubernetes/pkg/volume/util"
@@ -37,11 +36,12 @@ const (
var (
registerMetrics sync.Once
totalVolumesDesc = prometheus.NewDesc(
totalVolumesDesc = metrics.NewDesc(
volumeManagerTotalVolumes,
"Number of volumes in Volume Manager",
[]string{"plugin_name", "state"},
nil,
metrics.ALPHA, "",
)
)
@@ -60,36 +60,34 @@ func (v volumeCount) add(state, plugin string) {
// Register registers Volume Manager metrics.
func Register(asw cache.ActualStateOfWorld, dsw cache.DesiredStateOfWorld, pluginMgr *volume.VolumePluginMgr) {
registerMetrics.Do(func() {
legacyregistry.RawMustRegister(&totalVolumesCollector{asw, dsw, pluginMgr})
legacyregistry.CustomMustRegister(&totalVolumesCollector{asw: asw, dsw: dsw, pluginMgr: pluginMgr})
})
}
type totalVolumesCollector struct {
metrics.BaseStableCollector
asw cache.ActualStateOfWorld
dsw cache.DesiredStateOfWorld
pluginMgr *volume.VolumePluginMgr
}
var _ prometheus.Collector = &totalVolumesCollector{}
var _ metrics.StableCollector = &totalVolumesCollector{}
// Describe implements the prometheus.Collector interface.
func (c *totalVolumesCollector) Describe(ch chan<- *prometheus.Desc) {
// DescribeWithStability implements the metrics.StableCollector interface.
func (c *totalVolumesCollector) DescribeWithStability(ch chan<- *metrics.Desc) {
ch <- totalVolumesDesc
}
// Collect implements the prometheus.Collector interface.
func (c *totalVolumesCollector) Collect(ch chan<- prometheus.Metric) {
// CollectWithStability implements the metrics.StableCollector interface.
func (c *totalVolumesCollector) CollectWithStability(ch chan<- metrics.Metric) {
for stateName, pluginCount := range c.getVolumeCount() {
for pluginName, count := range pluginCount {
metric, err := prometheus.NewConstMetric(totalVolumesDesc,
prometheus.GaugeValue,
ch <- metrics.NewLazyConstMetric(totalVolumesDesc,
metrics.GaugeValue,
float64(count),
pluginName,
stateName)
if err != nil {
klog.Warningf("Failed to create metric : %v", err)
}
ch <- metric
}
}
}

2
pkg/kubelet/volumemanager/metrics/metrics_test.go
View File

@@ -83,7 +83,7 @@ func TestMetricCollection(t *testing.T) {
t.Fatalf("AddPodToVolume failed. Expected: <no error> Actual: <%v>", err)
}
metricCollector := &totalVolumesCollector{asw, dsw, volumePluginMgr}
metricCollector := &totalVolumesCollector{asw: asw, dsw: dsw, pluginMgr: volumePluginMgr}
// Check if getVolumeCount returns correct data
count := metricCollector.getVolumeCount()