Feature/periodically scan disk

adapters/handlers/rest/configure_api.go

@@ -132,9 +132,11 @@ func configureAPI(api *operations.WeaviateAPI) http.Handler {
 	// TODO: configure http transport for efficient intra-cluster comm
 	remoteIndexClient := clients.NewRemoteIndex(clusterHttpClient)
 	repo := db.New(appState.Logger, db.Config{
-		RootPath:            appState.ServerConfig.Config.Persistence.DataPath,
-		QueryLimit:          appState.ServerConfig.Config.QueryDefaults.Limit,
-		QueryMaximumResults: appState.ServerConfig.Config.QueryMaximumResults,
+		RootPath:                  appState.ServerConfig.Config.Persistence.DataPath,
+		QueryLimit:                appState.ServerConfig.Config.QueryDefaults.Limit,
+		QueryMaximumResults:       appState.ServerConfig.Config.QueryMaximumResults,
+		DiskUseWarningPercentage:  appState.ServerConfig.Config.DiskUse.WarningPercentage,
+		DiskUseReadOnlyPercentage: appState.ServerConfig.Config.DiskUse.ReadOnlyPercentage,
 	}, remoteIndexClient, appState.Cluster) // TODO client
 	vectorMigrator = db.NewMigrator(repo, appState.Logger)
 	vectorRepo = repo

adapters/repos/db/index.go

@@ -144,8 +144,10 @@ func (i *Index) updateVectorIndexConfig(ctx context.Context,
 }
 
 type IndexConfig struct {
-	RootPath  string
-	ClassName schema.ClassName
+	RootPath                  string
+	ClassName                 schema.ClassName
+	DiskUseWarningPercentage  uint64
+	DiskUseReadOnlyPercentage uint64
 }
 
 func indexID(class schema.ClassName) string {
@@ -908,3 +910,9 @@ func (i *Index) updateShardStatus(shardName, targetStatus string) error {
 
 	return shard.updateStatus(targetStatus)
 }
+
+func (i *Index) notifyReady() {
+	for _, shd := range i.Shards {
+		shd.notifyReady()
+	}
+}

adapters/repos/db/init.go

@@ -46,8 +46,10 @@ func (d *DB) init(ctx context.Context) error {
 			}
 
 			idx, err := NewIndex(ctx, IndexConfig{
-				ClassName: schema.ClassName(class.Class),
-				RootPath:  d.config.RootPath,
+				ClassName:                 schema.ClassName(class.Class),
+				RootPath:                  d.config.RootPath,
+				DiskUseWarningPercentage:  d.config.DiskUseWarningPercentage,
+				DiskUseReadOnlyPercentage: d.config.DiskUseReadOnlyPercentage,
 			}, d.schemaGetter.ShardingState(class.Class), invertedConfig,
 				class.VectorIndexConfig.(schema.VectorIndexConfig),
 				d.schemaGetter, d, d.logger, d.nodeResolver, d.remoteClient)

adapters/repos/db/migrator.go

@@ -31,8 +31,10 @@ func (m *Migrator) AddClass(ctx context.Context, class *models.Class,
 	shardState *sharding.State) error {
 	idx, err := NewIndex(ctx,
 		IndexConfig{
-			ClassName: schema.ClassName(class.Class),
-			RootPath:  m.db.config.RootPath,
+			ClassName:                 schema.ClassName(class.Class),
+			RootPath:                  m.db.config.RootPath,
+			DiskUseWarningPercentage:  m.db.config.DiskUseWarningPercentage,
+			DiskUseReadOnlyPercentage: m.db.config.DiskUseReadOnlyPercentage,
 		},
 		shardState,
 		// no backward-compatibility check required, since newly added classes will

adapters/repos/db/repo.go

@@ -35,7 +35,16 @@ func (d *DB) SetSchemaGetter(sg schemaUC.SchemaGetter) {
 }
 
 func (d *DB) WaitForStartup(ctx context.Context) error {
-	return d.init(ctx)
+	err := d.init(ctx)
+	if err != nil {
+		return err
+	}
+
+	for _, idx := range d.indices {
+		idx.notifyReady()
+	}
+
+	return nil
 }
 
 func New(logger logrus.FieldLogger, config Config,
@@ -50,9 +59,11 @@ func New(logger logrus.FieldLogger, config Config,
 }
 
 type Config struct {
-	RootPath            string
-	QueryLimit          int64
-	QueryMaximumResults int64
+	RootPath                  string
+	QueryLimit                int64
+	QueryMaximumResults       int64
+	DiskUseWarningPercentage  uint64
+	DiskUseReadOnlyPercentage uint64
 }
 
 // GetIndex returns the index if it exists or nil if it doesn't

adapters/repos/db/shard.go

@@ -49,7 +49,7 @@ type Shard struct {
 	propertyIndices  propertyspecific.Indices
 	deletedDocIDs    *docid.InMemDeletedTracker
 	cleanupInterval  time.Duration
-	cleanupCancel    chan struct{}
+	cancel           chan struct{}
 	propLengths      *inverted.PropertyLengthTracker
 	randomSource     *bufferedRandomGen
 	versioner        *shardVersioner
@@ -72,8 +72,8 @@ func NewShard(ctx context.Context, shardName string, index *Index) (*Shard, erro
 		deletedDocIDs:    docid.NewInMemDeletedTracker(),
 		cleanupInterval: time.Duration(index.invertedIndexConfig.
 			CleanupIntervalSeconds) * time.Second,
-		cleanupCancel: make(chan struct{}),
-		randomSource:  rand,
+		cancel:       make(chan struct{}, 1),
+		randomSource: rand,
 	}
 
 	hnswUserConfig, ok := index.vectorIndexUserConfig.(hnsw.UserConfig)
@@ -173,6 +173,8 @@ func (s *Shard) drop(force bool) error {
 		return storagestate.ErrStatusReadOnly
 	}
 
+	s.cancel <- struct{}{}
+
 	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
 	defer cancel()
 
@@ -302,9 +304,20 @@ func (s *Shard) updateVectorIndexConfig(ctx context.Context,
 }
 
 func (s *Shard) shutdown(ctx context.Context) error {
+	s.cancel <- struct{}{}
+
 	if err := s.propLengths.Close(); err != nil {
 		return errors.Wrap(err, "close prop length tracker")
 	}
 
 	return s.store.Shutdown(ctx)
 }
+
+func (s *Shard) notifyReady() {
+	s.initStatus()
+	s.index.logger.
+		WithField("action", "startup").
+		Debugf("shard=%s is ready", s.name)
+
+	s.scanDiskUse()
+}

adapters/repos/db/shard_disk_use.go

@@ -0,0 +1,95 @@
+package db
+
+import (
+	"fmt"
+	"syscall"
+	"time"
+
+	"github.com/semi-technologies/weaviate/entities/storagestate"
+)
+
+type diskUse struct {
+	total uint64
+	avail uint64
+}
+
+func (d diskUse) percentUsed() float64 {
+	used := d.total - d.avail
+	return (float64(used) / float64(d.total)) * 100
+}
+
+func (d diskUse) String() string {
+	GB := 1024 * 1024 * 1024
+
+	return fmt.Sprintf("total: %.2fGB, avail: %.2fGB, used: %.2fGB",
+		float64(d.total)/float64(GB),
+		float64(d.avail)/float64(GB),
+		float64(d.total-d.avail)/float64(GB))
+}
+
+func (s *Shard) scanDiskUse() {
+	go func() {
+		for {
+			t := time.Tick(time.Second * 30)
+			select {
+			case <-s.cancel:
+				return
+			case <-t:
+				fs := syscall.Statfs_t{}
+				diskPath := s.index.Config.RootPath
+
+				err := syscall.Statfs(diskPath, &fs)
+				if err != nil {
+					s.index.logger.WithField("action", "read_disk_use").
+						WithField("path", diskPath).
+						Fatalf("failed to read disk usage: %s", err)
+				}
+
+				du := diskUse{
+					total: fs.Blocks * uint64(fs.Bsize),
+					avail: fs.Bavail * uint64(fs.Bsize),
+				}
+
+				s.diskUseWarn(du, diskPath)
+				s.diskUseReadonly(du, diskPath)
+			}
+		}
+	}()
+}
+
+// logs a warning if user-set threshold is surpassed
+func (s *Shard) diskUseWarn(du diskUse, diskPath string) {
+	if pu := du.percentUsed(); pu > float64(s.index.Config.DiskUseWarningPercentage) {
+		if s.index.Config.DiskUseWarningPercentage > 0 {
+			s.index.logger.WithField("action", "read_disk_use").
+				WithField("shard", s.name).
+				WithField("path", diskPath).
+				Warnf("disk usage currently at %.2f%%", pu)
+
+			s.index.logger.WithField("action", "disk_use_stats").
+				WithField("shard", s.name).
+				WithField("path", diskPath).
+				Debugf("%s", du.String())
+		}
+	}
+}
+
+// sets the shard to readonly if user-set threshold is surpassed
+func (s *Shard) diskUseReadonly(du diskUse, diskPath string) {
+	if pu := du.percentUsed(); pu > float64(s.index.Config.DiskUseReadOnlyPercentage) {
+		if !s.isReadOnly() && s.index.Config.DiskUseReadOnlyPercentage > 0 {
+			err := s.updateStatus(storagestate.StatusReadOnly.String())
+			if err != nil {
+				s.index.logger.WithField("action", "set_shard_read_only").
+					WithField("shard", s.name).
+					WithField("path", s.index.Config.RootPath).
+					Fatal("failed to set to READONLY")
+			}
+
+			s.index.logger.WithField("action", "set_shard_read_only").
+				WithField("shard", s.name).
+				WithField("path", diskPath).
+				Warnf("disk usage currently at %.2f%%, %s set READONLY", pu, s.name)
+		}
+	}
+}

adapters/repos/db/shard_status.go

@@ -18,6 +18,13 @@ import (
 	"github.com/semi-technologies/weaviate/entities/storagestate"
 )
 
+func (s *Shard) initStatus() {
+	s.statusLock.Lock()
+	defer s.statusLock.Unlock()
+
+	s.status = storagestate.StatusReady
+}
+
 func (s *Shard) getStatus() storagestate.Status {
 	s.statusLock.Lock()
 	defer s.statusLock.Unlock()

adapters/repos/db/shard_test.go

@@ -9,6 +9,9 @@
 //  CONTACT: hello@semi.technology
 //
 
+//go:build integrationTest
+// +build integrationTest
+
 package db
 
 import (

go.mod

@@ -1,35 +1,25 @@
 module github.com/semi-technologies/weaviate
 
 require (
-	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d // indirect
-	github.com/aws/aws-sdk-go v1.34.28 // indirect
 	github.com/bmatcuk/doublestar v1.1.3
 	github.com/buger/jsonparser v1.1.1
 	github.com/coreos/go-oidc v2.0.0+incompatible
 	github.com/danaugrs/go-tsne v0.0.0-20200708172100-6b7d1d577fd3
 	github.com/davecgh/go-spew v1.1.1
 	github.com/fatih/camelcase v1.0.0
-	github.com/globalsign/mgo v0.0.0-20181015135952-eeefdecb41b8 // indirect
 	github.com/go-openapi/errors v0.20.2
 	github.com/go-openapi/loads v0.21.1
 	github.com/go-openapi/runtime v0.23.2
 	github.com/go-openapi/spec v0.20.4
 	github.com/go-openapi/strfmt v0.21.2
 	github.com/go-openapi/swag v0.21.1
 	github.com/go-openapi/validate v0.21.0
-	github.com/go-stack/stack v1.8.1 // indirect
 	github.com/golang-jwt/jwt/v4 v4.0.0
 	github.com/google/uuid v1.2.0
 	github.com/graphql-go/graphql v0.7.9
 	github.com/hashicorp/memberlist v0.2.4
 	github.com/jessevdk/go-flags v1.4.0
-	github.com/kisielk/errcheck v1.2.0 // indirect
-	github.com/kr/pretty v0.3.0 // indirect
-	github.com/kr/pty v1.1.5 // indirect
-	github.com/mailru/easyjson v0.7.7 // indirect
-	github.com/mitchellh/mapstructure v1.4.3 // indirect
 	github.com/nyaruka/phonenumbers v1.0.54
-	github.com/pborman/uuid v1.2.0 // indirect
 	github.com/pkg/errors v0.9.1
 	github.com/pquerna/cachecontrol v0.0.0-20201205024021-ac21108117ac // indirect
 	github.com/rs/cors v1.5.0
@@ -39,17 +29,14 @@ require (
 	github.com/square/go-jose v2.3.0+incompatible
 	github.com/stretchr/objx v0.2.0 // indirect
 	github.com/stretchr/testify v1.7.0
-	github.com/vektah/gqlparser v1.1.2 // indirect
 	github.com/willf/bitset v1.1.11 // indirect
 	github.com/willf/bloom v2.0.3+incompatible
-	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c // indirect
-	github.com/xdg/stringprep v0.0.0-20180714160509-73f8eece6fdc // indirect
 	go.etcd.io/bbolt v1.3.5
 	go.mongodb.org/mongo-driver v1.8.4 // indirect
+	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 // indirect
 	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e
-	golang.org/x/tools v0.1.10 // indirect
 	gonum.org/v1/gonum v0.9.1
 	google.golang.org/grpc v1.24.0
 	gopkg.in/square/go-jose.v2 v2.5.1 // indirect