Merge #297

297: rr.service unit file r=48d90782 a=48d90782

resolves #296 

Co-authored-by: Valery Piashchynski <[email protected]>

1 files changed, 295 insertions, 196 deletions

diff --git a/service/limit/service_test.go b/service/limit/service_test.go
index 5f6ff183..856ff1a6 100644
--- a/service/limit/service_test.go
+++ b/service/limit/service_test.go
@@ -2,6 +2,7 @@ package limit
 
 import (
 	"fmt"
+	"github.com/cenkalti/backoff/v4"
 	json "github.com/json-iterator/go"
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
@@ -49,347 +50,445 @@ func (cfg *testCfg) Unmarshal(out interface{}) error {
 }
 
 func Test_Service_PidEcho(t *testing.T) {
-	logger, _ := test.NewNullLogger()
-	logger.SetLevel(logrus.DebugLevel)
+	bkoff := backoff.NewExponentialBackOff()
+	bkoff.MaxElapsedTime = time.Second * 15
 
-	c := service.NewContainer(logger)
-	c.Register(rrhttp.ID, &rrhttp.Service{})
-	c.Register(ID, &Service{})
+	err := backoff.Retry(func() error {
+		logger, _ := test.NewNullLogger()
+		logger.SetLevel(logrus.DebugLevel)
 
-	assert.NoError(t, c.Init(&testCfg{
-		httpCfg: `{
+		c := service.NewContainer(logger)
+		c.Register(rrhttp.ID, &rrhttp.Service{})
+		c.Register(ID, &Service{})
+
+		err := c.Init(&testCfg{
+			httpCfg: `{
 			"address": ":7029",
 			"workers":{
 				"command": "php ../../tests/http/client.php pid pipes",
 				"pool": {"numWorkers": 1}
 			}
 	}`,
-		limitCfg: `{
+			limitCfg: `{
 			"services": {
 				"http": {
 					"ttl": 1
 				}
 			}
 		}`,
-	}))
+		})
+		if err != nil {
+			return err
+		}
+
+		s, _ := c.Get(rrhttp.ID)
+		assert.NotNil(t, s)
+
+		go func() {
+			err := c.Serve()
+			if err != nil {
+				t.Errorf("error during the Serve: error %v", err)
+			}
+		}()
 
-	s, _ := c.Get(rrhttp.ID)
-	assert.NotNil(t, s)
+		time.Sleep(time.Millisecond * 100)
+		req, err := http.NewRequest("GET", "http://localhost:7029", nil)
+		if err != nil {
+			return err
+		}
 
-	go func() {
-		err := c.Serve()
+		r, err := http.DefaultClient.Do(req)
 		if err != nil {
-			t.Errorf("error during the Serve: error %v", err)
+			return err
 		}
-	}()
 
-	time.Sleep(time.Millisecond * 100)
-	req, err := http.NewRequest("GET", "http://localhost:7029", nil)
-	assert.NoError(t, err)
 
-	r, err := http.DefaultClient.Do(req)
-	assert.NoError(t, err)
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			return err
+		}
 
+		assert.Equal(t, getPID(s), string(b))
 
-	b, err := ioutil.ReadAll(r.Body)
-	assert.NoError(t, err)
+		err2 := r.Body.Close()
+		if err2 != nil {
+			t.Errorf("error during the body closing: error %v", err2)
+		}
+		c.Stop()
+		return nil
 
-	assert.NoError(t, err)
-	assert.Equal(t, getPID(s), string(b))
+	}, bkoff)
 
