Implement host collector as part of system-stats-monitor

Host collector report three things today:
1. Host OS uptime (in seconds)
2. Host kernel version (as a metric label)
3. Host OS version (as a metric label)

Author: Xuewei Zhang

config/system-stats-monitor.json

@@ -15,5 +15,12 @@
 		"includeAllAttachedBlk": true,
 		"lsblkTimeout": "5s"
 	},
+	"host": {
+		"metricsConfigs": {
+			"host/uptime": {
+				"displayName": "host/uptime"
+			}
+		}
+	},
 	"invokeInterval": "60s"
 }

pkg/systemstatsmonitor/disk_collector.go

@@ -46,28 +46,39 @@ type diskCollector struct {
 
 func NewDiskCollectorOrDie(diskConfig *ssmtypes.DiskStatsConfig) *diskCollector {
 	dc := diskCollector{config: diskConfig}
-	dc.keyDevice, _ = tag.NewKey("device")
-
-	dc.mIOTime = metrics.NewInt64Metric(
-		diskConfig.MetricsConfigs["disk/io_time"].DisplayName,
-		"The IO time spent on the disk",
-		"second",
-		view.LastValue(),
-		[]tag.Key{dc.keyDevice})
-
-	dc.mWeightedIO = metrics.NewInt64Metric(
-		diskConfig.MetricsConfigs["disk/weighted_io"].DisplayName,
-		"The weighted IO on the disk",
-		"second",
-		view.LastValue(),
-		[]tag.Key{dc.keyDevice})
-
-	dc.mAvgQueueLen = metrics.NewFloat64Metric(
-		diskConfig.MetricsConfigs["disk/avg_queue_len"].DisplayName,
-		"The average queue length on the disk",
-		"second",
-		view.LastValue(),
-		[]tag.Key{dc.keyDevice})
+
+	var err error
+	dc.keyDevice, err = tag.NewKey("device")
+	if err != nil {
+		glog.Fatalf("Failed to create device tag during initializing disk collector: %v", err)
+	}
+
+	if diskConfig.MetricsConfigs["disk/io_time"].DisplayName != "" {
+		dc.mIOTime = metrics.NewInt64Metric(
+			diskConfig.MetricsConfigs["disk/io_time"].DisplayName,
+			"The IO time spent on the disk",
+			"second",
+			view.LastValue(),
+			[]tag.Key{dc.keyDevice})
+	}
+
+	if diskConfig.MetricsConfigs["disk/weighted_io"].DisplayName != "" {
+		dc.mWeightedIO = metrics.NewInt64Metric(
+			diskConfig.MetricsConfigs["disk/weighted_io"].DisplayName,
+			"The weighted IO on the disk",
+			"second",
+			view.LastValue(),
+			[]tag.Key{dc.keyDevice})
+	}
+
+	if diskConfig.MetricsConfigs["disk/avg_queue_len"].DisplayName != "" {
+		dc.mAvgQueueLen = metrics.NewFloat64Metric(
+			diskConfig.MetricsConfigs["disk/avg_queue_len"].DisplayName,
+			"The average queue length on the disk",
+			"second",
+			view.LastValue(),
+			[]tag.Key{dc.keyDevice})
+	}
 
 	dc.historyIOTime = make(map[string]uint64)
 	dc.historyWeightedIO = make(map[string]uint64)
@@ -88,7 +99,11 @@ func (dc *diskCollector) collect() {
 		blks = append(blks, listAttachedBlockDevices()...)
 	}
 
-	ioCountersStats, _ := disk.IOCounters(blks...)
+	ioCountersStats, err := disk.IOCounters(blks...)
+	if err != nil {
+		glog.Errorf("Failed to retrieve disk IO counters: %v", err)
+		return
+	}
 
 	for deviceName, ioCountersStat := range ioCountersStats {
 		// Calculate average IO queue length since last measurement.
@@ -98,21 +113,26 @@ func (dc *diskCollector) collect() {
 		dc.historyIOTime[deviceName] = ioCountersStat.IoTime
 		dc.historyWeightedIO[deviceName] = ioCountersStat.WeightedIO
 
-		avg_queue_len := float64(0.0)
+		avgQueueLen := float64(0.0)
 		if lastIOTime != ioCountersStat.IoTime {
-			avg_queue_len = float64(ioCountersStat.WeightedIO-lastWeightedIO) / float64(ioCountersStat.IoTime-lastIOTime)
+			avgQueueLen = float64(ioCountersStat.WeightedIO-lastWeightedIO) / float64(ioCountersStat.IoTime-lastIOTime)
 		}
 
 		// Attach label {"device": deviceName} to the metrics.
-		device_ctx, _ := tag.New(context.Background(), tag.Upsert(dc.keyDevice, deviceName))
+		deviceCtx, err := tag.New(context.Background(), tag.Upsert(dc.keyDevice, deviceName))
+		if err != nil {
+			glog.Errorf("Failed to create context with device tag: %v", err)
+			deviceCtx = context.Background()
+		}
+
 		if dc.mIOTime != nil {
-			stats.Record(device_ctx, dc.mIOTime.M(int64(ioCountersStat.IoTime)))
+			stats.Record(deviceCtx, dc.mIOTime.M(int64(ioCountersStat.IoTime)))
 		}
 		if dc.mWeightedIO != nil {
-			stats.Record(device_ctx, dc.mWeightedIO.M(int64(ioCountersStat.WeightedIO)))
+			stats.Record(deviceCtx, dc.mWeightedIO.M(int64(ioCountersStat.WeightedIO)))
 		}
 		if dc.mAvgQueueLen != nil {
-			stats.Record(device_ctx, dc.mAvgQueueLen.M(avg_queue_len))
+			stats.Record(deviceCtx, dc.mAvgQueueLen.M(avgQueueLen))
 		}
 	}
 }
@@ -135,8 +155,14 @@ func listRootBlockDevices(timeout time.Duration) []string {
 
 // listAttachedBlockDevices lists all currently attached block devices.
 func listAttachedBlockDevices() []string {
-	partitions, _ := disk.Partitions(false)
 	blks := []string{}
+
+	partitions, err := disk.Partitions(false)
+	if err != nil {
+		glog.Errorf("Failed to retrieve the list of disk partitions: %v", err)
+		return blks
+	}
+
 	for _, partition := range partitions {
 		blks = append(blks, partition.Device)
 	}

pkg/systemstatsmonitor/host_collector.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2019 The Kubernetes Authors All rights reserved.
+
+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 systemstatsmonitor
+
+import (
+	"context"
+
+	"github.com/golang/glog"
+	"github.com/shirou/gopsutil/host"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+
+	ssmtypes "k8s.io/node-problem-detector/pkg/systemstatsmonitor/types"
+	"k8s.io/node-problem-detector/pkg/util"
+	"k8s.io/node-problem-detector/pkg/util/metrics"
+)
+
+type hostCollector struct {
+	tags   []tag.Mutator
+	uptime *stats.Int64Measure
+}
+
+func NewHostCollectorOrDie(hostConfig *ssmtypes.HostStatsConfig) *hostCollector {
+	hc := hostCollector{}
+
+	keyKernelVersion, err := tag.NewKey("kernel_version")
+	if err != nil {
+		glog.Fatalf("Failed to create kernel_version tag during initializing host collector: %v", err)
+	}
+	kernelVersion, err := host.KernelVersion()
+	if err != nil {
+		glog.Fatalf("Failed to retrieve kernel version: %v", err)
+	}
+	hc.tags = append(hc.tags, tag.Upsert(keyKernelVersion, kernelVersion))
+
+	keyOSVersion, err := tag.NewKey("os_version")
+	if err != nil {
+		glog.Fatalf("Failed to create os_version tag during initializing host collector: %v", err)
+	}
+	osVersion, err := util.GetOSVersion()
+	if err != nil {
+		glog.Fatalf("Failed to retrieve OS version: %v", err)
+	}
+	hc.tags = append(hc.tags, tag.Upsert(keyOSVersion, osVersion))
+
+	if hostConfig.MetricsConfigs["host/uptime"].DisplayName != "" {
+		hc.uptime = metrics.NewInt64Metric(
+			hostConfig.MetricsConfigs["host/uptime"].DisplayName,
+			"The uptime of the operating system",
+			"second",
+			view.LastValue(),
+			[]tag.Key{keyKernelVersion, keyOSVersion})
+	}
+
+	return &hc
+}
+
+func (hc *hostCollector) collect() {
+	if hc == nil {
+		return
+	}
+
+	uptime, err := host.Uptime()
+	if err != nil {
+		glog.Errorf("Failed to retrieve uptime of the host: %v", err)
+		return
+	}
+
+	if hc.uptime != nil {
+		err := stats.RecordWithTags(context.Background(), hc.tags, hc.uptime.M(int64(uptime)))
+		if err != nil {
+			glog.Errorf("Failed to record current uptime (%d seconds) of the host: %v", uptime, err)
+		}
+	}
+}

pkg/systemstatsmonitor/system_stats_monitor.go

@@ -40,6 +40,7 @@ func init() {
 type systemStatsMonitor struct {
 	config        ssmtypes.SystemStatsConfig
 	diskCollector *diskCollector
+	hostCollector *hostCollector
 	tomb          *tomb.Tomb
 }
 
@@ -69,10 +70,12 @@ func NewSystemStatsMonitorOrDie(configPath string) types.Monitor {
 		glog.Fatalf("Failed to validate configuration %+v: %v", ssm.config, err)
 	}
 
-	// Initialize diskCollector if needed.
 	if len(ssm.config.DiskConfig.MetricsConfigs) > 0 {
 		ssm.diskCollector = NewDiskCollectorOrDie(&ssm.config.DiskConfig)
 	}
+	if len(ssm.config.HostConfig.MetricsConfigs) > 0 {
+		ssm.hostCollector = NewHostCollectorOrDie(&ssm.config.HostConfig)
+	}
 	return &ssm
 }
 
@@ -94,12 +97,14 @@ func (ssm *systemStatsMonitor) monitorLoop() {
 		return
 	default:
 		ssm.diskCollector.collect()
+		ssm.hostCollector.collect()
 	}
 
 	for {
 		select {
 		case <-runTicker.C:
 			ssm.diskCollector.collect()
+			ssm.hostCollector.collect()
 		case <-ssm.tomb.Stopping():
 			glog.Infof("System stats monitor stopped")
 			return

pkg/systemstatsmonitor/types/config.go

@@ -38,8 +38,13 @@ type DiskStatsConfig struct {
 	LsblkTimeout          time.Duration           `json:"-"`
 }
 
+type HostStatsConfig struct {
+	MetricsConfigs map[string]MetricConfig `json:"metricsConfigs"`
+}
+
 type SystemStatsConfig struct {
 	DiskConfig           DiskStatsConfig `json:"disk"`
+	HostConfig           HostStatsConfig `json:"host"`
 	InvokeIntervalString string          `json:"invokeInterval"`
 	InvokeInterval       time.Duration   `json:"-"`
 }

pkg/util/helpers.go

@@ -20,9 +20,13 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/cobaugh/osrelease"
+
 	"k8s.io/node-problem-detector/pkg/types"
 )
 
+var osReleasePath = "/etc/os-release"
+
 // GenerateConditionChangeEvent generates an event for condition change.
 func GenerateConditionChangeEvent(t string, status types.ConditionStatus, reason string, timestamp time.Time) types.Event {
 	return types.Event{
@@ -70,3 +74,34 @@ func GetStartTime(now time.Time, uptimeDuration time.Duration, lookbackStr strin
 
 	return startTime, nil
 }
+
+// GetOSVersion retrieves the version of the current operating system.
+// For example: "cos 77-12293.0.0", "ubuntu 16.04.6 LTS (Xenial Xerus)".
+func GetOSVersion() (string, error) {
+	osReleaseMap, err := osrelease.ReadFile(osReleasePath)
+	if err != nil {
+		return "", err
+	}
+	switch osReleaseMap["ID"] {
+	case "cos":
+		return getCOSVersion(osReleaseMap), nil
+	case "debian":
+		return getDebianVersion(osReleaseMap), nil
+	case "ubuntu":
+		return getDebianVersion(osReleaseMap), nil
+	default:
+		return "", fmt.Errorf("Unsupported ID in /etc/os-release: %q", osReleaseMap["ID"])
+	}
+}
+
+func getCOSVersion(osReleaseMap map[string]string) string {
+	// /etc/os-release syntax for COS is defined here:
+	// https://chromium.git.corp.google.com/chromiumos/docs/+/8edec95a297edfd8f1290f0f03a8aa35795b516b/os_config.md
+	return fmt.Sprintf("%s %s-%s", osReleaseMap["ID"], osReleaseMap["VERSION"], osReleaseMap["BUILD_ID"])
+}
+
+func getDebianVersion(osReleaseMap map[string]string) string {
+	// /etc/os-release syntax for Debian is defined here:
+	// https://manpages.debian.org/testing/systemd/os-release.5.en.html
+	return fmt.Sprintf("%s %s", osReleaseMap["ID"], osReleaseMap["VERSION"])
+}

pkg/util/helpers_test.go

@@ -135,3 +135,65 @@ func TestGetStartTime(t *testing.T) {
 		})
 	}
 }
+
+func TestGetOSVersion(t *testing.T) {
+	testCases := []struct {
+		name              string
+		fakeOSReleasePath string
+		expectedOSVersion string
+		expectErr         bool
+	}{
+		{
+			name:              "COS",
+			fakeOSReleasePath: "testdata/os-release-cos",
+			expectedOSVersion: "cos 77-12293.0.0",
+			expectErr:         false,
+		},
+		{
+			name:              "Debian",
+			fakeOSReleasePath: "testdata/os-release-debian",
+			expectedOSVersion: "debian 9 (stretch)",
+			expectErr:         false,
+		},
+		{
+			name:              "Ubuntu",
+			fakeOSReleasePath: "testdata/os-release-ubuntu",
+			expectedOSVersion: "ubuntu 16.04.6 LTS (Xenial Xerus)",
+			expectErr:         false,
+		},
+		{
+			name:              "Unknown",
+			fakeOSReleasePath: "testdata/os-release-unknown",
+			expectedOSVersion: "",
+			expectErr:         true,
+		},
+		{
+			name:              "Empty",
+			fakeOSReleasePath: "testdata/os-release-empty",
+			expectedOSVersion: "",
+			expectErr:         true,
+		},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			originalOSReleasePath := osReleasePath
+			defer func() {
+				osReleasePath = originalOSReleasePath
+			}()
+
+			osReleasePath = test.fakeOSReleasePath
+			osVersion, err := GetOSVersion()
+
+			if test.expectErr && err == nil {
+				t.Errorf("Expect to get error, but got no returned error.")
+			}
+			if !test.expectErr && err != nil {
+				t.Errorf("Expect to get no error, but got returned error: %v", err)
+			}
+			if !test.expectErr && osVersion != test.expectedOSVersion {
+				t.Errorf("Wanted: %+v. \nGot: %+v", test.expectedOSVersion, osVersion)
+			}
+		})
+	}
+}