Merge branch 'master' into percentile-timestamp-decimal

This commit is contained in:
reesporte 2022-01-03 11:12:23 -06:00
commit 72adb177ae
45 changed files with 2185 additions and 453 deletions

View file

@ -151,7 +151,7 @@ jobs:
- checkout-plus
- skip-if-root-unchanged
- setup_remote_docker
- run: make clustertests-build
- run: make clustertests
release:
executor:
name: golang

View file

@ -11,7 +11,8 @@ include:
paths:
- .go/pkg/mod/
variables:
GOVERSION: "1.16.9"
GOVERSION: "1.16.10"
stages:
- lint
@ -19,19 +20,13 @@ stages:
- build
- integration
#before_script:
#- echo "before_script"
#- git version
#- go env -w GOPRIVATE=github.com/molecula
#- mkdir -p .go
#- go version
#- go env -w GO111MODULE=on
golangci-lint:
image: golangci/golangci-lint:v1.39.0
stage: lint
extends: .go-cache
allow_failure: false
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- echo "Checking for issues in new code"
- golangci-lint run -v
@ -41,6 +36,8 @@ build lattice:
image: node:14
variables:
CI: "false"
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- cd lattice
- yarn install
@ -59,6 +56,8 @@ run jest tests:
image: node:14
variables:
CI: "true"
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- echo "Testing lattice..."
- cd lattice
@ -70,8 +69,10 @@ run jest tests:
run go tests:
stage: test
image: golang:1.16.10
image: golang:$GOVERSION
extends: .go-cache
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- echo "Running featurebase unit tests..."
- PKG_LIST=$(go list ./... | grep -v internal/clustertests | paste -s -d, -)
@ -84,6 +85,8 @@ run go tests future:
stage: test
image: golang:1.17.3
extends: .go-cache
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- echo "Running featurebase unit tests..."
- PKG_LIST=$(go list ./... | grep -v internal/clustertests | paste -s -d, -)
@ -95,7 +98,9 @@ run go tests future:
run go tests with output:
stage: test
image: golang:1.16.10
image: golang:$GOVERSION
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- echo "Running featurebase unit tests to capture JSON output..."
- go test -json > test-report.out
@ -108,6 +113,8 @@ upload to sonarcloud:
image: sonarsource/sonar-scanner-cli:4.6
variables:
SONAR_TOKEN: $SONAR_TOKEN
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- sonar-scanner -Dsonar.projectKey=molecula_featurebase -Dsonar.organization=molecula -Dsonar.sources=. -Dsonar.host.url=https://sonarcloud.io -Dsonar.go.coverage.reportPaths=coverage.out -Dsonar.go.tests.reportPaths=test-report.out -Dsonar.javascript.lcov.reportPaths=lattice/coverage/lcov.info
needs:
@ -117,7 +124,9 @@ upload to sonarcloud:
build for linux amd64:
stage: build
image: golang:1.16.10
image: golang:$GOVERSION
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- rm -r lattice
- tar -xvf lattice.tar.gz
@ -130,7 +139,9 @@ build for linux amd64:
build for linux arm64:
stage: build
image: golang:1.16.10
image: golang:$GOVERSION
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- rm -r lattice
- tar -xvf lattice.tar.gz
@ -143,7 +154,9 @@ build for linux arm64:
build for darwin amd64:
stage: build
image: golang:1.16.10
image: golang:$GOVERSION
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- rm -r lattice
- tar -xvf lattice.tar.gz
@ -156,7 +169,9 @@ build for darwin amd64:
build for darwin arm64:
stage: build
image: golang:1.16.10
image: golang:$GOVERSION
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
script:
- rm -r lattice
- tar -xvf lattice.tar.gz
@ -169,7 +184,9 @@ build for darwin arm64:
package for linux amd64:
stage: build
image: golang:1.16.10
image: golang:$GOVERSION
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
variables:
GOOS: "linux"
GOARCH: "amd64"
@ -190,6 +207,8 @@ build container fb:
- "build for linux amd64"
tags:
- shell
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
before_script:
- echo "${DOCKER_DEPLOY_TOKEN}" | docker login -u ${DOCKER_DEPLOY_USER} --password-stdin ${CI_REGISTRY}
script:
@ -205,6 +224,8 @@ deploy node for linux amd64:
variables:
PROFILE: "default"
AWS_SSH_PRIVATE_KEY: $AWS_SSH_PRIVATE_KEY
rules:
- if: '$CI_PIPELINE_SOURCE == "push" && $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH'
before_script:
- aws configure set aws_access_key_id $AWS_ACCESS_KEY_ID
- aws configure set aws_secret_access_key $AWS_SECRET_ACCESS_KEY

View file

@ -149,12 +149,10 @@ DOCKER_COMPOSE=internal/clustertests/docker-compose.yml
clustertests: vendor
docker-compose -f $(DOCKER_COMPOSE) down
docker-compose -f $(DOCKER_COMPOSE) build
docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1
docker-compose -f $(DOCKER_COMPOSE) up -d pilosa1 pilosa2 pilosa3
docker-compose -f $(DOCKER_COMPOSE) run client1
docker-compose -f $(DOCKER_COMPOSE) down
# Like clustertests, but rebuilds all images.
clustertests-build: vendor
docker-compose -f $(DOCKER_COMPOSE) down -v
docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1 --build
# Install Pilosa
install:

39
api.go
View file

