Enhancement/disk utilization (#642)

* Add config and worker for blobber updates * Manage files as per its lookuphash * Use contenthash instead of lookuphash * Replace file path calculation with new logic Add temporary directory and files * Fix file delete while moving file to cloud Multiple files can have same content hash within an allocation. Remove iterProgress use Remove . import * Modify filestore setup * Add/modify/optimize file operations as per file manager * Reorder functions and implement lock for file write Modify tests * Restructure setup for testing Add mock for file manager * Check if a path is a mountpoint * Fix flag and mount point check for integration * Fix null scan error to int64 value db.commit to unblock postgres transaction * Add negating build tag to unit tests * Revert use of build tags * Fix/Modify download integration test * Skip integration test if not set in flag * Add build tags to separate integration and unit test partially * Distinguish from integration test Put common function in new file * Update comment, delete unused function Add nil checking for integration testing * Add minor code optimization * Validate actual file content hash w.r.t. clien'ts content hash * Delete connection object after its usage * Optimize file write to temporary directory * Remove some code redundancy * Readability change * Modify config for minio setup * Reorganize/Reorder storage related code * Reorder argument of Filestorer interface * Add temporary fix for tests * Change flag name * Update test * Add filestore mocker and update accordingly * Add file store mocker for tests in allocation package * Add unit test for filestore * Add/Modify/Fix/Optimize code * Fix commit/integration tests issues * Add/Optimize update blobber capacity and other functions * Add tag * Delete unnecessary file * Fix variable name * Remove unused function * Relax mountpoint requirement * Fix mountpoint check temporarily * Modify yaml file for custom docker build * Modify content hash calculation and size increment * Fix size update * Fix for padding less data * Move to files_dir flag temporarily for system testing * Fix file store unit test * Modify comparison to length * Fix validation/test * Change back to what staging has * Change error message * Calculate different content hash for thumbnail * Skip thumbnail hash checking * Fix hash assignment issue * Add comments and modify field * Disable lint check * Rename function * Use helper function to check if test is integration * Fix return issues Remove comment * Rename functions * Remove defer for smaller code blocks * Modify viper's config retrieval function call * Modify function to return error and avoid using go routine * Fix double unlock call * Query in batches rather than single query * Fix lint issue * Fix address issue * Set allocations in batch * Rename function

Author: lpoli

.github/workflows/tests.yml

@@ -103,4 +103,4 @@ jobs:
  #sudo make integration-tests 
  go=$(which go)
  root=$(pwd)
- sudo CGO_ENABLED=1 root=$root integration=1 $go test -tags bn256 ./...
+ sudo CGO_ENABLED=1 root=$root integration=1 $go test -tags bn256 -tags=integration  ./...

code/go/0chain.net/blobber/config.go

@@ -20,6 +20,17 @@ func setupConfig(configDir string, deploymentMode int) {
 // setup config file
 config.SetupConfig(configDir)
 
+if mountPoint != "" {
+config.Configuration.MountPoint = mountPoint
+} else {
+config.Configuration.MountPoint = viper.GetString("storage.files_dir")
+}
+
+if config.Configuration.MountPoint == "" {
+panic("Please specify mount point in flag or config file")
+}
+config.Configuration.AllocDirLevel = viper.GetIntSlice("storage.alloc_dir_level")
+config.Configuration.FileDirLevel = viper.GetIntSlice("storage.file_dir_level")
 config.Configuration.DeploymentMode = byte(deploymentMode)
 config.Configuration.ChainID = viper.GetString("server_chain.id")
 config.Configuration.SignatureScheme = viper.GetString("server_chain.signature_scheme")
@@ -40,16 +51,28 @@ func setupConfig(configDir string, deploymentMode int) {
 config.Configuration.ChallengeResolveNumWorkers = viper.GetInt("challenge_response.num_workers")
 config.Configuration.ChallengeMaxRetires = viper.GetInt("challenge_response.max_retries")
 
+config.Configuration.AutomaticUpdate = viper.GetBool("disk_update.automatic_update")
+blobberUpdateIntrv := viper.GetDuration("disk_update.blobber_update_interval")
+if blobberUpdateIntrv <= 0 {
+blobberUpdateIntrv = 5 * time.Minute
+}
+config.Configuration.BlobberUpdateInterval = blobberUpdateIntrv
+
 config.Configuration.ColdStorageMinimumFileSize = viper.GetInt64("cold_storage.min_file_size")
 config.Configuration.ColdStorageTimeLimitInHours = viper.GetInt64("cold_storage.file_time_limit_in_hours")
-config.Configuration.ColdStorageJobQueryLimit = viper.GetInt64("cold_storage.job_query_limit")
-config.Configuration.ColdStorageStartCapacitySize = viper.GetInt64("cold_storage.start_capacity_size")
+config.Configuration.ColdStorageJobQueryLimit = viper.GetInt("cold_storage.job_query_limit")
+config.Configuration.ColdStorageStartCapacitySize = viper.GetUint64("cold_storage.start_capacity_size")
 config.Configuration.ColdStorageDeleteLocalCopy = viper.GetBool("cold_storage.delete_local_copy")
 config.Configuration.ColdStorageDeleteCloudCopy = viper.GetBool("cold_storage.delete_cloud_copy")
 
 config.Configuration.MinioStart = viper.GetBool("minio.start")
 config.Configuration.MinioWorkerFreq = viper.GetInt64("minio.worker_frequency")
 config.Configuration.MinioUseSSL = viper.GetBool("minio.use_ssl")
+config.Configuration.MinioStorageUrl = viper.GetString("minio.storage_service_url")
+config.Configuration.MinioAccessID = viper.GetString("minio.access_id")
+config.Configuration.MinioSecretKey = viper.GetString("minio.secret_access_key")
+config.Configuration.MinioBucket = viper.GetString("minio.bucket_name")
+config.Configuration.MinioRegion = viper.GetString("minio.region")
 
 config.Configuration.DBAutoMigrate = viper.GetBool("db.automigrate")
 config.Configuration.PGUserName = viper.GetString("pg.user")

code/go/0chain.net/blobber/filestore.go

@@ -6,14 +6,22 @@ import (
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/filestore"
 )
 
-var fsStore filestore.FileStore //nolint:unused // global which might be needed somewhere
-
 func setupFileStore() (err error) {
 fmt.Print("> setup file store")
+var fs filestore.FileStorer
+if isIntegrationTest {
+fs = &filestore.MockStore{}
+} else {
+fs = &filestore.FileStore{}
+
+}
 
-fsStore, err = filestore.SetupFSStore(filesDir + "/files")
+err = fs.Initialize()
+if err != nil {
+return
+}
 
-fmt.Print("	[OK]\n")
+filestore.SetFileStore(fs)
 
-return err
+return nil
 }

code/go/0chain.net/blobber/flags.go

@@ -10,7 +10,7 @@ var (
 deploymentMode int
 keysFile string
 minioFile string
-filesDir  string
+mountPoint string
 metadataDB string
 logDir string
 httpPort int
@@ -28,7 +28,7 @@ func init() {
 flag.IntVar(&deploymentMode, "deployment_mode", 2, "deployment mode: 0=dev,1=test, 2=mainnet")
 flag.StringVar(&keysFile, "keys_file", "", "keys_file")
 flag.StringVar(&minioFile, "minio_file", "", "minio_file")
-flag.StringVar(&filesDir, "files_dir", "", "files_dir")
+flag.StringVar(&mountPoint, "files_dir", "", "Mounted partition where all files will be stored")
 flag.StringVar(&metadataDB, "db_dir", "", "db_dir")
 flag.StringVar(&logDir, "log_dir", "", "log_dir")
 flag.IntVar(&httpPort, "port", 0, "port")
@@ -47,10 +47,6 @@ func parseFlags() {
 fmt.Print("> load flags")
 flag.Parse()
 
-if filesDir == "" {
-panic("Please specify --files_dir absolute folder name option where uploaded files can be stored")
-}
-
 if metadataDB == "" {
 panic("Please specify --db_dir absolute folder name option where meta data db can be stored")
 }

code/go/0chain.net/blobber/main.go

@@ -23,11 +23,6 @@ func main() {
 panic(err)
 }
 
-if err := setupMinio(); err != nil {
-logging.Logger.Error("Error setting up minio " + err.Error())
-panic(err)
-}
-
 if err := setupNode(); err != nil {
 logging.Logger.Error("Error setting up blobber node " + err.Error())
 panic(err)

code/go/0chain.net/blobber/minio.go

@@ -1,12 +1,14 @@
 package main
 
 import (
+"context"
 "time"
 
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/allocation"
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/challenge"
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/config"
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/datastore"
+"github.com/0chain/blobber/code/go/0chain.net/blobbercore/filestore"
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/handler"
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/readmarker"
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/writemarker"
@@ -22,8 +24,10 @@ func setupWorkers() {
 challenge.SetupWorkers(root)
 readmarker.SetupWorkers(root)
 writemarker.SetupWorkers(root)
-allocation.StartUpdateWorker(root,
-config.Configuration.UpdateAllocationsInterval)
+allocation.StartUpdateWorker(root, config.Configuration.UpdateAllocationsInterval)
+if config.Configuration.AutomaticUpdate {
+go StartUpdateWorker(root, config.Configuration.BlobberUpdateInterval)
+}
 }
 
 func refreshPriceOnChain() {
@@ -64,3 +68,33 @@ func startRefreshSettings() {
 <-time.After(REPEAT_DELAY * time.Second)
 }
 }
+
+func StartUpdateWorker(ctx context.Context, interval time.Duration) {
+err := filestore.GetFileStore().CalculateCurrentDiskCapacity()
+if err != nil {
+panic(err)
+}
+currentCapacity := filestore.GetFileStore().GetCurrentDiskCapacity()
+
+ticker := time.NewTicker(config.Configuration.BlobberUpdateInterval)
+for {
+select {
+case <-ctx.Done():
+return
+case <-ticker.C:
+err := filestore.GetFileStore().CalculateCurrentDiskCapacity()
+if err != nil {
+logging.Logger.Error("Error while getting capacity", zap.Error(err))
+break
+}
+if currentCapacity != filestore.GetFileStore().GetCurrentDiskCapacity() {
+
+err := handler.UpdateBlobberOnChain(ctx)
+if err != nil {
+logging.Logger.Error("Error while updating blobber updates on chain", zap.Error(err))
+}
+}
+}
+}
+
+}

code/go/0chain.net/blobber/worker.go

@@ -5,6 +5,7 @@ import (
 "time"
 
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/config"
+"github.com/0chain/blobber/code/go/0chain.net/blobbercore/filestore"
 "github.com/0chain/blobber/code/go/0chain.net/blobbercore/handler"
 "github.com/0chain/blobber/code/go/0chain.net/core/chain"
 "github.com/0chain/blobber/code/go/0chain.net/core/common"
@@ -23,42 +24,46 @@ func setupOnChain() {
 fmt.Print("	+ connect to miners: ")
 if isIntegrationTest {
 fmt.Print("	[SKIP]\n")
-} else {
-if err := handler.WalletRegister(); err != nil {
-fmt.Println(err.Error() + "\n")
-panic(err)
-}
-fmt.Print("	[OK]\n")
+return
 }
 
+if err := handler.WalletRegister(); err != nil {
+fmt.Println(err.Error() + "\n")
+panic(err)
+}
+fmt.Print("	[OK]\n")
+
+var success bool
+var err error
 // setup blobber (add or update) on the blockchain (multiple attempts)
 for i := 1; i <= 10; i++ {
-if i == 1 {
-fmt.Printf("\r+ connect to sharders:")
-} else {
-fmt.Printf("\r+ [%v/10]connect to sharders:", i)
+fmt.Printf("\r+ [%v/10]connect to sharders:", i)
+if err = filestore.GetFileStore().CalculateCurrentDiskCapacity(); err != nil {
+fmt.Print("\n", err.Error()+"\n")
+goto sleep
 }
 
-if isIntegrationTest {
-fmt.Print("	[SKIP]\n")
-break
-} else {
-if err := handler.RegisterBlobber(common.GetRootContext()); err != nil {
-if i == 10 { // no more attempts
-panic(err)
-}
-fmt.Print("\n", err.Error()+"\n")
-} else {
-fmt.Print("	[OK]\n")
-break
-}
-for n := 0; n < ATTEMPT_DELAY; n++ {
-<-time.After(1 * time.Second)
-
-fmt.Printf("\r- wait %v seconds to retry", ATTEMPT_DELAY-n)
-}
+if err = handler.RegisterBlobber(common.GetRootContext()); err != nil {
+fmt.Print("\n", err.Error()+"\n")
+goto sleep
+}
+
+fmt.Print("	[OK]\n")
+success = true
+break
+
+sleep:
+for n := 0; n < ATTEMPT_DELAY; n++ {
+<-time.After(1 * time.Second)
+
+fmt.Printf("\r- wait %v seconds to retry", ATTEMPT_DELAY-n)
 }
 }
+
+if !success {
+panic(err)
+}
+
 if !isIntegrationTest {
 go setupWorkers()