exp/support/pipeline: Generic Pipeline #1414

bartekn · 2019-06-14T17:40:35Z

PR Checklist

PR Structure

This PR has reasonably narrow scope (if not, break it down into smaller PRs).
This PR avoids mixing refactoring changes with feature changes (split into two PRs
otherwise).
This PR's title starts with name of package that is most changed in the PR, ex.
services/friendbot

Thoroughness

This PR adds tests for the most critical parts of the new functionality or fixes.
I've updated any docs (developer docs, .md
files, etc... affected by this change). Take a look in the docs folder for a given service,
like this one.

Release planning

I've updated the relevant CHANGELOG (here for Horizon) if
needed with deprecations, added features, breaking changes, and DB schema changes.
I've decided if this PR requires a new major/minor version according to
semver, or if it's mainly a patch change. The PR is targeted at the next
release branch if it's not a patch change.

Summary

Creates a new exp/support/pipeline package that allows creating pipelines processing arbitrary objects to support two pipeline types in the ingestion system.

Close #1388.

Goal and scope

Ingestion system will operate on two pipelines in most cases: state updates pipeline (processing history archives data) and transactions pipeline (processing ledger transactions provided by ledger backend). To support this, the existing pipeline code had to be changed to be more generic to support different reader and writer types and to prevent code duplication.

Summary of changes

This commit extracts and updates the existing pipeline code to a separate exp/support/pipeline package and changes it to be generic. This not only allows us to create two pipeline types in exp/ingest but also enables developers to use it in their apps (or in other SDF apps). Additionally, a number of wrappers have been added exp/ingest/pipline to allow easier development (without type checking from interface{}).

Known limitations & issues

Docs of exp/support/pipeline should be improved with examples how to use it and how it works.

What shouldn't be reviewed

All should be reviewed.

ire-and-curses

Looks good, minor comments. A number of public methods are missing docstrings, worth a quick pass over for those.

ire-and-curses · 2019-06-18T21:49:52Z

exp/ingest/pipeline/main.go

+
+var _ supportPipeline.ReadCloser = &ledgerReadCloserWrapper{}
+
+// readCloserWrapperState wraps pipelinne.ReadCloser to implement StateReadCloser interface.


typo pipelinne

ire-and-curses · 2019-06-18T21:50:01Z

exp/ingest/pipeline/main.go

+
+var _ io.StateReadCloser = &readCloserWrapperState{}
+
+// readCloserWrapperLedger wraps pipelinne.ReadCloser to implement LedgerReadCloser interface.


typo pipelinne

ire-and-curses · 2019-06-18T21:50:13Z

exp/ingest/pipeline/main.go

+
+var _ io.LedgerReadCloser = &readCloserWrapperLedger{}
+
+// writeCloserWrapperState wraps pipelinne.WriteCloser to implement StateWriteCloser interface.


typo pipelinne

ire-and-curses · 2019-06-18T21:50:24Z

exp/ingest/pipeline/main.go

+
+var _ io.StateWriteCloser = &writeCloserWrapperState{}
+
+// writeCloserWrapperLedger wraps pipelinne.WriteCloser to implement LedgerWriteCloser interface.


typo pipelinne

ire-and-curses · 2019-06-18T22:22:02Z

exp/support/pipeline/buffered_read_write_closer.go

+	"io"
+)
+
+const bufferSize = 50000


worth a comment

ire-and-curses · 2019-06-18T22:23:28Z

exp/support/pipeline/buffered_read_write_closer.go

