Repository: brimsec/zq
Pull request #2004 (2021-01-22 21:57:04)

Update deploy job to use ZTest
This adds a new set of ZTests tagged with "cluster" to run the same tests that were previously run directly with zapi in cd.yaml.

A Makefile target, test-cluster, that runs the cluster tests has been introduced. The cluster tests can be more easily extended and maintained as ZTests.

In addition, it addresses two existing intermittent failures in the Deploy test in cd.yaml.
(1) On some test failures, the temporary namespace (e.g. ci-408) was not being removed after the test. This PR consolidates three previous jobs: pre-clean, eks-test-deploy, and eks-test-zapi-1, into a single job that shares the same cleanup code that removes the temporary namespace.
(2) There was a "sleep 2" after the port-forward commands. It is replaced by two loops that verify the port-forward with netcat before proceeding.


Patch:
diff --git a/.github/workflows/cd.yaml b/.github/workflows/cd.yaml
index 9c28cbc84..36070782d 100644
--- a/.github/workflows/cd.yaml
+++ b/.github/workflows/cd.yaml
@@ -6,25 +6,8 @@ on:
       - master
 
 jobs:
-  pre-clean:
-    # Delete existing S3 folder and K8s namespace
-    # if they already exist from a previous run.
-    runs-on: ubuntu-latest
-    steps:
-      - uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
-          aws-region: us-east-2
-      - run: |
-          set -x
-          ZQD_DATA_URI=s3://zq-ci-test/$(date +%y-%m-%d).${{ github.run_number }}
-          aws eks --region us-east-2 update-kubeconfig --name zq-test
-          aws s3 rm --recursive $ZQD_DATA_URI || :
-          kubectl delete ns ci-${{ github.run_number }} || :
   ecr-image-push:
     # Build and push a Docker image for zqd.
-    needs: pre-clean
     runs-on: ubuntu-latest
     steps:
       - uses: aws-actions/configure-aws-credentials@v1
@@ -41,11 +24,14 @@ jobs:
           ZQD_ECR_HOST: ${{ secrets.AWS_ECR_HOST }}
         run: |
           make docker-push-ecr
-  eks-test-deploy:
+  eks-deploy-test:
     # Deploys this build to a temporary namespace in order to run smoke tests.
     needs: ecr-image-push
     runs-on: ubuntu-latest
     steps:
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '1.15'
       - uses: aws-actions/configure-aws-credentials@v1
         with:
           aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
@@ -56,64 +42,32 @@ jobs:
           ref: ${{ github.event.pull_request.head.sha }}
       - run: |
           set -x
-          NAMESPACE=ci-${{ github.run_number }}
+          namespace=ci-${{ github.run_number }}
           export ZQD_DATA_URI=s3://zq-ci-test/$(date +%y-%m-%d).${{ github.run_number }}
           export ZQD_ECR_HOST=${{ secrets.AWS_ECR_HOST }}
+          aws s3 rm --recursive $ZQD_DATA_URI || : # clear data from previous run
           aws eks --region us-east-2 update-kubeconfig --name zq-test
-          kubectl create ns $NAMESPACE || :
-          kubectl config set-context --current --namespace=$NAMESPACE
+          kubectl create ns $namespace || :
+          kubectl config set-context --current --namespace=$namespace
           helm uninstall zsrv || :
           k8s/postgres-secret.sh
           make helm-install PG_PERSIST=false
-          kubectl get pod
+          kubectl scale --replicas=4 deployment/zsrv-worker
           kubectl wait --for=condition=Ready --timeout=60s pods/zsrv-postgresql-0
-  eks-test-zapi-1:
-    # Run smoke tests for zqd image prior to deploying to ci-master.
-    needs: eks-test-deploy
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/setup-go@v2
-        with:
-          go-version: '1.15'
-      - uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
-          aws-region: us-east-2
-      - uses: actions/checkout@v1
-        with:
-          ref: ${{ github.event.pull_request.head.sha }}
-      - run: |
-          set -x
-          SPACE=files
-          ZQD_DATA_URI=s3://zq-ci-test/$(date +%y-%m-%d).${{ github.run_number }}
-          mkdir testlog
+          kubectl wait --for=condition=available --timeout=120s deployment/zsrv-recruiter
+          kubectl wait --for=condition=available --timeout=120s deployment/zsrv-root
+          kubectl wait --for=condition=available --timeout=120s deployment/zsrv-worker
           make install
