feat: Coordinated quota

Quota package made composable and broken down into:
  * Node quota (total limits one node can handle)
  * Storage quota (maximum namespace size quots)
  * Per cluster namespace quota, with Datadog backend

Storage quota, size metrics and namespace rate updates
moved to background threads.

Observability code is generalized and Datadog backend
is used by both O11y and Quota.

Unrelated tests fixes

.golangci.yaml

@@ -1,61 +1,15 @@
 linters:
-  enable:
-    - asasalint
-    - asciicheck
-    - bidichk
-    - deadcode
-    - decorder
-    - depguard
-    - dogsled
-    - errcheck
-    - errname
-    - execinquery
-    - exportloopref
-    - forbidigo
-    - gci
-    - godot
-    - gofmt
-    - goheader
-    - goimports
-    - gomoddirectives
-    - gomodguard
-    - goprintffuncname
-    - gosimple
-    - govet
-    - grouper
-    - importas
-    - ineffassign
-    - misspell
-    - nolintlint
-    - promlinter
-    - staticcheck
-    - tenv
-    - tparallel
-    - typecheck
-    - unconvert
-    - unused
-    - usestdlibvars
-    - varcheck
-    - whitespace
-    - gofumpt
-    - gosec
-    - prealloc
-    - errchkjson
-    - gocritic
+  enable-all: true
   disable:
     # Work on fixing this
+    #- reassign
     - maintidx
-    - unparam
     - nakedret
-    - makezero
     - gocyclo
-    - durationcheck
-    - bodyclose
     - gocognit
     - funlen
     - goconst
     - nestif
-    - noctx
     - dupl
     - containedctx
     - cyclop
@@ -66,7 +20,6 @@ linters:
     - exhaustive
     - thelper
     - nonamedreturns
-    - predeclared
     - nilnil
     - godox
     - goerr113
@@ -76,6 +29,7 @@ linters:
     - revive
     - stylecheck
     - wsl
+    #- interfacebloat
 
     # Not working with generics
     # enable in the future
@@ -86,6 +40,8 @@ linters:
     - wastedassign
 
     # These are two strict or deprecated
+    - deadcode
+    - varcheck
     - testpackage
     - paralleltest
     - interfacer

Makefile

