209 ES adaptor stability #233

jipperinbham · 2017-01-19T18:26:16Z

CHANGELOG.md updated
README.md updated

switch from elastigo to elastic library and create underlying clients for each major version of elasticsearch.

trinchan

Initial look through, I'll give it another look later today

trinchan · 2017-01-23T15:55:07Z

pkg/adaptor/elasticsearch/clients/closer.go

+
+// Closer takes care of receiving on the done channel and then properly cleaning up
+// the session and WaitGroup.
+func Closer(done chan struct{}, wg *sync.WaitGroup, s client.Session) {


Closer sounds like it should be an interface by its name. Maybe should be Close

trinchan · 2017-01-23T15:56:56Z

pkg/adaptor/elasticsearch/clients/registry.go

+type Creator func(chan struct{}, *sync.WaitGroup, *ClientOptions) (client.Writer, error)
+
+// Clients contains the map of versioned clients
+var Clients = map[string]*VersionedClient{}


This can be private if we're going to delegate access through Add below

planning on addressing this in #234

trinchan · 2017-01-23T16:01:09Z

pkg/adaptor/elasticsearch/clients/registry.go

@@ -0,0 +1,36 @@
+package clients


I can see us having/wanting to use this sort of registry for other db types outside of ES. Let's make an issue for making this generic and have the other adaptors use the registry as well. I guess we would need to be able to get the version information through an API call like we do for ES. Should be possible.

issue created to track that work, #234

trinchan · 2017-01-23T16:07:10Z

pkg/adaptor/elasticsearch/clients/v5/writer.go

+		var id string
+		if _, ok := msg.Data()["_id"]; ok {
+			id = msg.ID()
+			msg.Data().Delete("_id")


Will we have to restore the id field after the write for nodes further down in the pipeline that may expect _id to be present? I'm not sure if msg.Data() returns a copy or not. I think not.

msg.Data() does not return a copy and I hadn't really thought about the use case for chaining the adaptors in such a way, it does bring up the issue of whether the msg.Data should be immutable because in instances were multiple Sink adaptors are attached to the same Source, this code could cause a race condition with a concurrent map read/write.

see #233 (comment)

trinchan · 2017-01-23T16:07:36Z

pkg/adaptor/elasticsearch/clients/v5/writer.go

+}
+
+func (w *Writer) Close() {
+	log.Infoln("flushing BulkProcessor")


Would be nice to have this be a little more descriptive (that it's an ES bulk processor)

trinchan · 2017-01-23T16:13:18Z

pkg/adaptor/elasticsearch/clients/v1/writer_test.go

+	log.Debugln("tests shutdown complete")
+}
+
+type CountResponse struct {


Make this private

trinchan · 2017-01-23T16:16:16Z

pkg/adaptor/elasticsearch/clients/v2/writer.go

+		var id string
+		if _, ok := msg.Data()["_id"]; ok {
+			id = msg.ID()
+			msg.Data().Delete("_id")


Same comment as with v1, do we need to restore?

addressed by making a copy of the original msg.Data() in the adaptor.

trinchan · 2017-01-23T16:16:51Z

pkg/adaptor/elasticsearch/clients/v2/writer.go

@@ -0,0 +1,91 @@
+package v2


I know you're aware of all the code duplication, just want to make sure we track it in an issue to investigate clever ways to do the versioning.

trinchan · 2017-01-23T16:19:28Z

pkg/adaptor/elasticsearch/elasticsearch.go

-	hostBits := strings.Split(e.uri.Host, ":")
-	if len(hostBits) > 1 {
-		client.SetPort(hostBits[1])
+	for _, vc := range clients.Clients {


If we make the registry generic, I think a function to return the proper client given the adaptor type and version would be better than having this logic in the adaptor. Even for just elasticsearch, I think it would be better to have this in the registry.

trinchan · 2017-01-23T16:22:41Z

pkg/log/logger.go

@@ -96,6 +98,10 @@ func (l logger) Errorf(format string, args ...interface{}) {
 	l.entry.Errorf(format, args...)
 }

+func (l logger) Printf(format string, args ...interface{}) {


Why is this necessary if we have logger.Infof?

this is to fulfill the Logger interface in the new elasticsearch lib, https://github.com/olivere/elastic/wiki/Logging

…ls Flush for Bulk, add es to travis

…st not do delete via bulk

…empty before flushing

trinchan · 2017-01-24T17:56:29Z

pkg/adaptor/elasticsearch/clients/v2/writer.go

+}
+
+func newWriter(client *elastic.Client, done chan struct{}, wg *sync.WaitGroup) *Writer {
+	p, _ := client.BulkProcessor().


Not sure what we'd do with the errors, but right now we're not catching any of the errors from the BulkProcessor. You can use a BulkAfterFunc and .After(callback) to get at the errors though. Would be bad to throw away errors without any output.

ah, thanks for the reminder, I wanted to log any errors so I'll get that in now

trinchan · 2017-01-24T17:57:12Z

pkg/adaptor/elasticsearch/clients/v5/writer.go

+		BulkActions(1000).               // commit if # requests >= 1000
+		BulkSize(2 << 20).               // commit if size of requests >= 2 MB
+		FlushInterval(30 * time.Second). // commit every 30s
+		Do()


Same here with the error handling

trinchan · 2017-01-24T17:58:15Z

pkg/adaptor/elasticsearch/elasticsearch.go

-		e.pipe.Stop()
-		e.indexer.Stop()
-	}
+	log.With("path", e.path).Infoln("adaptor Stopping...")


ah i noticed the e.path there, it will show the adaptor type, all good.

trinchan · 2017-01-24T18:00:08Z

pkg/adaptor/elasticsearch/transport.go

+
+func newTransport(accessKeyID, secretAccessKey string) http.RoundTripper {
+	t := http.DefaultTransport
+	if accessKeyID != "" && secretAccessKey != "" {


is there any way to test the creds and error sooner?

what would we be testing here? when adding support for AWS signed auth, this seemed like the best/cleanest route to go.

was just wondering if there was a way to bail out early if the creds were bad

gotcha, I honestly don't know

…o elastic.Client

coveralls · 2017-01-24T20:38:01Z

Coverage increased (+4.2%) to 57.076% when pulling 278d25a on 209-es-stability into 1dc0c3d on experimental.

addresses multiple issues with sending data to elasticsearch, handling of _id, major API version differences, and performance fixes #209

jipperinbham added the review label Jan 19, 2017

jipperinbham force-pushed the 209-es-stability branch from 559f011 to 39176ad Compare January 19, 2017 20:55

jipperinbham mentioned this pull request Jan 19, 2017

fatal error: concurrent map read and map write #231

Closed

jipperinbham force-pushed the 209-es-stability branch from 41dc0fc to 021cab2 Compare January 20, 2017 17:42

jipperinbham changed the title ~~[WIP] 209 ES adaptor stability~~ 209 ES adaptor stability Jan 21, 2017

jipperinbham requested a review from trinchan January 23, 2017 15:40

trinchan reviewed Jan 23, 2017

View reviewed changes

This was referenced Jan 23, 2017

Support request signing for AWS auth #203

Closed

FIX: delete _id key to allow elastic insertions #168

Closed

better client registration/versioning #234

Closed

Possible fix for #151 issue #156

Closed

jipperinbham added 19 commits January 24, 2017 11:28

initial changes

c1179d0

rebase from experimental, nuke elastigo

6b1c76d

base es clients for v2.x and v5.x

63eb1de

nuke client.WriteClient, add es v1 client

02e06b9

refactor out custom message structs in place of a Base one, Close cal…

90231aa

…ls Flush for Bulk, add es to travis

update mongodb/bulk_test after merge

9bc6640

test out using docker for running elasticsearch

eabfe2d

try running es based on comments from travis-ci/travis-ci#4542

401dc95

not sure what's happening with the mongodb service

7b142bd

try elasticsearch-2.x for apt add-on

961ff9c

add config directory for es files

40c0382

make sure scripts directory exists

49531d0

remove elasticsearch command, update branch filter to except master

0e0e55d

try elasticsearch 5.x

c2ef51e

lets try a different method

fd3cc9f

initial set of es v5 tests

d231bcc

need to flush bulkprocessor before adding a delete, may be best to ju…

639f400

…st not do delete via bulk

print some debug lines before cat'ing file

b7c4b2b

try and switch to docker based infrastructure

a46fa7a

jipperinbham added 19 commits January 24, 2017 11:28

attempt to run elasticsearch with docker

9dbced5

dont set a custom port yet

5e4f705

use IP instead of localhost, set IP in port forward in docker cmd

3fffebd

skip curl command for now

71c7a3d

attempt to run elasticsearch on different port inside the container

20dcc59

attempt to use custom port for es v5

3933d85

try out edge group on travis, buffer event channel and make sure its …

631f1de

…empty before flushing

attempt to run multiple versions of elasticsearch

536cf70

add tests for ES v1 and v2, switch v1 to not use bulk

6c7cc41

need to test ES_V1_URL in .travis.yml

239003a

add tests for pkg/adaptor/elasticsearch/clients

b6ec9db

remaining tests for elasticsearch adaptor

0d93778

add support AWS request signing for ES adaptor, fixes #159

a5a3ff1

update CHANGELOG and elasticsearch README

bed315f

cleanup exports in tests

27bdc92

rename clients.Closer -> clients.Close, update func comments

6fa22e4

make a copy of msg.Data before calling client.Write

7aa828c

add a test to ensure message copying works and _id is not touched

567c922

rebase from experimental

d9c82d1

jipperinbham force-pushed the 209-es-stability branch from c239adb to d9c82d1 Compare January 24, 2017 17:31

trinchan suggested changes Jan 24, 2017

View reviewed changes

use After callback to log any bulk errors, no need to pass a logger t…

278d25a

…o elastic.Client

jipperinbham merged commit a97392d into experimental Jan 25, 2017

jipperinbham removed the review label Jan 25, 2017

jipperinbham deleted the 209-es-stability branch January 25, 2017 16:05

jipperinbham mentioned this pull request Jan 25, 2017

Elasticsearch adaptor stability #209

Closed

jipperinbham added a commit that referenced this pull request Jan 27, 2017

refactor ES adaptor (#233)

7b7ce3e

addresses multiple issues with sending data to elasticsearch, handling of _id, major API version differences, and performance fixes #209

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

209 ES adaptor stability #233

209 ES adaptor stability #233

jipperinbham commented Jan 19, 2017 •

edited

trinchan left a comment

trinchan Jan 23, 2017

trinchan Jan 23, 2017

jipperinbham Jan 23, 2017

trinchan Jan 23, 2017

jipperinbham Jan 23, 2017

trinchan Jan 23, 2017

jipperinbham Jan 23, 2017

jipperinbham Jan 24, 2017

trinchan Jan 23, 2017

trinchan Jan 23, 2017

trinchan Jan 23, 2017

jipperinbham Jan 24, 2017

trinchan Jan 23, 2017

jipperinbham Jan 23, 2017

trinchan Jan 23, 2017

trinchan Jan 23, 2017

jipperinbham Jan 23, 2017

trinchan Jan 24, 2017

jipperinbham Jan 24, 2017

trinchan Jan 24, 2017

trinchan Jan 24, 2017

trinchan Jan 24, 2017

jipperinbham Jan 24, 2017

trinchan Jan 24, 2017

jipperinbham Jan 24, 2017

coveralls commented Jan 24, 2017

209 ES adaptor stability #233

209 ES adaptor stability #233

Conversation

jipperinbham commented Jan 19, 2017 • edited

trinchan 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

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

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

coveralls commented Jan 24, 2017

jipperinbham commented Jan 19, 2017 •

edited