-          aws eks --region us-east-2 update-kubeconfig --name zq-test
-          kubectl config set-context --current --namespace=ci-${{ github.run_number }}
-          kubectl scale --replicas=4 deployment/zsrv-worker
-          kubectl port-forward svc/zsrv-recruiter 8020:9867 &
-          kubectl port-forward svc/zsrv-root 9867:9867 &
-          sleep 2 # wait for port-forwards to complete
-          time zapi ls
-          time zapi new -k archivestore -d $ZQD_DATA_URI $SPACE
-          # Post an archive of about 1.6 GB to run test queries.
-          time zapi -s $SPACE postpath s3://brim-sampledata/wrccdc/zeek-logs/files.log.gz
+          source k8s/zsrv-port-forward.sh
           # There should be 4 workers running now.
           [[ 4 == $(curl http://localhost:8020/recruiter/listfree | jq -r ".workers | length") ]]
-          time zapi -s $SPACE get -workers 1 -t '"FgzbkQ15TOoOENn1lb" | sort tx_hosts' &> testlog/filter1.log
-          time zapi -s $SPACE get -workers 2 -t '"FgzbkQ15TOoOENn1lb" | sort tx_hosts' &> testlog/filter2.log
-          time zapi -s $SPACE get -workers 4 -t '"FgzbkQ15TOoOENn1lb" | sort tx_hosts' &> testlog/filter4.log
-          time zapi -s $SPACE get -workers 4 -t 'count()' &> testlog/count.tzng
-          time zapi -s $SPACE get -workers 4 -t \
-            '* | count() by tx_hosts | sort -r count | head 5' &> testlog/countbytxhost.tzng
-          diff testlog/count.tzng k8s/testdiff/count.tzng
-          diff testlog/countbytxhost.tzng k8s/testdiff/countbytxhost.tzng
+          make test-cluster
       - name: Get zqd logs and remove namespace on success or failure
         if: ${{ always() }}
         run: |
           set -x
           ZQD_DATA_URI=s3://zq-ci-test/$(date +%y-%m-%d).${{ github.run_number }}
+          mkdir testlog
           kubectl logs --tail -1 -l app.kubernetes.io/name=recruiter > testlog/recruiter.log
           kubectl logs --tail -1 -l app.kubernetes.io/name=root > testlog/root.log
           kubectl logs --tail -1 -l app.kubernetes.io/name=worker > testlog/worker.log
@@ -124,11 +78,10 @@ jobs:
         with:
           name: eks-testlog
           path: testlog/
-
   eks-master-deploy:
     # Deploys this build to ci-master namespace so it will be available
     # for manual testing. Only deploy if previous jobs succeed.
-    needs: eks-test-zapi-1
+    needs: eks-deploy-test
     runs-on: ubuntu-latest
     steps:
       - uses: aws-actions/configure-aws-credentials@v1
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ddca8d2db..4e0911a9b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,45 @@ These entries focus on changes we think are relevant to users of Brim,
 zq, or pcap.  For all changes to zqd, its API, or to other components in the
 zq repo, check the git log.
 
+## v0.28.0
+**NOTE** - Beginning with this release, a subset of the source code in the
+[github.com/brimsec/zq](https://github.com/brimsec/zq) GitHub repository is
+covered by a source-available style license, the
+[Polyform Perimeter License (PPL)](https://polyformproject.org/licenses/perimeter/1.0.0/).
+We've moved the PPL-covered code under a `ppl/` directory in the repository.
+The majority of our source code retains the existing BSD-3-Clause license.
+
+The overwhelming majority of zq/zqd users and developers will not be impacted
+by this change, including those using zq/zqd in commercial settings. The use of
+the source-available Polyform Perimeter license prevents use cases like
+marketing a work as a "as-a-service" style offering for server components like
+zqd while using material covered under the PPL.
+
+In general, we are making this change to ensure technology giants can't use the
+PPL-covered code to make replacement offerings of our projects. We believe
+users and developers should have access to the source code for our projects,
+and we need a sustainable business model to continue funding our work. Using
+the source-available Polyform Perimeter license on portions of the source code
+lets us realize both. 
+
+For more detail regarding licensing, see the
+[CONTRIBUTING.md](https://github.com/brimsec/zq/blob/master/CONTRIBUTING.md)
+doc, and feel free to come talk to us on
+[Slack](https://www.brimsecurity.com/join-slack/) if you have additional
+questions.
+
+---
+
+* zqd: Update Zeek pointer to [v3.2.1-brim9](https://github.com/brimsec/zeek/releases/tag/v3.2.1-brim9) which provides the latest [geolocation](https://github.com/brimsec/brim/wiki/Geolocation) data (#2010)
+* zqd: Update Suricata pointer to [v5.0.3-brim1](https://github.com/brimsec/build-suricata/releases/tag/v5.0.3-brim1) which disables checksum checks, allowing for alert creation on more types of pcaps (#1975)
+* zson: Update [Zeek Interoperability doc](https://github.com/brimsec/zq/blob/master/zng/docs/zeek-compat.md) to include current ZSON syntax (#1956)
+* zq: Ensure the output from the [`fuse`](https://github.com/brimsec/zq/tree/master/zql/docs/processors#fuse) processor is deterministic (#1958)
+* zq: Fix an issue where the presence of the Greek µ character caused a ZSON read parsing error (#1967)
+* zqd: Fix an issue where Zeek events generated during pcap import and written to an archivestore were only visible after ingest completion (#1973)
+* zqd: Change the logger configuration to output stacktraces on messages of level "warn" and higher (#1990)
+* zq: Update [performance results](https://github.com/brimsec/zq/blob/master/performance/README.md) to include ZSON read/write (#1974)
+
+
 ## v0.27.1
 * zq: Fix an issue where nested nulls caused a panic in CSV output (#1954)
 
diff --git a/Makefile b/Makefile
index 85406bd50..737dd703f 100644
--- a/Makefile
+++ b/Makefile
@@ -100,6 +100,17 @@ test-postgres-docker:
 		docker-compose -f $(CURDIR)/ppl/zqd/scripts/dkc-services.yaml down || exit; \
 		exit $$status
 
+# test-cluster target assumes zqd endpoint is available at port 9867
+.PHONY: test-cluster
+test-cluster: build install
+	-zapi rm files
+	-aws s3 rm --recursive $(ZQD_DATA_URI)
+	zapi new -k archivestore -d $(ZQD_DATA_URI) files
+	time zapi -s files postpath s3://brim-sampledata/wrccdc/zeek-logs/files.log.gz
+	@ZTEST_PATH="$(CURDIR)/dist:$(CURDIR)/bin" \
+		ZTEST_TAG=cluster \
+		go test -v -run TestZq/ppl/zqd/ztests/cluster .
+
 perf-compare: build $(SAMPLEDATA)
 	scripts/comparison-test.sh
 
diff --git a/cmd/zapi/cmd/auth/command.go b/cmd/zapi/cmd/auth/command.go
index 7231f822b..28cd6f3c4 100644
--- a/cmd/zapi/cmd/auth/command.go
+++ b/cmd/zapi/cmd/auth/command.go
@@ -22,6 +22,7 @@ func init() {
 	Auth.Add(Login)
 	Auth.Add(Logout)
 	Auth.Add(Method)
+	Auth.Add(Store)
 	Auth.Add(Verify)
 	cmd.CLI.Add(Auth)
 }
diff --git a/cmd/zapi/cmd/auth/login.go b/cmd/zapi/cmd/auth/login.go
index 3bdc3c8db..604b93ac4 100644
--- a/cmd/zapi/cmd/auth/login.go
+++ b/cmd/zapi/cmd/auth/login.go
@@ -71,11 +71,7 @@ func (c *LoginCommand) Run(args []string) error {
 		return err
 	}
 
-	cpath, err := cmd.UserStdCredentialsPath()
-	if err != nil {
-		return err
-	}
-	creds, err := cmd.LoadCredentials(cpath)
+	creds, err := c.LocalConfig.LoadCredentials()
 	if err != nil {
 		return fmt.Errorf("failed to load credentials file: %w", err)
 	}
@@ -84,9 +80,9 @@ func (c *LoginCommand) Run(args []string) error {
 		ID:      dar.IDToken,
 		Refresh: dar.RefreshToken,
 	})
-	if err := cmd.SaveCredentials(cpath, creds); err != nil {
+	if err := c.LocalConfig.SaveCredentials(creds); err != nil {
 		return fmt.Errorf("failed to save credentials file: %w", err)
 	}
-	fmt.Printf("Login successful, stored credentials for %s in %s\n", c.Host, cpath)
+	fmt.Printf("Login successful, stored credentials for %s\n", c.Host)
 	return nil
 }
diff --git a/cmd/zapi/cmd/auth/logout.go b/cmd/zapi/cmd/auth/logout.go
index 86d1a80a2..b0c4bd663 100644
--- a/cmd/zapi/cmd/auth/logout.go
+++ b/cmd/zapi/cmd/auth/logout.go
@@ -5,7 +5,6 @@ import (
 	"flag"
 	"fmt"
 
-	"github.com/brimsec/zq/cmd/zapi/cmd"
 	"github.com/mccanne/charm"
 )
 
@@ -35,18 +34,14 @@ func (c *LogoutCommand) Run(args []string) error {
 		return errors.New("logout command takes no arguments")
 	}
 
-	cpath, err := cmd.UserStdCredentialsPath()
-	if err != nil {
-		return err
-	}
-	svccreds, err := cmd.LoadCredentials(cpath)
+	creds, err := c.LocalConfig.LoadCredentials()
 	if err != nil {
 		return fmt.Errorf("failed to load credentials file: %w", err)
 	}
-	svccreds.RemoveTokens(c.Host)
-	if err := cmd.SaveCredentials(cpath, svccreds); err != nil {
+	creds.RemoveTokens(c.Host)
+	if err := c.LocalConfig.SaveCredentials(creds); err != nil {
 		return fmt.Errorf("failed to save credentials file: %w", err)
 	}
-	fmt.Printf("Logout successful, cleared credentials for %s in %s\n", c.Host, cpath)
+	fmt.Printf("Logout successful, cleared credentials for %s\n", c.Host)
 	return nil
 }
diff --git a/cmd/zapi/cmd/auth/store.go b/cmd/zapi/cmd/auth/store.go
new file mode 100644
index 000000000..c2c41a59d
--- /dev/null
+++ b/cmd/zapi/cmd/auth/store.go
@@ -0,0 +1,53 @@
+package auth
+
+import (
+	"errors"
+	"flag"
+	"fmt"
+
+	"github.com/mccanne/charm"
+)
+
+var Store = &charm.Spec{
+	Name:   "store",
+	Usage:  "auth store",
+	Short:  "store raw tokens",
+	Long:   ``,
+	New:    NewStore,
+	Hidden: true,
+}
+
+type StoreCommand struct {
+	*Command
+
+	accessToken string
+}
+
+func NewStore(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
+	c := &StoreCommand{Command: parent.(*Command)}
+	f.StringVar(&c.accessToken, "access", "", "raw access token as string")
+	return c, nil
+}
+
+func (c *StoreCommand) Run(args []string) error {
+	if len(args) > 0 {
+		return errors.New("store command takes no arguments")
+	}
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
+	}
+
+	creds, err := c.LocalConfig.LoadCredentials()
+	if err != nil {
+		return fmt.Errorf("failed to load credentials file: %w", err)
+	}
+	tokens, _ := creds.ServiceTokens(c.Host)
+	tokens.Access = c.accessToken
+	creds.AddTokens(c.Host, tokens)
+
+	if err := c.LocalConfig.SaveCredentials(creds); err != nil {
+		return fmt.Errorf("failed to save credentials file: %w", err)
+	}
+	return nil
+}
diff --git a/cmd/zapi/cmd/auth/verify.go b/cmd/zapi/cmd/auth/verify.go
index becd5f098..92bb06d58 100644
--- a/cmd/zapi/cmd/auth/verify.go
+++ b/cmd/zapi/cmd/auth/verify.go
@@ -6,7 +6,6 @@ import (
 	"flag"
 	"fmt"
 
-	"github.com/brimsec/zq/cmd/zapi/cmd"
 	"github.com/mccanne/charm"
 )
 
@@ -26,22 +25,6 @@ func NewVerify(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 	return &VerifyCommand{Command: parent.(*Command)}, nil
 }
 
-func LoadServiceCredentials(serviceURL string) (cmd.ServiceTokens, error) {
-	cpath, err := cmd.UserStdCredentialsPath()
-	if err != nil {
-		return cmd.ServiceTokens{}, err
-	}
-	svccreds, err := cmd.LoadCredentials(cpath)
-	if err != nil {
-		return cmd.ServiceTokens{}, fmt.Errorf("failed to load credentials file: %w", err)
-	}
-	creds, ok := svccreds.ServiceTokens(serviceURL)
-	if !ok {
-		return cmd.ServiceTokens{}, fmt.Errorf("no stored credentials for %v", serviceURL)
-	}
-	return creds, nil
-}
-
 func (c *VerifyCommand) Run(args []string) error {
 	defer c.Cleanup()
 	if err := c.Init(); err != nil {
@@ -52,11 +35,15 @@ func (c *VerifyCommand) Run(args []string) error {
 	}
 	conn := c.Connection()
 
-	creds, err := LoadServiceCredentials(c.Host)
+	creds, err := c.LocalConfig.LoadCredentials()
 	if err != nil {
 		return err
 	}
-	conn.SetAuthToken(creds.Access)
+	tokens, ok := creds.ServiceTokens(c.Host)
+	if !ok {
+		return fmt.Errorf("no stored credentials for %v", c.Host)
+	}
+	conn.SetAuthToken(tokens.Access)
 
 	res, err := conn.AuthIdentity(c.Context())
 	if err != nil {
diff --git a/cmd/zapi/cmd/cli.go b/cmd/zapi/cmd/cli.go
index e9741a8e7..c16a61c69 100644
--- a/cmd/zapi/cmd/cli.go
+++ b/cmd/zapi/cmd/cli.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"net"
 	"os"
 	"syscall"
 
@@ -58,18 +59,20 @@ func New(f *flag.FlagSet) (charm.Command, error) {
 	f.Var(&c.spaceID, "id", "<space_id>")
 	f.BoolVar(&c.NoFancy, "nofancy", c.NoFancy, "disable fancy CLI output (true if stdout is not a tty)")
 	c.cli.SetFlags(f)
+	c.LocalConfig.SetFlags(f)
 
 	return c, nil
 }
 
 type Command struct {
-	conn      *client.Connection
-	Host      string
-	Spacename string
-	NoFancy   bool
-	ctx       *signalCtx
-	spaceID   api.SpaceID
-	cli       cli.Flags
+	Host        string
+	LocalConfig LocalConfigFlags
+	NoFancy     bool
+	Spacename   string
+	cli         cli.Flags
+	conn        *client.Connection
+	ctx         *signalCtx
+	spaceID     api.SpaceID
 }
 
 func (c *Command) Context() context.Context {
@@ -78,9 +81,6 @@ func (c *Command) Context() context.Context {
 
 // Connection returns a central client.Connection instance.
 func (c *Command) Connection() *client.Connection {
-	if c.conn == nil {
-		c.conn = client.NewConnectionTo("http://" + c.Host)
-	}
 	return c.conn
 }
 
@@ -103,6 +103,19 @@ func (c *Command) Cleanup() {
 }
 
 func (c *Command) Init(all ...cli.Initializer) error {
+	if _, _, err := net.SplitHostPort(c.Host); err == nil {
+		c.Host = "http://" + c.Host
+	}
+	c.conn = client.NewConnectionTo(c.Host)
+
+	creds, err := c.LocalConfig.LoadCredentials()
+	if err != nil {
+		return err
+	}
+	if tokens, ok := creds.ServiceTokens(c.Host); ok {
+		c.conn.SetAuthToken(tokens.Access)
+	}
+
 	return c.cli.Init(all...)
 }
 
diff --git a/cmd/zapi/cmd/credentials.go b/cmd/zapi/cmd/credentials.go
index 2b937b141..5abdb691f 100644
--- a/cmd/zapi/cmd/credentials.go
+++ b/cmd/zapi/cmd/credentials.go
@@ -1,8 +1,9 @@
 package cmd
 
 import (
+	"flag"
+	"fmt"
 	"os"
-	"path"
 	"path/filepath"
 
 	"github.com/brimsec/zq/pkg/fs"
@@ -13,9 +14,32 @@ const (
 	credsFileName     = "credentials.json"
 )
 
-func LoadCredentials(path string) (*Credentials, error) {
+type LocalConfigFlags struct {
+	configPath string
+}
+
+func (f *LocalConfigFlags) SetFlags(fs *flag.FlagSet) {
+	fs.StringVar(&f.configPath, "configpath", "", "directory to store local configuration and credentials")
+}
+
+func (f *LocalConfigFlags) credentialsPath() (string, error) {
+	if f.configPath != "" {
+		return filepath.Abs(filepath.Join(f.configPath, credsFileName))
+	}
+	c, err := os.UserConfigDir()
+	if err != nil {
+		return "", fmt.Errorf("zapi: no user configuration directory available: %w", err)
+	}
+	return filepath.Join(c, zapiConfigDirName, credsFileName), nil
+}
+
+func (f *LocalConfigFlags) LoadCredentials() (*Credentials, error) {
+	cpath, err := f.credentialsPath()
+	if err != nil {
+		return nil, err
+	}
 	var cf Credentials
-	if err := fs.UnmarshalJSONFile(path, &cf); err != nil {
+	if err := fs.UnmarshalJSONFile(cpath, &cf); err != nil {
 		if os.IsNotExist(err) {
 			return &Credentials{}, nil
 		}
@@ -24,20 +48,15 @@ func LoadCredentials(path string) (*Credentials, error) {
 	return &cf, nil
 }
 
-func SaveCredentials(path string, cf *Credentials) error {
-	return fs.MarshalJSONFile(cf, path, 0600)
-}
-
-func UserStdCredentialsPath() (string, error) {
-	c, err := os.UserConfigDir()
+func (f *LocalConfigFlags) SaveCredentials(cf *Credentials) error {
+	cpath, err := f.credentialsPath()
 	if err != nil {
-		return "", err
+		return err
 	}
-	zcd := filepath.Join(c, zapiConfigDirName)
-	if err := os.MkdirAll(zcd, 0777); err != nil {
-		return "", err
+	if err := os.MkdirAll(filepath.Dir(cpath), 0700); err != nil {
+		return err
 	}
-	return path.Join(zcd, credsFileName), nil
+	return fs.MarshalJSONFile(cf, cpath, 0600)
 }
 
 type ServiceInfo struct {
diff --git a/cmd/zapi/cmd/index/create.go b/cmd/zapi/cmd/index/create.go
index 09b9b804c..9cc1bfff7 100644
--- a/cmd/zapi/cmd/index/create.go
+++ b/cmd/zapi/cmd/index/create.go
@@ -60,6 +60,10 @@ func (c *CreateCmd) Run(args []string) error {
 	if len(args) == 0 && c.zql == "" {
 		return errors.New("zapi index create: one or more indexing patterns must be specified")
 	}
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
+	}
 	id, err := c.SpaceID()
 	if err != nil {
 		return err
diff --git a/cmd/zapi/cmd/index/find.go b/cmd/zapi/cmd/index/find.go
index ead35e412..940bd14d0 100644
--- a/cmd/zapi/cmd/index/find.go
+++ b/cmd/zapi/cmd/index/find.go
@@ -73,6 +73,10 @@ func NewFind(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 }
 
 func (c *FindCmd) Run(args []string) error {
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
+	}
 	req := api.IndexSearchRequest{IndexName: c.indexFile, Patterns: args}
 	id, err := c.SpaceID()
 	if err != nil {
diff --git a/cmd/zapi/cmd/info/info.go b/cmd/zapi/cmd/info/info.go
index 3ca99381f..20dc81cab 100644
--- a/cmd/zapi/cmd/info/info.go
+++ b/cmd/zapi/cmd/info/info.go
@@ -41,6 +41,10 @@ func New(parent charm.Command, flags *flag.FlagSet) (charm.Command, error) {
 // Run lists all spaces in the current zqd host or if a parameter
 // is provided (in glob style) lists the info about that space.
 func (c *Command) Run(args []string) error {
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
+	}
 	conn := c.Connection()
 	var ids []api.SpaceID
 	if len(args) > 0 {
diff --git a/cmd/zapi/cmd/info/ls.go b/cmd/zapi/cmd/info/ls.go
index f6a4d26be..a7f4a707f 100644
--- a/cmd/zapi/cmd/info/ls.go
+++ b/cmd/zapi/cmd/info/ls.go
@@ -38,6 +38,10 @@ func NewLs(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 // Run lists all spaces in the current zqd host or if a parameter
 // is provided (in glob style) lists the info about that space.
 func (c *LsCommand) Run(args []string) error {
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
+	}
 	conn := c.Connection()
 	matches, err := cmd.SpaceGlob(c.Context(), conn, args...)
 	if err != nil {
diff --git a/cmd/zapi/cmd/new/new.go b/cmd/zapi/cmd/new/new.go
index c84f3badd..2bb189211 100644
--- a/cmd/zapi/cmd/new/new.go
+++ b/cmd/zapi/cmd/new/new.go
@@ -40,6 +40,7 @@ func (c *Command) Run(args []string) error {
 	if len(args) != 1 {
 		return errors.New("must specify a space name")
 	}
+	defer c.Cleanup()
 	if err := c.Init(&c.createFlags); err != nil {
 		return err
 	}
diff --git a/cmd/zapi/cmd/newsubspace/newsubspace.go b/cmd/zapi/cmd/newsubspace/newsubspace.go
index 4bfcbf76e..b1bbbde50 100644
--- a/cmd/zapi/cmd/newsubspace/newsubspace.go
+++ b/cmd/zapi/cmd/newsubspace/newsubspace.go
@@ -39,6 +39,10 @@ func (c *Command) Run(args []string) error {
 	if len(args) == 0 {
 		return errors.New("must specify at least one log from parent")
 	}
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
+	}
 	conn := c.Connection()
 	req := api.SubspacePostRequest{
 		Name: c.name,
diff --git a/cmd/zapi/cmd/post/path.go b/cmd/zapi/cmd/post/path.go
index d98e6fed6..ec88ab330 100644
--- a/cmd/zapi/cmd/post/path.go
+++ b/cmd/zapi/cmd/post/path.go
@@ -51,6 +51,7 @@ func (c *PostPathCommand) Run(args []string) (err error) {
 	if len(args) == 0 {
 		return errors.New("path arg(s) required")
 	}
+	defer c.Cleanup()
 	if err := c.Init(&c.postFlags); err != nil {
 		return err
 	}
diff --git a/cmd/zapi/cmd/post/pcap.go b/cmd/zapi/cmd/post/pcap.go
index 22a699b9b..faff90e6b 100644
--- a/cmd/zapi/cmd/post/pcap.go
+++ b/cmd/zapi/cmd/post/pcap.go
@@ -54,6 +54,7 @@ func (c *PostPcapCommand) Run(args []string) (err error) {
 	if len(args) == 0 {
 		return errors.New("path arg required")
 	}
+	defer c.Cleanup()
 	if err := c.Init(&c.postFlags); err != nil {
 		return err
 	}
diff --git a/cmd/zapi/cmd/post/post.go b/cmd/zapi/cmd/post/post.go
index 97afd83fc..ceed1d128 100644
--- a/cmd/zapi/cmd/post/post.go
+++ b/cmd/zapi/cmd/post/post.go
@@ -44,6 +44,7 @@ func (c *PostCommand) Run(args []string) (err error) {
 	if len(args) == 0 {
 		return errors.New("path arg(s) required")
 	}
+	defer c.Cleanup()
 	if err := c.Init(&c.postFlags); err != nil {
 		return err
 	}
diff --git a/cmd/zapi/cmd/rename/rename.go b/cmd/zapi/cmd/rename/rename.go
index 6a33d8010..39e403638 100644
--- a/cmd/zapi/cmd/rename/rename.go
+++ b/cmd/zapi/cmd/rename/rename.go
@@ -12,7 +12,7 @@ import (
 
 var Rename = &charm.Spec{
 	Name:  "rename",
-	Usage: "rename [old_name] [new_name]",
+	Usage: "rename [old_name] new_name",
 	Short: "renames a space",
 	Long:  `Renames a space, given the current space name and a desired new name.`,
 	New: func(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
@@ -29,18 +29,34 @@ type Command struct {
 }
 
 func (c *Command) Run(args []string) error {
-	if len(args) != 2 {
-		return errors.New("expected <old_name> <new_name>")
-	}
-	oldname := args[0]
-	newname := args[1]
-	id, err := cmd.GetSpaceID(c.Context(), c.Connection(), oldname)
-	if err != nil {
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
 		return err
 	}
+
+	var err error
+	var id api.SpaceID
+	var newname string
+	switch len(args) {
+	case 2:
+		newname = args[1]
+		id, err = cmd.GetSpaceID(c.Context(), c.Connection(), args[0])
+		if err != nil {
+			return err
+		}
+	case 1:
+		newname = args[0]
+		id, err = c.SpaceID()
+		if err != nil {
+			return err
+		}
+	default:
+		return errors.New("rename takes 1 or 2 arguments")
+	}
+
 	if err := c.Connection().SpacePut(c.Context(), id, api.SpacePutRequest{Name: newname}); err != nil {
 		return err
 	}
-	fmt.Printf("%s: space renamed to %s\n", oldname, newname)
+	fmt.Printf("space renamed to %s\n", newname)
 	return nil
 }
diff --git a/cmd/zapi/cmd/rm/rm.go b/cmd/zapi/cmd/rm/rm.go
index a9b1cd46b..8e61a9742 100644
--- a/cmd/zapi/cmd/rm/rm.go
+++ b/cmd/zapi/cmd/rm/rm.go
@@ -34,6 +34,10 @@ func (c *Command) Run(args []string) error {
 	if len(args) == 1 {
 		c.Spacename = args[0]
 	}
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
+	}
 	id, err := c.SpaceID()
 	if err != nil {
 		return err
diff --git a/cmd/zapi/cmd/version/command.go b/cmd/zapi/cmd/version/command.go
index e1121c52d..f7dd7ade6 100644
--- a/cmd/zapi/cmd/version/command.go
+++ b/cmd/zapi/cmd/version/command.go
@@ -34,10 +34,14 @@ func New(parent charm.Command, flags *flag.FlagSet) (charm.Command, error) {
 // Run lists all spaces in the current zqd host or if a parameter
 // is provided (in glob style) lists the info about that space.
 func (c *Command) Run(args []string) error {
-	conn := c.Connection()
 	if len(args) > 0 {
-		return errors.New("version command takes no arguemtns")
+		return errors.New("version command takes no arguments")
+	}
+	defer c.Cleanup()
+	if err := c.Init(); err != nil {
+		return err
 	}
+	conn := c.Connection()
 	version, err := conn.Version(c.Context())
 	if err != nil {
 		return err
diff --git a/compiler/expr.go b/compiler/expr.go
index be578fa04..f24d61d61 100644
--- a/compiler/expr.go
+++ b/compiler/expr.go
@@ -258,6 +258,48 @@ func compileCutter(zctx *resolver.Context, node ast.FunctionCall) (*expr.Cutter,
 	return expr.NewCutter(zctx, lhs, rhs)
 }
 
+func shaperOps(name string) expr.ShaperTransform {
+	switch name {
+	case "cast":
+		return expr.Cast
+	case "crop":
+		return expr.Crop
+	case "fill":
+		return expr.Fill
+	case "fit":
+		return expr.Crop | expr.Fill
+	case "order":
+		return expr.Order
+	case "shape":
+		return expr.Cast | expr.Crop | expr.Fill | expr.Order
+	default:
+		return 0
+	}
+
+}
+
+func isShaperFunc(name string) bool {
+	return shaperOps(name) != 0
+}
+
+func compileShaper(zctx *resolver.Context, node ast.FunctionCall) (*expr.Shaper, error) {
+	if len(node.Args) < 2 {
+		return nil, function.ErrTooFewArgs
+	}
+	if len(node.Args) > 2 {
+		return nil, function.ErrTooManyArgs
+	}
+	field, err := CompileExpr(zctx, node.Args[0])
+	if err != nil {
+		return nil, err
+	}
+	ev, err := CompileExpr(zctx, node.Args[1])
+	if err != nil {
+		return nil, err
+	}
+	return expr.NewShaper(zctx, field, ev, shaperOps(node.Function))
+}
+
 func compileCall(zctx *resolver.Context, node ast.FunctionCall) (expr.Evaluator, error) {
 	// For now, we special case cut and pick here.  We shuold generalize this
 	// as we will add many more stateful functions and also resolve this
@@ -274,6 +316,9 @@ func compileCall(zctx *resolver.Context, node ast.FunctionCall) (expr.Evaluator,
 	if node.Function == "pick" {
 		return compileCutter(zctx, node)
 	}
+	if isShaperFunc(node.Function) {
+		return compileShaper(zctx, node)
+	}
 	nargs := len(node.Args)
 	fn, err := function.New(zctx, node.Function, nargs)
 	if err != nil {
diff --git a/driver/parallel.go b/driver/parallel.go
index f7f4ca08a..2fbb8e20d 100644
--- a/driver/parallel.go
+++ b/driver/parallel.go
@@ -13,6 +13,7 @@ import (
 	"github.com/brimsec/zq/ppl/zqd/recruiter"
 	"github.com/brimsec/zq/proc"
 	"github.com/brimsec/zq/zbuf"
+	"go.uber.org/zap"
 )
 
 type parallelHead struct {
@@ -95,6 +96,7 @@ func (ph *parallelHead) Done() {
 
 type parallelGroup struct {
 	filter     SourceFilter
+	logger     *zap.Logger
 	mcfg       MultiConfig
 	msrc       MultiSource
 	once       sync.Once
@@ -212,6 +214,7 @@ func createParallelGroup(pctx *proc.Context, filter *compiler.Filter, msrc Multi
 			Filter: filter,
 			Span:   mcfg.Span,
 		},
+		logger:     mcfg.Logger,
 		mcfg:       mcfg,
 		msrc:       msrc,
 		pctx:       pctx,
diff --git a/expr/shaper.go b/expr/shaper.go
new file mode 100644
index 000000000..2e4548b55
--- /dev/null
+++ b/expr/shaper.go
@@ -0,0 +1,419 @@
+package expr
+
+import (
+	"fmt"
+
+	"github.com/brimsec/zq/zcode"
+	"github.com/brimsec/zq/zng"
+	"github.com/brimsec/zq/zng/resolver"
+	"github.com/brimsec/zq/zson"
+)
+
+// A ShaperTransform represents one of the different transforms that a
+// shaper can apply.
+type ShaperTransform int
+
+const (
+	Cast ShaperTransform = 1 << iota
+	Fill
+	Crop
+	Order
+)
+
+type step int
+
+const (
+	copyPrimitive step = iota // copy field fromIndex from input record
+	copyContainer
+	castPrimitive // cast field fromIndex from input record
+	null          // write null
+	record        // record into record below us
+)
+
+// A op is a recursive data structure encoding a series of
+// copy/cast steps to be carried out over an input record.
+type op struct {
+	op        step
+	fromIndex int
+	castTypes struct{ from, to zng.Type } // for op == castPrimitive
+	record    []op                        // for op == record
+}
+
+func (s *op) append(step op) {
+	s.record = append(s.record, step)
+}
+
+// create the op needed to build a record of type out from a
+// record of type in. The two types must be compatible, meaning that
+// the input type must be an unordered sub-tree of the input type
+// (where 'unordered' means that if the output type has record fields
+// [a b] and the input type has fields [b a] that is ok). It is also
+// ok for leaf primitive types to be different; if they are a casting
+// step is inserted.
+func createOp(in, out *zng.TypeRecord) (op, error) {
+	o := op{op: record}
+	for _, outCol := range out.Columns {
+		ind, ok := in.ColumnOfField(outCol.Name)
+		if !ok {
+			o.append(op{op: null})
+			continue
+		}
+
+		inCol := in.Columns[ind]
+
+		switch {
+		case inCol.Type.ID() == outCol.Type.ID():
+			if zng.IsContainerType(inCol.Type) {
+				o.append(op{fromIndex: ind, op: copyContainer})
+			} else {
+				o.append(op{fromIndex: ind, op: copyPrimitive})
+			}
+		case zng.IsRecordType(inCol.Type) && zng.IsRecordType(outCol.Type):
+			step, err := createOp(inCol.Type.(*zng.TypeRecord), outCol.Type.(*zng.TypeRecord))
+			if err != nil {
+				return op{}, err
+			}
+			step.fromIndex = ind
+			o.append(step)
+		case zng.IsPrimitiveType(inCol.Type) && zng.IsPrimitiveType(outCol.Type):
+			step := op{fromIndex: ind, op: castPrimitive, castTypes: struct{ from, to zng.Type }{inCol.Type, outCol.Type}}
+			o.append(step)
+		default:
+			return op{}, fmt.Errorf("createOp incompatible column types %s and %s\n", inCol.Type, outCol.Type)
+		}
+	}
+	return o, nil
+}
+
+func (s op) castPrimitive(in zcode.Bytes, b *zcode.Builder) {
+	pc := LookupPrimitiveCaster(s.castTypes.to)
+	v, err := pc(zng.Value{s.castTypes.from, in})
+	if err != nil {
+		b.AppendNull()
+		return
+	}
+	b.AppendPrimitive(v.Bytes)
+}
+
+func (s op) buildRecord(in zcode.Bytes, b *zcode.Builder) {
+	if s.op != record {
+		panic("bad op")
+	}
+	for _, step := range s.record {
+
+		switch step.op {
+		case null:
+			b.AppendNull()
+			continue
+		}
+		// Using getNthFromContainer means we iterate from the
+		// beginning of the record for each field. An
+		// optimization (for shapes that don't require field
+		// reordering) would be make direct use of a
+		// zcode.Iter along with keeping track of our
+		// position.
+		bytes, err := getNthFromContainer(in, uint(step.fromIndex))
+		if err != nil {
+			panic(err)
+		}
+
+		switch step.op {
+		case copyPrimitive:
+			b.AppendPrimitive(bytes)
+		case copyContainer:
+			b.AppendContainer(bytes)
+		case castPrimitive:
+			step.castPrimitive(bytes, b)
+		case record:
+			step.buildRecord(bytes, b)
+		}
+	}
+}
+
+// A shapeSpec is a per-input type ID "spec" that contains the output
+// type and the op to create an output record.
+type shapeSpec struct {
+	typ *zng.TypeRecord
+	op  op
+}
+
+type Shaper struct {
+	zctx       *resolver.Context
+	b          zcode.Builder
+	fieldExpr  Evaluator
+	typ        *zng.TypeRecord
+	shapeSpecs map[int]shapeSpec // map from type ID to shapeSpec
+	transforms ShaperTransform
+}
+
+// NewShaper returns a shaper that will shape the result of fieldExpr
+// to the provided typExpr. (typExpr should evaluate to a type value,
+// e.g. a value of type TypeType).
+func NewShaper(zctx *resolver.Context, fieldExpr, typExpr Evaluator, tf ShaperTransform) (*Shaper, error) {
+	lit, ok := typExpr.(*Literal)
+	if !ok {
+		return nil, fmt.Errorf("shaping functions (crop, fill, cast, order) take a literal as second parameter")
+	}
+
+	// Z doesn't yet have type value literals, so we accept a
+	// string literal and parse it. When type value lits are in,
+	// this only change will be to require zng.TypeType rather
+	// than zng.TypeString. Since a TypeType value holds the type
+	// as zson, parsing with the zson.TypeTable below will be
+	// identical.
+	//
+	// if lit.zv.Type != zng.TypeType {
+	if lit.zv.Type != zng.TypeString {
+		return nil, fmt.Errorf("shaper needs a type value as second parameter")
+	}
+	tt := zson.NewTypeTable(zctx)
+	shapeToType, err := tt.LookupType(string(lit.zv.Bytes))
+	if err != nil {
+		return nil, fmt.Errorf("shaper could not parse type value literal: %s", err)
+	}
+
+	recType, isRecord := shapeToType.(*zng.TypeRecord)
+	if !isRecord {
+		return nil, fmt.Errorf("shaper needs a record type value as second parameter")
+	}
+	return &Shaper{
+		zctx:       zctx,
+		fieldExpr:  fieldExpr,
+		typ:        recType,
+		shapeSpecs: make(map[int]shapeSpec),
+		transforms: tf,
+	}, nil
+}
+
+func (c *Shaper) Eval(in *zng.Record) (zng.Value, error) {
+	inVal, err := c.fieldExpr.Eval(in)
+	if err != nil {
+		return zng.Value{}, err
+	}
+	inType, ok := inVal.Type.(*zng.TypeRecord)
+	if !ok {
+		return inVal, nil
+	}
+	if _, ok := c.shapeSpecs[in.Type.ID()]; !ok {
+		spec, err := c.createShapeSpec(inType, c.typ)
+		if err != nil {
+			return zng.Value{}, err
+		}
+		c.shapeSpecs[in.Type.ID()] = spec
+	}
+	spec := c.shapeSpecs[in.Type.ID()]
+	if spec.typ.ID() == in.Type.ID() {
+		return inVal, nil
+	}
+
+	c.b.Reset()
+	spec.op.buildRecord(inVal.Bytes, &c.b)
+	return zng.Value{spec.typ, c.b.Bytes()}, nil
+}
+
+func (c *Shaper) createShapeSpec(inType, spec *zng.TypeRecord) (shapeSpec, error) {
+	var err error
+	typ := inType
+	if c.transforms&Cast > 0 {
+		typ, err = c.castRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	if c.transforms&Crop > 0 {
+		typ, err = c.cropRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	if c.transforms&Fill > 0 {
+		typ, err = c.fillRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	if c.transforms&Order > 0 {
+		typ, err = c.orderRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	op, err := createOp(inType, typ)
+	return shapeSpec{typ, op}, err
+}
+
+// cropRecordType applies a crop (as specified by the record type 'spec')
+// to a record type and returns the resulting record type.
+func (c *Shaper) cropRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
+	cols := make([]zng.Column, 0)
+	for _, inCol := range input.Columns {
+		ind, ok := spec.ColumnOfField(inCol.Name)
+		if !ok {
+			// 1. Field not in crop: drop.
+			continue
+		}
+
+		specCol := spec.Columns[ind]
+		switch {
+		case !zng.IsRecordType(inCol.Type):
+			// 2. Field is non-record in input: keep (regardless of crop record-ness)
+			cols = append(cols, inCol)
+		case zng.IsRecordType(specCol.Type):
+			// 3. Both records: recurse
+			out, err := c.cropRecordType(inCol.Type.(*zng.TypeRecord), specCol.Type.(*zng.TypeRecord))
+			if err != nil {
+				return nil, err
+			}
+			cols = append(cols, zng.Column{inCol.Name, out})
+		default:
+			// 4. record input but non-record in crop: keep crop
+			cols = append(cols, specCol)
+
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
+
+// orderRecordType applies a field order (as specified by the record type 'spec')
+// to a record type and returns the resulting record type.
+func (c *Shaper) orderRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
+	cols := make([]zng.Column, 0)
+	// Simple order algorithm creates a list with all specified
+	// 'order' fields present in input, followed by any other
+	// fields that were in input but not specified in order. Two
+	// examples:
+	//
+	// 1. (a b c d) order (c e b a) -> (c b a d)
+	//
+	// 2. (a b c d) order (d e c b) -> (d c b a)
+	//
+	// The second example with 'a' moving to the back suggests
+	// that we may want to use a different algorithm where
+	// unspecified fields "stay where they were". Specifically we
+	// might prefer that the result be (a d c b). We will learn
+	// with use, so starting with simpler algorithm for now.
+	//
+	for _, specCol := range spec.Columns {
+		if ind, ok := input.ColumnOfField(specCol.Name); ok {
+			inCol := input.Columns[ind]
+			if zng.IsRecordType(inCol.Type) && zng.IsRecordType(specCol.Type) {
+				if nested, err := c.orderRecordType(inCol.Type.(*zng.TypeRecord), specCol.Type.(*zng.TypeRecord)); err != nil {
+					return nil, err
+				} else {
+					cols = append(cols, zng.Column{specCol.Name, nested})
+				}
+				continue
+			}
+			cols = append(cols, inCol)
+		}
+	}
+	for _, inCol := range input.Columns {
+		if !spec.HasField(inCol.Name) {
+			cols = append(cols, inCol)
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
+
+// fillRecordType applies a fill (as specified by the record type 'spec')
+// to a record type and returns the resulting record type.
+func (c *Shaper) fillRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
+	cols := make([]zng.Column, 0)
+	names := make([]string, 0)
+	// Compute ordered list of fields. This takes all the input
+	// fields and adds the filled fields at the end. For
+	// standalone uses of 'fill()' (as opposed to shape() which
+	// includes order)', this might be surprising and it would
+	// likely be preferable to have the filled fields be woven
+	// into the input records. This can come later as needed.
+	for _, col := range input.Columns {
+		names = append(names, col.Name)
+	}
+	for _, specCol := range spec.Columns {
+		if !input.HasField(specCol.Name) {
+			names = append(names, specCol.Name)
+		}
+	}
+	for _, name := range names {
+		ind, _ := spec.ColumnOfField(name)
+		specCol := spec.Columns[ind]
+		if !input.HasField(name) {
+			// Field not in input: fill.
+			cols = append(cols, specCol)
+			continue
+		}
+		ind, _ = input.ColumnOfField(name)
+		inCol := input.Columns[ind]
+		_, inIsRec := inCol.Type.(*zng.TypeRecord)
+		_, specIsRec := specCol.Type.(*zng.TypeRecord)
+
+		switch {
+		case inIsRec && specIsRec:
+			col, err := c.fillRecordType(inCol.Type.(*zng.TypeRecord), specCol.Type.(*zng.TypeRecord))
+			if err != nil {
+				return nil, err
+			}
+			cols = append(cols, zng.Column{inCol.Name, col})
+
+		case !inIsRec && !specIsRec:
+			cols = append(cols, inCol)
+
+		case inIsRec && !specIsRec:
+			cols = append(cols, inCol)
+
+		case !inIsRec && specIsRec:
+			cols = append(cols, specCol)
+
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
+
+// castRecordType applies a cast (as specified by the record type 'spec')
+// to a record type and returns the resulting record type.
+func (c *Shaper) castRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
+	cols := make([]zng.Column, 0)
+
+	for _, inCol := range input.Columns {
+		// For each input column, check if we have a matching
+		// name in the cast spec.
+		ind, ok := spec.ColumnOfField(inCol.Name)
+		if !ok {
+			// 1. No match: output type unmodified.
+			cols = append(cols, inCol)
+			continue
+		}
+		specCol := spec.Columns[ind]
+
+		inRec, inIsRec := inCol.Type.(*zng.TypeRecord)
+		castRec, castIsRec := specCol.Type.(*zng.TypeRecord)
+
+		switch {
+		case inCol.Type.ID() == specCol.Type.ID():
+			// 2. Field has same type in cast: output type unmodified.
+			cols = append(cols, inCol)
+		case inIsRec && castIsRec:
+			// 3. Matching field is a record: recurse.
+			out, err := c.castRecordType(inRec, castRec)
+			if err != nil {
+				return nil, err
+			}
+			cols = append(cols, zng.Column{inCol.Name, out})
+		case zng.IsPrimitiveType(inCol.Type) && zng.IsPrimitiveType(specCol.Type):
+			// 4. Matching field is a primitive: output type is cast type.
+			if LookupPrimitiveCaster(specCol.Type) == nil {
+				return nil, fmt.Errorf("cast to %s not implemented", specCol.Type)
+			}
+			cols = append(cols, zng.Column{inCol.Name, specCol.Type})
+		default:
+			// 5. Non-castable type pair with at least one
+			// (non-record) container: output column is left
+			// unchanged.  Note that eventually, we should
+			// recognize and cast e.g. array[string] to array[ip].
+			// xxx before merge: file issue for this and mention
+			// it here.
+			cols = append(cols, inCol)
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
diff --git a/go.mod b/go.mod
index ff67603e8..83c8bfc0d 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,6 @@ require (
 	github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4
 	github.com/alexbrainman/ps v0.0.0-20171229230509-b3e1b4a15894
 	github.com/apache/thrift v0.0.0-20181112125854-24918abba929
-	github.com/auth0/go-jwt-middleware v0.0.0-20201030150249-d783b5c46b39
 	github.com/aws/aws-sdk-go v1.30.19
 	github.com/axiomhq/hyperloglog v0.0.0-20191112132149-a4c4c47bc57f
 	github.com/buger/jsonparser v0.0.0-20191004114745-ee4c978eae7e
diff --git a/go.sum b/go.sum
index 21511a2b1..f1526b1b3 100644
--- a/go.sum
+++ b/go.sum
@@ -52,8 +52,6 @@ github.com/alexbrainman/ps v0.0.0-20171229230509-b3e1b4a15894/go.mod h1:Wgrp3f69
 github.com/apache/arrow/go/arrow v0.0.0-20200601151325-b2287a20f230/go.mod h1:QNYViu/X0HXDHw7m3KXzWSVXIbfUvJqBFe6Gj8/pYA0=
 github.com/apache/thrift v0.0.0-20181112125854-24918abba929 h1:ubPe2yRkS6A/X37s0TVGfuN42NV2h0BlzWj0X76RoUw=
 github.com/apache/thrift v0.0.0-20181112125854-24918abba929/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
-github.com/auth0/go-jwt-middleware v0.0.0-20201030150249-d783b5c46b39 h1:FXGfTw25KQECao75qgpoCttPz8VTwKkQTi2L0iGh9Vk=
-github.com/auth0/go-jwt-middleware v0.0.0-20201030150249-d783b5c46b39/go.mod h1:mF0ip7kTEFtnhBJbd/gJe62US3jykNN+dcZoZakJCCA=
 github.com/aws/aws-sdk-go v1.17.7/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.30.19 h1:vRwsYgbUvC25Cb3oKXTyTYk3R5n1LRVk8zbvL4inWsc=
 github.com/aws/aws-sdk-go v1.30.19/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
@@ -82,7 +80,6 @@ github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58/go.mod h1:EOBUe0h
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
 github.com/cockroachdb/cockroach-go v0.0.0-20190925194419-606b3d062051/go.mod h1:XGLbWH/ujMcbPbhZq52Nv6UrCghb1yGn//133kEsvDk=
-github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0/go.mod h1:4Zcjuz89kmFXt9morQgcfYZAYZ5n8WHjt81YYWIwtTM=
 github.com/colinmarc/hdfs/v2 v2.1.1/go.mod h1:M3x+k8UKKmxtFu++uAZ0OtDU8jR3jnaZIAc6yK4Ue0c=
 github.com/containerd/containerd v1.4.0/go.mod h1:bC6axHOhabU15QhwfG7w5PipXdVtMXFTttgp+kVtyUA=
 github.com/containerd/containerd v1.4.1 h1:pASeJT3R3YyVn+94qEPk0SnU1OQ20Jd/T+SPKy9xehY=
@@ -125,7 +122,6 @@ github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2
 github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
-github.com/go-martini/martini v0.0.0-20170121215854-22fa46961aab/go.mod h1:/P9AEU963A2AYjv4d1V5eVL1CQbEJq6aCNHDDjibzu8=
 github.com/go-pg/pg/v10 v10.7.3 h1:oL/Hz5MJie/9epmwxlZjfReO+2wlLPOK6BeGb9I+XHk=
 github.com/go-pg/pg/v10 v10.7.3/go.mod h1:UsDYtA+ihbBNX1OeIvDejxkL4RXzL3wsZYoEv5NUEqM=
 github.com/go-pg/zerochecker v0.2.0 h1:pp7f72c3DobMWOb2ErtZsnrPaSvHd2W4o9//8HtF4mU=
@@ -207,9 +203,6 @@ github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
-github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00 h1:l5lAOZEym3oK3SQ2HBHWsJUfbNBiTXJDeW2QDxw9AQ0=
-github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg=
 github.com/gorilla/handlers v1.4.2/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/OAirnOIQ=
 github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
@@ -265,8 +258,6 @@ github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCV
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
-github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
-github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
 github.com/k0kubun/pp v2.3.0+incompatible/go.mod h1:GWse8YhT0p8pT4ir3ZgBbfZild3tgzSScAn6HmfYukg=
@@ -389,11 +380,6 @@ github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6Mwd
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
 github.com/sirupsen/logrus v1.7.0 h1:ShrD1U9pZB12TX0cVy0DtePoCH97K8EtX+mg7ZARUtM=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
-github.com/smartystreets/assertions v1.1.0 h1:MkTeG1DMwsrdH7QtLXy5W+fUxWq+vmb6cLmyJ7aRtF0=
-github.com/smartystreets/assertions v1.1.0/go.mod h1:tcbTF8ujkAEcZ8TElKY+i30BzYlVhC/LOxJk7iOWnoo=
-github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
-github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/snowflakedb/glog v0.0.0-20180824191149-f5055e6f21ce/go.mod h1:EB/w24pR5VKI60ecFnKqXzxX3dOorz1rnVicQTQrGM0=
 github.com/snowflakedb/gosnowflake v1.3.5/go.mod h1:13Ky+lxzIm3VqNDZJdyvu9MCGy+WgRdYFdXp96UcLZU=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
@@ -411,8 +397,6 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc h1:9lRDQMhESg+zvGYmW5DyG0UqvY96Bu5QYsTLvCHdrgo=
 github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc/go.mod h1:bciPuU6GHm1iF1pBvUfxfsH0Wmnc2VbpgvbI9ZWuIRs=
-github.com/urfave/negroni v1.0.0 h1:kIimOitoypq34K7TG7DUaJ9kq/N4Ofuwi1sjz0KipXc=
-github.com/urfave/negroni v1.0.0/go.mod h1:Meg73S6kFm/4PpbYdq35yYWoCZ9mS/YSx+lKnmiohz4=
 github.com/vmihailenco/bufpool v0.1.11 h1:gOq2WmBrq0i2yW5QJ16ykccQ4wH9UyEsgLm6czKAd94=
 github.com/vmihailenco/bufpool v0.1.11/go.mod h1:AFf/MOy3l2CFTKbxwt0mp2MwnqjNEs5H/UxrkA5jxTQ=
 github.com/vmihailenco/msgpack/v4 v4.3.11/go.mod h1:gborTTJjAo/GWTqqRjrLCn9pgNN+NXzzngzBKDPIqw4=
@@ -630,7 +614,6 @@ golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190312151545-0bb0c0a6e846/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190312170243-e65039ee4138/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
-golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190425163242-31fd60d6bfdc/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190506145303-2d16b83fe98c/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
diff --git a/k8s/demo.sh b/k8s/demo.sh
index 78fc0103f..177ded1ec 100755
--- a/k8s/demo.sh
+++ b/k8s/demo.sh
@@ -1,5 +1,6 @@
 #!/bin/bash
 # This is designed to be run from the zq root directory
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 
 make install
 helm uninstall zsrv
@@ -11,7 +12,7 @@ make helm-install
 
 sleep 15 # wait for services
 
-./k8s/zqd-port.sh
+$DIR/zqd-port.sh
 
 aws s3 rm --recursive s3://brim-scratch/mark/sp-m1 || :
 zapi new -k archivestore -d s3://brim-scratch/mark/sp-m1 -thresh 5MB sp-m1
diff --git a/k8s/zqd-port.sh b/k8s/zqd-port.sh
index 2062aa58c..930b694ed 100755
--- a/k8s/zqd-port.sh
+++ b/k8s/zqd-port.sh
@@ -3,8 +3,7 @@ set -x #echo on
 # start by killing existing port-forwards
 kill $(ps aux | grep '[p]ort-forward' | awk '{print $2}')
 sleep 1 # wait a second for the processes to get the signal
-kubectl port-forward svc/zsrv-recruiter 8020:9867 &
-kubectl port-forward svc/zsrv-root 9867:9867 &
-sleep 2 # wait longer for port-forwards to complete
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+source $DIR/zsrv-port-forward.sh
 curl http://localhost:8020/recruiter/listfree
 zapi ls 
diff --git a/k8s/zsrv-port-forward.sh b/k8s/zsrv-port-forward.sh
new file mode 100755
index 000000000..ae5a1bbda
--- /dev/null
+++ b/k8s/zsrv-port-forward.sh
@@ -0,0 +1,14 @@
+kubectl port-forward svc/zsrv-recruiter 8020:9867 &
+kubectl port-forward svc/zsrv-root 9867:9867 &
+for i in {1..5}; do
+  if nc -z localhost 8020; then
+    break
+  fi
+  sleep 1
+done
+for i in {1..5}; do
+  if nc -z localhost 9867; then
+    break
+  fi
+  sleep 1
+done
diff --git a/package.json b/package.json
index b69d08a81..80fe1c42d 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "zq",
-  "version": "0.27.1-dev",
+  "version": "0.28.0-dev",
   "scripts": {
       "prepack": "node npm/build"
   },
@@ -13,6 +13,6 @@
   },
   "brimDependencies": {
       "suricataTag": "v5.0.3-brim1",
-      "zeekTag": "v3.2.1-brim8"
+      "zeekTag": "v3.2.1-brim9"
   }
 }
diff --git a/ppl/archive/chunk/chunk.go b/ppl/archive/chunk/chunk.go
index 05e294877..73cce744a 100644
--- a/ppl/archive/chunk/chunk.go
+++ b/ppl/archive/chunk/chunk.go
@@ -19,9 +19,9 @@ type FileKind string
 
 const (
 	FileKindUnknown  FileKind = ""
-	FileKindData              = "d"
-	FileKindMetadata          = "m"
-	FileKindSeek              = "ts"
+	FileKindData     FileKind = "d"
+	FileKindMetadata FileKind = "m"
+	FileKindSeek     FileKind = "ts"
 )
 
 var fileRegex = regexp.MustCompile(`(d|m)-([0-9A-Za-z]{27}).zng$`)
diff --git a/ppl/archive/multisource.go b/ppl/archive/multisource.go
index edbccfb21..f91037544 100644
--- a/ppl/archive/multisource.go
+++ b/ppl/archive/multisource.go
@@ -21,6 +21,7 @@ import (
 	"github.com/brimsec/zq/zng/resolver"
 	"github.com/brimsec/zq/zqe"
 	"go.uber.org/multierr"
+	"golang.org/x/sync/errgroup"
 )
 
 type MultiSource interface {
@@ -105,14 +106,37 @@ func (m *spanMultiSource) OrderInfo() (field.Static, bool) {
 }
 
 func (m *spanMultiSource) SendSources(ctx context.Context, span nano.Span, srcChan chan driver.Source) error {
-	return SpanWalk(ctx, m.ark, span, func(si SpanInfo) error {
-		select {
-		case srcChan <- &spanSource{ark: m.ark, spanInfo: si, stats: m.stats}:
-			return nil
-		case <-ctx.Done():
-			return ctx.Err()
+	// We keep a channel of []SpanInfos filled to reduce the time
+	// query workers are waiting for the next driver.Source.
+	const tsDirPreFetch = 10
+	sinfosChan := make(chan []SpanInfo, tsDirPreFetch)
+	g, ctx := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		err := tsDirVisit(ctx, m.ark, span, func(_ tsDir, chunks []chunk.Chunk) error {
+			sinfos := mergeChunksToSpans(chunks, m.ark.DataOrder, span)
+			select {
+			case sinfosChan <- sinfos:
+				return nil
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		})
+		close(sinfosChan)
+		return err
+	})
+	g.Go(func() error {
+		for sinfos := range sinfosChan {
+			for _, si := range sinfos {
+				select {
+				case srcChan <- &spanSource{ark: m.ark, spanInfo: si, stats: m.stats}:
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			}
 		}
+		return nil
 	})
+	return g.Wait()
 }
 
 func (m *spanMultiSource) SourceFromRequest(ctx context.Context, req *api.WorkerChunkRequest) (driver.Source, error) {
diff --git a/ppl/cmd/gentoken/gentoken.go b/ppl/cmd/gentoken/gentoken.go
new file mode 100644
index 000000000..3f9f72fe0
--- /dev/null
+++ b/ppl/cmd/gentoken/gentoken.go
@@ -0,0 +1,65 @@
+package main
+
+import (
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/brimsec/zq/ppl/zqd/auth"
+	"github.com/mccanne/charm"
+)
+
+var CLI = &charm.Spec{
+	Name:  "gentoken",
+	Usage: "gentoken",
+	Short: "generate access token to test zqd auth",
+	New:   New,
+}
+
+type Command struct {
+	charm.Command
+
+	domain         string
+	expiration     time.Duration
+	privateKeyFile string
+	keyID          string
+	tenantID       string
+	userID         string
+}
+
+func New(_ charm.Command, fs *flag.FlagSet) (charm.Command, error) {
+	c := &Command{}
+	fs.StringVar(&c.domain, "domain", "", "domain to use to generate token issuer")
+	fs.DurationVar(&c.expiration, "expiration", 4*time.Hour, "expiry duration for generated token")
+	fs.StringVar(&c.privateKeyFile, "privatekeyfile", "", "path of file containing private key (required)")
+	fs.StringVar(&c.keyID, "keyid", "", "key identifier")
+	fs.StringVar(&c.tenantID, "tenantid", "", "tenant ID claim in generated token")
+	fs.StringVar(&c.userID, "userid", "", "user ID claim in generated token")
+	return c, nil
+}
+
+func (c *Command) Run(args []string) error {
+	if len(args) > 0 {
+		return errors.New("gentoken takes no arguments")
+	}
+	if c.privateKeyFile == "" {
+		return errors.New("must specify a keyfile")
+	}
+	token, err := auth.GenerateAccessToken(c.keyID, c.privateKeyFile, c.expiration, c.domain, auth.TenantID(c.tenantID), auth.UserID(c.userID))
+	if err != nil {
+		return fmt.Errorf("GenerateAccessToken failed: %w", err)
+	}
+	fmt.Println(token)
+	return nil
+}
+
+func main() {
+	CLI.Add(charm.Help)
+	_, err := CLI.ExecRoot(os.Args[1:])
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "%s\n", err)
+		os.Exit(1)
+	}
+}
diff --git a/ppl/zqd/apiserver/manager.go b/ppl/zqd/apiserver/manager.go
index 0eb1233dc..09a39900f 100644
--- a/ppl/zqd/apiserver/manager.go
+++ b/ppl/zqd/apiserver/manager.go
@@ -9,6 +9,7 @@ import (
 	"github.com/brimsec/zq/api"
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/nano"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/brimsec/zq/ppl/zqd/db"
 	"github.com/brimsec/zq/ppl/zqd/db/schema"
 	"github.com/brimsec/zq/ppl/zqd/pcapstorage"
@@ -69,7 +70,7 @@ func (m *Manager) Shutdown() {
 }
 
 func (m *Manager) spawnAlphaMigrations(ctx context.Context) {
-	rows, err := m.db.ListSpaces(ctx)
+	rows, err := m.db.ListSpaces(ctx, auth.AnonymousTenantID)
 	if err != nil {
 		return
 	}
@@ -128,11 +129,13 @@ func (m *Manager) CreateSpace(ctx context.Context, req api.SpacePostRequest) (ap
 		return api.Space{}, zqe.E(zqe.Invalid, "cannot create file storage space on non-file backed data path")
 	}
 
+	ident := auth.IdentityFromContext(ctx)
 	row := schema.SpaceRow{
-		ID:      id,
-		Name:    req.Name,
-		DataURI: datapath,
-		Storage: storecfg,
+		ID:       id,
+		Name:     req.Name,
+		DataURI:  datapath,
+		Storage:  storecfg,
+		TenantID: ident.TenantID,
 	}
 
 	name := row.Name
@@ -177,12 +180,14 @@ func (m *Manager) CreateSubspace(ctx context.Context, parentID api.SpaceID, req
 	if _, err := m.getStorage(ctx, id, parent.DataURI, cfg); err != nil {
 		return api.Space{}, zqe.ErrInvalid("invalid subspace storage config: %w", err)
 	}
+	ident := auth.IdentityFromContext(ctx)
 	row := schema.SpaceRow{
 		ID:       id,
 		ParentID: parentID,
 		Name:     req.Name,
 		DataURI:  parent.DataURI,
 		Storage:  cfg,
+		TenantID: ident.TenantID,
 	}
 	if err := m.db.CreateSubspace(ctx, row); err != nil {
 		return api.Space{}, err
@@ -192,7 +197,7 @@ func (m *Manager) CreateSubspace(ctx context.Context, parentID api.SpaceID, req
 }
 
 func (m *Manager) GetStorage(ctx context.Context, id api.SpaceID) (storage.Storage, error) {
-	sr, err := m.db.GetSpace(ctx, id)
+	sr, err := m.getSpacePermCheck(ctx, id)
 	if err != nil {
 		return nil, err
 	}
@@ -200,7 +205,7 @@ func (m *Manager) GetStorage(ctx context.Context, id api.SpaceID) (storage.Stora
 }
 
 func (m *Manager) GetPcapStorage(ctx context.Context, id api.SpaceID) (*pcapstorage.Store, error) {
-	sr, err := m.db.GetSpace(ctx, id)
+	sr, err := m.getSpacePermCheck(ctx, id)
 	if err != nil {
 		return nil, err
 	}
@@ -306,8 +311,20 @@ func (m *Manager) rowToSpaceInfo(ctx context.Context, sr schema.SpaceRow) (api.S
 	return spaceInfo, nil
 }
 
-func (m *Manager) GetSpace(ctx context.Context, id api.SpaceID) (api.SpaceInfo, error) {
+func (m *Manager) getSpacePermCheck(ctx context.Context, id api.SpaceID) (schema.SpaceRow, error) {
 	sr, err := m.db.GetSpace(ctx, id)
+	if err != nil {
+		return schema.SpaceRow{}, err
+	}
+	ident := auth.IdentityFromContext(ctx)
+	if sr.TenantID != ident.TenantID {
+		return schema.SpaceRow{}, zqe.ErrForbidden()
+	}
+	return sr, nil
+}
+
+func (m *Manager) GetSpace(ctx context.Context, id api.SpaceID) (api.SpaceInfo, error) {
+	sr, err := m.getSpacePermCheck(ctx, id)
 	if err != nil {
 		return api.SpaceInfo{}, err
 	}
@@ -315,7 +332,8 @@ func (m *Manager) GetSpace(ctx context.Context, id api.SpaceID) (api.SpaceInfo,
 }
 
 func (m *Manager) ListSpaces(ctx context.Context) ([]api.Space, error) {
-	rows, err := m.db.ListSpaces(ctx)
+	ident := auth.IdentityFromContext(ctx)
+	rows, err := m.db.ListSpaces(ctx, ident.TenantID)
 	if err != nil {
 		return nil, err
 	}
@@ -333,11 +351,14 @@ func (m *Manager) UpdateSpaceName(ctx context.Context, id api.SpaceID, name stri
 	if !schema.ValidSpaceName(name) {
 		return zqe.E(zqe.Invalid, "name may not contain '/' or non-printable characters")
 	}
+	if _, err := m.getSpacePermCheck(ctx, id); err != nil {
+		return err
+	}
 	return m.db.UpdateSpaceName(ctx, id, name)
 }
 
 func (m *Manager) DeleteSpace(ctx context.Context, id api.SpaceID) error {
-	sr, err := m.db.GetSpace(ctx, id)
+	sr, err := m.getSpacePermCheck(ctx, id)
 	if err != nil {
 		return err
 	}
diff --git a/ppl/zqd/auth.go b/ppl/zqd/auth.go
index f9bb2eb3b..a47e67086 100644
--- a/ppl/zqd/auth.go
+++ b/ppl/zqd/auth.go
@@ -2,35 +2,17 @@ package zqd
 
 import (
 	"context"
-	"crypto/rsa"
 	"errors"
 	"flag"
-	"fmt"
 	"net/http"
-	"net/url"
-	"time"
 
-	jwtmiddleware "github.com/auth0/go-jwt-middleware"
 	"github.com/brimsec/zq/api"
-	"github.com/brimsec/zq/pkg/fs"
-	"github.com/dgrijalva/jwt-go"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"go.uber.org/zap"
 )
 
-const (
-	// AudienceClaimValue is the value of the "aud" standard claim that clients
-	// should use when requesting access tokens for this api.
-	// Though formatted as a URL, it does not need to be a reachable location.
-	AudienceClaimValue = "https://app.brimsecurity.com"
-
-	// These are the namespaced custom claims we expect on any JWT
-	// access token.
-	TenantIDClaim = AudienceClaimValue + "/tenant_id"
-	UserIDClaim   = AudienceClaimValue + "/user_id"
-)
-
 type AuthConfig struct {
 	Enabled  bool
 	JWKSPath string
@@ -49,101 +31,57 @@ func (c *AuthConfig) SetFlags(fs *flag.FlagSet) {
 	fs.StringVar(&c.JWKSPath, "auth.jwkspath", "", "path to JSON Web Key Set file")
 }
 
-func (c *AuthConfig) Validate() (*url.URL, error) {
-	if !c.Enabled {
-		return nil, nil
-	}
-	if c.ClientID == "" || c.Domain == "" || c.JWKSPath == "" {
-		return nil, errors.New("auth.clientid, auth.domain, and auth.jwkspath must be set when auth enabled")
-	}
-	u, err := url.Parse(c.Domain)
-	if err != nil {
-		return nil, fmt.Errorf("bad auth.domain URL: %w", err)
-	}
-	return u, nil
-}
-
 type Auth0Authenticator struct {
-	checker        *jwtmiddleware.JWTMiddleware
+	logger         *zap.Logger
 	methodResponse api.AuthMethodResponse
+	unauthorized   prometheus.Counter
+	validator      *auth.TokenValidator
 }
 
-// newAuthenticator returns an Auth0Authenticator that checks for a JWT signed
+// NewAuthenticator returns an Auth0Authenticator that checks for a JWT signed
 // by a key referenced in the JWKS file, has the required audience and issuer
 // claims, and contains claims for a brim tenant and user id.
-func newAuthenticator(ctx context.Context, logger *zap.Logger, registerer prometheus.Registerer, config AuthConfig) (*Auth0Authenticator, error) {
-	domainURL, err := config.Validate()
-	if err != nil {
-		return nil, err
+func NewAuthenticator(ctx context.Context, logger *zap.Logger, registerer prometheus.Registerer, config AuthConfig) (*Auth0Authenticator, error) {
+	if config.ClientID == "" || config.Domain == "" || config.JWKSPath == "" {
+		return nil, errors.New("auth.clientid, auth.domain, and auth.jwkspath must be set when auth enabled")
 	}
-	// Auth0 issuer is always the domain URL with trailing "/".
-	// https://auth0.com/docs/tokens/access-tokens/get-access-tokens#custom-domains-and-the-management-api
-	expectedIssuer := domainURL.String() + "/"
-	keys, err := loadPublicKeys(config.JWKSPath)
+	validator, err := auth.NewTokenValidator(config.Domain, config.JWKSPath)
 	if err != nil {
-		return nil, fmt.Errorf("failed to load JWKS file: %w", err)
+		return nil, err
 	}
 	unauthorized := promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 		Name: "request_errors_unauthorized_total",
 		Help: "Number of request errors due to bad or missing authorization.",
 	})
-	checker := jwtmiddleware.New(jwtmiddleware.Options{
-		ValidationKeyGetter: func(token *jwt.Token) (interface{}, error) {
-			claims := token.Claims.(jwt.MapClaims)
-			// jwt-go (called from jwtmiddleware) verifies any expiry claim, but
-			// will not fail if the expiry claim is missing. The call here with
-			// req=true ensures that the claim is both present and valid.
-			if !claims.VerifyExpiresAt(time.Now().Unix(), true) {
-				return token, errors.New("invalid expiration")
-			}
-			if !claims.VerifyIssuer(expectedIssuer, true) {
-				return token, errors.New("invalid issuer")
-			}
-			if !verifyAPIAudience(claims) {
-				return token, errors.New("invalid audience")
-			}
-			if tid, _ := claims[TenantIDClaim].(string); tid == "" {
-				return token, errors.New("missing tenant id")
-			}
-			if uid, _ := claims[UserIDClaim].(string); uid == "" {
-				return token, errors.New("missing user id")
-			}
-			tokenKeyID, _ := token.Header["kid"].(string)
-			key, ok := keys[tokenKeyID]
-			if !ok {
-				return token, errors.New("unknown token key id")
-			}
-			return key, nil
-		},
-		UserProperty: authTokenContextValue,
-		ErrorHandler: func(w http.ResponseWriter, r *http.Request, errstr string) {
-			unauthorized.Inc()
-			logger.Info("unauthorized request",
-				zap.String("request_id", api.RequestIDFromContext(r.Context())),
-				zap.String("error", errstr))
-			http.Error(w, errstr, http.StatusUnauthorized)
-		},
-		SigningMethod: jwt.SigningMethodRS256,
-	})
 	return &Auth0Authenticator{
-		checker: checker,
+		logger: logger.Named("auth"),
 		methodResponse: api.AuthMethodResponse{
 			Kind: api.AuthMethodAuth0,
 			Auth0: &api.AuthMethodAuth0Details{
-				Audience: AudienceClaimValue,
+				Audience: auth.AudienceClaimValue,
 				Domain:   config.Domain,
 				ClientID: config.ClientID,
 			},
 		},
+		unauthorized: unauthorized,
+		validator:    validator,
 	}, nil
 }
 
 func (a *Auth0Authenticator) Middleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if a.checker.CheckJWT(w, r) != nil {
-			// response sent by jwtmiddleware.Options.ErrorHandler
+		token, ident, err := a.validator.ValidateRequest(r)
+		if err != nil {
+			a.unauthorized.Inc()
+			a.logger.Info("Unauthorized request",
+				zap.String("request_id", api.RequestIDFromContext(r.Context())),
+				zap.Error(err))
+			http.Error(w, err.Error(), http.StatusUnauthorized)
 			return
 		}
+		ctx := auth.ContextWithAuthToken(r.Context(), token)
+		ctx = auth.ContextWithIdentity(ctx, ident)
+		r = r.WithContext(ctx)
 		next.ServeHTTP(w, r)
 	})
 }
@@ -151,74 +89,3 @@ func (a *Auth0Authenticator) Middleware(next http.Handler) http.Handler {
 func (a *Auth0Authenticator) MethodResponse() api.AuthMethodResponse {
 	return a.methodResponse
 }
-
-func verifyAPIAudience(claims jwt.MapClaims) bool {
-	// Audience claim may either be a string, or a slice of interfaces that are
-	// strings.
-	// https://auth0.com/docs/tokens/access-tokens/get-access-tokens#multiple-audiences
-	if str, ok := claims["aud"].(string); ok {
-		return str == AudienceClaimValue
-	}
-	if arr, ok := claims["aud"].([]interface{}); ok {
-		for _, a := range arr {
-			s, _ := a.(string)
-			if s == AudienceClaimValue {
-				return true
-			}
-		}
-	}
-	return false
-}
-
-// authTokenContextValue is the string the jwtmiddleware library uses to
-// store the validated & parsed JWT in a request's context.
-const authTokenContextValue = "zqd-core-auth-token"
-
-type Identity struct {
-	TenantID string
-	UserID   string
-}
-
-func IdentifyFromContext(ctx context.Context) (Identity, bool) {
-	var token *jwt.Token
-	if token = ctx.Value(authTokenContextValue).(*jwt.Token); token == nil {
-		return Identity{}, false
-	}
-	mc := token.Claims.(jwt.MapClaims)
-	tid := mc[TenantIDClaim].(string)
-	uid := mc[UserIDClaim].(string)
-	return Identity{TenantID: tid, UserID: uid}, true
-}
-
-// jwks matches the format of a JSON Web Key Set file:
-// https://auth0.com/docs/tokens/json-web-tokens/json-web-key-sets
-type jwks struct {
-	Keys []struct {
-		Kty string   `json:"kty"`
-		Kid string   `json:"kid"`
-		Use string   `json:"use"`
-		N   string   `json:"n"`
-		E   string   `json:"e"`
-		X5c []string `json:"x5c"`
-	} `json:"keys"`
-}
-
-func loadPublicKeys(jwkspath string) (map[string]*rsa.PublicKey, error) {
-	var jwks jwks
-	if err := fs.UnmarshalJSONFile(jwkspath, &jwks); err != nil {
-		return nil, err
-	}
-	keys := make(map[string]*rsa.PublicKey)
-	for _, jwk := range jwks.Keys {
-		if len(jwk.X5c) == 0 {
-			continue
-		}
-		cert := "-----BEGIN CERTIFICATE-----\n" + jwk.X5c[0] + "\n-----END CERTIFICATE-----"
-		public, err := jwt.ParseRSAPublicKeyFromPEM([]byte(cert))
-		if err != nil {
-			return nil, err
-		}
-		keys[jwk.Kid] = public
-	}
-	return keys, nil
-}
diff --git a/ppl/zqd/auth/auth.go b/ppl/zqd/auth/auth.go
new file mode 100644
index 000000000..b83b660fe
--- /dev/null
+++ b/ppl/zqd/auth/auth.go
@@ -0,0 +1,43 @@
+package auth
+
+import (
+	"context"
+)
+
+type TenantID string
+type UserID string
+
+const (
+	AnonymousTenantID TenantID = "tenant_000000000000000000000000001"
+	AnonymousUserID   UserID   = "user_000000000000000000000000001"
+)
+
+type Identity struct {
+	TenantID TenantID
+	UserID   UserID
+}
+
+type identityKey struct{}
+
+func IdentityFromContext(ctx context.Context) Identity {
+	ident, ok := ctx.Value(identityKey{}).(Identity)
+	if !ok {
+		return Identity{TenantID: AnonymousTenantID, UserID: AnonymousUserID}
+	}
+	return ident
+}
+
+func ContextWithIdentity(ctx context.Context, ident Identity) context.Context {
+	return context.WithValue(ctx, identityKey{}, ident)
+}
+
+type authTokenKey struct{}
+
+func AuthTokenFromContext(ctx context.Context) (string, bool) {
+	token, ok := ctx.Value(authTokenKey{}).(string)
+	return token, ok
+}
+
+func ContextWithAuthToken(ctx context.Context, token string) context.Context {
+	return context.WithValue(ctx, authTokenKey{}, token)
+}
diff --git a/ppl/zqd/auth/generator.go b/ppl/zqd/auth/generator.go
new file mode 100644
index 000000000..92d0f198b
--- /dev/null
+++ b/ppl/zqd/auth/generator.go
@@ -0,0 +1,46 @@
+package auth
+
+import (
+	"crypto/rsa"
+	"fmt"
+	"io/ioutil"
+	"net/url"
+	"time"
+
+	"github.com/dgrijalva/jwt-go"
+)
+
+func loadPrivateKey(keyFile string) (*rsa.PrivateKey, error) {
+	b, err := ioutil.ReadFile(keyFile)
+	if err != nil {
+		return nil, err
+	}
+	return jwt.ParseRSAPrivateKeyFromPEM(b)
+}
+
+func makeToken(keyID string, keyFile string, claims jwt.MapClaims) (string, error) {
+	privateKey, err := loadPrivateKey(keyFile)
+	if err != nil {
+		return "", err
+	}
+	token := jwt.New(jwt.SigningMethodRS256)
+	token.Claims = claims
+	token.Header["kid"] = keyID
+	return token.SignedString(privateKey)
+}
+
+// GenerateAccessToken creates a JWT in string format with the expected audience,
+// issuer, and claims to pass zqd authentication checks.
+func GenerateAccessToken(keyID string, privateKeyFile string, expiration time.Duration, domain string, tenantID TenantID, userID UserID) (string, error) {
+	dstr, err := url.Parse(domain)
+	if err != nil {
+		return "", fmt.Errorf("bad domain URL: %w", err)
+	}
+	return makeToken(keyID, privateKeyFile, jwt.MapClaims{
+		"aud":         AudienceClaimValue,
+		"exp":         time.Now().Add(expiration).Unix(),
+		"iss":         dstr.String() + "/",
+		TenantIDClaim: string(tenantID),
+		UserIDClaim:   string(userID),
+	})
+}
diff --git a/ppl/zqd/auth/validator.go b/ppl/zqd/auth/validator.go
new file mode 100644
index 000000000..dc5e1db6e
--- /dev/null
+++ b/ppl/zqd/auth/validator.go
@@ -0,0 +1,168 @@
+package auth
+
+import (
+	"crypto/rsa"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/brimsec/zq/pkg/fs"
+	"github.com/brimsec/zq/zqe"
+	"github.com/dgrijalva/jwt-go"
+)
+
+const (
+	// AudienceClaimValue is the value of the "aud" standard claim that clients
+	// should use when requesting access tokens for this api.
+	// Though formatted as a URL, it does not need to be a reachable location.
+	AudienceClaimValue = "https://app.brimsecurity.com"
+
+	// These are the namespaced custom claims we expect on any JWT
+	// access token.
+	TenantIDClaim = AudienceClaimValue + "/tenant_id"
+	UserIDClaim   = AudienceClaimValue + "/user_id"
+)
+
+type TokenValidator struct {
+	keyGetter      jwt.Keyfunc
+	expectedIssuer string
+}
+
+func NewTokenValidator(domain, jwksPath string) (*TokenValidator, error) {
+	domainURL, err := url.Parse(domain)
+	if err != nil {
+		return nil, fmt.Errorf("bad auth.domain URL: %w", err)
+	}
+	keys, err := loadPublicKeys(jwksPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load JWKS file: %w", err)
+	}
+	// Auth0 issuer is always the domain URL with trailing "/".
+	// https://auth0.com/docs/tokens/access-tokens/get-access-tokens#custom-domains-and-the-management-api
+	expectedIssuer := domainURL.String() + "/"
+	keyGetter := func(token *jwt.Token) (interface{}, error) {
+		tokenKeyID, _ := token.Header["kid"].(string)
+		key, ok := keys[tokenKeyID]
+		if !ok {
+			return token, errors.New("unknown token key id")
+		}
+		return key, nil
+	}
+	return &TokenValidator{
+		expectedIssuer: expectedIssuer,
+		keyGetter:      keyGetter,
+	}, nil
+}
+
+func getBearerToken(r *http.Request) string {
+	hdr := r.Header.Get("Authorization")
+	if hdr == "" {
+		return ""
+	}
+	s := strings.Fields(hdr)
+	if len(s) != 2 || strings.ToLower(s[0]) != "bearer" {
+		return ""
+	}
+	return s[1]
+}
+
+func (v *TokenValidator) ValidateRequest(r *http.Request) (string, Identity, error) {
+	token := getBearerToken(r)
+	ident, err := v.Validate(token)
+	if err != nil {
+		return "", Identity{}, err
+	}
+	return token, ident, nil
+}
+
+func (v *TokenValidator) Validate(token string) (Identity, error) {
+	if token == "" {
+		return Identity{}, zqe.ErrNoCredentials()
+	}
+	parsed, err := jwt.Parse(token, v.keyGetter)
+	if err != nil || !parsed.Valid {
+		return Identity{}, zqe.ErrNoCredentials("invalid token")
+	}
+	if parsed.Header["alg"] != jwt.SigningMethodRS256.Alg() {
+		return Identity{}, zqe.ErrNoCredentials("invalid signing method")
+	}
+	claims := parsed.Claims.(jwt.MapClaims)
+	// jwt-go verifies any expiry claim, but will not fail if the expiry claim
+	// is missing. The call here with req=true ensures that the claim is both
+	// present and valid.
+	if !claims.VerifyExpiresAt(time.Now().Unix(), true) {
+		return Identity{}, zqe.ErrNoCredentials("invalid expiration")
+	}
+	if !claims.VerifyIssuer(v.expectedIssuer, true) {
+		return Identity{}, zqe.ErrNoCredentials("invalid issuer")
+	}
+	if !verifyAPIAudience(claims) {
+		return Identity{}, zqe.ErrNoCredentials("invalid audience")
+	}
+	tid, _ := claims[TenantIDClaim].(string)
+	if tid == "" || TenantID(tid) == AnonymousTenantID {
+		return Identity{}, zqe.ErrNoCredentials("invalid tenant id")
+	}
+	uid, _ := claims[UserIDClaim].(string)
+	if uid == "" || UserID(uid) == AnonymousUserID {
+		return Identity{}, zqe.ErrNoCredentials("invalid user id")
+	}
+	return Identity{
+		TenantID: TenantID(tid),
+		UserID:   UserID(uid),
+	}, nil
+}
+
+func verifyAPIAudience(claims jwt.MapClaims) bool {
+	// Audience claim may either be a string, or a slice of interfaces that are
+	// strings.
+	// https://auth0.com/docs/tokens/access-tokens/get-access-tokens#multiple-audiences
+	if str, ok := claims["aud"].(string); ok {
+		return str == AudienceClaimValue
+	}
+	if arr, ok := claims["aud"].([]interface{}); ok {
+		for _, a := range arr {
+			s, _ := a.(string)
+			if s == AudienceClaimValue {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// jwks matches the format of a JSON Web Key Set file:
+// https://auth0.com/docs/tokens/json-web-tokens/json-web-key-sets
+type jwks struct {
+	Keys []struct {
+		Kty string   `json:"kty"`
+		Kid string   `json:"kid"`
+		Use string   `json:"use"`
+		N   string   `json:"n"`
+		E   string   `json:"e"`
+		X5c []string `json:"x5c"`
+	} `json:"keys"`
+}
+
+func loadPublicKeys(jwkspath string) (map[string]*rsa.PublicKey, error) {
+	var jwks jwks
+	if err := fs.UnmarshalJSONFile(jwkspath, &jwks); err != nil {
+		return nil, err
+	}
+	keys := make(map[string]*rsa.PublicKey)
+	for _, jwk := range jwks.Keys {
+		if len(jwk.X5c) == 0 {
+			continue
+		}
+		cert := "-----BEGIN CERTIFICATE-----\n" + jwk.X5c[0] + "\n-----END CERTIFICATE-----"
+		public, err := jwt.ParseRSAPublicKeyFromPEM([]byte(cert))
+		if err != nil {
+			return nil, err
+		}
+		keys[jwk.Kid] = public
+	}
+	return keys, nil
+}
diff --git a/ppl/zqd/core.go b/ppl/zqd/core.go
index ed1773ace..87dba532c 100644
--- a/ppl/zqd/core.go
+++ b/ppl/zqd/core.go
@@ -49,10 +49,6 @@ type Config struct {
 	Zeek     pcapanalyzer.Launcher
 }
 
-type middleware interface {
-	Middleware(next http.Handler) http.Handler
-}
-
 type Core struct {
 	auth       *Auth0Authenticator
 	db         db.DB
@@ -81,10 +77,10 @@ func NewCore(ctx context.Context, conf Config) (*Core, error) {
 	registry := prometheus.NewRegistry()
 	registry.MustRegister(prometheus.NewGoCollector())
 
-	var auth *Auth0Authenticator
+	var authenticator *Auth0Authenticator
 	if conf.Auth.Enabled {
 		var err error
-		if auth, err = newAuthenticator(ctx, conf.Logger, registry, conf.Auth); err != nil {
+		if authenticator, err = NewAuthenticator(ctx, conf.Logger, registry, conf.Auth); err != nil {
 			return nil, err
 		}
 	}
@@ -116,7 +112,7 @@ func NewCore(ctx context.Context, conf Config) (*Core, error) {
 	}
 
 	c := &Core{
-		auth:     auth,
+		auth:     authenticator,
 		logger:   conf.Logger.Named("core").With(zap.String("personality", personality)),
 		registry: registry,
 		router:   router,
diff --git a/ppl/zqd/db/db.go b/ppl/zqd/db/db.go
index ccd0a0a9b..37753d000 100644
--- a/ppl/zqd/db/db.go
+++ b/ppl/zqd/db/db.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/brimsec/zq/api"
 	"github.com/brimsec/zq/pkg/iosrc"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/brimsec/zq/ppl/zqd/db/filedb"
 	"github.com/brimsec/zq/ppl/zqd/db/postgresdb"
 	"github.com/brimsec/zq/ppl/zqd/db/schema"
@@ -17,7 +18,7 @@ type DB interface {
 	CreateSpace(context.Context, schema.SpaceRow) error
 	CreateSubspace(context.Context, schema.SpaceRow) error
 	GetSpace(context.Context, api.SpaceID) (schema.SpaceRow, error)
-	ListSpaces(context.Context) ([]schema.SpaceRow, error)
+	ListSpaces(context.Context, auth.TenantID) ([]schema.SpaceRow, error)
 	UpdateSpaceName(context.Context, api.SpaceID, string) error
 	DeleteSpace(context.Context, api.SpaceID) error
 }
diff --git a/ppl/zqd/db/filedb/filedb.go b/ppl/zqd/db/filedb/filedb.go
index 9a94df282..4718cd3f0 100644
--- a/ppl/zqd/db/filedb/filedb.go
+++ b/ppl/zqd/db/filedb/filedb.go
@@ -9,7 +9,7 @@ import (
 
 	"github.com/brimsec/zq/api"
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/zqd/db/filedb/oldconfig"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/brimsec/zq/ppl/zqd/db/schema"
 	"github.com/brimsec/zq/zqe"
 	"go.uber.org/zap"
@@ -23,98 +23,40 @@ type FileDB struct {
 	path   iosrc.URI
 }
 
-func Create(ctx context.Context, logger *zap.Logger, path iosrc.URI, rows []schema.SpaceRow) (*FileDB, error) {
-	db := &FileDB{path: path, logger: logger}
-	if err := db.save(ctx, rows); err != nil {
-		return nil, err
-	}
-	db.logger.Info("Created", zap.String("kind", "file"), zap.String("uri", path.String()))
-	return db, nil
-}
-
 func Open(ctx context.Context, logger *zap.Logger, root iosrc.URI) (*FileDB, error) {
-	dburi := root.AppendPath(dbname)
-	exists, err := iosrc.Exists(ctx, dburi)
-	if err != nil {
+	if err := migrateOldConfig(ctx, logger, root); err != nil {
+		logger.Error("Error migrating old multifile configuration", zap.Error(err))
 		return nil, err
 	}
-	if exists {
-		return open(ctx, logger, dburi)
-	}
-
-	// Since the dbfile doesn't exist, we check if we need to migrate the older
-	// per-space config files into a new dbfile.
-	configs, err := oldconfig.LoadConfigs(ctx, logger, root)
-	if err != nil {
-		return nil, err
-	}
-	var rows []schema.SpaceRow
-	for id, config := range configs {
-		datauri := config.DataURI
-		if datauri.IsZero() {
-			datauri = root.AppendPath(string(id))
-		}
-		rows = append(rows, schema.SpaceRow{
-			ID:      id,
-			Name:    config.Name,
-			DataURI: datauri,
-			Storage: config.Storage,
-		})
-		for _, subcfg := range config.Subspaces {
-			openopts := subcfg.OpenOptions
-			rows = append(rows, schema.SpaceRow{
-				ID:       subcfg.ID,
-				ParentID: id,
-				Name:     subcfg.Name,
-				DataURI:  datauri,
-				Storage: api.StorageConfig{
-					Kind: api.ArchiveStore,
-					Archive: &api.ArchiveConfig{
-						OpenOptions: &openopts,
-					},
-				},
-			})
-		}
-	}
-	return Create(ctx, logger, dburi, rows)
-}
-
-func open(ctx context.Context, logger *zap.Logger, path iosrc.URI) (*FileDB, error) {
-	db := &FileDB{path: path, logger: logger}
-	// Verify file exists & is readable.
-	if _, err := db.load(ctx); err != nil {
+	dburi := root.AppendPath(dbname)
+	if err := migrateFileDatabase(ctx, dburi); err != nil {
+		logger.Error("Error migrating database file", zap.Error(err))
 		return nil, err
 	}
-	db.logger.Info("Loaded", zap.String("kind", "file"), zap.String("uri", path.String()))
+	db := &FileDB{path: dburi, logger: logger}
+	db.logger.Info("Loaded", zap.String("kind", "file"), zap.String("uri", dburi.String()))
 	return db, nil
 }
 
-const dbversion = 4
-
-type dbdataV4 struct {
-	Version   int               `json:"version"`
-	SpaceRows []schema.SpaceRow `json:"space_rows"`
-}
-
 func (db *FileDB) load(ctx context.Context) ([]schema.SpaceRow, error) {
 	b, err := iosrc.ReadFile(ctx, db.path)
 	if err != nil {
 		return nil, err
 	}
-	var data dbdataV4
+	var data dbdata
 	if err := json.Unmarshal(b, &data); err != nil {
 		return nil, err
 	}
-	if data.Version != dbversion {
-		return nil, fmt.Errorf("expected db version %d, found %d", dbversion, data.Version)
+	if data.Version != currentVersion {
+		return nil, fmt.Errorf("expected db version %d, found %d", currentVersion, data.Version)
 	}
 	return data.SpaceRows, nil
 }
 
 func (db *FileDB) save(ctx context.Context, lcs []schema.SpaceRow) error {
 	return iosrc.Replace(ctx, db.path, func(w io.Writer) error {
-		return json.NewEncoder(w).Encode(dbdataV4{
-			Version:   dbversion,
+		return json.NewEncoder(w).Encode(dbdata{
+			Version:   currentVersion,
 			SpaceRows: lcs,
 		})
 	})
@@ -133,12 +75,15 @@ func (db *FileDB) CreateSpace(ctx context.Context, row schema.SpaceRow) error {
 	}
 
 	for _, r := range rows {
-		if row.Name == r.Name {
-			return zqe.ErrConflict("space with name '%s' already exists", row.Name)
-		}
 		if row.ID == r.ID {
 			return zqe.ErrExists()
 		}
+		if r.TenantID != row.TenantID {
+			continue
+		}
+		if row.Name == r.Name {
+			return zqe.ErrConflict("space with name '%s' already exists", row.Name)
+		}
 	}
 
 	return db.save(ctx, append(rows, row))
@@ -161,12 +106,15 @@ func (db *FileDB) CreateSubspace(ctx context.Context, row schema.SpaceRow) error
 
 	parentIdx := -1
 	for i, r := range rows {
-		if row.Name == r.Name {
-			return zqe.ErrConflict("space with name '%s' already exists", row.Name)
-		}
 		if row.ID == r.ID {
 			return zqe.ErrExists()
 		}
+		if r.TenantID != row.TenantID {
+			continue
+		}
+		if row.Name == r.Name {
+			return zqe.ErrConflict("space with name '%s' already exists", row.Name)
+		}
 		if row.ParentID == r.ID {
 			parentIdx = i
 		}
@@ -193,10 +141,20 @@ func (db *FileDB) GetSpace(ctx context.Context, id api.SpaceID) (schema.SpaceRow
 	return schema.SpaceRow{}, zqe.ErrNotFound()
 }
 
-func (db *FileDB) ListSpaces(ctx context.Context) ([]schema.SpaceRow, error) {
+func (db *FileDB) ListSpaces(ctx context.Context, tenantID auth.TenantID) ([]schema.SpaceRow, error) {
 	db.mu.Lock()
 	defer db.mu.Unlock()
-	return db.load(ctx)
+	all, err := db.load(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var rows []schema.SpaceRow
+	for _, r := range all {
+		if r.TenantID == tenantID {
+			rows = append(rows, r)
+		}
+	}
+	return rows, nil
 }
 
 func (db *FileDB) UpdateSpaceName(ctx context.Context, id api.SpaceID, name string) error {
@@ -208,18 +166,23 @@ func (db *FileDB) UpdateSpaceName(ctx context.Context, id api.SpaceID, name stri
 	}
 
 	idx := -1
-	for i := range rows {
-		if rows[i].ID == id {
+	for i, r := range rows {
+		if r.ID == id {
 			idx = i
-			continue
-		}
-		if rows[i].Name == name {
-			return zqe.ErrConflict("space with name '%s' already exists", name)
+			break
 		}
 	}
 	if idx == -1 {
 		return zqe.ErrNotFound()
 	}
+	for _, r := range rows {
+		if r.TenantID != rows[idx].TenantID {
+			continue
+		}
+		if r.Name == name {
+			return zqe.ErrConflict("space with name '%s' already exists", name)
+		}
+	}
 
 	rows[idx].Name = name
 	return db.save(ctx, rows)
diff --git a/ppl/zqd/db/filedb/migrate.go b/ppl/zqd/db/filedb/migrate.go
new file mode 100644
index 000000000..6abee2091
--- /dev/null
+++ b/ppl/zqd/db/filedb/migrate.go
@@ -0,0 +1,142 @@
+package filedb
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+
+	"github.com/brimsec/zq/api"
+	"github.com/brimsec/zq/pkg/iosrc"
+	"github.com/brimsec/zq/ppl/zqd/auth"
+	"github.com/brimsec/zq/ppl/zqd/db/filedb/oldconfig"
+	"github.com/brimsec/zq/ppl/zqd/db/schema"
+	"github.com/brimsec/zq/zqe"
+	"go.uber.org/zap"
+)
+
+const currentVersion = 5
+
+type dbdata struct {
+	Version   int               `json:"version"`
+	SpaceRows []schema.SpaceRow `json:"space_rows"`
+}
+
+func migrateOldConfig(ctx context.Context, logger *zap.Logger, root iosrc.URI) error {
+	dburi := root.AppendPath(dbname)
+	exists, err := iosrc.Exists(ctx, dburi)
+	if err != nil || exists {
+		return err
+	}
+
+	// Since the dbfile doesn't exist, we check if we need to migrate the older
+	// per-space config files into a new dbfile.
+	configs, err := oldconfig.LoadConfigs(ctx, logger, root)
+	if err != nil {
+		return err
+	}
+	var rows []rowV4
+	for id, config := range configs {
+		datauri := config.DataURI
+		if datauri.IsZero() {
+			datauri = root.AppendPath(string(id))
+		}
+		rows = append(rows, rowV4{
+			ID:      id,
+			Name:    config.Name,
+			DataURI: datauri,
+			Storage: config.Storage,
+		})
+		for _, subcfg := range config.Subspaces {
+			openopts := subcfg.OpenOptions
+			rows = append(rows, rowV4{
+				ID:       subcfg.ID,
+				ParentID: id,
+				Name:     subcfg.Name,
+				DataURI:  datauri,
+				Storage: api.StorageConfig{
+					Kind: api.ArchiveStore,
+					Archive: &api.ArchiveConfig{
+						OpenOptions: &openopts,
+					},
+				},
+			})
+		}
+	}
+	return iosrc.Replace(ctx, dburi, func(w io.Writer) error {
+		return json.NewEncoder(w).Encode(dbdataV4{
+			Version:   4,
+			SpaceRows: rows,
+		})
+	})
+}
+
+func migrateFileDatabase(ctx context.Context, dburi iosrc.URI) error {
+	data, err := iosrc.ReadFile(ctx, dburi)
+	if err != nil {
+		if zqe.IsNotFound(err) {
+			return iosrc.Replace(ctx, dburi, func(w io.Writer) error {
+				return json.NewEncoder(w).Encode(dbdata{
+					Version: currentVersion,
+				})
+			})
+		}
+		return err
+	}
+	var vc struct {
+		Version int `json:"version"`
+	}
+	if err := json.Unmarshal(data, &vc); err != nil {
+		return err
+	}
+	var migrator func([]byte) (int, []byte, error)
+	version := vc.Version
+	for version < currentVersion {
+		switch version {
+		case 4:
+			migrator = migrateV5
+		default:
+			return fmt.Errorf("unsupported database version %d", version)
+		}
+		var err error
+		if version, data, err = migrator(data); err != nil {
+			return err
+		}
+	}
+	return iosrc.WriteFile(ctx, dburi, data)
+}
+
+type dbdataV4 struct {
+	Version   int     `json:"version"`
+	SpaceRows []rowV4 `json:"space_rows"`
+}
+
+type rowV4 struct {
+	ID       api.SpaceID       `json:"id"`
+	DataURI  iosrc.URI         `json:"data_uri"`
+	Name     string            `json:"name"`
+	ParentID api.SpaceID       `json:"parent_id"`
+	Storage  api.StorageConfig `json:"storage"`
+}
+
+func migrateV5(data []byte) (int, []byte, error) {
+	var dbv4 dbdataV4
+	if err := json.Unmarshal(data, &dbv4); err != nil {
+		return 0, nil, err
+	}
+	var db dbdata
+	db.Version = 5
+	db.SpaceRows = make([]schema.SpaceRow, 0, len(dbv4.SpaceRows))
+	for _, r := range dbv4.SpaceRows {
+		db.SpaceRows = append(db.SpaceRows, schema.SpaceRow{
+			TenantID: auth.AnonymousTenantID,
+			ID:       r.ID,
+			DataURI:  r.DataURI,
+			Name:     r.Name,
+			ParentID: r.ParentID,
+			Storage:  r.Storage,
+		})
+	}
+	data, err := json.Marshal(db)
+	return 5, data, err
+}
diff --git a/ppl/zqd/db/postgresdb/migrations/20210119172011_add_tenant_id.down.sql b/ppl/zqd/db/postgresdb/migrations/20210119172011_add_tenant_id.down.sql
new file mode 100644
index 000000000..d4a7b4a2d
--- /dev/null
+++ b/ppl/zqd/db/postgresdb/migrations/20210119172011_add_tenant_id.down.sql
@@ -0,0 +1,3 @@
+ALTER TABLE space DROP CONSTRAINT space_name_key;
+ALTER TABLE space ADD CONSTRAINT space_name_key UNIQUE (name);
+ALTER TABLE space DROP tenant_id;
diff --git a/ppl/zqd/db/postgresdb/migrations/20210119172011_add_tenant_id.up.sql b/ppl/zqd/db/postgresdb/migrations/20210119172011_add_tenant_id.up.sql
new file mode 100644
index 000000000..0cb4c221f
--- /dev/null
+++ b/ppl/zqd/db/postgresdb/migrations/20210119172011_add_tenant_id.up.sql
@@ -0,0 +1,3 @@
+ALTER TABLE space ADD tenant_id text;
+ALTER TABLE space DROP CONSTRAINT space_name_key;
+ALTER TABLE space ADD CONSTRAINT space_name_key UNIQUE (tenant_id, name);
diff --git a/ppl/zqd/db/postgresdb/postgresdb.go b/ppl/zqd/db/postgresdb/postgresdb.go
index e8388bb01..3c449f116 100644
--- a/ppl/zqd/db/postgresdb/postgresdb.go
+++ b/ppl/zqd/db/postgresdb/postgresdb.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 
 	"github.com/brimsec/zq/api"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/brimsec/zq/ppl/zqd/db/schema"
 	"github.com/brimsec/zq/zqe"
 	"github.com/go-pg/pg/v10"
@@ -61,9 +62,9 @@ func (d *PostgresDB) GetSpace(ctx context.Context, id api.SpaceID) (schema.Space
 	return space, err
 }
 
-func (d *PostgresDB) ListSpaces(ctx context.Context) ([]schema.SpaceRow, error) {
+func (d *PostgresDB) ListSpaces(ctx context.Context, tenantID auth.TenantID) ([]schema.SpaceRow, error) {
 	var spaces []schema.SpaceRow
-	_, err := d.db.QueryContext(ctx, &spaces, "SELECT * FROM space")
+	_, err := d.db.QueryContext(ctx, &spaces, "SELECT * FROM space WHERE tenant_id = ?", tenantID)
 	return spaces, err
 }
 
diff --git a/ppl/zqd/db/schema/spacerow.go b/ppl/zqd/db/schema/spacerow.go
index 2184cfc18..4ba926832 100644
--- a/ppl/zqd/db/schema/spacerow.go
+++ b/ppl/zqd/db/schema/spacerow.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/brimsec/zq/api"
 	"github.com/brimsec/zq/pkg/iosrc"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/segmentio/ksuid"
 )
 
@@ -17,6 +18,7 @@ type SpaceRow struct {
 	Name      string            `json:"name"`
 	ParentID  api.SpaceID       `json:"parent_id"`
 	Storage   api.StorageConfig `json:"storage"`
+	TenantID  auth.TenantID     `json:"tenant_id"`
 }
 
 func NewSpaceID() api.SpaceID {
diff --git a/ppl/zqd/handlers.go b/ppl/zqd/handlers.go
index 894059170..4957e5def 100644
--- a/ppl/zqd/handlers.go
+++ b/ppl/zqd/handlers.go
@@ -12,6 +12,7 @@ import (
 	"github.com/brimsec/zq/compiler"
 	"github.com/brimsec/zq/pcap"
 	"github.com/brimsec/zq/pkg/ctxio"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/brimsec/zq/ppl/zqd/ingest"
 	"github.com/brimsec/zq/ppl/zqd/jsonpipe"
 	"github.com/brimsec/zq/ppl/zqd/search"
@@ -46,6 +47,8 @@ func errorResponse(e error) (status int, ae *api.Error) {
 		status = http.StatusConflict
 	case zqe.NoCredentials:
 		status = http.StatusUnauthorized
+	case zqe.Forbidden:
+		status = http.StatusForbidden
 	}
 
 	ae.Kind = ze.Kind.String()
@@ -96,7 +99,7 @@ func handleSearch(c *Core, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	srch, err := search.NewSearchOp(req)
+	srch, err := search.NewSearchOp(req, c.requestLogger(r))
 	if err != nil {
 		respondError(c, w, r, err)
 		return
@@ -592,14 +595,10 @@ func extractSpaceID(c *Core, w http.ResponseWriter, r *http.Request) (api.SpaceI
 }
 
 func handleAuthIdentityGet(c *Core, w http.ResponseWriter, r *http.Request) {
-	ident, ok := IdentifyFromContext(r.Context())
-	if !ok {
-		respondError(c, w, r, zqe.ErrInvalid("no valid credentials"))
-		return
-	}
+	ident := auth.IdentityFromContext(r.Context())
 	respond(c, w, r, http.StatusOK, api.AuthIdentityResponse{
-		TenantID: ident.TenantID,
-		UserID:   ident.UserID,
+		TenantID: string(ident.TenantID),
+		UserID:   string(ident.UserID),
 	})
 }
 
diff --git a/ppl/zqd/search/search.go b/ppl/zqd/search/search.go
index 8b4aae877..4367f6ded 100644
--- a/ppl/zqd/search/search.go
+++ b/ppl/zqd/search/search.go
@@ -36,11 +36,12 @@ const (
 )
 
 type SearchOp struct {
+	logger  *zap.Logger
 	query   *Query
 	workers int // for distributed queries only
 }
 
-func NewSearchOp(req api.SearchRequest) (*SearchOp, error) {
+func NewSearchOp(req api.SearchRequest, logger *zap.Logger) (*SearchOp, error) {
 	if req.Span.Ts < 0 {
 		return nil, zqe.ErrInvalid("time span must have non-negative timestamp")
 	}
@@ -59,7 +60,10 @@ func NewSearchOp(req api.SearchRequest) (*SearchOp, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &SearchOp{query: query}, nil
+	return &SearchOp{
+		logger: logger,
+		query:  query,
+	}, nil
 }
 
 func (s *SearchOp) Run(ctx context.Context, store storage.Storage, output Output) (err error) {
@@ -83,9 +87,10 @@ func (s *SearchOp) Run(ctx context.Context, store storage.Storage, output Output
 	switch st := store.(type) {
 	case *archivestore.Storage:
 		return driver.MultiRun(ctx, d, s.query.Proc, zctx, st.MultiSource(), driver.MultiConfig{
+			Logger:    s.logger,
+			Order:     zbuf.OrderDesc,
 			Span:      s.query.Span,
 			StatsTick: statsTicker.C,
-			Order:     zbuf.OrderDesc,
 		})
 	case *filestore.Storage:
 		rc, err := st.Open(ctx, zctx, s.query.Span)
@@ -95,6 +100,7 @@ func (s *SearchOp) Run(ctx context.Context, store storage.Storage, output Output
 		defer rc.Close()
 
 		return driver.Run(ctx, d, s.query.Proc, zctx, rc, driver.Config{
+			Logger:            s.logger,
 			ReaderSortKey:     "ts",
 			ReaderSortReverse: true,
 			Span:              s.query.Span,
diff --git a/ppl/zqd/search/worker.go b/ppl/zqd/search/worker.go
index 4e69085b9..e55353bf6 100644
--- a/ppl/zqd/search/worker.go
+++ b/ppl/zqd/search/worker.go
@@ -12,16 +12,18 @@ import (
 	"github.com/brimsec/zq/ppl/zqd/storage/archivestore"
 	"github.com/brimsec/zq/zng/resolver"
 	"github.com/brimsec/zq/zqe"
+	"go.uber.org/zap"
 )
 
 type WorkerOp struct {
-	proc  ast.Proc
-	span  nano.Span
-	src   driver.Source
-	store *archivestore.Storage
+	logger *zap.Logger
+	proc   ast.Proc
+	span   nano.Span
+	src    driver.Source
+	store  *archivestore.Storage
 }
 
-func NewWorkerOp(ctx context.Context, req api.WorkerChunkRequest, st storage.Storage) (*WorkerOp, error) {
+func NewWorkerOp(ctx context.Context, req api.WorkerChunkRequest, st storage.Storage, logger *zap.Logger) (*WorkerOp, error) {
 	// XXX zqd only supports backwards searches, remove once this has been
 	// fixed.
 	if req.Dir == 1 {
@@ -42,7 +44,13 @@ func NewWorkerOp(ctx context.Context, req api.WorkerChunkRequest, st storage.Sto
 	if err != nil {
 		return nil, err
 	}
-	return &WorkerOp{proc: proc, span: req.Span, src: src, store: store}, nil
+	return &WorkerOp{
+		logger: logger,
+		proc:   proc,
+		span:   req.Span,
+		src:    src,
+		store:  store,
+	}, nil
 }
 
 func (w *WorkerOp) Run(ctx context.Context, output Output) (err error) {
@@ -64,6 +72,7 @@ func (w *WorkerOp) Run(ctx context.Context, output Output) (err error) {
 	zctx := resolver.NewContext()
 
 	return driver.MultiRun(ctx, d, w.proc, zctx, w.store.StaticSource(w.src), driver.MultiConfig{
+		Logger:    w.logger,
 		Span:      w.span,
 		StatsTick: statsTicker.C,
 	})
diff --git a/ppl/zqd/worker.go b/ppl/zqd/worker.go
index ccf0d9a32..8b3c41674 100644
--- a/ppl/zqd/worker.go
+++ b/ppl/zqd/worker.go
@@ -22,7 +22,7 @@ func handleWorkerRootSearch(c *Core, w http.ResponseWriter, r *http.Request) {
 		respondError(c, w, r, err)
 		return
 	}
-	srch, err := search.NewSearchOp(req.SearchRequest)
+	srch, err := search.NewSearchOp(req.SearchRequest, c.requestLogger(r))
 	if err != nil {
 		respondError(c, w, r, err)
 		return
@@ -56,7 +56,7 @@ func handleWorkerChunkSearch(c *Core, w http.ResponseWriter, httpReq *http.Reque
 		respondError(c, w, httpReq, err)
 		return
 	}
-	work, err := search.NewWorkerOp(ctx, req, archivestore.NewStorage(ark))
+	work, err := search.NewWorkerOp(ctx, req, archivestore.NewStorage(ark), c.requestLogger(httpReq))
 	if err != nil {
 		respondError(c, w, httpReq, err)
 		return
diff --git a/ppl/zqd/worker/registration.go b/ppl/zqd/worker/registration.go
index 4b20a09bf..3d99709b5 100644
--- a/ppl/zqd/worker/registration.go
+++ b/ppl/zqd/worker/registration.go
@@ -28,7 +28,7 @@ func (c *WorkerConfig) SetFlags(fs *flag.FlagSet) {
 	fs.StringVar(&c.BoundWorkers, "worker.bound", "", "bound workers as comma-separated [addr]:port list")
 	fs.BoolVar(&c.Fallback, "worker.fallback", false, "fallback to using fewer workers than requested")
 	fs.StringVar(&c.Host, "worker.host", "", "host ip of container")
-	fs.DurationVar(&c.IdleTime, "worker.idletime", 2*time.Second, "timeout duration for zombie worker processes to exit")
+	fs.DurationVar(&c.IdleTime, "worker.idletime", 10*time.Second, "timeout duration for zombie worker processes to exit")
 	fs.DurationVar(&c.LongPoll, "worker.longpoll", 30*time.Second, "timeout duration for long poll of /recruiter/register request")
 	fs.DurationVar(&c.MaxRetry, "worker.maxretry", 10*time.Second, "maximum retry wait duration for registration request")
 	fs.DurationVar(&c.MinRetry, "worker.minretry", 200*time.Millisecond, "minimum retry wait duration for registration request")
diff --git a/zng/type.go b/zng/type.go
index b14b08e1e..b8a898f49 100644
--- a/zng/type.go
+++ b/zng/type.go
@@ -337,6 +337,11 @@ func IsUnionType(typ Type) bool {
 	return ok
 }
 
+func IsRecordType(typ Type) bool {
+	_, ok := AliasedType(typ).(*TypeRecord)
+	return ok
+}
+
 func IsContainerType(typ Type) bool {
 	switch typ := typ.(type) {
 	case *TypeAlias:
@@ -348,6 +353,10 @@ func IsContainerType(typ Type) bool {
 	}
 }
 
+func IsPrimitiveType(typ Type) bool {
+	return !IsContainerType(typ)
+}
+
 func AliasTypes(typ Type) []*TypeAlias {
 	var aliases []*TypeAlias
 	switch typ := typ.(type) {
diff --git a/zqe/error.go b/zqe/error.go
index 65b0d97bc..44f4393e8 100644
--- a/zqe/error.go
+++ b/zqe/error.go
@@ -19,25 +19,28 @@ const (
 	Other Kind = iota
 	Conflict
 	Exists
+	Forbidden
 	Invalid
-	NotFound
 	NoCredentials
+	NotFound
 )
 
 func (k Kind) String() string {
 	switch k {
-	case Other:
-		return "other error"
 	case Conflict:
 		return "conflict with pending operation"
-	case Invalid:
-		return "invalid operation"
 	case Exists:
 		return "item already exists"
-	case NotFound:
-		return "item does not exist"
+	case Invalid:
+		return "invalid operation"
+	case Forbidden:
+		return "forbidden"
 	case NoCredentials:
 		return "missing authentication credentials"
+	case NotFound:
+		return "item does not exist"
+	case Other:
+		return "other error"
 	}
 	return "unknown error kind"
 }
@@ -135,18 +138,21 @@ func IsKind(err error, k Kind) bool {
 	return errors.As(err, &zerr) && zerr.Kind == k
 }
 
-func IsOther(err error) bool    { return IsKind(err, Other) }
-func IsConflict(err error) bool { return IsKind(err, Conflict) }
-func IsExists(err error) bool   { return IsKind(err, Exists) }
-func IsInvalid(err error) bool  { return IsKind(err, Invalid) }
-func IsNotFound(err error) bool { return IsKind(err, NotFound) }
+func IsConflict(err error) bool      { return IsKind(err, Conflict) }
+func IsExists(err error) bool        { return IsKind(err, Exists) }
+func IsForbidden(err error) bool     { return IsKind(err, Forbidden) }
+func IsInvalid(err error) bool       { return IsKind(err, Invalid) }
+func IsNoCredentials(err error) bool { return IsKind(err, NoCredentials) }
+func IsNotFound(err error) bool      { return IsKind(err, NotFound) }
+func IsOther(err error) bool         { return IsKind(err, Other) }
 
-func ErrOther(args ...interface{}) error         { return errKind(Other, args) }
 func ErrConflict(args ...interface{}) error      { return errKind(Conflict, args) }
 func ErrExists(args ...interface{}) error        { return errKind(Exists, args) }
+func ErrForbidden(args ...interface{}) error     { return errKind(Forbidden, args) }
 func ErrInvalid(args ...interface{}) error       { return errKind(Invalid, args) }
-func ErrNotFound(args ...interface{}) error      { return errKind(NotFound, args) }
 func ErrNoCredentials(args ...interface{}) error { return errKind(NoCredentials, args) }
+func ErrNotFound(args ...interface{}) error      { return errKind(NotFound, args) }
+func ErrOther(args ...interface{}) error         { return errKind(Other, args) }
 
 func errKind(k Kind, args []interface{}) error {
 	args = append([]interface{}{k}, args...)


Test patch:
diff --git a/expr/ztests/shape-cast.yaml b/expr/ztests/shape-cast.yaml
new file mode 100644
index 000000000..03ce6b1e6
--- /dev/null
+++ b/expr/ztests/shape-cast.yaml
@@ -0,0 +1,18 @@
+zql: |
+   put
+   // cast to type with same field order
+   id=cast(id, "({orig_h:ip,orig_p:port=(uint16),resp_h:ip,resp_p:port})"),
+   // cast to type with different field order, does not change output order
+   id2=cast(id, "({resp_h:ip,resp_p:port=(uint16),orig_h:ip,orig_p:port})")
+
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,resp_p:port,resp_h:string],other:float64]
+   0:[[ff02::fb;5353;5353;1.2.3.4;]123;]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,resp_p:port,resp_h:ip],other:float64,id2:record[orig_h:ip,orig_p:port,resp_p:port,resp_h:ip]]
+   0:[[ff02::fb;5353;5353;1.2.3.4;]123;[ff02::fb;5353;5353;1.2.3.4;]]
+
diff --git a/expr/ztests/shape-crop.yaml b/expr/ztests/shape-crop.yaml
new file mode 100644
index 000000000..da12224f4
--- /dev/null
+++ b/expr/ztests/shape-crop.yaml
@@ -0,0 +1,17 @@
+zql: |
+   put
+   // crop to type with same field order
+   id=crop(id, "({orig_h:ip,orig_p:port=(uint16)})"),
+   // crop to type with different field order, does not change output order
+   id2=crop(id, "({orig_p:port=(uint16),orig_h:ip})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,resp_h:ip,resp_p:port],other:float64]
+   0:[[ff02::fb;5353;1.2.3.4;5353;]123;]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port],other:float64,id2:record[orig_h:ip,orig_p:port]]
+   0:[[ff02::fb;5353;]123;[ff02::fb;5353;]]
+
diff --git a/expr/ztests/shape-fill-fit.yaml b/expr/ztests/shape-fill-fit.yaml
new file mode 100644
index 000000000..a5f39db22
--- /dev/null
+++ b/expr/ztests/shape-fill-fit.yaml
@@ -0,0 +1,13 @@
+zql: |
+   cut id_filled=fill(id, "({orig_h:ip,orig_p:port=(uint16)})"),  id_fitted=fit(id, "({orig_h:ip,orig_p:port=(uint16)})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,resp_h:string]]
+   0:[[ff02::fb;ff02::fb;]]
+
+output: |
+   #port=uint16
+   #0:record[id_filled:record[orig_h:string,resp_h:string,orig_p:port],id_fitted:record[orig_h:string,orig_p:port]]
+   0:[[ff02::fb;ff02::fb;-;][ff02::fb;-;]]
+
diff --git a/expr/ztests/shape-order.yaml b/expr/ztests/shape-order.yaml
new file mode 100644
index 000000000..c961fb933
--- /dev/null
+++ b/expr/ztests/shape-order.yaml
@@ -0,0 +1,13 @@
+zql: |
+   put id=order(id, "({orig_h: string, orig_p: port=(uint16), resp_h:ip,resp_p:port=(uint16)})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,orig_p:port,resp_p:port,resp_h:string],other:float64]
+   0:[[ff02::fb;5353;5354;1.2.3.4;]123;]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,orig_p:port,resp_h:string,resp_p:port],other:float64]
+   0:[[ff02::fb;5353;1.2.3.4;5354;]123;]
+
diff --git a/expr/ztests/shape.yaml b/expr/ztests/shape.yaml
new file mode 100644
index 000000000..4543b38d2
--- /dev/null
+++ b/expr/ztests/shape.yaml
@@ -0,0 +1,13 @@
+zql: |
+   put id=shape(id, "({orig_h: ip, orig_p: port=(uint16), vlan: uint16, resp_h:ip,resp_p:port})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,resp_h:string,orig_p:port,resp_p:port,tag:uint16]]
+   0:[[1.2.3.4;5.6.7.8;1234;53;0;]]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,vlan:uint16,resp_h:ip,resp_p:port]]
+   0:[[1.2.3.4;1234;-;5.6.7.8;53;]]
+
diff --git a/k8s/testdiff/count.tzng b/k8s/testdiff/count.tzng
deleted file mode 100644
index 65cfe129e..000000000
--- a/k8s/testdiff/count.tzng
+++ /dev/null
@@ -1,2 +0,0 @@
-#0:record[count:uint64]
-0:[12243237;]
diff --git a/k8s/testdiff/countbytxhost.tzng b/k8s/testdiff/countbytxhost.tzng
deleted file mode 100644
index 481ad163f..000000000
--- a/k8s/testdiff/countbytxhost.tzng
+++ /dev/null
@@ -1,6 +0,0 @@
-#0:record[tx_hosts:set[ip],count:uint64]
-0:[[10.47.2.156;]985198;]
-0:[[10.47.7.58;]827462;]
-0:[[10.47.5.58;]779783;]
-0:[[10.47.8.58;]659903;]
-0:[[10.47.1.58;]536395;]
diff --git a/ppl/archive/archive_test.go b/ppl/archive/archive_test.go
index 03ce2a5ea..f01d2b21f 100644
--- a/ppl/archive/archive_test.go
+++ b/ppl/archive/archive_test.go
@@ -151,24 +151,19 @@ func TestOpenOptions(t *testing.T) {
 }
 
 func TestSeekIndex(t *testing.T) {
-	datapath, err := ioutil.TempDir("", "")
-	require.NoError(t, err)
-	defer os.RemoveAll(datapath)
+	datapath := t.TempDir()
 
 	orig := ImportStreamRecordsMax
 	ImportStreamRecordsMax = 1
 	defer func() {
 		ImportStreamRecordsMax = orig
 	}()
-	createArchiveSpace(t, datapath, babble, &CreateOptions{
-		// Must use SortAscending: true until zq#1329 is addressed.
-		SortAscending: true,
-	})
-	_, err = OpenArchive(datapath, &OpenOptions{})
+	createArchiveSpace(t, datapath, babble, nil)
+	_, err := OpenArchive(datapath, &OpenOptions{})
 	require.NoError(t, err)
 
-	first1 := nano.Ts(1587508830068523240)
-	var idxUri iosrc.URI
+	first1 := nano.Ts(1587513592062544400)
+	var idxURI iosrc.URI
 	err = filepath.Walk(datapath, func(p string, fi os.FileInfo, err error) error {
 		if err != nil {
 			return err
@@ -179,23 +174,24 @@ func TestSeekIndex(t *testing.T) {
 				return err
 			}
 			uri.Path = path.Dir(uri.Path)
-			chunk, err := chunk.Open(context.Background(), uri, id, zbuf.OrderAsc)
+			chunk, err := chunk.Open(context.Background(), uri, id, zbuf.OrderDesc)
 			if err != nil {
 				return err
 			}
 			if chunk.First == first1 {
-				idxUri = chunk.SeekIndexPath()
+				idxURI = chunk.SeekIndexPath()
 			}
 		}
 		return nil
 	})
 	require.NoError(t, err)
-	finder, err := microindex.NewFinder(context.Background(), resolver.NewContext(), idxUri)
+	finder, err := microindex.NewFinder(context.Background(), resolver.NewContext(), idxURI)
 	require.NoError(t, err)
 	keys, err := finder.ParseKeys("1587508851")
 	require.NoError(t, err)
 	rec, err := finder.ClosestLTE(keys)
 	require.NoError(t, err)
+	require.NoError(t, finder.Close())
 
 	var buf bytes.Buffer
 	w := tzngio.NewWriter(zio.NopCloser(&buf))
@@ -203,7 +199,7 @@ func TestSeekIndex(t *testing.T) {
 
 	exp := `
 #0:record[ts:time,offset:int64]
-0:[1587508850.06466032;202;]
+0:[1587508850.06466032;23795;]
 `
 	require.Equal(t, test.Trim(exp), buf.String())
 }
diff --git a/ppl/zqd/auth/validator_test.go b/ppl/zqd/auth/validator_test.go
new file mode 100644
index 000000000..ba329a09d
--- /dev/null
+++ b/ppl/zqd/auth/validator_test.go
@@ -0,0 +1,224 @@
+package auth
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/dgrijalva/jwt-go"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testKeyID    = "testkey"
+	testKeyFile  = "../testdata/auth-private-key"
+	testJWKSFile = "../testdata/auth-public-jwks.json"
+)
+
+func testValidator(t *testing.T) *TokenValidator {
+	v, err := NewTokenValidator("https://testdomain", testJWKSFile)
+	require.NoError(t, err)
+	return v
+}
+
+func genToken(t *testing.T, claims jwt.MapClaims) string {
+	token, err := makeToken(testKeyID, testKeyFile, claims)
+	require.NoError(t, err)
+	return token
+}
+
+func TestValidate(t *testing.T) {
+	expectedIdent := Identity{
+		TenantID: "test_tenant_id",
+		UserID:   "test_user_id",
+	}
+	token, err := GenerateAccessToken(testKeyID, testKeyFile, 1*time.Hour,
+		"https://testdomain", "test_tenant_id", "test_user_id")
+	require.NoError(t, err)
+	validator := testValidator(t)
+
+	ident, err := validator.Validate(token)
+	require.NoError(t, err)
+	require.Equal(t, expectedIdent, ident)
+
+	req, err := http.NewRequest("GET", "https://testdomain", nil)
+	require.NoError(t, err)
+	req.Header.Add("Authorization", "Bearer "+token)
+	tokstr, ident, err := validator.ValidateRequest(req)
+	require.Equal(t, expectedIdent, ident)
+	require.Equal(t, token, tokstr)
+
+	req, err = http.NewRequest("GET", "https://testdomain", nil)
+	require.NoError(t, err)
+	tokstr, ident, err = validator.ValidateRequest(req)
+	require.Error(t, err)
+}
+
+func TestBadClaims(t *testing.T) {
+	var cases = []struct {
+		name  string
+		token string
+	}{
+		{
+			name: "missing audience",
+			token: genToken(t, jwt.MapClaims{
+				"exp":         time.Now().Add(1 * time.Hour).Unix(),
+				"iss":         "https://testdomain/",
+				TenantIDClaim: "test_tenant_id",
+				UserIDClaim:   "test_user_id",
+			}),
+		},
+		{
+			name: "invalid audience",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         "foo",
+				"exp":         time.Now().Add(1 * time.Hour).Unix(),
+				"iss":         "https://testdomain/",
+				TenantIDClaim: "test_tenant_id",
+				UserIDClaim:   "test_user_id",
+			}),
+		},
+		{
+			name: "missing expiration",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         AudienceClaimValue,
+				"iss":         "https://testdomain/",
+				TenantIDClaim: "test_tenant_id",
+				UserIDClaim:   "test_user_id",
+			}),
+		},
+		{
+			name: "expired expiration",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         AudienceClaimValue,
+				"exp":         time.Now().Add(-1 * time.Hour).Unix(),
+				"iss":         "https://testdomain/",
+				TenantIDClaim: "test_tenant_id",
+				UserIDClaim:   "test_user_id",
+			}),
+		},
+		{
+			name: "missing issuer",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         AudienceClaimValue,
+				"exp":         time.Now().Add(1 * time.Hour).Unix(),
+				TenantIDClaim: "test_tenant_id",
+				UserIDClaim:   "test_user_id",
+			}),
+		},
+		{
+			name: "invalid issuer",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         AudienceClaimValue,
+				"exp":         time.Now().Add(1 * time.Hour).Unix(),
+				"iss":         "foo",
+				TenantIDClaim: "test_tenant_id",
+				UserIDClaim:   "test_user_id",
+			}),
+		},
+		{
+			name: "missing user id",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         AudienceClaimValue,
+				"exp":         time.Now().Add(1 * time.Hour).Unix(),
+				"iss":         "https://testdomain/",
+				TenantIDClaim: "test_tenant_id",
+			}),
+		},
+		{
+			name: "anonymous user id",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         AudienceClaimValue,
+				"exp":         time.Now().Add(1 * time.Hour).Unix(),
+				"iss":         "https://testdomain/",
+				TenantIDClaim: "test_tenant_id",
+				UserIDClaim:   AnonymousUserID,
+			}),
+		},
+		{
+			name: "missing tenant id",
+			token: genToken(t, jwt.MapClaims{
+				"aud":       AudienceClaimValue,
+				"exp":       time.Now().Add(1 * time.Hour).Unix(),
+				"iss":       "https://testdomain/",
+				UserIDClaim: "test_user_id",
+			}),
+		},
+		{
+			name: "anonymous user id",
+			token: genToken(t, jwt.MapClaims{
+				"aud":         AudienceClaimValue,
+				"exp":         time.Now().Add(1 * time.Hour).Unix(),
+				"iss":         "https://testdomain/",
+				TenantIDClaim: AnonymousTenantID,
+				UserIDClaim:   "test_user_id",
+			}),
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			v := testValidator(t)
+			_, err := v.Validate(c.token)
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestKeyID(t *testing.T) {
+	claims := jwt.MapClaims{
+		"aud":         AudienceClaimValue,
+		"exp":         time.Now().Add(1 * time.Hour).Unix(),
+		"iss":         "https://testdomain/",
+		TenantIDClaim: "test_tenant_id",
+		UserIDClaim:   "test_user_id",
+	}
+	privateKey, err := loadPrivateKey(testKeyFile)
+	require.NoError(t, err)
+	v := testValidator(t)
+
+	// Bad key id
+	token := jwt.New(jwt.SigningMethodRS256)
+	token.Claims = claims
+	token.Header["kid"] = "foo"
+	tokenString, err := token.SignedString(privateKey)
+	require.NoError(t, err)
+	_, err = v.Validate(tokenString)
+	require.Error(t, err)
+
+	// No key id
+	token = jwt.New(jwt.SigningMethodRS256)
+	token.Claims = claims
+	tokenString, err = token.SignedString(privateKey)
+	require.NoError(t, err)
+	_, err = v.Validate(tokenString)
+	require.Error(t, err)
+}
+
+func TestAudienceSlice(t *testing.T) {
+	expectedIdent := Identity{
+		TenantID: "test_tenant_id",
+		UserID:   "test_user_id",
+	}
+	token := genToken(t, jwt.MapClaims{
+		"aud":         []string{AudienceClaimValue, "foobar"},
+		"exp":         time.Now().Add(1 * time.Hour).Unix(),
+		"iss":         "https://testdomain/",
+		TenantIDClaim: "test_tenant_id",
+		UserIDClaim:   "test_user_id",
+	})
+	validator := testValidator(t)
+	ident, err := validator.Validate(token)
+	require.NoError(t, err)
+	require.Equal(t, expectedIdent, ident)
+
+	token = genToken(t, jwt.MapClaims{
+		"aud":         []string{"foo", "bar"},
+		"exp":         time.Now().Add(1 * time.Hour).Unix(),
+		"iss":         "https://testdomain/",
+		TenantIDClaim: "test_tenant_id",
+		UserIDClaim:   "test_user_id",
+	})
+	_, err = validator.Validate(token)
+	require.Error(t, err)
+}
diff --git a/ppl/zqd/auth_test.go b/ppl/zqd/auth_test.go
index 63d136bb7..3ce1fedc2 100644
--- a/ppl/zqd/auth_test.go
+++ b/ppl/zqd/auth_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/brimsec/zq/api"
 	"github.com/brimsec/zq/api/client"
 	"github.com/brimsec/zq/ppl/zqd"
+	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/dgrijalva/jwt-go"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -25,6 +26,14 @@ func testAuthConfig() zqd.AuthConfig {
 	}
 }
 
+func genToken(t *testing.T, tenantID auth.TenantID, userID auth.UserID) string {
+	ac := testAuthConfig()
+	token, err := auth.GenerateAccessToken("testkey", "testdata/auth-private-key",
+		1*time.Hour, ac.Domain, tenantID, userID)
+	require.NoError(t, err)
+	return token
+}
+
 func makeToken(t *testing.T, kid string, c jwt.MapClaims) string {
 	b, err := ioutil.ReadFile("testdata/auth-private-key")
 	require.NoError(t, err)
@@ -58,13 +67,7 @@ func TestAuthIdentity(t *testing.T) {
 	require.True(t, errors.As(err, &identErr))
 	require.Equal(t, http.StatusUnauthorized, identErr.StatusCode())
 
-	token := makeToken(t, "testkey", map[string]interface{}{
-		"aud":             zqd.AudienceClaimValue,
-		"exp":             time.Now().Add(1 * time.Hour).Unix(),
-		"iss":             authConfig.Domain + "/",
-		zqd.TenantIDClaim: "test_tenant_id",
-		zqd.UserIDClaim:   "test_user_id",
-	})
+	token := genToken(t, "test_tenant_id", "test_user_id")
 	conn.SetAuthToken(token)
 	res, err := conn.AuthIdentity(context.Background())
 	require.NoError(t, err)
@@ -77,52 +80,6 @@ func TestAuthIdentity(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestAuthTokenExpiration(t *testing.T) {
-	authConfig := testAuthConfig()
-	var cases = []struct {
-		name  string
-		token string
-	}{
-		{
-			name: "missing",
-			token: makeToken(t, "testkey", map[string]interface{}{
-				"aud":             zqd.AudienceClaimValue,
-				"iss":             authConfig.Domain + "/",
-				zqd.TenantIDClaim: "test_tenant_id",
-				zqd.UserIDClaim:   "test_user_id",
-			}),
-		},
-		{
-			name: "expired",
-			token: makeToken(t, "testkey", map[string]interface{}{
-				"aud":             zqd.AudienceClaimValue,
-				"exp":             time.Now().Add(-1 * time.Hour).Unix(),
-				"iss":             authConfig.Domain + "/",
-				zqd.TenantIDClaim: "test_tenant_id",
-				zqd.UserIDClaim:   "test_user_id",
-			}),
-		},
-	}
-
-	for _, c := range cases {
-		t.Run(c.name, func(t *testing.T) {
-			_, conn := newCoreWithConfig(t, zqd.Config{
-				Auth:   authConfig,
-				Logger: zap.NewNop(),
-			})
-			conn.SetAuthToken(c.token)
-			_, err := conn.AuthIdentity(context.Background())
-
-			var identErr *client.ErrorResponse
-			_, err = conn.AuthIdentity(context.Background())
-			require.Error(t, err)
-			require.True(t, errors.As(err, &identErr))
-			require.Equal(t, http.StatusUnauthorized, identErr.StatusCode())
-			require.Regexp(t, "invalid expiration", identErr.Error())
-		})
-	}
-}
-
 func TestAuthMethodGet(t *testing.T) {
 	t.Run("none", func(t *testing.T) {
 		_, connNoAuth := newCoreWithConfig(t, zqd.Config{
@@ -146,7 +103,7 @@ func TestAuthMethodGet(t *testing.T) {
 		require.Equal(t, &api.AuthMethodResponse{
 			Kind: "auth0",
 			Auth0: &api.AuthMethodAuth0Details{
-				Audience: zqd.AudienceClaimValue,
+				Audience: auth.AudienceClaimValue,
 				Domain:   authConfig.Domain,
 				ClientID: authConfig.ClientID,
 			},
diff --git a/ppl/zqd/db/postgresdb/ztests/rename.yaml b/ppl/zqd/db/postgresdb/ztests/rename.yaml
index d14f1b41f..57db1226c 100644
--- a/ppl/zqd/db/postgresdb/ztests/rename.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/rename.yaml
@@ -23,7 +23,7 @@ outputs:
     data: |
       testsp: space created
       ===
-      testsp: space renamed to newname
+      space renamed to newname
       ===
       newname
   - name: stderr
diff --git a/ppl/zqd/db/postgresdb/ztests/startup.sh b/ppl/zqd/db/postgresdb/ztests/startup.sh
index 7f5878c6c..bc5c60b41 100644
--- a/ppl/zqd/db/postgresdb/ztests/startup.sh
+++ b/ppl/zqd/db/postgresdb/ztests/startup.sh
@@ -24,7 +24,7 @@ zqd listen -l=localhost:0 \
   -portfile="$portdir/zqd" \
   -db.kind="postgres" \
   -db.postgres.url=$PG_TEST \
-  -db.postgres.database=$dbname &> zqd.log &
+  -db.postgres.database=$dbname $ZQD_EXTRA_FLAGS &> zqd.log &
 
 zqdpid=$!
 awaitfile $portdir/zqd
diff --git a/ppl/zqd/db/postgresdb/ztests/tenant.yaml b/ppl/zqd/db/postgresdb/ztests/tenant.yaml
new file mode 100644
index 000000000..31d5f294c
--- /dev/null
+++ b/ppl/zqd/db/postgresdb/ztests/tenant.yaml
@@ -0,0 +1,86 @@
+tag: postgres
+
+script: |
+  # Start an authentication enabled zqd, and create 2 separate access tokens
+  # representing different users in different tenants.
+  ZQD_EXTRA_FLAGS="-auth.enabled=true -auth.clientid=testclient -auth.domain=https://testdomain -auth.jwkspath=./auth-public-jwks.json" source startup.sh
+  zapi -h $ZQD_HOST -configpath ./user1 auth store -access \
+    $(gentoken -domain https://testdomain -privatekeyfile ./auth-private-key -keyid testkey -tenantid tenant1 -userid user1)
+  zapi -h $ZQD_HOST -configpath ./user2 auth store -access \
+    $(gentoken -domain https://testdomain -privatekeyfile ./auth-private-key -keyid testkey -tenantid tenant2 -userid user2)
+  zapi -configpath ./user1 -h $ZQD_HOST auth verify
+  zapi -configpath ./user2 -h $ZQD_HOST auth verify
+  echo ===
+  # Create 2 spaces for each user, using the same name for one of the spaces,
+  # export their space id for later usage.
+  zapi -configpath ./user1 -h $ZQD_HOST new testsp0
+  export USER1_SP0=$(zapi -configpath ./user1 -h $ZQD_HOST info testsp0 | grep "id:" | head -n 1 | awk '{print $2}')
+  zapi -configpath ./user1 -h $ZQD_HOST new testsp1
+  export USER1_SP1=$(zapi -configpath ./user1 -h $ZQD_HOST info testsp1 | grep "id:" | head -n 1 | awk '{print $2}')
+  zapi -configpath ./user2 -h $ZQD_HOST new testsp0
+  export USER2_SP0=$(zapi -configpath ./user2 -h $ZQD_HOST info testsp0 | grep "id:" | head -n 1 | awk '{print $2}')
+  zapi -configpath ./user2 -h $ZQD_HOST new testsp2
+  export USER2_SP2=$(zapi -configpath ./user2 -h $ZQD_HOST info testsp2| grep "id:" | head -n 1 | awk '{print $2}')
+  echo ===
+  # List the spaces for each user, and verify that they see only the spaces
+  # that they created.
+  echo user1 ls
+  zapi -configpath ./user1 -h $ZQD_HOST ls
+  echo ===
+  echo user2 ls
+  zapi -configpath ./user2 -h $ZQD_HOST ls
+  echo ===
+  # User1 should not be able to access user2's space.
+  echo user1 info
+  zapi -configpath ./user1 -h $ZQD_HOST -id $USER2_SP0 info 2>&1
+  echo ===
+  # User1 should not be able to rename user2's space.
+  echo user1 rename
+  zapi -configpath ./user1 -h $ZQD_HOST -id $USER2_SP0 rename newname 2>&1
+  echo ===
+  # User1 should not be able to remove user2's space.
+  echo user1 rm
+  zapi -configpath ./user1 -h $ZQD_HOST -id $USER2_SP0 rm 2>&1
+
+
+inputs:
+  - name: startup.sh
+    source: startup.sh
+  - name: migrations
+    symlink: ../migrations
+  - name: auth-public-jwks.json
+    source: ../../../testdata/auth-public-jwks.json
+  - name: auth-private-key
+    source: ../../../testdata/auth-private-key
+
+outputs:
+  - name: stdout
+    data: |
+      {
+      	"tenant_id": "tenant1",
+      	"user_id": "user1"
+      }
+      {
+      	"tenant_id": "tenant2",
+      	"user_id": "user2"
+      }
+      ===
+      testsp0: space created
+      testsp1: space created
+      testsp0: space created
+      testsp2: space created
+      ===
+      user1 ls
+      testsp0   testsp1
+      ===
+      user2 ls
+      testsp0   testsp2
+      ===
+      user1 info
+      status code 403: forbidden
+      ===
+      user1 rename
+      status code 403: forbidden
+      ===
+      user1 rm
+      status code 403: forbidden
diff --git a/ppl/zqd/ztests/auth/tenant.yaml b/ppl/zqd/ztests/auth/tenant.yaml
new file mode 100644
index 000000000..a6b64f255
--- /dev/null
+++ b/ppl/zqd/ztests/auth/tenant.yaml
@@ -0,0 +1,82 @@
+script: |
+  # Start an authentication enabled zqd, and create 2 separate access tokens
+  # representing different users in different tenants.
+  ZQD_EXTRA_FLAGS="-auth.enabled=true -auth.clientid=testclient -auth.domain=https://testdomain -auth.jwkspath=./auth-public-jwks.json" source services.sh
+  zapi -h $ZQD_HOST -configpath ./user1 auth store -access \
+    $(gentoken -domain https://testdomain -privatekeyfile ./auth-private-key -keyid testkey -tenantid tenant1 -userid user1)
+  zapi -h $ZQD_HOST -configpath ./user2 auth store -access \
+    $(gentoken -domain https://testdomain -privatekeyfile ./auth-private-key -keyid testkey -tenantid tenant2 -userid user2)
+  zapi -configpath ./user1 -h $ZQD_HOST auth verify
+  zapi -configpath ./user2 -h $ZQD_HOST auth verify
+  echo ===
+  # Create 2 spaces for each user, using the same name for one of the spaces,
+  # export their space id for later usage.
+  zapi -configpath ./user1 -h $ZQD_HOST new testsp0
+  export USER1_SP0=$(zapi -configpath ./user1 -h $ZQD_HOST info testsp0 | grep "id:" | head -n 1 | awk '{print $2}')
+  zapi -configpath ./user1 -h $ZQD_HOST new testsp1
+  export USER1_SP1=$(zapi -configpath ./user1 -h $ZQD_HOST info testsp1 | grep "id:" | head -n 1 | awk '{print $2}')
+  zapi -configpath ./user2 -h $ZQD_HOST new testsp0
+  export USER2_SP0=$(zapi -configpath ./user2 -h $ZQD_HOST info testsp0 | grep "id:" | head -n 1 | awk '{print $2}')
+  zapi -configpath ./user2 -h $ZQD_HOST new testsp2
+  export USER2_SP2=$(zapi -configpath ./user2 -h $ZQD_HOST info testsp2| grep "id:" | head -n 1 | awk '{print $2}')
+  echo ===
+  # List the spaces for each user, and verify that they see only the spaces
+  # that they created.
+  echo user1 ls
+  zapi -configpath ./user1 -h $ZQD_HOST ls
+  echo ===
+  echo user2 ls
+  zapi -configpath ./user2 -h $ZQD_HOST ls
+  echo ===
+  # User1 should not be able to access user2's space.
+  echo user1 info
+  zapi -configpath ./user1 -h $ZQD_HOST -id $USER2_SP0 info 2>&1
+  echo ===
+  # User1 should not be able to rename user2's space.
+  echo user1 rename
+  zapi -configpath ./user1 -h $ZQD_HOST -id $USER2_SP0 rename newname 2>&1
+  echo ===
+  # User1 should not be able to remove user2's space.
+  echo user1 rm
+  zapi -configpath ./user1 -h $ZQD_HOST -id $USER2_SP0 rm 2>&1
+
+
+inputs:
+  - name: services.sh
+    source: ../services.sh
+  - name: auth-public-jwks.json
+    source: ../../testdata/auth-public-jwks.json
+  - name: auth-private-key
+    source: ../../testdata/auth-private-key
+
+outputs:
+  - name: stdout
+    data: |
+      {
+      	"tenant_id": "tenant1",
+      	"user_id": "user1"
+      }
+      {
+      	"tenant_id": "tenant2",
+      	"user_id": "user2"
+      }
+      ===
+      testsp0: space created
+      testsp1: space created
+      testsp0: space created
+      testsp2: space created
+      ===
+      user1 ls
+      testsp0   testsp1
+      ===
+      user2 ls
+      testsp0   testsp2
+      ===
+      user1 info
+      status code 403: forbidden
+      ===
+      user1 rename
+      status code 403: forbidden
+      ===
+      user1 rm
+      status code 403: forbidden
diff --git a/ppl/zqd/ztests/cluster/count.yaml b/ppl/zqd/ztests/cluster/count.yaml
new file mode 100644
index 000000000..1d1c004ac
--- /dev/null
+++ b/ppl/zqd/ztests/cluster/count.yaml
@@ -0,0 +1,11 @@
+tag: cluster
+
+script: |
+  # Test assumes a remote zqd instance with space "files" is available at 9867
+  time zapi -s files get -workers 4 -t 'count()' 2>&1
+
+outputs:
+  - name: stdout
+    data: |
+      #0:record[count:uint64]
+      0:[12243237;]
diff --git a/ppl/zqd/ztests/cluster/countbytxhost.yaml b/ppl/zqd/ztests/cluster/countbytxhost.yaml
new file mode 100644
index 000000000..14a888519
--- /dev/null
+++ b/ppl/zqd/ztests/cluster/countbytxhost.yaml
@@ -0,0 +1,15 @@
+tag: cluster
+
+script: |
+  # Test assumes a remote zqd instance with space "files" is available at 9867
+  zapi -s files get -workers 4 -t 'count() by tx_hosts | sort -r count | head 5' 2>&1
+
+outputs:
+  - name: stdout
+    data: |
+      #0:record[tx_hosts:set[ip],count:uint64]
+      0:[[10.47.2.156;]985198;]
+      0:[[10.47.7.58;]827462;]
+      0:[[10.47.5.58;]779783;]
+      0:[[10.47.8.58;]659903;]
+      0:[[10.47.1.58;]536395;]
diff --git a/ppl/zqd/ztests/cluster/filter1.yaml b/ppl/zqd/ztests/cluster/filter1.yaml
new file mode 100644
index 000000000..cf52f4bc4
--- /dev/null
+++ b/ppl/zqd/ztests/cluster/filter1.yaml
@@ -0,0 +1,11 @@
+tag: cluster
+
+script: |
+  # Test assumes a remote zqd instance with space "files" is available at 9867
+  zapi -s files get -workers 1 -t '"FgzbkQ15TOoOENn1lb" | sort tx_hosts' 2>&1 
+
+outputs:
+  - name: stdout
+    data: |
+      #0:record[_path:string,ts:time,fuid:bstring,tx_hosts:set[ip],rx_hosts:set[ip],conn_uids:set[bstring],source:bstring,depth:uint64,analyzers:set[bstring],mime_type:bstring,filename:bstring,duration:duration,local_orig:bool,is_orig:bool,seen_bytes:uint64,total_bytes:uint64,missing_bytes:uint64,overflow_bytes:uint64,timedout:bool,parent_fuid:bstring,md5:bstring,sha1:bstring,sha256:bstring,extracted:bstring,extracted_cutoff:bool,extracted_size:uint64]
+      0:[files;1521940131.094903;FgzbkQ15TOoOENn1lb;[198.199.95.203;][10.47.25.80;][Cc4q8L1Hhp9MjJiVcl;]SSL;0;[MD5;SHA1;X509;SHA256;]application/x-x509-ca-cert;-;0;-;F;1174;-;0;0;F;-;b15409274f54ad8f023d3b85a5ecec5d;e6a3b45b062d509b3382282d196efe97d5956ccb;25847d668eb4f04fdd40b12b6b0740c567da7d024308eb6c2c96fe41d9de218d;-;-;-;]
diff --git a/ppl/zqd/ztests/cluster/filter2.yaml b/ppl/zqd/ztests/cluster/filter2.yaml
new file mode 100644
index 000000000..9b5ba7cc1
--- /dev/null
+++ b/ppl/zqd/ztests/cluster/filter2.yaml
@@ -0,0 +1,11 @@
+tag: cluster
+
+script: |
+  # Test assumes a remote zqd instance with space "files" is available at 9867
+  zapi -s files get -workers 2 -t '"FgzbkQ15TOoOENn1lb" | sort tx_hosts' 2>&1 
+
+outputs:
+  - name: stdout
+    data: |
+      #0:record[_path:string,ts:time,fuid:bstring,tx_hosts:set[ip],rx_hosts:set[ip],conn_uids:set[bstring],source:bstring,depth:uint64,analyzers:set[bstring],mime_type:bstring,filename:bstring,duration:duration,local_orig:bool,is_orig:bool,seen_bytes:uint64,total_bytes:uint64,missing_bytes:uint64,overflow_bytes:uint64,timedout:bool,parent_fuid:bstring,md5:bstring,sha1:bstring,sha256:bstring,extracted:bstring,extracted_cutoff:bool,extracted_size:uint64]
+      0:[files;1521940131.094903;FgzbkQ15TOoOENn1lb;[198.199.95.203;][10.47.25.80;][Cc4q8L1Hhp9MjJiVcl;]SSL;0;[MD5;SHA1;X509;SHA256;]application/x-x509-ca-cert;-;0;-;F;1174;-;0;0;F;-;b15409274f54ad8f023d3b85a5ecec5d;e6a3b45b062d509b3382282d196efe97d5956ccb;25847d668eb4f04fdd40b12b6b0740c567da7d024308eb6c2c96fe41d9de218d;-;-;-;]
diff --git a/ppl/zqd/ztests/cluster/filter4.yaml b/ppl/zqd/ztests/cluster/filter4.yaml
new file mode 100644
index 000000000..8091ff32f
--- /dev/null
+++ b/ppl/zqd/ztests/cluster/filter4.yaml
@@ -0,0 +1,11 @@
+tag: cluster
+
+script: |
+  # Test assumes a remote zqd instance with space "files" is available at 9867
+  zapi -s files get -workers 4 -t '"FgzbkQ15TOoOENn1lb" | sort tx_hosts' 2>&1 
+
+outputs:
+  - name: stdout
+    data: |
+      #0:record[_path:string,ts:time,fuid:bstring,tx_hosts:set[ip],rx_hosts:set[ip],conn_uids:set[bstring],source:bstring,depth:uint64,analyzers:set[bstring],mime_type:bstring,filename:bstring,duration:duration,local_orig:bool,is_orig:bool,seen_bytes:uint64,total_bytes:uint64,missing_bytes:uint64,overflow_bytes:uint64,timedout:bool,parent_fuid:bstring,md5:bstring,sha1:bstring,sha256:bstring,extracted:bstring,extracted_cutoff:bool,extracted_size:uint64]
+      0:[files;1521940131.094903;FgzbkQ15TOoOENn1lb;[198.199.95.203;][10.47.25.80;][Cc4q8L1Hhp9MjJiVcl;]SSL;0;[MD5;SHA1;X509;SHA256;]application/x-x509-ca-cert;-;0;-;F;1174;-;0;0;F;-;b15409274f54ad8f023d3b85a5ecec5d;e6a3b45b062d509b3382282d196efe97d5956ccb;25847d668eb4f04fdd40b12b6b0740c567da7d024308eb6c2c96fe41d9de218d;-;-;-;]
diff --git a/ppl/zqd/ztests/common.sh b/ppl/zqd/ztests/common.sh
index 6adfa224b..7e3a6f037 100644
--- a/ppl/zqd/ztests/common.sh
+++ b/ppl/zqd/ztests/common.sh
@@ -7,10 +7,10 @@ function awaitfile {
     let i+=1
     if [ $i -gt 5 ]; then
       echo "timed out waiting for file \"$file\" to appear"
-      echo "minio log:"
-      cat minio.log
-      echo "zqd log:"
-      cat zqd.log
+      for f in *.log ; do
+        echo logfile: $f
+        cat $f
+      done
       exit 1
     fi
     sleep 1
diff --git a/ztest/shell.go b/ztest/shell.go
index 6268afadf..e1475ece1 100644
--- a/ztest/shell.go
+++ b/ztest/shell.go
@@ -23,6 +23,7 @@ func RunShell(dir *Dir, bindir, script string, stdin io.Reader, useenvs []string
 		}
 	}
 
+	cmd.Env = append(cmd.Env, "HOME="+dir.Path())
 	cmd.Env = append(cmd.Env, "PATH=/bin:/usr/bin:"+bindir)
 	cmd.Dir = dir.Path()
 	cmd.Stdin = stdin