feat: preliminary implementation of traffic routing

Author: saicaca

core/route/base/headers.go

@@ -0,0 +1,12 @@
+package base
+
+type Headers struct {
+	Request  *HeaderOperations
+	Response *HeaderOperations
+}
+
+type HeaderOperations struct {
+	Set    map[string]string
+	Add    map[string]string
+	Remove []string
+}

core/route/base/http_match_request.go

@@ -0,0 +1,61 @@
+package base
+
+import (
+	"net/url"
+	"strconv"
+)
+
+type HTTPMatchRequest struct {
+	Name        string
+	Headers     map[string]*StringMatch
+	Uri         *StringMatch
+	Scheme      *StringMatch
+	Authority   *StringMatch
+	Method      *StringMatch
+	Port        *int
+	QueryParams map[string]*StringMatch
+}
+
+// TODO
+func (h *HTTPMatchRequest) IsMatch(context *TrafficContext) bool {
+
+	for key, match := range h.Headers {
+		if v, ok := context.Headers[key]; ok && !match.IsMatch(v) {
+			return false
+		}
+	}
+
+	if h.Uri != nil && !h.Uri.IsMatch(context.Uri) {
+		return false
+	}
+
+	var parsedURL *url.URL
+	var err error
+
+	if h.Uri != nil || h.Scheme != nil || h.Authority != nil || h.Port != nil {
+		parsedURL, err = url.Parse(context.Uri)
+		if err != nil {
+			return false
+		}
+	}
+	if h.Uri != nil && !h.Uri.IsMatch(parsedURL.Path) {
+		return false
+	}
+	if h.Scheme != nil && !h.Scheme.IsMatch(parsedURL.Scheme) {
+		return false
+	}
+	if h.Authority != nil && !h.Authority.IsMatch(parsedURL.Host) {
+		return false
+	}
+	if h.Port != nil {
+		p, err := strconv.Atoi(parsedURL.Port())
+		if err != nil || *h.Port != p {
+			return false
+		}
+	}
+	if !h.Method.IsMatch(context.MethodName) {
+		return false
+	}
+
+	return true
+}

core/route/base/http_route.go

@@ -0,0 +1,16 @@
+package base
+
+type HTTPRoute struct {
+	Name  string
+	Match []*HTTPMatchRequest
+	Route []*HTTPRouteDestination
+}
+
+func (h *HTTPRoute) IsMatch(context *TrafficContext) bool {
+	for _, match := range h.Match {
+		if match.IsMatch(context) {
+			return true
+		}
+	}
+	return false
+}

core/route/base/http_route_destination.go

@@ -0,0 +1,20 @@
+package base
+
+import "fmt"
+
+type HTTPRouteDestination struct {
+	Weight      int
+	Destination *Destination
+	Headers     Headers // TODO modifies headers
+}
+
+func (H HTTPRouteDestination) String() string {
+	return fmt.Sprintf("{Weight: %v, Destination: %+v}\n", H.Weight, H.Destination)
+}
+
+type Destination struct {
+	Host     string
+	Subset   string
+	Port     uint32
+	Fallback *HTTPRouteDestination
+}

core/route/base/instance.go

@@ -0,0 +1,9 @@
+package base
+
+type Instance struct {
+	AppName        string
+	Host           string
+	Port           int
+	Metadata       map[string]string
+	TargetInstance interface{}
+}

core/route/base/string_match.go

@@ -0,0 +1,25 @@
+package base
+
+import "regexp"
+
+type StringMatch struct {
+	Exact  string
+	Prefix string
+	Regex  string
+}
+
+func (s *StringMatch) IsMatch(input string) bool {
+	if input == "" {
+		return false
+	}
+
+	if s.Exact != "" {
+		return input == s.Exact
+	} else if s.Prefix != "" {
+		return len(input) >= len(s.Prefix) && input[:len(s.Prefix)] == s.Prefix
+	} else if s.Regex != "" {
+		matched, _ := regexp.MatchString(s.Regex, input)
+		return matched
+	}
+	return true
+}

core/route/base/traffic_context.go

@@ -0,0 +1,14 @@
+package base
+
+type TrafficContext struct {
+	Path        string
+	Uri         string
+	ServiceName string
+	Group       string
+	Version     string
+	MethodName  string
+	ParamTypes  []string
+	Args        []interface{}
+	Headers     map[string]string
+	Baggage     map[string]string
+}

core/route/base/traffic_policy.go

