diff --git a/pkg/controller/volume/persistentvolume/BUILD b/pkg/controller/volume/persistentvolume/BUILD
index 3873ca2f630..733472f951f 100644
--- a/pkg/controller/volume/persistentvolume/BUILD
+++ b/pkg/controller/volume/persistentvolume/BUILD
@@ -24,6 +24,7 @@ go_library(
         "//pkg/cloudprovider:go_default_library",
         "//pkg/controller:go_default_library",
         "//pkg/controller/volume/events:go_default_library",
+        "//pkg/controller/volume/persistentvolume/metrics:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/util/goroutinemap:go_default_library",
         "//pkg/util/goroutinemap/exponentialbackoff:go_default_library",
@@ -117,6 +118,7 @@ filegroup(
     name = "all-srcs",
     srcs = [
         ":package-srcs",
+        "//pkg/controller/volume/persistentvolume/metrics:all-srcs",
         "//pkg/controller/volume/persistentvolume/options:all-srcs",
     ],
     tags = ["automanaged"],
diff --git a/pkg/controller/volume/persistentvolume/metrics/BUILD b/pkg/controller/volume/persistentvolume/metrics/BUILD
new file mode 100644
index 00000000000..e5a37cbfd42
--- /dev/null
+++ b/pkg/controller/volume/persistentvolume/metrics/BUILD
@@ -0,0 +1,30 @@
+package(default_visibility = ["//visibility:public"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["metrics.go"],
+    importpath = "k8s.io/kubernetes/pkg/controller/volume/persistentvolume/metrics",
+    deps = [
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+)
diff --git a/pkg/controller/volume/persistentvolume/metrics/metrics.go b/pkg/controller/volume/persistentvolume/metrics/metrics.go
new file mode 100644
index 00000000000..5035b1bb9e0
--- /dev/null
+++ b/pkg/controller/volume/persistentvolume/metrics/metrics.go
@@ -0,0 +1,184 @@
+/*
+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 (
+	"sync"
+
+	"k8s.io/api/core/v1"
+
+	"github.com/golang/glog"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	// Subsystem names.
+	pvControllerSubsystem = "pv_collector"
+
+	// Metric names.
+	boundPVKey    = "bound_pv_count"
+	unboundPVKey  = "unbound_pv_count"
+	boundPVCKey   = "bound_pvc_count"
+	unboundPVCKey = "unbound_pvc_count"
+
+	// Label names.
+	namespaceLabel    = "namespace"
+	storageClassLabel = "storage_class"
+)
+
+var registerMetrics sync.Once
+
+// PVLister used to list persistent volumes.
+type PVLister interface {
+	List() []interface{}
+}
+
+// PVCLister used to list persistent volume claims.
+type PVCLister interface {
+	List() []interface{}
+}
+
+// Register all metrics for pv controller.
+func Register(pvLister PVLister, pvcLister PVCLister) {
+	registerMetrics.Do(func() {
+		prometheus.MustRegister(newPVAndPVCCountCollector(pvLister, pvcLister))
+	})
+}
+
+func newPVAndPVCCountCollector(pvLister PVLister, pvcLister PVCLister) *pvAndPVCCountCollector {
+	return &pvAndPVCCountCollector{pvLister, pvcLister}
+}
+
+// Custom collector for current pod and container counts.
+type pvAndPVCCountCollector struct {
+	// Cache for accessing information about PersistentVolumes.
+	pvLister PVLister
+	// Cache for accessing information about PersistentVolumeClaims.
+	pvcLister PVCLister
+}
+
+var (
+	boundPVCountDesc = prometheus.NewDesc(
+		prometheus.BuildFQName("", pvControllerSubsystem, boundPVKey),
+		"Gauge measuring number of persistent volume currently bound",
+		[]string{storageClassLabel}, nil)
+	unboundPVCountDesc = prometheus.NewDesc(
+		prometheus.BuildFQName("", pvControllerSubsystem, unboundPVKey),
+		"Gauge measuring number of persistent volume currently unbound",
+		[]string{storageClassLabel}, nil)
+
+	boundPVCCountDesc = prometheus.NewDesc(
+		prometheus.BuildFQName("", pvControllerSubsystem, boundPVCKey),
+		"Gauge measuring number of persistent volume claim currently bound",
+		[]string{namespaceLabel}, nil)
+	unboundPVCCountDesc = prometheus.NewDesc(
+		prometheus.BuildFQName("", pvControllerSubsystem, unboundPVCKey),
+		"Gauge measuring number of persistent volume claim currently unbound",
+		[]string{namespaceLabel}, nil)
+)
+
+func (collector *pvAndPVCCountCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- boundPVCountDesc
+	ch <- unboundPVCountDesc
+	ch <- boundPVCCountDesc
+	ch <- unboundPVCCountDesc
+}
+
+func (collector *pvAndPVCCountCollector) Collect(ch chan<- prometheus.Metric) {
+	collector.pvCollect(ch)
+	collector.pvcCollect(ch)
+}
+
+func (collector *pvAndPVCCountCollector) pvCollect(ch chan<- prometheus.Metric) {
+	boundNumberByStorageClass := make(map[string]int)
+	unboundNumberByStorageClass := make(map[string]int)
+	for _, pvObj := range collector.pvLister.List() {
+		pv, ok := pvObj.(*v1.PersistentVolume)
+		if !ok {
+			continue
+		}
+		if pv.Status.Phase == v1.VolumeBound {
+			boundNumberByStorageClass[pv.Spec.StorageClassName]++
+		} else {
+			unboundNumberByStorageClass[pv.Spec.StorageClassName]++
+		}
+	}
+	for storageClassName, number := range boundNumberByStorageClass {
+		metric, err := prometheus.NewConstMetric(
+			boundPVCountDesc,
+			prometheus.GaugeValue,
+			float64(number),
+			storageClassName)
+		if err != nil {
+			glog.Warningf("Create bound pv number metric failed: %v", err)
+			continue
+		}
+		ch <- metric
+	}
+	for storageClassName, number := range unboundNumberByStorageClass {
+		metric, err := prometheus.NewConstMetric(
+			unboundPVCountDesc,
+			prometheus.GaugeValue,
+			float64(number),
+			storageClassName)
+		if err != nil {
+			glog.Warningf("Create unbound pv number metric failed: %v", err)
+			continue
+		}
+		ch <- metric
+	}
+}
+
+func (collector *pvAndPVCCountCollector) pvcCollect(ch chan<- prometheus.Metric) {
+	boundNumberByNamespace := make(map[string]int)
+	unboundNumberByNamespace := make(map[string]int)
+	for _, pvcObj := range collector.pvcLister.List() {
+		pvc, ok := pvcObj.(*v1.PersistentVolumeClaim)
+		if !ok {
+			continue
+		}
+		if pvc.Status.Phase == v1.ClaimBound {
+			boundNumberByNamespace[pvc.Namespace]++
+		} else {
+			unboundNumberByNamespace[pvc.Namespace]++
+		}
+	}
+	for namespace, number := range boundNumberByNamespace {
+		metric, err := prometheus.NewConstMetric(
+			boundPVCCountDesc,
+			prometheus.GaugeValue,
+			float64(number),
+			namespace)
+		if err != nil {
+			glog.Warningf("Create bound pvc number metric failed: %v", err)
+			continue
+		}
+		ch <- metric
+	}
+	for namespace, number := range unboundNumberByNamespace {
+		metric, err := prometheus.NewConstMetric(
+			unboundPVCCountDesc,
+			prometheus.GaugeValue,
+			float64(number),
+			namespace)
+		if err != nil {
+			glog.Warningf("Create unbound pvc number metric failed: %v", err)
+			continue
+		}
+		ch <- metric
+	}
+}
diff --git a/pkg/controller/volume/persistentvolume/pv_controller_base.go b/pkg/controller/volume/persistentvolume/pv_controller_base.go
index 8995d2b24b6..1b745ef937d 100644
--- a/pkg/controller/volume/persistentvolume/pv_controller_base.go
+++ b/pkg/controller/volume/persistentvolume/pv_controller_base.go
@@ -40,6 +40,7 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/controller/volume/persistentvolume/metrics"
 	"k8s.io/kubernetes/pkg/util/goroutinemap"
 	vol "k8s.io/kubernetes/pkg/volume"
 
@@ -277,6 +278,8 @@ func (ctrl *PersistentVolumeController) Run(stopCh <-chan struct{}) {
 	go wait.Until(ctrl.volumeWorker, time.Second, stopCh)
 	go wait.Until(ctrl.claimWorker, time.Second, stopCh)
 
+	metrics.Register(ctrl.volumes.store, ctrl.claims)
+
 	<-stopCh
 }
 
diff --git a/test/e2e/framework/pv_util.go b/test/e2e/framework/pv_util.go
index b4576fc02a1..fb5a194ccc9 100644
--- a/test/e2e/framework/pv_util.go
+++ b/test/e2e/framework/pv_util.go
@@ -263,6 +263,11 @@ func createPV(c clientset.Interface, pv *v1.PersistentVolume) (*v1.PersistentVol
 	return pv, nil
 }
 
+// create the PV resource. Fails test on error.
+func CreatePV(c clientset.Interface, pv *v1.PersistentVolume) (*v1.PersistentVolume, error) {
+	return createPV(c, pv)
+}
+
 // create the PVC resource. Fails test on error.
 func CreatePVC(c clientset.Interface, ns string, pvc *v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error) {
 	pvc, err := c.CoreV1().PersistentVolumeClaims(ns).Create(pvc)
diff --git a/test/e2e/storage/BUILD b/test/e2e/storage/BUILD
index dbd358d617c..7f43c80a9f6 100644
--- a/test/e2e/storage/BUILD
+++ b/test/e2e/storage/BUILD
@@ -48,6 +48,7 @@ go_library(
         "//vendor/github.com/ghodss/yaml:go_default_library",
         "//vendor/github.com/onsi/ginkgo:go_default_library",
         "//vendor/github.com/onsi/gomega:go_default_library",
+        "//vendor/github.com/prometheus/common/model:go_default_library",
         "//vendor/google.golang.org/api/googleapi:go_default_library",
         "//vendor/k8s.io/api/apps/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
diff --git a/test/e2e/storage/volume_metrics.go b/test/e2e/storage/volume_metrics.go
index 303df4855f8..34fdf17d3ae 100644
--- a/test/e2e/storage/volume_metrics.go
+++ b/test/e2e/storage/volume_metrics.go
@@ -22,6 +22,7 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/prometheus/common/model"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -165,6 +166,137 @@ var _ = utils.SIGDescribe("[Serial] Volume metrics", func() {
 		framework.Logf("Deleting pod %q/%q", pod.Namespace, pod.Name)
 		framework.ExpectNoError(framework.DeletePodWithWait(f, c, pod))
 	})
+
+	// Test for pv controller metrics, concretely: bound/unbound pv/pvc count.
+	Describe("PVController", func() {
+		const (
+			classKey     = "storage_class"
+			namespaceKey = "namespace"
+
+			boundPVKey    = "pv_collector_bound_pv_count"
+			unboundPVKey  = "pv_collector_unbound_pv_count"
+			boundPVCKey   = "pv_collector_bound_pvc_count"
+			unboundPVCKey = "pv_collector_unbound_pvc_count"
+		)
+
+		var (
+			pv  *v1.PersistentVolume
+			pvc *v1.PersistentVolumeClaim
+
+			className = "bound-unbound-count-test-sc"
+			pvConfig  = framework.PersistentVolumeConfig{
+				PVSource: v1.PersistentVolumeSource{
+					HostPath: &v1.HostPathVolumeSource{Path: "/data"},
+				},
+				NamePrefix:       "pv-test-",
+				StorageClassName: className,
+			}
+			pvcConfig = framework.PersistentVolumeClaimConfig{StorageClassName: &className}
+
+			metrics = []struct {
+				name      string
+				dimension string
+			}{
+				{boundPVKey, classKey},
+				{unboundPVKey, classKey},
+				{boundPVCKey, namespaceKey},
+				{unboundPVCKey, namespaceKey},
+			}
+
+			// Original metric values before we create any PV/PVCs. The length should be 4,
+			// and the elements should be bound pv count, unbound pv count, bound pvc count,
+			// unbound pvc count in turn.
+			// We use these values to calculate relative increment of each test.
+			originMetricValues []map[string]int64
+		)
+
+		// validator used to validate each metric's values, the length of metricValues
+		// should be 4, and the elements should be bound pv count, unbound pv count, bound
+		// pvc count, unbound pvc count in turn.
+		validator := func(metricValues []map[string]int64) {
+			Expect(len(metricValues)).To(Equal(4),
+				"Wrong metric size: %d", len(metricValues))
+
+			controllerMetrics, err := metricsGrabber.GrabFromControllerManager()
+			Expect(err).NotTo(HaveOccurred(), "Error getting c-m metricValues: %v", err)
+
+			for i, metric := range metrics {
+				expectValues := metricValues[i]
+				if expectValues == nil {
+					expectValues = make(map[string]int64)
+				}
+				// We using relative increment value instead of absolute value to reduce unexpected flakes.
+				// Concretely, we expect the difference of the updated values and original values for each
+				// test suit are equal to expectValues.
+				actualValues := calculateRelativeValues(originMetricValues[i],
+					getPVControllerMetrics(controllerMetrics, metric.name, metric.dimension))
+				Expect(actualValues).To(Equal(expectValues),
+					"Wrong pv controller metric %s(%s): wanted %v, got %v",
+					metric.name, metric.dimension, expectValues, actualValues)
+			}
+		}
+
+		BeforeEach(func() {
+			if !metricsGrabber.HasRegisteredMaster() {
+				framework.Skipf("Environment does not support getting controller-manager metrics - skipping")
+			}
+
+			pv = framework.MakePersistentVolume(pvConfig)
+			pvc = framework.MakePersistentVolumeClaim(pvcConfig, ns)
+
+			// Initializes all original metric values.
+			controllerMetrics, err := metricsGrabber.GrabFromControllerManager()
+			Expect(err).NotTo(HaveOccurred(), "Error getting c-m metricValues: %v", err)
+			for _, metric := range metrics {
+				originMetricValues = append(originMetricValues,
+					getPVControllerMetrics(controllerMetrics, metric.name, metric.dimension))
+			}
+		})
+
+		AfterEach(func() {
+			if err := framework.DeletePersistentVolume(c, pv.Name); err != nil {
+				framework.Failf("Error deleting pv: %v", err)
+			}
+			if err := framework.DeletePersistentVolumeClaim(c, pvc.Name, pvc.Namespace); err != nil {
+				framework.Failf("Error deleting pvc: %v", err)
+			}
+
+			// Clear original metric values.
+			originMetricValues = nil
+		})
+
+		It("should create none metrics for pvc controller before creating any PV or PVC", func() {
+			validator([]map[string]int64{nil, nil, nil, nil})
+		})
+
+		It("should create unbound pv count metrics for pvc controller after creating pv only",
+			func() {
+				var err error
+				pv, err = framework.CreatePV(c, pv)
+				Expect(err).NotTo(HaveOccurred(), "Error creating pv: %v", err)
+				waitForPVControllerSync(metricsGrabber, unboundPVKey, classKey)
+				validator([]map[string]int64{nil, {className: 1}, nil, nil})
+			})
+
+		It("should create unbound pvc count metrics for pvc controller after creating pvc only",
+			func() {
+				var err error
+				pvc, err = framework.CreatePVC(c, ns, pvc)
+				Expect(err).NotTo(HaveOccurred(), "Error creating pvc: %v", err)
+				waitForPVControllerSync(metricsGrabber, unboundPVCKey, namespaceKey)
+				validator([]map[string]int64{nil, nil, nil, {ns: 1}})
+			})
+
+		It("should create bound pv/pvc count metrics for pvc controller after creating both pv and pvc",
+			func() {
+				var err error
+				pv, pvc, err = framework.CreatePVPVC(c, pvConfig, pvcConfig, ns, true)
+				Expect(err).NotTo(HaveOccurred(), "Error creating pv pvc: %v", err)
+				waitForPVControllerSync(metricsGrabber, boundPVKey, classKey)
+				validator([]map[string]int64{{className: 1}, nil, {ns: 1}, nil})
+
+			})
+	})
 })
 
 func waitForDetachAndGrabMetrics(oldMetrics map[string]int64, metricsGrabber *metrics.MetricsGrabber) map[string]int64 {
@@ -270,3 +402,54 @@ func findVolumeStatMetric(metricKeyName string, namespace string, pvcName string
 	Expect(errCount).To(Equal(0), "Found invalid samples")
 	return found
 }
+
+// Wait for the count of a pv controller's metric specified by metricName and dimension bigger than zero.
+func waitForPVControllerSync(metricsGrabber *metrics.MetricsGrabber, metricName, dimension string) {
+	backoff := wait.Backoff{
+		Duration: 10 * time.Second,
+		Factor:   1.2,
+		Steps:    21,
+	}
+	verifyMetricFunc := func() (bool, error) {
+		updatedMetrics, err := metricsGrabber.GrabFromControllerManager()
+		if err != nil {
+			framework.Logf("Error fetching controller-manager metrics")
+			return false, err
+		}
+		return len(getPVControllerMetrics(updatedMetrics, metricName, dimension)) > 0, nil
+	}
+	waitErr := wait.ExponentialBackoff(backoff, verifyMetricFunc)
+	Expect(waitErr).NotTo(HaveOccurred(),
+		"Timeout error fetching pv controller metrics : %v", waitErr)
+}
+
+func getPVControllerMetrics(ms metrics.ControllerManagerMetrics, metricName, dimension string) map[string]int64 {
+	result := make(map[string]int64)
+	for method, samples := range ms {
+		if method != metricName {
+			continue
+		}
+		for _, sample := range samples {
+			count := int64(sample.Value)
+			dimensionName := string(sample.Metric[model.LabelName(dimension)])
+			result[dimensionName] = count
+		}
+	}
+	return result
+}
+
+func calculateRelativeValues(originValues, updatedValues map[string]int64) map[string]int64 {
+	relativeValues := make(map[string]int64)
+	for key, value := range updatedValues {
+		relativeValue := value - originValues[key]
+		if relativeValue != 0 {
+			relativeValues[key] = relativeValue
+		}
+	}
+	for key, value := range originValues {
+		if _, exist := updatedValues[key]; !exist && value > 0 {
+			relativeValues[key] = -value
+		}
+	}
+	return relativeValues
+}