-	err2 := r.Body.Close()
-	if err2 != nil {
-		t.Errorf("error during the body closing: error %v", err2)
+	if err != nil {
+		t.Fatal(err)
 	}
-	c.Stop()
+
 }
 
 func Test_Service_ListenerPlusTTL(t *testing.T) {
-	logger, _ := test.NewNullLogger()
-	logger.SetLevel(logrus.DebugLevel)
+	bkoff := backoff.NewExponentialBackOff()
+	bkoff.MaxElapsedTime = time.Second * 15
+
+	err := backoff.Retry(func() error {
+		logger, _ := test.NewNullLogger()
+		logger.SetLevel(logrus.DebugLevel)
 
-	c := service.NewContainer(logger)
-	c.Register(rrhttp.ID, &rrhttp.Service{})
-	c.Register(ID, &Service{})
+		c := service.NewContainer(logger)
+		c.Register(rrhttp.ID, &rrhttp.Service{})
+		c.Register(ID, &Service{})
 
-	assert.NoError(t, c.Init(&testCfg{
-		httpCfg: `{
+		err :=  c.Init(&testCfg{
+			httpCfg: `{
 			"address": ":7030",
 			"workers":{
 				"command": "php ../../tests/http/client.php pid pipes",
 				"pool": {"numWorkers": 1}
 			}
 	}`,
-		limitCfg: `{
+			limitCfg: `{
 			"services": {
 				"http": {
 					"ttl": 1
 				}
 			}
 		}`,
-	}))
+		})
+		if err != nil {
+			return err
+		}
 
-	s, _ := c.Get(rrhttp.ID)
-	assert.NotNil(t, s)
+		s, _ := c.Get(rrhttp.ID)
+		assert.NotNil(t, s)
 
-	l, _ := c.Get(ID)
-	captured := make(chan interface{})
-	l.(*Service).AddListener(func(event int, ctx interface{}) {
-		if event == EventTTL {
-			close(captured)
-		}
-	})
+		l, _ := c.Get(ID)
+		captured := make(chan interface{})
+		l.(*Service).AddListener(func(event int, ctx interface{}) {
+			if event == EventTTL {
+				close(captured)
+			}
+		})
 
-	go func() {
-		err := c.Serve()
-		if err != nil {
-			t.Errorf("error during the Serve: error %v", err)
-		}
-	}()
+		go func() {
+			err := c.Serve()
+			if err != nil {
+				t.Errorf("error during the Serve: error %v", err)
+			}
+		}()
 
 
-	time.Sleep(time.Millisecond * 100)
+		time.Sleep(time.Millisecond * 100)
 
-	lastPID := getPID(s)
+		lastPID := getPID(s)
 
-	req, err := http.NewRequest("GET", "http://localhost:7030", nil)
-	assert.NoError(t, err)
+		req, err := http.NewRequest("GET", "http://localhost:7030", nil)
+		if err != nil {
+			return err
+		}
+
+		r, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
 
-	r, err := http.DefaultClient.Do(req)
-	assert.NoError(t, err)
 
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			return err
+		}
+		assert.Equal(t, lastPID, string(b))
 
-	b, err := ioutil.ReadAll(r.Body)
-	assert.NoError(t, err)
-	assert.Equal(t, lastPID, string(b))
+		<-captured
 
-	<-captured
+		// clean state
+		req, err = http.NewRequest("GET", "http://localhost:7030?new", nil)
+		if err != nil {
+			return err
+		}
+
+		_, err = http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
 
-	// clean state
-	req, err = http.NewRequest("GET", "http://localhost:7030?new", nil)
-	assert.NoError(t, err)
+		assert.NotEqual(t, lastPID, getPID(s))
 
-	_, err = http.DefaultClient.Do(req)
-	assert.NoError(t, err)
+		c.Stop()
 
-	assert.NotEqual(t, lastPID, getPID(s))
+		err2 := r.Body.Close()
+		if err2 != nil {
+			t.Errorf("error during the body closing: error %v", err2)
+		}
 
-	c.Stop()
+		return nil
+	}, bkoff)
 
