Enhancement/disk utilization

.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/worker.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/zcn.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()