diff --git a/README.md b/README.md index f7b43538d..8a41a7c16 100644 --- a/README.md +++ b/README.md @@ -23,18 +23,16 @@ $ go install github.com/pilosa/pilosa/cmd/... Now run a single pilosa node with the default configuration: ```sh -pilosa +pilosa server ``` -If you would like to quickly create a multi-node pilosa cluster, see the `pilosactl create` documentation. - ## Configuration You can specify a configuration by setting the `-config` flag when running `pilosa`. ```sh -pilosa -config custom-config-file.cfg +pilosa server --config custom-config-file.cfg ``` The config file uses the [TOML](https://github.com/toml-lang/toml) configuration file format, @@ -54,6 +52,12 @@ host = "127.0.0.1:15000" host = "127.0.0.1:15001" ``` +You can generate a template config file with default values with: + +```sh +pilosa config +``` + The first two configuration options will be unique to each node in the cluster: `data-dir`: directory in which data is stored to disk @@ -244,6 +248,11 @@ Range(project=10, frame="collaboration", start="1970-01-01T00:00", end="2000-01- --- #### TopN() +``` +TopN(frame="geo") +``` +Returns all Bitmaps in the cache from frame `geo` sorted by the count of bits. + ``` TopN(frame="geo", n=20) ``` @@ -254,13 +263,17 @@ TopN(Bitmap(project=10, frame="collaboration"), frame="geo", n=20) ``` Returns the top 20 Bitmaps from `geo` sorted by the count of bits in the intersection with `Bitmap(project=10)`. - ``` TopN(Bitmap(project=10, frame="collaboration"), frame="geo", n=20, field="category", [81,82]) ``` +<<<<<<< HEAD Returns the top 20 Bitmaps from `geo`in attribute `category` with values `81 or 82` sorted by the count of bits in the intersection with `Bitmap(project=10)`. +======= +Returns the top 20 Bitmaps from `bar`in attribute `category` with values `81 or +82` sorted by the count of bits in the intersection with `Bitmap(id=10)`. +>>>>>>> master ## Development @@ -289,83 +302,3 @@ $ go install --ldflags="-X main.Version=1.0.0" ``` [Glide]: http://glide.sh/ - -## Pilosactl - -Pilosactl contains a suite of tools for interacting with pilosa. Run `pilosactl` for an overview of commands, and `pilosactl -h` for specific information on that command. - -### Create - -`pilosactl create` is used to create pilosa clusters. It has a number of options for controlling how the cluster is configured, what hosts it is on, and even the ability to build the pilosa binary locally and copy it to each cluster node automatically. To start pilosa on remote hosts, you only need `ssh` access to those hosts. See `pilosactl create -h` for a full list of options. - -Examples: - -Create a 5 node cluster locally (using 5 different ports), with a replication factor of 2. -``` -pilosactl create \ - -serverN 5 \ - -replicaN 2 -``` - -Create a cluster on 3 remote hosts - all logs will come to local stderr, pilosa binary must be available on remote hosts. The ssh user on the remote hosts needs to be the same as your local user. Otherwise use the `ssh-user` option. -``` -pilosactl create \ - -hosts="node1.example.com:15000,node2.example.com:15000,node3.example.com:15000" -``` - -Create a cluster on 3 remote hosts running OSX, but build the binary locally and copy it up. Stream the stderr of each node to a separate local log file. -``` -pilosactl create \ - -hosts="mac1.example.com:15000,mac2.example.com:15000,mac3.example.com:15000" \ - -copy-binary \ - -goos=darwin \ - -goarch=amd64 \ - -log-file-prefix=clusterlogs -``` - -### Bagent - -`pilosactl bagent` is what you want if you just want to run a simple benchmark against an existing cluster. Running it with no arguments will print some help, including the set of subcommands that it may be passed. Calling a subcommand with `-h'` will print the options for that subcommand. The `agent-num` flag can be passed an integer which can change the behavior the benchmarks that are run. This is useful when multiple invocations of the same benchmark are made by the `bspawn` command - they can each (for example) set different bits even though they all have the same arguments. - -E.G. -``` -pilosactl bagent \ - -hosts="localhost:15000,localhost:15001" \ - import -h -``` - -Multiple subcommands and their arguments may be concatenated at the command line and they will be run serially. This is useful (i.e.) for importing a bunch of data, and then executing queries against it. - -This will generate and import a bunch of data, and then execute random queries against it. - -``` -pilosactl bagent \ - -hosts="localhost:15000,localhost:15001" \ - import -max-bits-per-map=10000 \ - random-query -iterations 100 -``` - -### Bspawn -`pilosactl bspawn` allows you to automate the creation of clusters and the running of complex benchmarks which span multiple benchmark agents against them. It has a number of options which are described by `pilosactl bspawn` with no arguments, and also takes a config file which describes the Benchmark itself - this file is described below. - -#### Configuration Format - -The configuration file is a json object with the top level key `benchmarks`. This contains a list of objects each of which represents a `bagent` command (the `args` key) that will be run some number of times concurrently (the `num` key), and a `name` which should describe the overall effect that command. An example is below. -```json -{ - "benchmarks": [ - { - "num": 3, - "name": "set-diags", - "args": ["diagonal-set-bits", "-iterations", "30000", "-client-type", "round_robin"] - }, - { - "num": 2, - "name": "rand-plus-zipf", - "args": ["random-set-bits", "-iterations", "20000", "zipf", "-iterations", "100"] - } - ] -} -``` - -All of the benchmarks, and agents are run concurrently. Each agent will be passed an `agent-num` which can modify the behavior in a way that is benchmark specific. See the documentation for each benchmark to see how `agent-num` changes its behavior. diff --git a/attr.go b/attr.go index 1425676cf..c85f77cad 100644 --- a/attr.go +++ b/attr.go @@ -265,6 +265,8 @@ func txUpdateAttrs(tx *bolt.Tx, id uint64, m map[string]interface{}) (map[string attr[k] = uint64(v) case uint: attr[k] = uint64(v) + case float64: + attr[k] = uint64(v) case int64: attr[k] = uint64(v) case string, uint64, bool: diff --git a/bitmap.go b/bitmap.go index 55664b24b..2e27ded7a 100644 --- a/bitmap.go +++ b/bitmap.go @@ -174,6 +174,22 @@ func (b *Bitmap) InvalidateCount() { } } +//increment the bitmap cached counter, note this is an optimization that assumes that the caller is aware the size increased +func (b *Bitmap) IncrementCount(i uint64) { + seg := b.segment(i / SliceWidth) + if seg != nil { + seg.n++ + } +} +func (b *Bitmap) DecrementCount(i uint64) { + seg := b.segment(i / SliceWidth) + if seg != nil { + if seg.n > 0 { + seg.n-- + } + } +} + // Count returns the number of set bits in the bitmap. func (b *Bitmap) Count() uint64 { var n uint64 @@ -312,7 +328,6 @@ func (s *BitmapSegment) Difference(other *BitmapSegment) *BitmapSegment { // SetBit sets the i-th bit of the bitmap. func (s *BitmapSegment) SetBit(i uint64) (changed bool) { s.ensureWritable() - changed, _ = s.data.Add(i) if changed { s.n++ diff --git a/cache.go b/cache.go index a1cba3dec..7c2e96f2d 100644 --- a/cache.go +++ b/cache.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "sort" + "sync" "time" "github.com/golang/groupcache/lru" @@ -14,6 +15,7 @@ import ( // Cache represents a cache for bitmap counts. type Cache interface { Add(bitmapID uint64, n uint64) + BulkAdd(bitmapID uint64, n uint64) Get(bitmapID uint64) uint64 Len() int @@ -43,6 +45,10 @@ func NewLRUCache(maxEntries int) *LRUCache { return c } +func (c *LRUCache) BulkAdd(bitmapID, n uint64) { + c.Add(bitmapID, n) +} + // Add adds a bitmap to the cache. func (c *LRUCache) Add(bitmapID, n uint64) { c.cache.Add(bitmapID, n) @@ -92,6 +98,7 @@ var _ Cache = &LRUCache{} // RankCache represents a cache with sorted entries. type RankCache struct { + mu sync.Mutex entries map[uint64]uint64 rankings []BitmapPair // cached, ordered list @@ -112,33 +119,47 @@ func NewRankCache() *RankCache { // Add adds a bitmap to the cache. func (c *RankCache) Add(bitmapID uint64, n uint64) { + c.mu.Lock() + defer c.mu.Unlock() // Ignore if the bit count on the bitmap is below the threshold. if n < c.ThresholdValue { return } - // Add to cache. c.entries[bitmapID] = n - // If size is larger than the threshold then trim it. - if len(c.entries) > c.ThresholdLength { - c.update() - for id, n := range c.entries { - if n <= c.ThresholdValue { - delete(c.entries, id) - } - } + c.invalidate() +} + +// BulkAdd adds a bitmap to the cache unsorted. You should Invalidate after completion. +func (c *RankCache) BulkAdd(bitmapID uint64, n uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if n < c.ThresholdValue { + return } + + c.entries[bitmapID] = n } // Get returns a bitmap with a given id. -func (c *RankCache) Get(bitmapID uint64) uint64 { return c.entries[bitmapID] } +func (c *RankCache) Get(bitmapID uint64) uint64 { + c.mu.Lock() + defer c.mu.Unlock() + return c.entries[bitmapID] +} // Len returns the number of items in the cache. -func (c *RankCache) Len() int { return len(c.entries) } +func (c *RankCache) Len() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.entries) +} // BitmapIDs returns a list of all bitmap IDs in the cache. func (c *RankCache) BitmapIDs() []uint64 { + c.mu.Lock() + defer c.mu.Unlock() a := make([]uint64, 0, len(c.entries)) for id := range c.entries { a = append(a, id) @@ -147,22 +168,25 @@ func (c *RankCache) BitmapIDs() []uint64 { return a } -// Invalidate reorders the entries, if necessary. -func (c *RankCache) Invalidate() { - // Update if there aren't many items or it hasn't been updated recently. - if len(c.rankings) < 50 || (c.updateN > 0 && time.Since(c.updateTime) > 5*time.Minute) { - c.update() - } -} - // update reorders the entries by rank. -func (c *RankCache) update() { +func (c *RankCache) Invalidate() { + c.mu.Lock() + defer c.mu.Unlock() + c.invalidate() + +} +func (c *RankCache) invalidate() { + // Don't invalidate more than once every X seconds. + // TODO: consider making this configurable. + if time.Now().Sub(c.updateTime).Seconds() < 10 { + return + } // Convert cache to a sorted list. rankings := make([]BitmapPair, 0, len(c.entries)) - for id, n := range c.entries { + for id, cnt := range c.entries { rankings = append(rankings, BitmapPair{ ID: id, - Count: n, + Count: cnt, }) } sort.Sort(BitmapPairs(rankings)) @@ -177,6 +201,15 @@ func (c *RankCache) update() { // Reset counters. c.updateTime, c.updateN = time.Now(), 0 + + // If size is larger than the threshold then trim it. + if len(c.entries) > c.ThresholdLength { + for id, cnt := range c.entries { + if cnt <= c.ThresholdValue { + delete(c.entries, id) + } + } + } } // Top returns an ordered list of bitmaps. @@ -233,6 +266,26 @@ func (p Pairs) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p Pairs) Len() int { return len(p) } func (p Pairs) Less(i, j int) bool { return p[i].Count > p[j].Count } +type PairHeap struct { + Pairs +} + +func (p PairHeap) Less(i, j int) bool { return p.Pairs[i].Count < p.Pairs[j].Count } + +func (h *Pairs) Push(x interface{}) { + // Push and Pop use pointer receivers because they modify the slice's length, + // not just its contents. + *h = append(*h, x.(Pair)) +} + +func (h *Pairs) Pop() interface{} { + old := *h + n := len(old) + x := old[n-1] + *h = old[0 : n-1] + return x +} + // Add merges other into p and returns a new slice. func (p Pairs) Add(other []Pair) []Pair { // Create lookup of key/counts. @@ -327,3 +380,27 @@ func (p uint64Slice) merge(other []uint64) []uint64 { return ret } + +// BitmapCache provides an interface for caching full bitmaps. +type BitmapCache interface { + Fetch(id uint64) (*Bitmap, bool) + Add(id uint64, b *Bitmap) +} + +// SimpleCache implements BitmapCache +// it is meant to be a short-lived cache for cases where writes are continuing to access +// the same bit within a short time frame (i.e. good for write-heavy loads) +// A read-heavy use case would cause the cache to get bigger, potentially causing the +// node to run out of memory. +type SimpleCache struct { + cache map[uint64]*Bitmap +} + +func (s *SimpleCache) Fetch(id uint64) (*Bitmap, bool) { + m, ok := s.cache[id] + return m, ok +} + +func (s *SimpleCache) Add(id uint64, b *Bitmap) { + s.cache[id] = b +} diff --git a/cmd/backup.go b/cmd/backup.go new file mode 100644 index 000000000..418f6a720 --- /dev/null +++ b/cmd/backup.go @@ -0,0 +1,42 @@ +package cmd + +import ( + "context" + "fmt" + "log" + "os" + + "github.com/spf13/cobra" + "github.com/spf13/viper" + + "github.com/pilosa/pilosa/ctl" +) + +var backuper = ctl.NewBackupCommand(os.Stdin, os.Stdout, os.Stderr) + +var backupCmd = &cobra.Command{ + Use: "backup", + Short: "Backup data from pilosa.", + Long: ` +Backs up the database and frame from across the cluster into a single file. +`, + Run: func(cmd *cobra.Command, args []string) { + if err := backuper.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + backupCmd.Flags().StringVarP(&backuper.Host, "host", "", "localhost:15000", "host:port of Pilosa.") + backupCmd.Flags().StringVarP(&backuper.Database, "database", "d", "", "Pilosa database to backup into.") + backupCmd.Flags().StringVarP(&backuper.Frame, "frame", "f", "", "Frame to backup into.") + backupCmd.Flags().StringVarP(&backuper.Path, "output-file", "o", "", "File to write backup to - default stdout") + + err := viper.BindPFlags(backupCmd.Flags()) + if err != nil { + log.Fatalf("Error binding backup flags: %v", err) + } + + RootCmd.AddCommand(backupCmd) +} diff --git a/cmd/bench.go b/cmd/bench.go new file mode 100644 index 000000000..3e093d68e --- /dev/null +++ b/cmd/bench.go @@ -0,0 +1,43 @@ +package cmd + +import ( + "context" + "fmt" + "log" + "os" + + "github.com/spf13/cobra" + "github.com/spf13/viper" + + "github.com/pilosa/pilosa/ctl" +) + +var bencher = ctl.NewBenchCommand(os.Stdin, os.Stdout, os.Stderr) + +var benchCmd = &cobra.Command{ + Use: "bench", + Short: "Benchmark operations.", + Long: ` +Executes a benchmark for a given operation against the database. +`, + Run: func(cmd *cobra.Command, args []string) { + if err := bencher.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + benchCmd.Flags().StringVarP(&bencher.Host, "host", "", "localhost:15000", "host:port of Pilosa.") + benchCmd.Flags().StringVarP(&bencher.Database, "database", "d", "", "Pilosa database to benchmark.") + benchCmd.Flags().StringVarP(&bencher.Frame, "frame", "f", "", "Frame to benchmark.") + benchCmd.Flags().StringVarP(&bencher.Op, "operation", "o", "set-bit", "Operation to perform: choose from [set-bit]") + benchCmd.Flags().IntVarP(&bencher.N, "num", "n", 0, "Number of operations to perform.") + + err := viper.BindPFlags(benchCmd.Flags()) + if err != nil { + log.Fatalf("Error binding bench flags: %v", err) + } + + RootCmd.AddCommand(benchCmd) +} diff --git a/cmd/check.go b/cmd/check.go new file mode 100644 index 000000000..560130187 --- /dev/null +++ b/cmd/check.go @@ -0,0 +1,35 @@ +package cmd + +import ( + "context" + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/pilosa/pilosa/ctl" +) + +var checker = ctl.NewCheckCommand(os.Stdin, os.Stdout, os.Stderr) + +var checkCmd = &cobra.Command{ + Use: "check [path2]...", + Short: "Do a consistency check on a pilosa data file.", + Long: ` +Performs a consistency check on data files. +`, + Run: func(cmd *cobra.Command, args []string) { + if len(args) == 0 { + fmt.Println("path required") + return + } + checker.Paths = args + if err := checker.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + RootCmd.AddCommand(checkCmd) +} diff --git a/cmd/config.go b/cmd/config.go new file mode 100644 index 000000000..5fccbfc7d --- /dev/null +++ b/cmd/config.go @@ -0,0 +1,29 @@ +package cmd + +import ( + "context" + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/pilosa/pilosa/ctl" +) + +var conf = ctl.NewConfigCommand(os.Stdin, os.Stdout, os.Stderr) + +var confCmd = &cobra.Command{ + Use: "config", + Short: "Print the default configuration.", + Long: `config prints the default configuration to stdout +`, + Run: func(cmd *cobra.Command, args []string) { + if err := conf.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + RootCmd.AddCommand(confCmd) +} diff --git a/cmd/export.go b/cmd/export.go new file mode 100644 index 000000000..06a76118d --- /dev/null +++ b/cmd/export.go @@ -0,0 +1,49 @@ +package cmd + +import ( + "context" + "fmt" + "log" + "os" + + "github.com/spf13/cobra" + "github.com/spf13/viper" + + "github.com/pilosa/pilosa/ctl" +) + +var exporter = ctl.NewExportCommand(os.Stdin, os.Stdout, os.Stderr) + +var exportCmd = &cobra.Command{ + Use: "export", + Short: "Export data from pilosa.", + Long: ` +Bulk exports a fragment to a CSV file. If the OUTFILE is not specified then +the output is written to STDOUT. + +The format of the CSV file is: + + BITMAPID,PROFILEID + +The file does not contain any headers. +`, + Run: func(cmd *cobra.Command, args []string) { + if err := exporter.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + exportCmd.Flags().StringVarP(&exporter.Host, "host", "", "localhost:15000", "host:port of Pilosa.") + exportCmd.Flags().StringVarP(&exporter.Database, "database", "d", "", "Pilosa database to export into.") + exportCmd.Flags().StringVarP(&exporter.Frame, "frame", "f", "", "Frame to export into.") + exportCmd.Flags().StringVarP(&exporter.Path, "output-file", "o", "", "File to write export to - default stdout") + + err := viper.BindPFlags(exportCmd.Flags()) + if err != nil { + log.Fatalf("Error binding export flags: %v", err) + } + + RootCmd.AddCommand(exportCmd) +} diff --git a/cmd/import.go b/cmd/import.go new file mode 100644 index 000000000..5f878c431 --- /dev/null +++ b/cmd/import.go @@ -0,0 +1,50 @@ +package cmd + +import ( + "context" + "fmt" + "log" + "os" + + "github.com/spf13/cobra" + "github.com/spf13/viper" + + "github.com/pilosa/pilosa/ctl" +) + +var importer = ctl.NewImportCommand(os.Stdin, os.Stdout, os.Stderr) + +var importCmd = &cobra.Command{ + Use: "import", + Short: "Bulk load data into pilosa.", + Long: `Bulk imports one or more CSV files to a host's database and frame. The bits +of the CSV file are grouped by slice for the most efficient import. + +The format of the CSV file is: + + BITMAPID,PROFILEID,[TIME] + +The file should contain no headers. The TIME column is optional and can be +omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. +`, + Run: func(cmd *cobra.Command, args []string) { + importer.Paths = args + if err := importer.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + importCmd.Flags().StringVarP(&importer.Host, "host", "", "localhost:15000", "host:port of Pilosa.") + importCmd.Flags().StringVarP(&importer.Database, "database", "d", "", "Pilosa database to import into.") + importCmd.Flags().StringVarP(&importer.Frame, "frame", "f", "", "Frame to import into.") + importCmd.Flags().IntVarP(&importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.") + + err := viper.BindPFlags(importCmd.Flags()) + if err != nil { + log.Fatalf("Error binding import flags: %v", err) + } + + RootCmd.AddCommand(importCmd) +} diff --git a/cmd/inspect.go b/cmd/inspect.go new file mode 100644 index 000000000..666c59d5c --- /dev/null +++ b/cmd/inspect.go @@ -0,0 +1,38 @@ +package cmd + +import ( + "context" + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/pilosa/pilosa/ctl" +) + +var inspecter = ctl.NewInspectCommand(os.Stdin, os.Stdout, os.Stderr) + +var inspectCmd = &cobra.Command{ + Use: "inspect", + Short: "Get stats on a pilosa data file.", + Long: ` +Inspects a data file and provides stats. +`, + Run: func(cmd *cobra.Command, args []string) { + if len(args) == 0 { + fmt.Println("path required") + return + } else if len(args) > 1 { + fmt.Println("only one path allowed") + return + } + inspecter.Path = args[0] + if err := inspecter.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + RootCmd.AddCommand(inspectCmd) +} diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index 4f7474996..c3bef0622 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -1,197 +1,15 @@ package main import ( - "errors" - "flag" "fmt" - "io" - "math/rand" "os" - "os/signal" - "path/filepath" - "runtime/pprof" - "strings" - "time" - "github.com/BurntSushi/toml" - "github.com/pilosa/pilosa" -) - -// Version and BuildTime hold the version/build time information passed in at compile time. -var ( - Version string - BuildTime string -) - -func init() { - if Version == "" { - Version = "v0.0.0" - } - if BuildTime == "" { - BuildTime = "not recorded" - } - - rand.Seed(time.Now().UTC().UnixNano()) -} - -const ( - // DefaultDataDir is the default data directory. - DefaultDataDir = "~/.pilosa" + "github.com/pilosa/pilosa/cmd" ) func main() { - m := NewMain() - m.Server.Handler.Version = Version - fmt.Fprintf(m.Stderr, "Pilosa %s, build time %s\n", Version, BuildTime) - - // Parse command line arguments. - if err := m.ParseFlags(os.Args[1:]); err != nil { - fmt.Fprintln(m.Stderr, err) - os.Exit(2) - } - - // Start CPU profiling. - if m.CPUProfile != "" { - f, err := os.Create(m.CPUProfile) - if err != nil { - fmt.Fprintf(m.Stderr, "create cpu profile: %v", err) - os.Exit(1) - } - defer f.Close() - - fmt.Fprintln(m.Stderr, "Starting cpu profile") - pprof.StartCPUProfile(f) - time.AfterFunc(m.CPUTime, func() { - fmt.Fprintln(m.Stderr, "Stopping cpu profile") - pprof.StopCPUProfile() - f.Close() - }) - } - - // Execute the program. - if err := m.Run(); err != nil { - fmt.Fprintln(m.Stderr, err) - fmt.Fprintln(m.Stderr, "stopping profile") - os.Exit(1) - } - - // First SIGKILL causes server to shut down gracefully. - c := make(chan os.Signal, 2) - signal.Notify(c, os.Interrupt) - sig := <-c - fmt.Fprintf(m.Stderr, "Received %s; gracefully shutting down...\n", sig.String()) - - // Second signal causes a hard shutdown. - go func() { <-c; os.Exit(1) }() - - if err := m.Close(); err != nil { - fmt.Fprintln(m.Stderr, err) + if err := cmd.RootCmd.Execute(); err != nil { + fmt.Println(err) os.Exit(1) } } - -// Main represents the main program execution. -type Main struct { - Server *pilosa.Server - - // Configuration options. - ConfigPath string - Config *pilosa.Config - - // Profiling options. - CPUProfile string - CPUTime time.Duration - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewMain returns a new instance of Main. -func NewMain() *Main { - return &Main{ - Server: pilosa.NewServer(), - Config: pilosa.NewConfig(), - - Stdin: os.Stdin, - Stdout: os.Stdout, - Stderr: os.Stderr, - } -} - -// Run executes the main program execution. -func (m *Main) Run(args ...string) error { - // Notify user of config file. - if m.ConfigPath != "" { - fmt.Fprintf(m.Stdout, "Using config: %s\n", m.ConfigPath) - } - - // Setup logging output. - m.Server.LogOutput = m.Stderr - - // Configure index. - fmt.Fprintf(m.Stderr, "Using data from: %s\n", m.Config.DataDir) - m.Server.Index.Path = m.Config.DataDir - m.Server.Index.Stats = pilosa.NewExpvarStatsClient() - - // Build cluster from config file. - m.Server.Host = m.Config.Host - m.Server.Cluster = m.Config.PilosaCluster() - - // Set configuration options. - m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval) - - // Initialize server. - if err := m.Server.Open(); err != nil { - return err - } - - fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Server.Host) - - return nil -} - -// Close shuts down the server. -func (m *Main) Close() error { - return m.Server.Close() -} - -// ParseFlags parses command line flags from args. -func (m *Main) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosa", flag.ContinueOnError) - fs.StringVar(&m.CPUProfile, "cpuprofile", "", "cpu profile") - fs.DurationVar(&m.CPUTime, "cputime", 30*time.Second, "cpu profile duration") - fs.StringVar(&m.ConfigPath, "config", "", "config path") - fs.SetOutput(m.Stderr) - if err := fs.Parse(args); err != nil { - return err - } - - // Load config, if specified. - if m.ConfigPath != "" { - if _, err := toml.DecodeFile(m.ConfigPath, &m.Config); err != nil { - return err - } - } - - // Use default data directory if one is not specified. - if m.Config.DataDir == "" { - m.Config.DataDir = DefaultDataDir - } - - // Expand home directory. - prefix := "~" + string(filepath.Separator) - if strings.HasPrefix(m.Config.DataDir, prefix) { - // u, err := user.Current() - HomeDir := os.Getenv("HOME") - /*if err != nil { - return err - } else*/if HomeDir == "" { - return errors.New("data directory not specified and no home dir available") - } - m.Config.DataDir = filepath.Join(HomeDir, strings.TrimPrefix(m.Config.DataDir, prefix)) - } - - return nil -} diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go deleted file mode 100644 index 88fa6fea5..000000000 --- a/cmd/pilosactl/main.go +++ /dev/null @@ -1,1028 +0,0 @@ -package main - -import ( - "bufio" - "context" - "encoding/csv" - "errors" - "flag" - "fmt" - "io" - "io/ioutil" - "log" - "math/rand" - "os" - "path/filepath" - "sort" - "strconv" - "strings" - "syscall" - "text/tabwriter" - "time" - "unsafe" - - "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/pilosactl" - "github.com/pilosa/pilosa/roaring" -) - -var ( - // ErrUnknownCommand is returned when specifying an unknown command. - ErrUnknownCommand = errors.New("unknown command") - - // ErrPathRequired is returned when executing a command without a required path. - ErrPathRequired = errors.New("path required") - Version string - BuildTime string -) - -func init() { - if Version == "" { - Version = "v0.0.0" - } - if BuildTime == "" { - BuildTime = "not recorded" - } -} - -func main() { - m := NewMain() - - fmt.Fprintf(m.Stderr, "Pilosactl %s, build time %s\n", Version, BuildTime) - - // Parse command line arguments. - if err := m.ParseFlags(os.Args[1:]); err == flag.ErrHelp { - os.Exit(2) - } else if err != nil { - fmt.Fprintln(m.Stderr, err) - os.Exit(2) - } - - // Execute the program. - if err := m.Run(); err != nil { - fmt.Fprintln(m.Stderr, err) - os.Exit(1) - } -} - -// Main represents the main program execution. -type Main struct { - // Subcommand to execute. - Cmd Command - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewMain returns a new instance of Main. -func NewMain() *Main { - return &Main{ - Stdin: os.Stdin, - Stdout: os.Stdout, - Stderr: os.Stderr, - } -} - -// Usage returns the usage message to be printed. -func (m *Main) Usage() string { - return strings.TrimSpace(` -Pilosactl is a tool for interacting with a pilosa server. - -Usage: - - pilosactl command [arguments] - -The commands are: - - config prints the default configuration - import imports data from a CSV file - export exports data to a CSV file - sort sorts a data file for optimal import speed - backup backs up a frame to an archive file - restore restores a frame from an archive file - inspect inspects fragment data files - check performs a consistency check of data files - bench benchmarks operations - -Use the "-h" flag with any command for more information. -`) -} - -// Run executes the main program execution. -func (m *Main) Run() error { return m.Cmd.Run(context.Background()) } - -// ParseFlags parses command line flags from args. -func (m *Main) ParseFlags(args []string) error { - var command string - if len(args) > 0 { - command = args[0] - args = args[1:] - } - - switch command { - case "", "help", "-h": - fmt.Fprintln(m.Stderr, m.Usage()) - fmt.Fprintln(m.Stderr, "") - return flag.ErrHelp - case "config": - m.Cmd = NewConfigCommand(m.Stdin, m.Stdout, m.Stderr) - case "import": - m.Cmd = pilosactl.NewImportCommand(m.Stdin, m.Stdout, m.Stderr) - case "export": - m.Cmd = NewExportCommand(m.Stdin, m.Stdout, m.Stderr) - case "sort": - m.Cmd = NewSortCommand(m.Stdin, m.Stdout, m.Stderr) - case "backup": - m.Cmd = NewBackupCommand(m.Stdin, m.Stdout, m.Stderr) - case "restore": - m.Cmd = NewRestoreCommand(m.Stdin, m.Stdout, m.Stderr) - case "inspect": - m.Cmd = NewInspectCommand(m.Stdin, m.Stdout, m.Stderr) - case "check": - m.Cmd = NewCheckCommand(m.Stdin, m.Stdout, m.Stderr) - case "bench": - m.Cmd = NewBenchCommand(m.Stdin, m.Stdout, m.Stderr) - default: - return ErrUnknownCommand - } - - // Parse command's flags. - if err := m.Cmd.ParseFlags(args); err == flag.ErrHelp { - fmt.Fprintln(m.Stderr, m.Cmd.Usage()) - fmt.Fprintln(m.Stderr, "") - return err - } else if err != nil { - return err - } - - return nil -} - -// Command represents an executable subcommand. -type Command interface { - Usage() string - ParseFlags(args []string) error - Run(context.Context) error -} - -// ConfigCommand represents a command for printing a default config. -type ConfigCommand struct { - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewConfigCommand returns a new instance of ConfigCommand. -func NewConfigCommand(stdin io.Reader, stdout, stderr io.Writer) *ConfigCommand { - return &ConfigCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *ConfigCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - if err := fs.Parse(args); err != nil { - return err - } - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *ConfigCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl config - -Prints the default configuration file to standard out. -`) -} - -// Run executes the main program execution. -func (cmd *ConfigCommand) Run(ctx context.Context) error { - fmt.Fprintln(cmd.Stdout, strings.TrimSpace(` -data-dir = "~/.pilosa" -host = "localhost:15000" - -[cluster] -replicas = 1 - -[[cluster.node]] -host = "localhost:15000" - -[plugins] -path = "" -`)+"\n") - return nil -} - -// ExportCommand represents a command for bulk exporting data from a server. -type ExportCommand struct { - // Remote host and port. - Host string - - // Name of the database & frame to export from. - Database string - Frame string - - // Filename to export to. - Path string - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewExportCommand returns a new instance of ExportCommand. -func NewExportCommand(stdin io.Reader, stdout, stderr io.Writer) *ExportCommand { - return &ExportCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *ExportCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - fs.StringVar(&cmd.Host, "host", "localhost:15000", "host:port") - fs.StringVar(&cmd.Database, "d", "", "database") - fs.StringVar(&cmd.Frame, "f", "", "frame") - fs.StringVar(&cmd.Path, "o", "", "output file") - if err := fs.Parse(args); err != nil { - return err - } - - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *ExportCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl export -host HOST -d database -f frame -o OUTFILE - -Bulk exports a fragment to a CSV file. If the OUTFILE is not specified then -the output is written to STDOUT. - -The format of the CSV file is: - - BITMAPID,PROFILEID - -The file does not contain any headers. -`) -} - -// Run executes the main program execution. -func (cmd *ExportCommand) Run(ctx context.Context) error { - logger := log.New(cmd.Stderr, "", log.LstdFlags) - - // Validate arguments. - if cmd.Database == "" { - return pilosa.ErrDatabaseRequired - } else if cmd.Frame == "" { - return pilosa.ErrFrameRequired - } - - // Use output file, if specified. - // Otherwise use STDOUT. - var w io.Writer = cmd.Stdout - if cmd.Path != "" { - f, err := os.Create(cmd.Path) - if err != nil { - return err - } - defer f.Close() - - w = f - } - - // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) - if err != nil { - return err - } - - // Determine slice count. - maxSlices, err := client.MaxSliceByDatabase(ctx) - if err != nil { - return err - } - - // Export each slice. - for slice := uint64(0); slice <= maxSlices[cmd.Database]; slice++ { - logger.Printf("exporting slice: %d", slice) - if err := client.ExportCSV(ctx, cmd.Database, cmd.Frame, slice, w); err != nil { - return err - } - } - - // Close writer, if applicable. - if w, ok := w.(io.Closer); ok { - if err := w.Close(); err != nil { - return err - } - } - - return nil -} - -// SortCommand represents a command for sorting import data. -type SortCommand struct { - // Filename to sort - Path string - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewSortCommand returns a new instance of SortCommand. -func NewSortCommand(stdin io.Reader, stdout, stderr io.Writer) *SortCommand { - return &SortCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *SortCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - if err := fs.Parse(args); err != nil { - return err - } - - // Extract the data path. - if fs.NArg() == 0 { - return errors.New("path required") - } else if fs.NArg() > 1 { - return errors.New("only one path allowed") - } - cmd.Path = fs.Arg(0) - - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *SortCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl sort PATH - -Sorts the import data at PATH into the optimal sort order for importing. - -The format of the CSV file is: - - BITMAPID,PROFILEID - -The file should contain no headers. -`) -} - -// Run executes the main program execution. -func (cmd *SortCommand) Run(ctx context.Context) error { - // Open file for reading. - f, err := os.Open(cmd.Path) - if err != nil { - return err - } - defer f.Close() - - // Read rows as bits. - r := csv.NewReader(f) - r.FieldsPerRecord = -1 - a := make([]pilosa.Bit, 0, 1000000) - for { - bitmapID, profileID, timestamp, err := readCSVRow(r) - if err == io.EOF { - break - } else if err == errBlank { - continue - } else if err != nil { - return err - } - a = append(a, pilosa.Bit{BitmapID: bitmapID, ProfileID: profileID, Timestamp: timestamp}) - } - - // Sort bits by position. - sort.Sort(pilosa.BitsByPos(a)) - - // Rewrite to STDOUT. - w := bufio.NewWriter(cmd.Stdout) - buf := make([]byte, 0, 1024) - for _, bit := range a { - // Write CSV to buffer. - buf = buf[:0] - buf = strconv.AppendUint(buf, bit.BitmapID, 10) - - buf = append(buf, ',') - buf = strconv.AppendUint(buf, bit.ProfileID, 10) - - if bit.Timestamp != 0 { - buf = append(buf, ',') - buf = append(buf, time.Unix(0, bit.Timestamp).UTC().Format(pilosa.TimeFormat)...) - } - - buf = append(buf, '\n') - - // Write to output. - if _, err := w.Write(buf); err != nil { - return err - } - } - - // Ensure buffer is flushed before exiting. - if err := w.Flush(); err != nil { - return err - } - - return nil -} - -// BackupCommand represents a command for backing up a frame. -type BackupCommand struct { - // Destination host and port. - Host string - - // Name of the database & frame to backup. - Database string - Frame string - - // Output file to write to. - Path string - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewBackupCommand returns a new instance of BackupCommand. -func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand { - return &BackupCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *BackupCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - fs.StringVar(&cmd.Host, "host", "localhost:15000", "host:port") - fs.StringVar(&cmd.Database, "d", "", "database") - fs.StringVar(&cmd.Frame, "f", "", "frame") - fs.StringVar(&cmd.Path, "o", "", "output file") - if err := fs.Parse(args); err != nil { - return err - } - - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *BackupCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl backup -host HOST -d database -f frame -o PATH - -Backs up the database and frame from across the cluster into a single file. -`) -} - -// Run executes the main program execution. -func (cmd *BackupCommand) Run(ctx context.Context) error { - // Validate arguments. - if cmd.Path == "" { - return errors.New("output file required") - } - - // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) - if err != nil { - return err - } - - // Open output file. - f, err := os.Create(cmd.Path) - if err != nil { - return err - } - defer f.Close() - - // Begin streaming backup. - if err := client.BackupTo(ctx, f, cmd.Database, cmd.Frame); err != nil { - return err - } - - // Sync & close file to ensure durability. - if err := f.Sync(); err != nil { - return err - } else if err = f.Close(); err != nil { - return err - } - - return nil -} - -// RestoreCommand represents a command for restoring a frame from a backup. -type RestoreCommand struct { - // Destination host and port. - Host string - - // Name of the database & frame to backup. - Database string - Frame string - - // Import file to read from. - Path string - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewRestoreCommand returns a new instance of RestoreCommand. -func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand { - return &RestoreCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *RestoreCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - fs.StringVar(&cmd.Host, "host", "localhost:15000", "host:port") - fs.StringVar(&cmd.Database, "d", "", "database") - fs.StringVar(&cmd.Frame, "f", "", "frame") - if err := fs.Parse(args); err != nil { - return err - } - - // Read input path from the args. - if fs.NArg() == 0 { - return errors.New("path required") - } else if fs.NArg() > 1 { - return errors.New("too many paths specified") - } - cmd.Path = fs.Arg(0) - - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *RestoreCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl restore -host HOST -d database -f frame PATH - -Restores a frame to the cluster from a backup file. -`) -} - -// Run executes the main program execution. -func (cmd *RestoreCommand) Run(ctx context.Context) error { - // Validate arguments. - if cmd.Path == "" { - return errors.New("backup file required") - } - - // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) - if err != nil { - return err - } - - // Open backup file. - f, err := os.Open(cmd.Path) - if err != nil { - return err - } - defer f.Close() - - // Restore backup file to the cluster. - if err := client.RestoreFrom(ctx, f, cmd.Database, cmd.Frame); err != nil { - return err - } - - return nil -} - -// InspectCommand represents a command for inspecting fragment data files. -type InspectCommand struct { - // Path to data file - Path string - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewInspectCommand returns a new instance of InspectCommand. -func NewInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *InspectCommand { - return &InspectCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *InspectCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - if err := fs.Parse(args); err != nil { - return err - } - - // Parse path. - if fs.NArg() == 0 { - return errors.New("path required") - } else if fs.NArg() > 1 { - return errors.New("only one path allowed") - } - cmd.Path = fs.Arg(0) - - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *InspectCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl inspect PATH - -Inspects a data file and provides stats. - -`) -} - -// Run executes the main program execution. -func (cmd *InspectCommand) Run(ctx context.Context) error { - // Open file handle. - f, err := os.Open(cmd.Path) - if err != nil { - return err - } - defer f.Close() - - fi, err := f.Stat() - if err != nil { - return err - } - - // Memory map the file. - data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED) - if err != nil { - return err - } - defer syscall.Munmap(data) - - // Attach the mmap file to the bitmap. - t := time.Now() - fmt.Fprintf(cmd.Stderr, "unmarshaling bitmap...") - bm := roaring.NewBitmap() - if err := bm.UnmarshalBinary(data); err != nil { - return err - } - fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) - - // Retrieve stats. - t = time.Now() - fmt.Fprintf(cmd.Stderr, "calculating stats...") - info := bm.Info() - fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) - - // Print top-level info. - fmt.Fprintf(cmd.Stdout, "== Bitmap Info ==\n") - fmt.Fprintf(cmd.Stdout, "Containers: %d\n", len(info.Containers)) - fmt.Fprintf(cmd.Stdout, "Operations: %d\n", info.OpN) - fmt.Fprintln(cmd.Stdout, "") - - // Print info for each container. - fmt.Fprintln(cmd.Stdout, "== Containers ==") - tw := tabwriter.NewWriter(cmd.Stdout, 0, 8, 0, '\t', 0) - fmt.Fprintf(tw, "%s\t%s\t% 8s \t% 8s\t%s\n", "KEY", "TYPE", "N", "ALLOC", "OFFSET") - for _, ci := range info.Containers { - fmt.Fprintf(tw, "%d\t%s\t% 8d \t% 8d \t0x%08x\n", - ci.Key, - ci.Type, - ci.N, - ci.Alloc, - uintptr(ci.Pointer)-uintptr(unsafe.Pointer(&data[0])), - ) - } - tw.Flush() - - return nil -} - -// CheckCommand represents a command for performing consistency checks on data files. -type CheckCommand struct { - // Data file paths. - Paths []string - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewCheckCommand returns a new instance of CheckCommand. -func NewCheckCommand(stdin io.Reader, stdout, stderr io.Writer) *CheckCommand { - return &CheckCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *CheckCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - if err := fs.Parse(args); err != nil { - return err - } - - // Parse path. - if fs.NArg() == 0 { - return errors.New("path required") - } - cmd.Paths = fs.Args() - - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *CheckCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl check PATHS... - -Performs a consistency check on data files. - -`) -} - -// Run executes the main program execution. -func (cmd *CheckCommand) Run(ctx context.Context) error { - for _, path := range cmd.Paths { - switch filepath.Ext(path) { - case "": - if err := cmd.checkBitmapFile(path); err != nil { - return err - } - - case ".cache": - if err := cmd.checkCacheFile(path); err != nil { - return err - } - - case ".snapshotting": - if err := cmd.checkSnapshotFile(path); err != nil { - return err - } - } - } - - return nil -} - -// checkBitmapFile performs a consistency check on path for a roaring bitmap file. -func (cmd *CheckCommand) checkBitmapFile(path string) error { - // Open file handle. - f, err := os.Open(path) - if err != nil { - return err - } - defer f.Close() - - fi, err := f.Stat() - if err != nil { - return err - } - - // Memory map the file. - data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED) - if err != nil { - return err - } - defer syscall.Munmap(data) - - // Attach the mmap file to the bitmap. - bm := roaring.NewBitmap() - if err := bm.UnmarshalBinary(data); err != nil { - return err - } - - // Perform consistency check. - if err := bm.Check(); err != nil { - // Print returned errors. - switch err := err.(type) { - case roaring.ErrorList: - for i := range err { - fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err[i].Error()) - } - default: - fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err.Error()) - } - } - - // Print success message if no errors were found. - fmt.Fprintf(cmd.Stdout, "%s: ok\n", path) - - return nil -} - -// checkCacheFile performs a consistency check on path for a cache file. -func (cmd *CheckCommand) checkCacheFile(path string) error { - fmt.Fprintf(cmd.Stderr, "%s: ignoring cache file\n", path) - return nil -} - -// checkSnapshotFile performs a consistency check on path for a snapshot file. -func (cmd *CheckCommand) checkSnapshotFile(path string) error { - fmt.Fprintf(cmd.Stderr, "%s: ignoring snapshot file\n", path) - return nil -} - -// BenchCommand represents a command for benchmarking database operations. -type BenchCommand struct { - // Destination host and port. - Host string - - // Name of the database & frame to execute against. - Database string - Frame string - - // Type of operation and number to execute. - Op string - N int - - // Standard input/output - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer -} - -// NewBenchCommand returns a new instance of BenchCommand. -func NewBenchCommand(stdin io.Reader, stdout, stderr io.Writer) *BenchCommand { - return &BenchCommand{ - Stdin: stdin, - Stdout: stdout, - Stderr: stderr, - } -} - -// ParseFlags parses command line flags from args. -func (cmd *BenchCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - fs.StringVar(&cmd.Host, "host", "localhost:15000", "host:port") - fs.StringVar(&cmd.Database, "d", "", "database") - fs.StringVar(&cmd.Frame, "f", "", "frame") - fs.StringVar(&cmd.Op, "op", "", "operation") - fs.IntVar(&cmd.N, "n", 0, "op count") - - if err := fs.Parse(args); err != nil { - return err - } - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *BenchCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl bench [args] - -Executes a benchmark for a given operation against the database. - -The following flags are allowed: - - -host HOSTPORT - hostname and port of running pilosa server - - -d DATABASE - database to execute operation against - - -f FRAME - frame to execute operation against - - -op OP - name of operation to execute - - -n COUNT - number of iterations to execute - -The following operations are available: - - set-bit - Sets a single random bit on the frame - -`) -} - -// Run executes the main program execution. -func (cmd *BenchCommand) Run(ctx context.Context) error { - // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) - if err != nil { - return err - } - - switch cmd.Op { - case "set-bit": - return cmd.runSetBit(ctx, client) - case "": - return errors.New("op required") - default: - return fmt.Errorf("unknown bench op: %q", cmd.Op) - } -} - -// runSetBit executes a benchmark of random SetBit() operations. -func (cmd *BenchCommand) runSetBit(ctx context.Context, client *pilosa.Client) error { - if cmd.N == 0 { - return errors.New("operation count required") - } else if cmd.Database == "" { - return pilosa.ErrDatabaseRequired - } else if cmd.Frame == "" { - return pilosa.ErrFrameRequired - } - - const maxBitmapID = 1000 - const maxProfileID = 100000 - - startTime := time.Now() - - // Execute operation continuously. - for i := 0; i < cmd.N; i++ { - bitmapID := rand.Intn(maxBitmapID) - profileID := rand.Intn(maxProfileID) - - q := fmt.Sprintf(`SetBit(id=%d, frame="%s", profileID=%d)`, bitmapID, cmd.Frame, profileID) - - if _, err := client.ExecuteQuery(ctx, cmd.Database, q, true); err != nil { - return err - } - } - - // Print results. - elapsed := time.Since(startTime) - fmt.Fprintf(cmd.Stdout, "Executed %d operations in %s (%0.3f op/sec)\n", cmd.N, elapsed, float64(cmd.N)/elapsed.Seconds()) - - return nil -} - -// readCSVRow reads a bitmap/profile pair from a CSV row. -func readCSVRow(r *csv.Reader) (bitmapID, profileID uint64, timestamp int64, err error) { - // Read CSV row. - record, err := r.Read() - if err != nil { - return 0, 0, 0, err - } - - // Ignore blank rows. - if record[0] == "" { - return 0, 0, 0, errBlank - } else if len(record) < 2 { - return 0, 0, 0, fmt.Errorf("bad column count: %d", len(record)) - } - - // Parse bitmap id. - bitmapID, err = strconv.ParseUint(record[0], 10, 64) - if err != nil { - return 0, 0, 0, fmt.Errorf("invalid bitmap id: %q", record[0]) - } - - // Parse bitmap id. - profileID, err = strconv.ParseUint(record[1], 10, 64) - if err != nil { - return 0, 0, 0, fmt.Errorf("invalid profile id: %q", record[1]) - } - - // Parse timestamp, if available. - if len(record) > 2 && record[2] != "" { - t, err := time.Parse(pilosa.TimeFormat, record[2]) - if err != nil { - return 0, 0, 0, fmt.Errorf("invalid timestamp: %q", record[2]) - } - timestamp = t.UnixNano() - } - - return bitmapID, profileID, timestamp, nil -} - -// errBlank indicates a blank row in a CSV file. -var errBlank = errors.New("blank row") diff --git a/cmd/pilosactl/main_test.go b/cmd/pilosactl/main_test.go deleted file mode 100644 index 0fee6f5dc..000000000 --- a/cmd/pilosactl/main_test.go +++ /dev/null @@ -1 +0,0 @@ -package main_test diff --git a/cmd/restore.go b/cmd/restore.go new file mode 100644 index 000000000..399122ff3 --- /dev/null +++ b/cmd/restore.go @@ -0,0 +1,42 @@ +package cmd + +import ( + "context" + "fmt" + "log" + "os" + + "github.com/spf13/cobra" + "github.com/spf13/viper" + + "github.com/pilosa/pilosa/ctl" +) + +var restorer = ctl.NewRestoreCommand(os.Stdin, os.Stdout, os.Stderr) + +var restoreCmd = &cobra.Command{ + Use: "restore", + Short: "Restore data to pilosa from a backup file.", + Long: ` +Restores a frame to the cluster from a backup file. +`, + Run: func(cmd *cobra.Command, args []string) { + if err := restorer.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + restoreCmd.Flags().StringVarP(&restorer.Host, "host", "", "localhost:15000", "host:port of Pilosa.") + restoreCmd.Flags().StringVarP(&restorer.Database, "database", "d", "", "Pilosa database to restore into.") + restoreCmd.Flags().StringVarP(&restorer.Frame, "frame", "f", "", "Frame to restore into.") + restoreCmd.Flags().StringVarP(&restorer.Path, "input-file", "i", "", "File to restore from.") + + err := viper.BindPFlags(restoreCmd.Flags()) + if err != nil { + log.Fatalf("Error binding restore flags: %v", err) + } + + RootCmd.AddCommand(restoreCmd) +} diff --git a/cmd/root.go b/cmd/root.go new file mode 100644 index 000000000..139214302 --- /dev/null +++ b/cmd/root.go @@ -0,0 +1,33 @@ +package cmd + +import "github.com/spf13/cobra" + +var ( + Version string + BuildTime string +) + +var RootCmd = &cobra.Command{ + Use: "pilosa", + Short: "Pilosa - A Distributed In-memory Binary Bitmap Index.", + // TODO - is documentation actually there? + Long: `Pilosa is a fast index to turbocharge your database. + +This binary contains Pilosa itself, as well as common +tools for administering pilosa, importing/exporting data, +backing up, and more. Complete documentation is available +at http://pilosa.com/docs + +`, +} + +func init() { + if Version == "" { + Version = "v0.0.0" + } + if BuildTime == "" { + BuildTime = "not recorded" + } + + RootCmd.Long = RootCmd.Long + "Version: " + Version + "\nBuild Time: " + BuildTime + "\n" +} diff --git a/cmd/server.go b/cmd/server.go new file mode 100644 index 000000000..efff0ac7d --- /dev/null +++ b/cmd/server.go @@ -0,0 +1,89 @@ +package cmd + +import ( + "fmt" + "log" + "os" + "os/signal" + "runtime/pprof" + "time" + + "github.com/spf13/cobra" + "github.com/spf13/viper" + + "github.com/pilosa/pilosa/server" +) + +var serve = server.NewCommand() + +var serveCmd = &cobra.Command{ + Use: "server", + Short: "Run Pilosa.", + Long: `pilosa server runs Pilosa. + +It will load existing data from the configured +directory, and start listening client connections +on the configured port.`, + Run: func(cmd *cobra.Command, args []string) { + serve.Server.Handler.Version = server.Version + fmt.Fprintf(serve.Stderr, "Pilosa %s, build time %s\n", server.Version, server.BuildTime) + + // Parse command line arguments. + if err := serve.SetupConfig(args); err != nil { + fmt.Fprintln(serve.Stderr, err) + os.Exit(2) + } + + // Start CPU profiling. + if serve.CPUProfile != "" { + f, err := os.Create(serve.CPUProfile) + if err != nil { + fmt.Fprintf(serve.Stderr, "create cpu profile: %v", err) + os.Exit(1) + } + defer f.Close() + + fmt.Fprintln(serve.Stderr, "Starting cpu profile") + pprof.StartCPUProfile(f) + time.AfterFunc(serve.CPUTime, func() { + fmt.Fprintln(serve.Stderr, "Stopping cpu profile") + pprof.StopCPUProfile() + f.Close() + }) + } + + // Execute the program. + if err := serve.Run(); err != nil { + fmt.Fprintln(serve.Stderr, err) + os.Exit(1) + } + + // First SIGKILL causes server to shut down gracefully. + c := make(chan os.Signal, 2) + signal.Notify(c, os.Interrupt) + sig := <-c + fmt.Fprintf(serve.Stderr, "Received %s; gracefully shutting down...\n", sig.String()) + + // Second signal causes a hard shutdown. + go func() { <-c; os.Exit(1) }() + + if err := serve.Close(); err != nil { + fmt.Fprintln(serve.Stderr, err) + os.Exit(1) + } + + }, +} + +func init() { + serveCmd.Flags().StringVarP(&serve.ConfigPath, "config", "c", "", "Configuration file to read from") + serveCmd.Flags().StringVarP(&serve.CPUProfile, "cpuprofile", "", "", "Where to store CPU profile") + serveCmd.Flags().DurationVarP(&serve.CPUTime, "cputime", "", 30*time.Second, "CPU profile duration") + + err := viper.BindPFlags(serveCmd.Flags()) + if err != nil { + log.Fatalf("Error binding server flags: %v", err) + } + + RootCmd.AddCommand(serveCmd) +} diff --git a/cmd/sort.go b/cmd/sort.go new file mode 100644 index 000000000..f9a1d682e --- /dev/null +++ b/cmd/sort.go @@ -0,0 +1,44 @@ +package cmd + +import ( + "context" + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/pilosa/pilosa/ctl" +) + +var sorter = ctl.NewSortCommand(os.Stdin, os.Stdout, os.Stderr) + +var sortCmd = &cobra.Command{ + Use: "sort ", + Short: "Sort import data for optimal import performance.", + Long: ` +Sorts the import data at PATH into the optimal sort order for importing. + +The format of the CSV file is: + + BITMAPID,PROFILEID + +The file should contain no headers. +`, + Run: func(cmd *cobra.Command, args []string) { + if len(args) == 0 { + fmt.Println("path required") + return + } else if len(args) > 1 { + fmt.Println("only one path supported") + return + } + sorter.Path = args[0] + if err := sorter.Run(context.Background()); err != nil { + fmt.Println(err) + } + }, +} + +func init() { + RootCmd.AddCommand(sortCmd) +} diff --git a/ctl/backup.go b/ctl/backup.go new file mode 100644 index 000000000..f0adf5766 --- /dev/null +++ b/ctl/backup.go @@ -0,0 +1,72 @@ +package ctl + +import ( + "context" + "errors" + "io" + "os" + + "github.com/pilosa/pilosa" +) + +// BackupCommand represents a command for backing up a frame. +type BackupCommand struct { + // Destination host and port. + Host string + + // Name of the database & frame to backup. + Database string + Frame string + + // Output file to write to. + Path string + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewBackupCommand returns a new instance of BackupCommand. +func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand { + return &BackupCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run executes the backup. +func (cmd *BackupCommand) Run(ctx context.Context) error { + // Validate arguments. + if cmd.Path == "" { + return errors.New("output file required") + } + + // Create a client to the server. + client, err := pilosa.NewClient(cmd.Host) + if err != nil { + return err + } + + // Open output file. + f, err := os.Create(cmd.Path) + if err != nil { + return err + } + defer f.Close() + + // Begin streaming backup. + if err := client.BackupTo(ctx, f, cmd.Database, cmd.Frame); err != nil { + return err + } + + // Sync & close file to ensure durability. + if err := f.Sync(); err != nil { + return err + } else if err = f.Close(); err != nil { + return err + } + + return nil +} diff --git a/ctl/bench.go b/ctl/bench.go new file mode 100644 index 000000000..2afe7bb27 --- /dev/null +++ b/ctl/bench.go @@ -0,0 +1,92 @@ +package ctl + +import ( + "context" + "errors" + "fmt" + "io" + "math/rand" + "time" + + "github.com/pilosa/pilosa" +) + +// BenchCommand represents a command for benchmarking database operations. +type BenchCommand struct { + // Destination host and port. + Host string + + // Name of the database & frame to execute against. + Database string + Frame string + + // Type of operation and number to execute. + Op string + N int + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewBenchCommand returns a new instance of BenchCommand. +func NewBenchCommand(stdin io.Reader, stdout, stderr io.Writer) *BenchCommand { + return &BenchCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run executes the bench command. +func (cmd *BenchCommand) Run(ctx context.Context) error { + // Create a client to the server. + client, err := pilosa.NewClient(cmd.Host) + if err != nil { + return err + } + + switch cmd.Op { + case "set-bit": + return cmd.runSetBit(ctx, client) + case "": + return errors.New("op required") + default: + return fmt.Errorf("unknown bench op: %q", cmd.Op) + } +} + +// runSetBit executes a benchmark of random SetBit() operations. +func (cmd *BenchCommand) runSetBit(ctx context.Context, client *pilosa.Client) error { + if cmd.N == 0 { + return errors.New("operation count required") + } else if cmd.Database == "" { + return pilosa.ErrDatabaseRequired + } else if cmd.Frame == "" { + return pilosa.ErrFrameRequired + } + + const maxBitmapID = 1000 + const maxProfileID = 100000 + + startTime := time.Now() + + // Execute operation continuously. + for i := 0; i < cmd.N; i++ { + bitmapID := rand.Intn(maxBitmapID) + profileID := rand.Intn(maxProfileID) + + q := fmt.Sprintf(`SetBit(id=%d, frame="%s", profileID=%d)`, bitmapID, cmd.Frame, profileID) + + if _, err := client.ExecuteQuery(ctx, cmd.Database, q, true); err != nil { + return err + } + } + + // Print results. + elapsed := time.Since(startTime) + fmt.Fprintf(cmd.Stdout, "Executed %d operations in %s (%0.3f op/sec)\n", cmd.N, elapsed, float64(cmd.N)/elapsed.Seconds()) + + return nil +} diff --git a/ctl/check.go b/ctl/check.go new file mode 100644 index 000000000..0893790f2 --- /dev/null +++ b/ctl/check.go @@ -0,0 +1,114 @@ +package ctl + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "syscall" + + "github.com/pilosa/pilosa/roaring" +) + +// CheckCommand represents a command for performing consistency checks on data files. +type CheckCommand struct { + // Data file paths. + Paths []string + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewCheckCommand returns a new instance of CheckCommand. +func NewCheckCommand(stdin io.Reader, stdout, stderr io.Writer) *CheckCommand { + return &CheckCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run executes the check command. +func (cmd *CheckCommand) Run(ctx context.Context) error { + for _, path := range cmd.Paths { + switch filepath.Ext(path) { + case "": + if err := cmd.checkBitmapFile(path); err != nil { + return err + } + + case ".cache": + if err := cmd.checkCacheFile(path); err != nil { + return err + } + + case ".snapshotting": + if err := cmd.checkSnapshotFile(path); err != nil { + return err + } + } + } + + return nil +} + +// checkBitmapFile performs a consistency check on path for a roaring bitmap file. +func (cmd *CheckCommand) checkBitmapFile(path string) error { + // Open file handle. + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + + fi, err := f.Stat() + if err != nil { + return err + } + + // Memory map the file. + data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED) + if err != nil { + return err + } + defer syscall.Munmap(data) + + // Attach the mmap file to the bitmap. + bm := roaring.NewBitmap() + if err := bm.UnmarshalBinary(data); err != nil { + return err + } + + // Perform consistency check. + if err := bm.Check(); err != nil { + // Print returned errors. + switch err := err.(type) { + case roaring.ErrorList: + for i := range err { + fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err[i].Error()) + } + default: + fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err.Error()) + } + } + + // Print success message if no errors were found. + fmt.Fprintf(cmd.Stdout, "%s: ok\n", path) + + return nil +} + +// checkCacheFile performs a consistency check on path for a cache file. +func (cmd *CheckCommand) checkCacheFile(path string) error { + fmt.Fprintf(cmd.Stderr, "%s: ignoring cache file\n", path) + return nil +} + +// checkSnapshotFile performs a consistency check on path for a snapshot file. +func (cmd *CheckCommand) checkSnapshotFile(path string) error { + fmt.Fprintf(cmd.Stderr, "%s: ignoring snapshot file\n", path) + return nil +} diff --git a/ctl/config.go b/ctl/config.go new file mode 100644 index 000000000..998c17b47 --- /dev/null +++ b/ctl/config.go @@ -0,0 +1,43 @@ +package ctl + +import ( + "context" + "fmt" + "io" + "strings" +) + +// ConfigCommand represents a command for printing a default config. +type ConfigCommand struct { + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewConfigCommand returns a new instance of ConfigCommand. +func NewConfigCommand(stdin io.Reader, stdout, stderr io.Writer) *ConfigCommand { + return &ConfigCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run prints out the default config. +func (cmd *ConfigCommand) Run(ctx context.Context) error { + fmt.Fprintln(cmd.Stdout, strings.TrimSpace(` +data-dir = "~/.pilosa" +host = "localhost:15000" + +[cluster] +replicas = 1 + +[[cluster.node]] +host = "localhost:15000" + +[plugins] +path = "" +`)+"\n") + return nil +} diff --git a/ctl/export.go b/ctl/export.go new file mode 100644 index 000000000..4eead3239 --- /dev/null +++ b/ctl/export.go @@ -0,0 +1,91 @@ +package ctl + +import ( + "context" + "io" + "log" + "os" + + "github.com/pilosa/pilosa" +) + +// ExportCommand represents a command for bulk exporting data from a server. +type ExportCommand struct { + // Remote host and port. + Host string + + // Name of the database & frame to export from. + Database string + Frame string + + // Filename to export to. + Path string + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewExportCommand returns a new instance of ExportCommand. +func NewExportCommand(stdin io.Reader, stdout, stderr io.Writer) *ExportCommand { + return &ExportCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run executes the export. +func (cmd *ExportCommand) Run(ctx context.Context) error { + logger := log.New(cmd.Stderr, "", log.LstdFlags) + + // Validate arguments. + if cmd.Database == "" { + return pilosa.ErrDatabaseRequired + } else if cmd.Frame == "" { + return pilosa.ErrFrameRequired + } + + // Use output file, if specified. + // Otherwise use STDOUT. + var w io.Writer = cmd.Stdout + if cmd.Path != "" { + f, err := os.Create(cmd.Path) + if err != nil { + return err + } + defer f.Close() + + w = f + } + + // Create a client to the server. + client, err := pilosa.NewClient(cmd.Host) + if err != nil { + return err + } + + // Determine slice count. + maxSlices, err := client.MaxSliceByDatabase(ctx) + if err != nil { + return err + } + + // Export each slice. + for slice := uint64(0); slice <= maxSlices[cmd.Database]; slice++ { + logger.Printf("exporting slice: %d", slice) + if err := client.ExportCSV(ctx, cmd.Database, cmd.Frame, slice, w); err != nil { + return err + } + } + + // Close writer, if applicable. + if w, ok := w.(io.Closer); ok { + if err := w.Close(); err != nil { + return err + } + } + + return nil +} diff --git a/pilosactl/import.go b/ctl/import.go similarity index 78% rename from pilosactl/import.go rename to ctl/import.go index 9d4ad63f4..107ddd84a 100644 --- a/pilosactl/import.go +++ b/ctl/import.go @@ -1,17 +1,14 @@ -package pilosactl +package ctl import ( "context" "encoding/csv" "errors" - "flag" "fmt" "io" - "io/ioutil" "log" "os" "strconv" - "strings" "time" "github.com/pilosa/pilosa" @@ -56,41 +53,6 @@ func (cmd *ImportCommand) String() string { return fmt.Sprint(*cmd) } -// ParseFlags parses command line flags from args. -func (cmd *ImportCommand) ParseFlags(args []string) error { - fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError) - fs.SetOutput(ioutil.Discard) - fs.StringVar(&cmd.Host, "host", "localhost:15000", "host:port") - fs.StringVar(&cmd.Database, "d", "", "database") - fs.StringVar(&cmd.Frame, "f", "", "frame") - fs.IntVar(&cmd.BufferSize, "buffer-size", cmd.BufferSize, "buffer size") - if err := fs.Parse(args); err != nil { - return err - } - - // Extract the import paths. - cmd.Paths = fs.Args() - - return nil -} - -// Usage returns the usage message to be printed. -func (cmd *ImportCommand) Usage() string { - return strings.TrimSpace(` -usage: pilosactl import -host HOST -d database -f frame paths - -Bulk imports one or more CSV files to a host's database and frame. The bits -of the CSV file are grouped by slice for the most efficient import. - -The format of the CSV file is: - - BITMAPID,PROFILEID,[TIME] - -The file should contain no headers. The TIME column is optional and can be -omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. -`) -} - // Run executes the main program execution. func (cmd *ImportCommand) Run(ctx context.Context) error { logger := log.New(cmd.Stderr, "", log.LstdFlags) @@ -104,7 +66,6 @@ func (cmd *ImportCommand) Run(ctx context.Context) error { } else if len(cmd.Paths) == 0 { return errors.New("path required") } - // Create a client to the server. client, err := pilosa.NewClient(cmd.Host) if err != nil { diff --git a/ctl/inspect.go b/ctl/inspect.go new file mode 100644 index 000000000..86434b131 --- /dev/null +++ b/ctl/inspect.go @@ -0,0 +1,94 @@ +package ctl + +import ( + "context" + "fmt" + "io" + "os" + "syscall" + "text/tabwriter" + "time" + "unsafe" + + "github.com/pilosa/pilosa/roaring" +) + +// InspectCommand represents a command for inspecting fragment data files. +type InspectCommand struct { + // Path to data file + Path string + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewInspectCommand returns a new instance of InspectCommand. +func NewInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *InspectCommand { + return &InspectCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run executes the inspect command. +func (cmd *InspectCommand) Run(ctx context.Context) error { + // Open file handle. + f, err := os.Open(cmd.Path) + if err != nil { + return err + } + defer f.Close() + + fi, err := f.Stat() + if err != nil { + return err + } + + // Memory map the file. + data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED) + if err != nil { + return err + } + defer syscall.Munmap(data) + + // Attach the mmap file to the bitmap. + t := time.Now() + fmt.Fprintf(cmd.Stderr, "unmarshaling bitmap...") + bm := roaring.NewBitmap() + if err := bm.UnmarshalBinary(data); err != nil { + return err + } + fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) + + // Retrieve stats. + t = time.Now() + fmt.Fprintf(cmd.Stderr, "calculating stats...") + info := bm.Info() + fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) + + // Print top-level info. + fmt.Fprintf(cmd.Stdout, "== Bitmap Info ==\n") + fmt.Fprintf(cmd.Stdout, "Containers: %d\n", len(info.Containers)) + fmt.Fprintf(cmd.Stdout, "Operations: %d\n", info.OpN) + fmt.Fprintln(cmd.Stdout, "") + + // Print info for each container. + fmt.Fprintln(cmd.Stdout, "== Containers ==") + tw := tabwriter.NewWriter(cmd.Stdout, 0, 8, 0, '\t', 0) + fmt.Fprintf(tw, "%s\t%s\t% 8s \t% 8s\t%s\n", "KEY", "TYPE", "N", "ALLOC", "OFFSET") + for _, ci := range info.Containers { + fmt.Fprintf(tw, "%d\t%s\t% 8d \t% 8d \t0x%08x\n", + ci.Key, + ci.Type, + ci.N, + ci.Alloc, + uintptr(ci.Pointer)-uintptr(unsafe.Pointer(&data[0])), + ) + } + tw.Flush() + + return nil +} diff --git a/ctl/restore.go b/ctl/restore.go new file mode 100644 index 000000000..a650df495 --- /dev/null +++ b/ctl/restore.go @@ -0,0 +1,65 @@ +package ctl + +import ( + "context" + "errors" + "io" + "os" + + "github.com/pilosa/pilosa" +) + +// RestoreCommand represents a command for restoring a frame from a backup. +type RestoreCommand struct { + // Destination host and port. + Host string + + // Name of the database & frame to backup. + Database string + Frame string + + // Import file to read from. + Path string + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewRestoreCommand returns a new instance of RestoreCommand. +func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand { + return &RestoreCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run executes the restore command. +func (cmd *RestoreCommand) Run(ctx context.Context) error { + // Validate arguments. + if cmd.Path == "" { + return errors.New("backup file required") + } + + // Create a client to the server. + client, err := pilosa.NewClient(cmd.Host) + if err != nil { + return err + } + + // Open backup file. + f, err := os.Open(cmd.Path) + if err != nil { + return err + } + defer f.Close() + + // Restore backup file to the cluster. + if err := client.RestoreFrom(ctx, f, cmd.Database, cmd.Frame); err != nil { + return err + } + + return nil +} diff --git a/ctl/sort.go b/ctl/sort.go new file mode 100644 index 000000000..66bd16372 --- /dev/null +++ b/ctl/sort.go @@ -0,0 +1,138 @@ +package ctl + +import ( + "bufio" + "context" + "encoding/csv" + "errors" + "fmt" + "io" + "os" + "sort" + "strconv" + "time" + + "github.com/pilosa/pilosa" +) + +// SortCommand represents a command for sorting import data. +type SortCommand struct { + // Filename to sort + Path string + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewSortCommand returns a new instance of SortCommand. +func NewSortCommand(stdin io.Reader, stdout, stderr io.Writer) *SortCommand { + return &SortCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// Run executes the sort command. +func (cmd *SortCommand) Run(ctx context.Context) error { + // Open file for reading. + f, err := os.Open(cmd.Path) + if err != nil { + return err + } + defer f.Close() + + // Read rows as bits. + r := csv.NewReader(f) + r.FieldsPerRecord = -1 + a := make([]pilosa.Bit, 0, 1000000) + for { + bitmapID, profileID, timestamp, err := readCSVRow(r) + if err == io.EOF { + break + } else if err == errBlank { + continue + } else if err != nil { + return err + } + a = append(a, pilosa.Bit{BitmapID: bitmapID, ProfileID: profileID, Timestamp: timestamp}) + } + + // Sort bits by position. + sort.Sort(pilosa.BitsByPos(a)) + + // Rewrite to STDOUT. + w := bufio.NewWriter(cmd.Stdout) + buf := make([]byte, 0, 1024) + for _, bit := range a { + // Write CSV to buffer. + buf = buf[:0] + buf = strconv.AppendUint(buf, bit.BitmapID, 10) + + buf = append(buf, ',') + buf = strconv.AppendUint(buf, bit.ProfileID, 10) + + if bit.Timestamp != 0 { + buf = append(buf, ',') + buf = append(buf, time.Unix(0, bit.Timestamp).UTC().Format(pilosa.TimeFormat)...) + } + + buf = append(buf, '\n') + + // Write to output. + if _, err := w.Write(buf); err != nil { + return err + } + } + + // Ensure buffer is flushed before exiting. + if err := w.Flush(); err != nil { + return err + } + + return nil +} + +// readCSVRow reads a bitmap/profile pair from a CSV row. +func readCSVRow(r *csv.Reader) (bitmapID, profileID uint64, timestamp int64, err error) { + // Read CSV row. + record, err := r.Read() + if err != nil { + return 0, 0, 0, err + } + + // Ignore blank rows. + if record[0] == "" { + return 0, 0, 0, errBlank + } else if len(record) < 2 { + return 0, 0, 0, fmt.Errorf("bad column count: %d", len(record)) + } + + // Parse bitmap id. + bitmapID, err = strconv.ParseUint(record[0], 10, 64) + if err != nil { + return 0, 0, 0, fmt.Errorf("invalid bitmap id: %q", record[0]) + } + + // Parse bitmap id. + profileID, err = strconv.ParseUint(record[1], 10, 64) + if err != nil { + return 0, 0, 0, fmt.Errorf("invalid profile id: %q", record[1]) + } + + // Parse timestamp, if available. + if len(record) > 2 && record[2] != "" { + t, err := time.Parse(pilosa.TimeFormat, record[2]) + if err != nil { + return 0, 0, 0, fmt.Errorf("invalid timestamp: %q", record[2]) + } + timestamp = t.UnixNano() + } + + return bitmapID, profileID, timestamp, nil +} + +// errBlank indicates a blank row in a CSV file. +var errBlank = errors.New("blank row") diff --git a/db.go b/db.go index 23f678bf6..9348c97b6 100644 --- a/db.go +++ b/db.go @@ -48,7 +48,12 @@ type DB struct { } // NewDB returns a new instance of DB. -func NewDB(path, name string) *DB { +func NewDB(path, name string) (*DB, error) { + err := ValidateName(name) + if err != nil { + return nil, err + } + return &DB{ path: path, name: name, @@ -61,7 +66,7 @@ func NewDB(path, name string) *DB { stats: NopStatsClient, LogOutput: ioutil.Discard, - } + }, nil } // Name returns name of the database. @@ -141,7 +146,10 @@ func (db *DB) openFrames() error { continue } - fr := db.newFrame(db.FramePath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + fr, err := db.newFrame(db.FramePath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + if err != nil { + return ErrName + } if err := fr.Open(); err != nil { return fmt.Errorf("open frame: name=%s, err=%s", fr.Name(), err) } @@ -312,12 +320,16 @@ func (db *DB) CreateFrameIfNotExists(name string, opt FrameOptions) (*Frame, err } func (db *DB) createFrame(name string, opt FrameOptions) (*Frame, error) { + if name == "" { return nil, errors.New("frame name required") } // Initialize frame. - f := db.newFrame(db.FramePath(name), name) + f, err := db.newFrame(db.FramePath(name), name) + if err != nil { + return nil, err + } // Open frame. if err := f.Open(); err != nil { @@ -335,11 +347,14 @@ func (db *DB) createFrame(name string, opt FrameOptions) (*Frame, error) { return f, nil } -func (db *DB) newFrame(path, name string) *Frame { - f := NewFrame(path, db.name, name) +func (db *DB) newFrame(path, name string) (*Frame, error) { + f, err := NewFrame(path, db.name, name) + if err != nil { + return nil, err + } f.LogOutput = db.LogOutput f.stats = db.stats.WithTags(fmt.Sprintf("frame:%s", name)) - return f + return f, nil } // DeleteFrame removes a frame from the database. diff --git a/db_test.go b/db_test.go index b95f89e62..96120dc1e 100644 --- a/db_test.go +++ b/db_test.go @@ -89,7 +89,11 @@ func NewDB() *DB { if err != nil { panic(err) } - return &DB{DB: pilosa.NewDB(path, "d")} + db, err := pilosa.NewDB(path, "d") + if err != nil { + panic(err) + } + return &DB{DB: db} } // MustOpenDB returns a new, opened database at a temporary path. Panic on error. @@ -109,12 +113,16 @@ func (db *DB) Close() error { // Reopen closes the database and reopens it. func (db *DB) Reopen() error { + var err error if err := db.DB.Close(); err != nil { return err } path, name := db.Path(), db.Name() - db.DB = pilosa.NewDB(path, name) + db.DB, err = pilosa.NewDB(path, name) + if err != nil { + return err + } if err := db.Open(); err != nil { return err @@ -130,3 +138,15 @@ func (db *DB) MustSetBit(name string, bitmapID, profileID uint64, t *time.Time) } return changed } + +// Ensure database can delete a frame. +func TestDB_InvalidName(t *testing.T) { + path, err := ioutil.TempDir("", "pilosa-db-") + if err != nil { + panic(err) + } + db, err := pilosa.NewDB(path, "ABC") + if db != nil { + t.Fatalf("unexpected db name %s", db) + } +} diff --git a/executor.go b/executor.go index 0ba1e7017..774e2443f 100644 --- a/executor.go +++ b/executor.go @@ -52,13 +52,15 @@ func (e *Executor) Execute(ctx context.Context, db string, q *pql.Query, slices // If slices aren't specified, then include all of them. if len(slices) == 0 { - // Round up the number of slices. - maxSlice := e.Index.DB(db).MaxSlice() + if needsSlices(q.Calls) { + // Round up the number of slices. + maxSlice := e.Index.DB(db).MaxSlice() - // Generate a slices of all slices. - slices = make([]uint64, maxSlice+1) - for i := range slices { - slices[i] = uint64(i) + // Generate a slices of all slices. + slices = make([]uint64, maxSlice+1) + for i := range slices { + slices[i] = uint64(i) + } } } @@ -168,6 +170,10 @@ func (e *Executor) executeBitmapCallSlice(ctx context.Context, db string, c *pql // requeries to retrieve the full counts for each of the top results. func (e *Executor) executeTopN(ctx context.Context, db string, c *pql.Call, slices []uint64, opt *ExecOptions) ([]Pair, error) { bitmapIDs, _ := c.Args["ids"].([]uint64) + var n uint64 + if nval, ok := c.Args["n"]; ok { + n = nval.(uint64) + } // Execute original query. pairs, err := e.executeTopNSlices(ctx, db, c, slices, opt) @@ -180,21 +186,29 @@ func (e *Executor) executeTopN(ctx context.Context, db string, c *pql.Call, slic if len(pairs) == 0 || len(bitmapIDs) > 0 || opt.Remote { return pairs, nil } - // Only the original caller should refetch the full counts. other := c.Clone() - other.Args["n"] = 0 + + // Double the size of n for other calls in order to... + // TODO: travis review + other.Args["n"] = len(bitmapIDs) * 2 ids := Pairs(pairs).Keys() sort.Sort(uint64Slice(ids)) other.Args["ids"] = ids - return e.executeTopNSlices(ctx, db, other, slices, opt) + trimmedList, err := e.executeTopNSlices(ctx, db, other, slices, opt) + if err != nil { + return nil, err + } + + if n != 0 && int(n) < len(trimmedList) { + trimmedList = trimmedList[0:n] + } + return trimmedList, nil } func (e *Executor) executeTopNSlices(ctx context.Context, db string, c *pql.Call, slices []uint64, opt *ExecOptions) ([]Pair, error) { - n, _ := c.Args["n"].(uint64) - // Execute calls in bulk on each remote node and merge. mapFn := func(slice uint64) (interface{}, error) { return e.executeTopNSlice(ctx, db, c, slice) @@ -215,11 +229,6 @@ func (e *Executor) executeTopNSlices(ctx context.Context, db string, c *pql.Call // Sort final merged results. sort.Sort(Pairs(results)) - // Only keep the top n after sorting. - if n > 0 && len(results) > int(n) { - results = results[0:n] - } - return results, nil } @@ -954,6 +963,7 @@ func (e *Executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod if n.Host == e.Host { resp.result, resp.err = e.mapperLocal(ctx, nodeSlices, mapFn, reduceFn) } else if !opt.Remote { + results, err := e.exec(ctx, n, db, &pql.Query{Calls: []*pql.Call{c}}, nodeSlices, opt) if len(results) > 0 { resp.result = results[0] @@ -1052,3 +1062,21 @@ func hasOnlySetBitmapAttrs(calls []*pql.Call) bool { } return true } + +func needsSlices(calls []*pql.Call) bool { + if len(calls) == 0 { + return false + } + for _, call := range calls { + switch call.Name { + case "ClearBit", "Profile", "SetBit", "SetBitmapAttrs", "SetProfileAttrs": + continue + case "Count", "TopN": + return true + // default catches Bitmap calls + default: + return true + } + } + return false +} diff --git a/executor_test.go b/executor_test.go index 1ad1fbbbb..1cc0985ac 100644 --- a/executor_test.go +++ b/executor_test.go @@ -182,7 +182,7 @@ func TestExecutor_Execute_SetBitmapAttrs(t *testing.T) { db := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{}) if _, err := db.CreateFrameIfNotExists("f", pilosa.FrameOptions{}); err != nil { t.Fatal(err) - } else if _, err := db.CreateFrameIfNotExists("XXX", pilosa.FrameOptions{}); err != nil { + } else if _, err := db.CreateFrameIfNotExists("xxx", pilosa.FrameOptions{}); err != nil { t.Fatal(err) } @@ -195,7 +195,7 @@ func TestExecutor_Execute_SetBitmapAttrs(t *testing.T) { if _, err := e.Execute(context.Background(), "d", MustParse(`SetBitmapAttrs(id=200, frame=f, YYY=1)`), nil, nil); err != nil { t.Fatal(err) } - if _, err := e.Execute(context.Background(), "d", MustParse(`SetBitmapAttrs(id=10, frame=XXX, YYY=1)`), nil, nil); err != nil { + if _, err := e.Execute(context.Background(), "d", MustParse(`SetBitmapAttrs(id=10, frame=xxx, YYY=1)`), nil, nil); err != nil { t.Fatal(err) } if _, err := e.Execute(context.Background(), "d", MustParse(`SetBitmapAttrs(id=10, frame=f, baz=123, bat=true)`), nil, nil); err != nil { @@ -260,6 +260,40 @@ func TestExecutor_Execute_TopN_fill(t *testing.T) { } } +// Ensure +func TestExecutor_Execute_TopN_fill_small(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(0, 2*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(0, 3*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 4).SetBit(0, 4*SliceWidth) + + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(1, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(1, 1) + + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(2, SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(2, SliceWidth+1) + + idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(3, 2*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(3, 2*SliceWidth+1) + + idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(4, 3*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(4, 3*SliceWidth+1) + + // Execute query. + e := NewExecutor(idx.Index, NewCluster(1)) + if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=1)`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{ + {Key: 0, Count: 5}, + }}) { + t.Fatalf("unexpected result: %s", spew.Sdump(result)) + } +} + // Ensure a TopN() query with a source bitmap can be executed. func TestExecutor_Execute_TopN_Src(t *testing.T) { idx := MustOpenIndex() @@ -293,6 +327,52 @@ func TestExecutor_Execute_TopN_Src(t *testing.T) { } } +//Ensure TopN handles Attribute filters +func TestExecutor_Execute_TopN_Attr(t *testing.T) { + // + idx := MustOpenIndex() + defer idx.Close() + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth) + + if err := idx.Frame("d", "f").BitmapAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil { + t.Fatal(err) + } + e := NewExecutor(idx.Index, NewCluster(1)) + if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{ + {Key: 10, Count: 1}, + }}) { + t.Fatalf("unexpected result: %s", spew.Sdump(result)) + } + +} + +//Ensure TopN handles Attribute filters with source bitmap +func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) { + // + idx := MustOpenIndex() + defer idx.Close() + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth) + + if err := idx.Frame("d", "f").BitmapAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil { + t.Fatal(err) + } + e := NewExecutor(idx.Index, NewCluster(1)) + if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(Bitmap(id=10,frame=f),frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{ + {Key: 10, Count: 1}, + }}) { + t.Fatalf("unexpected result: %s", spew.Sdump(result)) + } + +} + // Ensure a range query can be executed. func TestExecutor_Execute_Range(t *testing.T) { idx := MustOpenIndex() diff --git a/fragment.go b/fragment.go index 1181f1739..e70dab80c 100644 --- a/fragment.go +++ b/fragment.go @@ -4,6 +4,7 @@ import ( "archive/tar" "bufio" "bytes" + "container/heap" "context" "crypto/sha1" "encoding/binary" @@ -49,7 +50,7 @@ const ( const ( // DefaultFragmentMaxOpN is the default value for Fragment.MaxOpN. - DefaultFragmentMaxOpN = 1000 + DefaultFragmentMaxOpN = 2000 ) // Fragment represents the intersection of a frame and slice in a database. @@ -68,9 +69,12 @@ type Fragment struct { storageData []byte opN int // number of ops since snapshot - // Bitmap cache. + // Cache for bitmap counts. cache Cache + // Cache containing full bitmaps (not just counts). + bitmapCache BitmapCache + // Cached checksums for each block. checksums map[int][]byte @@ -203,6 +207,7 @@ func (f *Fragment) openStorage() error { // Attach the file to the bitmap to act as a write-ahead log. f.storage.OpWriter = f.file + f.bitmapCache = &SimpleCache{make(map[uint64]*Bitmap)} return nil @@ -239,9 +244,11 @@ func (f *Fragment) openCache() error { // Read in all bitmaps by ID. // This will cause them to be added to the cache. for _, bitmapID := range pb.BitmapIDs { - n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) - f.cache.Add(bitmapID, n) + //n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) + n := f.bitmap(bitmapID, true, true).Count() + f.cache.BulkAdd(bitmapID, n) } + f.cache.Invalidate() return nil } @@ -305,26 +312,37 @@ func (f *Fragment) logger() *log.Logger { return log.New(f.LogOutput, "", log.Ls func (f *Fragment) Bitmap(bitmapID uint64) *Bitmap { f.mu.Lock() defer f.mu.Unlock() - return f.bitmap(bitmapID) + return f.bitmap(bitmapID, true, true) } -func (f *Fragment) bitmap(bitmapID uint64) *Bitmap { +func (f *Fragment) bitmap(bitmapID uint64, checkBitmapCache bool, updateBitmapCache bool) *Bitmap { + + if checkBitmapCache { + r, ok := f.bitmapCache.Fetch(bitmapID) + if ok && r != nil { + return r + } + } + // Only use a subset of the containers. // NOTE: The start & end ranges must be divisible by data := f.storage.OffsetRange(f.slice*SliceWidth, bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) // Reference bitmap subrange in storage. + // We Clone() data because otherwise bm will contains pointers to containers in storage. + // This causes unexpected results when we cache the bitmap and try to use it later. bm := &Bitmap{ segments: []BitmapSegment{{ - data: *data, + data: *data.Clone(), slice: f.slice, writable: false, }}, } bm.InvalidateCount() - // Update cache. - f.cache.Add(bitmapID, bm.Count()) + if updateBitmapCache { + f.bitmapCache.Add(bitmapID, bm) + } return bm } @@ -337,31 +355,39 @@ func (f *Fragment) SetBit(bitmapID, profileID uint64) (changed bool, err error) return f.setBit(bitmapID, profileID) } -func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error) { - // Determine the position of the bit in the storage. +func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, err error) { changed = false + // Determine the position of the bit in the storage. pos, err := f.pos(bitmapID, profileID) if err != nil { return false, err } // Write to storage. + if changed, err = f.storage.Add(pos); err != nil { return false, err } + // Don't update the cache if nothing changed. + if !changed { + return changed, nil + } + // Invalidate block checksum. delete(f.checksums, int(bitmapID/HashBlockSize)) - // If the number of operations exceeds the limit then snapshot. + // Increment number of operations until snapshot is required. if err := f.incrementOpN(); err != nil { return false, err } + // Get the bitmap from bitmapCache or fragment.storage. + bm := f.bitmap(bitmapID, true, true) + bm.SetBit(profileID) + // Update the cache. - if f.bitmap(bitmapID).SetBit(profileID) { - changed = true - } + f.cache.Add(bitmapID, bm.Count()) f.stats.Count("setN", 1) @@ -376,7 +402,8 @@ func (f *Fragment) ClearBit(bitmapID, profileID uint64) (bool, error) { return f.clearBit(bitmapID, profileID) } -func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { +func (f *Fragment) clearBit(bitmapID, profileID uint64) (changed bool, err error) { + changed = false // Determine the position of the bit in the storage. pos, err := f.pos(bitmapID, profileID) if err != nil { @@ -384,11 +411,15 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { } // Write to storage. - changed, err := f.storage.Remove(pos) - if err != nil { + if changed, err = f.storage.Remove(pos); err != nil { return false, err } + // Don't update the cache if nothing changed. + if !changed { + return changed, nil + } + // Invalidate block checksum. delete(f.checksums, int(bitmapID/HashBlockSize)) @@ -397,10 +428,12 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { return false, err } + // Get the bitmap from bitmapCache or fragment.storage. + bm := f.bitmap(bitmapID, true, true) + bm.ClearBit(profileID) + // Update the cache. - if f.bitmap(bitmapID).ClearBit(profileID) { - return true, nil - } + f.cache.Add(bitmapID, bm.Count()) f.stats.Count("clearN", 1) @@ -453,7 +486,8 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { } // Iterate over rankings and add to results until we have enough. - results := make([]Pair, 0, opt.N) + //results := make(PairHeap, 0, opt.N) + results := &PairHeap{} for _, pair := range pairs { bitmapID, n := pair.ID, pair.Count @@ -477,7 +511,7 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { } // The initial n pairs should simply be added to the results. - if opt.N == 0 || len(results) < opt.N { + if opt.N == 0 || results.Len() < opt.N { // Calculate count and append. count := n if opt.Src != nil { @@ -486,26 +520,22 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { if count == 0 { continue } - results = append(results, Pair{Key: bitmapID, Count: count}) + heap.Push(results, Pair{Key: bitmapID, Count: count}) // If we reach the requested number of pairs and we are not computing // intersections then simply exit. If we are intersecting then sort // and then only keep pairs that are higher than the lowest count. - if opt.N > 0 && len(results) == opt.N { + if opt.N > 0 && results.Len() == opt.N { if opt.Src == nil { break } - sort.Sort(Pairs(results)) } continue } // Retrieve the lowest count we have. // If it's too low then don't try finding anymore pairs. - threshold := results[len(results)-1].Count - if threshold < MinThreshold { - break - } + threshold := results.Pairs[0].Count // If the bitmap doesn't have enough bits set before the intersection // then we can assume that any remaing bitmaps also have a count too low. @@ -515,22 +545,22 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { // Calculate the intersecting bit count and skip if it's below our // last bitmap in our current result set. + count := opt.Src.IntersectionCount(f.Bitmap(bitmapID)) if count < threshold { continue } - // Swap out the last pair for this new count. - results[len(results)-1] = Pair{Key: bitmapID, Count: count} - - // If it's count is also higher than the second to last item then resort. - if len(results) >= 2 && count > results[len(results)-2].Count { - sort.Sort(Pairs(results)) - } + heap.Push(results, Pair{Key: bitmapID, Count: count}) } - - sort.Sort(Pairs(results)) - return results, nil + r := make(Pairs, results.Len(), results.Len()) + x := results.Len() + i := 1 + for results.Len() > 0 { + r[x-i] = heap.Pop(results).(Pair) + i++ + } + return r, nil } func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair { @@ -543,23 +573,27 @@ func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair { } // Otherwise retrieve specific bitmaps. - pairs := make([]BitmapPair, len(bitmapIDs)) - for i, bitmapID := range bitmapIDs { + pairs := make([]BitmapPair, 0, len(bitmapIDs)) + for _, bitmapID := range bitmapIDs { // Look up cache first, if available. if n := f.cache.Get(bitmapID); n > 0 { - pairs[i] = BitmapPair{ + pairs = append(pairs, BitmapPair{ ID: bitmapID, Count: n, - } + }) continue } - // Otherwise load from storage. - pairs[i] = BitmapPair{ - ID: bitmapID, - Count: f.Bitmap(bitmapID).Count(), + bm := f.Bitmap(bitmapID) + if bm.Count() > 0 { + // Otherwise load from storage. + pairs = append(pairs, BitmapPair{ + ID: bitmapID, + Count: bm.Count(), + }) } } + sort.Sort(BitmapPairs(pairs)) return pairs } @@ -838,7 +872,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { // Process every bit. // If an error occurs then reopen the storage. lastID := uint64(0) - bmCounter := 0 if err := func() error { set := make(map[uint64]struct{}) for i := range bitmapIDs { @@ -851,7 +884,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { } // Write to storage. - changed, err := f.storage.Add(pos) + _, err = f.storage.Add(pos) if err != nil { return err } @@ -863,9 +896,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { lastID = bitmapID set[bitmapID] = struct{}{} } - if changed { - bmCounter += 1 - } // Invalidate block checksum. delete(f.checksums, int(bitmapID/HashBlockSize)) @@ -873,7 +903,10 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { // Update cache counts for all bitmaps. for bitmapID := range set { - f.cache.Add(bitmapID, f.bitmap(bitmapID).Count()) + // Import should ALWAYS have bitmap() load a new bm from fragment.storage + // because the bitmap that's in bitmapCache hasn't been updated with + // this import's data. + f.cache.BulkAdd(bitmapID, f.bitmap(bitmapID, false, false).Count()) } f.cache.Invalidate() @@ -912,10 +945,15 @@ func (f *Fragment) Snapshot() error { defer f.mu.Unlock() return f.snapshot() } +func track(start time.Time, name string, logger *log.Logger) { + elapsed := time.Since(start) + logger.Printf("%s took %s", name, elapsed) +} func (f *Fragment) snapshot() error { logger := f.logger() logger.Printf("fragment: snapshotting %s/%s/%d", f.db, f.frame, f.slice) + defer track(time.Now(), fmt.Sprintf("fragment: snapshot complete %s/%s/%d", f.db, f.frame, f.slice), logger) // Create a temporary file to snapshot to. snapshotPath := f.path + SnapshotExt @@ -1216,7 +1254,6 @@ func (s *FragmentSyncer) SyncFragment() error { // Determine replica set. nodes := s.Cluster.FragmentNodes(s.Fragment.DB(), s.Fragment.Slice()) if len(nodes) == 1 { - //fmt.Println("no place to replicate", s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice()) return nil } diff --git a/frame.go b/frame.go index efbf5f13c..fa47517c5 100644 --- a/frame.go +++ b/frame.go @@ -46,7 +46,12 @@ type Frame struct { } // NewFrame returns a new instance of frame. -func NewFrame(path, db, name string) *Frame { +func NewFrame(path, db, name string) (*Frame, error) { + err := ValidateName(name) + if err != nil { + return nil, err + } + return &Frame{ path: path, db: db, @@ -60,7 +65,7 @@ func NewFrame(path, db, name string) *Frame { rowLabel: DefaultRowLabel, LogOutput: ioutil.Discard, - } + }, nil } // Name returns the name the frame was initialized with. diff --git a/frame_test.go b/frame_test.go index 186db0c62..042bd751e 100644 --- a/frame_test.go +++ b/frame_test.go @@ -65,8 +65,11 @@ func NewFrame() *Frame { if err != nil { panic(err) } - - return &Frame{Frame: pilosa.NewFrame(path, "d", "f")} + frame, err := pilosa.NewFrame(path, "d", "f") + if err != nil { + panic(err) + } + return &Frame{Frame: frame} } // MustOpenFrame returns a new, opened frame at a temporary path. Panic on error. @@ -86,15 +89,31 @@ func (f *Frame) Close() error { // Reopen closes the database and reopens it. func (f *Frame) Reopen() error { + var err error if err := f.Frame.Close(); err != nil { return err } path, db, name := f.Path(), f.DB(), f.Name() - f.Frame = pilosa.NewFrame(path, db, name) + f.Frame, err = pilosa.NewFrame(path, db, name) + if err != nil { + return err + } if err := f.Open(); err != nil { return err } return nil } + +// NewFrame returns a new instance of Frame d/0. +func TestFrame_NameRestriction(t *testing.T) { + path, err := ioutil.TempDir("", "pilosa-frame-") + if err != nil { + panic(err) + } + frame, err := pilosa.NewFrame(path, "d", "ABC") + if frame != nil { + t.Fatalf("unexpected frame name %s", err) + } +} diff --git a/glide.lock b/glide.lock index af94d5502..001257bf1 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 469de49a1736f34a11e9b0e490f7c1da1d8cb0219fed4bf3ad9e71344ca7f58a -updated: 2017-02-09T17:03:01.816613507-06:00 +hash: 7de62dbaf3cc1dc4959f4f6d8213102cb182b4dd7a87b3ac29260ad6bc1b0cef +updated: 2017-03-03T12:25:48.088390296-06:00 imports: - name: github.com/boltdb/bolt version: 4b1ebc1869ad66568b313d0dc410e2be72670dda @@ -13,6 +13,8 @@ imports: version: 346938d642f2ec3594ed81d874461961cd0faa76 subpackages: - spew +- name: github.com/fsnotify/fsnotify + version: 7d7316ed6e1ed2de075aab8dfc76de5d158d66e1 - name: github.com/gogo/protobuf version: a9cd0c35b97daf74d0ebf3514c5254814b2703b4 subpackages: @@ -23,10 +25,54 @@ imports: - lru - name: github.com/golang/protobuf version: 888eb0692c857ec880338addf316bd662d5e630e + subpackages: + - proto +- name: github.com/hashicorp/hcl + version: 630949a3c5fa3c613328e1b8256052cbc2327c9b + subpackages: + - hcl/ast + - hcl/parser + - hcl/scanner + - hcl/strconv + - hcl/token + - json/parser + - json/scanner + - json/token +- name: github.com/inconshreveable/mousetrap + version: 76626ae9c91c4f2a10f34cad8ce83ea42c93bb75 +- name: github.com/magiconair/properties + version: b3b15ef068fd0b17ddf408a23669f20811d194d2 +- name: github.com/mitchellh/mapstructure + version: db1efb556f84b25a0a13a04aad883943538ad2e0 +- name: github.com/pelletier/go-buffruneio + version: c37440a7cf42ac63b919c752ca73a85067e05992 +- name: github.com/pelletier/go-toml + version: 13d49d4606eb801b8f01ae542b4afc4c6ee3d84a - name: github.com/satori/go.uuid version: 879c5887cd475cd7864858769793b2ceb0d44feb +- name: github.com/spf13/afero + version: 9be650865eab0c12963d8753212f4f9c66cdcf12 + subpackages: + - mem +- name: github.com/spf13/cast + version: 4f1683a2242a92e62d6ff705a30e435cbf2b50a3 +- name: github.com/spf13/cobra + version: fcd0c5a1df88f5d6784cb4feead962c3f3d0b66c +- name: github.com/spf13/jwalterweatherman + version: fa7ca7e836cf3a8bb4ebf799f472c12d7e903d66 +- name: github.com/spf13/pflag + version: 9ff6c6923cfffbcd502984b8e0c80539a94968b7 +- name: github.com/spf13/viper + version: 7538d73b4eb9511d85a9f1dfef202eeb8ac260f4 - name: golang.org/x/sys version: c200b10b5d5e122be351b67af224adc6128af5bf subpackages: - unix +- name: golang.org/x/text + version: 5a42fa2464759cbb7ee0af9de00b54d69f09a29c + subpackages: + - transform + - unicode/norm +- name: gopkg.in/yaml.v2 + version: a3f3340b5840cee44f372bddb5880fcbc419b46a testImports: [] diff --git a/glide.yaml b/glide.yaml index 10c8509d0..1c4f35dae 100644 --- a/glide.yaml +++ b/glide.yaml @@ -27,3 +27,5 @@ import: - package: github.com/golang/protobuf - package: github.com/satori/go.uuid version: ^1.1.0 +- package: github.com/spf13/cobra +- package: github.com/spf13/viper diff --git a/handler.go b/handler.go index 381cfeab7..1ccdf3711 100644 --- a/handler.go +++ b/handler.go @@ -205,7 +205,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) } - h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), time.Since(t).Seconds()) + dif := time.Since(t).Seconds() + if dif > 90 { + h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), dif) + } } // handleGetSchema handles GET /schema requests. diff --git a/index.go b/index.go index 7885d87dd..ee4710037 100644 --- a/index.go +++ b/index.go @@ -78,7 +78,10 @@ func (i *Index) Open() error { i.logger().Printf("opening database: %s", filepath.Base(fi.Name())) - db := i.newDB(i.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + db, err := i.newDB(i.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + if err != nil { + return ErrName + } if err := db.Open(); err != nil { return fmt.Errorf("open db: name=%s, err=%s", db.Name(), err) } @@ -193,7 +196,11 @@ func (i *Index) createDB(name string, opt DBOptions) (*DB, error) { } // Otherwise create a new database. - db := i.newDB(i.DBPath(name), name) + db, err := i.newDB(i.DBPath(name), name) + if err != nil { + return nil, err + } + if err := db.Open(); err != nil { return nil, err } @@ -208,11 +215,14 @@ func (i *Index) createDB(name string, opt DBOptions) (*DB, error) { return db, nil } -func (i *Index) newDB(path, name string) *DB { - db := NewDB(path, name) +func (i *Index) newDB(path, name string) (*DB, error) { + db, err := NewDB(path, name) + if err != nil { + return nil, err + } db.LogOutput = i.LogOutput db.stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name())) - return db + return db, nil } // DeleteDB removes a database from the index. diff --git a/pilosa.go b/pilosa.go index a136cb56f..b3620a900 100644 --- a/pilosa.go +++ b/pilosa.go @@ -4,6 +4,7 @@ import ( "errors" "github.com/pilosa/pilosa/internal" + "regexp" ) // System errors. @@ -18,10 +19,18 @@ var ( ErrFrameExists = errors.New("frame already exists") ErrFrameNotFound = errors.New("frame not found") + // ErrFrameRequired is returned when no frame is specified. + ErrName = errors.New("name restricted to [a-z0-9_-]") + + // ErrFragmentNotFound is returned when a fragment does not exist. ErrFragmentNotFound = errors.New("fragment not found") ErrQueryRequired = errors.New("query required") ) +// Regular expression to valuate db and frame's name +// Todo: remove . when frame doesn't require . for topN +var nameRegexp = regexp.MustCompile(`^([a-z0-9._-]{1,64}$)`) + // Profile represents vertical column in a database. // A profile can have a set of attributes attached to it. type Profile struct { @@ -74,3 +83,13 @@ func decodeProfile(pb *internal.Profile) *Profile { // TimeFormat is the go-style time format used to parse string dates. const TimeFormat = "2006-01-02T15:04" + + +// Restrict name using regex +func ValidateName(name string) error { + validName := nameRegexp.Match([]byte(name)) + if validName == false{ + return ErrName + } + return nil +} \ No newline at end of file diff --git a/roaring/roaring.go b/roaring/roaring.go index 2a0dc57d1..e74739311 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -444,33 +444,56 @@ func (b *Bitmap) removeEmptyContainers() { i++ } } +func (b *Bitmap) countEmptyContainers() int { + result := 0 + for i := 0; i < len(b.containers); { + c := b.containers[i] + + if c.n == 0 { + result++ + } + i++ + } + return result +} // WriteTo writes b to w. func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) { // Remove empty containers before persisting. - b.removeEmptyContainers() + //b.removeEmptyContainers() + containerCount := len(b.keys) - b.countEmptyContainers() // Build header before writing individual container blocks. - buf := make([]byte, headerSize+(len(b.keys)*(4+8+4))) + buf := make([]byte, headerSize+(containerCount*(4+8+4))) binary.LittleEndian.PutUint32(buf[0:], cookie) - binary.LittleEndian.PutUint32(buf[4:], uint32(len(b.keys))) - + binary.LittleEndian.PutUint32(buf[4:], uint32(containerCount)) + empty := 0 // Encode keys and cardinality. for i, key := range b.keys { c := b.containers[i] // Verify container count before writing. - count := c.count() - assert(c.count() == c.n, "cannot write container count, mismatch: count=%d, n=%d", count, c.n) - - binary.LittleEndian.PutUint64(buf[headerSize+i*12:], uint64(key)) - binary.LittleEndian.PutUint32(buf[headerSize+i*12+8:], uint32(c.n-1)) + // TODO: instead of commenting this out, we need to make it a configuration option + //count := c.count() + //assert(c.count() == c.n, "cannot write container count, mismatch: count=%d, n=%d", count, c.n) + if c.n > 0 { + binary.LittleEndian.PutUint64(buf[headerSize+(i-empty)*12:], uint64(key)) + binary.LittleEndian.PutUint32(buf[headerSize+(i-empty)*12+8:], uint32(c.n-1)) + } else { + empty++ + } } // Write the offset for each container block. offset := uint32(len(buf)) + empty = 0 for i, c := range b.containers { - binary.LittleEndian.PutUint32(buf[headerSize+(len(b.keys)*12)+(i*4):], uint32(offset)) + + if c.n > 0 { + binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset)) + } else { + empty++ + } offset += uint32(c.size()) } @@ -483,10 +506,12 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) { // Write each container block. for _, c := range b.containers { - nn, err := c.WriteTo(w) - n += nn - if err != nil { - return n, err + if c.n > 0 { + nn, err := c.WriteTo(w) + n += nn + if err != nil { + return n, err + } } } @@ -532,9 +557,10 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error { c := b.containers[i] if c.n <= ArrayMaxSize { c.array = (*[0xFFFFFFF]uint32)(unsafe.Pointer(&data[offset]))[:c.n] - for _, v := range c.array { - assert(lowbits(uint64(v)) == v, "array value out of range: %d", v) - } + // TODO: instead of commenting this out, we need to make it a configuration option + //for _, v := range c.array { + // assert(lowbits(uint64(v)) == v, "array value out of range: %d", v) + //} opsOffset = int(offset) + len(c.array)*4 } else { c.bitmap = (*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN] @@ -542,8 +568,9 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error { } // Verify container count on load. - count := c.count() - assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n) + // TODO: instead of commenting this out, we need to make it a configuration option + //count := c.count() + //assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n) } // Read ops log until the end of the file. @@ -1074,9 +1101,10 @@ func (c *container) arrayWriteTo(w io.Writer) (n int64, err error) { } // Verify all elements are valid. - for _, v := range c.array { - assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v) - } + // TODO: instead of commenting this out, we need to make it a configuration option + //for _, v := range c.array { + // assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v) + //} nn, err := w.Write((*[0xFFFFFFF]byte)(unsafe.Pointer(&c.array[0]))[:4*c.n]) return int64(nn), err diff --git a/server/server.go b/server/server.go new file mode 100644 index 000000000..845340485 --- /dev/null +++ b/server/server.go @@ -0,0 +1,131 @@ +package server + +import ( + "errors" + "fmt" + "io" + "math/rand" + "os" + "path/filepath" + "strings" + "time" + + "github.com/BurntSushi/toml" + "github.com/pilosa/pilosa" +) + +// Version and BuildTime hold the version/build time information passed in at compile time. +var ( + Version string + BuildTime string +) + +func init() { + if Version == "" { + Version = "v0.0.0" + } + if BuildTime == "" { + BuildTime = "not recorded" + } + + rand.Seed(time.Now().UTC().UnixNano()) +} + +const ( + // DefaultDataDir is the default data directory. + DefaultDataDir = "~/.pilosa" +) + +// Command represents the state of the pilosa server command. +type Command struct { + Server *pilosa.Server + + // Configuration options. + ConfigPath string + Config *pilosa.Config + + // Profiling options. + CPUProfile string + CPUTime time.Duration + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewMain returns a new instance of Main. +func NewCommand() *Command { + return &Command{ + Server: pilosa.NewServer(), + Config: pilosa.NewConfig(), + + Stdin: os.Stdin, + Stdout: os.Stdout, + Stderr: os.Stderr, + } +} + +// Run executes the pilosa server. +func (m *Command) Run(args ...string) error { + // Notify user of config file. + if m.ConfigPath != "" { + fmt.Fprintf(m.Stdout, "Using config: %s\n", m.ConfigPath) + } + + // Setup logging output. + m.Server.LogOutput = m.Stderr + + // Configure index. + fmt.Fprintf(m.Stderr, "Using data from: %s\n", m.Config.DataDir) + m.Server.Index.Path = m.Config.DataDir + m.Server.Index.Stats = pilosa.NewExpvarStatsClient() + + // Build cluster from config file. + m.Server.Host = m.Config.Host + m.Server.Cluster = m.Config.PilosaCluster() + + // Set configuration options. + m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval) + + // Initialize server. + if err := m.Server.Open(); err != nil { + return err + } + + fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Server.Host) + + return nil +} + +// Close shuts down the server. +func (m *Command) Close() error { + return m.Server.Close() +} + +// SetupConfig loads the config file if specified and sets state on the Command. +func (m *Command) SetupConfig(args []string) error { + // Load config, if specified. + if m.ConfigPath != "" { + if _, err := toml.DecodeFile(m.ConfigPath, &m.Config); err != nil { + return err + } + } + + // Use default data directory if one is not specified. + if m.Config.DataDir == "" { + m.Config.DataDir = DefaultDataDir + } + + // Expand home directory. + prefix := "~" + string(filepath.Separator) + if strings.HasPrefix(m.Config.DataDir, prefix) { + HomeDir := os.Getenv("HOME") + if HomeDir == "" { + return errors.New("data directory not specified and no home dir available") + } + m.Config.DataDir = filepath.Join(HomeDir, strings.TrimPrefix(m.Config.DataDir, prefix)) + } + + return nil +} diff --git a/cmd/pilosa/main_test.go b/server/server_test.go similarity index 96% rename from cmd/pilosa/main_test.go rename to server/server_test.go index 184f5c334..9e907522a 100644 --- a/cmd/pilosa/main_test.go +++ b/server/server_test.go @@ -1,4 +1,4 @@ -package main_test +package server_test import ( "bytes" @@ -18,7 +18,7 @@ import ( "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" - main "github.com/pilosa/pilosa/cmd/pilosa" + "github.com/pilosa/pilosa/server" ) // Ensure program can process queries and maintain consistency. @@ -304,7 +304,7 @@ path = "/path/to/plugins" // Main represents a test wrapper for main.Main. type Main struct { - *main.Main + *server.Command Stdin bytes.Buffer Stdout bytes.Buffer @@ -318,16 +318,16 @@ func NewMain() *Main { panic(err) } - m := &Main{Main: main.NewMain()} + m := &Main{Command: server.NewCommand()} m.Config.DataDir = path m.Config.Host = "localhost:0" - m.Main.Stdin = &m.Stdin - m.Main.Stdout = &m.Stdout - m.Main.Stderr = &m.Stderr + m.Command.Stdin = &m.Stdin + m.Command.Stdout = &m.Stdout + m.Command.Stderr = &m.Stderr if testing.Verbose() { - m.Main.Stdout = io.MultiWriter(os.Stdout, m.Main.Stdout) - m.Main.Stderr = io.MultiWriter(os.Stderr, m.Main.Stderr) + m.Command.Stdout = io.MultiWriter(os.Stdout, m.Command.Stdout) + m.Command.Stderr = io.MultiWriter(os.Stderr, m.Command.Stderr) } return m @@ -345,18 +345,18 @@ func MustRunMain() *Main { // Close closes the program and removes the underlying data directory. func (m *Main) Close() error { defer os.RemoveAll(m.Config.DataDir) - return m.Main.Close() + return m.Command.Close() } // Reopen closes the program and reopens it. func (m *Main) Reopen() error { - if err := m.Main.Close(); err != nil { + if err := m.Command.Close(); err != nil { return err } // Create new main with the same config. config := m.Config - m.Main = main.NewMain() + m.Command = server.NewCommand() m.Config = config // Run new program.