@ -23,6 +23,7 @@ import (
"github.com/molecula/featurebase/v2/disco"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/rbf"
//"github.com/molecula/featurebase/v2/pg"
"github.com/molecula/featurebase/v2/pql"
@ -130,9 +131,16 @@ func (api *API) SetAPIOptions(opts ...apiOption) error {
var validAPIMethods = map[disco.ClusterState]map[apiMethod]struct{}{
disco.ClusterStateStarting: methodsCommon,
disco.ClusterStateNormal: appendMap(methodsCommon, methodsNormal),
disco.ClusterStateDegraded: appendMap(methodsCommon, methodsDegraded),
// Ideally, this would be just `appendMap(methodsCommon, methodsDegraded)`,
// but in an attempt to reduce the influence that state (determined by etcd)
// has on a node under load, this is set to effectively allow all requests
// in a DEGRADED state.
disco.ClusterStateDegraded: appendMap(methodsCommon, methodsNormal),
disco.ClusterStateResizing: appendMap(methodsCommon, methodsResizing),
disco.ClusterStateDown: methodsCommon,
// Ideally, this would be just `methodsCommon`, but in an attempt to reduce
// the influence that state (determined by etcd) has on a node under load,
// this is set to effectively allow all requests in a DOWN state.
disco.ClusterStateDown: appendMap(methodsCommon, methodsNormal),
}
func appendMap(a, b map[apiMethod]struct{}) map[apiMethod]struct{} {
@ -2753,8 +2761,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
@ -2768,6 +2776,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
@ -3148,6 +3164,21 @@ func (api *API) Plan(ctx context.Context, q string) (*Stmt, error) {
return api.server.PlanSQL(ctx, q)
}
func (api *API) RBFDebugInfo() map[string]*rbf.DebugInfo {
infos := make(map[string]*rbf.DebugInfo)
for key, dbShard := range api.holder.Txf().dbPerShard.Flatmap {
wrapper, ok := dbShard.W.(*RbfDBWrapper)
if !ok {
continue
}
skey := fmt.Sprintf("%s/%d", key.index, key.shard)
infos[skey] = wrapper.db.DebugInfo()
}
return infos
}
type serverInfo struct {
ShardWidth uint64 `json:"shardWidth"`
ReplicaN int `json:"replicaN"`

View file

@ -1415,3 +1415,25 @@ func TestVariousApiTranslateCalls(t *testing.T) {
*/
}
}
func TestAPI_RBFDebugInfo(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
c := test.MustRunCluster(t, 1,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
coord := c.GetPrimary()
if _, err := coord.API.CreateIndex(ctx, "i", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if infos := coord.API.RBFDebugInfo(); infos == nil {
t.Fatal("expected info")
}
}

View file

@ -1,25 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package auth
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"`
}

150
authz/authorization.go Normal file
View file

@ -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
}

314
authz/authorization_test.go Normal file
View file

@ -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)
}
})
}
}

View file

@ -80,8 +80,14 @@ type InternalClient interface {
GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error)
GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error)
ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error
ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error
// ImportFieldKeys and ImportIndexKeys are mainly used when
// restoring a backup. They take a readerFunc which returns a
// reader rather than taking an io.Reader directly to allow for
// efficient retries (rather than reading the entire request body
// into a buffer and reusing it). Reader returned from the func
// must be properly closed by the implementation.
ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error
ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error
// SetInternalAPI tells the client the API it should use for internal/loopback ops
// where applicable.
@ -277,11 +283,11 @@ func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map
func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) {
return nil, nil
}
func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error {
func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error {
return nil
}
func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error {
func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error {
return nil
}

View file

@ -23,11 +23,13 @@ Backs up a FeatureBase server to a local, tar-formatted snapshot file.
}
flags := ccmd.Flags()
flags.StringVarP(&cmd.OutputDir, "output", "o", "", "output dir to write to")
flags.BoolVar(&cmd.NoSync, "no-sync", false, "disable file sync")
flags.IntVar(&cmd.Concurrency, "concurrency", cmd.Concurrency, "number of concurrent backup goroutines")
flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.")
flags.StringVar(&cmd.Index, "index", "", "index to backup, default backs up all indexes. ")
flags.StringVarP(&cmd.OutputDir, "output", "o", "", "Output directory to write to.")
flags.BoolVar(&cmd.NoSync, "no-sync", false, "Disable file sync")
flags.IntVar(&cmd.Concurrency, "concurrency", cmd.Concurrency, "Number of concurrent backup goroutines.")
flags.StringVar(&cmd.Host, "host", "localhost:10101", "The address (host:port) of FeatureBase (HTTP).")
flags.StringVar(&cmd.Index, "index", "", "Index to backup, default backs up all indexes. ")
flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.")
flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.")
ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification)
return ccmd
}

View file

@ -25,6 +25,8 @@ The Restore command will take a backup archive and restore it to a new, clean cl
flags.StringVarP(&cmd.Path, "source", "s", "", "backup file; specify '-' to restore from stdin tar stream")
flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.")
flags.IntVar(&cmd.Concurrency, "concurrency", 1, "number of concurrent uploads")
flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.")
flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.")
ctl.SetTLSConfig(
flags, "",
&cmd.TLS.CertificatePath,

View file

@ -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 {

View file

@ -92,8 +92,10 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
byteData, err := ioutil.ReadAll(tr)
vprint.PanicOn(err)
br := bytes.NewReader(byteData)
err = r.client.ImportFieldKeys(context.Background(), uri, index, fieldName, false, br)
readerFunc := func() (io.Reader, error) {
return bytes.NewReader(byteData), nil
}
err = r.client.ImportFieldKeys(context.Background(), uri, index, fieldName, false, readerFunc)
if err != nil {
return err
}
@ -106,9 +108,11 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
}
byteData, err := ioutil.ReadAll(tr)
vprint.PanicOn(err)
readerFunc := func() (io.Reader, error) {
return bytes.NewReader(byteData), nil
}
br := bytes.NewReader(byteData)
err = r.client.ImportIndexKeys(context.Background(), uri, index, int(partition), false, br)
err = r.client.ImportIndexKeys(context.Background(), uri, index, int(partition), false, readerFunc)
if err != nil {
return err
}

View file

@ -10,11 +10,13 @@ import (
"io/ioutil"
"os"
"path/filepath"
"time"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/http"
fb_http "github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/server"
"github.com/molecula/featurebase/v2/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -37,6 +39,12 @@ type BackupCommand struct { // nolint: maligned
// Number of concurrent backup goroutines running at a time.
Concurrency int
// Amount of time after first failed request to continue retrying.
RetryPeriod time.Duration `json:"retry-period"`
// Host:port on which to listen for pprof.
Pprof string `json:"pprof"`
// Reusable client.
client pilosa.InternalClient
@ -51,11 +59,20 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand
return &BackupCommand{
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
Concurrency: 1,
RetryPeriod: time.Minute,
Pprof: "localhost:43809",
}
}
// Run executes the main program execution.
func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
logger := cmd.Logger()
close, err := startProfilingServer(cmd.Pprof, logger)
if err != nil {
return errors.Wrap(err, "starting profiling server")
}
defer close()
// Validate arguments.
if cmd.OutputDir == "" {
return fmt.Errorf("-o flag required")
@ -70,7 +87,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
}
// Create a client to the server.
client, err := commandClient(cmd)
client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
if err != nil {
return fmt.Errorf("creating client: %w", err)
}
@ -262,7 +279,7 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string,
logger := cmd.Logger()
logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig))
client := fb_http.NewInternalClientFromURI(&node.URI, fb_http.GetHTTPClient(cmd.tlsConfig), fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
rc, err := client.ShardReader(ctx, indexName, shard)
if err != nil {
return fmt.Errorf("fetching shard reader: %w", err)

View file

@ -2,6 +2,11 @@
package ctl
import (
"net"
"time"
gohttp "net/http"
"github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/logger"
"github.com/molecula/featurebase/v2/server"
@ -25,14 +30,22 @@ func SetTLSConfig(flags *pflag.FlagSet, prefix string, certificatePath *string,
flags.BoolVarP(enableClientVerification, prefix+"tls.enable-client-verification", "", false, "Enable TLS certificate client verification for incoming connections")
}
// default dial timeout is 30s for some reason which makes testing
// failures/retries really awkward. I don't think we need it that
// high, so I set it to 1s here... let's see what happens.
func clientOptions(client *gohttp.Client, dialer *net.Dialer) *gohttp.Client {
dialer.Timeout = time.Second * 1
return client
}
// commandClient returns a pilosa.InternalHTTPClient for the command
func commandClient(cmd CommandWithTLSSupport) (*http.InternalClient, error) {
func commandClient(cmd CommandWithTLSSupport, opts ...http.InternalClientOption) (*http.InternalClient, error) {
tls := cmd.TLSConfiguration()
tlsConfig, err := server.GetTLSConfig(&tls, cmd.Logger())
if err != nil {
return nil, errors.Wrap(err, "getting tls config")
}
client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig))
client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig, clientOptions), opts...)
if err != nil {
return nil, errors.Wrap(err, "getting internal client")
}

View file

@ -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)
}
}