-	err2 := r.Body.Close()
-	if err2 != nil {
-		t.Errorf("error during the body closing: error %v", err2)
+	if err != nil {
+		t.Fatal(err)
 	}
+
 }
 
 func Test_Service_ListenerPlusIdleTTL(t *testing.T) {
-	logger, _ := test.NewNullLogger()
-	logger.SetLevel(logrus.DebugLevel)
+	bkoff := backoff.NewExponentialBackOff()
+	bkoff.MaxElapsedTime = time.Second * 15
+
+	err := backoff.Retry(func() error {
+		logger, _ := test.NewNullLogger()
+		logger.SetLevel(logrus.DebugLevel)
 
-	c := service.NewContainer(logger)
-	c.Register(rrhttp.ID, &rrhttp.Service{})
-	c.Register(ID, &Service{})
+		c := service.NewContainer(logger)
+		c.Register(rrhttp.ID, &rrhttp.Service{})
+		c.Register(ID, &Service{})
 
-	assert.NoError(t, c.Init(&testCfg{
-		httpCfg: `{
+		err := c.Init(&testCfg{
+			httpCfg: `{
 			"address": ":7031",
 			"workers":{
 				"command": "php ../../tests/http/client.php pid pipes",
 				"pool": {"numWorkers": 1}
 			}
 	}`,
-		limitCfg: `{
+			limitCfg: `{
 			"services": {
 				"http": {
 					"idleTtl": 1
 				}
 			}
 		}`,
-	}))
-
-	s, _ := c.Get(rrhttp.ID)
-	assert.NotNil(t, s)
-
-	l, _ := c.Get(ID)
-	captured := make(chan interface{})
-	l.(*Service).AddListener(func(event int, ctx interface{}) {
-		if event == EventIdleTTL {
-			close(captured)
-		}
-	})
-
-	go func() {
-		err := c.Serve()
+		})
 		if err != nil {
-			t.Errorf("error during the Serve: error %v", err)
+			return err
 		}
-	}()
 
+		s, _ := c.Get(rrhttp.ID)
+		assert.NotNil(t, s)
 
-	time.Sleep(time.Millisecond * 100)
+		l, _ := c.Get(ID)
+		captured := make(chan interface{})
+		l.(*Service).AddListener(func(event int, ctx interface{}) {
+			if event == EventIdleTTL {
+				close(captured)
+			}
+		})
 
-	lastPID := getPID(s)
+		go func() {
+			err := c.Serve()
+			if err != nil {
+				t.Errorf("error during the Serve: error %v", err)
+			}
+		}()
 
-	req, err := http.NewRequest("GET", "http://localhost:7031", nil)
-	assert.NoError(t, err)
 
-	r, err := http.DefaultClient.Do(req)
-	assert.NoError(t, err)
+		time.Sleep(time.Millisecond * 100)
 
-	b, err := ioutil.ReadAll(r.Body)
-	assert.NoError(t, err)
+		lastPID := getPID(s)
 
-	assert.NoError(t, err)
-	assert.Equal(t, lastPID, string(b))
+		req, err := http.NewRequest("GET", "http://localhost:7031", nil)
+		if err != nil {
+			return err
+		}
+
+		r, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
 
-	<-captured
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			return err
+		}
+		assert.Equal(t, lastPID, string(b))
 
-	// clean state
-	req, err = http.NewRequest("GET", "http://localhost:7031?new", nil)
-	assert.NoError(t, err)
+		<-captured
 
-	_, err = http.DefaultClient.Do(req)
-	assert.NoError(t, err)
+		// clean state
+		req, err = http.NewRequest("GET", "http://localhost:7031?new", nil)
+		if err != nil {
+			return err
+		}
+
+		_, err = http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
 
-	assert.NotEqual(t, lastPID, getPID(s))
+		assert.NotEqual(t, lastPID, getPID(s))
 
-	c.Stop()
-	err2 := r.Body.Close()
-	if err2 != nil {
-		t.Errorf("error during the body closing: error %v", err2)
+		c.Stop()
+		err2 := r.Body.Close()
+		if err2 != nil {
+			t.Errorf("error during the body closing: error %v", err2)
+		}
+		return nil
+	}, bkoff)
+	if err != nil {
+		t.Fatal(err)
 	}
 }
 
 func Test_Service_Listener_MaxExecTTL(t *testing.T) {
-	logger, _ := test.NewNullLogger()
-	logger.SetLevel(logrus.DebugLevel)
+	bkoff := backoff.NewExponentialBackOff()
+	bkoff.MaxElapsedTime = time.Second * 15
+
+	err := backoff.Retry(func() error {
+
+		logger, _ := test.NewNullLogger()
+		logger.SetLevel(logrus.DebugLevel)
 
-	c := service.NewContainer(logger)
-	c.Register(rrhttp.ID, &rrhttp.Service{})
-	c.Register(ID, &Service{})
+		c := service.NewContainer(logger)
+		c.Register(rrhttp.ID, &rrhttp.Service{})
+		c.Register(ID, &Service{})
 
-	assert.NoError(t, c.Init(&testCfg{
-		httpCfg: `{
+		err := c.Init(&testCfg{
+			httpCfg: `{
 			"address": ":7032",
 			"workers":{
 				"command": "php ../../tests/http/client.php stuck pipes",
 				"pool": {"numWorkers": 1}
 			}
 	}`,
-		limitCfg: `{
+			limitCfg: `{
 			"services": {
 				"http": {
 					"execTTL": 1
 				}
 			}
 		}`,
-	}))
+		})
+		if err != nil {
+			return err
+		}
 
-	s, _ := c.Get(rrhttp.ID)
-	assert.NotNil(t, s)
+		s, _ := c.Get(rrhttp.ID)
+		assert.NotNil(t, s)
 
-	l, _ := c.Get(ID)
-	captured := make(chan interface{})
-	l.(*Service).AddListener(func(event int, ctx interface{}) {
-		if event == EventExecTTL {
-			close(captured)
-		}
-	})
+		l, _ := c.Get(ID)
+		captured := make(chan interface{})
+		l.(*Service).AddListener(func(event int, ctx interface{}) {
+			if event == EventExecTTL {
+				close(captured)
+			}
+		})
+
+		go func() {
+			err := c.Serve()
+			if err != nil {
+				t.Errorf("error during the Serve: error %v", err)
+			}
+		}()
+
+		time.Sleep(time.Millisecond * 100)
 
-	go func() {
-		err := c.Serve()
+		req, err := http.NewRequest("GET", "http://localhost:7032", nil)
 		if err != nil {
-			t.Errorf("error during the Serve: error %v", err)
+			return err
 		}
-	}()
 
-	time.Sleep(time.Millisecond * 100)
-
-	req, err := http.NewRequest("GET", "http://localhost:7032", nil)
-	assert.NoError(t, err)
+		r, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
+		assert.Equal(t, 500, r.StatusCode)
 
-	r, err := http.DefaultClient.Do(req)
-	assert.NoError(t, err)
-	assert.Equal(t, 500, r.StatusCode)
+		<-captured
 
-	<-captured
+		c.Stop()
+		return nil
+	}, bkoff)
 
