Etcd Database Support

.github/workflows/ci.yml

@@ -75,6 +75,7 @@ jobs:
           AZURE_STORAGE_ACCOUNT: "devstoreaccount1"
           AZURE_STORAGE_ACCESS_KEY: "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
         run: |
+          sudo mkdir -p /srv ; sudo chown runner /srv
           COVERAGE_DIR=${{ runner.temp }} make
 
       - name: Merge code coverage

AUTHORS

@@ -61,3 +61,5 @@ List of contributors, in chronological order:
 * iofq (https://github.com/iofq)
 * Noa Resare (https://github.com/nresare)
 * Ramón N.Rodriguez (https://github.com/runitonmetal)
+* Golf Hu (https://github.com/hudeng-go)
+* Cookie Fei (https://github.com/wuhuang26)

Makefile

@@ -19,9 +19,6 @@ all: modules test bench check system-test
 help:  ## Print this help
 	@grep -E '^[a-zA-Z][a-zA-Z0-9_-]*:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
 
-prepare:
-	curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin $(GOLANGCI_LINT_VERSION)
-
 modules:
 	go mod download
 	go mod verify
@@ -53,9 +50,14 @@ endif
 system-test: install system/env
 ifeq ($(RUN_LONG_TESTS), yes)
 	go generate
+	test -d /srv/etcd || system/t13_etcd/install-etcd.sh
+	system/t13_etcd/start-etcd.sh &
 	go test -v -coverpkg="./..." -c -tags testruncli
+	kill `cat /tmp/etcd.pid`
+
 	if [ ! -e ~/aptly-fixture-db ]; then git clone https://github.com/aptly-dev/aptly-fixture-db.git ~/aptly-fixture-db/; fi
 	if [ ! -e ~/aptly-fixture-pool ]; then git clone https://github.com/aptly-dev/aptly-fixture-pool.git ~/aptly-fixture-pool/; fi
+	cd /home/runner; curl -O http://repo.aptly.info/system-tests/etcd.db
 	PATH=$(BINPATH)/:$(PATH) && . system/env/bin/activate && APTLY_VERSION=$(VERSION) FORCE_COLOR=1 $(PYTHON) system/run.py --long $(TESTS) --coverage-dir $(COVERAGE_DIR) $(CAPTURE)
 endif
 
@@ -69,7 +71,11 @@ docker-test: install
 	$(PYTHON) system/run.py --long $(TESTS) --coverage-dir $(COVERAGE_DIR) $(CAPTURE) $(TEST)
 
 test:
+	test -d /srv/etcd || system/t13_etcd/install-etcd.sh
+	system/t13_etcd/start-etcd.sh &
+	@echo Running go test
 	go test -v ./... -gocheck.v=true -coverprofile=unit.out
+	kill `cat /tmp/etcd.pid`
 
 bench:
 	go test -v ./deb -run=nothing -bench=. -benchmem
@@ -104,7 +110,7 @@ docker-build-system-tests:  ## Build system-test docker image
 	docker build -f system/Dockerfile . -t aptly-system-test
 
 docker-unit-tests:  ## Run unit tests in docker container
-	docker run -it --rm -v ${PWD}:/app aptly-system-test go test -v ./... -gocheck.v=true
+	docker run -it --rm -v ${PWD}:/app aptly-system-test /app/system/run-unit-tests
 
 docker-system-tests:  ## Run system tests in docker container (add TEST=t04_mirror to run only specific tests)
 	docker run -it --rm -v ${PWD}:/app aptly-system-test /app/system/run-system-tests $(TEST)

context/context.go

@@ -3,6 +3,7 @@
 
 import (
 	gocontext "context"
+	"errors"
 	"fmt"
 	"math/rand"
 	"os"
@@ -18,6 +19,7 @@
 	"github.com/aptly-dev/aptly/azure"
 	"github.com/aptly-dev/aptly/console"
 	"github.com/aptly-dev/aptly/database"
+	"github.com/aptly-dev/aptly/database/etcddb"
 	"github.com/aptly-dev/aptly/database/goleveldb"
 	"github.com/aptly-dev/aptly/deb"
 	"github.com/aptly-dev/aptly/files"
@@ -287,8 +289,18 @@
 func (context *AptlyContext) _database() (database.Storage, error) {
 	if context.database == nil {
 		var err error
-
-		context.database, err = goleveldb.NewDB(context.dbPath())
+		switch context.config().DatabaseBackend.Type {
+		case "leveldb":
+			if len(context.config().DatabaseBackend.DbPath) == 0 {
+				return nil, errors.New("leveldb databaseBackend config invalid")
+			}
+			dbPath := filepath.Join(context.config().RootDir, context.config().DatabaseBackend.DbPath)
+			context.database, err = goleveldb.NewDB(dbPath)
+		case "etcd":
+			context.database, err = etcddb.NewDB(context.config().DatabaseBackend.URL)
+		default:
+			context.database, err = goleveldb.NewDB(context.dbPath())
+		}
 		if err != nil {
 			return nil, fmt.Errorf("can't instantiate database: %s", err)
 		}

database/etcddb/batch.go

@@ -0,0 +1,53 @@
+package etcddb
+
+import (
+	"github.com/aptly-dev/aptly/database"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+type EtcDBatch struct {
+	s   *EtcDStorage
+	ops []clientv3.Op
+}
+
+type WriteOptions struct {
+	NoWriteMerge bool
+	Sync         bool
+}
+
+func (b *EtcDBatch) Put(key []byte, value []byte) (err error) {
+	b.ops = append(b.ops, clientv3.OpPut(string(key), string(value)))
+	return
+}
+
+func (b *EtcDBatch) Delete(key []byte) (err error) {
+	b.ops = append(b.ops, clientv3.OpDelete(string(key)))
+	return
+}
+
+func (b *EtcDBatch) Write() (err error) {
+	kv := clientv3.NewKV(b.s.db)
+
+	batchSize := 128
+	for i := 0; i < len(b.ops); i += batchSize {
+		txn := kv.Txn(Ctx)
+		end := i + batchSize
+		if end > len(b.ops) {
+			end = len(b.ops)
+		}
+
+		batch := b.ops[i:end]
+		txn.Then(batch...)
+		_, err = txn.Commit()
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	return
+}
+
+// batch should implement database.Batch
+var (
+	_ database.Batch = &EtcDBatch{}
+)

database/etcddb/database.go

@@ -0,0 +1,32 @@
+package etcddb
+
+import (
+	"context"
+	"time"
+
+	"github.com/aptly-dev/aptly/database"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+var Ctx = context.TODO()
+
+func internalOpen(url string) (cli *clientv3.Client, err error) {
+	cfg := clientv3.Config{
+		Endpoints:            []string{url},
+		DialTimeout:          30 * time.Second,
+		MaxCallSendMsgSize:   2147483647, // (2048 * 1024 * 1024) - 1
+		MaxCallRecvMsgSize:   2147483647,
+		DialKeepAliveTimeout: 7200 * time.Second,
+	}
+
+	cli, err = clientv3.New(cfg)
+	return
+}
+
+func NewDB(url string) (database.Storage, error) {
+	cli, err := internalOpen(url)
+	if err != nil {
+		return nil, err
+	}
+	return &EtcDStorage{url, cli, ""}, nil
+}

database/etcddb/database_test.go

@@ -0,0 +1,158 @@
+package etcddb_test
+
+import (
+	"testing"
+
+	"github.com/aptly-dev/aptly/database"
+	"github.com/aptly-dev/aptly/database/etcddb"
+	. "gopkg.in/check.v1"
+)
+
+// Launch gocheck tests
+func Test(t *testing.T) {
+	TestingT(t)
+}
+
+type EtcDDBSuite struct {
+	url string
+	db  database.Storage
+}
+
+var _ = Suite(&EtcDDBSuite{})
+
+func (s *EtcDDBSuite) SetUpTest(c *C) {
+	var err error
+	s.db, err = etcddb.NewDB("127.0.0.1:2379")
+	c.Assert(err, IsNil)
+}
+
+func (s *EtcDDBSuite) TestSetUpTest(c *C) {
+	var err error
+	s.db, err = etcddb.NewDB("127.0.0.1:2379")
+	c.Assert(err, IsNil)
+}
+
+func (s *EtcDDBSuite) TestGetPut(c *C) {
+	var (
+		key   = []byte("key")
+		value = []byte("value")
+	)
+	var err error
+
+	err = s.db.Put(key, value)
+	c.Assert(err, IsNil)
+
+	result, err := s.db.Get(key)
+	c.Assert(err, IsNil)
+	c.Assert(result, DeepEquals, value)
+}
+
+func (s *EtcDDBSuite) TestDelete(c *C) {
+	var (
+		key   = []byte("key")
+		value = []byte("value")
+	)
+
+	err := s.db.Put(key, value)
+	c.Assert(err, IsNil)
+
+	_, err = s.db.Get(key)
+	c.Assert(err, IsNil)
+
+	err = s.db.Delete(key)
+	c.Assert(err, IsNil)
+
+}
+
+func (s *EtcDDBSuite) TestByPrefix(c *C) {
+	//c.Check(s.db.FetchByPrefix([]byte{0x80}), DeepEquals, [][]byte{})
+
+	s.db.Put([]byte{0x80, 0x01}, []byte{0x01})
+	s.db.Put([]byte{0x80, 0x03}, []byte{0x03})
+	s.db.Put([]byte{0x80, 0x02}, []byte{0x02})
+	c.Check(s.db.FetchByPrefix([]byte{0x80}), DeepEquals, [][]byte{{0x01}, {0x02}, {0x03}})
+	c.Check(s.db.KeysByPrefix([]byte{0x80}), DeepEquals, [][]byte{{0x80, 0x01}, {0x80, 0x02}, {0x80, 0x03}})
+
+	s.db.Put([]byte{0x90, 0x01}, []byte{0x04})
+	c.Check(s.db.FetchByPrefix([]byte{0x80}), DeepEquals, [][]byte{{0x01}, {0x02}, {0x03}})
+	c.Check(s.db.KeysByPrefix([]byte{0x80}), DeepEquals, [][]byte{{0x80, 0x01}, {0x80, 0x02}, {0x80, 0x03}})
+
+	s.db.Put([]byte{0x00, 0x01}, []byte{0x05})
+	c.Check(s.db.FetchByPrefix([]byte{0x80}), DeepEquals, [][]byte{{0x01}, {0x02}, {0x03}})
+	c.Check(s.db.KeysByPrefix([]byte{0x80}), DeepEquals, [][]byte{{0x80, 0x01}, {0x80, 0x02}, {0x80, 0x03}})
+
+	keys := [][]byte{}
+	values := [][]byte{}
+
+	c.Check(s.db.ProcessByPrefix([]byte{0x80}, func(k, v []byte) error {
+		keys = append(keys, append([]byte(nil), k...))
+		values = append(values, append([]byte(nil), v...))
+		return nil
+	}), IsNil)
+
+	c.Check(values, DeepEquals, [][]byte{{0x01}, {0x02}, {0x03}})
+	c.Check(keys, DeepEquals, [][]byte{{0x80, 0x01}, {0x80, 0x02}, {0x80, 0x03}})
+
+	c.Check(s.db.ProcessByPrefix([]byte{0x80}, func(k, v []byte) error {
+		return database.ErrNotFound
+	}), Equals, database.ErrNotFound)
+
+	c.Check(s.db.ProcessByPrefix([]byte{0xa0}, func(k, v []byte) error {
+		return database.ErrNotFound
+	}), IsNil)
+
+	c.Check(s.db.FetchByPrefix([]byte{0xa0}), DeepEquals, [][]byte{})
+	c.Check(s.db.KeysByPrefix([]byte{0xa0}), DeepEquals, [][]byte{})
+}
+
+func (s *EtcDDBSuite) TestHasPrefix(c *C) {
+	//c.Check(s.db.HasPrefix([]byte(nil)), Equals, false)
+	//c.Check(s.db.HasPrefix([]byte{0x80}), Equals, false)
+
+	s.db.Put([]byte{0x80, 0x01}, []byte{0x01})
+
+	c.Check(s.db.HasPrefix([]byte(nil)), Equals, true)
+	c.Check(s.db.HasPrefix([]byte{0x80}), Equals, true)
+	c.Check(s.db.HasPrefix([]byte{0x79}), Equals, false)
+}
+
+func (s *EtcDDBSuite) TestTransactionCommit(c *C) {
+	var (
+		key    = []byte("key")
+		key2   = []byte("key2")
+		value  = []byte("value")
+		value2 = []byte("value2")
+	)
+	transaction, err := s.db.OpenTransaction()
+
+	err = s.db.Put(key, value)
+	c.Assert(err, IsNil)
+
+	c.Assert(err, IsNil)
+	transaction.Put(key2, value2)
+	v, err := s.db.Get(key)
+	c.Check(v, DeepEquals, value)
+        err = transaction.Delete(key)
+	c.Assert(err, IsNil)
+
+	_, err = transaction.Get(key2)
+	c.Assert(err, IsNil)
+
+	v2, err := transaction.Get(key2)
+	c.Check(err, IsNil)
+	c.Check(v2, DeepEquals, value2)
+
+	_, err = transaction.Get(key)
+	c.Assert(err, IsNil)
+
+	err = transaction.Commit()
+	c.Check(err, IsNil)
+
+	v2, err = transaction.Get(key2)
+	c.Check(err, IsNil)
+	c.Check(v2, DeepEquals, value2)
+
+	_, err = transaction.Get(key)
+	c.Assert(err, NotNil)
+}
+