View file

@ -5,18 +5,23 @@ import (
"context"
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"io"
"math"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/hashicorp/go-retryablehttp"
pilosa "github.com/molecula/featurebase/v2"
fb_http "github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/server"
"github.com/molecula/featurebase/v2/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -29,6 +34,13 @@ type RestoreCommand struct {
// Filepath to the backup file.
Path string
// Amount of time after first failed request to continue retrying.
RetryPeriod time.Duration `json:"retry-period"`
// Host:port on which to listen for pprof.
Pprof string `json:"pprof"`
// Reusable client.
client pilosa.InternalClient
@ -41,13 +53,20 @@ type RestoreCommand struct {
func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand {
return &RestoreCommand{
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
RetryPeriod: time.Second * 30,
Concurrency: 1,
Pprof: "localhost:43809",
}
}
// Run executes the restore.
func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
logger := cmd.Logger()
close, err := startProfilingServer(cmd.Pprof, logger)
if err != nil {
return errors.Wrap(err, "starting profiling server")
}
defer close()
// Validate arguments.
if cmd.Path == "" {
@ -62,7 +81,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
return fmt.Errorf("parsing tls config: %w", err)
}
// Create a client to the server.
client, err := commandClient(cmd)
client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod))
if err != nil {
return fmt.Errorf("creating client: %w", err)
}
@ -119,7 +138,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
if len(existingSchema) == 0 {
cmd.Logger().Printf("Load Schema")
url := primary.URI.Path("/schema")
var client http.Client
client := cmd.newClient()
_, err = client.Post(url, "application/json", f)
} else {
schema := &pilosa.Schema{}
@ -159,6 +178,34 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
return err
}
func retryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) {
if resp != nil && resp.StatusCode >= 400 { // we have some dumb status codes
return true, nil
}
return retryablehttp.DefaultRetryPolicy(ctx, resp, err)
}
// This logic is taken from featurebase/http/client.go If this logic
// is not the same as what's there, that could be a problem. Ideally
// all network calls from restore would go through the client and this
// would not longer be needed.
func (cmd *RestoreCommand) newClient() *retryablehttp.Client {
min := time.Millisecond * 100
// do some math to figure out how many attempts we need to get our
// total sleep time close to the period
attempts := math.Log2(float64(cmd.RetryPeriod)) - math.Log2(float64(min))
attempts += 0.3 // mmmm, fudge
if attempts < 1 {
attempts = 1
}
client := retryablehttp.NewClient()
client.RetryWaitMin = min
client.RetryMax = int(attempts)
client.CheckRetry = retryWith400
return client
}
func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error {
logger := cmd.Logger()
@ -174,7 +221,7 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology
logger.Printf("Load idalloc")
url := primary.URI.Path("/internal/idalloc/restore")
var client http.Client
client := cmd.newClient()
_, err = client.Post(url, "application/octet-stream", f)
return err
}
@ -244,14 +291,14 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er
defer f.Close()
url := node.URI.Path(fmt.Sprintf("/internal/restore/%v/%v", indexName, shard))
req, err := http.NewRequest("POST", url, f)
req, err := retryablehttp.NewRequest("POST", url, f)
if err != nil {
return err
}
req = req.WithContext(ctx)
req.Header.Set("Content-Type", "application/octet-stream")
var client http.Client
client := cmd.newClient()
resp, err := client.Do(req)
if err != nil {
return err
@ -319,13 +366,11 @@ func (cmd *RestoreCommand) restoreIndexTranslationFile(ctx context.Context, file
for _, node := range nodes {
if err := func() error {
f, err := os.Open(filename)
if err != nil {
return err
readerFunc := func() (io.Reader, error) {
return os.Open(filename) // gets used as an HTTP request body and closed by http library
}
defer f.Close()
return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, f)
return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, readerFunc)
}(); err != nil {
return err
}
@ -380,13 +425,11 @@ func (cmd *RestoreCommand) restoreFieldTranslationFile(ctx context.Context, node
for _, node := range nodes {
if err := func() error {
f, err := os.Open(filename)
if err != nil {
return err
readerFunc := func() (io.Reader, error) {
return os.Open(filename)
}
defer f.Close()
return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, f)
return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, readerFunc)
}(); err != nil {
return err
}

View file

@ -45,7 +45,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
// Etcd
// Etcd.Name used Config.Name for its value.
// Etcd.Dir defaults to a directory under the pilosa data directory.
flags.StringVar(&srv.Config.Etcd.Dir, "etcd.dir", srv.Config.Etcd.Dir, "Directory to store etcd data files. If not provided, a directory will be created under the main data-dir directory.")
// Etcd.ClusterName uses Cluster.Name for its value
flags.StringVar(&srv.Config.Etcd.LClientURL, "etcd.listen-client-address", srv.Config.Etcd.LClientURL, "Listen client address.")
flags.StringVar(&srv.Config.Etcd.AClientURL, "etcd.advertise-client-address", srv.Config.Etcd.AClientURL, "Advertise client address. If not provided, uses the listen client address.")
@ -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)
@ -117,5 +117,5 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.StringVar(&srv.Config.Auth.TokenURL, "auth.token-url", srv.Config.Auth.TokenURL, "Identity Provider's Token URL.")
flags.StringVar(&srv.Config.Auth.GroupEndpointURL, "auth.group-endpoint-url", srv.Config.Auth.GroupEndpointURL, "Identity Provider's Group endpoint URL.")
flags.StringVar(&srv.Config.Auth.ScopeURL, "auth.scope-url", srv.Config.Auth.ScopeURL, "Identity Provider's Scope URL.")
flags.StringVar(&srv.Config.Auth.PermissionsFile, "auth.permissions", srv.Config.Auth.PermissionsFile, "Permissions' file with group authorization.")
}

56
ctl/util.go Normal file
View file

@ -0,0 +1,56 @@
package ctl
import (
"context"
"net"
"net/http"
"net/http/pprof"
"runtime"
"time"
"github.com/felixge/fgprof"
"github.com/molecula/featurebase/v2/logger"
"github.com/pkg/errors"
)
// startProfilingServer starts a server which handles /debug/pprof and
// /debug/fgprof for use in utilities we might want to profile but
// wouldn't otherwise be running an http server. Caller should call
// the returned close function before exiting to release resources.
func startProfilingServer(addr string, logger logger.Logger) (close func() error, err error) {
if addr == "" {
return func() error { return nil }, nil
}
sm := http.NewServeMux()
sm.Handle("/debug/fgprof", fgprof.Handler())
sm.HandleFunc("/debug/pprof/", pprof.Index)
sm.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
sm.HandleFunc("/debug/pprof/profile", pprof.Profile)
sm.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
sm.HandleFunc("/debug/pprof/trace", pprof.Trace)
s := &http.Server{
Addr: addr,
Handler: sm,
}
runtime.SetBlockProfileRate(10000000) // 1 sample per 10 ms
runtime.SetMutexProfileFraction(100) // 1% sampling
ln, err := net.Listen("tcp", addr)
if err != nil {
return nil, err
}
go func() {
logger.Printf("Listening for /debug/pprof/ and /debug/fgprof on '%s'", addr)
logger.Printf("%v", s.Serve(ln))
}()
return func() error {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
defer cancel()
err := s.Shutdown(ctx)
if err != nil {
return errors.Wrap(err, "shutting down profiling server")
}
return s.Close()
}, nil
}

View file

@ -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,68 @@ 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)
n := atomic.AddInt64(&e.currentWorkers, 1)
if e.Holder != nil {
e.Holder.Stats.Gauge("worker_total", float64(n), 0)
}
go func() {
defer e.workersWG.Done()
e.worker(e.work)
n := atomic.AddInt64(&e.currentWorkers, -1)
if e.Holder != nil {
e.Holder.Stats.Gauge("worker_total", float64(n), 0)
}
}()
}
func (e *executor) Close() error {
e.workMu.Lock()
defer e.workMu.Unlock()
@ -154,6 +211,14 @@ func (e *executor) Close() error {
return nil
}
// InitStats initializes stats counters. Must be called after Holder set.
func (e *executor) InitStats() {
if e.Holder != nil {
e.Holder.Stats.Count("job_total", 0, 0)
e.Holder.Stats.Gauge("worker_total", float64(atomic.LoadInt64(&e.currentWorkers)), 0)
}
}
// Execute executes a PQL query.
func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) (QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.Execute")
@ -4518,7 +4583,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.
@ -4557,15 +4626,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()
@ -4597,6 +4672,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()
@ -4847,6 +4927,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
@ -5940,10 +6023,16 @@ 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)
e.Holder.Stats.Count("job_total", 1, 0)
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.

