Add StatFs and the df [-h] command

cmd/hdfs/df.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"text/tabwriter"
+
+	"github.com/colinmarc/hdfs"
+)
+
+func df(humanReadable bool) {
+	client, err := getClient("")
+	if err != nil {
+		fatal(err)
+	}
+
+	var fs hdfs.FsInfo
+
+	fs, err = client.StatFs()
+	if err != nil {
+		fatal(err)
+	}
+
+	tw := tabwriter.NewWriter(os.Stdout, 3, 8, 0, ' ', tabwriter.AlignRight)
+	fmt.Fprintf(tw, "Filesystem \tSize \tUsed \tAvailable \t Use%%\n")
+	if humanReadable {
+		fmt.Fprintf(tw, "%v \t%v \t%v \t%v \t%d%%\n",
+			os.Getenv("HADOOP_NAMENODE"),
+			formatBytes(fs.Capacity),
+			formatBytes(fs.Used),
+			formatBytes(fs.Remaining),
+			100 * fs.Used / fs.Capacity)
+	} else {
+		fmt.Fprintf(tw, "%v \t%v \t %v \t %v \t%d%%\n",
+			os.Getenv("HADOOP_NAMENODE"),
+			fs.Capacity,
+			fs.Used,
+			fs.Remaining,
+			100 * fs.Used / fs.Capacity)
+	}
+	tw.Flush()
+}

cmd/hdfs/du.go

@@ -93,7 +93,7 @@ func duDir(client *hdfs.Client, tw *tabwriter.Writer, dir string, humanReadable
 
 func printSize(tw *tabwriter.Writer, size int64, name string, humanReadable bool) {
 	if humanReadable {
-		formattedSize := formatBytes(size)
+		formattedSize := formatBytes(uint64(size))
 		fmt.Fprintf(tw, "%s \t%s\n", formattedSize, name)
 	} else {
 		fmt.Fprintf(tw, "%d \t%s\n", size, name)

cmd/hdfs/ls.go

@@ -136,7 +136,7 @@ func printLong(tw *tabwriter.Writer, name string, info os.FileInfo, humanReadabl
 	group := fi.OwnerGroup()
 	size := strconv.FormatInt(fi.Size(), 10)
 	if humanReadable {
-		size = formatBytes(fi.Size())
+		size = formatBytes(uint64(fi.Size()))
 	}
 
 	modtime := fi.ModTime()

cmd/hdfs/main.go

@@ -9,7 +9,7 @@ import (
 	"github.com/pborman/getopt"
 )
 
-// TODO: cp, df, tree, test, trash
+// TODO: cp, tree, test, trash
 
 var (
 	version string
@@ -32,6 +32,7 @@ Valid commands:
   get SOURCE [DEST]
   getmerge SOURCE DEST
   put SOURCE DEST
+  df [-h]
 `, os.Args[0])
 
 	lsOpts = getopt.New()
@@ -70,6 +71,9 @@ Valid commands:
 	getmergeOpts = getopt.New()
 	getmergen    = getmergeOpts.Bool('n')
 
+	dfOpts = getopt.New()
+	dfh    = dfOpts.Bool('h')
+
 	cachedClient *hdfs.Client
 	status       = 0
 )
@@ -84,6 +88,7 @@ func init() {
 	headTailOpts.SetUsage(printHelp)
 	duOpts.SetUsage(printHelp)
 	getmergeOpts.SetUsage(printHelp)
+	dfOpts.SetUsage(printHelp)
 }
 
 func main() {
@@ -134,6 +139,9 @@ func main() {
 		getmerge(getmergeOpts.Args(), *getmergen)
 	case "put":
 		put(argv[1:])
+	case "df":
+		dfOpts.Parse(argv)
+		df(*dfh)
 	// it's a seeeeecret command
 	case "complete":
 		complete(argv)

cmd/hdfs/test/df.bats

@@ -0,0 +1,8 @@
+#!/usr/bin/env bats
+
+load helper
+
+@test "df" {
+  run $HDFS df
+  assert_success
+}

cmd/hdfs/util.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 )
 
-func formatBytes(i int64) string {
+func formatBytes(i uint64) string {
 	switch {
 	case i > (1024 * 1024 * 1024 * 1024):
 		return fmt.Sprintf("%#.1fT", float64(i)/1024/1024/1024/1024)

stat_fs.go

@@ -0,0 +1,49 @@
+package hdfs
+
+import (
+	"errors"
+
+	hdfs "github.com/colinmarc/hdfs/protocol/hadoop_hdfs"
+	"github.com/colinmarc/hdfs/rpc"
+)
+
+var StatFsError = errors.New("Failed to get HDFS usage")
+
+// FsInfo provides information about HDFS
+type FsInfo struct {
+	Capacity              uint64
+	Used                  uint64
+	Remaining             uint64
+	UnderReplicated       uint64
+	CorruptBlocks         uint64
+	MissingBlocks         uint64
+	MissingReplOneBlocks  uint64
+	BlocksInFuture        uint64
+	PendingDeletionBlocks uint64
+}
+
+func (c *Client) StatFs() (FsInfo, error) {
+	req  := &hdfs.GetFsStatusRequestProto{}
+	resp := &hdfs.GetFsStatsResponseProto{}
+
+	err := c.namenode.Execute("getFsStats", req, resp)
+	if err != nil {
+		if nnErr, ok := err.(*rpc.NamenodeError); ok {
+			err = interpretException(nnErr.Exception, err)
+		}
+		return FsInfo{}, err
+	}
+
+	var fs FsInfo
+	fs.Capacity              = resp.GetCapacity()
+	fs.Used                  = resp.GetUsed()
+	fs.Remaining             = resp.GetRemaining()
+	fs.UnderReplicated       = resp.GetUnderReplicated()
+	fs.CorruptBlocks         = resp.GetCorruptBlocks()
+	fs.MissingBlocks         = resp.GetMissingBlocks()
+	fs.MissingReplOneBlocks  = resp.GetMissingReplOneBlocks()
+	fs.BlocksInFuture        = resp.GetBlocksInFuture()
+	fs.PendingDeletionBlocks = resp.GetPendingDeletionBlocks()
+
+	return fs, nil
+}

stat_fs_test.go

@@ -0,0 +1,14 @@
+package hdfs
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestStatFs(t *testing.T) {
+	client := getClient(t)
+
+	_, err := client.StatFs()
+	require.NoError(t, err)
+}