diff --git a/elasticsearch.go b/elasticsearch.go index 57fa863..c111238 100644 --- a/elasticsearch.go +++ b/elasticsearch.go @@ -1,18 +1,20 @@ package main import ( + "net/url" "encoding/json" "github.com/buger/goreplay/proto" "github.com/mattbaird/elastigo/lib" "log" - "regexp" + "strings" + //"regexp" "time" ) type ESUriErorr struct{} func (e *ESUriErorr) Error() string { - return "Wrong ElasticSearch URL format. Expected to be: host:port/index_name" + return "Wrong ElasticSearch URL format. Expected to be: scheme://host/index_name" } type ESPlugin struct { @@ -52,17 +54,27 @@ type ESRequestResponse struct { // Parse ElasticSearch URI // -// Proper format is: host:port/index_name -func parseURI(URI string) (err error, host string, port string, index string) { - rURI := regexp.MustCompile("(.+):([0-9]+)/(.+)") - match := rURI.FindAllStringSubmatch(URI, -1) +// Proper format is: scheme://[userinfo@]host/index_name +// userinfo is: user[:password] +// net/url.Parse() does not fail if scheme is not provided but actualy does not +// handle URI properly. +// So we must 'validate' URI format to match requirements to use net/url.Parse() +func parseURI(URI string) (err error, index string) { - if len(match) == 0 { + parsedUrl, parseErr := url.Parse(URI) + + if parseErr != nil { + err = new(ESUriErorr) + } + + // check URL validity by extracting host and undex values. + host := parsedUrl.Host + urlPathParts := strings.Split(parsedUrl.Path, "/") + index = urlPathParts[len(urlPathParts) - 1 ] + + // force index specification in uri : ie no implicit index + if (host == "" || index == "") { err = new(ESUriErorr) - } else { - host = match[0][1] - port = match[0][2] - index = match[0][3] } return @@ -71,14 +83,15 @@ func parseURI(URI string) (err error, host string, port string, index string) { func (p *ESPlugin) Init(URI string) { var err error - err, p.Host, p.ApiPort, p.Index = parseURI(URI) + err, p.Index = parseURI(URI) if err != nil { log.Fatal("Can't initialize ElasticSearch plugin.", err) } + p.eConn = elastigo.NewConn() - p.eConn.SetPort(p.ApiPort) - p.eConn.SetHosts([]string{p.Host}) + + p.eConn.SetFromUrl(URI) p.indexor = p.eConn.NewBulkIndexerErrors(50, 60) p.done = make(chan bool) diff --git a/elasticsearch_test.go b/elasticsearch_test.go new file mode 100644 index 0000000..701c38f --- /dev/null +++ b/elasticsearch_test.go @@ -0,0 +1,140 @@ +package main + +import ( + "testing" +) + +const expectedIndex = "gor" + +func assertExpectedGorIndex (index string, t *testing.T) { + if expectedIndex != index { + t.Fatalf("Expected index %s but got %s", expectedIndex, index) + } +} + +func assertExpectedIndex (expectedIndex string, index string, t *testing.T) { + if expectedIndex != index { + t.Fatalf("Expected index %s but got %s", expectedIndex, index) + } +} + +func assertExpectedError (returnedError error, t *testing.T) { + expectedError := new(ESUriErorr) + + if expectedError != returnedError { + t.Errorf("Expected err %s but got %s", expectedError, returnedError) + } +} + +func assertNoError (returnedError error, t *testing.T) { + if nil != returnedError { + t.Errorf("Expected err %s but got %s", nil, returnedError) + } +} + +// Argument host:port/index_name +// i.e : localhost:9200/gor +// Fail because scheme is mandatory +func TestElasticConnectionBuildFailWithoutScheme(t *testing.T) { + uri := "localhost:9200/" + expectedIndex + + err, _ := parseURI(uri) + assertExpectedError(err, t) +} + +// Argument scheme://host:port +// i.e : http://localhost:9200 +// Fail : explicit index is required +func TestElasticConnectionBuildFailWithoutIndex(t *testing.T) { + uri := "http://localhost:9200" + + err, index := parseURI(uri) + + assertExpectedIndex("", index, t) + + assertExpectedError(err, t) +} + +// Argument scheme://host/index_name +// i.e : http://localhost/gor +func TestElasticConnectionBuildFailWithoutPort(t *testing.T) { + uri := "http://localhost/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : http://localhost:9200/gor +func TestElasticLocalConnectionBuild(t *testing.T) { + uri := "http://localhost:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : http://localhost.local:9200/gor or https://localhost.local:9200/gor +func TestElasticSimpleLocalWithSchemeConnectionBuild(t *testing.T) { + uri := "http://localhost.local:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : http://localhost.local:9200/gor or https://localhost.local:9200/gor +func TestElasticSimpleLocalWithHTTPSConnectionBuild(t *testing.T) { + uri := "https://localhost.local:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : localhost.local:9200/pathtoElastic/gor +func TestElasticLongPathConnectionBuild(t *testing.T) { + uri := "http://localhost.local:9200/pathtoElastic/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:userinfo@port/index_name +// i.e : http://user:password@localhost.local:9200/gor +func TestElasticBasicAuthConnectionBuild(t *testing.T) { + uri := "http://user:password@localhost.local:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/path/index_name +// i.e : http://localhost.local:9200/path/gor or https://localhost.local:9200/path/gor +func TestElasticComplexPathConnectionBuild(t *testing.T) { + uri := "http://localhost.local:9200/path/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +}