Move compaction to separate goroutine (#122)

flipkart-incubator · Mar 16, 2022 · b7d8acb · b7d8acb
1 parent 31dc2ff
commit b7d8acb
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 7 deletions.
diff --git a/go.mod b/go.mod
@@ -11,7 +11,7 @@ require (
 	github.com/dgraph-io/ristretto v0.0.3 // indirect
 	github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect
 	github.com/flipkart-incubator/gorocksdb v0.0.0-20210920082714-1f7dcbb7b2e4
-	github.com/flipkart-incubator/nexus v0.0.0-20220303113731-529d5203982c
+	github.com/flipkart-incubator/nexus v0.0.0-20220316072727-c44c4b25144a
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2
 	github.com/golang/snappy v0.0.2 // indirect

diff --git a/go.sum b/go.sum
@@ -96,8 +96,8 @@ github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/flipkart-incubator/gorocksdb v0.0.0-20210920082714-1f7dcbb7b2e4 h1:9zPLm5QKcBp5xUOBMWhRZPU+iwfl6xJtbSmbazmgy2g=
 github.com/flipkart-incubator/gorocksdb v0.0.0-20210920082714-1f7dcbb7b2e4/go.mod h1:kvJSXc90Ifw0rxuTxEHKq6UH/7hQ/gd9RKCyD94ctJ0=
-github.com/flipkart-incubator/nexus v0.0.0-20220303113731-529d5203982c h1:jTj5fpxO7d5O4k5Kq0svH4DFUQP1nDLSzRTJt7Aiuek=
-github.com/flipkart-incubator/nexus v0.0.0-20220303113731-529d5203982c/go.mod h1:B85NLNwdayqmodWfNcJXZEBelZ7LXHee6jyRDdrfoyY=
+github.com/flipkart-incubator/nexus v0.0.0-20220316072727-c44c4b25144a h1:PrHQUQEqNV0tP8j9yvJWJkOwVLTwcsxqvt33dpWxt4Q=
+github.com/flipkart-incubator/nexus v0.0.0-20220316072727-c44c4b25144a/go.mod h1:eEnRO5Yjl4OgxHbLjtmGdx/iQSgUqtLEOEGEGhka8Wo=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWpgI=
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
@@ -146,6 +146,7 @@ github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.2 h1:aeE13tS0IiQgFjYdoL8qN3K1N2bXXtI6Vi51/y7BpMw=
 github.com/golang/snappy v0.0.2/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/sync v0.0.0-20190911185100-cd5d95a43a6e h1:Wi0y4ZAnk+WvzvZ26frs641N6DpG0e5Se3whDjLs9zU=
 github.com/golang/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 github.com/golang/sys v0.0.0-20200317113312-5766fd39f98d h1:8zzUweJJxbKvLGxvjbyJsIimIUysmU3Xa27wwAG04s0=
 github.com/golang/sys v0.0.0-20200317113312-5766fd39f98d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

diff --git a/internal/master/service.go b/internal/master/service.go
@@ -554,12 +554,10 @@ func (ds *distributedService) GetStatus(context context.Context, request *emptyp
 			regionInfo.Status = serverpb.RegionStatus_PRIMARY_FOLLOWER
 		}
 	}
-	ds.opts.Logger.Debug("Current Info", zap.String("Status", regionInfo.Status.String()))
 	return regionInfo, nil
 }
 
 func (ds *distributedService) Check(ctx context.Context, healthCheckReq *health.HealthCheckRequest) (*health.HealthCheckResponse, error) {
-	regionInfo := ds.DKVService.(*standaloneService).regionInfo
 	if ds.isClosed {
 		return &health.HealthCheckResponse{Status: health.HealthCheckResponse_NOT_SERVING}, nil
 	} else {
@@ -572,7 +570,6 @@ func (ds *distributedService) Check(ctx context.Context, healthCheckReq *health.
 			return &health.HealthCheckResponse{Status: health.HealthCheckResponse_NOT_SERVING}, nil
 		}
 	}
-	ds.opts.Logger.Debug("Current Info", zap.String("Status", regionInfo.Status.String()))
 	return &health.HealthCheckResponse{Status: health.HealthCheckResponse_NOT_SERVING}, nil
 }
 

diff --git a/internal/storage/rocksdb/store.go b/internal/storage/rocksdb/store.go
@@ -429,13 +429,15 @@ func (rdb *rocksDB) generateSST(snap *gorocksdb.Snapshot, cf *gorocksdb.ColumnFa
 type checkPointSnapshot struct {
 	tar *utils.StreamingTar
 	dir string
+	lgr *zap.Logger
 }
 
 func (r *checkPointSnapshot) Read(p []byte) (n int, err error) {
 	return r.tar.Read(p)
 }
 
 func (r *checkPointSnapshot) Close() error {
+	r.lgr.Info(fmt.Sprintf("GetSnapshot: Closing snapshot, will delete folder %s", r.dir))
 	r.tar.Close()
 	return os.RemoveAll(r.dir)
 }
@@ -497,7 +499,7 @@ func (rdb *rocksDB) GetSnapshot() (io.ReadCloser, error) {
 		rdb.opts.lgr.Error("GetSnapshot: Failed to archive checkpoint files", zap.Error(err))
 		return nil, err
 	}
-	return &checkPointSnapshot{tar: tarF, dir: sstDir}, nil
+	return &checkPointSnapshot{tar: tarF, dir: sstDir, lgr: rdb.opts.lgr}, nil
 }
 
 func (rdb *rocksDB) PutSnapshot(snap io.ReadCloser) error {