Merge pull request #32 from spiral/feature/env-manager

Feature/env manager

18 files changed, 275 insertions, 38 deletions

diff --git a/.travis.yml b/.travis.yml
index 4f7741f6..8fd4f66d 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -25,6 +25,7 @@ install:
 script:
   - go test -race -v -coverprofile=lib.txt -covermode=atomic
   - go test ./service -race -v -coverprofile=service.txt -covermode=atomic
+  - go test ./service/env -race -v -coverprofile=env.txt -covermode=atomic
   - go test ./service/rpc -race -v -coverprofile=rpc.txt -covermode=atomic
   - go test ./service/http -race -v -coverprofile=http.txt -covermode=atomic
   - go test ./service/static -race -v -coverprofile=static.txt -covermode=atomic
@@ -32,6 +33,7 @@ script:
 after_success:
   - bash <(curl -s https://codecov.io/bash) -f lib.txt
   - bash <(curl -s https://codecov.io/bash) -f service.txt
+  - bash <(curl -s https://codecov.io/bash) -f env.txt
   - bash <(curl -s https://codecov.io/bash) -f rpc.txt
   - bash <(curl -s https://codecov.io/bash) -f http.txt
   - bash <(curl -s https://codecov.io/bash) -f static.txt
\ No newline at end of file
diff --git a/Makefile b/Makefile
index 0ee41909..bcac2b29 100644
--- a/Makefile
+++ b/Makefile
@@ -11,6 +11,7 @@ uninstall:
 test:
 	go test -v -race -cover
 	go test -v -race -cover ./service
+	go test -v -race -cover ./service/env
 	go test -v -race -cover ./service/rpc
 	go test -v -race -cover ./service/http
 	go test -v -race -cover ./service/static
diff --git a/README.md b/README.md
index a1c7bfd9..d4d79f7c 100644
--- a/README.md
+++ b/README.md
@@ -55,6 +55,10 @@ Using RoadRunner:
 In order to use RoadRunner you only have to place a `.rr.yaml` config file in the root of your PHP project:
 
 ```yaml
+# defines environment variables for all underlying php processes
+env:
+  key: value
+
 # rpc bus allows php application and external clients to talk to rr services.
 rpc:
   # enable rpc server
diff --git a/cmd/rr/.rr.yaml b/cmd/rr/.rr.yaml
index 5ea6b345..401a42da 100644
--- a/cmd/rr/.rr.yaml
+++ b/cmd/rr/.rr.yaml
@@ -1,3 +1,7 @@
+# defines environment variables for all underlying php processes
+env:
+  key: value
+
 # rpc bus allows php application and external clients to talk to rr services.
 rpc:
   # enable rpc server
diff --git a/cmd/rr/cmd/root.go b/cmd/rr/cmd/root.go
index 79f43398..595395c0 100644
--- a/cmd/rr/cmd/root.go
+++ b/cmd/rr/cmd/root.go
@@ -24,11 +24,11 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"github.com/spiral/roadrunner/cmd/rr/debug"
 	"github.com/spiral/roadrunner/cmd/rr/utils"
 	"github.com/spiral/roadrunner/service"
-	"os"
 	"github.com/spiral/roadrunner/service/http"
-	"github.com/spiral/roadrunner/cmd/rr/debug"
+	"os"
 )
 
 // Service bus for all the commands.
diff --git a/cmd/rr/main.go b/cmd/rr/main.go
index 2a16b195..170d58bb 100644
--- a/cmd/rr/main.go
+++ b/cmd/rr/main.go
@@ -23,8 +23,6 @@
 package main
 
 import (
-	"github.com/sirupsen/logrus"
-	
 	rr "github.com/spiral/roadrunner/cmd/rr/cmd"
 
 	// services (plugins)
@@ -33,15 +31,19 @@ import (
 	"github.com/spiral/roadrunner/service/static"
 
 	// additional command handlers
+	"github.com/sirupsen/logrus"
 	_ "github.com/spiral/roadrunner/cmd/rr/http"
+	"github.com/spiral/roadrunner/service/env"
 )
 
 func main() {
+	rr.Logger.Formatter = &logrus.TextFormatter{ForceColors: true}
+
 	rr.Container.Register(rpc.ID, &rpc.Service{})
+	rr.Container.Register(env.ID, env.NewService(rr.Version))
 	rr.Container.Register(http.ID, &http.Service{})
 	rr.Container.Register(static.ID, &static.Service{})
 
 	// you can register additional commands using cmd.CLI
-        rr.Logger.Formatter = &logrus.TextFormatter{ForceColors: true}
 	rr.Execute()
 }
diff --git a/php-src/tests/http/env.php b/php-src/tests/http/env.php
new file mode 100644
index 00000000..1e29926f
--- /dev/null
+++ b/php-src/tests/http/env.php
@@ -0,0 +1,10 @@
+<?php
+
+use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\ServerRequestInterface;
+
+function handleRequest(ServerRequestInterface $req, ResponseInterface $resp): ResponseInterface
+{
+    $resp->getBody()->write($_SERVER['ENV_KEY']);
+    return $resp;
+}
\ No newline at end of file
diff --git a/server_config.go b/server_config.go
index b927c8c6..88d15e1f 100644
--- a/server_config.go
+++ b/server_config.go
@@ -2,7 +2,9 @@ package roadrunner
 
 import (
 	"errors"
+	"fmt"
 	"net"
+	"os"
 	"os/exec"
 	"strings"
 	"syscall"
@@ -26,6 +28,9 @@ type ServerConfig struct {
 	// Pool defines worker pool configuration, number of workers, timeouts and etc. This config section might change
 	// while server is running.
 	Pool *Config
+
+	// values defines set of values to be passed to the command context.
+	env []string
 }
 
 // Differs returns true if configuration has changed but ignores pool or cmd changes.
@@ -33,11 +38,18 @@ func (cfg *ServerConfig) Differs(new *ServerConfig) bool {
 	return cfg.Relay != new.Relay || cfg.RelayTimeout != new.RelayTimeout
 }
 
+// SetEnv sets new environment variable. Value is automatically uppercase-d.
+func (cfg *ServerConfig) SetEnv(k, v string) {
+	cfg.env = append(cfg.env, fmt.Sprintf("%s=%s", strings.ToUpper(k), v))
+}
+
 // makeCommands returns new command provider based on configured options.
 func (cfg *ServerConfig) makeCommand() func() *exec.Cmd {
 	var cmd = strings.Split(cfg.Command, " ")
 	return func() *exec.Cmd {
-		return exec.Command(cmd[0], cmd[1:]...)
+		cmd := exec.Command(cmd[0], cmd[1:]...)
+		cmd.Env = append(os.Environ(), cfg.env...)
+		return cmd
 	}
 }
 
diff --git a/service/container.go b/service/container.go
index 436d2e5f..3450a18c 100644
--- a/service/container.go
+++ b/service/container.go
@@ -8,7 +8,7 @@ import (
 	"sync"
 )
 
-var noConfig = fmt.Errorf("no config has been provided")
+var errNoConfig = fmt.Errorf("no config has been provided")
 
 // InitMethod contains name of the method to be automatically invoked while service initialization. Must return
 // (bool, error). Container can be requested as well. Config can be requested in a form
@@ -132,7 +132,7 @@ func (c *container) Init(cfg Config) error {
 		// inject service dependencies
 		if ok, err := c.initService(e.svc, cfg.Get(e.name)); err != nil {
 			// soft error (skipping)
-			if err == noConfig {
+			if err == errNoConfig {
 				c.log.Warningf("[%s]: no config has been provided", e.name)
 				continue
 			}
@@ -253,7 +253,7 @@ func (c *container) resolveValues(s interface{}, m reflect.Method, cfg Config) (
 
 		case v.Implements(reflect.TypeOf((*HydrateConfig)(nil)).Elem()): // injectable config
 			if cfg == nil {
-				return nil, noConfig
+				return nil, errNoConfig
 			}
 
 			sc := reflect.New(v.Elem())
@@ -265,27 +265,18 @@ func (c *container) resolveValues(s interface{}, m reflect.Method, cfg Config) (
 
 		case v.Implements(reflect.TypeOf((*Config)(nil)).Elem()): // generic config section
 			if cfg == nil {
-				return nil, noConfig
+				return nil, errNoConfig
 			}
 
 			values = append(values, reflect.ValueOf(cfg))
 
 		default: // dependency on other service (resolution to nil if service can't be found)
-			found := false
-			for _, e := range c.services {
-				if !e.hasStatus(StatusOK) || !v.ConvertibleTo(reflect.ValueOf(e.svc).Type()) {
-					continue
-				}
-
-				found = true
-				values = append(values, reflect.ValueOf(e.svc))
-				break
+			value, err := c.resolveValue(v)
+			if err != nil {
+				return nil, err
 			}
 
-			if !found {
-				// placeholder (make sure to check inside the method)
-				values = append(values, reflect.New(v).Elem())
-			}
+			values = append(values, value)
 		}
 	}
 
@@ -308,3 +299,35 @@ func (c *container) verifySignature(m reflect.Method) error {
 
 	return nil
 }
+
+func (c *container) resolveValue(v reflect.Type) (reflect.Value, error) {
+	value := reflect.Value{}
+	for _, e := range c.services {
+		if !e.hasStatus(StatusOK) {
+			continue
+		}
+
+		if v.Kind() == reflect.Interface && reflect.TypeOf(e.svc).Implements(v) {
+			if value.IsValid() {
+				return value, fmt.Errorf("disambiguous dependency `%s`", v)
+			}
+
+			value = reflect.ValueOf(e.svc)
+		}
+
+		if v.ConvertibleTo(reflect.ValueOf(e.svc).Type()) {
+			if value.IsValid() {
+				return value, fmt.Errorf("disambiguous dependency `%s`", v)
+			}
+
+			value = reflect.ValueOf(e.svc)
+		}
+	}
+
+	if !value.IsValid() {
+		// placeholder (make sure to check inside the method)
+		value = reflect.New(v).Elem()
+	}
+
+	return value, nil
+}
diff --git a/service/env/config.go b/service/env/config.go
new file mode 100644
index 00000000..d0ba686b
--- /dev/null
+++ b/service/env/config.go
@@ -0,0 +1,16 @@
+package env
+
+import (
+	"github.com/spiral/roadrunner/service"
+)
+
+// Config defines set of env values for RR workers.
+type Config struct {
+	// values to set as worker _ENV.
+	Values map[string]string
+}
+
+// Hydrate must populate Config values using given Config source. Must return error if Config is not valid.
+func (c *Config) Hydrate(cfg service.Config) error {
+	return cfg.Unmarshal(&c.Values)
+}
diff --git a/service/env/config_test.go b/service/env/config_test.go
new file mode 100644
index 00000000..3ae2afbc
--- /dev/null
+++ b/service/env/config_test.go
@@ -0,0 +1,29 @@
+package env
+
+import (
+	"encoding/json"
+	"github.com/spiral/roadrunner/service"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+type mockCfg struct{ cfg string }
+
+func (cfg *mockCfg) Get(name string) service.Config  { return nil }
+func (cfg *mockCfg) Unmarshal(out interface{}) error { return json.Unmarshal([]byte(cfg.cfg), out) }
+
+func Test_Config_Hydrate(t *testing.T) {
+	cfg := &mockCfg{`{"key":"value"}`}
+	c := &Config{}
+
+	assert.NoError(t, c.Hydrate(cfg))
+	assert.Len(t, c.Values, 1)
+}
+
+func Test_Config_Hydrate_Empty(t *testing.T) {
+	cfg := &mockCfg{`{}`}
+	c := &Config{}
+
+	assert.NoError(t, c.Hydrate(cfg))
+	assert.Len(t, c.Values, 0)
+}
diff --git a/service/env/provider.go b/service/env/provider.go
new file mode 100644
index 00000000..2918f18c
--- /dev/null
+++ b/service/env/provider.go
@@ -0,0 +1,8 @@
+package env
+
+// Provider aggregates list of environment variables. This interface can be used in custom implementation to drive
+// values from external sources.
+type Provider interface {
+	// GetEnv must return list of env variables.
+	GetEnv() (map[string]string, error)
+}
diff --git a/service/env/service.go b/service/env/service.go
new file mode 100644
index 00000000..9fb110c3
--- /dev/null
+++ b/service/env/service.go
@@ -0,0 +1,30 @@
+package env
+
+// ID contains default svc name.
+const ID = "env"
+
+// Service provides ability to map _ENV values from config file.
+type Service struct {
+	// values is default set of values.
+	values map[string]string
+}
+
+// NewService creates new env service instance for given rr version.
+func NewService(version string) *Service {
+	return &Service{values: map[string]string{"rr": version}}
+}
+
+// Init must return configure svc and return true if svc hasStatus enabled. Must return error in case of
+// misconfiguration. Services must not be used without proper configuration pushed first.
+func (s *Service) Init(cfg *Config) (bool, error) {
+	for k, v := range cfg.Values {
+		s.values[k] = v
+	}
+
+	return true, nil
+}
+
+// GetEnv must return list of env variables.
+func (s *Service) GetEnv() (map[string]string, error) {
+	return s.values, nil
+}
diff --git a/service/env/service_test.go b/service/env/service_test.go
new file mode 100644
index 00000000..f25e56c7
--- /dev/null
+++ b/service/env/service_test.go
@@ -0,0 +1,24 @@
+package env
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func Test_NewService(t *testing.T) {
+	s := NewService("version")
+	assert.Len(t, s.values, 1)
+}
+
+func Test_Extend(t *testing.T) {
+	s := NewService("version")
+
+	s.Init(&Config{Values: map[string]string{"key": "value"}})
+	assert.Len(t, s.values, 2)
+
+	values, err := s.GetEnv()
+	assert.NoError(t, err)
+	assert.Len(t, values, 2)
+	assert.Equal(t, "version", values["rr"])
+	assert.Equal(t, "value", values["key"])
+}
diff --git a/service/http/attributes/attributes.go b/service/http/attributes/attributes.go
index 94d0e9c1..77d6ea69 100644
--- a/service/http/attributes/attributes.go
+++ b/service/http/attributes/attributes.go
@@ -6,7 +6,9 @@ import (
 	"net/http"
 )
 
-const contextKey = "psr:attributes"
+type attrKey int
+
+const contextKey attrKey = iota
 
 type attrs map[string]interface{}
 
@@ -41,7 +43,7 @@ func All(r *http.Request) map[string]interface{} {
 	return v.(attrs)
 }
 
-// get gets the value from request context. It replaces any existing
+// Get gets the value from request context. It replaces any existing
 // values.
 func Get(r *http.Request, key string) interface{} {
 	v := r.Context().Value(contextKey)
@@ -52,7 +54,7 @@ func Get(r *http.Request, key string) interface{} {
 	return v.(attrs).get(key)
 }
 
-// set sets the key to value. It replaces any existing
+// Set sets the key to value. It replaces any existing
 // values. Context specific.
 func Set(r *http.Request, key string, value interface{}) error {
 	v := r.Context().Value(contextKey)
diff --git a/service/http/request.go b/service/http/request.go
index 6d5cc126..531a1efd 100644
--- a/service/http/request.go
+++ b/service/http/request.go
@@ -126,7 +126,7 @@ func (r *Request) Close() {
 	r.Uploads.Clear()
 }
 
-// Payload request marshaled RoadRunner payload based on PSR7 data. Default encode method is JSON. Make sure to open
+// Payload request marshaled RoadRunner payload based on PSR7 data. values encode method is JSON. Make sure to open
 // files prior to calling this method.
 func (r *Request) Payload() (p *roadrunner.Payload, err error) {
 	p = &roadrunner.Payload{}
diff --git a/service/http/service.go b/service/http/service.go
index f7fdf2ab..9f62f5af 100644
--- a/service/http/service.go
+++ b/service/http/service.go
@@ -3,6 +3,7 @@ package http
 import (
 	"context"
 	"github.com/spiral/roadrunner"
+	"github.com/spiral/roadrunner/service/env"
 	"github.com/spiral/roadrunner/service/http/attributes"
 	"github.com/spiral/roadrunner/service/rpc"
 	"net/http"
@@ -18,10 +19,10 @@ type middleware func(f http.HandlerFunc) http.HandlerFunc
 
 // Service manages rr, http servers.
 type Service struct {
-	cfg  *Config
-	lsns []func(event int, ctx interface{})
-	mdws []middleware
-
+	cfg      *Config
+	env      env.Provider
+	lsns     []func(event int, ctx interface{})
+	mdws     []middleware
 	mu       sync.Mutex
 	rr       *roadrunner.Server
 	stopping int32
@@ -41,12 +42,13 @@ func (s *Service) AddListener(l func(event int, ctx interface{})) {
 
 // Init must return configure svc and return true if svc hasStatus enabled. Must return error in case of
 // misconfiguration. Services must not be used without proper configuration pushed first.
-func (s *Service) Init(cfg *Config, r *rpc.Service) (bool, error) {
+func (s *Service) Init(cfg *Config, r *rpc.Service, e env.Provider) (bool, error) {
 	if !cfg.Enable {
 		return false, nil
 	}
 
 	s.cfg = cfg
+	s.env = e
 	if r != nil {
 		r.Register(ID, &rpcServer{s})
 	}
@@ -57,6 +59,18 @@ func (s *Service) Init(cfg *Config, r *rpc.Service) (bool, error) {
 // Serve serves the svc.
 func (s *Service) Serve() error {
 	s.mu.Lock()
+
+	if s.env != nil {
+		values, err := s.env.GetEnv()
+		if err != nil {
+			return err
+		}
+
+		for k, v := range values {
+			s.cfg.Workers.SetEnv(k, v)
+		}
+	}
+
 	rr := roadrunner.NewServer(s.cfg.Workers)
 
 	s.rr = rr
@@ -116,11 +130,7 @@ func (s *Service) listener(event int, ctx interface{}) {
 	}
 
 	if event == roadrunner.EventServerFailure {
-		if atomic.LoadInt32(&s.stopping) != 0 {
-			// attempting rr server restart
-			if err := s.rr.Start(); err != nil {
-				s.Stop()
-			}
-		}
+		// underlying rr server is dead
+		s.Stop()
 	}
 }
diff --git a/service/http/service_test.go b/service/http/service_test.go
index b442ae51..29c263b9 100644
--- a/service/http/service_test.go
+++ b/service/http/service_test.go
@@ -13,11 +13,13 @@ import (
 	"os"
 	"testing"
 	"time"
+	"github.com/spiral/roadrunner/service/env"
 )
 
 type testCfg struct {
 	httpCfg string
 	rpcCfg  string
+	envCfg  string
 	target  string
 }
 
@@ -29,6 +31,11 @@ func (cfg *testCfg) Get(name string) service.Config {
 	if name == rpc.ID {
 		return &testCfg{target: cfg.rpcCfg}
 	}
+
+	if name == env.ID {
+		return &testCfg{target: cfg.envCfg}
+	}
+
 	return nil
 }
 func (cfg *testCfg) Unmarshal(out interface{}) error {
@@ -163,6 +170,59 @@ func Test_Service_Echo(t *testing.T) {
 	assert.Equal(t, "WORLD", string(b))
 }
 
+func Test_Service_Env(t *testing.T) {
+	logger, _ := test.NewNullLogger()
+	logger.SetLevel(logrus.DebugLevel)
+
+	c := service.NewContainer(logger)
+	c.Register(env.ID, env.NewService("test"))
+	c.Register(ID, &Service{})
+
+	assert.NoError(t, c.Init(&testCfg{httpCfg: `{
+			"enable": true,
+			"address": ":6029",
+			"maxRequest": 1024,
+			"uploads": {
+				"dir": ` + tmpDir() + `,
+				"forbid": []
+			},
+			"workers":{
+				"command": "php ../../php-src/tests/http/client.php env pipes",
+				"relay": "pipes",
+				"pool": {
+					"numWorkers": 1, 
+					"allocateTimeout": 10000000,
+					"destroyTimeout": 10000000 
+				}
+			}
+	}`, envCfg: `{"env_key":"ENV_VALUE"}`}))
+
+	s, st := c.Get(ID)
+	assert.NotNil(t, s)
+	assert.Equal(t, service.StatusOK, st)
+
+	// should do nothing
+	s.(*Service).Stop()
+
+	go func() { c.Serve() }()
+	time.Sleep(time.Millisecond * 100)
+	defer c.Stop()
+
+	req, err := http.NewRequest("GET", "http://localhost:6029", nil)
+	assert.NoError(t, err)
+
+	r, err := http.DefaultClient.Do(req)
+	assert.NoError(t, err)
+	defer r.Body.Close()
+
+	b, err := ioutil.ReadAll(r.Body)
+	assert.NoError(t, err)
+
+	assert.NoError(t, err)
+	assert.Equal(t, 200, r.StatusCode)
+	assert.Equal(t, "ENV_VALUE", string(b))
+}
+
 func Test_Service_ErrorEcho(t *testing.T) {
 	logger, _ := test.NewNullLogger()
 	logger.SetLevel(logrus.DebugLevel)