forward request to the upstream (#6)

* feat: parse upstream rules from config

* feat: added forwarding request to the upstream by certain conditions

* feat: added tests for forwarding on proxy server

* fix: close upstream when the client sent EOF

* fix: removed redundant comment

* fix: send headers from the upstream to client

* perf: cleaned up code a bit

Author: Semior001

_example/mock.yaml

@@ -80,3 +80,11 @@ rules:
       metadata:
         header: { X-Request-Id: "123"   }
         trailer: { Powered-By:   "groxy" }
+
+    # The next rule will forward the request to the specified upstream.
+  - match: { uri: "com.github.Semior001.groxy.example.mock.Upstream/Get" }
+    forward:
+      upstream: example-1
+      header:
+        X-Request-Id: "123"
+

pkg/discovery/discovery.go

@@ -24,11 +24,20 @@ type Provider interface {
 	// It returns the name of the provider to update the routing rules.
 	Events(ctx context.Context) <-chan string
 
-	// Rules returns the routing rules.
-	Rules(ctx context.Context) ([]*Rule, error)
+	// State returns the current state of the provider.
+	State(ctx context.Context) (*State, error)
+}
+
+// State contains the state of the provider.
+type State struct {
+	// Name is the name of the provider.
+	Name string
+
+	// Rules contains the routing rules.
+	Rules []*Rule
 
-	// Upstreams returns the upstreams.
-	Upstreams(ctx context.Context) ([]Upstream, error)
+	// Upstreams contains the upstreams.
+	Upstreams []Upstream
 }
 
 // Mock contains the details of how the handler should reply to the downstream.
@@ -50,6 +59,16 @@ type Rule struct {
 
 	// Mock defines the details of how the handler should reply to the downstream.
 	Mock *Mock
+
+	// Forward specifies the upstream to forward the request.
+	Forward *Forward
+}
+
+// Forward specifies the upstream to forward the request and the parameters
+// to invoke the upstream.
+type Forward struct {
+	Upstream Upstream
+	Header   metadata.MD
 }
 
 // String returns the name of the rule.
@@ -107,15 +126,15 @@ type Upstream interface {
 	grpc.ClientConnInterface
 }
 
-// NamedClosableClientConn is a named closable client connection.
-type NamedClosableClientConn struct {
+// ClientConn is a named closable client connection.
+type ClientConn struct {
 	ConnName        string
 	ServeReflection bool
 	*grpc.ClientConn
 }
 
 // Name returns the name of the connection.
-func (n NamedClosableClientConn) Name() string { return n.ConnName }
+func (n ClientConn) Name() string { return n.ConnName }
 
 // Reflection returns true if the connection serves reflection.
-func (n NamedClosableClientConn) Reflection() bool { return n.ServeReflection }
+func (n ClientConn) Reflection() bool { return n.ServeReflection }

pkg/discovery/fileprovider/config.go

@@ -22,7 +22,14 @@ type Rule struct {
 		Header map[string]string `yaml:"header"`
 		Body   *string           `yaml:"body"`
 	} `yaml:"match"`
-	Respond Respond `yaml:"respond"`
+	Respond *Respond `yaml:"respond,omitempty"`
+	Forward *Forward `yaml:"forward,omitempty"`
+}
+
+// Forward specifies how the service should forward the request.
+type Forward struct {
+	Upstream string            `yaml:"upstream"`
+	Header   map[string]string `yaml:"header"`
 }
 
 // Respond specifies how the service should respond to the request.

pkg/discovery/fileprovider/file.go

@@ -21,6 +21,7 @@ import (
 	"google.golang.org/grpc/credentials"
 	"crypto/tls"
 	"sort"
+	"github.com/Semior001/groxy/pkg/grpcx"
 )
 
 // File discovers the changes in routing rules from a file.
@@ -90,8 +91,8 @@ func (d *File) Events(ctx context.Context) <-chan string {
 	return res
 }
 
-// Rules parses the file and returns the routing rules from it.
-func (d *File) Rules(context.Context) ([]*discovery.Rule, error) {
+// State parses the file and returns the current state of the provider.
+func (d *File) State(ctx context.Context) (*discovery.State, error) {
 	f, err := os.Open(d.FileName)
 	if err != nil {
 		return nil, fmt.Errorf("open file: %w", err)
@@ -100,70 +101,36 @@ func (d *File) Rules(context.Context) ([]*discovery.Rule, error) {
 	defer f.Close()
 
 	var cfg Config
-	if err = yaml.NewDecoder(f).Decode(&cfg); err != nil {
+	if err := yaml.NewDecoder(f).Decode(&cfg); err != nil {
 		return nil, fmt.Errorf("decode file: %w", err)
 	}
 
 	if cfg.Version != "1" {
 		return nil, fmt.Errorf("unsupported version: %s", cfg.Version)
 	}
 
-	parseRespond := func(r Respond) (result *discovery.Mock, err error) {
-		result = &discovery.Mock{}
-
-		if r.Metadata != nil {
-			result.Header = metadata.New(r.Metadata.Header)
-			result.Trailer = metadata.New(r.Metadata.Trailer)
-		}
-
-		switch {
-		case r.Status != nil && r.Body != nil:
-			return nil, fmt.Errorf("can't set both status and body in rule")
-		case r.Status != nil:
-			var code codes.Code
-			if err = code.UnmarshalJSON([]byte(fmt.Sprintf("%q", r.Status.Code))); err != nil {
-				return nil, fmt.Errorf("unmarshal status code: %w", err)
-			}
-			result.Status = status.New(code, r.Status.Message)
-		case r.Body != nil:
-			if result.Body, err = protodef.BuildMessage(*r.Body); err != nil {
-				return nil, fmt.Errorf("build respond message: %w", err)
-			}
-		default:
-			return nil, fmt.Errorf("empty response in rule")
-		}
-
-		return result, nil
+	upstreams, err := d.upstreams(ctx, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("get upstreams: %w", err)
 	}
 
-	parseRule := func(r Rule) (result discovery.Rule, err error) {
-		if r.Match.URI == "" {
-			return discovery.Rule{}, fmt.Errorf("empty URI in rule")
-		}
-
-		result.Name = r.Match.URI
-		if result.Match.URI, err = regexp.Compile(r.Match.URI); err != nil {
-			return discovery.Rule{}, fmt.Errorf("compile URI regexp: %w", err)
-		}
-
-		result.Match.IncomingMetadata = metadata.New(r.Match.Header)
-
-		if r.Match.Body != nil {
-			if result.Match.Message, err = protodef.BuildMessage(*r.Match.Body); err != nil {
-				return discovery.Rule{}, fmt.Errorf("build request matcher message: %w", err)
-			}
-		}
-
-		if result.Mock, err = parseRespond(r.Respond); err != nil {
-			return discovery.Rule{}, fmt.Errorf("parse respond: %w", err)
-		}
-
-		return result, nil
+	rules, err := d.rules(cfg, upstreams)
+	if err != nil {
+		return nil, fmt.Errorf("get rules: %w", err)
 	}
 
+	return &discovery.State{
+		Name:      d.Name(),
+		Rules:     rules,
+		Upstreams: upstreams,
+	}, nil
+}
+
+// Rules parses the file and returns the routing rules from it.
+func (d *File) rules(cfg Config, upstreams []discovery.Upstream) ([]*discovery.Rule, error) {
 	rules := make([]*discovery.Rule, 0, len(cfg.Rules)+1)
 	for idx, r := range cfg.Rules {
-		rule, err := parseRule(r)
+		rule, err := parseRule(r, upstreams)
 		if err != nil {
 			return nil, fmt.Errorf("parse rule #%d: %w", idx, err)
 		}
@@ -172,37 +139,22 @@ func (d *File) Rules(context.Context) ([]*discovery.Rule, error) {
 	}
 
 	if cfg.NotMatched != nil {
+		mock, err := parseRespond(cfg.NotMatched)
+		if err != nil {
+			return nil, fmt.Errorf("parse respond: %w", err)
+		}
 		rule := discovery.Rule{
 			Name:  "not matched",
 			Match: discovery.RequestMatcher{URI: regexp.MustCompile(".*")},
-		}
-		if rule.Mock, err = parseRespond(*cfg.NotMatched); err != nil {
-			return nil, fmt.Errorf("parse respond: %w", err)
+			Mock:  mock,
 		}
 		rules = append(rules, &rule)
 	}
 
 	return rules, nil
 }
 
-// Upstreams parses the file and returns the upstreams from it.
-func (d *File) Upstreams(ctx context.Context) ([]discovery.Upstream, error) {
-	f, err := os.Open(d.FileName)
-	if err != nil {
-		return nil, fmt.Errorf("open file: %w", err)
-	}
-
-	defer f.Close()
-
-	var cfg Config
-	if err = yaml.NewDecoder(f).Decode(&cfg); err != nil {
-		return nil, fmt.Errorf("decode file: %w", err)
-	}
-
-	if cfg.Version != "1" {
-		return nil, fmt.Errorf("unsupported version: %s", cfg.Version)
-	}
-
+func (d *File) upstreams(ctx context.Context, cfg Config) ([]discovery.Upstream, error) {
 	res := make([]discovery.Upstream, 0, len(cfg.Upstreams))
 	for name, u := range cfg.Upstreams {
 		cred := insecure.NewCredentials()
@@ -219,12 +171,15 @@ func (d *File) Upstreams(ctx context.Context) ([]discovery.Upstream, error) {
 			slog.String("address", u.Addr),
 			slog.Bool("tls", u.TLS))
 
-		cc, err := grpc.DialContext(ctx, u.Addr, grpc.WithTransportCredentials(cred))
+		cc, err := grpc.DialContext(ctx, u.Addr,
+			grpc.WithTransportCredentials(cred),
+			grpc.WithStreamInterceptor(grpcx.ClientLogInterceptor(slog.Default())),
+		)
 		if err != nil {
 			return nil, fmt.Errorf("dial upstream %q: %w", name, err)
 		}
 
-		res = append(res, discovery.NamedClosableClientConn{
+		res = append(res, discovery.ClientConn{
 			ConnName:        name,
 			ServeReflection: u.ServeReflection,
 			ClientConn:      cc,
@@ -253,3 +208,74 @@ func (d *File) getModifTime(ctx context.Context) (modif time.Time, ok bool) {
 
 	return fi.ModTime(), true
 }
+
+func parseRule(r Rule, upstreams []discovery.Upstream) (result discovery.Rule, err error) {
+	if r.Match.URI == "" {
+		return discovery.Rule{}, fmt.Errorf("empty URI in rule")
+	}
+
+	result.Name = r.Match.URI
+	if result.Match.URI, err = regexp.Compile(r.Match.URI); err != nil {
+		return discovery.Rule{}, fmt.Errorf("compile URI regexp: %w", err)
+	}
+
+	result.Match.IncomingMetadata = metadata.New(r.Match.Header)
+
+	if r.Match.Body != nil {
+		if result.Match.Message, err = protodef.BuildMessage(*r.Match.Body); err != nil {
+			return discovery.Rule{}, fmt.Errorf("build request matcher message: %w", err)
+		}
+	}
+
+	if r.Forward != nil {
+		result.Forward = &discovery.Forward{Header: metadata.New(r.Forward.Header)}
+		for _, up := range upstreams {
+			if up.Name() == r.Forward.Upstream {
+				result.Forward.Upstream = up
+				break
+			}
+		}
+	}
+
+	if result.Mock, err = parseRespond(r.Respond); err != nil {
+		return discovery.Rule{}, fmt.Errorf("parse respond: %w", err)
+	}
+
+	if result.Mock != nil && result.Forward != nil {
+		return discovery.Rule{}, fmt.Errorf("can't set both mock and forward in rule")
+	}
+
+	return result, nil
+}
+
+func parseRespond(r *Respond) (result *discovery.Mock, err error) {
+	if r == nil {
+		return nil, nil
+	}
+
+	result = &discovery.Mock{}
+
+	if r.Metadata != nil {
+		result.Header = metadata.New(r.Metadata.Header)
+		result.Trailer = metadata.New(r.Metadata.Trailer)
+	}
+
+	switch {
+	case r.Status != nil && r.Body != nil:
+		return nil, fmt.Errorf("can't set both status and body in rule")
+	case r.Status != nil:
+		var code codes.Code
+		if err = code.UnmarshalJSON([]byte(fmt.Sprintf("%q", r.Status.Code))); err != nil {
+			return nil, fmt.Errorf("unmarshal status code: %w", err)
+		}
+		result.Status = status.New(code, r.Status.Message)
+	case r.Body != nil:
+		if result.Body, err = protodef.BuildMessage(*r.Body); err != nil {
+			return nil, fmt.Errorf("build respond message: %w", err)
+		}
+	default:
+		return nil, fmt.Errorf("empty response in rule")
+	}
+
+	return result, nil
+}