View file

@ -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
}

View file

@ -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.

3
go.mod
View file

@ -25,6 +25,7 @@ require (
github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00 // indirect
github.com/gorilla/handlers v1.3.0
github.com/gorilla/mux v1.7.0
github.com/hashicorp/go-retryablehttp v0.7.0
github.com/improbable-eng/grpc-web v0.13.0
github.com/lib/pq v1.8.0
github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b
@ -54,7 +55,7 @@ require (
golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d // indirect
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

5
go.sum
View file

@ -186,10 +186,15 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t
github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q=
github.com/hashicorp/consul/sdk v0.1.1/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8=
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
github.com/hashicorp/go-cleanhttp v0.5.1 h1:dH3aiDG9Jvb5r5+bYHsikaOUIpcM0xvgMXVoDkXMzJM=
github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
github.com/hashicorp/go-retryablehttp v0.7.0 h1:eu1EI/mbirUgP5C8hVsTNaGZreBDlYiwC1FZWkvQPQ4=
github.com/hashicorp/go-retryablehttp v0.7.0/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY=
github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=

View file

@ -8,18 +8,22 @@ import (
"fmt"
"io"
"io/ioutil"
"math"
"math/rand"
"net/http"
"net/url"
"os"
"path"
"sort"
"strconv"
"strings"
"time"
"github.com/hashicorp/go-retryablehttp"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/encoding/proto"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/logger"
pnet "github.com/molecula/featurebase/v2/net"
"github.com/molecula/featurebase/v2/topology"
"github.com/molecula/featurebase/v2/tracing"
@ -31,8 +35,11 @@ type InternalClient struct {
defaultURI *pnet.URI
serializer pilosa.Serializer
log logger.Logger
// The client to use for HTTP communication.
httpClient *http.Client
httpClient *http.Client
retryableClient *retryablehttp.Client
// the local node's API, used for operations that we can short-circuit that way
api *pilosa.API
}
@ -40,7 +47,7 @@ type InternalClient struct {
// NewInternalClient returns a new instance of InternalClient to connect to host.
// If api is non-nil, the client uses it for some same-host operations instead
// of going through http.
func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, error) {
func NewInternalClient(host string, remoteClient *http.Client, opts ...InternalClientOption) (*InternalClient, error) {
if host == "" {
return nil, pilosa.ErrHostRequired
}
@ -50,16 +57,75 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient,
return nil, errors.Wrap(err, "getting URI")
}
client := NewInternalClientFromURI(uri, remoteClient)
client := NewInternalClientFromURI(uri, remoteClient, opts...)
return client, nil
}
func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client) *InternalClient {
return &InternalClient{
type InternalClientOption func(c *InternalClient)
// WithClientRetryPeriod is the max amount of total time the client will
// retry failed requests using exponential backoff.
func WithClientRetryPeriod(period time.Duration) InternalClientOption {
min := time.Millisecond * 100
// do some math to figure out how many attempts we need to get our
// total sleep time close to the period
attempts := math.Log2(float64(period)) - math.Log2(float64(min))
attempts += 0.3 // mmmm, fudge
if attempts < 1 {
attempts = 1
}
fmt.Println("attempts: ", int(attempts))
return func(c *InternalClient) {
rc := retryablehttp.NewClient()
rc.HTTPClient = c.httpClient
rc.RetryWaitMin = min
rc.RetryMax = int(attempts)
rc.CheckRetry = retryWith400Policy
c.retryableClient = rc
}
}
func WithClientLogger(log logger.Logger) InternalClientOption {
return func(c *InternalClient) {
c.log = log
}
}
func noRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, error) {
return false, nil
}
// retryWith400Policy wraps retryablehttp's default retry policy to
// also retry on 4XX errors which *should* be client errors and
// therefore useless to retry, but we have some incorrect status codes.
// TODO: fix the incorrect status codes so we can get rid of this.
func retryWith400Policy(ctx context.Context, resp *http.Response, err error) (bool, error) {
if resp != nil && resp.StatusCode >= 400 {
return true, nil
}
return retryablehttp.DefaultRetryPolicy(ctx, resp, err)
}
func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, opts ...InternalClientOption) *InternalClient {
ic := &InternalClient{
defaultURI: defaultURI,
serializer: proto.Serializer{},
httpClient: remoteClient,
log: logger.NewStandardLogger(os.Stderr),
}
for _, opt := range opts {
opt(ic)
}
if ic.retryableClient == nil {
rc := retryablehttp.NewClient()
rc.HTTPClient = ic.httpClient
rc.CheckRetry = noRetryPolicy
ic.retryableClient = rc
}
return ic
}
// MaxShardByIndex returns the number of shards on a server by index.
@ -1717,19 +1783,36 @@ func giveRawResponse(b bool) executeRequestOption {
}
}
type nopCloser struct {
*bytes.Reader
}
func (n nopCloser) Close() error {
return nil
}
// executeRequest executes the given request and checks the Response. For
// responses with non-2XX status, the body is read and closed, and an error is
// returned. If the error is nil, the caller must ensure that the response body
// is closed.
func (c *InternalClient) executeRequest(req *http.Request, opts ...executeRequestOption) (*http.Response, error) {
return c.executeRetryableRequest(&retryablehttp.Request{Request: req}, opts...)
}
func (c *InternalClient) executeRetryableRequest(req *retryablehttp.Request, opts ...executeRequestOption) (*http.Response, error) {
tracing.GlobalTracer.InjectHTTPHeaders(req.Request)
req.Close = false
eo := &executeOpts{}
for _, opt := range opts {
opt(eo)
}
tracing.GlobalTracer.InjectHTTPHeaders(req)
req.Close = false
resp, err := c.httpClient.Do(req)
resp, err := c.retryableClient.Do(req)
return c.handleResponse(req.Request, eo, resp, err)
}
func (c *InternalClient) handleResponse(req *http.Request, eo *executeOpts, resp *http.Response, err error) (*http.Response, error) {
if err != nil {
if resp != nil {
resp.Body.Close()
@ -2009,7 +2092,7 @@ func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context,
return resp.Body, nil
}
func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error {
func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportIndexKeys")
defer span.Finish()
@ -2026,14 +2109,14 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind
url := fmt.Sprintf("%s/internal/translate/index/%s/%d", uri, index, partitionID)
// Generate HTTP request.
httpReq, err := http.NewRequest("POST", url, rddbdata)
httpReq, err := retryablehttp.NewRequest("POST", url, readerFunc)
if err != nil {
return errors.Wrap(err, "creating request")
}
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
resp, err := c.executeRequest(httpReq.WithContext(ctx))
resp, err := c.executeRetryableRequest(httpReq.WithContext(ctx))
if err != nil {
return err
}
@ -2041,7 +2124,7 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind
return nil
}
func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error {
func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportFieldKeys")
defer span.Finish()
@ -2058,14 +2141,14 @@ func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, ind
url := fmt.Sprintf("%s/internal/translate/field/%s/%s", uri, index, field)
// Generate HTTP request.
httpReq, err := http.NewRequest("POST", url, rddbdata)
httpReq, err := retryablehttp.NewRequest("POST", url, readerFunc)
if err != nil {
return errors.Wrap(err, "creating request")
}
httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
resp, err := c.executeRequest(httpReq.WithContext(ctx))
resp, err := c.executeRetryableRequest(httpReq.WithContext(ctx))
if err != nil {
return err
}

View file

@ -441,6 +441,9 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/internal/idalloc/data", handler.handleIDAllocData).Methods("GET").Name("IDAllocData")
router.HandleFunc("/internal/restore/{index}/{shardID}", handler.handlePostRestore).Methods("POST").Name("Restore")
router.HandleFunc("/internal/debug/rbf", handler.handleGetInternalDebugRBFJSON).Methods("GET").Name("GetInternalDebugRBFJSON")
// endpoints for collecting cpu profiles from a chosen begin point to
// when the client wants to stop. Used for profiling imports that
// could be long or short.
@ -2064,6 +2067,18 @@ func validateProtobufHeader(r *http.Request) (error string, code int) {
return
}
// handleGetInternalDebugRBFJSON handles /internal/debug/rbf requests.
func (h *Handler) handleGetInternalDebugRBFJSON(w http.ResponseWriter, r *http.Request) {
buf, err := json.MarshalIndent(h.api.RBFDebugInfo(), "", " ")
if err != nil {
http.Error(w, "marshal json: "+err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
w.Write(buf)
}
// handleGetMetricsJSON handles /metrics.json requests, translating text metrics results to more consumable JSON.
func (h *Handler) handleGetMetricsJSON(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
@ -2578,14 +2593,17 @@ func (s queryValidationSpec) validate(query url.Values) error {
return nil
}
func GetHTTPClient(t *tls.Config) *http.Client {
type ClientOption func(client *http.Client, dialer *net.Dialer) *http.Client
func GetHTTPClient(t *tls.Config, opts ...ClientOption) *http.Client {
dialer := &net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}
transport := &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}).DialContext,
Proxy: http.ProxyFromEnvironment,
DialContext: dialer.DialContext,
MaxIdleConns: 1000,
MaxIdleConnsPerHost: 200,
IdleConnTimeout: 90 * time.Second,
@ -2595,7 +2613,12 @@ func GetHTTPClient(t *tls.Config) *http.Client {
if t != nil {
transport.TLSClientConfig = t
}
return &http.Client{Transport: transport}
client := &http.Client{Transport: transport}
for _, opt := range opts {
client = opt(client, dialer)
}
return client
}
// handlePostImportAtomicRecord handles /import-atomic-record requests
@ -3348,7 +3371,7 @@ func (h *Handler) handlePostRestore(w http.ResponseWriter, r *http.Request) {
//validate shard for this node
err = h.api.RestoreShard(ctx, indexName, shard, r.Body)
if err != nil {
http.Error(w, fmt.Sprintf("failed to restore shared %v %v err:%v", indexName, shard, err), http.StatusBadRequest)
http.Error(w, fmt.Sprintf("failed to restore shard %v %v err:%v", indexName, shard, err), http.StatusBadRequest)
return
}

View file

@ -380,4 +380,5 @@ log-path = "/var/log/molecula/featurebase.log"
# authorize-url = ""
# token-url = ""
# group-endpoint-url = ""
# scope-url = ""
# scope-url = ""
# permissions = ""

View file

@ -3,6 +3,8 @@ package clustertest
import (
"context"
"fmt"
"net/http"
"os"
"os/exec"
"testing"
@ -30,56 +32,53 @@ func TestClusterStuff(t *testing.T) {
t.Fatalf("getting client: %v", err)
}
t.Run("long pause", func(t *testing.T) {
err := cli1.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
err = cli1.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100})
if err != nil {
t.Fatalf("creating field: %v", err)
}
if err := cli1.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{}); err != nil {
t.Fatalf("creating index: %v", err)
}
if err := cli1.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100}); err != nil {
t.Fatalf("creating field: %v", err)
}
req := &pilosa.ImportRequest{
Index: "testidx",
Field: "testf",
}
req.ColumnIDs = make([]uint64, 10)
req.RowIDs = make([]uint64, 10)
req := &pilosa.ImportRequest{
Index: "testidx",
Field: "testf",
}
req.ColumnIDs = make([]uint64, 10)
req.RowIDs = make([]uint64, 10)
for i := 0; i < 1000; i++ {
req.RowIDs[i%10] = 0
req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10)
req.Shard = uint64(i / 10)
if i%10 == 9 {
err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{})
if err != nil {
t.Fatalf("importing: %v", err)
}
}
}
// Check query results from each node.
for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
for i := 0; i < 1000; i++ {
req.RowIDs[i%10] = 0
req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10)
req.Shard = uint64(i / 10)
if i%10 == 9 {
err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{})
if err != nil {
t.Fatalf("count querying pilosa%d: %v", i, err)
}
if r.Results[0].(uint64) != 1000 {
t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64))
t.Fatalf("importing: %v", err)
}
}
}
// Check query results from each node.
for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} {
r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})
if err != nil {
t.Fatalf("count querying pilosa%d: %v", i, err)
}
if r.Results[0].(uint64) != 1000 {
t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64))
}
}
t.Run("long pause", func(t *testing.T) {
pcmd := exec.Command("/pumba", "pause", "clustertests_pilosa3_1", "--duration", "10s")
pcmd.Stdout = os.Stdout
pcmd.Stderr = os.Stderr
t.Log("pausing pilosa3 for 10s")
err = pcmd.Start()
if err != nil {
if err := pcmd.Start(); err != nil {
t.Fatalf("starting pumba command: %v", err)
}
err = pcmd.Wait()
if err != nil {
if err := pcmd.Wait(); err != nil {
t.Fatalf("waiting on pumba pause cmd: %v", err)
}
@ -98,6 +97,89 @@ func TestClusterStuff(t *testing.T) {
}
}
})
t.Run("backup", func(t *testing.T) {
// do backup with node 1 down, but restart it after a few seconds
if err := sendCmd("docker", "stop", "clustertests_pilosa1_1"); err != nil {
t.Fatalf("sending stop command: %v", err)
}
var backupCmd *exec.Cmd
tmpdir := t.TempDir()
if backupCmd, err = startCmd(
"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest")); err != nil {
t.Fatalf("sending backup command: %v", err)
}
time.Sleep(time.Second * 5)
if err = sendCmd("docker", "start", "clustertests_pilosa1_1"); err != nil {
t.Fatalf("sending start command: %v", err)
}
if err = backupCmd.Wait(); err != nil {
t.Fatalf("waiting on backup to finish: %v", err)
}
fmt.Println("STARTING RESTORE")
client := http.Client{}
if req, err := http.NewRequest(http.MethodDelete, "http://pilosa1:10101/index/testidx", nil); err != nil {
t.Fatalf("getting req: %v", err)
} else if resp, err := client.Do(req); err != nil {
t.Fatalf("doing request: %v", err)
} else if resp.StatusCode >= 400 {
t.Fatalf("bad response: %v", resp)
}
var restoreCmd *exec.Cmd
if restoreCmd, err = startCmd("featurebase", "restore", "-s", tmpdir+"/backuptest", "--host", "pilosa1:10101"); err != nil {
t.Fatalf("starting restore: %v", err)
}
time.Sleep(time.Millisecond * 50)
if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil {
t.Fatalf("sending stop command: %v", err)
}
time.Sleep(time.Second * 10)
if err = sendCmd("docker", "start", "clustertests_pilosa2_1"); err != nil {
t.Fatalf("sending stop command: %v", err)
}
if err := restoreCmd.Wait(); err != nil {
t.Fatalf("restore failed: %v", err)
}
// now do backup with all nodes down and too short a timeout
// so it fails. Has be to be all 3 because the cluster has
// replicas=3 and the backup command will retry on replicas.
if backupCmd, err = startCmd(
"featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=200ms"); err != nil {
t.Fatalf("sending second backup command: %v", err)
}
time.Sleep(time.Millisecond * 10) // want the backup to get started, then fail
if err = sendCmd("docker", "stop", "clustertests_pilosa1_1"); err != nil {
t.Fatalf("sending stop command: %v", err)
}
if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil {
t.Fatalf("sending stop command: %v", err)
}
if err = sendCmd("docker", "stop", "clustertests_pilosa3_1"); err != nil {
t.Fatalf("sending stop command: %v", err)
}
time.Sleep(time.Second * 5)
if err = sendCmd("docker", "start", "clustertests_pilosa1_1"); err != nil {
t.Fatalf("sending start command: %v", err)
}
if err = sendCmd("docker", "start", "clustertests_pilosa2_1"); err != nil {
t.Fatalf("sending start command: %v", err)
}
if err = sendCmd("docker", "start", "clustertests_pilosa3_1"); err != nil {
t.Fatalf("sending start command: %v", err)
}
if err = backupCmd.Wait(); err == nil {
t.Fatal("backup command should have errored but didn't")
}
})
}
func waitForStatus(t *testing.T, stator func(context.Context) (string, error), status string, n int, sleep time.Duration) {

View file

@ -9,6 +9,7 @@ services:
- "33455:10101"
environment:
- PILOSA_NAME=pilosa1
- PILOSA_ETCD_DIR=/root/.etcd
- PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
- PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa1:10201
- PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
@ -28,6 +29,7 @@ services:
- "33456:10101"
environment:
- PILOSA_NAME=pilosa2
- PILOSA_ETCD_DIR=/root/.etcd
- PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
- PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa2:10201
- PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
@ -47,6 +49,7 @@ services:
- "33457:10101"
environment:
- PILOSA_NAME=pilosa3
- PILOSA_ETCD_DIR=/root/.etcd
- PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
- PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa3:10201
- PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301

View file

@ -23,11 +23,16 @@ import (
"github.com/pkg/errors"
)
func sendCmd(cmd string, args ...string) error {
func startCmd(cmd string, args ...string) (*exec.Cmd, error) {
pcmd := exec.Command(cmd, args...)
pcmd.Stdout = os.Stdout
pcmd.Stderr = os.Stderr
err := pcmd.Start()
return pcmd, err
}
func sendCmd(cmd string, args ...string) error {
pcmd, err := startCmd(cmd, args...)
if err != nil {
return errors.Wrap(err, "starting cmd")
}

View file

@ -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 {

View file

@ -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
}

View file

@ -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) {

411
rbf/db.go
View file

@ -7,10 +7,13 @@ import (
"io"
"os"
"path/filepath"
"runtime/debug"
"sort"
"sync"
"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 +30,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 +51,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 +81,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 +157,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 +186,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 +199,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 +297,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 +606,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{
@ -467,9 +639,15 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
pageMap: db.pageMap,
walPageN: db.walPageN,
writable: writable,
stack: debug.Stack(), // DEBUG
DeleteEmptyContainer: true,
}
defer func() {
if err != nil {
tx.rollback(true)
}
}()
if writable {
tx.dirtyPages = make(map[uint32][]byte)
@ -480,10 +658,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 +673,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 +794,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.
@ -583,6 +818,20 @@ func (db *DB) getCursor(tx *Tx) *Cursor {
return c
}
func (db *DB) DebugInfo() *DebugInfo {
info := &DebugInfo{Path: db.Path}
for tx := range db.txs {
info.Txs = append(info.Txs, tx.DebugInfo())
}
sort.Slice(info.Txs, func(i, j int) bool { return info.Txs[i].Ptr < info.Txs[j].Ptr })
return info
}
type DebugInfo struct {
Path string `json:"path"`
Txs []*TxDebugInfo `json:"txs"`
}
// Shared pool for in-memory database pages.
// These are used before being flushed to disk.
var pagePool = &sync.Pool{

View file

@ -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,134 @@ func TestDB_MultiTx(t *testing.T) {
}
}
func TestDB_DebugInfo(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer tx.Rollback()
info := db.DebugInfo()
if got, want := info.Path, db.Path; got != want {
t.Fatalf("Path=%q, want %q", got, want)
} else if got, want := len(info.Txs), 1; got != want {
t.Fatalf("len(Txs)=%d, want %d", got, want)
}
}
// 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")

View file

@ -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)

View file

@ -65,6 +65,9 @@ type Tx struct {
// manages to trigger a *deallocation* (which I don't think should be
// happening), we'll process that one after the current list is processed.
pendingFreelistAdds []uint32
// DEBUG
stack []byte
}
func (tx *Tx) DBPath() string {
@ -109,24 +112,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 +151,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 +744,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 +854,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 +870,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
}
@ -2011,6 +2045,20 @@ func (tx *Tx) GetSortedFieldViewList() (fvs []txkey.FieldView, _ error) {
return
}
func (tx *Tx) DebugInfo() *TxDebugInfo {
return &TxDebugInfo{
Ptr: fmt.Sprintf("%p", tx),
Writable: tx.writable,
Stack: string(tx.stack),
}
}
type TxDebugInfo struct {
Ptr string `json:"ptr"`
Writable bool `json:"writable"`
Stack string `json:"stack,omitempty"`
}
// SnapshotReader returns a reader that provides a snapshot for the current database state.
func (tx *Tx) SnapshotReader() (io.Reader, error) {
if tx.db == nil {

View file

@ -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)
}
}

View file

@ -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
@ -507,6 +506,10 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.holder.schemator = s.schemator
s.holder.sharder = s.sharder
s.holder.serializer = s.serializer
// Initial stats must be invoked after the executor obtains reference to the holder.
s.executor.InitStats()
return s, nil
}

View file

@ -4,15 +4,18 @@ package server
import (
"context"
"fmt"
"io"
"log"
"net"
"net/url"
"os"
"path/filepath"
"runtime"
"strconv"
"strings"
"time"
"github.com/molecula/featurebase/v2/auth"
"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"
@ -200,9 +203,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.
@ -231,7 +234,7 @@ type Config struct {
SchemaDetailsOn bool `toml:"schema-details-on"`
// Enable AuthZ/AuthN
Auth auth.Auth `toml:"auth"`
Auth authz.Auth `toml:"auth"`
}
// Namespace returns the namespace to use based on the Future flag.
@ -596,9 +599,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,
@ -609,7 +612,6 @@ func (c *Config) ValidateAuth() ([]error, error) {
"ScopeURL": c.Auth.ScopeURL,
}
errors := make([]error, 0)
for name, value := range authConfig {
if value == "" {
errors = append(errors, fmt.Errorf("empty string for auth config %s", name))
@ -624,17 +626,95 @@ func (c *Config) ValidateAuth() ([]error, error) {
}
}
}
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"))
}
}

