remove pkg/transformers; extract shared files needed to

libraries/shared; work on automated db migration management
2019-02-24 21:38:47 -06:00
parent 55fa9b8364
commit decc2a3caf
545 changed files with 431 additions and 15922 deletions
@@ -22,26 +22,28 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	log "github.com/sirupsen/logrus"

+	chunk "github.com/vulcanize/vulcanizedb/libraries/shared/chunker"
+	"github.com/vulcanize/vulcanizedb/libraries/shared/constants"
+	fetch "github.com/vulcanize/vulcanizedb/libraries/shared/fetcher"
+	repo "github.com/vulcanize/vulcanizedb/libraries/shared/repository"
 	"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/constants"
 )

 type EventWatcher struct {
-	Transformers  []transformer.Transformer
+	Transformers  []transformer.EventTransformer
 	DB            *postgres.DB
-	Fetcher       shared.LogFetcher
-	Chunker       shared.Chunker
+	Fetcher       fetch.LogFetcher
+	Chunker       chunk.Chunker
 	Addresses     []common.Address
 	Topics        []common.Hash
 	StartingBlock *int64
 }

 func NewEventWatcher(db *postgres.DB, bc core.BlockChain) EventWatcher {
-	chunker := shared.NewLogChunker()
-	fetcher := shared.NewFetcher(bc)
+	chunker := chunk.NewLogChunker()
+	fetcher := fetch.NewFetcher(bc)
 	return EventWatcher{
 		DB:      db,
 		Fetcher: fetcher,
@@ -83,13 +85,13 @@ func (watcher *EventWatcher) Execute(recheckHeaders constants.TransformerExecuti
 		return fmt.Errorf("No transformers added to watcher")
 	}

-	checkedColumnNames, err := shared.GetCheckedColumnNames(watcher.DB)
+	checkedColumnNames, err := repo.GetCheckedColumnNames(watcher.DB)
 	if err != nil {
 		return err
 	}
-	notCheckedSQL := shared.CreateNotCheckedSQL(checkedColumnNames, recheckHeaders)
+	notCheckedSQL := repo.CreateNotCheckedSQL(checkedColumnNames, recheckHeaders)

-	missingHeaders, err := shared.MissingHeaders(*watcher.StartingBlock, -1, watcher.DB, notCheckedSQL)
+	missingHeaders, err := repo.MissingHeaders(*watcher.StartingBlock, -1, watcher.DB, notCheckedSQL)
 	if err != nil {
 		log.Error("Fetching of missing headers failed in watcher!")
 		return err
@@ -25,14 +25,14 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"

+	"github.com/vulcanize/vulcanizedb/libraries/shared/constants"
+	"github.com/vulcanize/vulcanizedb/libraries/shared/mocks"
 	"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
 	"github.com/vulcanize/vulcanizedb/libraries/shared/watcher"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
 	"github.com/vulcanize/vulcanizedb/pkg/fakes"
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/constants"
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks"
 	"github.com/vulcanize/vulcanizedb/test_config"
 )

@@ -0,0 +1,90 @@
+// VulcanizeDB
+// Copyright © 2018 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package watcher
+
+import (
+	"strings"
+	"reflect"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
+	"github.com/vulcanize/vulcanizedb/libraries/shared/utils"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/fs"
+)
+
+type StorageWatcher struct {
+	db           *postgres.DB
+	tailer       fs.Tailer
+	Queue        IStorageQueue
+	Transformers map[common.Address]transformer.StorageTransformer
+}
+
+func NewStorageWatcher(tailer fs.Tailer, db *postgres.DB) StorageWatcher {
+	transformers := make(map[common.Address]transformer.StorageTransformer)
+	queue := NewStorageQueue(db)
+	return StorageWatcher{
+		db:           db,
+		tailer:       tailer,
+		Queue:        queue,
+		Transformers: transformers,
+	}
+}
+
+func (watcher StorageWatcher) AddTransformers(initializers []transformer.StorageTransformerInitializer) {
+	for _, initializer := range initializers {
+		transformer := initializer(watcher.db)
+		watcher.Transformers[transformer.ContractAddress()] = transformer
+	}
+}
+
+func (watcher StorageWatcher) Execute() error {
+	t, tailErr := watcher.tailer.Tail()
+	if tailErr != nil {
+		return tailErr
+	}
+	for line := range t.Lines {
+		row, parseErr := utils.FromStrings(strings.Split(line.Text, ","))
+		if parseErr != nil {
+			return parseErr
+		}
+		transformer, ok := watcher.Transformers[row.Contract]
+		if !ok {
+			logrus.Warn(utils.ErrContractNotFound{Contract: row.Contract.Hex()}.Error())
+			continue
+		}
+		executeErr := transformer.Execute(row)
+		if executeErr != nil {
+			if isKeyNotFound(executeErr) {
+				queueErr := watcher.Queue.Add(row)
+				if queueErr != nil {
+					logrus.Warn(queueErr.Error())
+				}
+			} else {
+				logrus.Warn(executeErr.Error())
+			}
+			continue
+		}
+	}
+	return nil
+}
+
+func isKeyNotFound(executeErr error) bool {
+	return reflect.TypeOf(executeErr) == reflect.TypeOf(utils.ErrStorageKeyNotFound{})
+}
@@ -0,0 +1,212 @@
+// VulcanizeDB
+// Copyright © 2018 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package watcher_test
+
+import (
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/hpcloud/tail"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/mocks"
+	"github.com/vulcanize/vulcanizedb/libraries/shared/transformer"
+	"github.com/vulcanize/vulcanizedb/libraries/shared/utils"
+	"github.com/vulcanize/vulcanizedb/libraries/shared/watcher"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/fakes"
+	"github.com/vulcanize/vulcanizedb/test_config"
+)
+
+var _ = Describe("Storage Watcher", func() {
+	It("adds transformers", func() {
+		fakeAddress := common.HexToAddress("0x12345")
+		fakeTransformer := &mocks.MockStorageTransformer{Address: fakeAddress}
+		w := watcher.NewStorageWatcher(&fakes.MockTailer{}, test_config.NewTestDB(core.Node{}))
+
+		w.AddTransformers([]transformer.StorageTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
+
+		Expect(w.Transformers[fakeAddress]).To(Equal(fakeTransformer))
+	})
+
+	It("reads the tail of the storage diffs file", func() {
+		mockTailer := fakes.NewMockTailer()
+		w := watcher.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
+
+		assert(func(err error) {
+			Expect(err).To(BeNil())
+			Expect(mockTailer.TailCalled).To(BeTrue())
+		}, w, mockTailer, []*tail.Line{})
+	})
+
+	It("returns error if row parsing fails", func() {
+		mockTailer := fakes.NewMockTailer()
+		w := watcher.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
+		line := &tail.Line{Text: "oops"}
+
+		assert(func(err error) {
+			Expect(err).To(HaveOccurred())
+			Expect(err).To(MatchError(utils.ErrRowMalformed{Length: 1}))
+		}, w, mockTailer, []*tail.Line{line})
+	})
+
+	It("logs error if no transformer can parse storage row", func() {
+		mockTailer := fakes.NewMockTailer()
+		address := common.HexToAddress("0x12345")
+		line := getFakeLine(address.Bytes())
+		w := watcher.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
+		tempFile, err := ioutil.TempFile("", "log")
+		defer os.Remove(tempFile.Name())
+		Expect(err).NotTo(HaveOccurred())
+		logrus.SetOutput(tempFile)
+
+		assert(func(err error) {
+			Expect(err).NotTo(HaveOccurred())
+			logContent, readErr := ioutil.ReadFile(tempFile.Name())
+			Expect(readErr).NotTo(HaveOccurred())
+			Expect(string(logContent)).To(ContainSubstring(utils.ErrContractNotFound{Contract: address.Hex()}.Error()))
+		}, w, mockTailer, []*tail.Line{line})
+	})
+
+	It("executes transformer with storage row", func() {
+		address := []byte{1, 2, 3}
+		line := getFakeLine(address)
+		mockTailer := fakes.NewMockTailer()
+		w := watcher.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
+		fakeTransformer := &mocks.MockStorageTransformer{Address: common.BytesToAddress(address)}
+		w.AddTransformers([]transformer.StorageTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
+
+		assert(func(err error) {
+			Expect(err).To(BeNil())
+			expectedRow, err := utils.FromStrings(strings.Split(line.Text, ","))
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeTransformer.PassedRow).To(Equal(expectedRow))
+		}, w, mockTailer, []*tail.Line{line})
+	})
+
+	Describe("when executing transformer fails", func() {
+		It("queues row when error is storage key not found", func() {
+			address := []byte{1, 2, 3}
+			line := getFakeLine(address)
+			mockTailer := fakes.NewMockTailer()
+			w := watcher.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
+			mockQueue := &mocks.MockStorageQueue{}
+			w.Queue = mockQueue
+			keyNotFoundError := utils.ErrStorageKeyNotFound{Key: "unknown_storage_key"}
+			fakeTransformer := &mocks.MockStorageTransformer{Address: common.BytesToAddress(address), ExecuteErr: keyNotFoundError}
+			w.AddTransformers([]transformer.StorageTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
+
+			assert(func(err error) {
+				Expect(err).NotTo(HaveOccurred())
+				Expect(mockQueue.AddCalled).To(BeTrue())
+			}, w, mockTailer, []*tail.Line{line})
+		})
+
+		It("logs error if queuing row fails", func() {
+			address := []byte{1, 2, 3}
+			line := getFakeLine(address)
+			mockTailer := fakes.NewMockTailer()
+			w := watcher.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
+			mockQueue := &mocks.MockStorageQueue{}
+			mockQueue.AddError = fakes.FakeError
+			w.Queue = mockQueue
+			keyNotFoundError := utils.ErrStorageKeyNotFound{Key: "unknown_storage_key"}
+			fakeTransformer := &mocks.MockStorageTransformer{Address: common.BytesToAddress(address), ExecuteErr: keyNotFoundError}
+			w.AddTransformers([]transformer.StorageTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
+			tempFile, err := ioutil.TempFile("", "log")
+			defer os.Remove(tempFile.Name())
+			Expect(err).NotTo(HaveOccurred())
+			logrus.SetOutput(tempFile)
+
+			assert(func(err error) {
+				Expect(err).NotTo(HaveOccurred())
+				Expect(mockQueue.AddCalled).To(BeTrue())
+				logContent, readErr := ioutil.ReadFile(tempFile.Name())
+				Expect(readErr).NotTo(HaveOccurred())
+				Expect(string(logContent)).To(ContainSubstring(fakes.FakeError.Error()))
+			}, w, mockTailer, []*tail.Line{line})
+		})
+
+		It("logs any other error", func() {
+			address := []byte{1, 2, 3}
+			line := getFakeLine(address)
+			mockTailer := fakes.NewMockTailer()
+			w := watcher.NewStorageWatcher(mockTailer, test_config.NewTestDB(core.Node{}))
+			executionError := errors.New("storage watcher failed attempting to execute transformer")
+			fakeTransformer := &mocks.MockStorageTransformer{Address: common.BytesToAddress(address), ExecuteErr: executionError}
+			w.AddTransformers([]transformer.StorageTransformerInitializer{fakeTransformer.FakeTransformerInitializer})
+			tempFile, err := ioutil.TempFile("", "log")
+			defer os.Remove(tempFile.Name())
+			Expect(err).NotTo(HaveOccurred())
+			logrus.SetOutput(tempFile)
+
+			assert(func(err error) {
+				Expect(err).NotTo(HaveOccurred())
+				logContent, readErr := ioutil.ReadFile(tempFile.Name())
+				Expect(readErr).NotTo(HaveOccurred())
+				Expect(string(logContent)).To(ContainSubstring(executionError.Error()))
+			}, w, mockTailer, []*tail.Line{line})
+		})
+	})
+})
+
+func assert(assertion func(err error), watcher watcher.StorageWatcher, mockTailer *fakes.MockTailer, lines []*tail.Line) {
+	errs := make(chan error, 1)
+	done := make(chan bool, 1)
+	go execute(watcher, errs, done)
+	for _, line := range lines {
+		mockTailer.Lines <- line
+	}
+	close(mockTailer.Lines)
+
+	select {
+	case err := <-errs:
+		assertion(err)
+		break
+	case <-done:
+		assertion(nil)
+		break
+	}
+}
+
+func execute(w watcher.StorageWatcher, errs chan error, done chan bool) {
+	err := w.Execute()
+	if err != nil {
+		errs <- err
+	} else {
+		done <- true
+	}
+}
+
+func getFakeLine(address []byte) *tail.Line {
+	blockHash := []byte{4, 5, 6}
+	blockHeight := int64(789)
+	storageKey := []byte{9, 8, 7}
+	storageValue := []byte{6, 5, 4}
+	return &tail.Line{
+		Text: fmt.Sprintf("%s,%s,%d,%s,%s", common.Bytes2Hex(address), common.Bytes2Hex(blockHash), blockHeight, common.Bytes2Hex(storageKey), common.Bytes2Hex(storageValue)),
+		Time: time.Time{},
+		Err:  nil,
+	}
+}