+		b.readEntries++
+		b.readEntriesMutex.Unlock()
+		return entry, nil
+	} else {


could drop the else here

ire-and-curses · 2019-06-18T22:28:43Z

exp/ingest/pipeline/main.go

+	// `ctx.Done()` channel and exit when it returns a value. This can happen when
+	// pipeline execution is interrupted, ex. due to an error.
+	//
+	// Given all information above `ProcessLedger` should always look like this:


This large standard pattern makes me wonder whether we could provide an interface for a custom processing object which would plug in in the right places in this pattern. Then we just execute the generic pattern with the custom user-supplied code.

Good call but I'm not sure how we could change the code to support it. This comment (and all the *Wrapper structs) are for ease of use for developers as Golang does not have a support for generics. We could use generic pipeline processor but I really want to prevent having developers to work with interface{} type.

Let me think about it and create another PR when we have a solution.

tamirms · 2019-06-19T18:59:36Z

in tools/accounts-for-signers/main.go should the "github.com/stellar/go/exp/ingest/adapters" import be aliased to ingestadapters?

bartekn · 2019-06-24T10:33:43Z

in tools/accounts-for-signers/main.go should the "github.com/stellar/go/exp/ingest/adapters" import be aliased to ingestadapters?

We can but it's ingest package demo so I don't think it's necessary. Also, possible that we'll remove adapters package (#1405).

This commit introduces a new `exp/support/pipeline` package that allows creating pipelines processing arbitrary objects to support two pipeline types in the ingestion system. Ingestion system will operate on two pipelines in most cases: state updates pipeline (processing history archives data) and transactions pipeline (processing ledger transactions provided by ledger backend). To support this, the existing pipeline code had to be changed to be more generic to support different reader and writer types and to prevent code duplication. We extract and update the existing pipeline code to a separate `exp/support/pipeline` package and change it to be generic. This not only allows us to create two pipeline types in `exp/ingest` but also enables developers to use it in their apps (or in other SDF apps). Additionally, a number of wrappers have been added to `exp/ingest/pipeline` to allow easier development (without type checking from `interface{}`).

bartekn added the ingest New ingestion system label Jun 14, 2019

bartekn requested review from nikhilsaraf, tomquisel and ire-and-curses June 14, 2019 17:40

bartekn force-pushed the generic-pipeline branch from b5ae8a6 to 97c1024 Compare June 17, 2019 14:03

bartekn added 4 commits June 18, 2019 15:11

Generic Pipeline

c708396

Fixes

c359349

Type checks

16ba9fc

Ledger pipeline

3ddaa56

bartekn force-pushed the generic-pipeline branch from 97c1024 to 3ddaa56 Compare June 18, 2019 13:47

bartekn requested a review from tamirms June 18, 2019 13:49

ire-and-curses approved these changes Jun 18, 2019

View reviewed changes

review

54ec21a

bartekn merged commit ff38810 into master Jun 24, 2019

bartekn deleted the generic-pipeline branch June 24, 2019 10:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

exp/support/pipeline: Generic Pipeline #1414

exp/support/pipeline: Generic Pipeline #1414

bartekn commented Jun 14, 2019

ire-and-curses left a comment

ire-and-curses Jun 18, 2019

ire-and-curses Jun 18, 2019

ire-and-curses Jun 18, 2019

ire-and-curses Jun 18, 2019

ire-and-curses Jun 18, 2019

ire-and-curses Jun 18, 2019

ire-and-curses Jun 18, 2019

bartekn Jun 24, 2019

tamirms commented Jun 19, 2019

bartekn commented Jun 24, 2019


		var _ supportPipeline.ReadCloser = &ledgerReadCloserWrapper{}

		// readCloserWrapperState wraps pipelinne.ReadCloser to implement StateReadCloser interface.


		var _ io.StateReadCloser = &readCloserWrapperState{}

		// readCloserWrapperLedger wraps pipelinne.ReadCloser to implement LedgerReadCloser interface.


		var _ io.LedgerReadCloser = &readCloserWrapperLedger{}

		// writeCloserWrapperState wraps pipelinne.WriteCloser to implement StateWriteCloser interface.


		var _ io.StateWriteCloser = &writeCloserWrapperState{}

		// writeCloserWrapperLedger wraps pipelinne.WriteCloser to implement LedgerWriteCloser interface.

exp/support/pipeline: Generic Pipeline #1414

exp/support/pipeline: Generic Pipeline #1414

Conversation

bartekn commented Jun 14, 2019

PR Structure

Thoroughness

Release planning

Summary

Goal and scope

Summary of changes

Known limitations & issues

What shouldn't be reviewed

ire-and-curses left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tamirms commented Jun 19, 2019

bartekn commented Jun 24, 2019