diff --git a/.circleci/config.yml b/.circleci/config.yml index 498c56ea3..c3e1a43e4 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -144,16 +144,6 @@ jobs: - run: command: make test-external-lookup EXTERNAL_LOOKUP_DSN=postgresql://postgres:password@localhost/circle_test?sslmode=disable no_output_timeout: 30m - test-backup-restore: - executor: - name: golang - steps: - - checkout-plus - - skip-if-root-unchanged - - setup_remote_docker - - run: echo -n $DOCKER_PASS | docker login -u $DOCKER_USER --password-stdin - - run: make backuptests-build - - run: make backuptests cluster-tests: executor: name: golang @@ -273,10 +263,6 @@ workflows: context: molecula requires: - setup - - test-backup-restore: - context: molecula - requires: - - setup - cluster-tests: context: molecula requires: @@ -292,8 +278,6 @@ workflows: filters: tags: only: /^v.*/ - branches: - only: master - publish_release: context: molecula requires: diff --git a/Dockerfile.pilosa b/Dockerfile.pilosa deleted file mode 100644 index 6ee2e0bbf..000000000 --- a/Dockerfile.pilosa +++ /dev/null @@ -1,37 +0,0 @@ -ARG GO_VERSION=latest - -###################### -### Pilosa builder ### -###################### - -FROM golang:${GO_VERSION} as pilosa-builder -ARG MAKE_FLAGS -WORKDIR /pilosa - -COPY . ./ - -RUN make build FLAGS="-o build/featurebase" ${MAKE_FLAGS} - -##################### -### Pilosa runner ### -##################### - -FROM alpine:3.13.2 as runner - -LABEL maintainer "dev@molecula.com" - -RUN apk add --no-cache curl jq - -COPY --from=pilosa-builder /pilosa/build/featurebase / - -COPY NOTICE /NOTICE - -EXPOSE 10101 -VOLUME /data - -ENV PILOSA_DATA_DIR /data -ENV PILOSA_BIND 0.0.0.0:10101 -ENV PILOSA_BIND_GRPC 0.0.0.0:20101 - -ENTRYPOINT ["/featurebase"] -CMD ["server"] diff --git a/Dockerfile.runner b/Dockerfile.runner deleted file mode 100644 index 12f49e13c..000000000 --- a/Dockerfile.runner +++ /dev/null @@ -1,23 +0,0 @@ -ARG GO_VERSION=latest - -###################### -### Pilosa builder ### -###################### - -FROM golang:${GO_VERSION} as pilosa-builder -ARG MAKE_FLAGS -WORKDIR /pilosa - -COPY . ./ - -RUN make build FLAGS="-o build/featurebase" ${MAKE_FLAGS} - -FROM moleculacorp/idk as idk -LABEL maintainer "dev@molecula.com" -RUN apt-get update -y -RUN apt-get install -y bash curl jq - - -COPY --from=pilosa-builder /pilosa/build/featurebase / -COPY testBackupRestore.sh / -CMD ["bash","/testBackupRestore.sh"] diff --git a/Makefile b/Makefile index f72517684..a24d8c40a 100644 --- a/Makefile +++ b/Makefile @@ -155,15 +155,6 @@ clustertests: vendor clustertests-build: vendor docker-compose -f $(DOCKER_COMPOSE) down -v docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1 --build -# Test Cluster backup and restore -backuptests-build: vendor - docker-compose -f docker-compose-3.yml down - docker-compose -f docker-compose-3.yml build - -backuptests: vendor - docker-compose -f docker-compose-3.yml down -v - docker-compose -f docker-compose-3.yml up --exit-code-from=client1 --abort-on-container-exit - # Install Pilosa install: diff --git a/api.go b/api.go index 07e971599..47558678d 100644 --- a/api.go +++ b/api.go @@ -834,6 +834,15 @@ func (api *API) FragmentData(ctx context.Context, indexName, fieldName, viewName return f, nil } +type RedirectError struct { + HostPort string + error string +} + +func (r RedirectError) Error() string { + return r.error +} + // TranslateData returns all translation data in the specified partition. func (api *API) TranslateData(ctx context.Context, indexName string, partition int) (io.WriterTo, error) { span, _ := tracing.StartSpanFromContext(ctx, "API.TranslateData") @@ -849,6 +858,15 @@ func (api *API) TranslateData(ctx context.Context, indexName string, partition i return nil, newNotFoundError(ErrIndexNotFound, indexName) } + snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) + nodes := snap.PartitionNodes(partition) + if nodes[0].ID != api.server.NodeID() { + return nil, RedirectError{ + HostPort: nodes[0].URI.HostPort(), + error: fmt.Sprintf("can't translate data, this node(%s) does not partition %d", api.server.uri, partition), + } + } + // Retrieve translatestore from holder. store := idx.TranslateStore(partition) if store == nil { @@ -978,7 +996,10 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e return resp, nil } - if api.usageCache.lastCalcDuration < usageCacheMinDuration { + api.usageCache.muAssign.Lock() + lastCalc := api.usageCache.lastCalcDuration + api.usageCache.muAssign.Unlock() + if lastCalc < usageCacheMinDuration { err := api.ResetUsageCache() if err != nil { api.server.logger.Infof("could not reset usageCache: %s", err) @@ -2732,8 +2753,8 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64 for _, flv := range flvs { fld := idx.field(flv.Field) - view, ok := fld.viewMap[flv.View] - if !ok { + view := fld.view(flv.View) + if view == nil { view, err = fld.createViewIfNotExists(flv.View) if err != nil { return err @@ -2747,6 +2768,14 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64 if err != nil { return err } + bd, err := view.bitDepth([]uint64{shard}) + if err != nil { + return err + } + err = fld.cacheBitDepth(bd) + if err != nil { + return err + } } return nil diff --git a/authz/authorization.go b/authz/authorization.go new file mode 100644 index 000000000..77bd67ade --- /dev/null +++ b/authz/authorization.go @@ -0,0 +1,150 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package authz + +import ( + "fmt" + "io" + "io/ioutil" + + "gopkg.in/yaml.v2" +) + +type Auth struct { + // Enable AuthZ/AuthN for featurebase server + Enable bool `toml:"enable"` + + // Application/Client ID + ClientId string `toml:"client-id"` + + // Client Secret + ClientSecret string `toml:"client-secret"` + + // Authorize URL + AuthorizeURL string `toml:"authorize-url"` + + // Token URL + TokenURL string `toml:"token-url"` + + // Group Endpoint URL + GroupEndpointURL string `toml:"group-endpoint-url"` + + // Scope URL + ScopeURL string `toml:"scope-url"` + + // Permissions file for groups + PermissionsFile string `toml:"permissions"` +} + +type GroupPermissions struct { + Permissions map[string]map[string]string `yaml:"user-groups"` + Admin string `yaml:"admin"` +} + +type Group struct { + UserID string + GroupID string `json:"id"` + GroupName string `json:"displayName"` +} + +func (p *GroupPermissions) ReadPermissionsFile(permsFile io.Reader) (err error) { + permsData, err := ioutil.ReadAll(permsFile) + + if err != nil { + return fmt.Errorf("reading permissions failed with error: %s", err) + } + + err = yaml.UnmarshalStrict(permsData, &p) + if err != nil { + return fmt.Errorf("unmarshalling permissions failed with error: %s", err) + } + + return +} + +func (p *GroupPermissions) GetPermissions(groups []Group, index string) (permission string, errors error) { + + if admin := p.IsAdmin(groups); admin { + return "admin", nil + } + + allPermissions := map[string]bool{ + "write": false, + "read": false, + } + + if len(groups) == 0 { + return "", fmt.Errorf("user is not part of any groups in identity provider") + } + + var groupsDenied []string + for _, group := range groups { + if _, ok := p.Permissions[group.GroupID]; ok { + if perm, ok := p.Permissions[group.GroupID][index]; ok { + allPermissions[perm] = true + } else { + return "", fmt.Errorf("user %s does not have permission to index %s", group.UserID, index) + } + } else { + groupsDenied = append(groupsDenied, group.GroupID) + } + } + + if len(groupsDenied) == len(groups) { + return "", fmt.Errorf("group(s) %s does not have permission to FeatureBase", groupsDenied) + } + + if allPermissions["write"] { + return "write", nil + } else if allPermissions["read"] { + return "read", nil + } else { + return "", fmt.Errorf("no permissions found") + } +} + +func (p *GroupPermissions) IsAdmin(groups []Group) bool { + for _, group := range groups { + if p.Admin == group.GroupID { + return true + } + } + return false +} + +func (p *GroupPermissions) GetAuthorizedIndexList(groups []Group, desiredPermission string) (indexList []string) { + // if user is admin, find all indexes in permissions file and return them + if admin := p.IsAdmin(groups); admin { + for groupId := range p.Permissions { + for index := range p.Permissions[groupId] { + indexList = append(indexList, index) + } + } + return indexList + } + + for _, group := range groups { + if _, ok := p.Permissions[group.GroupID]; ok { + for index, permission := range p.Permissions[group.GroupID] { + if permission == desiredPermission { + indexList = append(indexList, index) + } else if permission == "write" && desiredPermission == "read" { + indexList = append(indexList, index) + } + } + } + } + return indexList +} diff --git a/authz/authorization_test.go b/authz/authorization_test.go new file mode 100644 index 000000000..dab33dbe1 --- /dev/null +++ b/authz/authorization_test.go @@ -0,0 +1,314 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +package authz_test + +import ( + "fmt" + "reflect" + "sort" + "strings" + "testing" + + "github.com/molecula/featurebase/v2/authz" +) + +func TestAuth_ReadPermissionsFile(t *testing.T) { + + singleInput := `user-groups: + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + multiInput := `user-groups: + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "read" + "test2": "write" + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "write" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + singlePermission := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": {"test": "read"}, + }, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + multiPermission := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": {"test": "read", "test2": "write"}, + "dca35310-ecda-4f23-86cd-876aee559900": {"test": "write"}}, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + tests := []struct { + input string + output authz.GroupPermissions + }{ + {singleInput, singlePermission}, + {multiInput, multiPermission}, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + permFile := strings.NewReader(test.input) + + var p authz.GroupPermissions + err := p.ReadPermissionsFile(permFile) + if err != nil { + t.Fatalf("readPermissionsFile error: %s", err) + } + + if !reflect.DeepEqual(p, test.output) { + t.Fatalf("expected output %s, but got %s", test.output, p) + } + }, + ) + } +} + +func TestAuth_GetPermissions(t *testing.T) { + + // initializes different example of permissions file in yaml + permissions1 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions2 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "write" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions3 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee55906b": + "test": "write" + "test2": "read" + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions4 := `"user-groups": + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + // initializes groups that are returned from identity provider + groupName := "name" + userId := "user-id" + groupsList1 := []authz.Group{} + groupsList2 := []authz.Group{{userId, "fake-group", groupName}} + groupsList3 := []authz.Group{ + {userId, "dca35310-ecda-4f23-86cd-876aee55906b", groupName}, + {userId, "dca35310-ecda-4f23-86cd-876aee559900", groupName}, + } + groupsList4 := []authz.Group{{userId, "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", groupName}} + + tests := []struct { + yamlData string + groups []authz.Group + index string + userAccess string + err string + }{ + { + permissions1, + groupsList1, + "test", + "", + "user is not part of any groups in identity provider", + }, + { + permissions1, + groupsList3, + "test1", + "", + "does not have permission to index", + }, + { + permissions2, + groupsList2, + "test", + "", + "does not have permission to FeatureBase", + }, + { + permissions1, + groupsList3, + "test", + "read", + "", + }, + { + permissions2, + groupsList3, + "test", + "write", + "", + }, + { + permissions3, + groupsList4, + "test", + "admin", + "", + }, + { + permissions4, + groupsList3, + "test", + "", + "no permissions found", + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + + permFile := strings.NewReader(test.yamlData) + + var p authz.GroupPermissions + if err := p.ReadPermissionsFile(permFile); err != nil { + t.Errorf("Error: %s", err) + } + + p1, err := p.GetPermissions(test.groups, test.index) + + if p1 != test.userAccess { + t.Errorf("expected permission to be %s, but got %s", test.userAccess, p1) + } + + if err != nil { + if !strings.Contains(err.Error(), test.err) { + t.Errorf("expected error to contain %s, but got %s", test.err, err.Error()) + } + } + + }) + } +} + +func TestAuth_IsAdmin(t *testing.T) { + + group1 := []authz.Group{ + {"admin-user-id", "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", "admin-group"}, + } + + group2 := []authz.Group{ + {"user-id", "dca35310-ecda-4f23-86cd-876aee55906b", "group-name"}, + } + + groupPermissions := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": {"test": "write"}, + }, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + tests := []struct { + groups []authz.Group + groupPermissions authz.GroupPermissions + output bool + }{ + { + group1, groupPermissions, true, + }, + { + group2, groupPermissions, false, + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + p := test.groupPermissions + resp := p.IsAdmin(test.groups) + if resp != test.output { + t.Errorf("expected %t, but got %t", test.output, resp) + } + }) + } +} + +func TestAuth_GetAuthorizedIndexList(t *testing.T) { + + group1 := []authz.Group{ + {"user-id", "dca35310-ecda-4f23-86cd-876aee55906b", "group-name"}, + } + + group2 := []authz.Group{ + {"admin-user-id", "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", "admin-group"}, + } + + group3 := []authz.Group{ + {"user-id", "dca35310-ecda-4f23-86cd-876aee559900", "group-name"}, + } + + p := authz.GroupPermissions{ + Permissions: map[string]map[string]string{ + "dca35310-ecda-4f23-86cd-876aee55906b": { + "test1": "read", + "test2": "write", + }, + "dca35310-ecda-4f23-86cd-876aee559900": { + "test3": "read", + }, + }, + Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", + } + + tests := []struct { + groups []authz.Group + permission string + output []string + }{ + { + group1, + "read", + []string{"test1", "test2"}, + }, + { + group1, + "write", + []string{"test2"}, + }, + { + group3, + "write", + nil, + }, + { + group2, + "read", + []string{"test1", "test2", "test3"}, + }, + { + group2, + "write", + []string{"test1", "test2", "test3"}, + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + + indexList := p.GetAuthorizedIndexList(test.groups, test.permission) + sort.Strings(indexList) + + if !reflect.DeepEqual(indexList, test.output) { + t.Errorf("expected %s, but got %s", test.output, indexList) + } + }) + } + +} diff --git a/client/client.go b/client/client.go index 5fd3efa9c..211e7c3f7 100644 --- a/client/client.go +++ b/client/client.go @@ -808,7 +808,6 @@ func (c *Client) shardsMax() (map[string]uint64, error) { } // HTTPRequest sends an HTTP request to the Pilosa server (used by idk) -// nolint: deadcode func (c *Client) HTTPRequest(method string, path string, data []byte, headers map[string]string) (status int, body []byte, err error) { span := c.tracer.StartSpan("Client.HTTPRequest") diff --git a/cmd/roaring-migrate/main.go b/cmd/roaring-migrate/main.go index 3ccc909a6..4ca91bec0 100644 --- a/cmd/roaring-migrate/main.go +++ b/cmd/roaring-migrate/main.go @@ -273,21 +273,19 @@ func Migrate(dataDir, backupPath string) error { }) //raw is now sorted by shard - // need index/field/shard - // make rbf file in backup - rowSize := uint64(0) //? - clear := false - log := false cache := &rbfFile{ temp: filepath.Join(backupPath, "_SCRATCH"), } + bm := roaring.NewSliceBitmap() for _, filename := range raw { index, field, view, shard := Extract(filename) + content, err := ioutil.ReadFile(dataDir + filename) if err != nil { return err } - itr, err := roaring.NewRoaringIterator(content) + err = bm.UnmarshalBinary(content) + if err != nil { return err } @@ -300,10 +298,13 @@ func Migrate(dataDir, backupPath string) error { return err } key := string(txkey.Prefix(index, field, view, shard)) - _, _, err = tx.ImportRoaringBits(key, itr, clear, log, rowSize) - if err != nil { - tx.Rollback() - return err + itr, ok := bm.Containers.Iterator(0) + if ok { + for itr.Next() { + k, v := itr.Value() + tx.PutContainer(key, k, v) + + } } err = tx.Commit() if err != nil { diff --git a/ctl/backup.go b/ctl/backup.go index 6e7e52e83..0e8a257f1 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -18,7 +18,7 @@ import ( "golang.org/x/sync/errgroup" ) -// BackupCommand represents a command for backing up a Pilosa node. +// BackupCommand represents a command for backing up a FeatureBase node. type BackupCommand struct { // nolint: maligned tlsConfig *tls.Config @@ -183,12 +183,17 @@ func (cmd *BackupCommand) backupIDAllocData(ctx context.Context) error { func (cmd *BackupCommand) backupIndexTranslation(ctx context.Context, ii *pilosa.IndexInfo) error { logger := cmd.Logger() logger.Printf("backing up index translation: %q", ii.Name) - if err := cmd.backupIndexTranslateData(ctx, ii.Name); err != nil { - return err + if ii.Options.Keys { + if err := cmd.backupIndexTranslateData(ctx, ii.Name); err != nil { + return err + } } // Back up field translation data. for _, fi := range ii.Fields { + if !fi.Options.Keys { + continue + } if err := cmd.backupFieldTranslateData(ctx, ii.Name, fi.Name); err != nil { return fmt.Errorf("cannot backup field translation data for field %q on index %q: %w", fi.Name, ii.Name, err) } @@ -286,7 +291,6 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, name string) error { partitionN := topology.DefaultPartitionN - // Back up all bitmap data for the index. ch := make(chan int, partitionN) for partitionID := 0; partitionID < partitionN; partitionID++ { ch <- partitionID @@ -318,9 +322,7 @@ func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context, logger.Printf("backing up index translation data: %s/%d", name, partitionID) rc, err := cmd.client.IndexTranslateDataReader(ctx, name, partitionID) - if err == pilosa.ErrTranslateStoreNotFound { - return nil - } else if err != nil { + if err != nil { return fmt.Errorf("fetching translate data reader: %w", err) } defer rc.Close() @@ -349,9 +351,7 @@ func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, indexNam logger.Printf("backing up field translation data: %s/%s", indexName, fieldName) rc, err := cmd.client.FieldTranslateDataReader(ctx, indexName, fieldName) - if err == pilosa.ErrTranslateStoreNotFound { - return nil - } else if err != nil { + if err != nil { return fmt.Errorf("fetching translate data reader: %w", err) } defer rc.Close() diff --git a/ctl/chksum.go b/ctl/chksum.go index 5514fc362..af2430efb 100644 --- a/ctl/chksum.go +++ b/ctl/chksum.go @@ -8,7 +8,7 @@ import ( "io" "github.com/cespare/xxhash" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/server" ) @@ -61,14 +61,19 @@ func (cmd *ChkSumCommand) Run(ctx context.Context) (err error) { h := xxhash.New() for _, ii := range schema.Indexes { - qa := &pilosa.QueryRequest{Index: ii.Name, Query: "Count(All())"} + qa := &pilosa.QueryRequest{Index: ii.Name, Query: "All()"} rs, err := client.Query(ctx, ii.Name, qa) if err != nil { return err } - all := rs.Results[0].(uint64) - as := fmt.Sprintf("all=%v", all) - _, _ = h.Write([]byte(as)) + + all := rs.Results[0].(*pilosa.Row) + if len(all.Keys) > 0 { + allString := fmt.Sprintf("%v", all.Keys) + _, _ = h.Write([]byte(allString)) + } else { + _, _ = h.Write(all.Roaring()) + } for _, field := range ii.Fields { switch field.Options.Type { @@ -92,7 +97,7 @@ func (cmd *ChkSumCommand) Run(ctx context.Context) (err error) { } for _, item := range res.Results { rowids := item.(*pilosa.RowIdentifiers) - //either rowids or keys + // either rowids or keys for _, row := range rowids.Keys { countPql := fmt.Sprintf(`Count(Row(%v="%v"))`, field.Name, row) qr := &pilosa.QueryRequest{Index: ii.Name, Query: countPql} @@ -121,7 +126,7 @@ func (cmd *ChkSumCommand) Run(ctx context.Context) (err error) { } } - fmt.Printf("hash:%x\n", h.Sum(nil)) + fmt.Fprintf(cmd.Stdout, "hash:%x\n", h.Sum(nil)) } return nil diff --git a/ctl/rbf_pages.go b/ctl/rbf_pages.go index 2e831892a..ca3484f31 100644 --- a/ctl/rbf_pages.go +++ b/ctl/rbf_pages.go @@ -69,9 +69,9 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { // Print one line for each page. for pgno, info := range infos { + fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) switch info := info.(type) { case *rbf.MetaPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "meta") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", "") @@ -79,7 +79,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "pageN=%d,walid=%d,rootrec=%d,freelist=%d\n", info.PageN, info.WALID, info.RootRecordPageNo, info.FreelistPageNo) case *rbf.RootRecordPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "rootrec") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", "") @@ -87,7 +86,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "next=%d\n", info.Next) case *rbf.LeafPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "leaf") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree)) @@ -95,7 +93,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "flags=x%x,celln=%d\n", info.Flags, info.CellN) case *rbf.BranchPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "branch") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree)) @@ -103,7 +100,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "flags=x%x,celln=%d\n", info.Flags, info.CellN) case *rbf.BitmapPageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "bitmap") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree)) @@ -111,7 +107,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "-\n") case *rbf.FreePageInfo: - fmt.Fprintf(cmd.Stdout, "%-8d ", pgno) fmt.Fprintf(cmd.Stdout, "%-10s ", "free") if cmd.WithTree { fmt.Fprintf(cmd.Stdout, "%-30q ", "") @@ -119,7 +114,7 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error { fmt.Fprintf(cmd.Stdout, "-\n") default: - panic(fmt.Sprintf("unexpected page info type %T", info)) + fmt.Fprintf(cmd.Stdout, "unknown [%T]\n", info) } } diff --git a/ctl/server.go b/ctl/server.go index 6e65c6ef1..f2118b08c 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -85,7 +85,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk") // RowcacheOn - flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "turn on the rowcache for all backends (may speed some queries)") + flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "Do not use, permanently disabled. Flag exists for backwards compatibility and will be removed.") // RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions. srv.Config.RBFConfig.DefineFlags(flags) @@ -120,5 +120,6 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringSliceVar(&srv.Config.Auth.Scopes, "auth.scopes", srv.Config.Auth.Scopes, "Comma separated list of scopes obtained from IdP") flags.StringVar(&srv.Config.Auth.HashKey, "auth.hash-key", srv.Config.Auth.HashKey, "First Secret for Auth.") flags.StringVar(&srv.Config.Auth.BlockKey, "auth.block-key", srv.Config.Auth.BlockKey, "Second Secret for Auth.") + flags.StringVar(&srv.Config.Auth.PermissionsFile, "auth.permissions", srv.Config.Auth.PermissionsFile, "Permissions' file with group authorization.") } diff --git a/docker-compose-3.yml b/docker-compose-3.yml deleted file mode 100644 index ef31424d0..000000000 --- a/docker-compose-3.yml +++ /dev/null @@ -1,97 +0,0 @@ -version: "3" -services: - pilosa0: - image: build/pilosa - build: - context: . - dockerfile: Dockerfile.pilosa - environment: - PILOSA_ADVERTISE: pilosa0:10101 - PILOSA_ADVERTISE_GRPC: pilosa0:20101 - PILOSA_CLUSTER_REPLICAS: 1 - PILOSA_DATA_DIR: /data/pilosa0 - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosa0:10201 - PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosa0:10301 - PILOSA_ETCD_INITIAL_CLUSTER: pilosa0=http://pilosa0:10301,pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301 - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201 - PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301 - PILOSA_NAME: pilosa0 - PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf} - volumes: - - data:/data - healthcheck: - test: x=$$(curl -s localhost:10101/status | jq -r ".state") && [[ "$$x" == "NORMAL" ]] || $$(exit 1) - interval: 10s - timeout: 5s - retries: 5 - pilosa1: - image: build/pilosa - build: - context: . - dockerfile: Dockerfile.pilosa - environment: - PILOSA_ADVERTISE: pilosa1:10101 - PILOSA_ADVERTISE_GRPC: pilosa1:20101 - PILOSA_CLUSTER_REPLICAS: 1 - PILOSA_DATA_DIR: /data/pilosa1 - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosa1:10201 - PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosa1:10301 - PILOSA_ETCD_INITIAL_CLUSTER: pilosa0=http://pilosa0:10301,pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301 - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201 - PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301 - PILOSA_NAME: pilosa1 - PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf} - volumes: - - data:/data - pilosa2: - image: build/pilosa - build: - context: . - dockerfile: Dockerfile.pilosa - environment: - PILOSA_ADVERTISE: pilosa2:10101 - PILOSA_ADVERTISE_GRPC: pilosa2:20101 - PILOSA_CLUSTER_REPLICAS: 1 - PILOSA_DATA_DIR: /data/pilosa2 - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosa2:10201 - PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosa2:10301 - PILOSA_ETCD_INITIAL_CLUSTER: pilosa0=http://pilosa0:10301,pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301 - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201 - PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301 - PILOSA_NAME: pilosa2 - PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf} - volumes: - - data:/data - pilosax: - image: build/pilosa - build: - context: . - dockerfile: Dockerfile.pilosa - environment: - PILOSA_ADVERTISE: pilosax:10101 - PILOSA_ADVERTISE_GRPC: pilosax:20101 - PILOSA_CLUSTER_REPLICAS: 1 - PILOSA_DATA_DIR: /data/pilosax - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosax:10201 - PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosax:10301 - PILOSA_ETCD_INITIAL_CLUSTER: pilosax=http://pilosax:10301 - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201 - PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301 - PILOSA_NAME: pilosax - PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf} - volumes: - - data:/data - client1: - image: tgruben/bash - build: - context: . - dockerfile: Dockerfile.runner - environment: - - GO111MODULE=on - volumes: - - /var/run/docker.sock:/var/run/docker.sock - depends_on: - - pilosa0 - -volumes: - data: diff --git a/executor.go b/executor.go index 226759b55..5e03e1a72 100644 --- a/executor.go +++ b/executor.go @@ -51,6 +51,9 @@ type executor struct { Node *topology.Node Cluster *cluster + // how many jobs the work queue has seen + workCounter uint64 + // Client used for remote requests. client InternalQueryClient @@ -61,6 +64,7 @@ type executor struct { workMu sync.RWMutex workersWG sync.WaitGroup workerPoolSize int + currentWorkers int64 work chan job // Maximum per-request memory usage (Extract() only) @@ -128,15 +132,61 @@ func newExecutor(opts ...executorOption) *executor { e.work = make(chan job, e.workerPoolSize) _ = testhook.Opened(NewAuditor(), e, nil) for i := 0; i < e.workerPoolSize; i++ { - e.workersWG.Add(1) - go func() { - defer e.workersWG.Done() - worker(e.work) - }() + e.addWorker() } + go func() { + // background task: every so often, check to see whether we have + // work in the queue but none has been taken for a while. if so, we + // need more workers. + prev := atomic.LoadUint64(&e.workCounter) + periodic := time.NewTicker(50 * time.Millisecond) + defer periodic.Stop() + running := true + idle := 0 + for running { + <-periodic.C + func() { + e.workMu.RLock() + defer e.workMu.RUnlock() + if e.shutdown { + running = false + return + } + if len(e.work) == 0 { + idle++ + if idle > 10 && atomic.LoadInt64(&e.currentWorkers) > int64(e.workerPoolSize*2) { + select { + case e.work <- job{idleHands: true}: + // we closed an excess worker + default: + // somehow between our test above and now the work + // queue FILLED UP and we stoically accept this + } + idle = 0 + } + return + } + next := atomic.LoadUint64(&e.workCounter) + if next == prev { + e.addWorker() + } + prev = next + }() + } + }() return e } +func (e *executor) addWorker() { + e.workersWG.Add(1) + atomic.AddInt64(&e.currentWorkers, 1) + go func() { + defer e.workersWG.Done() + e.worker(e.work) + atomic.AddInt64(&e.currentWorkers, -1) + }() +} + func (e *executor) Close() error { e.workMu.Lock() defer e.workMu.Unlock() @@ -4493,7 +4543,11 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, return nil, err } defer finisher(&err0) - return frag.row(tx, rowID) + row, err := frag.row(tx, rowID) + if qcx.write && err == nil { + row = row.Clone() + } + return row, err } // If no quantum exists then return an empty bitmap. @@ -4532,15 +4586,21 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, if len(rows) == 0 { return &Row{}, nil } else if len(rows) == 1 { + if qcx.write { + return rows[0].Clone(), nil + } return rows[0], nil } row := rows[0].Union(rows[1:]...) + if qcx.write { + row = row.Clone() + } return row, nil } // executeRowBSIGroupShard executes a range(bsiGroup) call for a local shard. -func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err0 error) { +func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (cloneable *Row, err0 error) { span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeRowBSIGroupShard") defer span.Finish() @@ -4572,6 +4632,11 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index return nil, err } defer finisher(&err0) + defer func() { + if qcx.write && cloneable != nil { + cloneable = cloneable.Clone() + } + }() // EQ null _exists - frag.NotNull() // NEQ null frag.NotNull() @@ -4822,6 +4887,9 @@ func (e *executor) executeNotShard(ctx context.Context, qcx *Qcx, index string, if existenceRow, err = existenceFrag.row(tx, 0); err != nil { return nil, err } + if qcx.write { + existenceRow = existenceRow.Clone() + } } // the finishers returned by a write tx, which we might be in if there's // a higher-level write in this call OR ANY OTHER CALL, are safe to @@ -5915,10 +5983,15 @@ type job struct { ctx context.Context memoryAvailable *int64 // shared, atomic value resultChan chan mapResponse + idleHands bool } -func worker(work chan job) { +func (e *executor) worker(work chan job) { for j := range work { + atomic.AddUint64(&e.workCounter, 1) + if j.idleHands { + return + } // Skip out early if the context is done, but still send // an ack so mapperLocal can be sure we aren't about to // work on something it sent us. diff --git a/executor_test.go b/executor_test.go index 56023107e..432499511 100644 --- a/executor_test.go +++ b/executor_test.go @@ -27,6 +27,7 @@ import ( "github.com/google/go-cmp/cmp/cmpopts" pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/boltdb" + "github.com/molecula/featurebase/v2/ctl" "github.com/molecula/featurebase/v2/disco" "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/pql" @@ -6748,7 +6749,7 @@ func variousQueriesCountDistinctTimestamp(t *testing.T, c *test.Cluster) { field := "ts" // create an index and timestamp field - c.CreateField(t, index, pilosa.IndexOptions{}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s")) + c.CreateField(t, index, pilosa.IndexOptions{TrackExistence: true}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s")) // add some data data := []string{"2010-01-02T12:32:00Z", "2010-04-20T12:32:00Z", "2011-04-20T12:32:00Z"} @@ -7015,19 +7016,90 @@ func TestMissingKeyRegression(t *testing.T) { // (single and multi-node clusters, different endpoints for the // queries (HTTP, GRPC, Postgres), etc.). func TestVariousQueries(t *testing.T) { - for _, clusterSize := range []int{1, 3, 4, 7} { + for _, clusterSize := range []int{1, 3, 5} { clusterSize := clusterSize t.Run(fmt.Sprintf("%d-node", clusterSize), func(t *testing.T) { c := test.MustRunCluster(t, clusterSize) defer c.Close() + + // put a variety of data into the cluster + populateTestData(t, c) + backupTest(t, c, usersIndex) + variousQueries(t, c) variousQueriesOnTimeFields(t, c) variousQueriesOnPercentiles(t, c) variousQueriesCountDistinctTimestamp(t, c) + backupTest(t, c, "") // test backup/restore of all indexes }) } } +func backupTest(t *testing.T, c *test.Cluster, index string) { + // should this really be in executor? No. But all these + // integration-y query tests probably shouldn't be either. My goal + // putting this here is to take advantage of already-existing + // clusters and data. + + sum := chkSumCluster(t, c) + + backupDir := backupCluster(t, c, index) + + cnew := test.MustRunCluster(t, 3) // this way we test 1->3 3->3 5->3 + defer cnew.Close() + + restoreCluster(t, backupDir, cnew) + + sumNew := chkSumCluster(t, cnew) + + if sum != sumNew { + t.Fatalf("old/new checksum mismatch, old:\n%s\nnew:\n%s", sum, sumNew) + } +} + +func chkSumCluster(t *testing.T, c *test.Cluster) string { + buf := &bytes.Buffer{} + + chkSum := ctl.NewChkSumCommand(nil, buf, buf) + chkSum.Host = c.Nodes[len(c.Nodes)-1].URL() + if err := chkSum.Run(context.Background()); err != nil { + t.Fatalf("running checksum: %v", err) + } + + return buf.String() +} + +func backupCluster(t *testing.T, c *test.Cluster, index string) (backupDir string) { + td, err := testhook.TempDir(t, "backupTest") + if err != nil { + t.Fatalf("can't even get a temp dir, what a ripoff: %v", err) + } + td = td + "/backupTest" + + buf := &bytes.Buffer{} + backupCommand := ctl.NewBackupCommand(nil, buf, buf) + backupCommand.Host = c.Nodes[len(c.Nodes)-1].URL() // don't pick node 0 so we don't always get primary (better code coverage) + backupCommand.Index = index + backupCommand.OutputDir = td + + if err := backupCommand.Run(context.Background()); err != nil { + t.Log(buf.String()) + t.Fatalf("running backup: %v", err) + } + return td +} + +func restoreCluster(t *testing.T, backupDir string, c *test.Cluster) { + buf := &bytes.Buffer{} + + restore := ctl.NewRestoreCommand(nil, buf, buf) + restore.Host = c.Nodes[len(c.Nodes)-1].URL() + restore.Path = backupDir + if err := restore.Run(context.Background()); err != nil { + t.Fatalf("restoring: %v", err) + } +} + // tests for abbreviating time values in queries func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) { // todo, make rand more random, 42 isnt the answer to everything @@ -7332,10 +7404,12 @@ func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) { } } -func variousQueries(t *testing.T, c *test.Cluster) { +var usersIndex = "users" + +func populateTestData(t *testing.T, c *test.Cluster) { // Create and populate "likenums" similar to "likes", but without keys on the field. - c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likenums") - c.ImportIDKey(t, "users", "likenums", []test.KeyID{ + c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likenums") + c.ImportIDKey(t, usersIndex, "likenums", []test.KeyID{ {ID: 1, Key: "userA"}, {ID: 2, Key: "userB"}, {ID: 3, Key: "userC"}, @@ -7353,8 +7427,8 @@ func variousQueries(t *testing.T, c *test.Cluster) { }) // Create and populate "likes" field. - c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likes", pilosa.OptFieldKeys()) - c.ImportKeyKey(t, "users", "likes", [][2]string{ + c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likes", pilosa.OptFieldKeys()) + c.ImportKeyKey(t, usersIndex, "likes", [][2]string{ {"molecula", "userA"}, {"pilosa", "userB"}, {"pangolin", "userC"}, @@ -7370,8 +7444,8 @@ func variousQueries(t *testing.T, c *test.Cluster) { }) // Create and populate "dinner" field. - c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "dinner", pilosa.OptFieldKeys()) - c.ImportKeyKey(t, "users", "dinner", [][2]string{ + c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "dinner", pilosa.OptFieldKeys()) + c.ImportKeyKey(t, usersIndex, "dinner", [][2]string{ {"leftovers", "userB"}, {"pizza", "userA"}, {"pizza", "userB"}, @@ -7381,11 +7455,11 @@ func variousQueries(t *testing.T, c *test.Cluster) { }) // Create and populate "places_visited" time field. - c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "places_visited", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YM"))) + c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "places_visited", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YM"))) ts2019Jan01 := int64(1546300800) * 1e+9 // 2019 January 1st 0:00:00 ts2019Aug01 := int64(1564617600) * 1e+9 // 2019 August 1st 0:00:00 ts2020Jan01 := int64(1577836800) * 1e+9 // 2020 January 1st 0:00:00 - c.ImportTimeQuantumKey(t, "users", "places_visited", []test.TimeQuantumKey{ + c.ImportTimeQuantumKey(t, usersIndex, "places_visited", []test.TimeQuantumKey{ // 2019 January: nairobi, paris, austin, toronto {RowKey: "nairobi", ColKey: "userB", Ts: ts2019Jan01}, {RowKey: "paris", ColKey: "userC", Ts: ts2019Jan01}, @@ -7405,8 +7479,8 @@ func variousQueries(t *testing.T, c *test.Cluster) { }) // Create and populate "affinity" int field with negative, positive, zero and null values. - c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "affinity", pilosa.OptFieldTypeInt(-1000, 1000)) - c.ImportIntKey(t, "users", "affinity", []test.IntKey{ + c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "affinity", pilosa.OptFieldTypeInt(-1000, 1000)) + c.ImportIntKey(t, usersIndex, "affinity", []test.IntKey{ {Val: 10, Key: "userA"}, {Val: -10, Key: "userB"}, {Val: 5, Key: "userC"}, @@ -7415,8 +7489,8 @@ func variousQueries(t *testing.T, c *test.Cluster) { }) // Create and populate "net_worth" int field with positive values. - c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "net_worth", pilosa.OptFieldTypeInt(-100000000, 100000000)) - c.ImportIntKey(t, "users", "net_worth", []test.IntKey{ + c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "net_worth", pilosa.OptFieldTypeInt(-100000000, 100000000)) + c.ImportIntKey(t, usersIndex, "net_worth", []test.IntKey{ {Val: 1, Key: "userA"}, {Val: 10, Key: "userB"}, {Val: 100, Key: "userC"}, @@ -7425,8 +7499,8 @@ func variousQueries(t *testing.T, c *test.Cluster) { {Val: 100000, Key: "userF"}, }) - c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "zip_code", pilosa.OptFieldTypeInt(0, 100000)) - c.ImportIntKey(t, "users", "zip_code", []test.IntKey{ + c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "zip_code", pilosa.OptFieldTypeInt(0, 100000)) + c.ImportIntKey(t, usersIndex, "zip_code", []test.IntKey{ {Val: 78739, Key: "userA"}, {Val: 78739, Key: "userB"}, {Val: 19707, Key: "userC"}, @@ -7434,7 +7508,12 @@ func variousQueries(t *testing.T, c *test.Cluster) { {Val: 86753, Key: "userE"}, {Val: 78739, Key: "userG"}, }) +} +func variousQueries(t *testing.T, c *test.Cluster) { + // NOTE: this relies on populateTestData being called first + + // define and run a bunch of tests tests := []struct { query string qrVerifier func(t *testing.T, resp pilosa.QueryResponse) @@ -7781,8 +7860,8 @@ leftovers,1 for i, tst := range tests { t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) { - resp := c.Query(t, "users", tst.query) - tr := c.QueryGRPC(t, "users", tst.query) + resp := c.Query(t, usersIndex, tst.query) + tr := c.QueryGRPC(t, usersIndex, tst.query) if tst.qrVerifier != nil { tst.qrVerifier(t, resp) } @@ -8352,7 +8431,7 @@ func MinMaxTimestampNodeTester(t *testing.T, numNodes int) { defer c.Close() // create an index and timestamp field - c.CreateField(t, index, pilosa.IndexOptions{}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s")) + c.CreateField(t, index, pilosa.IndexOptions{TrackExistence: true}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s")) // add some data expected := "2010-01-02T12:32:00Z" diff --git a/field.go b/field.go index 7076c6e98..d325c477e 100644 --- a/field.go +++ b/field.go @@ -705,8 +705,11 @@ func (f *Field) cacheBitDepth(bd uint64) error { f.mu.Lock() defer f.mu.Unlock() - f.options.BitDepth = bd - if bsig != nil { + if f.options.BitDepth < bd { + f.options.BitDepth = bd + } + + if bsig != nil && bsig.BitDepth < bd { bsig.BitDepth = bd } @@ -1884,7 +1887,7 @@ func applyDefaultOptions(o *FieldOptions) FieldOptions { // are included. func (o *FieldOptions) MarshalJSON() ([]byte, error) { switch o.Type { - case FieldTypeSet: + case FieldTypeSet, "": return json.Marshal(struct { Type string `json:"type"` CacheType string `json:"cacheType"` @@ -1975,7 +1978,7 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { o.Type, }) } - return nil, errors.New("invalid field type") + return nil, errors.Errorf("invalid field type: '%s'", o.Type) } // MinTimestamp returns the minimum value for a timestamp field. diff --git a/fragment.go b/fragment.go index 83514edad..8dfa6749c 100644 --- a/fragment.go +++ b/fragment.go @@ -218,6 +218,8 @@ func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragm func (f *fragment) cachePath() string { return f.path() + cacheExt } func (f *fragment) bitDepth() (uint64, error) { + f.mu.RLock() + defer f.mu.RUnlock() tx, err := f.holder.BeginTx(false, f.idx, f.shard) if err != nil { return 0, errors.Wrapf(err, "beginning new tx(false, %s, %d)", f.index(), f.shard) @@ -593,8 +595,8 @@ func (f *fragment) mutexCheck(tx Tx, details bool, limit int) (map[uint64][]uint // row returns a row by ID. func (f *fragment) row(tx Tx, rowID uint64) (*Row, error) { - f.mu.Lock() - defer f.mu.Unlock() + f.mu.RLock() + defer f.mu.RUnlock() return f.unprotectedRow(tx, rowID) } @@ -937,9 +939,12 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e return changed, nil } -// unprotectedClearBlock clears all rows for a given block. +// clearBlock clears all rows for a given block. // This updates both the on-disk storage and the in-cache bitmap. -func (f *fragment) unprotectedClearBlock(tx Tx, block int) (changed bool, err error) { +func (f *fragment) clearBlock(tx Tx, block int) (changed bool, err error) { + f.mu.Lock() + defer f.mu.Unlock() + firstRow := uint64(block * HashBlockSize) var wp *io.Writer if f.storage != nil { @@ -2708,20 +2713,24 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep func (f *fragment) importRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error { span, ctx := tracing.StartSpanFromContext(ctx, "fragment.importRoaring") defer span.Finish() - span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.AcquireFragmentLock") - f.mu.Lock() - defer f.mu.Unlock() - span.Finish() - return f.unprotectedImportRoaring(ctx, tx, data, clear) + rowSet, updateCache, err := f.doImportRoaring(ctx, tx, data, clear) + if err != nil { + return errors.Wrap(err, "doImportRoaring") + } + if updateCache { + return f.updateCachePostImport(ctx, rowSet) + } + return nil } -func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error { +func (f *fragment) doImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) (map[uint64]int, bool, error) { + f.mu.RLock() + defer f.mu.RUnlock() rowSize := uint64(1 << shardVsContainerExponent) span, ctx := tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits") + defer span.Finish() - useRowCache := storage.RowCacheEnabled() - var changed int var rowSet map[uint64]int var wp *io.Writer if f.storage != nil { @@ -2734,37 +2743,37 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b return err } - changed, rowSet, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, rit, clear, true, rowSize) + _, rowSet, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, rit, clear, true, rowSize) return err }) - span.Finish() if err != nil { - return err + return nil, false, err } updateCache := f.CacheType != CacheTypeNone + return rowSet, updateCache, err +} + +func (f *fragment) updateCachePostImport(ctx context.Context, rowSet map[uint64]int) error { + f.mu.Lock() + defer f.mu.Unlock() anyChanged := false for rowID, changes := range rowSet { if changes == 0 { continue } - if useRowCache && f.rowCache != nil { - f.rowCache.Add(rowID, nil) - } - if updateCache { - anyChanged = true - if changes < 0 { - absChanges := uint64(-1 * changes) - if absChanges <= f.cache.Get(rowID) { - f.cache.BulkAdd(rowID, f.cache.Get(rowID)-absChanges) - } else { - f.cache.BulkAdd(rowID, 0) - } + anyChanged = true + if changes < 0 { + absChanges := uint64(-1 * changes) + if absChanges <= f.cache.Get(rowID) { + f.cache.BulkAdd(rowID, f.cache.Get(rowID)-absChanges) } else { - f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes)) + f.cache.BulkAdd(rowID, 0) } + } else { + f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes)) } } // we only set this if we need to update the cache @@ -2772,26 +2781,18 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b f.cache.Invalidate() } - span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.incrementOpN") - - f.incrementOpN(changed) - - span.Finish() return nil } // importRoaringOverwrite overwrites the specified block with the provided data. func (f *fragment) importRoaringOverwrite(ctx context.Context, tx Tx, data []byte, block int) error { - f.mu.Lock() - defer f.mu.Unlock() - // Clear the existing data from fragment block. - if _, err := f.unprotectedClearBlock(tx, block); err != nil { + if _, err := f.clearBlock(tx, block); err != nil { return errors.Wrapf(err, "clearing block: %d", block) } // Union the new block data with the fragment data. - return f.unprotectedImportRoaring(ctx, tx, data, false) + return f.importRoaring(ctx, tx, data, false) } // incrementOpN increase the operation count by one. diff --git a/go.mod b/go.mod index 82ac7a3d6..24e125060 100644 --- a/go.mod +++ b/go.mod @@ -57,7 +57,7 @@ require ( golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c google.golang.org/grpc v1.28.0 - gopkg.in/yaml.v2 v2.3.0 // indirect + gopkg.in/yaml.v2 v2.3.0 modernc.org/mathutil v1.0.0 modernc.org/strutil v1.0.0 sigs.k8s.io/yaml v1.2.0 // indirect diff --git a/http/handler.go b/http/handler.go index 79c45b734..895b9c368 100644 --- a/http/handler.go +++ b/http/handler.go @@ -2342,6 +2342,12 @@ func (h *Handler) handleGetTranslateData(w http.ResponseWriter, r *http.Request) // Retrieve partition data from holder. p, err := h.api.TranslateData(r.Context(), q.Get("index"), int(partition)) + if redir, ok := err.(pilosa.RedirectError); ok { + newURL := *r.URL + newURL.Host = redir.HostPort + http.Redirect(w, r, newURL.String(), http.StatusSeeOther) + return + } if err != nil { http.Error(w, err.Error(), http.StatusNotFound) return diff --git a/install/featurebase.conf b/install/featurebase.conf index c824098df..033db191d 100644 --- a/install/featurebase.conf +++ b/install/featurebase.conf @@ -384,4 +384,5 @@ log-path = "/var/log/molecula/featurebase.log" # logout-url = "" # scopes = ["", ""] # hash-key = "" -# block-key = "" \ No newline at end of file +# block-key = "" +# permissions = "" diff --git a/rbf/cfg/cfg.go b/rbf/cfg/cfg.go index cc2cf7a8a..671c6fe43 100644 --- a/rbf/cfg/cfg.go +++ b/rbf/cfg/cfg.go @@ -2,6 +2,7 @@ package cfg import ( + "github.com/molecula/featurebase/v2/logger" "github.com/spf13/pflag" ) @@ -35,6 +36,11 @@ type Config struct { // CursorCacheSize is the number of copies of Cursor{} to keep in our // readyCursorCh arena to avoid GC pressure. CursorCacheSize int64 `toml:"cursor-cache-size"` + + // Logger specifies a logger for asynchronous errors, such as + // background checkpoints. It cannot be set from toml. The default is + // to use stderr. + Logger logger.Logger `toml:"-"` } func NewDefaultConfig() *Config { diff --git a/rbf/cursor.go b/rbf/cursor.go index 68c165d94..eda6bec1d 100644 --- a/rbf/cursor.go +++ b/rbf/cursor.go @@ -774,6 +774,25 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) { cells[len(cells)-1] = branchCell{} cells = cells[:len(cells)-1] + // Branches are not allowed to have zero element so we must remove the page + // or, in the case of the root page, convert to a leaf page. + if len(cells) == 0 { + // If this is the root page, convert to leaf page. + if stackIndex == 0 { + var buf [PageSize]byte + writePageNo(buf[:], elem.pgno) + writeFlags(buf[:], PageTypeLeaf) + writeCellN(buf[:], len(cells)) + return c.tx.writePage(buf[:]) + } + + // If this is a non-root page, free and remove from parent. + if err := c.tx.freePgno(elem.pgno); err != nil { + return err + } + return c.deleteBranchCell(stackIndex-1, oldPageKey) + } + // If the root only has one node, replace it with its child. if stackIndex == 0 && len(cells) == 1 { target, _, err := c.tx.readPage(cells[0].ChildPgno) @@ -802,6 +821,9 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) { writeBranchCell(buf[:], j, offset, cell) offset += align8(branchCellSize) } + + assert(readCellN(buf[:]) > 0) // must have at least one cell + if err := c.tx.writePage(buf[:]); err != nil { return err } diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go index 4f7464bd1..940798c04 100644 --- a/rbf/cursor_test.go +++ b/rbf/cursor_test.go @@ -973,8 +973,8 @@ func TestCursor_SplitBranchCells(t *testing.T) { } // c, _ := tx.Cursor("x") //added just for dot code coverage - c.Dump("ignore for coverage") - + c.Dump("test.dump") + os.Remove("test.dump") } func TestCursor_RemoveCells(t *testing.T) { diff --git a/rbf/db.go b/rbf/db.go index cc4b65700..65dd4fbea 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -11,6 +11,7 @@ import ( "syscall" "github.com/benbjohnson/immutable" + "github.com/molecula/featurebase/v2/logger" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/syswrap" ) @@ -27,6 +28,16 @@ var cursorSyncPool = &sync.Pool{ }, } +// txWaiter is a representation of "i need to wait for txs to complete". +// it is created with a function, and will run that function, with the db +// lock held, at some point after every Tx that was open when it was created +// has closed. WARNING: A txWaiter may hold db.rwmu. +type txWaiter struct { + ready chan struct{} + waitingOn map[*Tx]struct{} + callback func() +} + // DB options like MaxSize, FsyncEnabled, DoAllocZero // can be set before calling DB.Open(). type DB struct { @@ -38,15 +49,21 @@ type DB struct { pageMap *PageMap // pgno-to-WALID mapping txs map[*Tx]struct{} // active transactions opened bool // true if open + logger logger.Logger // for diagnostics from async things - wal []byte // wal mmap - walFile *os.File // wal file descriptor - walPageN int // wal page count + wal []byte // wal mmap + walFile *os.File // wal file descriptor + walPageN int // wal page count + baseWALID int64 // WAL ID of first page mu sync.RWMutex // general mutex rwmu sync.Mutex // mutex for restricting single writer haltCond *sync.Cond // condition for resuming txs after checkpoint + txWaiters []*txWaiter // things waiting for Txs to close + + isDead error // this database died in an unrecoverable way, error out opens + // Path represents the path to the database file. Path string } @@ -62,6 +79,11 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB { txs: make(map[*Tx]struct{}), pageMap: NewPageMap(), Path: path, + logger: cfg.Logger, + } + if db.logger == nil { + // default to writing to stdout if not told otherwise + db.logger = logger.NewStandardLogger(os.Stderr) } db.haltCond = sync.NewCond(&db.mu) @@ -133,8 +155,12 @@ func (db *DB) Open() (err error) { // Open write-ahead log & checkpoint to the end since no transactions are open. if err := db.openWAL(); err != nil { return fmt.Errorf("wal open: %w", err) - } else if err := db.checkpoint(); err != nil { - return fmt.Errorf("checkpoint: %w", err) + } else { + // checkpoint wants to hold the rwmu lock. + db.rwmu.Lock() + if err := db.checkpoint(); err != nil { + return fmt.Errorf("startup checkpoint: %w", err) + } } return nil @@ -158,10 +184,12 @@ func (db *DB) openWAL() (err error) { // Determine the number of whole pages in the WAL. var pageN int + var fileSize int64 if fi, err := db.walFile.Stat(); err != nil { return fmt.Errorf("wal stat: %w", err) } else { - pageN = int(fi.Size() / PageSize) + fileSize = fi.Size() + pageN = int(fileSize / PageSize) } // Read backwards through the WAL to find the last valid meta page. @@ -169,28 +197,96 @@ func (db *DB) openWAL() (err error) { if page, err := db.readWALPageAt(pageN - 1); err != nil { return err } else if IsMetaPage(page) { + // We now face a challenge. Probably this is a meta page. + // But consider a sequence of pages written which gets + // interrupted right before the meta page is written. + // If the last page is a bitmap page, it could LOOK LIKE a meta + // page. So we have to check the page before it. If that page + // is a bitmap header, then actually this is a bitmap page, right? + // If that page doesn't exist, of course, we're fine, except + // for the philosophical question of why we wrote a meta page + // when no pages had changed. + if pageN > 1 { + if page, err = db.readWALPageAt(pageN - 2); err != nil { + return err + } + if IsBitmapHeader(page) { + // But wait! + // What if this *is* a meta page, and the page before it is + // actually a *bitmap page* that looks like a bitmap header? And + // so on. + // + // Rather than try to resolve this, in this insanely unlikely + // situation, we read from the beginning which allows us to + // always know what we're seeing, because every bitmap page + // comes *after* a bitmap header page, and thus, we know when + // we might be seeing one. + pageN, err = db.methodicalWALPageN(pageN) + if err != nil { + return err + } + } + } break } } - - // Truncate WAL to the last valid meta page. - if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil { - return fmt.Errorf("wal truncate: %w", err) - } else if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil { + if fileSize != int64(pageN*PageSize) { + if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil { + return fmt.Errorf("wal truncate: %w", err) + } + } + if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil { return fmt.Errorf("wal seek: %w", err) } db.walPageN = pageN + db.baseWALID = readMetaWALID(db.data) return nil } -// checkpoint moves all WAL pages to the main DB file. -// Must be called by a write transaction while under db.mu lock. -func (db *DB) checkpoint() error { +// methodicalWALPageN tries to determine the last meta page in a very reliable +// but slow way. This handles the theoretical but hard to imagine creating +// edge case where we have a bitmap page which happens to look like a meta +// page, and the write got interrupted before the meta page got written. +func (db *DB) methodicalWALPageN(pageN int) (lastMeta int, err error) { + for i := 0; i < pageN; i++ { + var page []byte + if page, err = db.readWALPageAt(i); err != nil { + return -1, err + } + switch { + case IsMetaPage(page): + lastMeta = i + case IsBitmapHeader(page): + // skip the bitmap page, which we can't usefully evaluate + i++ + } + } + return lastMeta, nil +} + +// Checkpoint performs a manual checkpoint. This is not necessary except for tests. +func (db *DB) Checkpoint() error { + db.mu.Lock() + defer db.mu.Unlock() + db.rwmu.Lock() + return db.checkpoint() +} + +// checkpoint moves all WAL pages to the main DB file. Must be called +// while holding both db.mu and db.rwmu. Should release db.rwmu, but not +// db.mu. +func (db *DB) checkpoint() (err error) { + // if we don't spin off a possible async waiter, we should release the + // write lock, if we do, that will release it. + releaseLock := true + defer func() { + if releaseLock { + db.rwmu.Unlock() + } + }() if !db.opened { return nil - } else if len(db.txs) > 0 { - return nil // skip if transactions open } // Check if there are any WAL pages, if not do nothing as @@ -199,48 +295,112 @@ func (db *DB) checkpoint() error { if db.walPageN == 0 { return nil } - - for i := 0; i < db.walPageN; i++ { - page, err := db.readWALPageAt(i) - if err != nil { - return err + // wake up things waiting on haltCond when we're done, even if we fail. + // Otherwise, we deadlock with them all stuck waiting on that forever. + defer func() { + if err != nil && db.isDead == nil { + db.isDead = err } + db.haltCond.Broadcast() + }() - // Determine page number. Meta pages are always on zero & bitmap - // headers specify the page number of the next page in the WAL. - // All other pages have their page number in the page data. - var pgno uint32 - if IsBitmapHeader(page) { - pgno = readPageNo(page) - if page, err = db.readWALPageAt(i + 1); err != nil { - return err + // Copy the pages from the WAL back to the database outside of the lock. + if err := func() error { + db.mu.Unlock() // This is intentionally reversed so run w/o lock + defer db.mu.Lock() + + var page []byte + // We might have either a *PageMap or just the file. If we have the file, + // building the PageMap is fairly expensive because it's fancy and immutable. + // If we have the PageMap *or* some other map, that's two different things + // to iterate. If we have the PageMap, building a map from it is relatively + // cheap, so we'll do it that way. + pages := make(map[uint32]int) + + if db.pageMap.size == 0 { + // you'd think we're done, but actually this PROBABLY means that + // this is initial startup, and we haven't read the file yet. We scan + // the file for pages, because it turns out most of them probably + // got overwritten. + for i := 0; i < db.walPageN; i++ { + page, err = db.readWALPageAt(i) + if err != nil { + return fmt.Errorf("reading WAL page %d: %w", i, err) + } + + // Determine page number. Meta pages are always on zero & bitmap + // headers specify the page number of the next page in the WAL. + // All other pages have their page number in the page data. + var pgno uint32 + if IsBitmapHeader(page) { + pgno = readPageNo(page) + if i+1 < db.walPageN { + if page, err = db.readWALPageAt(i + 1); err != nil { + return err + } + } else { + return fmt.Errorf("last page of WAL file (%d) is bitmap header", i) + } + i++ // bitmaps in WAL are two pages + } else if !IsMetaPage(page) { + pgno = readPageNo(page) + } + // record where in the file we have this page + pages[pgno] = i + } + } else { + itr := db.pageMap.Iterator() + itr.First() + for k, v, ok := itr.Next(); ok; k, v, ok = itr.Next() { + pages[k] = int(v - db.baseWALID - 1) } - i++ // bitmaps in WAL are two pages - } else if !IsMetaPage(page) { - pgno = readPageNo(page) } - // Write data to the data file. - if err := db.writeDBPage(pgno, page); err != nil { - return err + // fmt.Printf("checkpoint: walPageN %d, PageMap size %d\n", db.walPageN, db.pageMap.size) + for pgno, walID := range pages { + page, err = db.readWALPageAt(walID) + if err != nil { + return fmt.Errorf("reading page %d [page number %d]: %v", walID, pgno, err) + } + + // Write data to the data file. + if err = db.writeDBPage(pgno, page); err != nil { + return fmt.Errorf("writing page %d: %v", pgno, err) + } } + + // Ensure database file is synced and then truncate the WAL file. + if err = db.fsync(db.file); err != nil { + return fmt.Errorf("db file sync: %w", err) + } + + return nil + }(); err != nil { + return err } - // Ensure database file is synced and then truncate the WAL file. - if err := db.fsync(db.file); err != nil { - return fmt.Errorf("db file sync: %w", err) - } else if err := db.walFile.Truncate(0); err != nil { - return fmt.Errorf("truncate wal file: %w", err) - } else if err := db.fsync(db.walFile); err != nil { - return fmt.Errorf("wal file sync: %w", err) - } else if _, err := db.walFile.Seek(0, io.SeekStart); err != nil { - return fmt.Errorf("seek wal file: %w", err) - } + // now we've updated the file. There are existing transactions that are still + // using the WAL, though. So we wait for them to terminate before we unlock + // the rwmu and update the metadata about the WAL. + releaseLock = false db.walPageN = 0 db.pageMap = NewPageMap() - // Notify halted transactions that the WAL has been checkpointed. - db.haltCond.Broadcast() + db.afterCurrentTx(func() { + defer db.rwmu.Unlock() + db.baseWALID = readMetaWALID(db.data) + db.mu.Unlock() + defer db.mu.Lock() + + if err = db.walFile.Truncate(0); err != nil { + db.logger.Errorf("truncate wal file: %w", err) + } else if err = db.fsync(db.walFile); err != nil { + db.logger.Errorf("wal file sync: %w", err) + } else if _, err = db.walFile.Seek(0, io.SeekStart); err != nil { + db.logger.Errorf("seek wal file: %w", err) + } + + }) return nil } @@ -444,21 +604,31 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { } db.mu.Lock() - // note: We cannot defer db.mu.Unlock() here because - // we call tx.Rollback() before if db.readMetaPage - // returns an error, and thus we will deadlock against - // ourselves when the Rollback tries to acquire the db.mu. - // This is why db.mu.Unlock() is done manually below. + defer db.mu.Unlock() if !db.opened { cleanup() - db.mu.Unlock() return nil, ErrClosed } + if db.isDead != nil { + err := db.isDead + cleanup() + return nil, err + } - // Wait for WAL size to be below threshold. - for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize { - db.haltCond.Wait() + // Wait for WAL size to be below threshold, if we're going to write. + // Reads don't care. + if writable { + for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize { + if db.isDead != nil { + err := db.isDead + cleanup() + return nil, err + } + // This implicitly releases db.mu.Lock and comes back with it + // held again. + db.haltCond.Wait() + } } tx := &Tx{ @@ -470,6 +640,11 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { DeleteEmptyContainer: true, } + defer func() { + if err != nil { + tx.rollback(true) + } + }() if writable { tx.dirtyPages = make(map[uint32][]byte) @@ -480,10 +655,6 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { // This page is only written at the end of a dirty transaction. page, err := db.readMetaPage() if err != nil { - // we will deadlock in tx.Rollback() - // on db.mu.Lock unless we manually db.mu.Unlock first. - db.mu.Unlock() - tx.Rollback() return nil, err } copy(tx.meta[:], page) @@ -499,36 +670,102 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { // this avoids recomputing the cache if there are no write txs for a while. if db.rootRecords == nil { if db.rootRecords, err = tx.RootRecords(); err != nil { - db.mu.Unlock() - tx.Rollback() return nil, err } } - db.mu.Unlock() return tx, nil } -// removeTx removes an active transaction from the database. -func (db *DB) removeTx(tx *Tx) error { - // Release writer lock if tx is writable. - if tx.writable { - tx.db.rwmu.Unlock() +// afterCurrentTx produces runs the provided callback, with the db lock +// held, after all current Tx terminate. It should be called with the db +// lock held. +func (db *DB) afterCurrentTx(callback func()) { + if len(db.txs) == 0 { + callback() + return } + txw := &txWaiter{} + txw.ready = make(chan struct{}) + txw.callback = callback + txw.waitingOn = make(map[*Tx]struct{}, len(db.txs)) + for k := range db.txs { + txw.waitingOn[k] = struct{}{} + } + db.txWaiters = append(db.txWaiters, txw) + go func() { + <-txw.ready + // fmt.Printf("afterCurrentTx: locking db\n") + db.mu.Lock() + defer db.mu.Unlock() + // fmt.Printf("afterCurrentTx: running callback\n") + txw.callback() + }() + return +} +// removeTx removes an active transaction from the database. it obtains +// the db lock, and currently drops it, but will later possibly be leaving +// it retained by an asynchronous op that wants to happen before we start +// running new tx. +func (db *DB) removeTx(tx *Tx) error { + // We might want to trigger a checkpoint. Only for writable + // transactions, and only when either there's nothing else open or we + // really need to. + checkpoint := false + if tx.writable { + walSize := db.walSize() + if walSize > db.cfg.MinWALCheckpointSize { + // Might be a good time for a checkpoint. We'll do a checkpoint + // if we're the only transaction, or if we have to. + if len(db.txs) == 1 || walSize > db.cfg.MaxWALCheckpointSize { + checkpoint = true + } + } + // During checkpointing, we'll be preventing writes, but allowing reads. + if !checkpoint { + tx.db.rwmu.Unlock() + } + } + // remove ourselves from the list of transactions the db is keeping. delete(tx.db.txs, tx) + for i := 0; i < len(tx.db.txWaiters); i++ { + txw := tx.db.txWaiters[i] + // in practice this probably never matters, but theoretically the + // goroutine that's waiting on the condition variable may + // not have performed its first test on len(txw.waitingOn) yet. + delete(txw.waitingOn, tx) + // let it know we're done. we've still got db.mu.lock, so it won't + // happen just yet, but it'll be able to continue. + if len(txw.waitingOn) == 0 { + // remove us from the db's list + copy(db.txWaiters[i:], db.txWaiters[i+1:]) + db.txWaiters = db.txWaiters[:len(db.txWaiters)-1] + close(txw.ready) + // decrement i so we don't skip an entry we just copied in to [i] + i-- + } + } // Disassociate from db. tx.db = nil - // Write pages from WAL to DB. - // TODO(bbj): Move this to an async goroutine. - if len(db.txs) == 0 && db.walSize() > db.cfg.MinWALCheckpointSize { - if err := db.checkpoint(); err != nil { - return fmt.Errorf("checkpoint: %w", err) - } + if checkpoint { + // We need to run a checkpoint. This can be semi-asynchronous. + // It needs to wait until every existing transaction has finished, + // because every existing transaction could want to look up pages + // which are in the database before our operations, but which should + // now be in the WAL. We want them to use the WAL instead. + // fmt.Printf("possibly-async checkpoint...\n") + db.afterCurrentTx(func() { + // We still hold db.rwmu here. checkpoint unlocks it when it's + // ready. + // fmt.Printf("checkpoint starting\n") + if err := db.checkpoint(); err != nil { + db.logger.Errorf("async checkpoint: %v", err) + } + }) } - return nil } @@ -554,14 +791,9 @@ func (db *DB) readDBPage(pgno uint32) ([]byte, error) { return db.data[offset : offset+PageSize], nil } -// baseWALID returns the WAL ID stored in the database file meta page. -func (db *DB) baseWALID() int64 { - return readMetaWALID(db.data) -} - // readWALPageByID reads a WAL page by WAL ID. func (db *DB) readWALPageByID(id int64) ([]byte, error) { - return db.readWALPageAt(int(id - db.baseWALID() - 1)) + return db.readWALPageAt(int(id - db.baseWALID - 1)) } // readWALPageAt reads the i-th page in the WAL file. diff --git a/rbf/db_test.go b/rbf/db_test.go index 56170d6eb..c0eb3b3c7 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -13,6 +13,7 @@ import ( _ "net/http/pprof" + "github.com/felixge/fgprof" "github.com/molecula/featurebase/v2/rbf" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "golang.org/x/sync/errgroup" @@ -290,7 +291,8 @@ func TestDB_MultiTx(t *testing.T) { time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) - for i := 0; i < rand.Intn(1000); i++ { + n := rand.Intn(500) + 500 + for i := 0; i < n; i++ { v := rand.Intn(1 << 20) if _, err := tx.Contains("x", uint64(v)); err != nil { return err @@ -315,7 +317,8 @@ func TestDB_MultiTx(t *testing.T) { } defer tx.Rollback() - for j := 0; j < rand.Intn(100); j++ { + n := rand.Intn(90) + 10 + for j := 0; j < n; j++ { v := rand.Intn(1 << 20) if _, err := tx.Add("x", uint64(v)); err != nil { t.Fatal(err) @@ -336,6 +339,119 @@ func TestDB_MultiTx(t *testing.T) { } } +// premake pool of random values +const randPool = (1 << 18) + +// benchmarkOneCheckpoint +func benchmarkOneCheckpoint(b *testing.B, randInts []int) { + cfg := rbfcfg.NewDefaultConfig() + // extremely low to force checkpointing + cfg.MinWALCheckpointSize = rbf.PageSize * 16 + cfg.MaxWALCheckpointSize = rbf.PageSize * 64 + var _ rbfcfg.Config + db := MustOpenDB(b, cfg) + defer MustCloseDB(b, db) + + // Run multiple readers in separate goroutines. + ctx, cancel := context.WithCancel(context.Background()) + g, ctx := errgroup.WithContext(ctx) + for i := 0; i < 8; i++ { + i := i + g.Go(func() error { + for { + if ctx.Err() != nil { + return nil // cancelled, return no error + } else if err := func() error { + tx, err := db.Begin(false) + if err != nil { + return err + } + defer tx.Rollback() + + time.Sleep(time.Duration(rand.Intn(int(3 * time.Millisecond)))) + + times := rand.Intn(1000) + 1 + for j := 0; j < times; j++ { + v := randInts[((i<<10)+j)%(randPool-1)] + if _, err := tx.Contains("x", uint64(v)); err != nil { + return err + } + } + return nil + }(); err != nil { + return err + } + // time.Sleep(time.Duration(rand.Intn(int(3 * time.Millisecond)))) + } + }) + } + + // Continuously set/clear bits while readers are executing. + next := 0 + for i := 0; i < 1000; i++ { + func() { + tx, err := db.Begin(true) + if err != nil { + b.Fatal(err) + } + defer tx.Rollback() + + times := rand.Intn(100) + for j := 0; j < times; j++ { + v := randInts[next] + next = (next + 1) % (randPool - 1) + if j&7 == 0 { + // some removes but they're less frequent + if _, err := tx.Remove("x", uint64(v)); err != nil { + b.Fatal(err) + } + } else { + if _, err := tx.Add("x", uint64(v)); err != nil { + b.Fatal(err) + } + } + + } + if err := tx.Commit(); err != nil { + b.Fatal(err) + } + }() + } + + // Stop readers & wait. + cancel() + if err := g.Wait(); err != nil { + b.Fatal(err) + } +} + +func BenchmarkDbCheckpoint(b *testing.B) { + out, err := os.Create("cp.out") + if err != nil { + b.Fatalf("creating log file: %v", err) + } + done := fgprof.Start(out, fgprof.FormatPprof) + b.StopTimer() + // premake these because otherwise it's >5% of CPU in the reads + randInts := make([]int, randPool) + for i := range randInts { + v1, v2 := rand.Intn(1<<24), rand.Intn(1<<24) + // minimum gives us a skewed distribution which makes lower values more + // likely than higher values, so we get a mix of container types + if v1 < v2 { + randInts[i] = v1 + } else { + randInts[i] = v2 + } + } + b.StartTimer() + for i := 0; i < b.N; i++ { + benchmarkOneCheckpoint(b, randInts) + } + b.StopTimer() + done() +} + // better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests. func TestMain(m *testing.M) { l, err := net.Listen("tcp", ":0") diff --git a/rbf/rbf_test.go b/rbf/rbf_test.go index 48f16335a..4071a1a95 100644 --- a/rbf/rbf_test.go +++ b/rbf/rbf_test.go @@ -11,6 +11,7 @@ import ( "sort" "testing" + "github.com/molecula/featurebase/v2/logger" "github.com/molecula/featurebase/v2/rbf" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/testhook" @@ -65,6 +66,13 @@ func NewDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB { // MustOpenDB returns a db opened on a temporary file. On error, fail test. func MustOpenDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB { tb.Helper() + if len(cfg) == 0 || cfg[0] == nil { + newconf := rbfcfg.NewDefaultConfig() + newconf.Logger = logger.NewLogfLogger(tb) + cfg = []*rbfcfg.Config{newconf} + } else if cfg[0].Logger == nil { + cfg[0].Logger = logger.NewLogfLogger(tb) + } db := NewDB(tb, cfg...) if err := db.Open(); err != nil { tb.Fatal(err) @@ -78,7 +86,14 @@ func MustCloseDB(tb testing.TB, db *rbf.DB) { tb.Helper() if err := db.Check(); err != nil && err != rbf.ErrClosed { tb.Fatal(err) - } else if n := db.TxN(); n != 0 { + } + MustCloseDBNoCheck(tb, db) +} + +// MustCloseDBNoCheck closes db. On error, fail test. +func MustCloseDBNoCheck(tb testing.TB, db *rbf.DB) { + tb.Helper() + if n := db.TxN(); n != 0 { tb.Fatalf("db still has %d active transactions; must closed before closing db", n) } else if err := db.Close(); err != nil && err != rbf.ErrClosed { tb.Fatal(err) diff --git a/rbf/tx.go b/rbf/tx.go index 37c76a351..5c6c7f7c6 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -109,24 +109,31 @@ func (tx *Tx) Commit() error { // future plan: after checkpoint is moved to background // or not every removeTx, then we can move the // tx.db.rootRecords = tx.rootRecords into removeTx(). - + // + // ... or maybe not: let's do that part here, and then removeTx + // may or may not start a checkpoint, possibly asynchronously. + // // avoid race detector firing on a write race here - // vs the read of rootRecords at db.Begin() + // vs the read of rootRecords at db.Begin(), then release + // the lock, because we need removeTx to grab the lock to + // work, but if it wants to checkpoint, it wants to be able to return + // to us here and still be holding the lock. tx.db.mu.Lock() - defer tx.db.mu.Unlock() tx.db.rootRecords = tx.rootRecords tx.db.pageMap = tx.pageMap tx.db.walPageN = tx.walPageN - return tx.db.removeTx(tx) + tx.db.mu.Unlock() } - // Disconnect transaction from DB. tx.db.mu.Lock() defer tx.db.mu.Unlock() + // Disconnect transaction from DB. return tx.db.removeTx(tx) } -func (tx *Tx) Rollback() { +func (tx *Tx) Rollback() { tx.rollback(false) } + +func (tx *Tx) rollback(hasDBLock bool) { tx.mu.Lock() defer tx.mu.Unlock() @@ -141,8 +148,10 @@ func (tx *Tx) Rollback() { } // Disconnect transaction from DB. - tx.db.mu.Lock() - defer tx.db.mu.Unlock() + if !hasDBLock { + tx.db.mu.Lock() + defer tx.db.mu.Unlock() + } vprint.PanicOn(tx.db.removeTx(tx)) } @@ -732,6 +741,27 @@ func (tx *Tx) Check() error { return nil } +func (tx *Tx) checkPage(pgno, parent, typ uint32) error { + switch typ { + case PageTypeBranch: + return tx.checkBranchPage(pgno, parent, typ) + default: + return nil + } +} + +func (tx *Tx) checkBranchPage(pgno, parent, typ uint32) error { + page, _, err := tx.readPage(pgno) + if err != nil { + return err + } + + if readCellN(page) == 0 { + return fmt.Errorf("branch page %d is empty", pgno) + } + return nil +} + // checkPageAllocations ensures that all pages are either in-use or on the freelist. func (tx *Tx) checkPageAllocations() error { freePageSet, err := tx.freePageSet() @@ -821,7 +851,7 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) { // Traverse freelist and mark pages as in-use. if err := tx.walkTree(readMetaFreelistPageNo(tx.meta[:]), 0, func(pgno, parent, typ uint32) error { m[pgno] = struct{}{} - return nil + return tx.checkPage(pgno, parent, typ) }); err != nil { return m, err } @@ -837,7 +867,8 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) { if err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32) error { m[pgno] = struct{}{} - return nil + + return tx.checkPage(pgno, parent, typ) }); err != nil { return m, err } diff --git a/rbf/tx_test.go b/rbf/tx_test.go index 1004437a3..5be3a50ae 100644 --- a/rbf/tx_test.go +++ b/rbf/tx_test.go @@ -2,8 +2,11 @@ package rbf_test import ( + "encoding/binary" "fmt" "math/rand" + "os" + "strings" "sync" "testing" "time" @@ -433,6 +436,52 @@ func TestTx_DeallocateToFreeList(t *testing.T) { } } +func TestTx_Remove(t *testing.T) { + t.Parallel() + + db := MustOpenDB(t) + defer MustCloseDB(t, db) + + tx := MustBegin(t, db, true) + defer tx.Rollback() + + if err := tx.CreateBitmap("x"); err != nil { + t.Fatal(err) + } + + // Insert large array values. + var values []uint64 + for i := 0; i < 1000; i++ { + for j := 0; j < rbf.ArrayMaxSize; j++ { + v := uint64((i << 16) + j) + values = append(values, v) + + if _, err := tx.Add("x", v); err != nil { + t.Fatalf("Add(%d) err=%q", v, err) + } + } + } + + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + + tx = MustBegin(t, db, true) + defer tx.Rollback() + + // Remove all array values. + for _, i := range rand.Perm(len(values)) { + v := values[i] + if _, err := tx.Remove("x", v); err != nil { + t.Fatalf("Remove(%d) err=%q", v, err) + } + } + + if err := tx.Commit(); err != nil { + t.Fatal(err) + } +} + func TestTx_AddRemove_Quick(t *testing.T) { if testing.Short() { t.Skip("-short enabled, skipping") @@ -770,3 +819,83 @@ func TestTx_DeleteBitmapsWithPrefix(t *testing.T) { checkInfos() } + +func TestTx_Check(t *testing.T) { + t.Run("EmptyBranchPage", func(t *testing.T) { + t.Parallel() + + db := MustOpenDB(t) + defer MustCloseDBNoCheck(t, db) + tx := MustBegin(t, db, true) + defer tx.Rollback() + + if err := tx.CreateBitmap("x"); err != nil { + t.Fatal(err) + } + + // Insert enough array containers to split page. + for i := 0; i < 1000; i++ { + if _, err := tx.Add("x", uint64(i<<16)); err != nil { + t.Fatalf("Add(%d) err=%q", i<<16, err) + } + } + + // Read page types for all pages. + infos, err := tx.PageInfos() + if err != nil { + t.Fatal(err) + } + + // Commit & checkpoint to flush to the data file. + if err := tx.Commit(); err != nil { + t.Fatal(err) + } else if err := db.Checkpoint(); err != nil { + t.Fatal(err) + } + + // Corrupt first branch page found by zeroing out the cell count. + var pgno uint32 + for _, info := range infos { + if info, ok := info.(*rbf.BranchPageInfo); ok { + pgno = info.Pgno + page := mustReadPage(t, db.DataPath(), pgno) + binary.BigEndian.PutUint16(page[8:10], 0) // zero cell count + mustWritePage(t, db.DataPath(), pgno, page) + break + } + } + + // Verify that check now returns an error. + if err := db.Check(); err == nil || !strings.Contains(err.Error(), fmt.Sprintf("branch page %d is empty", pgno)) { + t.Fatalf("unexpected error: %#v", err) + } + }) +} + +func mustReadPage(tb testing.TB, path string, pgno uint32) []byte { + tb.Helper() + f, err := os.Open(path) + if err != nil { + tb.Fatal(err) + } + defer f.Close() + + buf := make([]byte, rbf.PageSize) + if _, err := f.ReadAt(buf, int64(pgno)*rbf.PageSize); err != nil { + tb.Fatal(err) + } + return buf +} + +func mustWritePage(tb testing.TB, path string, pgno uint32, buf []byte) { + tb.Helper() + f, err := os.OpenFile(path, os.O_WRONLY, 0666) + if err != nil { + tb.Fatal(err) + } + defer f.Close() + + if _, err := f.WriteAt(buf, int64(pgno)*rbf.PageSize); err != nil { + tb.Fatal(err) + } +} diff --git a/server.go b/server.go index 8858ab122..0c74a2fa5 100644 --- a/server.go +++ b/server.go @@ -476,7 +476,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { } s.holder = NewHolder(path, s.holderConfig) s.holder.Stats.SetLogger(s.logger) - s.holder.Logger.Infof("RowCacheOn: %v", s.holderConfig.RowcacheOn) cwd, err := os.Getwd() if err != nil { return nil, err diff --git a/server/config.go b/server/config.go index 45cf56a54..95eabd727 100644 --- a/server/config.go +++ b/server/config.go @@ -4,14 +4,19 @@ package server import ( "context" "fmt" + "io" "log" "net" "net/url" + "os" + "path/filepath" "runtime" "strconv" "strings" "time" + + "github.com/molecula/featurebase/v2/authz" petcd "github.com/molecula/featurebase/v2/etcd" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/storage" @@ -199,9 +204,9 @@ type Config struct { // "rbf". Storage *storage.Config `toml:"storage"` - // RowcacheOn, if true, turns on the row cache for all storage backends. - // The default is now off because it makes rbf queries faster and uses - // much less memory. + // RowcacheOn permanently disabled. No longer useful w/ RBF. Left + // for backward compatibility but will be removed in a future + // version. RowcacheOn bool `toml:"rowcache-on"` // RBFConfig defines all externally configurable RBF flags. @@ -229,16 +234,15 @@ type Config struct { // Toggles /schema/details endpoint. If off, it returns empty. SchemaDetailsOn bool `toml:"schema-details-on"` + Auth Auth } type Auth struct { // Enable AuthZ/AuthN for featurebase server Enable bool `toml:"enable"` - - // Application/Client ID + ClientId string `toml:"client-id"` - ClientSecret string `toml:"client-secret"` AuthorizeURL string `toml:"authorize-url"` TokenURL string `toml:"token-url"` @@ -247,6 +251,8 @@ type Auth struct { Scopes []string `toml:"scopes"` HashKey string `toml:"hash-key"` BlockKey string `toml:"block-key"` + PermissionsFile string `toml:"permissions"` + Auth authz.Auth `toml:"auth"` } // Namespace returns the namespace to use based on the Future flag. @@ -611,9 +617,9 @@ func lookupAddr(ctx context.Context, resolver *net.Resolver, host string) (strin return addrs[0].String(), nil } -func (c *Config) ValidateAuth() ([]error, error) { +func (c *Config) ValidateAuth() (errors []error) { if !c.Auth.Enable { - return []error{}, nil + return } authConfig := map[string]string{ "ClientId": c.Auth.ClientId, @@ -626,7 +632,6 @@ func (c *Config) ValidateAuth() ([]error, error) { "BlockKey": c.Auth.BlockKey, } - errors := make([]error, 0) for name, value := range authConfig { if value == "" { errors = append(errors, fmt.Errorf("empty string for auth config %s", name)) @@ -647,20 +652,101 @@ func (c *Config) ValidateAuth() ([]error, error) { } } } + if len(c.Auth.Scopes) == 0 { errors = append(errors, fmt.Errorf("must provide scope for authentication with IdP - for access and refresh token")) } - if len(errors) > 0 { - return errors, fmt.Errorf("there were errors validating config") + + return errors +} + +func (c *Config) ValidatePermissions(permsFile io.Reader) (errors []error) { + + var p authz.GroupPermissions + if err := p.ReadPermissionsFile(permsFile); err != nil { + return append(errors, err) } - return errors, nil + + if len(p.Permissions) == 0 { + return append(errors, fmt.Errorf("no group permissions found in permissions file: %s", c.Auth.PermissionsFile)) + } + + for groupId, indexPerm := range p.Permissions { + if groupId == "" { + errors = append(errors, fmt.Errorf("empty string for group id in permissions file %s", c.Auth.PermissionsFile)) + continue + } + + for index, perm := range indexPerm { + if index == "" { + errors = append(errors, fmt.Errorf("empty string for index for group id %s in permissions file %s ", groupId, c.Auth.PermissionsFile)) + continue + } + + if perm == "" { + errors = append(errors, fmt.Errorf("empty string for permission for group id %s and index %s in permissions file %s", groupId, index, c.Auth.PermissionsFile)) + continue + } + + if !((perm == "write") || (perm == "read")) { + errors = append(errors, fmt.Errorf("not a valid permission %s for group id %s and index %s in permissions file %s; expected permissions are read or write", perm, groupId, index, c.Auth.PermissionsFile)) + continue + } + } + } + + if p.Admin == "" { + errors = append(errors, fmt.Errorf("empty string for admin in permissions file: %s", c.Auth.PermissionsFile)) + + } + + return errors +} + +func (c *Config) ValidatePermissionsFile() (err error) { + + if c.Auth.PermissionsFile == "" { + return fmt.Errorf("empty string for auth config permissions file") + } + + fileExt := filepath.Ext(c.Auth.PermissionsFile) + if (fileExt != ".yaml") && (fileExt != ".yml") { + return fmt.Errorf("invalid file extension for auth config permissions file: %s", c.Auth.PermissionsFile) + } + return } func (c *Config) MustValidateAuth() { - if errors, err := c.ValidateAuth(); err != nil { - for _, e := range errors { + + errorsAuth := c.ValidateAuth() + if len(errorsAuth) > 0 { + for _, e := range errorsAuth { log.Println(e) } - log.Fatal(err) + } + + var errorsPerm []error + errorsPermFile := c.ValidatePermissionsFile() + if errorsPermFile == nil { + permsFile, err := os.Open(c.Auth.PermissionsFile) + if err != nil { + log.Println(err) + } + + defer permsFile.Close() + + errorsPerm = c.ValidatePermissions(permsFile) + if len(errorsPerm) > 0 { + for _, e := range errorsPerm { + log.Println(e) + } + } + + } else { + log.Println(errorsPermFile) + } + + if len(errorsAuth) > 0 || len(errorsPerm) > 0 || errorsPermFile != nil { + log.Fatal(fmt.Errorf("there were errors validating authN/authZ config and/or permissions")) } } diff --git a/server/config_internal_test.go b/server/config_internal_test.go index ff262ee4b..0d5c17778 100644 --- a/server/config_internal_test.go +++ b/server/config_internal_test.go @@ -8,6 +8,7 @@ import ( "os" "strings" "testing" + "github.com/molecula/featurebase/v2/authz" ) type addrs struct{ bind, advertise string } @@ -280,11 +281,12 @@ func TestConfig_validateAuth(t *testing.T) { errorMesgEmpty := "empty string" errorMesgURL := "invalid URL" errorMesgScope := "must provide scope" + errorMesgKey := "invalid key length" validTestURL := "https://url.com/" validClientID := "clientid" validClientSecret := "clientSecret" validKey := "3db6665be8b860af422155acf2346d4fcb46678fca42e60d934abe0b7ce43600" - notValidURL := "not-a-url" + invalidURL := "not-a-url" emptyString := "" validStringSlice := []string{"https://graph.microsoft.com/.default", "offline_access"} validString := "asdfqwer1234asdfzxcv" @@ -297,7 +299,6 @@ func TestConfig_validateAuth(t *testing.T) { expErrs []string input Auth }{ - { // Auth enabled, all configs are set to empty string []string{ @@ -309,7 +310,6 @@ func TestConfig_validateAuth(t *testing.T) { errorMesgEmpty, errorMesgEmpty, errorMesgEmpty, - errorMesgScope, }, Auth{ Enable: enable, @@ -319,168 +319,65 @@ func TestConfig_validateAuth(t *testing.T) { TokenURL: emptyString, GroupEndpointURL: emptyString, LogoutURL: emptyString, - Scopes: emptySlice, + Scopes: validStringSlice, HashKey: emptyString, BlockKey: emptyString, }, }, { - // Auth enabled, some configs are set to empty string + // Auth enabled, keys are invalid length []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgScope, - }, - Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: emptyString, - AuthorizeURL: emptyString, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - LogoutURL: emptyString, - Scopes: emptySlice, - HashKey: emptyString, - BlockKey: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgScope, - }, - Auth{ - Enable: enable, - ClientId: emptyString, - ClientSecret: validClientSecret, - AuthorizeURL: emptyString, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - LogoutURL: emptyString, - Scopes: emptySlice, - HashKey: emptyString, - BlockKey: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgURL, - errorMesgScope, - }, - Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: validClientSecret, - AuthorizeURL: emptyString, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - LogoutURL: notValidURL, - Scopes: emptySlice, - HashKey: emptyString, - BlockKey: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgScope, + errorMesgKey, + errorMesgKey, }, Auth{ Enable: enable, ClientId: validClientID, ClientSecret: validClientSecret, AuthorizeURL: validTestURL, - TokenURL: emptyString, - GroupEndpointURL: emptyString, - LogoutURL: emptyString, - Scopes: emptySlice, - HashKey: emptyString, - BlockKey: emptyString, - }, - }, - { - // Auth enabled, some configs are set to empty string - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgEmpty, - errorMesgScope, - }, - Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: validClientSecret, - AuthorizeURL: validTestURL, - TokenURL: validTestURL, - GroupEndpointURL: emptyString, - LogoutURL: validTestURL, - Scopes: emptySlice, - HashKey: emptyString, - BlockKey: emptyString, - }, - }, - { - // Auth enabled, some strings are set to invalid URL - []string{ - errorMesgEmpty, - errorMesgEmpty, - errorMesgURL, - errorMesgURL, - }, - Auth{ - Enable: enable, - ClientId: validClientID, - ClientSecret: validClientSecret, - AuthorizeURL: notValidURL, TokenURL: validTestURL, GroupEndpointURL: validTestURL, - LogoutURL: notValidURL, + LogoutURL: validTestURL, Scopes: validStringSlice, - HashKey: emptyString, - BlockKey: emptyString, + HashKey: validString, + BlockKey: validString, }, }, { - // Auth enabled, some strings are set to invalid URL + // Auth enabled, some URLs are set to invalid URL []string{ errorMesgURL, errorMesgURL, errorMesgURL, - errorMesgEmpty, }, Auth{ Enable: enable, ClientId: validClientID, ClientSecret: validClientSecret, AuthorizeURL: validTestURL, - TokenURL: notValidURL, - GroupEndpointURL: notValidURL, - LogoutURL: notValidURL, + TokenURL: invalidURL, + GroupEndpointURL: invalidURL, + LogoutURL: invalidURL, Scopes: validStringSlice, - HashKey: emptyString, + HashKey: validKey, + BlockKey: validKey, + }, + }, + { + // Auth enabled, all configs are set properly except scope + []string{ + errorMesgScope, + }, + Auth{ + Enable: enable, + ClientId: validClientID, + ClientSecret: validClientSecret, + AuthorizeURL: validTestURL, + TokenURL: validTestURL, + GroupEndpointURL: validTestURL, + LogoutURL: validTestURL, + Scopes: emptySlice, + HashKey: validKey, BlockKey: validKey, }, }, @@ -501,19 +398,19 @@ func TestConfig_validateAuth(t *testing.T) { }, }, { - // Auth disabled, all configs are set to some values + // Auth disabled, some configs are set to values []string{}, Auth{ Enable: disable, - ClientId: validString, + ClientId: emptyString, ClientSecret: validString, - AuthorizeURL: validString, - TokenURL: validTestURL, - GroupEndpointURL: validTestURL, + AuthorizeURL: emptyString, + TokenURL: emptyString, + GroupEndpointURL: invalidURL, LogoutURL: validTestURL, Scopes: validStringSlice, HashKey: validKey, - BlockKey: validKey, + BlockKey: emptyString, }, }, } @@ -523,9 +420,9 @@ func TestConfig_validateAuth(t *testing.T) { c := NewConfig() c.Auth = test.input - errors, err := c.ValidateAuth() + errors := c.ValidateAuth() if len(test.expErrs) > 0 { - if err == nil { + if errors == nil { t.Fatal("expected errors, but none were found") } } @@ -535,11 +432,121 @@ func TestConfig_validateAuth(t *testing.T) { t.Fatalf("expected %v errors but got %v", len(test.expErrs), len(errors)) } - // for i, e := range errors { - // if !strings.Contains(e.Error(), test.expErrs[i]) { - // t.Errorf("expected error to contain %s, but got %s", test.expErrs[i], e.Error()) - // } - // } + for i, e := range errors { + if !strings.Contains(e.Error(), test.expErrs[i]) { + t.Errorf("expected error to contain %s, but got %s", test.expErrs[i], e.Error()) + } + } + }) + } +} + +func TestConfig_validatePermissions(t *testing.T) { + permissions0 := `` + + permissions1 := `user-groups: + "": + "test": "read" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions2 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "": "write" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions3 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions4 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "readwrite" +admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` + + permissions5 := `user-groups: + "dca35310-ecda-4f23-86cd-876aee559900": + "test": "read"` + + tests := []struct { + err string + input string + }{ + { + "no group permissions found in permissions file", + permissions0, + }, + { + "empty string for group id", + permissions1, + }, + { + "empty string for index", + permissions2, + }, + { + "empty string for permission", + permissions3, + }, + { + "not a valid permission", + permissions4, + }, + { + "empty string for admin in permissions file", + permissions5, + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + + c := NewConfig() + c.Auth.PermissionsFile = "test.yaml" + + permFile := strings.NewReader(test.input) + errors := c.ValidatePermissions(permFile) + + if errors == nil { + t.Fatal("expected errors, but none were found") + } + + for _, err := range errors { + if !strings.Contains(err.Error(), test.err) { + t.Errorf("expected error to contain %s, but got %s", test.err, err.Error()) + + } + } + }) + } +} + +func TestConfig_validatePermissionsFilename(t *testing.T) { + + tests := []struct { + err string + input string + }{ + { + "empty string for auth config permissions file", + "", + }, + { + "invalid file extension for auth config permissions file", + "permissions.txt", + }, + } + + for i, test := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + c := NewConfig() + c.Auth.PermissionsFile = test.input + + if err := c.ValidatePermissionsFile(); err != nil { + if !strings.Contains(err.Error(), test.err) { + t.Errorf("expected error to contain %s, but got %s", test.err, err.Error()) + } + } }) } } diff --git a/server/server.go b/server/server.go index 2b2f4ddc8..911f82177 100644 --- a/server/server.go +++ b/server/server.go @@ -30,6 +30,7 @@ import ( pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/authn" + "github.com/molecula/featurebase/v2/authz" "github.com/molecula/featurebase/v2/boltdb" "github.com/molecula/featurebase/v2/encoding/proto" petcd "github.com/molecula/featurebase/v2/etcd" @@ -481,7 +482,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerClusterName(m.Config.Cluster.Name), pilosa.OptServerSerializer(proto.Serializer{}), pilosa.OptServerStorageConfig(m.Config.Storage), - pilosa.OptServerRowcacheOn(m.Config.RowcacheOn), + pilosa.OptServerRowcacheOn(false), pilosa.OptServerRBFConfig(m.Config.RBFConfig), pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory), pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength), @@ -524,12 +525,22 @@ func (m *Command) SetupServer() error { if m.Config.Auth.Enable { m.Config.MustValidateAuth() + permsFile, err := os.Open(m.Config.Auth.PermissionsFile) + if err != nil { + return err + } + defer permsFile.Close() + + var p authz.GroupPermissions + if err = p.ReadPermissionsFile(permsFile); err != nil { + return err + } + ac := m.Config.Auth m.auth, err = authn.NewAuth(m.logger, m.listenURI.String(), ac.Scopes, ac.AuthorizeURL, ac.TokenURL, ac.GroupEndpointURL, ac.LogoutURL, ac.ClientId, ac.ClientSecret, ac.HashKey, ac.BlockKey) if err != nil { return errors.Wrap(err, "instantiating authN object") } - } m.Handler, err = http.NewHandler( diff --git a/testBackupRestore.sh b/testBackupRestore.sh deleted file mode 100755 index 7bcbe3894..000000000 --- a/testBackupRestore.sh +++ /dev/null @@ -1,57 +0,0 @@ -#!/bin/bash - -set -eux - -declare STATUS="NORMAL" -declare TIMEOUT=30 -sleep 4 -STATUS=$STATUS timeout -s TERM $TIMEOUT bash -c \ - 'while [[ ${STATUS_RECEIVED} != ${STATUS} ]];\ - do STATUS_RECEIVED=$(curl --connect-timeout 1 -s pilosa0:10101/status | jq -r ".state") && \ - echo "received status: $STATUS_RECEIVED" && \ - sleep 1;\ - done;' -echo "NOW DO STUFF" -datagen --source kitchensink_keyed -e 9999 --pilosa.index sink --pilosa.batch-size 10000 --pilosa.hosts pilosa0:10101 -before=$(/featurebase chksum --host pilosa0:10101) -/featurebase backup -o backupdir --host pilosa0:10101 -curl -X DELETE -s pilosa0:10101/index/sink -/featurebase restore -s backupdir --host pilosa0:10101 -after=$(/featurebase chksum --host pilosa0:10101) -if [ "$before" = "$after" ]; then - echo "PASS Cluster" -else - echo "FAIL Single" - exit 1 -fi -/featurebase restore -s backupdir --host pilosax:10101 -single=$(/featurebase chksum --host pilosax:10101) -if [ "$before" = "$single" ]; then - echo "PASS Single" - exit 0 -else - echo "FAIL Single" - exit 1 -fi - -datagen --source texas_health -e 9999 --pilosa.index newsink --pilosa.batch-size 10000 --pilosa.hosts pilosa0:10101 -before=$(/featurebase chksum --host pilosa0:10101) -/featurebase backup -o newbackupdir --host pilosa0:10101 --index newsink -curl -X DELETE -s pilosa0:10101/index/newsink -/featurebase restore -s newbackupdir --host pilosa0:10101 -after=$(/featurebase chksum --host pilosa0:10101) -if [ "$before" = "$after" ]; then - echo "PASS Cluster Table" -else - echo "FAIL Single Table" - exit 1 -fi -/featurebase restore -s newbackupdir --host pilosax:10101 -single=$(/featurebase chksum --host pilosax:10101) -if [ "$before" = "$single" ]; then - echo "PASS Single Table" - exit 0 -else - echo "FAIL Single Table" - exit 1 -fi diff --git a/translate.go b/translate.go index a9d547cb7..be1c47306 100644 --- a/translate.go +++ b/translate.go @@ -568,6 +568,8 @@ func (s *InMemTranslateStore) WriteTo(w io.Writer) (int64, error) { // don't expect to use InMemTranslateStore much, it's mostly there to // avoid disk load during testing. func (s *InMemTranslateStore) ReadFrom(r io.Reader) (count int64, err error) { + s.mu.Lock() + defer s.mu.Unlock() var bytes []byte bytes, err = ioutil.ReadAll(r) count = int64(len(bytes)) diff --git a/txfactory.go b/txfactory.go index 62f527653..12fa12981 100644 --- a/txfactory.go +++ b/txfactory.go @@ -242,10 +242,15 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) { } // qcx.write reflects the top executor determination - // if a write will be done at the end, so we upgrade - // the "local" read Tx to be writes, so that they - // don't deadlock against themselves. - o.Write = o.Write || qcx.write + // if a write will be happen at some point, in which case, to avoid + // locking problems with multi-shard things, we (probably incorrectly) + // treat every Tx as its own individual separate Tx. + // + // But we still want to open non-write transactions individually, we + // just can't recycle them (because write operations will come in and + // we want them to work and commit right away so we're not holding a write + // lock for long). + writeLogic := o.Write || qcx.write // In general, we make ALL write transactions local, and never reuse them // below. Previously this was to help lmdb. @@ -273,7 +278,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) { return *qcx.RequiredForAtomicWriteTx, NoopFinisher, nil } - if !o.Write && qcx.Grp != nil { + if !writeLogic && qcx.Grp != nil { // read, with a group in place. finisher = func(perr *error) {} // finisher is a returned value diff --git a/view.go b/view.go index f3bd27b3a..5a8e23ae1 100644 --- a/view.go +++ b/view.go @@ -619,7 +619,9 @@ func (v *view) bitDepth(shards []uint64) (uint64, error) { var maxBitDepth uint64 for _, shard := range shards { + v.mu.RLock() frag, ok := v.fragments[shard] + v.mu.RUnlock() if !ok || frag == nil { continue }