Merge branch 'master' into fb-1291

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

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.

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

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:

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

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
 	}

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 {

ingest_testdata/sample.tc

@@ -87,3 +87,13 @@ ingest-error:
     }
   }
 ]
+ingest:
+[
+  {
+    "action": "delete",
+    "record_ids": [ 1 ]
+  }
+]
+queries:
+Row(setkey="a")
+[]

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

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

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 

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"

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"

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."
+
+
+

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

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 

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

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?

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", ]
+}

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

qa/tf/gauntlet/backuprestore/provider.tf

@@ -0,0 +1,4 @@
+provider "aws" {
+  region  = var.region
+  profile = var.profile
+}

qa/tf/gauntlet/backuprestore/tf.auto.tfvars

@@ -0,0 +1,2 @@
+region  = "us-east-2"
+profile = "service-terraform"

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"
+}

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

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

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,