@@ -0,0 +1,5 @@
+package base
+
+type TrafficPolicy struct {
+	LoadBalancer string
+}

core/route/base/traffic_router.go

@@ -0,0 +1,11 @@
+package base
+
+type TrafficRouter struct {
+	Host []string
+	Http []*HTTPRoute
+}
+
+type Fallback struct {
+	Host   string
+	Subset string
+}

core/route/base/virtual_workload.go

@@ -0,0 +1,12 @@
+package base
+
+type VirtualWorkload struct {
+	Host          string
+	trafficPolicy *TrafficPolicy
+	Subsets       []*Subset
+}
+
+type Subset struct {
+	Name   string
+	Labels map[string]string
+}

core/route/cluster_manager.go

@@ -0,0 +1,51 @@
+package route
+
+import (
+	"github.com/alibaba/sentinel-golang/core/route/base"
+	"github.com/pkg/errors"
+)
+
+type ClusterManager struct {
+	InstanceManager  InstanceManager
+	RouterFilterList []RouterFilter
+	LoadBalancer     LoadBalancer
+}
+
+func NewClusterManager(instanceManager InstanceManager, routerFilters []RouterFilter, loadBalancer LoadBalancer) *ClusterManager {
+	return &ClusterManager{
+		InstanceManager:  instanceManager,
+		RouterFilterList: routerFilters,
+		LoadBalancer:     loadBalancer,
+	}
+}
+
+func (m *ClusterManager) Route(context *base.TrafficContext) ([]*base.Instance, error) {
+	instances := m.InstanceManager.GetInstances()
+
+	var err error
+	for _, routerFilter := range m.RouterFilterList {
+		instances, err = routerFilter.Filter(instances, context)
+		if err != nil {
+			return nil, err
+		}
+	}
+	if len(instances) == 0 {
+		return nil, errors.New("no matching instances")
+	}
+	return instances, nil
+}
+
+func (m *ClusterManager) GetOne(context *base.TrafficContext) (*base.Instance, error) {
+	instances, err := m.Route(context)
+	if err != nil {
+		return nil, err
+	}
+	if m.LoadBalancer == nil {
+		return instances[0], nil
+	}
+	instance, err := m.LoadBalancer.Select(instances, context)
+	if err != nil {
+		return nil, err
+	}
+	return instance, nil
+}

core/route/instance_manager.go

@@ -0,0 +1,24 @@
+package route
+
+import "github.com/alibaba/sentinel-golang/core/route/base"
+
+type InstanceManager interface {
+	StoreInstances(instances []*base.Instance)
+	GetInstances() []*base.Instance
+}
+
+type BasicInstanceManager struct {
+	Instances []*base.Instance
+}
+
+func NewBasicInstanceManager() *BasicInstanceManager {
+	return &BasicInstanceManager{}
+}
+
+func (b *BasicInstanceManager) StoreInstances(instances []*base.Instance) {
+	b.Instances = instances
+}
+
+func (b *BasicInstanceManager) GetInstances() []*base.Instance {
+	return b.Instances
+}

core/route/load_balancer.go

@@ -0,0 +1,47 @@
+package route
+
+import (
+	"github.com/alibaba/sentinel-golang/core/route/base"
+	"math/rand"
+	"sync"
+)
+
+type LoadBalancer interface {
+	Select(instances []*base.Instance, context *base.TrafficContext) (*base.Instance, error)
+}
+
+type RandomLoadBalancer struct {
+}
+
+func NewRandomLoadBalancer() *RandomLoadBalancer {
+	return &RandomLoadBalancer{}
+}
+
+func (r *RandomLoadBalancer) Select(instances []*base.Instance, context *base.TrafficContext) (*base.Instance, error) {
+	if len(instances) == 0 {
+		return nil, nil
+	}
+
+	return instances[rand.Intn(len(instances))], nil
+}
+
+type RoundRobinLoadBalancer struct {
+	idx int
+	mu  sync.Mutex
+}
+
+func NewRoundRobinLoadBalancer() *RoundRobinLoadBalancer {
+	return &RoundRobinLoadBalancer{idx: 0}
+}
+
+func (r *RoundRobinLoadBalancer) Select(instances []*base.Instance, context *base.TrafficContext) (*base.Instance, error) {
+	if len(instances) == 0 {
+		return nil, nil
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	r.idx = (r.idx + 1) % len(instances)
+	return instances[r.idx], nil
+}