-	c.Stop()
+	if err != nil {
+		t.Fatal(err)
+	}
 }
 
 func Test_Service_Listener_MaxMemoryUsage(t *testing.T) {
-	logger, _ := test.NewNullLogger()
-	logger.SetLevel(logrus.DebugLevel)
+	bkoff := backoff.NewExponentialBackOff()
+	bkoff.MaxElapsedTime = time.Second * 15
+
+	err := backoff.Retry(func() error {
+		logger, _ := test.NewNullLogger()
+		logger.SetLevel(logrus.DebugLevel)
 
-	c := service.NewContainer(logger)
-	c.Register(rrhttp.ID, &rrhttp.Service{})
-	c.Register(ID, &Service{})
+		c := service.NewContainer(logger)
+		c.Register(rrhttp.ID, &rrhttp.Service{})
+		c.Register(ID, &Service{})
 
-	assert.NoError(t, c.Init(&testCfg{
-		httpCfg: `{
+		err := c.Init(&testCfg{
+			httpCfg: `{
 			"address": ":7033",
 			"workers":{
 				"command": "php ../../tests/http/client.php memleak pipes",
 				"pool": {"numWorkers": 1}
 			}
 	}`,
-		limitCfg: `{
+			limitCfg: `{
 			"services": {
 				"http": {
 					"maxMemory": 10
 				}
 			}
 		}`,
-	}))
-
-	s, _ := c.Get(rrhttp.ID)
-	assert.NotNil(t, s)
-
-	l, _ := c.Get(ID)
-	captured := make(chan interface{})
-	once := false
-	l.(*Service).AddListener(func(event int, ctx interface{}) {
-		if event == EventMaxMemory && !once {
-			close(captured)
-			once = true
-		}
-	})
-
-	go func() {
-		err := c.Serve()
+		})
 		if err != nil {
-			t.Errorf("error during the Serve: error %v", err)
+			return err
 		}
-	}()
-
-	time.Sleep(time.Millisecond * 100)
 
-	lastPID := getPID(s)
+		s, _ := c.Get(rrhttp.ID)
+		assert.NotNil(t, s)
 
-	req, err := http.NewRequest("GET", "http://localhost:7033", nil)
-	assert.NoError(t, err)
+		l, _ := c.Get(ID)
+		captured := make(chan interface{})
+		once := false
+		l.(*Service).AddListener(func(event int, ctx interface{}) {
+			if event == EventMaxMemory && !once {
+				close(captured)
+				once = true
+			}
+		})
 
-	for {
-		select {
-		case <-captured:
-			_, err := http.DefaultClient.Do(req)
+		go func() {
+			err := c.Serve()
 			if err != nil {
-				c.Stop()
-				t.Errorf("error during sending the http request: error %v", err)
+				t.Errorf("error during the Serve: error %v", err)
 			}
-			assert.NotEqual(t, lastPID, getPID(s))
-			c.Stop()
-			return
-		default:
-			_, err := http.DefaultClient.Do(req)
-			if err != nil {
+		}()
+
+		time.Sleep(time.Millisecond * 100)
+
+		lastPID := getPID(s)
+
+		req, err := http.NewRequest("GET", "http://localhost:7033", nil)
+		if err != nil {
+			return err
+		}
+
+		for {
+			select {
+			case <-captured:
+				_, err := http.DefaultClient.Do(req)
+				if err != nil {
+					c.Stop()
+					t.Errorf("error during sending the http request: error %v", err)
+				}
+				assert.NotEqual(t, lastPID, getPID(s))
+				c.Stop()
+				return nil
+			default:
+				_, err := http.DefaultClient.Do(req)
+				if err != nil {
+					c.Stop()
+					t.Errorf("error during sending the http request: error %v", err)
+				}
 				c.Stop()
-				t.Errorf("error during sending the http request: error %v", err)
+				return nil
 			}
-			c.Stop()
-			return
 		}
+	}, bkoff)
+
+	if err != nil {
+		t.Fatal(err)
 	}
+
 }
 func getPID(s interface{}) string {
 	if len(s.(*rrhttp.Service).Server().Workers()) > 0 {