2019-06-03 18:44:43 +01:00
|
|
|
// Package mucp transparently forwards the incoming request using a go-micro client.
|
|
|
|
package mucp
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
2019-07-24 19:03:13 +01:00
|
|
|
"fmt"
|
2019-06-03 18:44:43 +01:00
|
|
|
"io"
|
2019-08-23 14:05:11 +01:00
|
|
|
"sort"
|
2019-06-03 18:44:43 +01:00
|
|
|
"strings"
|
2019-06-27 12:56:52 +01:00
|
|
|
"sync"
|
2019-09-20 16:25:29 +01:00
|
|
|
"time"
|
2019-06-03 18:44:43 +01:00
|
|
|
|
|
|
|
"github.com/micro/go-micro/client"
|
2019-10-22 11:53:49 +01:00
|
|
|
"github.com/micro/go-micro/client/selector"
|
2019-06-03 18:44:43 +01:00
|
|
|
"github.com/micro/go-micro/codec"
|
|
|
|
"github.com/micro/go-micro/codec/bytes"
|
2019-06-12 12:45:42 +01:00
|
|
|
"github.com/micro/go-micro/config/options"
|
2019-08-23 14:05:11 +01:00
|
|
|
"github.com/micro/go-micro/errors"
|
2019-10-31 22:34:06 +00:00
|
|
|
"github.com/micro/go-micro/metadata"
|
2019-08-05 17:44:33 +01:00
|
|
|
"github.com/micro/go-micro/proxy"
|
|
|
|
"github.com/micro/go-micro/router"
|
2019-06-03 18:44:43 +01:00
|
|
|
"github.com/micro/go-micro/server"
|
2019-10-24 17:51:41 +01:00
|
|
|
"github.com/micro/go-micro/util/log"
|
2019-06-03 18:44:43 +01:00
|
|
|
)
|
|
|
|
|
2019-06-07 13:42:39 +01:00
|
|
|
// Proxy will transparently proxy requests to an endpoint.
|
|
|
|
// If no endpoint is specified it will call a service using the client.
|
|
|
|
type Proxy struct {
|
|
|
|
// embed options
|
|
|
|
options.Options
|
|
|
|
|
2019-07-24 18:13:51 +01:00
|
|
|
// Endpoint specifies the fixed service endpoint to call.
|
2019-06-03 18:44:43 +01:00
|
|
|
Endpoint string
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
// The client to use for outbound requests in the local network
|
2019-06-03 18:44:43 +01:00
|
|
|
Client client.Client
|
2019-06-27 12:56:52 +01:00
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
// Links are used for outbound requests not in the local network
|
|
|
|
Links map[string]client.Client
|
|
|
|
|
2019-06-27 12:56:52 +01:00
|
|
|
// The router for routes
|
|
|
|
Router router.Router
|
|
|
|
|
|
|
|
// A fib of routes service:address
|
|
|
|
sync.RWMutex
|
2019-07-27 13:58:51 +01:00
|
|
|
Routes map[string]map[uint64]router.Route
|
2019-06-03 18:44:43 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// read client request and write to server
|
|
|
|
func readLoop(r server.Request, s client.Stream) error {
|
|
|
|
// request to backend server
|
|
|
|
req := s.Request()
|
|
|
|
|
|
|
|
for {
|
|
|
|
// get data from client
|
|
|
|
// no need to decode it
|
|
|
|
body, err := r.Read()
|
|
|
|
if err == io.EOF {
|
|
|
|
return nil
|
|
|
|
}
|
2019-06-18 18:51:52 +01:00
|
|
|
|
2019-06-03 18:44:43 +01:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// get the header from client
|
|
|
|
hdr := r.Header()
|
|
|
|
msg := &codec.Message{
|
|
|
|
Type: codec.Request,
|
|
|
|
Header: hdr,
|
|
|
|
Body: body,
|
|
|
|
}
|
2019-06-18 18:51:52 +01:00
|
|
|
|
2019-06-03 18:44:43 +01:00
|
|
|
// write the raw request
|
|
|
|
err = req.Codec().Write(msg, nil)
|
|
|
|
if err == io.EOF {
|
|
|
|
return nil
|
|
|
|
} else if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-07-24 18:13:51 +01:00
|
|
|
// toNodes returns a list of node addresses from given routes
|
2019-08-23 14:05:11 +01:00
|
|
|
func toNodes(routes []router.Route) []string {
|
2019-07-24 18:13:51 +01:00
|
|
|
var nodes []string
|
|
|
|
for _, node := range routes {
|
|
|
|
address := node.Address
|
|
|
|
if len(node.Gateway) > 0 {
|
|
|
|
address = node.Gateway
|
2019-06-27 12:56:52 +01:00
|
|
|
}
|
2019-07-24 18:13:51 +01:00
|
|
|
nodes = append(nodes, address)
|
2019-06-27 12:56:52 +01:00
|
|
|
}
|
2019-07-24 18:13:51 +01:00
|
|
|
return nodes
|
|
|
|
}
|
2019-06-27 12:56:52 +01:00
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
func toSlice(r map[uint64]router.Route) []router.Route {
|
2019-11-04 10:33:53 +00:00
|
|
|
routes := make([]router.Route, 0, len(r))
|
2019-10-25 22:46:43 +01:00
|
|
|
for _, v := range r {
|
|
|
|
routes = append(routes, v)
|
|
|
|
}
|
|
|
|
|
|
|
|
// sort the routes in order of metric
|
|
|
|
sort.Slice(routes, func(i, j int) bool { return routes[i].Metric < routes[j].Metric })
|
|
|
|
|
|
|
|
return routes
|
|
|
|
}
|
|
|
|
|
2019-10-31 22:34:06 +00:00
|
|
|
func (p *Proxy) filterRoutes(ctx context.Context, routes []router.Route) []router.Route {
|
|
|
|
md, ok := metadata.FromContext(ctx)
|
|
|
|
if !ok {
|
|
|
|
return routes
|
|
|
|
}
|
|
|
|
|
|
|
|
var filteredRoutes []router.Route
|
|
|
|
|
|
|
|
// filter the routes based on our headers
|
|
|
|
for _, route := range routes {
|
|
|
|
// process only routes for this id
|
|
|
|
if id := md["Micro-Router"]; len(id) > 0 {
|
|
|
|
if route.Router != id {
|
|
|
|
// skip routes that don't mwatch
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// only process routes with this network
|
|
|
|
if net := md["Micro-Network"]; len(net) > 0 {
|
|
|
|
if route.Network != net {
|
|
|
|
// skip routes that don't mwatch
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// process only this gateway
|
|
|
|
if gw := md["Micro-Gateway"]; len(gw) > 0 {
|
|
|
|
// if the gateway matches our address
|
|
|
|
// special case, take the routes with no gateway
|
|
|
|
// TODO: should we strip the gateway from the context?
|
|
|
|
if gw == p.Router.Options().Address {
|
|
|
|
if len(route.Gateway) > 0 && route.Gateway != gw {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
// otherwise its a local route and we're keeping it
|
|
|
|
} else {
|
|
|
|
// gateway does not match our own
|
|
|
|
if route.Gateway != gw {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// TODO: address based filtering
|
|
|
|
// address := md["Micro-Address"]
|
|
|
|
|
|
|
|
// TODO: label based filtering
|
|
|
|
// requires new field in routing table : route.Labels
|
|
|
|
|
|
|
|
// passed the filter checks
|
|
|
|
filteredRoutes = append(filteredRoutes, route)
|
|
|
|
}
|
|
|
|
|
|
|
|
return filteredRoutes
|
|
|
|
}
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
func (p *Proxy) getLink(r router.Route) (client.Client, error) {
|
|
|
|
if r.Link == "local" || len(p.Links) == 0 {
|
|
|
|
return p.Client, nil
|
|
|
|
}
|
|
|
|
l, ok := p.Links[r.Link]
|
|
|
|
if !ok {
|
2019-08-23 14:09:57 +01:00
|
|
|
return nil, errors.InternalServerError("go.micro.proxy", "link not found")
|
2019-08-23 14:05:11 +01:00
|
|
|
}
|
|
|
|
return l, nil
|
|
|
|
}
|
|
|
|
|
2019-10-31 22:34:06 +00:00
|
|
|
func (p *Proxy) getRoute(ctx context.Context, service string) ([]router.Route, error) {
|
2019-06-27 12:56:52 +01:00
|
|
|
// lookup the route cache first
|
2019-07-24 18:13:51 +01:00
|
|
|
p.Lock()
|
2019-10-25 22:46:43 +01:00
|
|
|
cached, ok := p.Routes[service]
|
2019-06-27 12:56:52 +01:00
|
|
|
if ok {
|
|
|
|
p.Unlock()
|
2019-10-31 22:34:06 +00:00
|
|
|
routes := toSlice(cached)
|
|
|
|
return p.filterRoutes(ctx, routes), nil
|
2019-06-27 12:56:52 +01:00
|
|
|
}
|
2019-07-24 18:13:51 +01:00
|
|
|
p.Unlock()
|
2019-06-27 12:56:52 +01:00
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
// cache routes for the service
|
|
|
|
routes, err := p.cacheRoutes(service)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2019-10-31 22:34:06 +00:00
|
|
|
return p.filterRoutes(ctx, routes), nil
|
2019-10-25 22:46:43 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
func (p *Proxy) cacheRoutes(service string) ([]router.Route, error) {
|
2019-07-24 18:13:51 +01:00
|
|
|
// lookup the routes in the router
|
2019-10-09 17:13:52 +01:00
|
|
|
results, err := p.Router.Lookup(router.QueryService(service))
|
2019-07-24 18:13:51 +01:00
|
|
|
if err != nil {
|
2019-08-12 17:06:08 +01:00
|
|
|
// check the status of the router
|
|
|
|
if status := p.Router.Status(); status.Code == router.Error {
|
|
|
|
return nil, status.Error
|
|
|
|
}
|
|
|
|
// otherwise return the error
|
2019-07-24 18:13:51 +01:00
|
|
|
return nil, err
|
2019-06-27 12:56:52 +01:00
|
|
|
}
|
|
|
|
|
2019-07-24 18:13:51 +01:00
|
|
|
// update the proxy cache
|
|
|
|
p.Lock()
|
|
|
|
for _, route := range results {
|
2019-08-23 14:05:11 +01:00
|
|
|
// create if does not exist
|
|
|
|
if _, ok := p.Routes[service]; !ok {
|
|
|
|
p.Routes[service] = make(map[uint64]router.Route)
|
|
|
|
}
|
2019-07-24 18:13:51 +01:00
|
|
|
p.Routes[service][route.Hash()] = route
|
2019-06-27 12:56:52 +01:00
|
|
|
}
|
2019-10-25 22:46:43 +01:00
|
|
|
routes := p.Routes[service]
|
2019-07-24 18:13:51 +01:00
|
|
|
p.Unlock()
|
2019-06-27 12:56:52 +01:00
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
return toSlice(routes), nil
|
2019-06-27 12:56:52 +01:00
|
|
|
}
|
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
// refreshMetrics will refresh any metrics for our local cached routes.
|
|
|
|
// we may not receive new watch events for these as they change.
|
|
|
|
func (p *Proxy) refreshMetrics() {
|
2019-11-04 10:33:53 +00:00
|
|
|
services := make([]string, 0, len(p.Routes))
|
2019-10-25 22:46:43 +01:00
|
|
|
|
|
|
|
// get a list of services to update
|
|
|
|
p.RLock()
|
2019-11-01 15:07:53 +00:00
|
|
|
for service := range p.Routes {
|
2019-10-25 22:46:43 +01:00
|
|
|
services = append(services, service)
|
|
|
|
}
|
|
|
|
p.RUnlock()
|
|
|
|
|
|
|
|
// get and cache the routes for the service
|
|
|
|
for _, service := range services {
|
|
|
|
p.cacheRoutes(service)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// manageRoutes applies action on a given route to Proxy route cache
|
|
|
|
func (p *Proxy) manageRoutes(route router.Route, action string) error {
|
|
|
|
// we only cache what we are actually concerned with
|
|
|
|
p.Lock()
|
|
|
|
defer p.Unlock()
|
|
|
|
|
2019-07-24 19:03:13 +01:00
|
|
|
switch action {
|
|
|
|
case "create", "update":
|
|
|
|
if _, ok := p.Routes[route.Service]; !ok {
|
2019-10-25 22:46:43 +01:00
|
|
|
return fmt.Errorf("not called %s", route.Service)
|
2019-07-24 19:03:13 +01:00
|
|
|
}
|
|
|
|
p.Routes[route.Service][route.Hash()] = route
|
|
|
|
case "delete":
|
|
|
|
delete(p.Routes[route.Service], route.Hash())
|
|
|
|
default:
|
|
|
|
return fmt.Errorf("unknown action: %s", action)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// watchRoutes watches service routes and updates proxy cache
|
|
|
|
func (p *Proxy) watchRoutes() {
|
|
|
|
// route watcher
|
|
|
|
w, err := p.Router.Watch()
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
event, err := w.Next()
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
if err := p.manageRoutes(event.Route, fmt.Sprintf("%s", event.Type)); err != nil {
|
2019-07-24 19:03:13 +01:00
|
|
|
// TODO: should we bail here?
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
func (p *Proxy) SendRequest(ctx context.Context, req client.Request, rsp client.Response) error {
|
|
|
|
return errors.InternalServerError("go.micro.proxy", "SendRequest is unsupported")
|
|
|
|
}
|
|
|
|
|
2019-06-03 18:44:43 +01:00
|
|
|
// ServeRequest honours the server.Router interface
|
2019-06-07 13:42:39 +01:00
|
|
|
func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server.Response) error {
|
2019-08-23 14:05:11 +01:00
|
|
|
// determine if its local routing
|
|
|
|
var local bool
|
|
|
|
// address to call
|
|
|
|
var addresses []string
|
|
|
|
// routes
|
|
|
|
var routes []router.Route
|
|
|
|
// service name to call
|
2019-06-03 18:44:43 +01:00
|
|
|
service := req.Service()
|
2019-08-23 14:05:11 +01:00
|
|
|
// endpoint to call
|
2019-06-03 18:44:43 +01:00
|
|
|
endpoint := req.Endpoint()
|
2019-08-23 14:05:11 +01:00
|
|
|
|
2019-08-29 13:10:06 +01:00
|
|
|
if len(service) == 0 {
|
|
|
|
return errors.BadRequest("go.micro.proxy", "service name is blank")
|
|
|
|
}
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
// are we network routing or local routing
|
|
|
|
if len(p.Links) == 0 {
|
|
|
|
local = true
|
|
|
|
}
|
2019-06-03 18:44:43 +01:00
|
|
|
|
2019-06-27 12:56:52 +01:00
|
|
|
// call a specific backend endpoint either by name or address
|
2019-06-07 13:42:39 +01:00
|
|
|
if len(p.Endpoint) > 0 {
|
2019-06-03 18:44:43 +01:00
|
|
|
// address:port
|
2019-06-07 13:42:39 +01:00
|
|
|
if parts := strings.Split(p.Endpoint, ":"); len(parts) > 1 {
|
2019-06-27 12:56:52 +01:00
|
|
|
addresses = []string{p.Endpoint}
|
2019-06-03 18:44:43 +01:00
|
|
|
} else {
|
2019-06-27 12:56:52 +01:00
|
|
|
// get route for endpoint from router
|
2019-10-31 22:34:06 +00:00
|
|
|
addr, err := p.getRoute(ctx, p.Endpoint)
|
2019-06-27 12:56:52 +01:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
// set the address
|
2019-08-23 14:05:11 +01:00
|
|
|
routes = addr
|
2019-06-27 12:56:52 +01:00
|
|
|
// set the name
|
2019-06-07 13:42:39 +01:00
|
|
|
service = p.Endpoint
|
2019-06-03 18:44:43 +01:00
|
|
|
}
|
2019-06-27 12:56:52 +01:00
|
|
|
} else {
|
|
|
|
// no endpoint was specified just lookup the route
|
|
|
|
// get route for endpoint from router
|
2019-10-31 22:34:06 +00:00
|
|
|
addr, err := p.getRoute(ctx, service)
|
2019-06-27 12:56:52 +01:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2019-08-23 14:05:11 +01:00
|
|
|
routes = addr
|
2019-06-27 12:56:52 +01:00
|
|
|
}
|
|
|
|
|
2019-10-22 11:53:49 +01:00
|
|
|
var opts []client.CallOption
|
|
|
|
|
|
|
|
// set strategy to round robin
|
|
|
|
opts = append(opts, client.WithSelectOption(selector.WithStrategy(selector.RoundRobin)))
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
// if the address is already set just serve it
|
|
|
|
// TODO: figure it out if we should know to pick a link
|
2019-06-27 12:56:52 +01:00
|
|
|
if len(addresses) > 0 {
|
2019-10-22 11:53:49 +01:00
|
|
|
opts = append(opts, client.WithAddress(addresses...))
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
// serve the normal way
|
2019-10-22 11:53:49 +01:00
|
|
|
return p.serveRequest(ctx, p.Client, service, endpoint, req, rsp, opts...)
|
2019-08-23 14:05:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// there's no links e.g we're local routing then just serve it with addresses
|
|
|
|
if local {
|
|
|
|
var opts []client.CallOption
|
|
|
|
|
|
|
|
// set address if available via routes or specific endpoint
|
|
|
|
if len(routes) > 0 {
|
|
|
|
addresses := toNodes(routes)
|
|
|
|
opts = append(opts, client.WithAddress(addresses...))
|
|
|
|
}
|
|
|
|
|
|
|
|
// serve the normal way
|
|
|
|
return p.serveRequest(ctx, p.Client, service, endpoint, req, rsp, opts...)
|
2019-06-03 18:44:43 +01:00
|
|
|
}
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
var gerr error
|
|
|
|
|
|
|
|
// we're routing globally with multiple links
|
|
|
|
// so we need to pick a link per route
|
|
|
|
for _, route := range routes {
|
|
|
|
// pick the link or error out
|
|
|
|
link, err := p.getLink(route)
|
|
|
|
if err != nil {
|
|
|
|
// ok let's try again
|
|
|
|
gerr = err
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2019-10-24 17:51:41 +01:00
|
|
|
log.Debugf("Proxy using route %+v\n", route)
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
// set the address to call
|
|
|
|
addresses := toNodes([]router.Route{route})
|
2019-10-22 11:53:49 +01:00
|
|
|
opts = append(opts, client.WithAddress(addresses...))
|
2019-08-23 14:05:11 +01:00
|
|
|
|
|
|
|
// do the request with the link
|
2019-10-22 11:53:49 +01:00
|
|
|
gerr = p.serveRequest(ctx, link, service, endpoint, req, rsp, opts...)
|
2019-08-23 14:05:11 +01:00
|
|
|
// return on no error since we succeeded
|
|
|
|
if gerr == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// return where the context deadline was exceeded
|
|
|
|
if gerr == context.Canceled || gerr == context.DeadlineExceeded {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// otherwise attempt to do it all over again
|
|
|
|
}
|
|
|
|
|
|
|
|
// if we got here something went really badly wrong
|
|
|
|
return gerr
|
|
|
|
}
|
|
|
|
|
|
|
|
func (p *Proxy) serveRequest(ctx context.Context, link client.Client, service, endpoint string, req server.Request, rsp server.Response, opts ...client.CallOption) error {
|
2019-06-03 18:44:43 +01:00
|
|
|
// read initial request
|
|
|
|
body, err := req.Read()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// create new request with raw bytes body
|
2019-08-23 14:05:11 +01:00
|
|
|
creq := link.NewRequest(service, endpoint, &bytes.Frame{body}, client.WithContentType(req.ContentType()))
|
2019-06-03 18:44:43 +01:00
|
|
|
|
2019-08-16 17:24:17 +01:00
|
|
|
// not a stream so make a client.Call request
|
2019-08-16 16:46:29 +01:00
|
|
|
if !req.Stream() {
|
2019-08-16 17:24:17 +01:00
|
|
|
crsp := new(bytes.Frame)
|
|
|
|
|
|
|
|
// make a call to the backend
|
2019-08-23 14:05:11 +01:00
|
|
|
if err := link.Call(ctx, creq, crsp, opts...); err != nil {
|
2019-08-16 17:24:17 +01:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// write the response
|
|
|
|
if err := rsp.Write(crsp.Data); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
2019-08-16 16:46:29 +01:00
|
|
|
}
|
|
|
|
|
2019-06-03 18:44:43 +01:00
|
|
|
// create new stream
|
2019-08-23 14:05:11 +01:00
|
|
|
stream, err := link.Stream(ctx, creq, opts...)
|
2019-06-03 18:44:43 +01:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
defer stream.Close()
|
|
|
|
|
2019-08-16 16:46:29 +01:00
|
|
|
// create client request read loop if streaming
|
2019-08-16 17:24:17 +01:00
|
|
|
go readLoop(req, stream)
|
2019-06-03 18:44:43 +01:00
|
|
|
|
|
|
|
// get raw response
|
|
|
|
resp := stream.Response()
|
|
|
|
|
|
|
|
// create server response write loop
|
|
|
|
for {
|
2019-10-25 22:46:43 +01:00
|
|
|
// read backend response body
|
|
|
|
body, err := resp.Read()
|
|
|
|
if err == io.EOF {
|
|
|
|
return nil
|
|
|
|
} else if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2019-06-03 18:44:43 +01:00
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
// read backend response header
|
|
|
|
hdr := resp.Header()
|
2019-06-03 18:44:43 +01:00
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
// write raw response header to client
|
|
|
|
rsp.WriteHeader(hdr)
|
2019-06-03 18:44:43 +01:00
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
// write raw response body to client
|
|
|
|
err = rsp.Write(body)
|
|
|
|
if err == io.EOF {
|
|
|
|
return nil
|
|
|
|
} else if err != nil {
|
|
|
|
return err
|
2019-06-03 18:44:43 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2019-06-07 13:42:39 +01:00
|
|
|
// NewSingleHostProxy returns a proxy which sends requests to a single backend
|
|
|
|
func NewSingleHostProxy(endpoint string) *Proxy {
|
|
|
|
return &Proxy{
|
|
|
|
Options: options.NewOptions(),
|
|
|
|
Endpoint: endpoint,
|
2019-06-03 18:44:43 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-06-07 13:42:39 +01:00
|
|
|
// NewProxy returns a new proxy which will route based on mucp headers
|
|
|
|
func NewProxy(opts ...options.Option) proxy.Proxy {
|
|
|
|
p := new(Proxy)
|
2019-08-23 14:05:11 +01:00
|
|
|
p.Links = map[string]client.Client{}
|
2019-06-07 13:42:39 +01:00
|
|
|
p.Options = options.NewOptions(opts...)
|
|
|
|
p.Options.Init(options.WithString("mucp"))
|
|
|
|
|
|
|
|
// get endpoint
|
|
|
|
ep, ok := p.Options.Values().Get("proxy.endpoint")
|
|
|
|
if ok {
|
|
|
|
p.Endpoint = ep.(string)
|
|
|
|
}
|
|
|
|
|
|
|
|
// get client
|
|
|
|
c, ok := p.Options.Values().Get("proxy.client")
|
|
|
|
if ok {
|
|
|
|
p.Client = c.(client.Client)
|
|
|
|
}
|
|
|
|
|
2019-07-24 18:13:51 +01:00
|
|
|
// set the default client
|
|
|
|
if p.Client == nil {
|
|
|
|
p.Client = client.DefaultClient
|
|
|
|
}
|
|
|
|
|
2019-08-23 14:05:11 +01:00
|
|
|
// get client
|
|
|
|
links, ok := p.Options.Values().Get("proxy.links")
|
|
|
|
if ok {
|
|
|
|
p.Links = links.(map[string]client.Client)
|
|
|
|
}
|
|
|
|
|
2019-06-26 16:12:57 +01:00
|
|
|
// get router
|
|
|
|
r, ok := p.Options.Values().Get("proxy.router")
|
|
|
|
if ok {
|
2019-06-27 12:56:52 +01:00
|
|
|
p.Router = r.(router.Router)
|
2019-06-26 16:12:57 +01:00
|
|
|
}
|
|
|
|
|
2019-07-24 18:13:51 +01:00
|
|
|
// create default router and start it
|
|
|
|
if p.Router == nil {
|
|
|
|
p.Router = router.DefaultRouter
|
|
|
|
}
|
|
|
|
|
|
|
|
// routes cache
|
2019-07-27 13:58:51 +01:00
|
|
|
p.Routes = make(map[string]map[uint64]router.Route)
|
2019-07-24 18:13:51 +01:00
|
|
|
|
2019-09-20 16:25:29 +01:00
|
|
|
go func() {
|
|
|
|
// continuously attempt to watch routes
|
|
|
|
for {
|
|
|
|
// watch the routes
|
|
|
|
p.watchRoutes()
|
|
|
|
// in case of failure just wait a second
|
|
|
|
time.Sleep(time.Second)
|
|
|
|
}
|
|
|
|
}()
|
2019-07-24 19:03:13 +01:00
|
|
|
|
2019-10-25 22:46:43 +01:00
|
|
|
go func() {
|
|
|
|
t := time.NewTicker(time.Minute)
|
|
|
|
defer t.Stop()
|
|
|
|
|
|
|
|
// we must refresh route metrics since they do not trigger new events
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-t.C:
|
|
|
|
// refresh route metrics
|
|
|
|
p.refreshMetrics()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2019-06-07 13:42:39 +01:00
|
|
|
return p
|
2019-06-03 18:44:43 +01:00
|
|
|
}
|