@@ -41,7 +41,7 @@ lint: generate
 	yq --exit-status 'tag == "!!map" or tag== "!!seq"' .github/workflows/*.yaml config/*.yaml >/dev/null
 	shellcheck scripts/*
 	shellcheck test/docker/grafana/*
-	golangci-lint --timeout=$(LINT_TIMEOUT) run
+	golangci-lint --timeout=$(LINT_TIMEOUT) run --fix
 
 # dependency on generate needed to create generated file outside of docker with
 # current user owner instead of root

api/server/v1/tx.go

@@ -20,25 +20,40 @@ import (
 	"google.golang.org/grpc"
 )
 
-var (
-	methodPrefix             = "/tigrisdata.v1.Tigris/"
-	insert                   = methodPrefix + "Insert"
-	replace                  = methodPrefix + "Replace"
-	update                   = methodPrefix + "Update"
-	delete                   = methodPrefix + "Delete"
-	read                     = methodPrefix + "Read"
-	commitTransaction        = methodPrefix + "CommitTransaction"
-	rollbackTransaction      = methodPrefix + "RollbackTransaction"
-	dropCollection           = methodPrefix + "DropCollection"
-	listCollection           = methodPrefix + "ListCollections"
-	createOrUpdateCollection = methodPrefix + "CreateOrUpdateCollection"
+const (
+	methodPrefix = "/tigrisdata.v1.Tigris/"
+
+	InsertMethodName  = methodPrefix + "Insert"
+	ReplaceMethodName = methodPrefix + "Replace"
+	UpdateMethodName  = methodPrefix + "Update"
+	DeleteMethodName  = methodPrefix + "Delete"
+	ReadMethodName    = methodPrefix + "Read"
+
+	SearchMethodName = methodPrefix + "Search"
+
+	SubscribeMethodName = methodPrefix + "Subscribe"
+
+	EventsMethodName = methodPrefix + "Events"
+
+	CommitTransactionMethodName   = methodPrefix + "CommitTransaction"
+	RollbackTransactionMethodName = methodPrefix + "RollbackTransaction"
+
+	CreateOrUpdateCollectionMethodName = methodPrefix + "CreateOrUpdateCollection"
+	DropCollectionMethodName           = methodPrefix + "DropCollection"
+
+	ListDatabasesMethodName   = methodPrefix + "ListDatabases"
+	ListCollectionsMethodName = methodPrefix + "ListCollections"
+
+	DescribeDatabaseMethodName   = methodPrefix + "DescribeDatabase"
+	DescribeCollectionMethodName = methodPrefix + "DescribeCollection"
 )
 
 func IsTxSupported(ctx context.Context) bool {
 	m, _ := grpc.Method(ctx)
 	switch m {
-	case insert, replace, update, delete, read, commitTransaction, rollbackTransaction,
-		dropCollection, listCollection, createOrUpdateCollection:
+	case InsertMethodName, ReplaceMethodName, UpdateMethodName, DeleteMethodName, ReadMethodName,
+		CommitTransactionMethodName, RollbackTransactionMethodName,
+		DropCollectionMethodName, ListCollectionsMethodName, CreateOrUpdateCollectionMethodName:
 		return true
 	default:
 		return false

go.mod

@@ -3,6 +3,7 @@ module github.com/tigrisdata/tigris
 go 1.18
 
 require (
+	github.com/DataDog/datadog-api-client-go v1.16.0
 	github.com/apple/foundationdb/bindings/go v0.0.0-20220521054011-a88e049b28d8
 	github.com/auth0/go-auth0 v0.9.3
 	github.com/auth0/go-jwt-middleware/v2 v2.0.1
@@ -51,6 +52,7 @@ require (
 	github.com/DataDog/datadog-go/v5 v5.1.1 // indirect
 	github.com/DataDog/gostackparse v0.5.0 // indirect
 	github.com/DataDog/sketches-go v1.4.1 // indirect
+	github.com/DataDog/zstd v1.5.0 // indirect
 	github.com/Microsoft/go-winio v0.5.2 // indirect
 	github.com/PuerkitoBio/rehttp v1.1.0 // indirect
 	github.com/ajg/form v1.5.1 // indirect

go.sum

@@ -72,6 +72,8 @@ github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym
 github.com/DataDog/datadog-agent/pkg/obfuscate v0.0.0-20211129110424-6491aa3bf583/go.mod h1:EP9f4GqaDJyP1F5jTNMtzdIpw3JpNs3rMSJOnYywCiw=
 github.com/DataDog/datadog-agent/pkg/obfuscate v0.38.0 h1:Mk1GqUitDrsZBvNIR4G/GF1HUw3gegs8h38/rtv0Els=
 github.com/DataDog/datadog-agent/pkg/obfuscate v0.38.0/go.mod h1:MxVcCIC42tBIjPm93BHdh9/vw2LivRiptj3HygI+GGQ=
+github.com/DataDog/datadog-api-client-go v1.16.0 h1:5jOZv1m98criCvYTa3qpW8Hzv301nbZX3K9yJtwGyWY=
+github.com/DataDog/datadog-api-client-go v1.16.0/go.mod h1:PgrP2ABuJWL3Auw2iEkemAJ/r72ghG4DQQmb5sgnKW4=
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
 github.com/DataDog/datadog-go v4.8.2+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
 github.com/DataDog/datadog-go/v5 v5.0.2/go.mod h1:ZI9JFB4ewXbw1sBnF4sxsR2k1H3xjV+PUAOUsHvKpcU=
@@ -84,6 +86,8 @@ github.com/DataDog/sketches-go v1.2.1/go.mod h1:1xYmPLY1So10AwxV6MJV0J53XVH+WL9A
 github.com/DataDog/sketches-go v1.4.1 h1:j5G6as+9FASM2qC36lvpvQAj9qsv/jUs3FtO8CwZNAY=
 github.com/DataDog/sketches-go v1.4.1/go.mod h1:xJIXldczJyyjnbDop7ZZcLxJdV3+7Kra7H1KMgpgkLk=
 github.com/DataDog/zstd v1.3.5/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo=
+github.com/DataDog/zstd v1.5.0 h1:+K/VEwIAaPcHiMtQvpLD4lqW7f0Gk3xdYZmI1hD+CXo=
+github.com/DataDog/zstd v1.5.0/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
 github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=
 github.com/Microsoft/go-winio v0.5.0/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
 github.com/Microsoft/go-winio v0.5.1/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=

query/filter/key_builder.go

@@ -135,8 +135,8 @@ func (s *StrictEqKeyComposer) Compose(selectors []*Selector, userDefinedKeys []*
 		for j := 1; j < len(repeatedFields); j++ {
 			keyPartsCopy := make([]*Selector, len(compositeKeys[0])-1)
 			copy(keyPartsCopy, compositeKeys[0][0:len(compositeKeys[0])-1])
-			keyPartsCopy = append(keyPartsCopy, repeatedFields[j])
-			compositeKeys = append(compositeKeys, keyPartsCopy)
+			keyPartsCopy = append(keyPartsCopy, repeatedFields[j]) //nolint:golint,makezero
+			compositeKeys = append(compositeKeys, keyPartsCopy)    //nolint:golint,makezero
 		}
 	}
 

server/config/config.go

@@ -16,9 +16,12 @@ package config
 
 import (
 	"bytes"
+	"context"
+	"fmt"
 	"os"
 	"os/exec"
 	"strings"
+	"time"
 
 	"github.com/davecgh/go-spew/spew"
 	"github.com/fsnotify/fsnotify"
@@ -148,13 +151,17 @@ func GetTestFDBConfig(path string) (*FoundationDBConfig, error) {
 		fn = path + "/test/config/fdb.cluster"
 	}
 
-	cmd := exec.Command("fdbcli", "-C", fn, "--exec", "configure new single memory")
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, "fdbcli", "-C", fn, "--exec", "configure new single memory")
 	_, err := cmd.Output()
 	if err != nil {
-		cmd := exec.Command("fdbcli", "-C", fn, "--exec", "configure single memory")
+		cmd := exec.CommandContext(ctx, "fdbcli", "-C", fn, "--exec", "configure single memory")
 		_, err = cmd.Output()
 	}
 	if err != nil {
+		fmt.Printf("\nRun `make local_run` in the terminal to start FDB instance for tests\n") //nolint:golint,forbidigo
 		return nil, err
 	}
 

server/config/options.go

@@ -166,6 +166,11 @@ type ObservabilityConfig struct {
 	ProviderUrl string `mapstructure:"provider_url" yaml:"provider_url" json:"provider_url"`
 }
 
+var (
+	WriteUnitSize = 1024
+	ReadUnitSize  = 4096
+)
+
 var DefaultConfig = Config{
 	Log: log.LogConfig{
 		Level:      "info",
@@ -275,17 +280,38 @@ var DefaultConfig = Config{
 		EnableHeap: true,
 	},
 	Quota: QuotaConfig{
-		Enabled:                   false,
-		RateLimit:                 1000,        // requests per second both reads and writes
-		WriteThroughputLimit:      10000000,    // bytes per second
-		ReadThroughputLimit:       10000000,    // bytes per second
-		DataSizeLimit:             10000000000, // bytes
-		LimitUpdateInterval:       5,           // seconds
-		TenantSizeRefreshInterval: 60,          // seconds
-		AllTenantsRefreshInternal: 300,         // seconds
+		// Maximum limits single node can handle. Across namespaces.
+		Node: LimitsConfig{
+			Enabled: false,
+
+			ReadUnits:  4000, // read requests per second per namespace
+			WriteUnits: 1000, // write requests per second
+		},
+		Namespace: NamespaceLimitsConfig{
+			// Per cluster limits for single namespace
+			Default: LimitsConfig{
+				Enabled: false,
+
+				ReadUnits:  100, // read requests per second per namespace
+				WriteUnits: 25,  // write requests per second
+			},
+			// Maximum per node quota for single namespace
+			Node: LimitsConfig{
+				ReadUnits:  100, // read requests per second per namespace
+				WriteUnits: 25,  // write requests per second
+			},
+			RefreshInterval: 60 * time.Second,
+		},
+		Storage: StorageLimitsConfig{
+			Enabled:         false,
+			DataSizeLimit:   100 * 1024 * 1024,
+			RefreshInterval: 60 * time.Second,
+		},
 	},
 	Observability: ObservabilityConfig{
-		Enabled: false,
+		Enabled:     false,
+		Provider:    "datadog",
+		ProviderUrl: "https://us3.datadoghq.com",
 	},
 	Management: ManagementConfig{
 		Enabled: true,
@@ -305,17 +331,48 @@ type SearchConfig struct {
 	WriteEnabled bool   `mapstructure:"write_enabled" yaml:"write_enabled" json:"write_enabled"`
 }
 
+type LimitsConfig struct {
+	Enabled bool
+
+	ReadUnits  int `mapstructure:"read_units" yaml:"read_units" json:"read_units"`
+	WriteUnits int `mapstructure:"write_units" yaml:"write_units" json:"write_units"`
+}
+
+type NamespaceLimitsConfig struct {
+	Enabled    bool
+	Default    LimitsConfig            // default per namespace limit
+	Node       LimitsConfig            // max per node per namespace limit
+	Namespaces map[string]LimitsConfig // individual namespaces configuration
+
+	RefreshInterval time.Duration `mapstructure:"refresh_interval" yaml:"refresh_interval" json:"refresh_interval"`
+}
+
+func (n *NamespaceLimitsConfig) NamespaceLimits(ns string) *LimitsConfig {
+	cfg, ok := n.Namespaces[ns]
+	if ok {
+		return &cfg
+	}
+	return &n.Default
+}
+
+type StorageLimitsConfig struct {
+	Enabled         bool
+	DataSizeLimit   int64         `mapstructure:"data_size_limit" yaml:"data_size_limit" json:"data_size_limit"`
+	RefreshInterval time.Duration `mapstructure:"refresh_interval" yaml:"refresh_interval" json:"refresh_interval"`
+
+	// Per namespace limits
+	Namespaces map[string]int64
+}
+
 type QuotaConfig struct {
-	Enabled                   bool
-	RateLimit                 int   `mapstructure:"rate_limit" yaml:"rate_limit" json:"rate_limit"`
-	WriteThroughputLimit      int   `mapstructure:"write_throughput_limit" yaml:"write_throughput_limit" json:"write_throughput_limit"`
-	ReadThroughputLimit       int   `mapstructure:"read_throughput_limit" yaml:"read_throughput_limit" json:"read_throughput_limit"`
-	DataSizeLimit             int64 `mapstructure:"data_size_limit" yaml:"data_size_limit" json:"data_size_limit"`
-	LimitUpdateInterval       int64 `mapstructure:"limit_update_interval" yaml:"limit_update_interval" json:"limit_update_interval"`
-	TenantSizeRefreshInterval int64 `mapstructure:"tenant_size_refresh_interval" yaml:"tenant_size_refresh_interval" json:"tenant_size_refresh_interval"`
-	AllTenantsRefreshInternal int64 `mapstructure:"all_tenants_refresh_interval" yaml:"all_tenants_refresh_interval" json:"all_tenants_refresh_interval"`
+	Node      LimitsConfig          // maximum rates per node. protects the node from overloading
+	Namespace NamespaceLimitsConfig // user quota across all the nodes
+	Storage   StorageLimitsConfig
+
+	WriteUnitSize int
+	ReadUnitSize  int
 }
 
-func IsIndexingStoreReadEnabled() bool {
-	return DefaultConfig.Search.WriteEnabled && DefaultConfig.Search.ReadEnabled
+func (s *SearchConfig) IsReadEnabled() bool {
+	return s.WriteEnabled && s.ReadEnabled
 }

server/main.go

@@ -80,18 +80,19 @@ func mainWithCode() int {
 		return 1
 	}
 
-	tenantMgr := metadata.NewTenantManager(kvStore, searchStore)
-	log.Info().Msg("initialized tenant manager")
-
 	txMgr := transaction.NewManager(kvStore)
 	log.Info().Msg("initialized transaction manager")
 
-	if err = tenantMgr.EnsureDefaultNamespace(txMgr); err != nil {
+	tenantMgr := metadata.NewTenantManager(kvStore, searchStore, txMgr)
+	log.Info().Msg("initialized tenant manager")
+
+	if err = tenantMgr.EnsureDefaultNamespace(); err != nil {
 		log.Error().Err(err).Msg("error initializing default namespace")
 		return 1
 	}
 
-	quota.Init(tenantMgr, txMgr, &config.DefaultConfig.Quota)
+	_ = quota.Init(tenantMgr, &config.DefaultConfig)
+	defer quota.Cleanup()
 
 	mx := muxer.NewMuxer(&config.DefaultConfig)
 	mx.RegisterServices(kvStore, searchStore, tenantMgr, txMgr)