diff --git a/.gitlab/.gitlab-ci.yml b/.gitlab/.gitlab-ci.yml index 6b9de74c0..8919322a9 100644 --- a/.gitlab/.gitlab-ci.yml +++ b/.gitlab/.gitlab-ci.yml @@ -184,7 +184,7 @@ build for linux arm64: paths: - featurebase_linux_arm64 - roaring-migrate_linux_arm64 - + build for darwin amd64: stage: build image: golang:$GOVERSION @@ -400,7 +400,7 @@ smoke test: junit: report.xml -gauntlet: +samsung-gauntlet: stage: gauntlet timeout: 4h image: registry.gitlab.com/gitlab-org/cloud-deploy/aws-base:latest @@ -441,8 +441,8 @@ gauntlet: - chmod 700 /root/.ssh - '[[ -f /.dockerenv ]] && echo -e "Host *\n\tStrictHostKeyChecking no\n\n" > ~/.ssh/config' - apt update && apt -y install jq wget - - wget -q https://go.dev/dl/go1.17.5.linux-amd64.tar.gz - - tar -C /usr/local -xzf go1.17.5.linux-amd64.tar.gz + - wget -q https://go.dev/dl/go1.17.8.linux-amd64.tar.gz + - tar -C /usr/local -xzf go1.17.8.linux-amd64.tar.gz - export PATH=$PATH:/usr/local/go/bin - TF_VAR_cluster_prefix="gauntlet-$(openssl rand -base64 12 | tr -d /=+ | cut -c -16)" - echo "Cluster Prefix --> $TF_VAR_cluster_prefix" @@ -455,7 +455,65 @@ gauntlet: - ./qa/scripts/teardownSamsungGauntlet.sh || true # leaving dangling resources is better than dangling ASG instances that can't be terminated - export INSTANCE_ID=$(cat instance_id) - aws autoscaling set-instance-protection --instance-ids "$INSTANCE_ID" --auto-scaling-group-name $ASG_NAME --no-protected-from-scale-in --profile $INFRA_PROFILE - + +backup-restore-gauntlet: + stage: gauntlet + timeout: 4h + image: registry.gitlab.com/gitlab-org/cloud-deploy/aws-base:latest + variables: + FBCI_PROFILE: "service-terraform" + INFRA_PROFILE: "service-gitlab" + AWS_SSH_PRIVATE_KEY: $AWS_FBCI_SSH_KEY + AWS_ACCESS_KEY_ID: $AWS_FBCI_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY: $AWS_FBCI_SECRET_ACCESS_KEY + ASG_NAME: "gitlab-runners" + TF_VAR_cluster_prefix: "" + rules: + - if: '$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH && ($CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web")' + tags: + - aws + - docker + - fbsmoke + before_script: + - apt-get update && apt-get install -y gnupg software-properties-common curl git + - curl -fsSL https://apt.releases.hashicorp.com/gpg | apt-key add - + - apt-add-repository "deb [arch=amd64] https://apt.releases.hashicorp.com $(lsb_release -cs) main" + - apt-get update && apt-get install terraform + - aws configure set aws_access_key_id $AWS_FBCI_ACCESS_KEY_ID --profile $FBCI_PROFILE + - aws configure set aws_secret_access_key $AWS_FBCI_SECRET_ACCESS_KEY --profile $FBCI_PROFILE + - aws configure set region "us-east-2" --profile $FBCI_PROFILE + - aws configure set aws_profile $FBCI_PROFILE + - aws configure set aws_access_key_id $AWS_INFRA_ACCESS_KEY_ID --profile $INFRA_PROFILE + - aws configure set aws_secret_access_key $AWS_INFRA_SECRET_ACCESS_KEY --profile $INFRA_PROFILE + - aws configure set region "us-east-2" --profile $INFRA_PROFILE + - echo $AWS_FBCI_SSH_KEY > gitlab-featurebase-ci.pem + - chmod 400 gitlab-featurebase-ci.pem + - 'which ssh-agent || ( apt-get update -y && apt-get install openssh-client -y )' + - eval $(ssh-agent -s) + - mkdir -p ~/.ssh + - echo $AWS_FBCI_SSH_KEY > /root/.ssh/gitlab-featurebase-ci.pem + - chmod 400 /root/.ssh/gitlab-featurebase-ci.pem + - echo "$AWS_FBCI_SSH_KEY" | ssh-add - + - chmod 700 /root/.ssh + - '[[ -f /.dockerenv ]] && echo -e "Host *\n\tStrictHostKeyChecking no\n\n" > ~/.ssh/config' + - apt update && apt -y install jq wget + - wget -q https://go.dev/dl/go1.17.8.linux-amd64.tar.gz + - tar -C /usr/local -xzf go1.17.8.linux-amd64.tar.gz + - export PATH=$PATH:/usr/local/go/bin + - TF_VAR_cluster_prefix="gauntlet-$(openssl rand -base64 12 | tr -d /=+ | cut -c -16)" + - echo "Cluster Prefix --> $TF_VAR_cluster_prefix" + - export INSTANCE_ID=$(curl --silent --fail "http://169.254.169.254/latest/meta-data/instance-id" | tee instance_id) + - aws autoscaling set-instance-protection --instance-ids "$INSTANCE_ID" --auto-scaling-group-name $ASG_NAME --protected-from-scale-in --profile $INFRA_PROFILE + script: + - ./qa/scripts/setupBackupRestoreGauntlet.sh + - ./qa/scripts/testBackupRestoreGauntlet.sh + after_script: + - ./qa/scripts/teardownBackupRestoreGauntlet.sh || true # leaving dangling resources is better than dangling ASG instances that can't be terminated + - export INSTANCE_ID=$(cat instance_id) + - aws autoscaling set-instance-protection --instance-ids "$INSTANCE_ID" --auto-scaling-group-name $ASG_NAME --no-protected-from-scale-in --profile $INFRA_PROFILE + needs: + - build for linux arm64 + s3 dump: stage: post build variables: @@ -547,8 +605,8 @@ perf_delete: - chmod 700 /root/.ssh - '[[ -f /.dockerenv ]] && echo -e "Host *\n\tStrictHostKeyChecking no\n\n" > ~/.ssh/config' - apt update && apt -y install jq wget - - wget -q https://go.dev/dl/go1.17.5.linux-amd64.tar.gz - - tar -C /usr/local -xzf go1.17.5.linux-amd64.tar.gz + - wget -q https://go.dev/dl/go1.17.8.linux-amd64.tar.gz + - tar -C /usr/local -xzf go1.17.8.linux-amd64.tar.gz - export PATH=$PATH:/usr/local/go/bin - TF_VAR_cluster_prefix="delete-$(openssl rand -base64 12 | tr -d /=+ | cut -c -16)" - echo "Cluster Prefix --> $TF_VAR_cluster_prefix" diff --git a/README.md b/README.md index eba94455b..bdec85672 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,9 @@ # FeatureBase, a distributed bitmap index +[![Coverage](https://sonarcloud.io/api/project_badges/measure?project=molecula_featurebase&metric=coverage&token=8e09e593b40570b544ed7defb47018add4eb9e7b)](https://sonarcloud.io/summary/new_code?id=molecula_featurebase) +[![SecurityRating](https://sonarcloud.io/api/project_badges/measure?project=molecula_featurebase&metric=security_rating&token=8e09e593b40570b544ed7defb47018add4eb9e7b)](https://sonarcloud.io/summary/new_code?id=molecula_featurebase) +[![Quality Gate Status](https://sonarcloud.io/api/project_badges/measure?project=molecula_featurebase&metric=alert_status&token=8e09e593b40570b544ed7defb47018add4eb9e7b)](https://sonarcloud.io/summary/new_code?id=molecula_featurebase) + See our [internal documentation](https://internal-docs.molecula.cloud), which includes all [external documentation](https://docs.molecula.cloud), plus many internal-only pages, listed under the "Internal" heading in the main navigation bar. Follow along with the [Sample Project](https://internal-docs.molecula.cloud/tutorials/getting-started) to get a better understanding of FeatureBase's capabilities. diff --git a/api.go b/api.go index 34155ee32..f0417c010 100644 --- a/api.go +++ b/api.go @@ -861,16 +861,31 @@ func (api *API) TranslateData(ctx context.Context, indexName string, partition i return nil, newNotFoundError(ErrIndexNotFound, indexName) } + // Find the node that can service the request. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) nodes := snap.PartitionNodes(partition) - if nodes[0].ID != api.server.NodeID() { + var upNode *topology.Node + for _, node := range nodes { + if node.State == disco.NodeStateStarted { + upNode = node + break + } + } + + // If there is no upNode, then we can't service the request. + if upNode == nil { + return nil, fmt.Errorf("can't get translate data, no nodes available for partition %d", partition) + } + + // If we're not the upNode, we need to redirect to it. + if upNode.ID != api.server.NodeID() { return nil, RedirectError{ - HostPort: nodes[0].URI.HostPort(), + HostPort: upNode.URI.HostPort(), error: fmt.Sprintf("can't translate data, this node(%s) does not partition %d", api.server.uri, partition), } } - // Retrieve translatestore from holder. + // We are the upNode! store := idx.TranslateStore(partition) if store == nil { return nil, ErrTranslateStoreNotFound @@ -1919,7 +1934,7 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha // We delete from the existence field unconditionally and other fields // if we know they exist. if op.OpType == ingest.OpDelete { - err = clearExistenceColumns(qcx, index, op.ClearRecordIDs, shard) + err = clearExistenceColumns(tx, index, op.ClearRecordIDs, shard) if err != nil { return fmt.Errorf("clearing existence columns: %w", err) } @@ -1997,21 +2012,20 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard ui return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options) } -func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error { +func clearExistenceColumns(tx Tx, index *Index, columnIDs []uint64, shard uint64) error { ef := index.existenceField() if ef == nil { return nil } - - existenceRowIDs := make([]uint64, len(columnIDs)) - // If we don't gratuitously hand-duplicate things in field.Import, - // the fact that fragment.bulkImport rewrites its row and column - // lists can burn us if we don't make a copy before doing the - // existence field write. - columnCopy := make([]uint64, len(columnIDs)) - copy(columnCopy, columnIDs) - options := ImportOptions{Clear: true} - return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options) + v := ef.view("standard") + if v == nil { + return nil + } + f := v.Fragment(shard) + if f == nil { + return nil + } + return f.ClearRecords(tx, columnIDs) } // ShardDistribution returns an object representing the distribution of shards diff --git a/api_test.go b/api_test.go index 0ebf16b5f..73792bf8d 100644 --- a/api_test.go +++ b/api_test.go @@ -1475,7 +1475,6 @@ func makeUser(t *testing.T, groups []authn.Group, name, secret string) *authn.Us if err != nil { t.Fatalf("signing string %v", err) } - validToken = "Bearer " + validToken return &authn.UserInfo{ UserID: "fake" + name, @@ -1520,7 +1519,7 @@ admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` return } g := []authn.Group{} - switch token[0] { + switch strings.TrimPrefix(token[0], "Bearer ") { case adminUser.Token: g = adminUser.Groups case readUser.Token: diff --git a/catcher.go b/catcher.go index 0a75ac9f7..a1f128d65 100644 --- a/catcher.go +++ b/catcher.go @@ -26,11 +26,6 @@ func init() { var _ Tx = (*catcherTx)(nil) -func (c *catcherTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) { - c.b.RemoveChannel(index, field, view, shard, a, resChan) - return -} - func (c *catcherTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator { return c.b.NewTxIterator(index, field, view, shard) } diff --git a/executor.go b/executor.go index 4a3345393..0f52e09d8 100644 --- a/executor.go +++ b/executor.go @@ -8308,24 +8308,30 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, index strin func DeleteRows(ctx context.Context, src *Row, idx *Index, shard uint64) (bool, error) { return DeleteRowsWithFlow(ctx, src, idx, shard, false) } -func clearFragment(writeTx Tx, columns *roaring.Bitmap, frag *fragment, resChan chan countResults) (changed bool, err error) { - posChan := make(chan uint64, 8192) - findExisting := roaring.NewBitmapBitmapFilter(columns, func(pos uint64) error { - posChan <- pos - return nil - }) - go writeTx.RemoveChannel(frag.index(), frag.field(), frag.view(), frag.shard, posChan, resChan) +func clearFragment(writeTx Tx, columns *roaring.Bitmap, frag *fragment, toClear []uint64) (changed bool, err error) { + rowSet := make(map[uint64]struct{}) + toClear = toClear[:0] + callback := func(pos uint64) error { + toClear = append(toClear, pos) + rowID := pos / ShardWidth + rowSet[rowID] = struct{}{} + return nil + } + findExisting := roaring.NewBitmapBitmapFilter(columns, callback) err = writeTx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting) - close(posChan) + if err != nil { return false, err } - r := <-resChan - - changed = r.changeCount > 0 - err = r.err - return + if len(toClear) > 0 { + err = frag.importPositions(writeTx, []uint64{}, toClear, rowSet) + if err != nil { + return false, err + } + return true, nil + } + return false, nil } func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, idx *Index, shard uint64, normalFlow bool) (bool, error) { @@ -8378,8 +8384,8 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id } }() - resChan := make(chan countResults) + toClear := make([]uint64, 0) for _, field := range idx.Fields() { for _, view := range field.views() { @@ -8387,7 +8393,7 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id if !ok { continue } - c, err := clearFragment(writeTx, columns, frag, resChan) + c, err := clearFragment(writeTx, columns, frag, toClear) if err != nil { return false, err } @@ -8397,7 +8403,6 @@ func DeleteRowsWithFlowWithKeys(ctx context.Context, columns *roaring.Bitmap, id } } - close(resChan) if existenceFragment != nil { //a string keys have been deleted and the deleteRow was created if normalFlow { existenceFragment.clearRow(writeTx, deletedRowID) @@ -8431,7 +8436,7 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx return } }() - resChan := make(chan countResults) + toClear := make([]uint64, 0) for _, field := range idx.Fields() { for _, view := range field.views() { @@ -8439,7 +8444,7 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx if !ok { continue } - c, err := clearFragment(writeTx, columns, frag, resChan) + c, err := clearFragment(writeTx, columns, frag, toClear) if err != nil { return false, err } @@ -8449,7 +8454,6 @@ func DeleteRowsWithOutKeysFlow(ctx context.Context, columns *roaring.Bitmap, idx } } - close(resChan) if existenceFragment == nil { //a string keys have been deleted and the deleteRow was created return changed, nil } diff --git a/ingest_test.go b/ingest_test.go index 712b834e6..7339f75e9 100644 --- a/ingest_test.go +++ b/ingest_test.go @@ -146,6 +146,9 @@ func parseExpectedResults(data []byte) (ints []uint64, keys []string, err error) return nil, nil, errors.New("expecting [] results") } words := bytes.Split(data[1:len(data)-1], []byte{','}) + if len(words) == 1 && len(words[0]) == 0 { + return nil, nil, nil + } for _, word := range words { word = bytes.TrimSpace(word) if len(word) == 0 { diff --git a/ingest_testdata/sample.tc b/ingest_testdata/sample.tc index 6ddff22a4..764965025 100644 --- a/ingest_testdata/sample.tc +++ b/ingest_testdata/sample.tc @@ -87,3 +87,13 @@ ingest-error: } } ] +ingest: +[ + { + "action": "delete", + "record_ids": [ 1 ] + } +] +queries: +Row(setkey="a") +[] diff --git a/internal_client.go b/internal_client.go index 1bff1e90a..cbfeb355f 100644 --- a/internal_client.go +++ b/internal_client.go @@ -144,10 +144,15 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o return ic } +// AddAuthToken checks in a couple spots for our authorization token and adds it to +// the Authorization Header in the request if it finds it. func AddAuthToken(ctx context.Context, req *http.Request) *http.Request { - token, ok := ctx.Value("token").(string) - if ok && token != "" { + if token, ok := ctx.Value("token").(string); ok && token != "" { + // the "token" value should be prefixed with "Bearer" req.Header.Set("Authorization", token) + } else if uinfo := ctx.Value("userinfo"); uinfo != nil { + // UserInfo.Token is not prefixed with "Bearer" + req.Header.Set("Authorization", "Bearer "+uinfo.(*authn.UserInfo).Token) } return req } @@ -606,12 +611,6 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index str return nil, errors.Wrap(err, "creating request") } - uinfo := ctx.Value("userinfo") - if uinfo != nil { - token := uinfo.(*authn.UserInfo).Token - req.Header.Set("Authorization", token) - } - req = AddAuthToken(ctx, req) req.Header.Set("Content-Length", strconv.Itoa(len(buf))) diff --git a/internal_client_test.go b/internal_client_test.go index 204a85cae..8acb74751 100644 --- a/internal_client_test.go +++ b/internal_client_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/hex" "fmt" + "net/http" gohttp "net/http" "reflect" "strings" @@ -15,6 +16,7 @@ import ( "github.com/davecgh/go-spew/spew" pilosa "github.com/molecula/featurebase/v3" + "github.com/molecula/featurebase/v3/authn" "github.com/molecula/featurebase/v3/encoding/proto" "github.com/molecula/featurebase/v3/pql" "github.com/molecula/featurebase/v3/server" @@ -1568,3 +1570,38 @@ func TestClient_ImportRoaringExists(t *testing.T) { } } + +func TestAddAuthToken(t *testing.T) { + t.Run("none", func(t *testing.T) { + req, err := http.NewRequest("GET", "dontmatternone", strings.NewReader("this doesn't matter")) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + pilosa.AddAuthToken(context.Background(), req) + if req.Header.Get("Authorization") != "" { + t.Fatalf("Authorization header set when it should be empty") + } + }) + t.Run("userinfo", func(t *testing.T) { + req, err := http.NewRequest("GET", "dontmatternone", strings.NewReader("this doesn't matter")) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + uinfo := &authn.UserInfo{Token: "ayo"} + pilosa.AddAuthToken(context.WithValue(context.Background(), "userinfo", uinfo), req) + if got := req.Header.Get("Authorization"); got != "Bearer "+uinfo.Token { + t.Fatalf("got '%v', expected 'Bearer %v'", got, uinfo.Token) + } + }) + t.Run("token", func(t *testing.T) { + req, err := http.NewRequest("GET", "dontmatternone", strings.NewReader("this doesn't matter")) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + tok := "Bearer thisisatoken" + pilosa.AddAuthToken(context.WithValue(context.Background(), "token", tok), req) + if got := req.Header.Get("Authorization"); got != tok { + t.Fatalf("got '%v', expected '%v'", got, tok) + } + }) +} diff --git a/qa/scripts/backupRestoreTest.sh b/qa/scripts/backupRestoreTest.sh new file mode 100755 index 000000000..794898e41 --- /dev/null +++ b/qa/scripts/backupRestoreTest.sh @@ -0,0 +1,112 @@ +#!/bin/bash + +HOSTS=($@) +/data/datagen --source custom --custom-config /data/backup_test_datagen.yaml --pilosa.index=backup_test --pilosa.hosts=$HOSTS --pilosa.batch-size=1000 +# make sure backup doesn't fail +if (( $? != 0 )); then + echo "datagen failed" + exit 1 +fi + +# kill a node somehow +echo "hosts we've got" +for host in ${HOSTS[@]}; do + echo $host; +done + +KILLNODE=${HOSTS[1]} +echo "getting checksum from $KILLNODE" +firstCheckSum=$(featurebase chksum --host $KILLNODE) +echo "first check sum: $firstCheckSum" +if (( $? != 0 )) +then + echo "getting checksum from $KILLNODE failed" + exit 1 +fi + +echo "pausing node $KILLNODE" +ssh -A -o "StrictHostKeyChecking no" ec2-user@${KILLNODE} "sudo systemctl stop featurebase" +if (( $? != 0 )) +then + echo "pausing node $KILLNODE failed" + exit 1 +fi + +# backup from ingest node +featurebase backup --host=${HOSTS[2]} --retry-period=0s --output=backupWOOO +# make sure backup doesn't fail +if (( $? != 0 )); then + echo "backup failed!!!!!!!!!!" + exit 1 +fi + +echo "stopping all featurebase nodes" +# kill all featurebase nodes and wipe the data directory +for host in ${HOSTS[@]}; do + echo "stopping featurebase on ${host}" + ssh -A -o "StrictHostKeyChecking no" ec2-user@${host} "sudo systemctl stop featurebase && sudo find /data/featurebase/ -mindepth 1 -delete" + if (( $? != 0 )); then + echo "emptying and stopping node failed" + exit 1 + fi +done + +echo "starting all featurebase nodes" +# start all featurebase nodes - systemd +for host in ${HOSTS[@]}; do + echo "starting featurebase on ${host}" + ssh -A -o "StrictHostKeyChecking no" ec2-user@${host} "sudo systemctl restart featurebase" + if (( $? != 0 )); then + echo "restarting featurebase failed" + exit 1 + fi +done + +echo "waiting for featurebase to start" +# wait until we can connect to one of the hosts +for i in {0..24}; do + echo "checking ${HOSTS[1]}:10101/status" + curl -v ${HOSTS[1]}:10101/status + S=$(curl -s ${HOSTS[1]}:10101/status | jq -r ".state"); + if [[ $S == "NORMAL" ]]; then + echo "hosts up after $i tries"; + break; + fi + echo "attempt $i resulted in $S" + sleep 5 +done + + +if [[ $S != "NORMAL" ]]; then + echo "couldn't connect, featurebase never stable" + exit 1 +fi + +echo "restoring featurebase" +# featurebase restore +featurebase restore -s=backupWOOO --host=${HOSTS[1]} + +# make sure it doesn't fail +if (( $? != 0 )); then + echo "restore failed!!!!" + exit 1 +fi + +# make sure it's the same data +echo "getting checksum from $KILLNODE" +secondChkSum=$(featurebase chksum --host $KILLNODE) +echo "second checkSum $secondChkSum" +if (( $? != 0 )); then + echo "getting checksum from $KILLNODE failed" + exit 1 +fi + +if [[ $firstCheckSum == $secondChkSum ]]; then + echo "checksums match" + exit 0 +else + echo "first check sum $firstCheckSum" + echo "second check sum $secondCheckSum" + echo "check sums don't match" + exit 1 +fi diff --git a/qa/scripts/backup_test_datagen.yaml b/qa/scripts/backup_test_datagen.yaml new file mode 100644 index 000000000..c9b8292c5 --- /dev/null +++ b/qa/scripts/backup_test_datagen.yaml @@ -0,0 +1,61 @@ +fields: + - name: "a_random_string" + type: "string" # (default StringField (non-mutex)) + generator_type: "random-string" # used to generate random strings rather than pulling from known set + min_len: 8 + max_len: 12 + charset: "AB" # set of possible characters to pull from when generating random string + - name: "id" + type: "uint" + distribution: "sequential" + min: 0 + max: 1000 # 2%24 + step: 1 + repeat: false + - name: "type" + type: "int" # (default IntField) + min: 0 + max: 3 + distribution: "zipfian" + s: 1.1 + v: 5.1 + - name: "ts" + type: "timestamp" + min_date: 2006-01-02T15:04:05.001Z # RFC3339Nano + max_date: 2007-01-02T15:04:05.001Z # RFC3339Nano + distribution: "increasing" # only "increasing" is supported right now + min_step_duration: "10ns" + max_step_duration: "200ms" + - name: "slice" + type: "uint-set" # (default IDArrayField) + min: 0 + max: 35000 + distribution: "zipfian" + s: 1.1 + v: 5.1 + min_num: 1 + max_num: 50 + +# idk_params describe how data from "fields" should be ingested by IDK +idk_params: + primary_key_config: + field: "a_random_string" # if this is a single field named "id" then we'll use uint IDs, if it's empty we'll autogen ids, and if it's anything else we'll do string keys... yes this is a bit hacky, needs to be cleaned up. + # fields is keyed by names of fields from top level "fields". It is + # not required that all fields appear here, those that don't will + # use the default ingestion. + fields: + id: + - type: "ID" + type: + - type: "ID" + a_decimal_field: + - type: "Decimal" + scale: 4 + ts: + - type: "RecordTime" + layout: "2006-01-02T15:04:05Z" + epoch: 1970-01-01T00:00:00.0Z + name: "na" + slice: + - type: "IDArray" + time_quantum: "YMD" diff --git a/qa/scripts/perf/delete/deleteSetup.sh b/qa/scripts/perf/delete/deleteSetup.sh index 6cadbcd6f..f3194217b 100755 --- a/qa/scripts/perf/delete/deleteSetup.sh +++ b/qa/scripts/perf/delete/deleteSetup.sh @@ -50,7 +50,7 @@ echo "}" DEPLOYED_INGEST_IPS_LEN=`echo "$DEPLOYED_INGEST_IPS" | wc -l` -#wait until we can connect to one of the hosts +# wait until we can connect to one of the hosts for i in {0..24} do ssh -A -i ~/.ssh/gitlab-featurebase-ci.pem -o StrictHostKeyChecking=no -o ConnectTimeout=10 ec2-user@${DATANODE0} "pwd" diff --git a/qa/scripts/setupBackupRestoreGauntlet.sh b/qa/scripts/setupBackupRestoreGauntlet.sh new file mode 100755 index 000000000..2fa611ff9 --- /dev/null +++ b/qa/scripts/setupBackupRestoreGauntlet.sh @@ -0,0 +1,79 @@ +#!/bin/bash + +# To run script: ./setupBackupRestoreGauntlet.sh +export TF_IN_AUTOMATION=1 + +SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) +source $SCRIPT_DIR/utilCluster.sh + +pushd ./qa/tf/gauntlet/backuprestore +echo "Running terraform init..." +terraform init -input=false +echo "Running terraform apply..." +terraform apply -input=false -auto-approve +terraform output -json > outputs.json +popd + +# get the first ingest host +INGESTNODE0=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.ingest_ips][0]["value"][0]') +echo "using INGESTNODE0 ${INGESTNODE0}" + +# get the first data host +DATANODE0=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.data_node_ips][0]["value"][0]') +echo "using DATANODE0 ${DATANODE0}" + + +DEPLOYED_CLUSTER_PREFIX=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.cluster_prefix][0]["value"]') +echo "Using DEPLOYED_CLUSTER_PREFIX: ${DEPLOYED_CLUSTER_PREFIX}" + +DEPLOYED_CLUSTER_REPLICA_COUNT=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.fb_cluster_replica_count][0]["value"]') +echo "Using DEPLOYED_CLUSTER_REPLICA_COUNT: ${DEPLOYED_CLUSTER_REPLICA_COUNT}" + +DEPLOYED_DATA_IPS=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.data_node_ips][0]["value"][]') +echo "DEPLOYED_DATA_IPS: {" +echo "${DEPLOYED_DATA_IPS}" +echo "}" + +DEPLOYED_DATA_IPS_LEN=`echo "$DEPLOYED_DATA_IPS" | wc -l` + +DEPLOYED_INGEST_IPS=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.ingest_ips][0]["value"][]') +echo "DEPLOYED_INGEST_IPS: {" +echo "${DEPLOYED_INGEST_IPS}" +echo "}" + +DEPLOYED_INGEST_IPS_LEN=`echo "$DEPLOYED_INGEST_IPS" | wc -l` + +#wait until we can connect to one of the hosts +for i in {0..24} +do + ssh -A -i ~/.ssh/gitlab-featurebase-ci.pem -o StrictHostKeyChecking=no -o ConnectTimeout=10 ec2-user@${DATANODE0} "pwd" + if [ $? -eq 0 ] + then + echo "Cluster is up after ${i} tries." + break + fi + sleep 10 +done + +ssh -A -i ~/.ssh/gitlab-featurebase-ci.pem -o StrictHostKeyChecking=no -o ConnectTimeout=10 ec2-user@${DATANODE0} "pwd" +if [ $? -ne 0 ] +then + echo "Unable to connect to cluster - giving up" + exit 1 +fi + +setupClusterNodes + +# verify featurebase running +echo "Verifying featurebase cluster running..." +curl -s http://${DATANODE0}:10101/status +if (( $? != 0 )) +then + echo "Featurebase cluster not running" + exit 1 +fi + +echo "Cluster running." + + + diff --git a/qa/scripts/teardownBackupRestoreGauntlet.sh b/qa/scripts/teardownBackupRestoreGauntlet.sh new file mode 100755 index 000000000..0baa5c74b --- /dev/null +++ b/qa/scripts/teardownBackupRestoreGauntlet.sh @@ -0,0 +1,7 @@ +#!/bin/bash + +# To run script: ./teardownBackupRestoreGauntlet.sh + +cd qa/tf/gauntlet/backuprestore +export TF_IN_AUTOMATION=1 +terraform destroy -auto-approve diff --git a/qa/scripts/testBackupRestoreGauntlet.sh b/qa/scripts/testBackupRestoreGauntlet.sh new file mode 100755 index 000000000..a814901c9 --- /dev/null +++ b/qa/scripts/testBackupRestoreGauntlet.sh @@ -0,0 +1,51 @@ +#!/bin/bash + +source ./qa/scripts/utilCluster.sh + +# get the first ingest host +INGESTNODE0=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.ingest_ips][0]["value"][0]') +echo "using INGESTNODE0 ${INGESTNODE0}" + +# get the first data host +DATANODE0=$(cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '[.data_node_ips][0]["value"][0]') +echo "using DATANODE0 ${DATANODE0}" + +# get the data node host ips +HOSTS=($( cat ./qa/tf/gauntlet/backuprestore/outputs.json | jq -r '.data_node_ips.value' | tr -d '[],"')) + +echo "using hosts:" +for host in ${HOSTS[@]}; do + echo $host; +done + +installDatagen $INGESTNODE0 + +# copy the tests over to ingest node +scp -r -i ~/.ssh/gitlab-featurebase-ci.pem ./qa/scripts/backupRestoreTest.sh ec2-user@${INGESTNODE0}:/data +if (( $? != 0 )) +then + echo "test copy failed" + exit 1 +fi + +# copy the datagen over to ingest node +scp -r -i ~/.ssh/gitlab-featurebase-ci.pem ./qa/scripts/backup_test_datagen.yaml ec2-user@${INGESTNODE0}:/data +if (( $? != 0 )) +then + echo "test copy failed" + exit 1 +fi + + +echo "running backup test with hosts: ${HOSTS[@]}" +ssh -A -i ~/.ssh/gitlab-featurebase-ci.pem -o "StrictHostKeyChecking no" ec2-user@${INGESTNODE0} "cd /data; ./backupRestoreTest.sh ${HOSTS[@]}" +TESTRESULT=$? + +if (( $TESTRESULT != 0 )) +then + echo "backup test failed" +else + echo "backup test complete" +fi + +exit $TESTRESULT diff --git a/qa/scripts/utilCluster.sh b/qa/scripts/utilCluster.sh index e083e082e..625bace9a 100644 --- a/qa/scripts/utilCluster.sh +++ b/qa/scripts/utilCluster.sh @@ -230,3 +230,55 @@ setupClusterNodes() { setupIngestNodes } + +installDatagen() { + INGESTNODE0=$1 + # download datagen + aws s3 cp s3://molecula-artifact-storage/idk/master/_latest/idk-linux-arm64/datagen datagen + if (( $? != 0 )) + then + echo "datagen binary copy failed" + exit 1 + fi + + # make it executable + chmod +x datagen + if (( $? != 0 )) + then + echo "couldn't make datagen executable" + exit 1 + fi + + # copy it over to the ingest node + scp -r -i ~/.ssh/gitlab-featurebase-ci.pem ./datagen ec2-user@${INGESTNODE0}:/data + if (( $? != 0 )) + then + echo "datagen copy failed" + exit 1 + fi + + # setup the yum repo needed for librdkafka onto the ingest node + scp -r -i ~/.ssh/gitlab-featurebase-ci.pem ./qa/scripts/perf/delete/confluent ec2-user@${INGESTNODE0}:/data + if (( $? != 0 )) + then + echo "confluent repo setup copy failed" + exit 1 + fi + + echo "setting up confluent repo" + ssh -A -i ~/.ssh/gitlab-featurebase-ci.pem -o "StrictHostKeyChecking no" ec2-user@${INGESTNODE0} "sudo mv /data/confluent /etc/yum.repos.d" + if (( $? != 0 )) + then + echo "setting up confluent repo failed" + exit 1 + fi + + + echo "installing librdkafka on ingest node" + ssh -A -i ~/.ssh/gitlab-featurebase-ci.pem -o "StrictHostKeyChecking no" ec2-user@${INGESTNODE0} "sudo rpm --import http://packages.confluent.io/rpm/3.1/archive.key && sudo yum clean all && sudo yum install librdkafka-devel -y" + if (( $? != 0 )) + then + echo "librdkafka install failed" + exit 1 + fi +} diff --git a/qa/tf/gauntlet/backuprestore/README.md b/qa/tf/gauntlet/backuprestore/README.md new file mode 100644 index 000000000..e7b633516 --- /dev/null +++ b/qa/tf/gauntlet/backuprestore/README.md @@ -0,0 +1,35 @@ +With terraform installed (`brew install terraform` if not)... + +You can do `terraform plan` -> `terraform apply` to spin up a cluster, `terraform destroy` to tear one down. + +## Other prerequisites: +Please read these carefully. + +Be in the `tf` directory (e.g., when you try to run a `terraform` command, the output of `pwd` should be `.../featurebase/qa/tf`) + +Currently, the path to the terraform module is using a local reference, i.e., in `main.tf`, the source line is assuming that you have `molecula-terraform` project installed locally, such that the `molecular-terraform` project and `featurebase` have the same parent directory (e.g., `...A/featurebase/qa/tf` and `...A/molecular-terraform/aws/.modules/featurebase-cluster` should both be valid paths). + +In addition, you must currently have a local copy of the `fb901` branch for the `molecular-terraform` project (located in the previously specified directory). + +Last thing, there is a key that is currently in 1Password (in the `Shared` vault, called `gitlab-featurebase-ci AWS key`) that must be in `~/.ssh/`, `chmod 400`, named `gitlab-featurebase-ci.pem`. You need this key to SSH to these instances. Assuming an `~/.ssh/config` like the following (append to the top of yours) +``` +Host test_* + User ec2-user + IdentityFile ~/.ssh/gitlab-featurebase-ci.pem +Host test_ingest + HostName 3.143.237.165 +Host test_node + HostName 10.0.1.142 + ProxyJump test_ingest +``` +except with the `test_ingest`'s `HostName` being the public, `ingest_ips` output from `terraform output` and `test_node`'s `HostName` being one of the private, `data_node_ips` output from `terraform output`. (Hopefully the rationale to use the ssh config to do the jumping like this makes sense; you can do `ssh test_ingest` or `ssh test_node` with minimal further fiddling.) + +OR specify cert to us directly thus: + +`ssh -A -i ~/.ssh/gitlab-featurebase-ci.pem ec2-user@ip_address` + +-A is used to ensure key forwarding. + +### TODOs +* We need a `user-data.sh` script which sets up/installs featurebase (possibly installs go, most likely pulls the artifacts from GitLab; sets up featurebase on both the node and data workers). +* Logs get sent to DataDog? diff --git a/qa/tf/gauntlet/backuprestore/main.tf b/qa/tf/gauntlet/backuprestore/main.tf new file mode 100644 index 000000000..87a68d0f0 --- /dev/null +++ b/qa/tf/gauntlet/backuprestore/main.tf @@ -0,0 +1,17 @@ +module "samsung-cluster" { + source = "../../.modules/featurebase-cluster" + cluster_prefix = var.cluster_prefix + region = var.region + profile = var.profile + fb_data_node_type = "m6g.xlarge" + fb_data_disk_iops = 10000 + fb_data_node_count = 3 + fb_ingest_type = "m6g.large" + fb_cluster_replica_count = 3 + fb_ingest_disk_iops = 10000 + fb_ingest_node_count = 1 + vpc_id = "vpc-05a26a122f961dc2b" + vpc_cidr_block = "10.0.0.0/16" + vpc_public_subnets = ["subnet-066b4b922b54e51a2", "subnet-037b8884269a69025", "subnet-08482631514426210", ] + vpc_private_subnets = ["subnet-0319dde319380326f", "subnet-0517ca9a646d80f88", "subnet-05a7b685ed27eb1cf", ] +} diff --git a/qa/tf/gauntlet/backuprestore/outputs.tf b/qa/tf/gauntlet/backuprestore/outputs.tf new file mode 100644 index 000000000..c00860a34 --- /dev/null +++ b/qa/tf/gauntlet/backuprestore/outputs.tf @@ -0,0 +1,19 @@ +output "ingest_ips" { + description = "List of ingest IPs" + value = module.samsung-cluster.ingest_ips +} + +output "data_node_ips" { + description = "List of data node IPs" + value = module.samsung-cluster.data_node_ips +} + +output "cluster_prefix" { + description = "The cluster prefix used" + value = module.samsung-cluster.cluster_prefix +} + +output "fb_cluster_replica_count" { + description = "The cluster replica count used" + value = module.samsung-cluster.fb_cluster_replica_count +} diff --git a/qa/tf/gauntlet/backuprestore/provider.tf b/qa/tf/gauntlet/backuprestore/provider.tf new file mode 100644 index 000000000..c0fc95d9d --- /dev/null +++ b/qa/tf/gauntlet/backuprestore/provider.tf @@ -0,0 +1,4 @@ +provider "aws" { + region = var.region + profile = var.profile +} \ No newline at end of file diff --git a/qa/tf/gauntlet/backuprestore/tf.auto.tfvars b/qa/tf/gauntlet/backuprestore/tf.auto.tfvars new file mode 100644 index 000000000..ac6de62a6 --- /dev/null +++ b/qa/tf/gauntlet/backuprestore/tf.auto.tfvars @@ -0,0 +1,2 @@ +region = "us-east-2" +profile = "service-terraform" \ No newline at end of file diff --git a/qa/tf/gauntlet/backuprestore/variables.tf b/qa/tf/gauntlet/backuprestore/variables.tf new file mode 100644 index 000000000..e55c7936d --- /dev/null +++ b/qa/tf/gauntlet/backuprestore/variables.tf @@ -0,0 +1,14 @@ +variable "region" { + description = "The AWS region in which the VPC should be built" + type = string +} + +variable "profile" { + description = "The name of the AWS profile Terraform should use for auth." + type = string +} + +variable "cluster_prefix" { + type = string + description = "This is a identifier that will be prefixed to created resources" +} diff --git a/rbf.go b/rbf.go index 1cdc59261..a6e81e463 100644 --- a/rbf.go +++ b/rbf.go @@ -228,71 +228,6 @@ type countResults struct { err error } -// RemoveChannel provides a method of streaming in bits or positions and not requiring a large buffer like add and remove -// the bits are input via the posChanel and the results are returned via the retChannel -func (tx *RBFTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) { - name := rbfName(index, field, view, shard) - var lastHi uint64 = math.MaxUint64 // highbits is always less than this starter. - var rc *roaring.Container - var hi uint64 - var lo uint16 - var err error - changeCount := 0 - i := 0 - for v := range a { - hi, lo = highbits(v), lowbits(v) - if hi != lastHi { - // either first time through, or changed to a different container. - // do we need put the last updated container now? - if i > 0 { - // not first time through, write what we got. - if rc == nil || (rc.N() == 0) { - err = tx.tx.RemoveContainer(name, lastHi) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to remove container")} - return - } - } else { - rc = roaring.Optimize(rc) - - err = tx.tx.PutContainer(name, lastHi, rc) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to put container")} - return - } - } - } - // get the next container - rc, err = tx.tx.Container(name, hi) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to retrieve container")} - return - } - } // else same container, keep adding bits to rct. - chng := false - rc, chng = rc.Remove(lo) - if chng { - changeCount++ - } - lastHi = hi - i++ - } - // write the last updates. - if rc == nil || rc.N() == 0 { - err = tx.tx.RemoveContainer(name, hi) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "failed to remove container")} - return - } - } else { - err = tx.tx.PutContainer(name, hi, rc) - if err != nil { - resChan <- countResults{0, errors.Wrap(err, "put to remove container")} - return - } - } - resChan <- countResults{changeCount, nil} -} func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove bool, a ...uint64) (changeCount int, err error) { if len(a) == 0 { return 0, nil diff --git a/stattx.go b/stattx.go index 5ce265e90..4780f70bc 100644 --- a/stattx.go +++ b/stattx.go @@ -159,7 +159,6 @@ const ( kOffsetRange kLast // mark the end, always keep this last. The following aren't tracked atm: kType - kRemoveChannel ) func (k kall) String() string { @@ -206,8 +205,6 @@ func (k kall) String() string { return "kLast" case kType: return "kType" - case kRemoveChannel: - return "kRemoveChannel" } vprint.PanicOn(fmt.Sprintf("unknown kall '%v'", int(k))) return "" @@ -224,15 +221,6 @@ func (c *statTx) NewTxIterator(index, field, view string, shard uint64) *roaring }() return c.b.NewTxIterator(index, field, view, shard) } -func (c *statTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) { - me := kRemoveChannel - t0 := time.Now() - defer func() { - c.stats.add(me, time.Since(t0)) - }() - c.b.RemoveChannel(index, field, view, shard, a, resChan) - return -} func (c *statTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) { me := kImportRoaringBits diff --git a/tx.go b/tx.go index 7be5a54b7..194b3e8e9 100644 --- a/tx.go +++ b/tx.go @@ -155,7 +155,6 @@ type Tx interface { GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) GetFieldSizeBytes(index, field string) (uint64, error) - RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) } // GenericApplyFilter implements ApplyFilter in terms of tx.ContainerIterator,