From d1f3b58861b83f0cadd0c047b0a979c0ba3e5738 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Thu, 3 Feb 2022 11:25:08 -0600 Subject: [PATCH] remove inspect command --- api.go | 4 - cmd/convert.go | 43 --- cmd/inspect_test.go | 29 -- cmd/root.go | 1 - ctl/inspect.go | 394 ----------------------- ctl/inspect_test.go | 48 --- fragment.go | 12 - handler.go | 28 -- holder.go | 437 -------------------------- holder_internal_test.go | 150 --------- internal/clustertests/cluster_test.go | 1 - 11 files changed, 1147 deletions(-) delete mode 100644 cmd/convert.go delete mode 100644 cmd/inspect_test.go delete mode 100644 ctl/inspect.go delete mode 100644 ctl/inspect_test.go diff --git a/api.go b/api.go index 5766701a9..45892dce1 100644 --- a/api.go +++ b/api.go @@ -2307,10 +2307,6 @@ func (api *API) Info() serverInfo { } } -func (api *API) Inspect(ctx context.Context, req *InspectRequest) (*HolderInfo, error) { - return api.holder.Inspect(ctx, req) -} - // GetTranslateEntryReader provides an entry reader for key translation logs starting at offset. func (api *API) GetTranslateEntryReader(ctx context.Context, offsets TranslateOffsetMap) (_ TranslateEntryReader, err error) { span, ctx := tracing.StartSpanFromContext(ctx, "API.GetTranslateEntryReader") diff --git a/cmd/convert.go b/cmd/convert.go deleted file mode 100644 index 2a9f5711a..000000000 --- a/cmd/convert.go +++ /dev/null @@ -1,43 +0,0 @@ -// Copyright 2021 Molecula Corp. All rights reserved. -package cmd - -import ( - "context" - "fmt" - "io" - - "github.com/spf13/cobra" - - "github.com/molecula/featurebase/v3/ctl" -) - -var inspector *ctl.InspectCommand - -func newInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command { - inspector = ctl.NewInspectCommand(stdin, stdout, stderr) - - inspectCmd := &cobra.Command{ - Use: "inspect", - Short: "Get stats on a FeatureBase data file.", - Long: ` -Inspects a data file and provides stats. -`, - RunE: func(cmd *cobra.Command, args []string) error { - if len(args) == 0 { - return fmt.Errorf("path required") - } else if len(args) > 1 { - return fmt.Errorf("only one path allowed") - } - inspector.Path = args[0] - return inspector.Run(context.Background()) - }, - } - flags := inspectCmd.Flags() - flags.BoolVarP(&inspector.Quiet, "quiet", "q", false, "don't list details of containers") - flags.IntVarP(&inspector.Max, "max", "n", 0, "list at most max items (0 = unlimited)") - flags.StringVarP(&inspector.InspectOpts.Indexes, "index", "i", "", "filter indexes") - flags.StringVarP(&inspector.InspectOpts.Views, "view", "v", "", "filter views") - flags.StringVarP(&inspector.InspectOpts.Fields, "field", "f", "", "filter fields") - flags.StringVarP(&inspector.InspectOpts.Shards, "shard", "s", "", "filter shards") - return inspectCmd -} diff --git a/cmd/inspect_test.go b/cmd/inspect_test.go deleted file mode 100644 index 33dd616d6..000000000 --- a/cmd/inspect_test.go +++ /dev/null @@ -1,29 +0,0 @@ -// Copyright 2021 Molecula Corp. All rights reserved. -package cmd_test - -import ( - "strings" - "testing" -) - -func TestInspectHelp(t *testing.T) { - output, err := ExecNewRootCommand(t, "inspect", "--help") - if !strings.Contains(output, "Usage:") || - !strings.Contains(output, "featurebase inspect") || err != nil { - t.Fatalf("Command 'inspect --help' not working, err: '%v', output: '%s'", err, output) - } -} - -func TestInspectNoPath(t *testing.T) { - output, err := ExecNewRootCommand(t, "inspect") - if !strings.Contains(err.Error(), "path required") { - t.Fatalf("Command 'inspect' without args should error but: err: '%v', output: '%v'", err, output) - } -} - -func TestInspectMultiPath(t *testing.T) { - output, err := ExecNewRootCommand(t, "inspect", "one", "two") - if !strings.Contains(err.Error(), "only one path") { - t.Fatalf("Command 'inspect' without args should error but: err: '%v', output: '%v'", err, output) - } -} diff --git a/cmd/root.go b/cmd/root.go index c6c6492b3..5feee57d0 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -57,7 +57,6 @@ at https://docs.molecula.cloud/. rc.AddCommand(newExportCommand(stdin, stdout, stderr)) rc.AddCommand(newGenerateConfigCommand(stdin, stdout, stderr)) rc.AddCommand(newImportCommand(stdin, stdout, stderr)) - rc.AddCommand(newInspectCommand(stdin, stdout, stderr)) rc.AddCommand(newRBFCommand(stdin, stdout, stderr)) rc.AddCommand(newServeCmd(stdin, stdout, stderr)) rc.AddCommand(newHolderCmd(stdin, stdout, stderr)) diff --git a/ctl/inspect.go b/ctl/inspect.go deleted file mode 100644 index 73d197501..000000000 --- a/ctl/inspect.go +++ /dev/null @@ -1,394 +0,0 @@ -// Copyright 2021 Molecula Corp. All rights reserved. -package ctl - -import ( - "context" - "encoding/binary" - "fmt" - "hash/fnv" - "io" - "io/ioutil" - "os" - "path/filepath" - "sort" - "strconv" - "strings" - "syscall" - "text/tabwriter" - "time" - "unsafe" - - "github.com/gogo/protobuf/proto" - "github.com/molecula/featurebase/v3" - "github.com/molecula/featurebase/v3/pb" - "github.com/molecula/featurebase/v3/roaring" - "github.com/pkg/errors" -) - -// InspectCommand represents a command for inspecting fragment data files. -type InspectCommand struct { - // Path to data file - Path string - // don't list details of objects - Quiet bool - // list only this many objects - Max int - // Filters: - InspectOpts pilosa.InspectRequest - - // Standard input/output - *pilosa.CmdIO -} - -// NewInspectCommand returns a new instance of InspectCommand. -func NewInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *InspectCommand { - return &InspectCommand{ - CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), - } -} - -type pointerContext struct { - from, to uintptr -} - -func (p *pointerContext) pretty(c roaring.ContainerInfo) string { - var pointer string - if c.Mapped { - if c.Pointer >= p.from && c.Pointer < p.to { - pointer = fmt.Sprintf("@+0x%x", c.Pointer-p.from) - } else { - pointer = fmt.Sprintf("!0x%x!", c.Pointer) - } - } else { - pointer = fmt.Sprintf("0x%x", c.Pointer) - } - return fmt.Sprintf("%s \t%d \t%d \t%s ", c.Type, c.N, c.Alloc, pointer) -} - -func (cmd *InspectCommand) PrintOps(info roaring.BitmapInfo) { - fmt.Fprintln(cmd.Stdout, " Ops:") - tw := tabwriter.NewWriter(cmd.Stdout, 0, 8, 0, '\t', 0) - fmt.Fprintf(tw, " \t%s\t%s\t%s\t\n", "TYPE", "OpN", "SIZE") - printed := 0 - for _, op := range info.OpDetails { - fmt.Fprintf(tw, "\t%s\t%d\t%d\t\n", op.Type, op.OpN, op.Size) - printed++ - if cmd.Max != 0 && printed >= cmd.Max { - break - } - } - tw.Flush() -} - -func (cmd *InspectCommand) PrintContainers(info roaring.BitmapInfo, pC pointerContext) { - fmt.Fprintln(cmd.Stdout, " Containers:") - tw := tabwriter.NewWriter(cmd.Stdout, 0, 8, 0, '\t', 0) - fmt.Fprintf(tw, " \t\tRoaring\t\t\t\tOps\t\t\t\tFlags\t\n") - fmt.Fprintf(tw, "\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t\n", "KEY", "TYPE", "N", "ALLOC", "OFFSET", "TYPE", "N", "ALLOC", "OFFSET", "FLAGS") - c1s := info.Containers - c2s := info.OpContainers - l1 := len(c1s) - l2 := len(c2s) - i1 := 0 - i2 := 0 - var c1, c2 roaring.ContainerInfo - c1.Key = ^uint64(0) - c2.Key = ^uint64(0) - c1e := false - c2e := false - if i1 < l1 { - c1 = c1s[i1] - i1++ - c1e = true - } - if i2 < l2 { - c2 = c2s[i2] - i2++ - c2e = true - } - printed := 0 - for c1e || c2e { - c1used := false - c2used := false - var key uint64 - c1fmt := "-\t\t\t" - c2fmt := "-\t\t\t" - // If c2 exists, we'll always prefer its flags, - // if it doesn't, this gets overwritten. - flags := c2.Flags - if !c2e || (c1e && c1.Key < c2.Key) { - c1fmt = pC.pretty(c1) - key = c1.Key - c1used = true - flags = c1.Flags - } else if !c1e || (c2e && c2.Key < c1.Key) { - c2fmt = pC.pretty(c2) - key = c2.Key - c2used = true - } else { - // c1e and c2e both set, and neither key is < the other. - c1fmt = pC.pretty(c1) - c2fmt = pC.pretty(c2) - key = c1.Key - c1used = true - c2used = true - } - if c1used { - if i1 < l1 { - c1 = c1s[i1] - i1++ - } else { - c1e = false - } - } - if c2used { - if i2 < l2 { - c2 = c2s[i2] - i2++ - } else { - c2e = false - } - } - fmt.Fprintf(tw, "\t%d\t%s\t%s\t%s\t\n", key, c1fmt, c2fmt, flags) - printed++ - if cmd.Max > 0 && printed >= cmd.Max { - break - } - } - tw.Flush() -} - -// 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 errors.Wrap(err, "opening file") - } - defer f.Close() - - fi, err := f.Stat() - if err != nil { - return errors.Wrap(err, "statting file") - } - if fi.IsDir() { - total := 0 - infos, err := f.Readdir(0) - if err != nil { - return err - } - if len(infos) == 0 { - return errors.New("directory contains no files") - } - - names := make([]string, len(infos)) - nameToInfo := make(map[string]os.FileInfo, len(infos)) - // find numeric-only names; we'll operate on - // either those, or the whole holder if we find - // a .topology file. - n := 0 - for _, fi := range infos { - name := fi.Name() - if name == ".topology" { - return cmd.InspectHolder(ctx, cmd.Path) - } - if _, err := strconv.Atoi(name); err == nil { - names[n] = name - nameToInfo[name] = fi - n++ - } - } - if n == 0 { - return fmt.Errorf("directory contains no fragments (looking for numeric names)") - } - names = names[:n] - fmt.Fprintf(cmd.Stdout, "%s contains %d fragments:\n", cmd.Path, n) - for _, name := range names { - f2, err := os.Open(filepath.Join(cmd.Path, name)) - if err != nil { - return fmt.Errorf("opening %q: %v", name, err) - } - fmt.Fprintf(cmd.Stdout, "%s/%s:\n", cmd.Path, name) - err = cmd.InspectFile(f2, nameToInfo[name]) - total++ - f2.Close() - if err != nil { - return fmt.Errorf("inspecting %q: %v", name, err) - } - } - return nil - } - return cmd.InspectFile(f, fi) -} - -// loadTopology is copied almost exactly from pilosa/cluster.go. -func loadTopology(path string) (topology pb.Topology, myID string, err error) { - buf, err := ioutil.ReadFile(filepath.Join(path, ".topology")) - if os.IsNotExist(err) { - return topology, myID, err - } else if err != nil { - return topology, myID, errors.Wrap(err, "reading file") - } - if err := proto.Unmarshal(buf, &topology); err != nil { - return topology, myID, errors.Wrap(err, "unmarshalling") - } - sort.Slice(topology.NodeIDs, - func(i, j int) bool { - return topology.NodeIDs[i] < topology.NodeIDs[j] - }) - buf, err = ioutil.ReadFile(filepath.Join(path, ".id")) - if os.IsNotExist(err) { - return topology, myID, err - } else if err != nil { - return topology, myID, nil - } - myID = strings.TrimSpace(string(buf)) - return topology, myID, nil -} - -var partitions = make(map[string]map[uint64]int) - -func findPartition(index string, shard uint64, partitionN int) (partition int) { - var shardMap map[uint64]int - var ok bool - if shardMap, ok = partitions[index]; !ok { - shardMap = make(map[uint64]int) - partitions[index] = shardMap - } - if partition, ok = shardMap[shard]; !ok { - var buf [8]byte - binary.BigEndian.PutUint64(buf[:], shard) - - // Hash the bytes and mod by partition count. - h := fnv.New64a() - _, _ = h.Write([]byte(index)) - _, _ = h.Write(buf[:]) - partition = int(h.Sum64() % uint64(partitionN)) - shardMap[shard] = partition - } - return partition -} - -func findPartitionPath(path string, partitionN int) (int, error) { - parts := strings.Split(path, "/") - shard, err := strconv.ParseUint(parts[len(parts)-1], 10, 64) - if err != nil { - return 0, err - } - return findPartition(parts[0], shard, partitionN), nil -} - -func (cmd *InspectCommand) InspectHolder(ctx context.Context, path string) error { - holder := pilosa.NewHolder(path, nil) - holder.Opts.Inspect = true - holder.Opts.ReadOnly = true - err := holder.Open() - if err != nil { - return fmt.Errorf("%s: holder open: %v", path, err) - } - holderInfo, err := holder.Inspect(ctx, &cmd.InspectOpts) - if err != nil { - return fmt.Errorf("%s: inspect: %v", path, err) - } - myPartition := 0 - topology, myID, err := loadTopology(path) - if err == nil { - fmt.Fprintf(cmd.Stdout, "Cluster ID: %q\n", topology.ClusterID) - if len(topology.NodeIDs) > 1 { - fmt.Fprintf(cmd.Stdout, "Cluster of %d nodes, this node %q\n", len(topology.NodeIDs), myID) - } else { - fmt.Fprintf(cmd.Stdout, "Cluster has only one node: %q\n", myID) - } - found := false - for i := range topology.NodeIDs { - if topology.NodeIDs[i] == myID { - found = true - myPartition = i - break - } - } - if !found { - fmt.Fprintf(cmd.Stdout, "Warning: node ID %q not found in topology (%q)\n", myID, topology.NodeIDs) - } - } else { - fmt.Fprintf(cmd.Stdout, "warning: reading topology failed: %v\n", err) - } - for _, name := range holderInfo.FragmentNames { - partition, err := findPartitionPath(name, len(topology.NodeIDs)) - if err != nil { - fmt.Fprintf(cmd.Stdout, "%s: [can't find partition: %v]\n", name, err) - } else { - if partition == myPartition { - fmt.Fprintf(cmd.Stdout, "%s:\n", name) - } else { - fmt.Fprintf(cmd.Stdout, "%s: [primary node %q]\n", name, topology.NodeIDs[partition]) - } - } - details := holderInfo.FragmentInfo[name] - cmd.DisplayInfo(details.BitmapInfo) - if details.BlockChecksums != nil { - fmt.Fprintf(cmd.Stdout, " Checksums [%d total]:\n", len(details.BlockChecksums)) - for _, block := range details.BlockChecksums { - fmt.Fprintf(cmd.Stdout, " %8d: %x\n", block.ID, block.Checksum) - } - } - } - return nil -} - -func (cmd *InspectCommand) InspectFile(f *os.File, fi os.FileInfo) error { - // 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 errors.Wrap(err, "mmapping") - } - defer func() { - err := syscall.Munmap(data) - if err != nil { - fmt.Fprintf(cmd.Stderr, "inspect command: munmap failed: %v", err) - } - }() - mappedFrom := uintptr(unsafe.Pointer(&data[0])) - mappedTo := mappedFrom + uintptr(len(data)) - // Attach the mmap file to the bitmap. - t := time.Now() - fmt.Fprintf(cmd.Stderr, "inspecting bitmap...") - var info roaring.BitmapInfo - bitmap, _, err := roaring.InspectBinary(data, true, &info) - fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) - cmd.DisplayInfo(info) - if err != nil { - return errors.Wrap(err, "inspecting") - } - mappedIn, mappedOut, unmappedIn, errs, err := bitmap.SanityCheckMapping(mappedFrom, mappedTo) - if err != nil { - fmt.Fprintf(cmd.Stderr, "sanity check: %d mapped in, %d mapped out, %d unmapped in, %d errors\n", - mappedIn, mappedOut, unmappedIn, errs) - fmt.Fprintf(cmd.Stderr, "last error: %v\n", err) - } - return nil -} - -func (cmd *InspectCommand) DisplayInfo(info roaring.BitmapInfo) { - pC := pointerContext{ - from: info.From, - to: info.To, - } - - // Print top-level info. - fmt.Fprintf(cmd.Stdout, " Bitmap Info:\n") - fmt.Fprintf(cmd.Stdout, " Bits: %d\n", info.BitCount) - fmt.Fprintf(cmd.Stdout, " Containers: %d (%d roaring)\n", info.ContainerCount, len(info.Containers)) - fmt.Fprintf(cmd.Stdout, " Operations: %d (%d bits)\n", info.Ops, info.OpN) - fmt.Fprintln(cmd.Stdout, "") - - // Print info for each container. - if !cmd.Quiet { - if info.ContainerCount > 0 { - cmd.PrintContainers(info, pC) - } - if info.Ops > 0 { - cmd.PrintOps(info) - } - } -} diff --git a/ctl/inspect_test.go b/ctl/inspect_test.go deleted file mode 100644 index 3528edbe7..000000000 --- a/ctl/inspect_test.go +++ /dev/null @@ -1,48 +0,0 @@ -// Copyright 2021 Molecula Corp. All rights reserved. -package ctl - -import ( - "bytes" - "context" - "io" - "os" - "strings" - "testing" - - "github.com/molecula/featurebase/v3/testhook" -) - -func TestInspectCommand_Run(t *testing.T) { - rder := []byte{} - stdin := bytes.NewReader(rder) - r, w, _ := os.Pipe() - - cm := NewInspectCommand(stdin, w, w) - file, err := testhook.TempFile(t, "inspectTest") - if err != nil { - t.Fatalf("Error creating tempfile: %s", err) - } - _, err = file.Write([]byte("12358267538963")) - if err != nil { - t.Fatalf("writing to tempfile: %v", err) - } - file.Close() - cm.Path = file.Name() - err = cm.Run(context.Background()) - expectedError := "inspecting: " - if !strings.Contains(err.Error(), expectedError) { - t.Fatalf("expected error '%s', got '%v'", expectedError, err) - } - - w.Close() - var buf bytes.Buffer - _, err = io.Copy(&buf, r) - if err != nil { - t.Fatalf("copying data: %v", err) - } - if !strings.Contains(buf.String(), "inspecting bitmap...") { - t.Fatalf("Inspect doesn't work: %s", err) - } - - // Todo: need correct roaring file for happy path -} diff --git a/fragment.go b/fragment.go index 2997264ba..3e8e45741 100644 --- a/fragment.go +++ b/fragment.go @@ -214,18 +214,6 @@ func (f *fragment) Index() *Index { return f.holder.Index(f.index()) } -func (f *fragment) inspect(params InspectRequestParams) (fi FragmentInfo) { - if f.bitmapInfo == nil { - fi.BitmapInfo = f.storage.Info(params.Containers) - } else { - fi.BitmapInfo = *f.bitmapInfo - } - if params.Checksum { - fi.BlockChecksums, _ = f.Blocks() - } - return fi -} - // Open opens the underlying storage. func (f *fragment) Open() error { f.mu.Lock() diff --git a/handler.go b/handler.go index 725225f5f..c74296b45 100644 --- a/handler.go +++ b/handler.go @@ -410,31 +410,3 @@ type TranslateIDsRequest struct { type TranslateIDsResponse struct { Keys []string } - -// InspectRequestParams represents the parts of an InspectRequest that -// aren't generic holder filtering attributes. -type InspectRequestParams struct { - Containers bool // include container details - Checksum bool // perform checksums -} - -// InspectRequest represents a request for a possibly-partial -// holder inspection, using a provided holder filter and inspect-specific -// parameters. -type InspectRequest struct { - HolderFilterParams - InspectRequestParams -} - -// InspectResponse contains the structured results for an InspectRequest. -// It may some day be expanded to include metadata about views or indexes. -type InspectResponse struct { - Fragments []struct { - Index string - Field string - View string - Shard int64 - Path string - Info *FragmentInfo - } -} diff --git a/holder.go b/holder.go index 1c4d3c940..18cd7154c 100644 --- a/holder.go +++ b/holder.go @@ -7,10 +7,8 @@ import ( "fmt" "os" "path/filepath" - "regexp" "runtime" "sort" - "strconv" "strings" "sync" "time" @@ -137,14 +135,6 @@ type Holder struct { // HolderOpts holds information about the holder which other things might want // to look up later while using the holder. type HolderOpts struct { - // ReadOnly indicates that this holder's contents should not produce - // disk writes under any circumstances. It must be set before Open - // is called, and changing it is not supported. - ReadOnly bool - // If Inspect is set, we'll try to obtain additional information - // about fragments when opening them. - Inspect bool - // StorageBackend controls the tx/storage engine we instatiate. Set by // server.go OptServerStorageConfig StorageBackend string @@ -295,280 +285,6 @@ func (h *Holder) IndexesPath() string { return filepath.Join(h.path, IndexesDir) } -type HolderInfo struct { - FragmentInfo map[string]FragmentInfo - FragmentNames []string -} - -type regexpList []*regexp.Regexp - -func newRegexpList(regexes string) (results regexpList, err error) { - if regexes == "" { - return nil, nil - } - for _, sub := range strings.Split(regexes, ",") { - re, err := regexp.Compile(sub) - if err != nil { - return nil, err - } - results = append(results, re) - } - return results, nil -} - -func (rl regexpList) Match(haystack string) bool { - if rl == nil { - return true - } - for _, re := range rl { - if re.MatchString(haystack) { - return true - } - } - return false -} - -// shardRange represents a series of shards -type shardRange struct { - min, max uint64 -} - -type shardRangeList []shardRange - -func newShardRangeList(shards string) (results shardRangeList, err error) { - if shards == "" { - return nil, nil - } - for _, sub := range strings.Split(shards, ",") { - var sr shardRange - minMax := strings.Split(sub, "-") - if len(minMax) > 2 { - return nil, fmt.Errorf("invalid range %q", sub) - } - sr.min, err = strconv.ParseUint(minMax[0], 10, 64) - if err != nil { - return nil, err - } - sr.max = sr.min - if len(minMax) == 2 { - sr.max, err = strconv.ParseUint(minMax[0], 10, 64) - if err != nil { - return nil, err - } - } - if sr.max < sr.min { - return nil, fmt.Errorf("invalid range %q: max < min", sub) - } - results = append(results, sr) - } - return results, nil -} - -func (sl shardRangeList) Match(shard uint64) bool { - if sl == nil { - return true - } - for _, sr := range sl { - if shard >= sr.min && shard <= sr.max { - return true - } - } - return false -} - -// HolderFilter represents something that potentially filters out -// parts of a holder, indicating whether or not to process them, -// or recurse into them. It is permissible to recurse a thing -// without processing it, or process it without recursing it. -// For instance, something looking to accumulate statistics -// about views might return (true, false) from CheckView, -// while a fragment scanning operation would return (false, true) -// from everything above CheckFrag. -type HolderFilter interface { - CheckIndex(iname string) (process bool, recurse bool) - CheckField(iname, fname string) (process bool, recurse bool) - CheckView(iname, fname, vname string) (process bool, recurse bool) - CheckFragment(iname, fname, vname string, shard uint64) (process bool) -} - -// HolderFilterAll is a placeholder type which always returns true for the -// check functions. You can embed it to make a HolderOperator which processes -// everything. -type HolderFilterAll struct{} - -func (HolderFilterAll) CheckIndex(string) (bool, bool) { - return true, true -} - -func (HolderFilterAll) CheckField(string, string) (bool, bool) { - return true, true -} - -func (HolderFilterAll) CheckView(string, string, string) (bool, bool) { - return true, true -} - -func (HolderFilterAll) CheckFragment(string, string, string, uint64) bool { - return true -} - -// HolderProcessNone is a placeholder type which does nothing for the -// process functions. You can embed it to make a HolderOperator which -// does nothing, or embed it and provide your own ProcessFragment to -// do just that. -type HolderProcessNone struct{} - -func (HolderProcessNone) ProcessIndex(*Index) error { - return nil -} - -func (HolderProcessNone) ProcessField(*Field) error { - return nil -} - -func (HolderProcessNone) ProcessView(*view) error { - return nil -} - -func (HolderProcessNone) ProcessFragment(*fragment) error { - return nil -} - -// HolderProcess represents something that has operations which can be -// performed on indexes, fields, views, and/or fragments. -type HolderProcess interface { - ProcessIndex(*Index) error - ProcessField(*Field) error - ProcessView(*view) error - ProcessFragment(*fragment) error -} - -// HolderOperator is both a filter and a process. This is the general -// form of "I want to do something to some part of a holder." -type HolderOperator interface { - HolderFilter - HolderProcess -} - -var _ HolderOperator = (*holderInspector)(nil) - -type HolderFilterParams struct { - Indexes string - Fields string - Views string - Shards string -} - -type holderFilterFull struct { - HolderFilterParams - indexRegexps regexpList - fieldRegexps regexpList - viewRegexps regexpList - shardRanges shardRangeList -} - -type inspectRequestFull struct { - HolderFilter - params InspectRequestParams -} - -func (i *holderFilterFull) CheckIndex(iname string) (process, recurse bool) { - return true, i.indexRegexps.Match(iname) -} - -func (i *holderFilterFull) CheckField(iname, fname string) (process, recurse bool) { - return true, i.fieldRegexps.Match(fname) -} - -func (i *holderFilterFull) CheckView(iname, fname, vname string) (process, recurse bool) { - return true, i.viewRegexps.Match(vname) -} - -func (i *holderFilterFull) CheckFragment(iname, fname, vname string, shard uint64) (process bool) { - return i.shardRanges.Match(shard) -} - -func NewHolderFilter(params HolderFilterParams) (result HolderFilter, err error) { - filter := &holderFilterFull{ - HolderFilterParams: params, - } - filter.indexRegexps, err = newRegexpList(params.Indexes) - if err != nil { - return nil, err - } - filter.fieldRegexps, err = newRegexpList(params.Fields) - if err != nil { - return nil, err - } - filter.viewRegexps, err = newRegexpList(params.Views) - if err != nil { - return nil, err - } - filter.shardRanges, err = newShardRangeList(params.Shards) - if err != nil { - return nil, err - } - return filter, nil -} - -func expandInspectRequest(req *InspectRequest) (*inspectRequestFull, error) { - filter, err := NewHolderFilter(req.HolderFilterParams) - if err != nil { - return nil, err - } - irf := &inspectRequestFull{ - HolderFilter: filter, - params: req.InspectRequestParams, - } - return irf, nil -} - -type holderInspector struct { - *inspectRequestFull - pathParts [3]string - path string - hi *HolderInfo -} - -func (h *holderInspector) ProcessIndex(i *Index) error { - h.pathParts[0] = i.name - return nil -} - -func (h *holderInspector) ProcessField(f *Field) error { - h.pathParts[1] = f.name - return nil -} - -func (h *holderInspector) ProcessView(v *view) error { - h.pathParts[2] = v.name - h.path = strings.Join(h.pathParts[:], "/") - return nil -} - -func (h *holderInspector) ProcessFragment(f *fragment) error { - path := h.path + "/" + strconv.FormatUint(f.shard, 10) - h.hi.FragmentInfo[path] = f.inspect(h.inspectRequestFull.params) - h.hi.FragmentNames = append(h.hi.FragmentNames, path) - return nil -} - -func (h *Holder) Inspect(ctx context.Context, req *InspectRequest) (*HolderInfo, error) { - fullReq, err := expandInspectRequest(req) - if err != nil { - return nil, err - } - inspector := &holderInspector{ - inspectRequestFull: fullReq, - hi: &HolderInfo{ - FragmentInfo: make(map[string]FragmentInfo), - }, - } - err = h.Process(ctx, inspector) - sort.Strings(inspector.hi.FragmentNames) - return inspector.hi, err -} - // Open initializes the root data directory for the holder. func (h *Holder) Open() error { h.opening = true @@ -773,7 +489,6 @@ func (h *Holder) processForeignIndexFields() error { // Close closes all open fragments. func (h *Holder) Close() error { - if h == nil { return nil } @@ -1951,130 +1666,6 @@ func uint64InSlice(i uint64, s []uint64) bool { return false } -// Process loops through a holder based on the Check functions in op, calling -// the Process functions in op when indicated. -func (h *Holder) Process(ctx context.Context, op HolderOperator) (err error) { - var fieldNames, viewNames []string - var fragNums []uint64 - - indexes := h.Indexes() - for _, idx := range indexes { - if err = ctx.Err(); err != nil { - return err - } - if idx == nil { - continue - } - indexName := idx.name - process, recurse := op.CheckIndex(indexName) - if !process && !recurse { - continue - } - - if err = ctx.Err(); err != nil { - return err - } - if process { - err = op.ProcessIndex(idx) - if err != nil { - return err - } - } - if !recurse { - continue - } - fieldNames = fieldNames[:0] - idx.mu.Lock() - for fieldName := range idx.fields { - fieldNames = append(fieldNames, fieldName) - } - idx.mu.Unlock() - for _, fieldName := range fieldNames { - if err = ctx.Err(); err != nil { - return err - } - process, recurse := op.CheckField(idx.name, fieldName) - if !process && !recurse { - continue - } - idx.mu.Lock() - field := idx.fields[fieldName] - idx.mu.Unlock() - if field == nil { - continue - } - if err = ctx.Err(); err != nil { - return err - } - if process { - err = op.ProcessField(field) - if err != nil { - return err - } - } - if !recurse { - continue - } - viewNames = viewNames[:0] - field.mu.Lock() - for viewName := range field.viewMap { - viewNames = append(viewNames, viewName) - } - field.mu.Unlock() - for _, viewName := range viewNames { - if err = ctx.Err(); err != nil { - return err - } - process, recurse := op.CheckView(indexName, fieldName, viewName) - if !process && !recurse { - continue - } - field.mu.Lock() - view := field.viewMap[viewName] - field.mu.Unlock() - if view == nil { - continue - } - if err = ctx.Err(); err != nil { - return err - } - if process { - err = op.ProcessView(view) - if err != nil { - return err - } - } - if !recurse { - continue - } - fragNums = fragNums[:0] - view.mu.Lock() - for fragNum := range view.fragments { - fragNums = append(fragNums, fragNum) - } - view.mu.Unlock() - for _, fragNum := range fragNums { - if err = ctx.Err(); err != nil { - return err - } - process := op.CheckFragment(indexName, fieldName, viewName, fragNum) - if !process { - continue - } - view.mu.Lock() - frag := view.fragments[fragNum] - view.mu.Unlock() - err = op.ProcessFragment(frag) - if err != nil { - return err - } - } - } - } - } - return nil -} - // used by Index.openFields(), enabling Tx / Txf by telling // the holder about its own indexes. func (h *Holder) addIndex(idx *Index) { @@ -2095,34 +1686,6 @@ func (h *Holder) BeginTx(writable bool, idx *Index, shard uint64) (Tx, error) { return h.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}), nil } -func (h *Holder) HasRoaringData() (has bool, err error) { - idxs := h.Indexes() - for _, idx := range idxs { - paths, err := listFilesUnderDir(idx.path, false, "", true) - if err != nil { - return false, errors.Wrap(err, "HasRoaringData listFilesUnderDir") - } - index := idx.name - - for _, relpath := range paths { - field, view, shard, err := fragmentSpecFromRoaringPath(relpath) - if err != nil { - continue // ignore .meta paths - } - abspath := idx.path + sep + relpath - - hasData, err := roaringFragmentHasData(abspath, index, field, view, shard) - if err != nil { - return false, errors.Wrap(err, "HasRoaringData roaringFragmentHasData") - } - if hasData { - return true, nil - } - } - } - return -} - func decodeCreateIndexMessage(ser Serializer, b []byte) (*CreateIndexMessage, error) { var cim CreateIndexMessage if err := ser.Unmarshal(b, &cim); err != nil { diff --git a/holder_internal_test.go b/holder_internal_test.go index a762408b8..4204dddc1 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -2,82 +2,11 @@ package pilosa import ( - "context" - "fmt" - "os" "testing" "github.com/molecula/featurebase/v3/disco" - "github.com/molecula/featurebase/v3/testhook" ) -var _ = fmt.Printf - -type testHolderOperator struct { - indexSeen, indexProcessed int - fieldSeen, fieldProcessed int - viewSeen, viewProcessed int - fragmentSeen, fragmentProcessed int - waitHere chan struct{} -} - -func (t *testHolderOperator) CheckIndex(string) (bool, bool) { - t.indexSeen++ - return true, true -} - -func (t *testHolderOperator) CheckField(string, string) (bool, bool) { - t.fieldSeen++ - return true, true -} - -func (t *testHolderOperator) CheckView(string, string, string) (bool, bool) { - t.viewSeen++ - return true, true -} - -func (t *testHolderOperator) CheckFragment(string, string, string, uint64) bool { - t.fragmentSeen++ - return true -} - -func (t *testHolderOperator) ProcessIndex(*Index) error { - t.indexProcessed++ - return nil -} - -func (t *testHolderOperator) ProcessField(*Field) error { - t.fieldProcessed++ - return nil -} - -func (t *testHolderOperator) ProcessView(*view) error { - t.viewProcessed++ - return nil -} - -func (t *testHolderOperator) ProcessFragment(*fragment) error { - if t.waitHere != nil { - <-t.waitHere - } - t.fragmentProcessed++ - return nil -} - -func makeHolder(tb testing.TB, backend string) (*Holder, string, error) { - path, err := testhook.TempDir(tb, "pilosa-") - if err != nil { - return nil, "", err - } - cfg := mustHolderConfig() - if backend != "" { - cfg.StorageConfig.Backend = backend - cfg.StorageConfig.FsyncEnabled = false - } - h := NewHolder(path, cfg) - return h, path, h.Open() -} - func testSetBit(t *testing.T, h *Holder, index, field string, rowID, columnID uint64) { idx, err := h.CreateIndexIfNotExists(index, IndexOptions{}) @@ -95,85 +24,6 @@ func testSetBit(t *testing.T, h *Holder, index, field string, rowID, columnID ui } } -func TestHolderOperatorProcess(t *testing.T) { - h, path, err := makeHolder(t, "") - if err != nil { - t.Fatalf("creating holder: %v", err) - } - defer os.RemoveAll(path) - defer h.Close() - - // Write bits to separate indexes. - testSetBit(t, h, "i0", "f", 100, 200) - testSetBit(t, h, "i1", "f", 100, 200) - testSetBit(t, h, "i1", "f", 100, 12345678) - - testOp := testHolderOperator{} - ctx := context.Background() - err = h.Process(ctx, &testOp) - if err != nil { - t.Fatalf("processing holder: %v", err) - } - expected := testHolderOperator{ - indexSeen: 2, indexProcessed: 2, - fieldSeen: 2, fieldProcessed: 2, - viewSeen: 2, viewProcessed: 2, - fragmentSeen: 3, fragmentProcessed: 3, - } - if testOp != expected { - t.Fatalf("holder processor did not process as expected. expected %#v, got %#v", expected, testOp) - } -} - -func TestHolderOperatorCancel(t *testing.T) { - h, path, err := makeHolder(t, "") - if err != nil { - t.Fatalf("creating holder: %v", err) - } - defer os.RemoveAll(path) - defer h.Close() - - // Write bits to separate indexes. - testSetBit(t, h, "i0", "f", 100, 200) - testSetBit(t, h, "i1", "f", 100, 200) - testSetBit(t, h, "i1", "f", 100, 12345678) - - // Here, we want to ensure that the operation gets cancelled - // successfully. In practice we expect it to process one fragment, then - // end up blocked on the waitHere, then get cancelled... But the - // waitHere blockage isn't really something holder.Process can do - // anything about, so we close the channel, so two fragments are - // processed. But in theory you could end up with only one fragment - // processed if this goroutine managed to cancel before the processor - // gets to the next fragment. Point is, it shouldn't hit all three, - // because the checks against the cancellation should fire before it - // gets there. - testOp := testHolderOperator{waitHere: make(chan struct{})} - ctx, cancel := context.WithCancel(context.Background()) - done := make(chan struct{}) - go func() { - err = h.Process(ctx, &testOp) - close(done) - }() - testOp.waitHere <- struct{}{} - cancel() - close(testOp.waitHere) - <-done - if err != context.Canceled { - t.Fatalf("processing holder: expected context.Canceled, got %v", err) - } - testOp.waitHere = nil - expected := testHolderOperator{ - indexSeen: 2, indexProcessed: 2, - fieldSeen: 2, fieldProcessed: 2, - viewSeen: 2, viewProcessed: 2, - fragmentSeen: 3, fragmentProcessed: 3, - } - if testOp == expected { - t.Fatalf("holder processor did not cancel. expected something other than %#v", expected) - } -} - // mustHolderConfig sets up a default holder config for tests. func mustHolderConfig() *HolderConfig { cfg := DefaultHolderConfig() diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 1d97d85f2..99a653ca9 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -236,7 +236,6 @@ func TestClusterStuff(t *testing.T) { t.Fatalf("restore failed: %v", err) } - fmt.Println("pausing all featurebasen") if err = sendCmd("docker", "pause", container(t, "pilosa1")); err != nil { t.Fatalf("sending pause command: %v", err) }