Merge pull request #29 from caesarxuchao/add-metrics

Adding metrics for the core migrator

cmd/migrator/app/migrator.go

@@ -2,10 +2,12 @@ package app
 
 import (
 	"fmt"
+	"net/http"
 	"os"
 
 	migrationclient "github.com/kubernetes-sigs/kube-storage-version-migrator/pkg/clients/clientset"
 	"github.com/kubernetes-sigs/kube-storage-version-migrator/pkg/controller"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
@@ -30,6 +32,9 @@ func NewMigratorCommand() *cobra.Command {
 }
 
 func Run(stopCh <-chan struct{}) error {
+	http.Handle("/metrics", promhttp.Handler())
+	go func() { http.ListenAndServe(":2112", nil) }()
+
 	// creates the in-cluster config
 	config, err := rest.InClusterConfig()
 	if err != nil {

pkg/migrator/core.go

@@ -22,6 +22,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/kubernetes-sigs/kube-storage-version-migrator/pkg/migrator/metrics"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -42,12 +43,12 @@ const (
 type migrator struct {
 	resource    schema.GroupVersionResource
 	client      dynamic.Interface
-	progress    *progressTracker
+	progress    progressInterface
 	concurrency int
 }
 
 // NewMigrator creates a migrator that can migrate a single resource type.
-func NewMigrator(resource schema.GroupVersionResource, client dynamic.Interface, progress *progressTracker) *migrator {
+func NewMigrator(resource schema.GroupVersionResource, client dynamic.Interface, progress progressInterface) *migrator {
 	return &migrator{
 		resource:    resource,
 		client:      client,
@@ -117,6 +118,8 @@ func (m *migrator) Run() error {
 		if err != nil {
 			return err
 		}
+		metrics.Metrics.ObserveObjectsMigrated(len(list.Items), m.resource.String())
+		// TODO: call ObserveObjectsRemaining as well, once https://github.com/kubernetes/kubernetes/pull/75993 is in.
 		if len(token) == 0 {
 			return nil
 		}

pkg/migrator/core_test.go

@@ -21,7 +21,10 @@ import (
 	"fmt"
 	"testing"
 
-	"k8s.io/api/core/v1"
+	"github.com/kubernetes-sigs/kube-storage-version-migrator/pkg/migrator/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	ptype "github.com/prometheus/client_model/go"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -96,6 +99,7 @@ func toUnstructuredListOrDie(l interface{}) *unstructured.UnstructuredList {
 }
 
 func TestMigrateList(t *testing.T) {
+	metrics.Metrics.Reset()
 	podList := newPodList(100)
 	client := fake.NewSimpleDynamicClient(scheme.Scheme, &podList)
 
@@ -198,6 +202,7 @@ func TestMigrateList(t *testing.T) {
 }
 
 func TestMigrateListClusterScoped(t *testing.T) {
+	metrics.Metrics.Reset()
 	nodeList := newNodeList(100)
 	client := fake.NewSimpleDynamicClient(scheme.Scheme, &nodeList)
 
@@ -235,3 +240,75 @@ func TestMigrateListClusterScoped(t *testing.T) {
 		}
 	}
 }
+
+type fakeProgress struct{}
+
+func (f *fakeProgress) load() (string, error) {
+	return "", nil
+}
+
+func (f *fakeProgress) save(string) error {
+	return nil
+}
+
+func TestMetrics(t *testing.T) {
+	metrics.Metrics.Reset()
+	// fake client doesn't support pagination, so we can't test complex behavior.
+	nodeList := newNodeList(100)
+	client := fake.NewSimpleDynamicClient(scheme.Scheme, toUnstructuredListOrDie(nodeList))
+	migrator := NewMigrator(v1.SchemeGroupVersion.WithResource("nodes"), client, &fakeProgress{})
+	migrator.Run()
+	expectCounterCount(t,
+		"storage_migrator_core_migrator_migrated_objects",
+		map[string]string{
+			"resource": "/v1, Resource=nodes",
+		},
+		100,
+	)
+
+}
+
+func labelsMatch(metric *ptype.Metric, labelFilter map[string]string) bool {
+NEXT_FILTER:
+	for k, v := range labelFilter {
+		for _, lp := range metric.GetLabel() {
+			if lp.GetName() == k && lp.GetValue() != v {
+				return false
+			}
+			if lp.GetName() == k && lp.GetValue() == v {
+				continue NEXT_FILTER
+			}
+		}
+		return false
+	}
+	return true
+}
+
+func expectCounterCount(t *testing.T, name string, labelFilter map[string]string, wantCount int) {
+	metrics, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatalf("Failed to gather metrics: %s", err)
+	}
+	counterSum := 0
+	for _, mf := range metrics {
+		if mf.GetName() != name {
+			continue // Ignore other metrics.
+		}
+		for _, metric := range mf.GetMetric() {
+			if !labelsMatch(metric, labelFilter) {
+				continue
+			}
+			counterSum += int(metric.GetCounter().GetValue())
+		}
+	}
+	if wantCount != counterSum {
+		t.Errorf("Wanted count %d, got %d for metric %s with labels %#+v", wantCount, counterSum, name, labelFilter)
+		for _, mf := range metrics {
+			if mf.GetName() == name {
+				for _, metric := range mf.GetMetric() {
+					t.Logf("\tnear match: %s", metric.String())
+				}
+			}
+		}
+	}
+}

pkg/migrator/metrics/metrics.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2017 The Kubernetes 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 (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	namespace = "storage_migrator"
+	subsystem = "core_migrator"
+)
+
+var (
+	// Metrics provides access to all core migrator metrics.
+	Metrics = newCoreMigratorMetrics()
+)
+
+// CoreMigratorMetrics instruments core migrator with prometheus metrics.
+type CoreMigratorMetrics struct {
+	migrated  *prometheus.CounterVec
+	remaining *prometheus.GaugeVec
+}
+
+// newCoreMigratorMetrics create a new CoreMigratorMetrics, configured with default metric names.
+func newCoreMigratorMetrics() *CoreMigratorMetrics {
+	migrated := prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "migrated_objects",
+			Help:      "The number of objects that have been migrated, labeled with the full resource name.",
+		}, []string{"resource"})
+	prometheus.MustRegister(migrated)
+
+	remaining := prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "remaining_objects",
+			Help:      "The number of objects that still require migration, labeled with the full resource name",
+		}, []string{"resource"})
+	prometheus.MustRegister(remaining)
+
+	return &CoreMigratorMetrics{
+		migrated:  migrated,
+		remaining: remaining,
+	}
+}
+
+func (m *CoreMigratorMetrics) Reset() {
+	m.migrated.Reset()
+	m.remaining.Reset()
+}
+
+// ObserveObjectsMigrated adds the number of migrated objects for a resource type..
+func (m *CoreMigratorMetrics) ObserveObjectsMigrated(added int, resource string) {
+	m.migrated.WithLabelValues(resource).Add(float64(added))
+}
+
+// ObserveObjectsRemaining records the number of objects pending-migration for a particular resource type.
+func (m *CoreMigratorMetrics) ObserveObjectsRemaining(count int, resource string) {
+	m.remaining.WithLabelValues(resource).Set(float64(count))
+}

pkg/migrator/progress.go

@@ -22,13 +22,18 @@ import (
 	"k8s.io/client-go/util/retry"
 )
 
+type progressInterface interface {
+	save(continueToken string) error
+	load() (continueToken string, err error)
+}
+
 type progressTracker struct {
 	client migrationclient.StorageVersionMigrationInterface
 	name   string
 }
 
 // NewProgressTracker returns a progress tracker.
-func NewProgressTracker(client migrationclient.StorageVersionMigrationInterface, name string) *progressTracker {
+func NewProgressTracker(client migrationclient.StorageVersionMigrationInterface, name string) progressInterface {
 	return &progressTracker{
 		client: client,
 		name:   name,