Replace info w/ report command

info was not safe to run and not really useful.

Godeps

@@ -15,3 +15,4 @@ github.com/paulbellamy/ratecounter 5a11f585a31379765c190c033b6ad39956584447
 github.com/peterh/liner 8975875355a81d612fafb9f5a6037bdcc2d9b073
 github.com/rakyll/statik 274df120e9065bdd08eb1120e0375e3dc1ae8465
 golang.org/x/crypto c197bcf24cde29d3f73c7b4ac6fd41f4384e8af6
+github.com/retailnext/hllpp 38a7bb71b483e855d35010808143beaf05b67f9d

cmd/influx_inspect/info.go

@@ -1,101 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"io/ioutil"
-	"log"
-	"os"
-	"path/filepath"
-	"sort"
-	"strings"
-	"text/tabwriter"
-
-	"github.com/influxdata/influxdb/influxql"
-	"github.com/influxdata/influxdb/tsdb"
-)
-
-func cmdInfo(path string) {
-	tstore := tsdb.NewStore(filepath.Join(path, "data"))
-	tstore.Logger = log.New(ioutil.Discard, "", log.LstdFlags)
-	tstore.EngineOptions.Config.Dir = filepath.Join(path, "data")
-	tstore.EngineOptions.Config.WALLoggingEnabled = false
-	tstore.EngineOptions.Config.WALDir = filepath.Join(path, "wal")
-	if err := tstore.Open(); err != nil {
-		fmt.Printf("Failed to open dir: %v\n", err)
-		os.Exit(1)
-	}
-
-	size, err := tstore.DiskSize()
-	if err != nil {
-		fmt.Printf("Failed to determine disk usage: %v\n", err)
-	}
-
-	// Summary stats
-	fmt.Printf("Shards: %d, Indexes: %d, Databases: %d, Disk Size: %d, Series: %d\n\n",
-		tstore.ShardN(), tstore.DatabaseIndexN(), len(tstore.Databases()), size, countSeries(tstore))
-
-	tw := tabwriter.NewWriter(os.Stdout, 16, 8, 0, '\t', 0)
-
-	fmt.Fprintln(tw, strings.Join([]string{"Shard", "DB", "Measurement", "Tags [#K/#V]", "Fields [Name:Type]", "Series"}, "\t"))
-
-	shardIDs := tstore.ShardIDs()
-
-	databases := tstore.Databases()
-	sort.Strings(databases)
-
-	for _, db := range databases {
-		index := tstore.DatabaseIndex(db)
-		measurements := index.Measurements()
-		sort.Sort(measurements)
-		for _, m := range measurements {
-			tags := m.TagKeys()
-			tagValues := 0
-			for _, tag := range tags {
-				tagValues += len(m.TagValues(tag))
-			}
-			fields := m.FieldNames()
-			sort.Strings(fields)
-			series := m.SeriesKeys()
-			sort.Sort(ShardIDs(shardIDs))
-
-			// Sample a point from each measurement to determine the field types
-			sources := []influxql.Source{&influxql.Measurement{Name: m.Name}}
-			for _, shardID := range shardIDs {
-				shard := tstore.Shard(shardID)
-				fieldSet, _, err := shard.FieldDimensions(sources)
-				if err != nil {
-					continue
-				}
-				for _, name := range fields {
-					ft := fmt.Sprintf("%s:%s", name, fieldSet[name])
-					fmt.Fprintf(tw, "%d\t%s\t%s\t%d/%d\t%d [%s]\t%d\n", shardID, db, m.Name, len(tags), tagValues,
-						len(fields), ft, len(series))
-
-				}
-
-			}
-		}
-	}
-	tw.Flush()
-}
-
-func countSeries(tstore *tsdb.Store) int {
-	var count int
-	for _, shardID := range tstore.ShardIDs() {
-		shard := tstore.Shard(shardID)
-		cnt, err := shard.SeriesCount()
-		if err != nil {
-			fmt.Printf("series count failed: %v\n", err)
-			continue
-		}
-		count += cnt
-	}
-	return count
-}
-
-// ShardIDs is a collection of UINT 64 that represent shard ids.
-type ShardIDs []uint64
-
-func (a ShardIDs) Len() int           { return len(a) }
-func (a ShardIDs) Less(i, j int) bool { return a[i] < a[j] }
-func (a ShardIDs) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }

cmd/influx_inspect/main.go

@@ -15,9 +15,9 @@ Displays detailed information about InfluxDB data files.
 `)
 
 	println(`Commands:
-  info - displays series meta-data for all shards.  Default location [$HOME/.influxdb]
   dumptsm - dumps low-level details about tsm1 files.
-  export - exports a tsm file to line protocol`)
+  export - exports a tsm file to line protocol
+  report - displays a shard level report`)
 	println()
 }
 
