Memleak fix

This commit is contained in:
2023-04-08 20:45:05 +01:00
parent 7623410bfe
commit 7af4e1f96a
5 changed files with 110 additions and 62 deletions
+30 -6
View File
@@ -43,9 +43,7 @@ func NewConnection() *BaseClient {
b.retries.max = envutil.GetInt("GRAPHQL_RETRIES_MAX", 1) b.retries.max = envutil.GetInt("GRAPHQL_RETRIES_MAX", 1)
b.retries.delay = envutil.GetInt("GRAPHQL_RETRIES_DELAY", 300) b.retries.delay = envutil.GetInt("GRAPHQL_RETRIES_DELAY", 300)
if b.cache.enabled { b.enableCache()
b.enableCache()
}
b_tmp_log_level := envutil.Getenv("LOG_LEVEL", "info") b_tmp_log_level := envutil.Getenv("LOG_LEVEL", "info")
@@ -68,9 +66,20 @@ func NewConnection() *BaseClient {
}) })
var httpClient *http.Client var httpClient *http.Client
httpTransport := &http.Transport{
MaxIdleConns: 10,
IdleConnTimeout: 15 * time.Second,
ResponseHeaderTimeout: 15 * time.Second,
DisableKeepAlives: false,
}
if strings.HasPrefix(b.endpoint, "http://") { if strings.HasPrefix(b.endpoint, "http://") {
httpClient = &http.Client{ httpClient = &http.Client{
Transport: http.DefaultTransport, Transport: httpTransport,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
return http.ErrUseLastResponse
},
} }
} else { } else {
tlsClientConfig := &tls.Config{} tlsClientConfig := &tls.Config{}
@@ -80,6 +89,8 @@ func NewConnection() *BaseClient {
http2Transport := &http2.Transport{ http2Transport := &http2.Transport{
AllowHTTP: true, AllowHTTP: true,
TLSClientConfig: tlsClientConfig, TLSClientConfig: tlsClientConfig,
ReadIdleTimeout: 15 * time.Second,
PingTimeout: 15 * time.Second,
} }
httpClient = &http.Client{ httpClient = &http.Client{
Transport: http2Transport, Transport: http2Transport,
@@ -102,7 +113,21 @@ func (b *BaseClient) SetOutput(output string) {
func (b *BaseClient) enableCache() { func (b *BaseClient) enableCache() {
var err error var err error
b.cache.client, err = bigcache.NewBigCache(bigcache.DefaultConfig(time.Duration(b.cache.ttl) * time.Second)) b.cache.client, err = bigcache.NewBigCache(
// bigcache.DefaultConfig(time.Duration(b.cache.ttl) * time.Second)
bigcache.Config{
Shards: 1024,
LifeWindow: time.Duration(b.cache.ttl) * time.Second,
CleanWindow: time.Duration(b.cache.ttl) * 2 * time.Second,
OnRemove: func(key string, entry []byte) {
b.Logger.Debug(b, "Removing cache entry;", "key", key)
},
MaxEntriesInWindow: 1000 * 10 * 60,
MaxEntrySize: 500,
Verbose: true,
HardMaxCacheSize: 8192,
},
)
if err != nil { if err != nil {
fmt.Println(">> Error while creating cache client;", "error", err.Error()) fmt.Println(">> Error while creating cache client;", "error", err.Error())
panic(err) panic(err)
@@ -112,5 +137,4 @@ func (b *BaseClient) enableCache() {
func (b *BaseClient) disableCache() { func (b *BaseClient) disableCache() {
b.Logger.Debug(b, "Disabling cache") b.Logger.Debug(b, "Disabling cache")
b.cache.enabled = false b.cache.enabled = false
b.cache.client = nil
} }
+23 -55
View File
@@ -5,9 +5,8 @@ import (
"fmt" "fmt"
"strconv" "strconv"
"github.com/lukaszraczylo/go-simple-graphql/utils/helpers"
"github.com/gookit/goutil/strutil" "github.com/gookit/goutil/strutil"
"github.com/lukaszraczylo/go-simple-graphql/utils/helpers"
jsoniter "github.com/json-iterator/go" jsoniter "github.com/json-iterator/go"
) )
@@ -69,68 +68,36 @@ func (c *BaseClient) NewQuery(q ...any) *Query {
// It looks a bit weird because of the backward compatibility // It looks a bit weird because of the backward compatibility
func (c *BaseClient) Query(queryContent string, queryVariables interface{}, queryHeaders map[string]interface{}) (any, error) { func (c *BaseClient) Query(queryContent string, queryVariables interface{}, queryHeaders map[string]interface{}) (any, error) {
compiledQuery := c.NewQuery(queryContent, queryVariables)
parseQueryHeaders, enabledCache, headersModified := c.parseQueryHeaders(queryHeaders)
var should_cache bool
var queryHash string var queryHash string
var cachedResponse []byte
query := c.NewQuery(queryContent, queryVariables) if headersModified {
should_cache = enabledCache
localClientlocal := *c } else {
localClient := &localClientlocal should_cache = c.cache.enabled
// Check for library specific headers
if len(queryHeaders) > 0 {
queryHeadersModified, cache_enabled, headers_modified := c.parseQueryHeaders(queryHeaders)
if headers_modified {
queryHeaders = queryHeadersModified
c.Logger.Debug(c, fmt.Sprintf("Switching cache %s as per single-request header", strconv.FormatBool(cache_enabled)))
if cache_enabled {
// this doesn't make sense - cache is created but its ttl is longer than query execution so it'll be disappeared before it's used
localClient.enableCache()
} else {
localClient.disableCache()
}
}
} }
if localClient.cache.enabled { if should_cache {
queryHash = strutil.Md5(fmt.Sprintf("%s-%+v", query.compiledQuery, queryHeaders)) queryHash = strutil.Md5(fmt.Sprintf("%s-%+v", compiledQuery.compiledQuery, queryHeaders))
c.Logger.Debug(c, "Hash calculated;", "hash:", queryHash)
cachedResponse = c.cacheLookup(queryHash)
if cachedResponse != nil {
c.Logger.Debug(c, "Found cached response")
return c.decodeResponse(cachedResponse), nil
} else {
c.Logger.Debug(c, "No cached response found")
}
} }
response, err := localClient.executeQuery(query.compiledQuery, queryHeaders) q := &queryExecutor{
if err != nil { client: c,
c.Logger.Error(c, "Error while executing query;", "error", err.Error()) query: compiledQuery.compiledQuery,
return nil, err headers: parseQueryHeaders,
context: context.Background(),
should_cache: should_cache,
hash: queryHash,
} }
jsonData, err := json.Marshal(response) q.execute()
if err != nil { defer q.done()
c.Logger.Error(c, "Error while converting to json;", "error", err.Error())
return nil, err
}
if localClient.cache.enabled && c.cache.client != nil && jsonData != nil && queryHash != "" { return q.result.data, q.result.errors
err = c.cache.client.Set(queryHash, jsonData)
if err != nil {
c.Logger.Error(c, "Error while setting cache key;", "error", err.Error())
}
} else if localClient.cache.enabled && c.cache.enabled && jsonData == nil {
c.Logger.Warn(c, "Response is empty")
} else if localClient.cache.enabled && c.cache.enabled && queryHash == "" {
c.Logger.Warn(c, "Query hash is empty")
}
return c.decodeResponse(jsonData), err
} }
func (c *BaseClient) decodeResponse(jsonData []byte) any { func (c *BaseClient) decodeResponse(jsonData []byte) any {
@@ -155,6 +122,7 @@ func (c *BaseClient) decodeResponse(jsonData []byte) any {
func (c *BaseClient) parseQueryHeaders(queryHeaders map[string]interface{}) (returnHeaders map[string]interface{}, cache_enabled bool, headers_modified bool) { func (c *BaseClient) parseQueryHeaders(queryHeaders map[string]interface{}) (returnHeaders map[string]interface{}, cache_enabled bool, headers_modified bool) {
returnHeaders = make(map[string]interface{}) returnHeaders = make(map[string]interface{})
for k, v := range queryHeaders { for k, v := range queryHeaders {
if k == "gqlcache" { if k == "gqlcache" {
cache_enabled, _ = strconv.ParseBool(fmt.Sprintf("%v", v)) cache_enabled, _ = strconv.ParseBool(fmt.Sprintf("%v", v))
+1 -1
View File
@@ -298,10 +298,10 @@ func TestBaseClient_parseQueryHeaders(t *testing.T) {
queryHeaders map[string]interface{} queryHeaders map[string]interface{}
} }
tests := []struct { tests := []struct {
name string
fields fields fields fields
args args args args
wantReturnHeaders map[string]interface{} wantReturnHeaders map[string]interface{}
name string
}{ }{
{ {
name: "Test parseQueryHeaders - no change", name: "Test parseQueryHeaders - no change",
+41
View File
@@ -3,6 +3,7 @@ package gql
import ( import (
"bytes" "bytes"
"fmt" "fmt"
"io"
"io/ioutil" "io/ioutil"
"net/http" "net/http"
"time" "time"
@@ -31,6 +32,7 @@ func (c *BaseClient) executeQuery(query []byte, headers any) (result any, err er
c.Logger.Error(c, "Error while executing http request;", "error", err.Error()) c.Logger.Error(c, "Error while executing http request;", "error", err.Error())
return err return err
} }
defer io.Copy(io.Discard, httpResponse.Body) // equivalent to `cp body /dev/null`
defer httpResponse.Body.Close() defer httpResponse.Body.Close()
if httpResponse.StatusCode <= 200 && httpResponse.StatusCode >= 204 { if httpResponse.StatusCode <= 200 && httpResponse.StatusCode >= 204 {
@@ -65,3 +67,42 @@ func (c *BaseClient) executeQuery(query []byte, headers any) (result any, err er
} }
return queryResult.Data, nil return queryResult.Data, nil
} }
func (q *queryExecutor) execute() {
if q.should_cache {
cachedResponse := q.client.cacheLookup(q.hash)
if cachedResponse != nil {
q.client.Logger.Debug(q.client, "Found cached response")
q.result.data = q.client.decodeResponse(cachedResponse)
} else {
q.client.Logger.Debug(q.client, "No cached response found")
}
}
response, err := q.client.executeQuery(q.query, q.headers)
if err != nil {
q.client.Logger.Error(q.client, "Error while executing query;", "error", err.Error())
q.result.errors = err
return
}
jsonData, err := json.Marshal(response)
if err != nil {
q.client.Logger.Error(q.client, "Error while converting to json;", "error", err.Error())
q.result.errors = err
return
}
if q.should_cache {
err = q.client.cache.client.Set(q.hash, jsonData)
if err != nil {
q.client.Logger.Error(q.client, "Error while setting cache key;", "error", err.Error())
}
}
q.result.data = q.client.decodeResponse(jsonData)
}
func (q *queryExecutor) done() {
q = nil
}
+15
View File
@@ -44,6 +44,21 @@ type Query struct {
compiledQuery []byte `json:"compiledQuery"` compiledQuery []byte `json:"compiledQuery"`
} }
type queryExecutor struct {
result queryExecutorResult
context context.Context
client *BaseClient
headers map[string]interface{}
hash string
query []byte
should_cache bool
}
type queryExecutorResult struct {
data interface{}
errors error
}
type request struct { type request struct {
Variables any `json:"variables"` Variables any `json:"variables"`
Query string `json:"query"` Query string `json:"query"`