View file

@ -9,7 +9,7 @@ import (
"strings"
"testing"
"github.com/molecula/featurebase/v2/auth"
"github.com/molecula/featurebase/v2/authz"
)
type addrs struct{ bind, advertise string }
@ -284,14 +284,14 @@ func TestConfig_validateAuth(t *testing.T) {
validTestURL := "https://url.com/"
validClientID := "clientid"
validClientSecret := "clientSecret"
notValidURL := "not-a-url"
invalidURL := "not-a-url"
emptyString := ""
enable := true
disable := false
tests := []struct {
expErrs []string
input auth.Auth
input authz.Auth
}{
{
@ -304,7 +304,7 @@ func TestConfig_validateAuth(t *testing.T) {
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
authz.Auth{
Enable: enable,
ClientId: emptyString,
ClientSecret: emptyString,
@ -314,130 +314,25 @@ func TestConfig_validateAuth(t *testing.T) {
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: emptyString,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: emptyString,
ClientSecret: validClientSecret,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: validTestURL,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: validTestURL,
TokenURL: validTestURL,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some strings are set to invalid URL
[]string{
errorMesgURL,
},
auth.Auth{
authz.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: notValidURL,
AuthorizeURL: invalidURL,
TokenURL: validTestURL,
GroupEndpointURL: validTestURL,
ScopeURL: validTestURL,
},
},
{
// Auth enabled, some strings are set to invalid URL
[]string{
errorMesgURL,
errorMesgURL,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: validTestURL,
TokenURL: notValidURL,
GroupEndpointURL: notValidURL,
ScopeURL: validTestURL,
},
},
{
// Auth enabled, all configs are set properly
[]string{},
auth.Auth{
authz.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
@ -450,7 +345,7 @@ func TestConfig_validateAuth(t *testing.T) {
{
// Auth disabled, all configs are set to empty string
[]string{},
auth.Auth{
authz.Auth{
Enable: disable,
ClientId: emptyString,
ClientSecret: emptyString,
@ -467,9 +362,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")
}
}
@ -487,3 +382,113 @@ func TestConfig_validateAuth(t *testing.T) {
})
}
}
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())
}
}
})
}
}