@@ -32,13 +32,14 @@ func main() {
 	}
 
 	switch flag.Args()[0] {
-	case "info":
+	case "report":
-		var path string
+		opts := &reportOpts{}
-		fs := flag.NewFlagSet("info", flag.ExitOnError)
+		fs := flag.NewFlagSet("report", flag.ExitOnError)
-		fs.StringVar(&path, "dir", os.Getenv("HOME")+"/.influxdb", "Root storage path. [$HOME/.influxdb]")
+		fs.StringVar(&opts.pattern, "pattern", "", "Include only files matching a pattern")
+		fs.BoolVar(&opts.detailed, "detailed", false, "Report detailed cardinality estimates")
 
 		fs.Usage = func() {
-			println("Usage: influx_inspect info [options]\n\n   Displays series meta-data for all shards.")
+			println("Usage: influx_inspect report [options]\n\n   Displays shard level report")
 			println()
 			println("Options:")
 			fs.PrintDefaults()
@@ -48,7 +49,8 @@ func main() {
 			fmt.Printf("%v", err)
 			os.Exit(1)
 		}
-		cmdInfo(path)
+		opts.dir = fs.Arg(0)
+		cmdReport(opts)
 	case "dumptsmdev":
 		fmt.Fprintf(os.Stderr, "warning: dumptsmdev is deprecated, use dumptsm instead.\n")
 		fallthrough

cmd/influx_inspect/report.go

@@ -0,0 +1,139 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"text/tabwriter"
+	"time"
+
+	"github.com/influxdata/influxdb/models"
+	"github.com/influxdata/influxdb/tsdb/engine/tsm1"
+	"github.com/retailnext/hllpp"
+)
+
+type reportOpts struct {
+	dir      string
+	pattern  string
+	detailed bool
+}
+
+func cmdReport(opts *reportOpts) {
+	start := time.Now()
+
+	files, err := filepath.Glob(filepath.Join(opts.dir, fmt.Sprintf("*.%s", tsm1.TSMFileExtension)))
+	if err != nil {
+		fmt.Printf("%v\n", err)
+		os.Exit(1)
+	}
+
+	var filtered []string
+	if opts.pattern != "" {
+		for _, f := range files {
+			if strings.Contains(f, opts.pattern) {
+				filtered = append(filtered, f)
+			}
+		}
+		files = filtered
+	}
+
+	if len(files) == 0 {
+		fmt.Printf("no tsm files at %v\n", opts.dir)
+		os.Exit(1)
+	}
+
+	tw := tabwriter.NewWriter(os.Stdout, 8, 8, 1, '\t', 0)
+	fmt.Fprintln(tw, strings.Join([]string{"File", "Series"}, "\t"))
+
+	totalSeries := hllpp.New()
+	tagCardialities := map[string]*hllpp.HLLPP{}
+	measCardinalities := map[string]*hllpp.HLLPP{}
+	fieldCardinalities := map[string]*hllpp.HLLPP{}
+
+	ordering := make([]chan struct{}, 0, len(files))
+	for range files {
+		ordering = append(ordering, make(chan struct{}))
+	}
+
+	for _, f := range files {
+		file, err := os.OpenFile(f, os.O_RDONLY, 0600)
+		if err != nil {
+			fmt.Printf("%v", err)
+			os.Exit(1)
+		}
+
+		reader, err := tsm1.NewTSMReader(file)
+		if err != nil {
+			fmt.Printf("%v", err)
+			os.Exit(1)
+		}
+
+		seriesCount := reader.KeyCount()
+		for i := 0; i < seriesCount; i++ {
+			key, _ := reader.KeyAt(i)
+			totalSeries.Add([]byte(key))
+
+			if opts.detailed {
+				sep := strings.Index(key, "#!~#")
+				seriesKey, field := key[:sep], key[sep+4:]
+				measurement, tags, _ := models.ParseKey(seriesKey)
+
+				measCount, ok := measCardinalities[measurement]
+				if !ok {
+					measCount = hllpp.New()
+					measCardinalities[measurement] = measCount
+				}
+				measCount.Add([]byte(key))
+
+				fieldCount, ok := fieldCardinalities[measurement]
+				if !ok {
+					fieldCount = hllpp.New()
+					fieldCardinalities[measurement] = fieldCount
+				}
+				fieldCount.Add([]byte(field))
+
+				for t, v := range tags {
+					tagCount, ok := tagCardialities[t]
+					if !ok {
+						tagCount = hllpp.New()
+						tagCardialities[t] = tagCount
+					}
+					tagCount.Add([]byte(v))
+				}
+			}
+		}
+		reader.Close()
+
+		fmt.Fprintln(tw, strings.Join([]string{
+			filepath.Base(file.Name()),
+			strconv.FormatInt(int64(seriesCount), 10),
+		}, "\t"))
+		tw.Flush()
+	}
+
+	tw.Flush()
+	println()
+	fmt.Printf("Statistics\n")
+	fmt.Printf("  Series:\n")
+	fmt.Printf("    Total (est): %d\n", totalSeries.Count())
+	if opts.detailed {
+		fmt.Printf("  Measurements (est):\n")
+		for t, card := range measCardinalities {
+			fmt.Printf("    %v: %d (%d%%)\n", t, card.Count(), int((float64(card.Count())/float64(totalSeries.Count()))*100))
+		}
+
+		fmt.Printf("  Fields (est):\n")
+		for t, card := range fieldCardinalities {
+			fmt.Printf("    %v: %d\n", t, card.Count())
+		}
+
+		fmt.Printf("  Tags (est):\n")
+		for t, card := range tagCardialities {
+			fmt.Printf("    %v: %d\n", t, card.Count())
+		}
+	}
+
+	fmt.Printf("Completed in %s\n", time.Since(start))
+}