The client does not reuse HTTP connections

[SamuraiPrinciple]

Greetings,

The documentation (and issue tracker, see #118) seems to imply that HTTP connections should be reused, but that doesn't seem to be the case. Here is a small app:

package main

import (
	"context"
	"fmt"
	"log"
	"strings"

	"github.com/elastic/go-elasticsearch/v7"
	"github.com/elastic/go-elasticsearch/v7/esapi"
)

func main() {
	es, err := elasticsearch.NewClient(elasticsearch.Config{
		Addresses: []string{"http://localhost:9200"},
		//Transport: &http.Transport{MaxIdleConns: 8, MaxIdleConnsPerHost: 8, MaxConnsPerHost: 16, IdleConnTimeout: 10 * time.Second},
	})
	if err != nil {
		panic(err)
	}
	for i := 0; i < 1000; i++ {
		func() {
			log.Printf("Saving %d\n", i)
			req := esapi.IndexRequest{
				Index:      "test",
				DocumentID: fmt.Sprintf("%d", i),
				Body:       strings.NewReader(`{}`),
				Refresh:    "true",
			}
			res, err := req.Do(context.Background(), es)
			if err != nil {
				log.Println(err)
			}
			defer func() {
				err := res.Body.Close()
				if err != nil {
					log.Println(err)
				}
			}()
			if res.IsError() {
				log.Println(res.Status())
			}
		}()
	}
}

Before running the app netstat -an | grep 9200 shows:

tcp4       0      0  127.0.0.1.9200         *.*                    LISTEN     
tcp6       0      0  ::1.9200               *.*                    LISTEN     

If I then execute netstat -an | grep 9200 | awk '{print $6}' | sort | uniq -c while the app is running, the number of connections in TIME_WAIT state keeps growing. Running it immediately after the app finishes:

   2 LISTEN
1000 TIME_WAIT

Reproduced on both linux and MacOS with Elasticsearch 7.5.1, go 1.13.6 and elastic/go-elasticsearch 7.5.0.

Many thanks,
damjan

[karmi]

Hey, yes, the connections should definitely be reused (assuming the body is closed). There's this test:

go-elasticsearch/elasticsearch_integration_test.go

Line 29 in 5398a82

t.Run("Persistent", func(t *testing.T) {

which essentially runs what you have in the example, and uses the nodes stats in Elasticsearch to check that the number of connections is not growing — did you try to run that as well?

[SamuraiPrinciple]

Hi, thanks for the quick reply (and reformatting my message too :)).

Yes, that's the test mentioned in #118, but it's asserting a slightly different thing (that number of open connections isn't growing). And yes, it's behaving in the same way:

 damjan@iMac  ~/Downloads/go-elasticsearch-master  go test elasticsearch_integration_test.go                  
ok  	command-line-arguments	1.335s
 damjan@iMac  ~/Downloads/go-elasticsearch-master  netstat -an | grep 9200 | awk '{print $6}' | sort | uniq -c
   2 LISTEN
 132 TIME_WAIT

Hope this helps,
damjan

[SamuraiPrinciple]

It seems like closing HTTP response body without consuming the response is causing a connection to be closed by go HTTP stack. So invoking res.String(), just for the side effect, makes the issue go away. Which is obviously a workaround, but quite silly...

package main

import (
	"context"
	"fmt"
	"log"
	"strings"

	"github.com/elastic/go-elasticsearch/v7"
	"github.com/elastic/go-elasticsearch/v7/esapi"
)

func main() {
	es, err := elasticsearch.NewClient(elasticsearch.Config{
		Addresses: []string{"http://localhost:9200"},
		//Transport: &http.Transport{MaxIdleConns: 8, MaxIdleConnsPerHost: 8, MaxConnsPerHost: 16, IdleConnTimeout: 10 * time.Second},
	})
	if err != nil {
		panic(err)
	}
	for i := 0; i < 1000; i++ {
		func() {
			log.Printf("Saving %d\n", i)
			req := esapi.IndexRequest{
				Index:      "test",
				DocumentID: fmt.Sprintf("%d", i),
				Body:       strings.NewReader(`{}`),
				Refresh:    "true",
			}
			res, err := req.Do(context.Background(), es)
			if err != nil {
				log.Println(err)
			}
			defer func() {
				log.Println("Closing")
				err := res.Body.Close()
				if err != nil {
					log.Println(err)
				}
			}()
			res.String()
			if res.IsError() {
				log.Println(res.Status())
			}
		}()
	}
}

This now produces (the expected):

netstat -an | grep 9200 | awk '{print $6}' | sort | uniq -c
   2 LISTEN
   1 TIME_WAIT

[SamuraiPrinciple]

This seems "by design" - see here:

golang/go@ce83415

And here:

https://golang.org/pkg/net/http/#Response

It is the caller's responsibility to close Body. The default HTTP client's Transport may not reuse HTTP/1.x "keep-alive" TCP connections if the Body is not read to completion and closed.

I guess the HTTP usage pattern with this elastic client is slightly different - you're mostly not interested in response body (just the HTTP status code) when you're indexing documents. Not sure if you'd be keen to change the client so that it internally consumes the response body "eagerly", thus allowing the connection to be reused.

This is quite problematic in high-throughput scenarios because it can easily cause the exhaustion of (ip, port) touples space when a new connection is created for each request.

[karmi]

Right, that's "obvious" from the Go design, but perhaps not from the usual client usage. Thanks for all the info — I'll think about some good way to highlight this in the documentation.

[turboezh]

Hi. Just for note. Found this thread after two hours of debugging gorouting/mem leak. In my case I've just ignored response from bulk op: _, err = Client.Bulk(...).

[karmi]

I've added a note about closing and consuming the reponse body in order to re-use connections into the README in bbd6b17 — thanks for raising the issue.

---

@turboezh Curious about a use case where you send a bulk request, but you're not interested in the response, ie. you don't care if documents have been indexed succesfully or not?

[turboezh]

@karmi I'm new to ES and this API, wasn't sure how to properly use bulk and there is some kind of a prototype code here. I assumed that err indicates any non-success response. I see now that op response is just a wrapped http-response and I must close the body and inspect status code also (and look into the body for each op result).

[karmi]

@turboezh , right, err is only returned when the response cannot be retrieved (host down, etc).

Have a look at _examples/bulk, it has a lot of code for usual scenarios, and an example of the esutil.BulkIndexer helper, which is preferable to calling the Bulk API manually.

[turboezh]

@karmi Thank you!

[c5433137]

hi when i use context to control request timeout，i also find goroutinue leak。
the most goroutinue number is ：
41423 @ 0x488780 0x498e40 0x498e2b 0x498ba7 0x4c539c 0xf965b8 0xf965c1 0xd500f1 0xeb8349 0x116697e 0x1172590 0x4b6881

0x498ba6 sync.runtime_SemacquireMutex+0x46 /usr/local/go/src/runtime/sema.go:71

0x4c539b sync.(*Mutex).lockSlow+0xfb /usr/local/go/src/sync/mutex.go:138

0xf965b7 sync.(*Mutex).Lock+0xb47 /usr/local/go/src/sync/mutex.go:81

0xf965c0 github.com/elastic/go-elasticsearch/v7/estransport.(*Client).Perform+0xb50 /mnt/hgfs/go/pkg/mod/github.com/elastic/go-elasticsearch/v7@v7.6.0/estransport/estransport.go:209

0xd500f0 github.com/elastic/go-elasticsearch/v7/esapi.MgetRequest.Do+0x920 /mnt/hgfs/go/pkg/mod/github.com/elastic/go-elasticsearch/v7@v7.6.0/esapi/api.mget.go:168

0xeb8348 github.com/elastic/go-elasticsearch/v7/esapi.newMgetFunc.func1+0x198 /mnt/hgfs/go/pkg/mod/github.com/elastic/go-elasticsearch/v7@v7.6.0/esapi/api.mget.go:23