View file

@ -29,6 +29,7 @@ import (
"golang.org/x/sync/errgroup"
pilosa "github.com/molecula/featurebase/v2"
"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"
@ -224,6 +225,18 @@ func (m *Command) Start() (err 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
}
}
// Initialize server.
@ -482,7 +495,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),

View file

@ -17,7 +17,7 @@ import (
"testing"
"time"
"github.com/molecula/featurebase/v2"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/disco"
"github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/pql"
@ -26,6 +26,7 @@ import (
"github.com/molecula/featurebase/v2/test"
"github.com/molecula/featurebase/v2/testhook"
"github.com/pkg/errors"
"github.com/stretchr/testify/require"
"golang.org/x/sync/errgroup"
)
@ -504,6 +505,30 @@ func TestClusteringNodesReplica1(t *testing.T) {
t.Fatalf("starting cluster: %v", err)
}
indexName := "idx"
fieldName := "fld"
// Create the schema.
if _, err := cluster.GetPrimary().API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
t.Fatalf("creating index: %v", err)
}
if _, err := cluster.GetPrimary().API.CreateField(context.Background(), indexName, fieldName); err != nil {
t.Fatalf("creating field: %v", err)
}
// Set some columns across shards to ensure that the Row query will require
// data from all nodes.
data := []string{}
for rowID := 1; rowID < 2; rowID++ {
for columnID := 1; columnID < 10; columnID++ {
data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, columnID*pilosa.ShardWidth, fieldName, rowID))
}
}
if _, err := cluster.GetPrimary().Query(t, indexName, "", strings.Join(data, "")); err != nil {
t.Fatalf("setting columns: %v", err)
}
// Shut down a node.
if err := cluster.GetNonPrimary().Command.Close(); err != nil {
t.Fatalf("closing third node: %v", err)
}
@ -513,7 +538,12 @@ func TestClusteringNodesReplica1(t *testing.T) {
}
// confirm that cluster stops accepting queries after one node closes
if _, err := cluster.GetPrimary().API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
qry := &pilosa.QueryRequest{
Index: "idx",
Query: fmt.Sprintf("Row(%s=1)", fieldName),
}
if _, err := cluster.GetPrimary().API.Query(context.Background(), qry); !strings.Contains(err.Error(), "shard unavailable") {
t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
}
}
@ -540,8 +570,34 @@ func TestClusteringNodesReplica2(t *testing.T) {
}
defer cluster.Close()
indexName := "idx"
fieldName := "fld"
coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries()
// Create the schema.
if _, err := coord.API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
t.Fatalf("creating index: %v", err)
}
if _, err := coord.API.CreateField(context.Background(), indexName, fieldName); err != nil {
t.Fatalf("creating field: %v", err)
}
// Set some columns across shards to ensure that the Row query will require
// data from all nodes.
data := []string{}
cols := []uint64{}
for rowID := 1; rowID < 2; rowID++ {
for columnID := 1; columnID < 30; columnID++ {
col := uint64(columnID * pilosa.ShardWidth)
cols = append(cols, col)
data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, col, fieldName, rowID))
}
}
if _, err := coord.Query(t, indexName, "", strings.Join(data, "")); err != nil {
t.Fatalf("setting columns: %v", err)
}
if err := others[0].Close(); err != nil {
t.Fatalf("closing third node: %v", err)
}
@ -569,8 +625,30 @@ func TestClusteringNodesReplica2(t *testing.T) {
t.Fatalf("after closing second server: %v", err)
}
if _, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
qry := &pilosa.QueryRequest{
Index: "idx",
Query: fmt.Sprintf("Row(%s=1)", fieldName),
}
// Because we no longer block queries when the cluster is in state DOWN,
// there are cases where a DOWN cluster can still respond to a query. In
// that case, we want the test to pass. But if the unavailable node(s) cause
// the query to result in an error, we check that it's the error we expect.
resp, err := coord.API.Query(context.Background(), qry)
if err != nil {
if !strings.Contains(err.Error(), "shard unavailable") {
t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
}
} else {
if len(resp.Results) == 0 {
t.Fatal("got no results")
}
row, ok := resp.Results[0].(*pilosa.Row)
if !ok {
t.Fatalf("expected a *pilosa.Row, but got %T", resp.Results[0])
}
require.Equal(t, row.Columns(), cols)
}
}

View file

@ -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

View file

@ -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
}