mosuka · vniche · Jun 4, 2020 · Jun 5, 2020 · Jul 10, 2020
diff --git a/.github/workflows/PR.yaml b/.github/workflows/PR.yaml
@@ -0,0 +1,32 @@
+on:
+  pull_request:
+    branches:
+      - master
+
+jobs:
+  build:
+    name: build
+    runs-on: ubuntu-latest
+    container:
+      image: golang:1.14-buster
+      volumes:
+        - "/home/runner/work/$GITHUB_REPOSITORY:/go/src/github.com/$GITHUB_REPOSITORY"
+    steps:
+      - uses: actions/checkout@v1
+
+      - name: restore from cache
+        uses: actions/cache@v1
+        with:
+          path: /go/pkg/mod
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: |
+            ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+
+      - name: download dependencies if not cached
+        run: |
+          if [ ! -d "/go/pkg/mod" ]; then
+            go mod tidy
+          fi
+
+      - name: test
+        run: go test -race -covermode=atomic -coverprofile coverage.out -v $(go list ./...)
diff --git a/CHANGES.md b/CHANGES.md
@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+- adds scan endpoint #35 @vniche
 - Fix bug in getting leader ID #34 @mosuka
 
 ## [v0.3.1] 2020-04-01

diff --git a/protobuf/kvs.pb.go b/protobuf/kvs.pb.go
diff --git a/protobuf/kvs.pb.gw.go b/protobuf/kvs.pb.gw.go
diff --git a/protobuf/kvs.proto b/protobuf/kvs.proto
@@ -55,12 +55,20 @@ service KVS {
             get: "/v1/data/{key=**}"
         };
     }
+
+    rpc Scan (ScanRequest) returns (ScanResponse) {
+        option (google.api.http) = {
+            get: "/v1/data/{prefix=**}"
+        };
+    }
+
     rpc Set (SetRequest) returns (google.protobuf.Empty) {
         option (google.api.http) = {
             put: "/v1/data/{key=**}"
             body: "*"
         };
     }
+
     rpc Delete (DeleteRequest) returns (google.protobuf.Empty) {
         option (google.api.http) = {
             delete: "/v1/data/{key=**}"
@@ -125,6 +133,14 @@ message GetResponse {
     bytes value = 1;
 }
 
+message ScanRequest {
+    string prefix = 1;
+}
+
+message ScanResponse {
+    repeated bytes values = 1;
+}
+
 message SetRequest {
     string key = 1;
     bytes value = 2;

diff --git a/server/grpc_service.go b/server/grpc_service.go
@@ -349,6 +349,23 @@ func (s *GRPCService) Get(ctx context.Context, req *protobuf.GetRequest) (*proto
 	return resp, nil
 }
 
+func (s *GRPCService) Scan(ctx context.Context, req *protobuf.ScanRequest) (*protobuf.ScanResponse, error) {
+	resp := &protobuf.ScanResponse{}
+
+	var err error
+
+	resp, err = s.raftServer.Scan(req)
+	if err != nil {
+		switch err {
+		default:
+			s.logger.Debug("failed to scan data", zap.String("prefix", req.Prefix), zap.String("err", err.Error()))
+			return resp, status.Error(codes.Internal, err.Error())
+		}
+	}
+
+	return resp, nil
+}
+
 func (s *GRPCService) Set(ctx context.Context, req *protobuf.SetRequest) (*empty.Empty, error) {
 	resp := &empty.Empty{}
 

diff --git a/server/raft_fsm.go b/server/raft_fsm.go
@@ -71,6 +71,16 @@ func (f *RaftFSM) Get(key string) ([]byte, error) {
 	return value, nil
 }
 
+func (f *RaftFSM) Scan(prefix string) ([][]byte, error) {
+	values, err := f.kvs.Scan(prefix)
+	if err != nil {
+		f.logger.Error("failed to scan values", zap.String("prefix", prefix), zap.Error(err))
+		return nil, err
+	}
+
+	return values, nil
+}
+
 func (f *RaftFSM) applySet(key string, value []byte) interface{} {
 	err := f.kvs.Set(key, value)
 	if err != nil {

diff --git a/server/raft_server.go b/server/raft_server.go
@@ -606,6 +606,20 @@ func (s *RaftServer) Get(req *protobuf.GetRequest) (*protobuf.GetResponse, error
 	return resp, nil
 }
 
+func (s *RaftServer) Scan(req *protobuf.ScanRequest) (*protobuf.ScanResponse, error) {
+	values, err := s.fsm.Scan(req.Prefix)
+	if err != nil {
+		s.logger.Error("failed to scan", zap.Any("prefix", req.Prefix), zap.Error(err))
+		return nil, err
+	}
+
+	resp := &protobuf.ScanResponse{
+		Values: values,
+	}
+
+	return resp, nil
+}
+
 func (s *RaftServer) Set(req *protobuf.SetRequest) error {
 	kvpAny := &any.Any{}
 	if err := marshaler.UnmarshalAny(req, kvpAny); err != nil {

diff --git a/storage/kvs.go b/storage/kvs.go
@@ -79,6 +79,34 @@ func (k *KVS) Get(key string) ([]byte, error) {
 	return value, nil
 }
 
+func (k *KVS) Scan(prefix string) ([][]byte, error) {
+	start := time.Now()
+
+	var value [][]byte
+	if err := k.db.View(func(txn *badger.Txn) error {
+		it := txn.NewIterator(badger.DefaultIteratorOptions)
+		defer it.Close()
+		prefixBytes := []byte(prefix)
+		for it.Seek(prefixBytes); it.ValidForPrefix(prefixBytes); it.Next() {
+			item := it.Item()
+			err := item.Value(func(val []byte) error {
+				value = append(value, append([]byte{}, val...))
+				return nil
+			})
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		k.logger.Error("failed to scan value", zap.String("prefix", prefix), zap.Error(err))
+		return nil, err
+	}
+
+	k.logger.Debug("scan", zap.String("prefix", prefix), zap.Float64("time", float64(time.Since(start))/float64(time.Second)))
+	return value, nil
+}
+
 func (k *KVS) Set(key string, value []byte) error {
 	start := time.Now()