message namespace no longer includes a "db"

CHANGELOG.md

@@ -1,3 +1,12 @@
+## v0.1.3 [UNRELEASED]
+
+### Breaking changes:
+- namespace processing no longer expects their the be a "db" portion (i.e. "database.collection")
+but an attempt to maintain backwards compatibility is still there for the time being.
+[#258](https://github.com/compose/transporter/pull/258)
+
+### Bugfixes
+
 ## v0.1.2 [2017-01-27]
 
 This release is primarily aimed at getting the MongoDB and Elasticsearch adaptors into a

pkg/adaptor/elasticsearch/clients/registry.go

@@ -33,4 +33,5 @@ type ClientOptions struct {
 	UserInfo   *url.Userinfo
 	HTTPClient *http.Client
 	Path       string
+	Index      string
 }

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

@@ -22,6 +22,7 @@ var (
 // Writer implements client.Writer and client.Session for sending requests to an elasticsearch
 // cluster in individual requests.
 type Writer struct {
+	index    string
 	esClient *elastic.Client
 	logger   log.Logger
 }
@@ -45,6 +46,7 @@ func init() {
 			return nil, err
 		}
 		w := &Writer{
+			index:    opts.Index,
 			esClient: esClient,
 			logger:   log.With("path", opts.Path).With("writer", "elasticsearch").With("version", 1),
 		}
@@ -56,7 +58,7 @@ func init() {
 
 func (w *Writer) Write(msg message.Msg) func(client.Session) error {
 	return func(s client.Session) error {
-		i, t, _ := message.SplitNamespace(msg)
+		indexType := msg.Namespace()
 		var id string
 		if _, ok := msg.Data()["_id"]; ok {
 			id = msg.ID()
@@ -65,11 +67,11 @@ func (w *Writer) Write(msg message.Msg) func(client.Session) error {
 		var err error
 		switch msg.OP() {
 		case ops.Delete:
-			_, err = w.esClient.Delete().Index(i).Type(t).Id(id).Do(context.TODO())
+			_, err = w.esClient.Delete().Index(w.index).Type(indexType).Id(id).Do(context.TODO())
 		case ops.Insert:
-			_, err = w.esClient.Index().Index(i).Type(t).Id(id).BodyJson(msg.Data()).Do(context.TODO())
+			_, err = w.esClient.Index().Index(w.index).Type(indexType).Id(id).BodyJson(msg.Data()).Do(context.TODO())
 		case ops.Update:
-			_, err = w.esClient.Index().Index(i).Type(t).BodyJson(msg.Data()).Id(id).Do(context.TODO())
+			_, err = w.esClient.Index().Index(w.index).Type(indexType).BodyJson(msg.Data()).Id(id).Do(context.TODO())
 		}
 		return err
 	}

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

@@ -18,6 +18,7 @@ import (
 const (
 	defaultURL   = "http://127.0.0.1:9200"
 	defaultIndex = "test_v1"
+	testType     = "test"
 )
 
 var (
@@ -28,10 +29,6 @@ func fullURL(suffix string) string {
 	return fmt.Sprintf("%s/%s%s", testURL, defaultIndex, suffix)
 }
 
-func testNS() string {
-	return fmt.Sprintf("%s.%s", defaultIndex, "test")
-}
-
 func setup() error {
 	log.Debugln("setting up tests")
 	return clearTestData()
@@ -75,13 +72,14 @@ func TestWriter(t *testing.T) {
 		URLs:       []string{testURL},
 		HTTPClient: http.DefaultClient,
 		Path:       defaultIndex,
+		Index:      defaultIndex,
 	}
 	vc := clients.Clients["v1"]
 	w, _ := vc.Creator(done, &wg, opts)
-	w.Write(message.From(ops.Insert, testNS(), map[string]interface{}{"hello": "world"}))(nil)
-	w.Write(message.From(ops.Insert, testNS(), map[string]interface{}{"_id": "booya", "hello": "world"}))(nil)
-	w.Write(message.From(ops.Update, testNS(), map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
-	w.Write(message.From(ops.Delete, testNS(), map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
+	w.Write(message.From(ops.Insert, testType, map[string]interface{}{"hello": "world"}))(nil)
+	w.Write(message.From(ops.Insert, testType, map[string]interface{}{"_id": "booya", "hello": "world"}))(nil)
+	w.Write(message.From(ops.Update, testType, map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
+	w.Write(message.From(ops.Delete, testType, map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
 	close(done)
 	wg.Wait()
 

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

@@ -23,6 +23,7 @@ var (
 // Writer implements client.Writer and client.Session for sending requests to an elasticsearch
 // cluster via its _bulk API.
 type Writer struct {
+	index  string
 	bp     *elastic.BulkProcessor
 	logger log.Logger
 }
@@ -46,6 +47,7 @@ func init() {
 			return nil, err
 		}
 		w := &Writer{
+			index:  opts.Index,
 			logger: log.With("writer", "elasticsearch").With("version", 2).With("path", opts.Path),
 		}
 		p, err := esClient.BulkProcessor().
@@ -68,7 +70,7 @@ func init() {
 
 func (w *Writer) Write(msg message.Msg) func(client.Session) error {
 	return func(s client.Session) error {
-		i, t, _ := message.SplitNamespace(msg)
+		indexType := msg.Namespace()
 		var id string
 		if _, ok := msg.Data()["_id"]; ok {
 			id = msg.ID()
@@ -81,11 +83,11 @@ func (w *Writer) Write(msg message.Msg) func(client.Session) error {
 			// we need to flush any pending writes here or this could fail because we're using
 			// more than 1 worker
 			w.bp.Flush()
-			br = elastic.NewBulkDeleteRequest().Index(i).Type(t).Id(id)
+			br = elastic.NewBulkDeleteRequest().Index(w.index).Type(indexType).Id(id)
 		case ops.Insert:
-			br = elastic.NewBulkIndexRequest().Index(i).Type(t).Id(id).Doc(msg.Data())
+			br = elastic.NewBulkIndexRequest().Index(w.index).Type(indexType).Id(id).Doc(msg.Data())
 		case ops.Update:
-			br = elastic.NewBulkUpdateRequest().Index(i).Type(t).Id(id).Doc(msg.Data())
+			br = elastic.NewBulkUpdateRequest().Index(w.index).Type(indexType).Id(id).Doc(msg.Data())
 		}
 		w.bp.Add(br)
 		return nil

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

@@ -18,6 +18,7 @@ import (
 const (
 	defaultURL   = "http://127.0.0.1:9200"
 	defaultIndex = "test_v2"
+	testType     = "test"
 )
 
 var (
@@ -28,10 +29,6 @@ func fullURL(suffix string) string {
 	return fmt.Sprintf("%s/%s%s", testURL, defaultIndex, suffix)
 }
 
-func testNS() string {
-	return fmt.Sprintf("%s.%s", defaultIndex, "test")
-}
-
 func setup() error {
 	log.Debugln("setting up tests")
 	return clearTestData()
@@ -75,13 +72,14 @@ func TestWriter(t *testing.T) {
 		URLs:       []string{testURL},
 		HTTPClient: http.DefaultClient,
 		Path:       defaultIndex,
+		Index:      defaultIndex,
 	}
 	vc := clients.Clients["v2"]
 	w, _ := vc.Creator(done, &wg, opts)
-	w.Write(message.From(ops.Insert, testNS(), map[string]interface{}{"hello": "world"}))(nil)
-	w.Write(message.From(ops.Insert, testNS(), map[string]interface{}{"_id": "booya", "hello": "world"}))(nil)
-	w.Write(message.From(ops.Update, testNS(), map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
-	w.Write(message.From(ops.Delete, testNS(), map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
+	w.Write(message.From(ops.Insert, testType, map[string]interface{}{"hello": "world"}))(nil)
+	w.Write(message.From(ops.Insert, testType, map[string]interface{}{"_id": "booya", "hello": "world"}))(nil)
+	w.Write(message.From(ops.Update, testType, map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
+	w.Write(message.From(ops.Delete, testType, map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
 	close(done)
 	wg.Wait()
 

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

@@ -23,6 +23,7 @@ var (
 // Writer implements client.Writer and client.Session for sending requests to an elasticsearch
 // cluster via its _bulk API.
 type Writer struct {
+	index  string
 	bp     *elastic.BulkProcessor
 	logger log.Logger
 }
@@ -46,6 +47,7 @@ func init() {
 			return nil, err
 		}
 		w := &Writer{
+			index:  opts.Index,
 			logger: log.With("writer", "elasticsearch").With("version", 5).With("path", opts.Path),
 		}
 		p, err := esClient.BulkProcessor().
@@ -68,7 +70,7 @@ func init() {
 
 func (w *Writer) Write(msg message.Msg) func(client.Session) error {
 	return func(s client.Session) error {
-		i, t, _ := message.SplitNamespace(msg)
+		indexType := msg.Namespace()
 		var id string
 		if _, ok := msg.Data()["_id"]; ok {
 			id = msg.ID()
@@ -81,11 +83,11 @@ func (w *Writer) Write(msg message.Msg) func(client.Session) error {
 			// we need to flush any pending writes here or this could fail because we're using
 			// more than 1 worker
 			w.bp.Flush()
-			br = elastic.NewBulkDeleteRequest().Index(i).Type(t).Id(id)
+			br = elastic.NewBulkDeleteRequest().Index(w.index).Type(indexType).Id(id)
 		case ops.Insert:
-			br = elastic.NewBulkIndexRequest().Index(i).Type(t).Id(id).Doc(msg.Data())
+			br = elastic.NewBulkIndexRequest().Index(w.index).Type(indexType).Id(id).Doc(msg.Data())
 		case ops.Update:
-			br = elastic.NewBulkUpdateRequest().Index(i).Type(t).Id(id).Doc(msg.Data())
+			br = elastic.NewBulkUpdateRequest().Index(w.index).Type(indexType).Id(id).Doc(msg.Data())
 		}
 		w.bp.Add(br)
 		return nil

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

@@ -18,6 +18,7 @@ import (
 const (
 	defaultURL   = "http://127.0.0.1:9200"
 	defaultIndex = "test_v5"
+	testType     = "test"
 )
 
 var (
@@ -28,10 +29,6 @@ func fullURL(suffix string) string {
 	return fmt.Sprintf("%s/%s%s", testURL, defaultIndex, suffix)
 }
 
-func testNS() string {
-	return fmt.Sprintf("%s.%s", defaultIndex, "test")
-}
-
 func setup() error {
 	log.Debugln("setting up tests")
 	return clearTestData()
@@ -75,13 +72,14 @@ func TestWriter(t *testing.T) {
 		URLs:       []string{testURL},
 		HTTPClient: http.DefaultClient,
 		Path:       defaultIndex,
+		Index:      defaultIndex,
 	}
 	vc := clients.Clients["v5"]
 	w, _ := vc.Creator(done, &wg, opts)
-	w.Write(message.From(ops.Insert, testNS(), map[string]interface{}{"hello": "world"}))(nil)
-	w.Write(message.From(ops.Insert, testNS(), map[string]interface{}{"_id": "booya", "hello": "world"}))(nil)
-	w.Write(message.From(ops.Update, testNS(), map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
-	w.Write(message.From(ops.Delete, testNS(), map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
+	w.Write(message.From(ops.Insert, testType, map[string]interface{}{"hello": "world"}))(nil)
+	w.Write(message.From(ops.Insert, testType, map[string]interface{}{"_id": "booya", "hello": "world"}))(nil)
+	w.Write(message.From(ops.Update, testType, map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
+	w.Write(message.From(ops.Delete, testType, map[string]interface{}{"_id": "booya", "hello": "goodbye"}))(nil)
 	close(done)
 	wg.Wait()
 

pkg/adaptor/elasticsearch/elasticsearch.go

@@ -71,15 +71,6 @@ func (e VersionError) Error() string {
 	return fmt.Sprintf("%s running %s, %s", e.uri, e.v, e.err)
 }
 
-// InvalidTimeoutError wraps the underlying error when the provided is not parsable time.ParseDuration
-// type InvalidTimeoutError struct {
-// 	timeout string
-// }
-//
-// func (e InvalidTimeoutError) Error() string {
-// 	return fmt.Sprintf("Invalid Timeout, %s", e.timeout)
-// }
-
 // Elasticsearch is an adaptor to connect a pipeline to
 // an elasticsearch cluster.
 type Elasticsearch struct {
@@ -164,24 +155,19 @@ func (e *Elasticsearch) Stop() error {
 }
 
 func (e *Elasticsearch) applyOp(msg message.Msg) (message.Msg, error) {
-	_, msgColl, _ := message.SplitNamespace(msg)
 	msgCopy := make(map[string]interface{})
 	// Copy from the original map to the target map
 	for key, value := range msg.Data() {
 		msgCopy[key] = value
 	}
-	err := e.client.Write(message.From(msg.OP(), e.computeNamespace(msgColl), msgCopy))(nil)
+	err := e.client.Write(message.From(msg.OP(), msg.Namespace(), msgCopy))(nil)
 
 	if err != nil {
 		e.pipe.Err <- adaptor.NewError(adaptor.ERROR, e.path, fmt.Sprintf("write message error (%s)", err), msg.Data)
 	}
 	return msg, err
 }
 
-func (e *Elasticsearch) computeNamespace(Type string) string {
-	return fmt.Sprintf("%s.%s", e.index, Type)
-}
-
 func (e *Elasticsearch) setupClient(conf Config) error {
 	uri, err := url.Parse(conf.URI)
 	if err != nil {
@@ -221,6 +207,7 @@ func (e *Elasticsearch) setupClient(conf Config) error {
 				UserInfo:   uri.User,
 				HTTPClient: httpClient,
 				Path:       e.path,
+				Index:      e.index,
 			}
 			versionedClient, _ := vc.Creator(e.doneChannel, &e.wg, opts)
 			e.client = versionedClient

pkg/adaptor/etcd/etcd.go

@@ -6,10 +6,10 @@ import (
 	"strings"
 	"time"
 
-	log "github.com/Sirupsen/logrus"
 	"golang.org/x/net/context"
 
 	"github.com/compose/transporter/pkg/adaptor"
+	"github.com/compose/transporter/pkg/log"
 	"github.com/compose/transporter/pkg/message"
 	"github.com/compose/transporter/pkg/message/adaptor/etcd"
 	"github.com/compose/transporter/pkg/message/data"
@@ -77,7 +77,7 @@ func init() {
 		if conf.Timeout != "" {
 			t, err := time.ParseDuration(conf.Timeout)
 			if err != nil {
-				log.Printf("error parsing timeout, defaulting to 10s, %v", err)
+				log.Errorf("error parsing timeout, defaulting to 10s, %v", err)
 			} else {
 				e.sessionTimeout = t
 			}

pkg/adaptor/mongodb/bulk.go

@@ -25,6 +25,7 @@ var (
 // Bulk implements client.Writer for use with MongoDB and takes advantage of the Bulk API for
 // performance improvements.
 type Bulk struct {
+	db      string
 	bulkMap map[string]*bulkOperation
 	*sync.RWMutex
 }
@@ -40,8 +41,9 @@ type bulkOperation struct {
 	*sync.Mutex
 }
 
-func newBulker(done chan struct{}, wg *sync.WaitGroup) *Bulk {
+func newBulker(db string, done chan struct{}, wg *sync.WaitGroup) *Bulk {
 	b := &Bulk{
+		db:      db,
 		bulkMap: make(map[string]*bulkOperation),
 		RWMutex: &sync.RWMutex{},
 	}
@@ -52,15 +54,15 @@ func newBulker(done chan struct{}, wg *sync.WaitGroup) *Bulk {
 
 func (b *Bulk) Write(msg message.Msg) func(client.Session) error {
 	return func(s client.Session) error {
-		db, coll, _ := message.SplitNamespace(msg)
+		coll := msg.Namespace()
 		b.RLock()
 		bOp, ok := b.bulkMap[coll]
 		b.RUnlock()
 		if !ok {
 			s := s.(*Session).mgoSession.Copy()
 			bOp = &bulkOperation{
 				s:     s,
-				bulk:  s.DB(db).C(coll).Bulk(),
+				bulk:  s.DB(b.db).C(coll).Bulk(),
 				Mutex: &sync.Mutex{},
 			}
 			b.Lock()