Merge pull request #1204 from pmorie/dm-support

devicemapper thin_ls support

container/docker/factory.go

@@ -24,9 +24,11 @@ import (
 
 	"github.com/google/cadvisor/container"
 	"github.com/google/cadvisor/container/libcontainer"
+	"github.com/google/cadvisor/devicemapper"
 	"github.com/google/cadvisor/fs"
 	info "github.com/google/cadvisor/info/v1"
 	"github.com/google/cadvisor/manager/watcher"
+	dockerutil "github.com/google/cadvisor/utils/docker"
 
 	docker "github.com/docker/engine-api/client"
 	"github.com/golang/glog"
@@ -68,7 +70,6 @@ func RootDir() string {
 type storageDriver string
 
 const (
-	// TODO: Add support for devicemapper storage usage.
 	devicemapperStorageDriver storageDriver = "devicemapper"
 	aufsStorageDriver         storageDriver = "aufs"
 	overlayStorageDriver      storageDriver = "overlay"
@@ -92,6 +93,8 @@ type dockerFactory struct {
 	dockerVersion []int
 
 	ignoreMetrics container.MetricSet
+
+	thinPoolWatcher *devicemapper.ThinPoolWatcher
 }
 
 func (self *dockerFactory) String() string {
@@ -118,6 +121,7 @@ func (self *dockerFactory) NewContainerHandler(name string, inHostNamespace bool
 		metadataEnvs,
 		self.dockerVersion,
 		self.ignoreMetrics,
+		self.thinPoolWatcher,
 	)
 	return
 }
@@ -187,7 +191,30 @@ func Register(factory info.MachineInfoFactory, fsInfo fs.FsInfo, ignoreMetrics c
 		return fmt.Errorf("failed to get cgroup subsystems: %v", err)
 	}
 
-	glog.Infof("Registering Docker factory")
+	var (
+		dockerStorageDriver                               = storageDriver(dockerInfo.Driver)
+		thinPoolWatcher     *devicemapper.ThinPoolWatcher = nil
+	)
+
+	if dockerStorageDriver == devicemapperStorageDriver {
+		// If the storage drive is devicemapper, create and start a
+		// ThinPoolWatcher to monitor the size of container CoW layers with
+		// thin_ls.
+		dockerThinPoolName, err := dockerutil.DockerThinPoolName(*dockerInfo)
+		if err != nil {
+			return fmt.Errorf("couldn't find device mapper thin pool name: %v", err)
+		}
+
+		dockerMetadataDevice, err := dockerutil.DockerMetadataDevice(*dockerInfo)
+		if err != nil {
+			return fmt.Errorf("couldn't determine devicemapper metadata device")
+		}
+
+		thinPoolWatcher = devicemapper.NewThinPoolWatcher(dockerThinPoolName, dockerMetadataDevice)
+		go thinPoolWatcher.Start()
+	}
+
+	glog.Infof("registering Docker factory")
 	f := &dockerFactory{
 		cgroupSubsystems:   cgroupSubsystems,
 		client:             client,
@@ -197,6 +224,7 @@ func Register(factory info.MachineInfoFactory, fsInfo fs.FsInfo, ignoreMetrics c
 		storageDriver:      storageDriver(dockerInfo.Driver),
 		storageDir:         RootDir(),
 		ignoreMetrics:      ignoreMetrics,
+		thinPoolWatcher:    thinPoolWatcher,
 	}
 
 	container.RegisterContainerHandlerFactory(f, []watcher.ContainerWatchSource{watcher.Raw})

container/docker/handler.go

@@ -25,11 +25,14 @@ import (
 	"github.com/google/cadvisor/container"
 	"github.com/google/cadvisor/container/common"
 	containerlibcontainer "github.com/google/cadvisor/container/libcontainer"
+	"github.com/google/cadvisor/devicemapper"
 	"github.com/google/cadvisor/fs"
 	info "github.com/google/cadvisor/info/v1"
+	dockerutil "github.com/google/cadvisor/utils/docker"
 
 	docker "github.com/docker/engine-api/client"
 	dockercontainer "github.com/docker/engine-api/types/container"
+	"github.com/golang/glog"
 	"github.com/opencontainers/runc/libcontainer/cgroups"
 	cgroupfs "github.com/opencontainers/runc/libcontainer/cgroups/fs"
 	libcontainerconfigs "github.com/opencontainers/runc/libcontainer/configs"
@@ -57,10 +60,18 @@ type dockerContainerHandler struct {
 	// Manager of this container's cgroups.
 	cgroupManager cgroups.Manager
 
+	// the docker storage driver
 	storageDriver    storageDriver
 	fsInfo           fs.FsInfo
 	rootfsStorageDir string
 
+	// devicemapper state
+
+	// the devicemapper poolname
+	poolName string
+	// the devicemapper device id for the container
+	deviceID string
+
 	// Time at which this container was created.
 	creationTime time.Time
 
@@ -84,8 +95,13 @@ type dockerContainerHandler struct {
 	fsHandler common.FsHandler
 
 	ignoreMetrics container.MetricSet
+
+	// thin pool watcher
+	thinPoolWatcher *devicemapper.ThinPoolWatcher
 }
 
+var _ container.ContainerHandler = &dockerContainerHandler{}
+
 func getRwLayerID(containerID, storageDir string, sd storageDriver, dockerVersion []int) (string, error) {
 	const (
 		// Docker version >=1.10.0 have a randomized ID for the root fs of a container.
@@ -103,6 +119,7 @@ func getRwLayerID(containerID, storageDir string, sd storageDriver, dockerVersio
 	return string(bytes), err
 }
 
+// newDockerContainerHandler returns a new container.ContainerHandler
 func newDockerContainerHandler(
 	client *docker.Client,
 	name string,
@@ -115,6 +132,7 @@ func newDockerContainerHandler(
 	metadataEnvs []string,
 	dockerVersion []int,
 	ignoreMetrics container.MetricSet,
+	thinPoolWatcher *devicemapper.ThinPoolWatcher,
 ) (container.ContainerHandler, error) {
 	// Create the cgroup paths.
 	cgroupPaths := make(map[string]string, len(cgroupSubsystems.MountPoints))
@@ -146,14 +164,27 @@ func newDockerContainerHandler(
 	if err != nil {
 		return nil, err
 	}
-	var rootfsStorageDir string
+
+	// Determine the rootfs storage dir OR the pool name to determine the device
+	var (
+		rootfsStorageDir string
+		poolName         string
+	)
 	switch storageDriver {
 	case aufsStorageDriver:
 		rootfsStorageDir = path.Join(storageDir, string(aufsStorageDriver), aufsRWLayer, rwLayerID)
 	case overlayStorageDriver:
 		rootfsStorageDir = path.Join(storageDir, string(overlayStorageDriver), rwLayerID)
+	case devicemapperStorageDriver:
+		status, err := Status()
+		if err != nil {
+			return nil, fmt.Errorf("unable to determine docker status: %v", err)
+		}
+
+		poolName = status.DriverStatus[dockerutil.DriverStatusPoolName]
 	}
 
+	// TODO: extract object mother method
 	handler := &dockerContainerHandler{
 		id:                 id,
 		client:             client,
@@ -164,13 +195,11 @@ func newDockerContainerHandler(
 		storageDriver:      storageDriver,
 		fsInfo:             fsInfo,
 		rootFs:             rootFs,
+		poolName:           poolName,
 		rootfsStorageDir:   rootfsStorageDir,
 		envs:               make(map[string]string),
 		ignoreMetrics:      ignoreMetrics,
-	}
+		thinPoolWatcher:    thinPoolWatcher,
-
-	if !ignoreMetrics.Has(container.DiskUsageMetrics) {
-		handler.fsHandler = common.NewFsHandler(time.Minute, rootfsStorageDir, otherStorageDir, fsInfo)
 	}
 
 	// We assume that if Inspect fails then the container is not known to docker.
@@ -191,6 +220,15 @@ func newDockerContainerHandler(
 	handler.labels = ctnr.Config.Labels
 	handler.image = ctnr.Config.Image
 	handler.networkMode = ctnr.HostConfig.NetworkMode
+	handler.deviceID = ctnr.GraphDriver.Data["DeviceId"]
+
+	if !ignoreMetrics.Has(container.DiskUsageMetrics) {
+		handler.fsHandler = &dockerFsHandler{
+			fsHandler:       common.NewFsHandler(time.Minute, rootfsStorageDir, otherStorageDir, fsInfo),
+			thinPoolWatcher: thinPoolWatcher,
+			deviceID:        handler.deviceID,
+		}
+	}
 
 	// split env vars to get metadata map.
 	for _, exposedEnv := range metadataEnvs {
@@ -205,6 +243,48 @@ func newDockerContainerHandler(
 	return handler, nil
 }
 
+// dockerFsHandler is a composite FsHandler implementation the incorporates
+// the common fs handler and a devicemapper ThinPoolWatcher.
+type dockerFsHandler struct {
+	fsHandler common.FsHandler
+
+	// thinPoolWatcher is the devicemapper thin pool watcher
+	thinPoolWatcher *devicemapper.ThinPoolWatcher
+	// deviceID is the id of the container's fs device
+	deviceID string
+}
+
+var _ common.FsHandler = &dockerFsHandler{}
+
+func (h *dockerFsHandler) Start() {
+	h.fsHandler.Start()
+}
+
+func (h *dockerFsHandler) Stop() {
+	h.fsHandler.Stop()
+}
+
+func (h *dockerFsHandler) Usage() (uint64, uint64) {
+	baseUsage, usage := h.fsHandler.Usage()
+
+	// When devicemapper is the storage driver, the base usage of the container comes from the thin pool.
+	// We still need the result of the fsHandler for any extra storage associated with the container.
+	// To correctly factor in the thin pool usage, we should:
+	// * Usage the thin pool usage as the base usage
+	// * Calculate the overall usage by adding the overall usage from the fs handler to the thin pool usage
+	if h.thinPoolWatcher != nil {
+		thinPoolUsage, err := h.thinPoolWatcher.GetUsage(h.deviceID)
+		if err != nil {
+			glog.Errorf("unable to get fs usage from thin pool for device %v: %v", h.deviceID, err)
+		} else {
+			baseUsage = thinPoolUsage
+			usage += thinPoolUsage
+		}
+	}
+
+	return baseUsage, usage
+}
+
 func (self *dockerContainerHandler) Start() {
 	if self.fsHandler != nil {
 		self.fsHandler.Start()
@@ -249,17 +329,22 @@ func (self *dockerContainerHandler) getFsStats(stats *info.ContainerStats) error
 	if self.ignoreMetrics.Has(container.DiskUsageMetrics) {
 		return nil
 	}
+	var device string
 	switch self.storageDriver {
+	case devicemapperStorageDriver:
+		// Device has to be the pool name to correlate with the device name as
+		// set in the machine info filesystems.
+		device = self.poolName
 	case aufsStorageDriver, overlayStorageDriver, zfsStorageDriver:
+		deviceInfo, err := self.fsInfo.GetDirFsDevice(self.rootfsStorageDir)
+		if err != nil {
+			return fmt.Errorf("unable to determine device info for dir: %v: %v", self.rootfsStorageDir, err)
+		}
+		device = deviceInfo.Device
 	default:
 		return nil
 	}
 
-	deviceInfo, err := self.fsInfo.GetDirFsDevice(self.rootfsStorageDir)
-	if err != nil {
-		return err
-	}
-
 	mi, err := self.machineInfoFactory.GetMachineInfo()
 	if err != nil {
 		return err
@@ -272,16 +357,16 @@ func (self *dockerContainerHandler) getFsStats(stats *info.ContainerStats) error
 
 	// Docker does not impose any filesystem limits for containers. So use capacity as limit.
 	for _, fs := range mi.Filesystems {
-		if fs.Device == deviceInfo.Device {
+		if fs.Device == device {
 			limit = fs.Capacity
 			fsType = fs.Type
 			break
 		}
 	}
 
-	fsStat := info.FsStats{Device: deviceInfo.Device, Type: fsType, Limit: limit}
+	fsStat := info.FsStats{Device: device, Type: fsType, Limit: limit}
-
 	fsStat.BaseUsage, fsStat.Usage = self.fsHandler.Usage()
+
 	stats.Filesystem = append(stats.Filesystem, fsStat)
 
 	return nil

devicemapper/dmsetup_client.go

@@ -0,0 +1,56 @@
+// Copyright 2016 Google Inc. 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 devicemapper
+
+import (
+	"os/exec"
+	"strconv"
+	"strings"
+
+	"github.com/golang/glog"
+)
+
+// DmsetupClient is a low-level client for interacting with devicemapper via
+// the dmsetup utility.
+type DmsetupClient interface {
+	Table(deviceName string) ([]byte, error)
+	Message(deviceName string, sector int, message string) ([]byte, error)
+	Status(deviceName string) ([]byte, error)
+}
+
+func NewDmsetupClient() DmsetupClient {
+	return &defaultDmsetupClient{}
+}
+
+// defaultDmsetupClient implements the standard behavior for interacting with dmsetup.
+type defaultDmsetupClient struct{}
+
+var _ DmsetupClient = &defaultDmsetupClient{}
+
+func (c *defaultDmsetupClient) Table(deviceName string) ([]byte, error) {
+	return c.dmsetup("table", deviceName)
+}
+
+func (c *defaultDmsetupClient) Message(deviceName string, sector int, message string) ([]byte, error) {
+	return c.dmsetup("message", deviceName, strconv.Itoa(sector), message)
+}
+
+func (c *defaultDmsetupClient) Status(deviceName string) ([]byte, error) {
+	return c.dmsetup("status", deviceName)
+}
+
+func (*defaultDmsetupClient) dmsetup(args ...string) ([]byte, error) {
+	glog.V(5).Infof("running dmsetup %v", strings.Join(args, " "))
+	return exec.Command("dmsetup", args...).Output()
+}

devicemapper/doc.go

@@ -0,0 +1,16 @@
+// Copyright 2016 Google Inc. 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 devicemapper contains code for working with devicemapper
+package devicemapper

devicemapper/fake/dmsetup_client_fake.go

@@ -0,0 +1,64 @@
+// Copyright 2016 Google Inc. 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 fake
+
+import (
+	"testing"
+)
+
+type DmsetupCommand struct {
+	Name   string
+	Result string
+	Err    error
+}
+
+func NewFakeDmsetupClient(t *testing.T, commands ...DmsetupCommand) *FakeDmsetupClient {
+	if len(commands) == 0 {
+		commands = make([]DmsetupCommand, 0)
+	}
+	return &FakeDmsetupClient{t: t, commands: commands}
+}
+
+// FakeDmsetupClient is a thread-unsafe fake implementation of the DmsetupClient interface
+type FakeDmsetupClient struct {
+	t        *testing.T
+	commands []DmsetupCommand
+}
+
+func (c *FakeDmsetupClient) Table(deviceName string) ([]byte, error) {
+	return c.dmsetup("table")
+}
+
+func (c *FakeDmsetupClient) Message(deviceName string, sector int, message string) ([]byte, error) {
+	return c.dmsetup("message")
+}
+
+func (c *FakeDmsetupClient) Status(deviceName string) ([]byte, error) {
+	return c.dmsetup("status")
+}
+
+func (c *FakeDmsetupClient) AddCommand(name string, result string, err error) {
+	c.commands = append(c.commands, DmsetupCommand{name, result, err})
+}
+
+func (c *FakeDmsetupClient) dmsetup(inputCommand string) ([]byte, error) {
+	var nextCommand DmsetupCommand
+	nextCommand, c.commands = c.commands[0], c.commands[1:]
+	if nextCommand.Name != inputCommand {
+		c.t.Fatalf("unexpected dmsetup command; expected: %q, got %q", nextCommand.Name, inputCommand)
+		// should be unreachable in a test context.
+	}
+
+	return []byte(nextCommand.Result), nextCommand.Err
+}

devicemapper/fake/thin_ls_client_fake.go

@@ -0,0 +1,27 @@
+// Copyright 2016 Google Inc. 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 fake
+
+type FakeThinLsClient struct {
+	result map[string]uint64
+	err    error
+}
+
+func NewFakeThinLsClient(result map[string]uint64, err error) *FakeThinLsClient {
+	return &FakeThinLsClient{result, err}
+}
+
+func (c *FakeThinLsClient) ThinLs(deviceName string) (map[string]uint64, error) {
+	return c.result, c.err
+}

devicemapper/thin_ls_client.go

@@ -0,0 +1,77 @@
+// Copyright 2016 Google Inc. 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 devicemapper
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"os/exec"
+	"strconv"
+	"strings"
+
+	"github.com/golang/glog"
+)
+
+// thinLsClient knows how to run a thin_ls very specific to CoW usage for containers.
+type thinLsClient interface {
+	ThinLs(deviceName string) (map[string]uint64, error)
+}
+
+func newThinLsClient() thinLsClient {
+	return &defaultThinLsClient{}
+}
+
+type defaultThinLsClient struct{}
+
+var _ thinLsClient = &defaultThinLsClient{}
+
+func (*defaultThinLsClient) ThinLs(deviceName string) (map[string]uint64, error) {
+	args := []string{"--no-headers", "-m", "-o", "DEV,EXCLUSIVE_BYTES", deviceName}
+	glog.V(4).Infof("running command: thin_ls %v", strings.Join(args, " "))
+
+	output, err := exec.Command("thin_ls", args...).Output()
+	if err != nil {
+		return nil, fmt.Errorf("Error running command `thin_ls %v`: %v\noutput:\n\n%v", strings.Join(args, " "), err, string(output))
+	}
+
+	return parseThinLsOutput(output), nil
+}
+
+// parseThinLsOutput parses the output returned by thin_ls to build a map of device id -> usage.
+func parseThinLsOutput(output []byte) map[string]uint64 {
+	cache := map[string]uint64{}
+
+	// parse output
+	scanner := bufio.NewScanner(bytes.NewReader(output))
+	for scanner.Scan() {
+		output := scanner.Text()
+		fields := strings.Fields(output)
+		if len(fields) != 2 {
+			continue
+		}
+
+		deviceID := fields[0]
+		usage, err := strconv.ParseUint(fields[1], 10, 64)
+		if err != nil {
+			glog.Warning("unexpected error parsing thin_ls output: %v", err)
+			continue
+		}
+
+		cache[deviceID] = usage
+	}
+
+	return cache
+
+}

devicemapper/thin_ls_client_test.go

@@ -0,0 +1,61 @@
+// Copyright 2016 Google Inc. 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 devicemapper
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestParseThinLsOutput(t *testing.T) {
+	cases := []struct {
+		name           string
+		input          string
+		expectedResult map[string]uint64
+	}{
+		{
+			name: "ok",
+			input: `
+  1         2293760
+  2         2097152
+  3          131072
+  4         2031616`,
+			expectedResult: map[string]uint64{
+				"1": 2293760,
+				"2": 2097152,
+				"3": 131072,
+				"4": 2031616,
+			},
+		},
+		{
+			name: "skip bad rows",
+			input: `
+  1         2293760
+  2         2097152
+  3          131072ads
+  4d dsrv         2031616`,
+			expectedResult: map[string]uint64{
+				"1": 2293760,
+				"2": 2097152,
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		actualResult := parseThinLsOutput([]byte(tc.input))
+		if e, a := tc.expectedResult, actualResult; !reflect.DeepEqual(e, a) {
+			t.Errorf("%v: unexpected result: expected %+v got %+v", tc.name, e, a)
+		}
+	}
+}

devicemapper/thin_pool_watcher.go

@@ -0,0 +1,164 @@
+// Copyright 2016 Google Inc. 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 devicemapper
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/golang/glog"
+)
+
+// ThinPoolWatcher maintains a cache of device name -> usage stats for a devicemapper thin-pool using thin_ls.
+type ThinPoolWatcher struct {
+	poolName       string
+	metadataDevice string
+	lock           *sync.RWMutex
+	cache          map[string]uint64
+	period         time.Duration
+	stopChan       chan struct{}
+	dmsetup        DmsetupClient
+	thinLsClient   thinLsClient
+}
+
+// NewThinPoolWatcher returns a new ThinPoolWatcher for the given devicemapper thin pool name and metadata device.
+func NewThinPoolWatcher(poolName, metadataDevice string) *ThinPoolWatcher {
+	return &ThinPoolWatcher{poolName: poolName,
+		metadataDevice: metadataDevice,
+		lock:           &sync.RWMutex{},
+		cache:          make(map[string]uint64),
+		period:         15 * time.Second,
+		stopChan:       make(chan struct{}),
+		dmsetup:        NewDmsetupClient(),
+		thinLsClient:   newThinLsClient(),
+	}
+}
+
+// Start starts the thin pool watcher.
+func (w *ThinPoolWatcher) Start() {
+	err := w.Refresh()
+	if err != nil {
+		glog.Errorf("encountered error refreshing thin pool watcher: %v", err)
+	}
+
+	for {
+		select {
+		case <-w.stopChan:
+			return
+		case <-time.After(w.period):
+			start := time.Now()
+			err = w.Refresh()
+			if err != nil {
+				glog.Errorf("encountered error refreshing thin pool watcher: %v", err)
+			}
+
+			// print latency for refresh
+			duration := time.Since(start)
+			glog.V(3).Infof("thin_ls(%d) took %s", start.Unix(), duration)
+		}
+	}
+}
+
+func (w *ThinPoolWatcher) Stop() {
+	close(w.stopChan)
+}
+
+// GetUsage gets the cached usage value of the given device.
+func (w *ThinPoolWatcher) GetUsage(deviceId string) (uint64, error) {
+	w.lock.RLock()
+	defer w.lock.RUnlock()
+	v, ok := w.cache[deviceId]
+	if !ok {
+		return 0, fmt.Errorf("no cached value for usage of device %v", deviceId)
+	}
+
+	return v, nil
+}
+
+const (
+	reserveMetadataMessage = "reserve_metadata_snap"
+	releaseMetadataMessage = "release_metadata_snap"
+)
+
+// Refresh performs a `thin_ls` of the pool being watched and refreshes the
+// cached data with the result.
+func (w *ThinPoolWatcher) Refresh() error {
+	w.lock.Lock()
+	defer w.lock.Unlock()
+
+	currentlyReserved, err := w.checkReservation(w.poolName)
+	if err != nil {
+		err = fmt.Errorf("error determining whether snapshot is reserved: %v", err)
+		return err
+	}
+
+	if currentlyReserved {
+		glog.V(4).Infof("metadata for %v is currently reserved; releasing", w.poolName)
+		_, err = w.dmsetup.Message(w.poolName, 0, releaseMetadataMessage)
+		if err != nil {
+			err = fmt.Errorf("error releasing metadata snapshot for %v: %v", w.poolName, err)
+			return err
+		}
+	}
+
+	glog.Infof("reserving metadata snapshot for thin-pool %v", w.poolName)
+	// NOTE: "0" in the call below is for the 'sector' argument to 'dmsetup message'.  It's not needed for thin pools.
+	if output, err := w.dmsetup.Message(w.poolName, 0, reserveMetadataMessage); err != nil {
+		err = fmt.Errorf("error reserving metadata for thin-pool %v: %v output: %v", w.poolName, err, string(output))
+		return err
+	} else {
+		glog.V(5).Infof("reserved metadata snapshot for thin-pool %v", w.poolName)
+	}
+
+	defer func() {
+		glog.V(5).Infof("releasing metadata snapshot for thin-pool %v", w.poolName)
+		w.dmsetup.Message(w.poolName, 0, releaseMetadataMessage)
+	}()
+
+	glog.V(5).Infof("running thin_ls on metadata device %v", w.metadataDevice)
+	newCache, err := w.thinLsClient.ThinLs(w.metadataDevice)
+	if err != nil {
+		err = fmt.Errorf("error performing thin_ls on metadata device %v: %v", w.metadataDevice, err)
+		return err
+	}
+
+	w.cache = newCache
+	return nil
+}
+
+const (
+	thinPoolDmsetupStatusTokens           = 11
+	thinPoolDmsetupStatusHeldMetadataRoot = 6
+)
+
+// checkReservation checks to see whether the thin device is currently holding userspace metadata.
+func (w *ThinPoolWatcher) checkReservation(poolName string) (bool, error) {
+	glog.V(5).Infof("checking whether the thin-pool is holding a metadata snapshot")
+	output, err := w.dmsetup.Status(poolName)
+	if err != nil {
+		return false, err
+	}
+
+	tokens := strings.Split(string(output), " ")
+	// Split returns the input as the last item in the result, adjust the number of tokens by one
+	if len(tokens) != thinPoolDmsetupStatusTokens+1 {
+		return false, fmt.Errorf("unexpected output of dmsetup status command; expected 11 fields, got %v; output: %v", len(tokens), string(output))
+	}
+
+	heldMetadataRoot := tokens[thinPoolDmsetupStatusHeldMetadataRoot]
+	currentlyReserved := heldMetadataRoot != "-"
+	return currentlyReserved, nil
+}

devicemapper/thin_pool_watcher_test.go

@@ -0,0 +1,209 @@
+// Copyright 2016 Google Inc. 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 devicemapper
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/cadvisor/devicemapper/fake"
+)
+
+func TestRefresh(t *testing.T) {
+	usage := map[string]uint64{
+		"1": 12345,
+		"2": 23456,
+		"3": 34567,
+	}
+
+	cases := []struct {
+		name            string
+		dmsetupCommands []fake.DmsetupCommand
+		thinLsOutput    map[string]uint64
+		thinLsErr       error
+		expectedError   bool
+		deviceId        string
+		expectedUsage   uint64
+	}{
+		{
+			name: "check reservation fails",
+			dmsetupCommands: []fake.DmsetupCommand{
+				{"status", "", fmt.Errorf("not gonna work")},
+			},
+			expectedError: true,
+		},
+		{
+			name: "no existing reservation - ok",
+			dmsetupCommands: []fake.DmsetupCommand{
+				{"status", "0 75497472 thin-pool 65 327/524288 14092/589824 - rw no_discard_passdown error_if_no_space - ", nil}, // status check
+				{"message", "", nil}, // make reservation
+				{"message", "", nil}, // release reservation
+			},
+			thinLsOutput:  usage,
+			expectedError: false,
+			deviceId:      "2",
+			expectedUsage: 23456,
+		},
+		{
+			name: "existing reservation - ok",
+			dmsetupCommands: []fake.DmsetupCommand{
+				// status check
+				{"status", "0 75497472 thin-pool 65 327/524288 14092/589824 39 rw no_discard_passdown error_if_no_space - ", nil},
+				// release reservation
+				{"message", "", nil},
+				// make reservation
+				{"message", "", nil},
+				// release reservation
+				{"message", "", nil},
+			},
+			thinLsOutput:  usage,
+			expectedError: false,
+			deviceId:      "3",
+			expectedUsage: 34567,
+		},
+		{
+			name: "failure releasing existing reservation",
+			dmsetupCommands: []fake.DmsetupCommand{
+				// status check
+				{"status", "0 75497472 thin-pool 65 327/524288 14092/589824 39 rw no_discard_passdown error_if_no_space - ", nil},
+				// release reservation
+				{"message", "", fmt.Errorf("not gonna work")},
+			},
+			expectedError: true,
+		},
+		{
+			name: "failure making reservation",
+			dmsetupCommands: []fake.DmsetupCommand{
+				// status check
+				{"status", "0 75497472 thin-pool 65 327/524288 14092/589824 39 rw no_discard_passdown error_if_no_space - ", nil},
+				// release reservation
+				{"message", "", nil},
+				// make reservation
+				{"message", "", fmt.Errorf("not gonna work")},
+			},
+			expectedError: true,
+		},
+		{
+			name: "failure running thin_ls",
+			dmsetupCommands: []fake.DmsetupCommand{
+				// status check
+				{"status", "0 75497472 thin-pool 65 327/524288 14092/589824 39 rw no_discard_passdown error_if_no_space - ", nil},
+				// release reservation
+				{"message", "", nil},
+				// make reservation
+				{"message", "", nil},
+				// release reservation
+				{"message", "", nil},
+			},
+			thinLsErr:     fmt.Errorf("not gonna work"),
+			expectedError: true,
+		},
+	}
+
+	for _, tc := range cases {
+		dmsetup := fake.NewFakeDmsetupClient(t, tc.dmsetupCommands...)
+		thinLsClient := fake.NewFakeThinLsClient(tc.thinLsOutput, tc.thinLsErr)
+		watcher := &ThinPoolWatcher{
+			poolName:       "test pool name",
+			metadataDevice: "/dev/mapper/metadata-device",
+			lock:           &sync.RWMutex{},
+			period:         15 * time.Second,
+			stopChan:       make(chan struct{}),
+			dmsetup:        dmsetup,
+			thinLsClient:   thinLsClient,
+		}
+
+		err := watcher.Refresh()
+		if err != nil {
+			if !tc.expectedError {
+				t.Errorf("%v: unexpected error: %v", tc.name, err)
+			}
+			continue
+		} else if tc.expectedError {
+			t.Errorf("%v: unexpected success", tc.name)
+			continue
+		}
+
+		actualUsage, err := watcher.GetUsage(tc.deviceId)
+		if err != nil {
+			t.Errorf("%v: device ID not found: %v", tc.deviceId, err)
+			continue
+		}
+
+		if e, a := tc.expectedUsage, actualUsage; e != a {
+			t.Errorf("%v: actual usage did not match expected usage: expected: %v got: %v", tc.name, e, a)
+		}
+	}
+}
+
+func TestCheckReservation(t *testing.T) {
+	cases := []struct {
+		name           string
+		statusResult   string
+		statusErr      error
+		expectedResult bool
+		expectedErr    error
+	}{
+		{
+			name:           "existing reservation 1",
+			statusResult:   "0 75497472 thin-pool 65 327/524288 14092/589824 36 rw no_discard_passdown queue_if_no_space - ",
+			expectedResult: true,
+		},
+		{
+			name:           "existing reservation 2",
+			statusResult:   "0 12345 thin-pool 65 327/45678 14092/45678 36 rw discard_passdown error_if_no_space needs_check ",
+			expectedResult: true,
+		},
+		{
+			name:           "no reservation 1",
+			statusResult:   "0 75497472 thin-pool 65 327/524288 14092/589824 - rw no_discard_passdown error_if_no_space - ",
+			expectedResult: false,
+		},
+		{
+			name:           "no reservation 2",
+			statusResult:   "0 75 thin-pool 65 327/12345 14092/589824 - rw no_discard_passdown queue_if_no_space - ",
+			expectedResult: false,
+		},
+		{
+			name:           "no reservation 2",
+			statusResult:   "0 75 thin-pool 65 327/12345 14092/589824 - rw no_discard_passdown queue_if_no_space - ",
+			expectedResult: false,
+		},
+		{
+			name:         "malformed input",
+			statusResult: "0 12345 14092/45678 36 rw discard_passdown error_if_no_space needs_check ",
+			expectedErr:  fmt.Errorf("not gonna work"),
+		},
+	}
+
+	for _, tc := range cases {
+		fakeDmsetupClient := fake.NewFakeDmsetupClient(t)
+		fakeDmsetupClient.AddCommand("status", tc.statusResult, tc.statusErr)
+		watcher := &ThinPoolWatcher{dmsetup: fakeDmsetupClient}
+		actualResult, err := watcher.checkReservation("test pool")
+		if err != nil {
+			if tc.expectedErr == nil {
+				t.Errorf("%v: unexpected error running checkReservation: %v", tc.name, err)
+			}
+		} else if tc.expectedErr != nil {
+			t.Errorf("%v: unexpected success running checkReservation", tc.name)
+		}
+
+		if e, a := tc.expectedResult, actualResult; e != a {
+			t.Errorf("%v: unexpected result from checkReservation: expected: %v got: %v", tc.name, e, a)
+		}
+	}
+}

fs/fs.go

@@ -33,6 +33,8 @@ import (
 
 	"github.com/docker/docker/pkg/mount"
 	"github.com/golang/glog"
+	"github.com/google/cadvisor/devicemapper"
+	dockerutil "github.com/google/cadvisor/utils/docker"
 	zfs "github.com/mistifyio/go-zfs"
 )
 
@@ -56,8 +58,8 @@ type RealFsInfo struct {
 	// Map from label to block device path.
 	// Labels are intent-specific tags that are auto-detected.
 	labels map[string]string
-
+	// devicemapper client
-	dmsetup dmsetupClient
+	dmsetup devicemapper.DmsetupClient
 }
 
 type Context struct {
@@ -80,13 +82,9 @@ func NewFsInfo(context Context) (FsInfo, error) {
 	fsInfo := &RealFsInfo{
 		partitions: make(map[string]partition, 0),
 		labels:     make(map[string]string, 0),
-		dmsetup:    &defaultDmsetupClient{},
+		dmsetup:    devicemapper.NewDmsetupClient(),
 	}
 
-	fsInfo.addSystemRootLabel(mounts)
-	fsInfo.addDockerImagesLabel(context, mounts)
-	fsInfo.addRktImagesLabel(context, mounts)
-
 	supportedFsType := map[string]bool{
 		// all ext systems are checked through prefix.
 		"btrfs": true,
@@ -113,7 +111,13 @@ func NewFsInfo(context Context) (FsInfo, error) {
 		}
 	}
 
+	fsInfo.addRktImagesLabel(context, mounts)
+	// need to call this before the log line below printing out the partitions, as this function may
+	// add a "partition" for devicemapper to fsInfo.partitions
+	fsInfo.addDockerImagesLabel(context, mounts)
+
 	glog.Infof("Filesystem partitions: %+v", fsInfo.partitions)
+	fsInfo.addSystemRootLabel(mounts)
 	return fsInfo, nil
 }
 
@@ -126,7 +130,7 @@ func (self *RealFsInfo) getDockerDeviceMapperInfo(context DockerContext) (string
 		return "", nil, nil
 	}
 
-	dataLoopFile := context.DriverStatus["Data loop file"]
+	dataLoopFile := context.DriverStatus[dockerutil.DriverStatusDataLoopFile]
 	if len(dataLoopFile) > 0 {
 		return "", nil, nil
 	}
@@ -274,6 +278,7 @@ func (self *RealFsInfo) GetFsInfoForPath(mountSet map[string]struct{}) ([]Fs, er
 			switch partition.fsType {
 			case DeviceMapper.String():
 				fs.Capacity, fs.Free, fs.Available, err = getDMStats(device, partition.blockSize)
+				glog.V(5).Infof("got devicemapper fs capacity stats: capacity: %v free: %v available: %v:", fs.Capacity, fs.Free, fs.Available)
 				fs.Type = DeviceMapper
 			case ZFS.String():
 				fs.Capacity, fs.Free, fs.Available, err = getZfstats(device)
@@ -434,30 +439,15 @@ func getVfsStats(path string) (total uint64, free uint64, avail uint64, inodes u
 	return total, free, avail, inodes, inodesFree, nil
 }
 
-// dmsetupClient knows to to interact with dmsetup to retrieve information about devicemapper.
-type dmsetupClient interface {
-	table(poolName string) ([]byte, error)
-	//TODO add status(poolName string) ([]byte, error) and use it in getDMStats so we can unit test
-}
-
-// defaultDmsetupClient implements the standard behavior for interacting with dmsetup.
-type defaultDmsetupClient struct{}
-
-var _ dmsetupClient = &defaultDmsetupClient{}
-
-func (*defaultDmsetupClient) table(poolName string) ([]byte, error) {
-	return exec.Command("dmsetup", "table", poolName).Output()
-}
-
 // Devicemapper thin provisioning is detailed at
 // https://www.kernel.org/doc/Documentation/device-mapper/thin-provisioning.txt
-func dockerDMDevice(driverStatus map[string]string, dmsetup dmsetupClient) (string, uint, uint, uint, error) {
+func dockerDMDevice(driverStatus map[string]string, dmsetup devicemapper.DmsetupClient) (string, uint, uint, uint, error) {
-	poolName, ok := driverStatus["Pool Name"]
+	poolName, ok := driverStatus[dockerutil.DriverStatusPoolName]
 	if !ok || len(poolName) == 0 {
 		return "", 0, 0, 0, fmt.Errorf("Could not get dm pool name")
 	}
 
-	out, err := dmsetup.table(poolName)
+	out, err := dmsetup.Table(poolName)
 	if err != nil {
 		return "", 0, 0, 0, err
 	}
@@ -470,6 +460,8 @@ func dockerDMDevice(driverStatus map[string]string, dmsetup dmsetupClient) (stri
 	return poolName, major, minor, dataBlkSize, nil
 }
 
+// parseDMTable parses a single line of `dmsetup table` output and returns the
+// major device, minor device, block size, and an error.
 func parseDMTable(dmTable string) (uint, uint, uint, error) {
 	dmTable = strings.Replace(dmTable, ":", " ", -1)
 	dmFields := strings.Fields(dmTable)

fs/fs_test.go

@@ -193,7 +193,15 @@ type testDmsetup struct {
 	err  error
 }
 
-func (t *testDmsetup) table(poolName string) ([]byte, error) {
+func (*testDmsetup) Message(deviceName string, sector int, message string) ([]byte, error) {
+	return nil, nil
+}
+
+func (*testDmsetup) Status(deviceName string) ([]byte, error) {
+	return nil, nil
+}
+
+func (t *testDmsetup) Table(poolName string) ([]byte, error) {
 	return t.data, t.err
 }
 

integration/tests/api/docker_test.go

@@ -333,7 +333,7 @@ func TestDockerFilesystemStats(t *testing.T) {
 	}
 	needsBaseUsageCheck := false
 	switch storageDriver {
-	case framework.Aufs, framework.Overlay:
+	case framework.Aufs, framework.Overlay, framework.DeviceMapper:
 		needsBaseUsageCheck = true
 	}
 	pass := false

utils/docker/docker.go

@@ -0,0 +1,58 @@
+// Copyright 2016 Google Inc. 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 docker
+
+import (
+	"fmt"
+	"strings"
+
+	dockertypes "github.com/docker/engine-api/types"
+)
+
+const (
+	DockerInfoDriver         = "Driver"
+	DockerInfoDriverStatus   = "DriverStatus"
+	DriverStatusPoolName     = "Pool Name"
+	DriverStatusDataLoopFile = "Data loop file"
+	DriverStatusMetadataFile = "Metadata file"
+)
+
+func DriverStatusValue(status [][2]string, target string) string {
+	for _, v := range status {
+		if strings.EqualFold(v[0], target) {
+			return v[1]
+		}
+	}
+
+	return ""
+}
+
+func DockerThinPoolName(info dockertypes.Info) (string, error) {
+	poolName := DriverStatusValue(info.DriverStatus, DriverStatusPoolName)
+	if len(poolName) == 0 {
+		return "", fmt.Errorf("Could not get devicemapper pool name")
+	}
+
+	return poolName, nil
+}
+
+func DockerMetadataDevice(info dockertypes.Info) (string, error) {
+	metadataDevice := DriverStatusValue(info.DriverStatus, DriverStatusMetadataFile)
+	if len(metadataDevice) == 0 {
+		return "", fmt.Errorf("Could not get the devicemapper metadata device")
+	}
+
+	return metadataDevice, nil
+}