TUN-3500: Integrate replace h2mux by http2 work with multiple origin support

Author: chungthuang

cmd/cloudflared/config/configuration.go

@@ -190,9 +190,9 @@ func ValidateUnixSocket(c *cli.Context) (string, error) {
 
 // ValidateUrl will validate url flag correctness. It can be either from --url or argument
 // Notice ValidateUnixSocket, it will enforce --unix-socket is not used with --url or argument
-func ValidateUrl(c *cli.Context, allowFromArgs bool) (*url.URL, error) {
+func ValidateUrl(c *cli.Context, allowURLFromArgs bool) (*url.URL, error) {
 	var url = c.String("url")
-	if allowFromArgs && c.NArg() > 0 {
+	if allowURLFromArgs && c.NArg() > 0 {
 		if c.IsSet("url") {
 			return nil, errors.New("Specified origin urls using both --url and argument. Decide which one you want, I can only support one.")
 		}

cmd/cloudflared/tunnel/cmd.go

@@ -367,12 +367,12 @@ func StartServer(
 		return errors.Wrap(err, "error setting up transport logger")
 	}
 
-	tunnelConfig, err := prepareTunnelConfig(c, buildInfo, version, log, transportLogger, namedTunnel, isUIEnabled)
+	tunnelConfig, ingressRules, err := prepareTunnelConfig(c, buildInfo, version, log, transportLogger, namedTunnel, isUIEnabled)
 	if err != nil {
 		return err
 	}
 
-	tunnelConfig.IngressRules.StartOrigins(&wg, log, shutdownC, errC)
+	ingressRules.StartOrigins(&wg, log, shutdownC, errC)
 
 	reconnectCh := make(chan origin.ReconnectSignal, 1)
 	if c.IsSet("stdin-control") {
@@ -391,8 +391,7 @@ func StartServer(
 			version,
 			hostname,
 			metricsListener.Addr().String(),
-			// TODO (TUN-3461): Update UI to show multiple origin URLs
-			&tunnelConfig.IngressRules,
+			&ingressRules,
 			tunnelConfig.HAConnections,
 		)
 		logLevels, err := logger.ParseLevelString(c.String("loglevel"))

cmd/cloudflared/tunnel/configuration.go

@@ -1,6 +1,7 @@
 package tunnel
 
 import (
+	"crypto/tls"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -160,27 +161,27 @@ func prepareTunnelConfig(
 	transportLogger logger.Service,
 	namedTunnel *connection.NamedTunnelConfig,
 	uiIsEnabled bool,
-) (*origin.TunnelConfig, error) {
+) (*origin.TunnelConfig, ingress.Ingress, error) {
 	isNamedTunnel := namedTunnel != nil
 
 	hostname, err := validation.ValidateHostname(c.String("hostname"))
 	if err != nil {
 		logger.Errorf("Invalid hostname: %s", err)
-		return nil, errors.Wrap(err, "Invalid hostname")
+		return nil, ingress.Ingress{}, errors.Wrap(err, "Invalid hostname")
 	}
 	isFreeTunnel := hostname == ""
 	clientID := c.String("id")
 	if !c.IsSet("id") {
 		clientID, err = generateRandomClientID(logger)
 		if err != nil {
-			return nil, err
+			return nil, ingress.Ingress{}, err
 		}
 	}
 
 	tags, err := NewTagSliceFromCLI(c.StringSlice("tag"))
 	if err != nil {
 		logger.Errorf("Tag parse failure: %s", err)
-		return nil, errors.Wrap(err, "Tag parse failure")
+		return nil, ingress.Ingress{}, errors.Wrap(err, "Tag parse failure")
 	}
 
 	tags = append(tags, tunnelpogs.Tag{Name: "ID", Value: clientID})
@@ -189,7 +190,7 @@ func prepareTunnelConfig(
 	if !isFreeTunnel {
 		originCert, err = getOriginCert(c, logger)
 		if err != nil {
-			return nil, errors.Wrap(err, "Error getting origin cert")
+			return nil, ingress.Ingress{}, errors.Wrap(err, "Error getting origin cert")
 		}
 	}
 
@@ -200,7 +201,7 @@ func prepareTunnelConfig(
 	if isNamedTunnel {
 		clientUUID, err := uuid.NewRandom()
 		if err != nil {
-			return nil, errors.Wrap(err, "can't generate clientUUID")
+			return nil, ingress.Ingress{}, errors.Wrap(err, "can't generate clientUUID")
 		}
 		namedTunnel.Client = tunnelpogs.ClientInfo{
 			ClientID: clientUUID[:],
@@ -210,10 +211,10 @@ func prepareTunnelConfig(
 		}
 		ingressRules, err = ingress.ParseIngress(config.GetConfiguration())
 		if err != nil && err != ingress.ErrNoIngressRules {
-			return nil, err
+			return nil, ingress.Ingress{}, err
 		}
 		if !ingressRules.IsEmpty() && c.IsSet("url") {
-			return nil, ingress.ErrURLIncompatibleWithIngress
+			return nil, ingress.Ingress{}, ingress.ErrURLIncompatibleWithIngress
 		}
 	} else {
 		classicTunnel = &connection.ClassicTunnelConfig{
@@ -226,36 +227,28 @@ func prepareTunnelConfig(
 
 	// Convert single-origin configuration into multi-origin configuration.
 	if ingressRules.IsEmpty() {
-		ingressRules, err = ingress.NewSingleOrigin(c, compatibilityMode, logger)
+		ingressRules, err = ingress.NewSingleOrigin(c, !isNamedTunnel, logger)
 		if err != nil {
-			return nil, err
+			return nil, ingress.Ingress{}, err
 		}
 	}
 
 	protocolSelector, err := connection.NewProtocolSelector(c.String("protocol"), namedTunnel, edgediscovery.HTTP2Percentage, origin.ResolveTTL, logger)
 	if err != nil {
-		return nil, err
+		return nil, ingress.Ingress{}, err
 	}
 	logger.Infof("Initial protocol %s", protocolSelector.Current())
 
 	edgeTLSConfigs := make(map[connection.Protocol]*tls.Config, len(connection.ProtocolList))
 	for _, p := range connection.ProtocolList {
 		edgeTLSConfig, err := tlsconfig.CreateTunnelConfig(c, p.ServerName())
 		if err != nil {
-			return nil, errors.Wrap(err, "unable to create TLS config to connect with edge")
+			return nil, ingress.Ingress{}, errors.Wrap(err, "unable to create TLS config to connect with edge")
 		}
 		edgeTLSConfigs[p] = edgeTLSConfig
 	}
 
-	proxyConfig := &origin.ProxyConfig{
-		Client:            httpTransport,
-		URL:               originURL,
-		TLSConfig:         httpTransport.TLSClientConfig,
-		HostHeader:        c.String("http-host-header"),
-		NoChunkedEncoding: c.Bool("no-chunked-encoding"),
-		Tags:              tags,
-	}
-	originClient := origin.NewClient(proxyConfig, logger)
+	originClient := origin.NewClient(ingressRules, tags, logger)
 	connectionConfig := &connection.Config{
 		OriginClient:    originClient,
 		GracePeriod:     c.Duration("grace-period"),
@@ -275,7 +268,6 @@ func prepareTunnelConfig(
 
 	return &origin.TunnelConfig{
 		ConnectionConfig: connectionConfig,
-		ProxyConfig:      proxyConfig,
 		BuildInfo:        buildInfo,
 		ClientID:         clientID,
 		EdgeAddrs:        c.StringSlice("edge"),
@@ -284,6 +276,7 @@ func prepareTunnelConfig(
 		IsAutoupdated:    c.Bool("is-autoupdated"),
 		IsFreeTunnel:     isFreeTunnel,
 		LBPool:           c.String("lb-pool"),
+		Tags:             tags,
 		Logger:           logger,
 		Observer:         connection.NewObserver(transportLogger, tunnelEventChan),
 		ReportedVersion:  version,
@@ -293,10 +286,9 @@ func prepareTunnelConfig(
 		ClassicTunnel:    classicTunnel,
 		MuxerConfig:      muxerConfig,
 		TunnelEventChan:  tunnelEventChan,
-		IngressRules:     ingressRules,
 		ProtocolSelector: protocolSelector,
 		EdgeTLSConfigs:   edgeTLSConfigs,
-	}, nil
+	}, ingressRules, nil
 }
 
 func isRunningFromTerminal() bool {

connection/h2mux.go

@@ -22,7 +22,6 @@ const (
 type h2muxConnection struct {
 	config      *Config
 	muxerConfig *MuxerConfig
-	originURL   string
 	muxer       *h2mux.Muxer
 	// connectionID is only used by metrics, and prometheus requires labels to be string
 	connIndexStr string
@@ -54,15 +53,13 @@ func (mc *MuxerConfig) H2MuxerConfig(h h2mux.MuxedStreamHandler, logger logger.S
 func NewH2muxConnection(ctx context.Context,
 	config *Config,
 	muxerConfig *MuxerConfig,
-	originURL string,
 	edgeConn net.Conn,
 	connIndex uint8,
 	observer *Observer,
 ) (*h2muxConnection, error, bool) {
 	h := &h2muxConnection{
 		config:       config,
 		muxerConfig:  muxerConfig,
-		originURL:    originURL,
 		connIndexStr: uint8ToString(connIndex),
 		connIndex:    connIndex,
 		observer:     observer,
@@ -188,7 +185,7 @@ func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
 }
 
 func (h *h2muxConnection) newRequest(stream *h2mux.MuxedStream) (*http.Request, error) {
-	req, err := http.NewRequest("GET", h.originURL, h2mux.MuxedStreamReader{MuxedStream: stream})
+	req, err := http.NewRequest("GET", "http://localhost:8080", h2mux.MuxedStreamReader{MuxedStream: stream})
 	if err != nil {
 		return nil, errors.Wrap(err, "Unexpected error from http.NewRequest")
 	}

connection/http2.go

@@ -7,7 +7,6 @@ import (
 	"math"
 	"net"
 	"net/http"
-	"net/url"
 	"strings"
 	"sync"
 
@@ -31,7 +30,6 @@ type HTTP2Connection struct {
 	conn          net.Conn
 	server        *http2.Server
 	config        *Config
-	originURL     *url.URL
 	namedTunnel   *NamedTunnelConfig
 	connOptions   *tunnelpogs.ConnectionOptions
 	observer      *Observer
@@ -44,7 +42,6 @@ type HTTP2Connection struct {
 func NewHTTP2Connection(
 	conn net.Conn,
 	config *Config,
-	originURL *url.URL,
 	namedTunnelConfig *NamedTunnelConfig,
 	connOptions *tunnelpogs.ConnectionOptions,
 	observer *Observer,
@@ -57,7 +54,6 @@ func NewHTTP2Connection(
 			MaxConcurrentStreams: math.MaxUint32,
 		},
 		config:        config,
-		originURL:     originURL,
 		namedTunnel:   namedTunnelConfig,
 		connOptions:   connOptions,
 		observer:      observer,
@@ -83,9 +79,6 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	c.wg.Add(1)
 	defer c.wg.Done()
 
-	r.URL.Scheme = c.originURL.Scheme
-	r.URL.Host = c.originURL.Host
-
 	respWriter := &http2RespWriter{
 		r: r.Body,
 		w: w,

hello/hello.go

@@ -22,6 +22,7 @@ const (
 	UptimeRoute    = "/uptime"
 	WSRoute        = "/ws"
 	SSERoute       = "/sse"
+	HealthRoute    = "/_health"
 	defaultSSEFreq = time.Second * 10
 )
 
@@ -114,6 +115,7 @@ func StartHelloWorldServer(logger logger.Service, listener net.Listener, shutdow
 	muxer.HandleFunc(UptimeRoute, uptimeHandler(time.Now()))
 	muxer.HandleFunc(WSRoute, websocketHandler(logger, upgrader))
 	muxer.HandleFunc(SSERoute, sseHandler(logger))
+	muxer.HandleFunc(HealthRoute, healthHandler())
 	muxer.HandleFunc("/", rootHandler(serverName))
 	httpServer := &http.Server{Addr: listener.Addr().String(), Handler: muxer}
 	go func() {
@@ -221,6 +223,12 @@ func sseHandler(logger logger.Service) http.HandlerFunc {
 	}
 }
 
+func healthHandler() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("ok"))
+	}
+}
+
 func rootHandler(serverName string) http.HandlerFunc {
 	responseTemplate := template.Must(template.New("index").Parse(indexTemplate))
 	return func(w http.ResponseWriter, r *http.Request) {

ingress/ingress.go

@@ -63,9 +63,9 @@ type Ingress struct {
 
 // NewSingleOrigin constructs an Ingress set with only one rule, constructed from
 // legacy CLI parameters like --url or --no-chunked-encoding.
-func NewSingleOrigin(c *cli.Context, compatibilityMode bool, logger logger.Service) (Ingress, error) {
+func NewSingleOrigin(c *cli.Context, allowURLFromArgs bool, logger logger.Service) (Ingress, error) {
 
-	service, err := parseSingleOriginService(c, compatibilityMode)
+	service, err := parseSingleOriginService(c, allowURLFromArgs)
 	if err != nil {
 		return Ingress{}, err
 	}
@@ -85,19 +85,15 @@ func NewSingleOrigin(c *cli.Context, compatibilityMode bool, logger logger.Servi
 }
 
 // Get a single origin service from the CLI/config.
-func parseSingleOriginService(c *cli.Context, compatibilityMode bool) (OriginService, error) {
+func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (OriginService, error) {
 	if c.IsSet("hello-world") {
 		return new(helloWorld), nil
 	}
 	if c.IsSet("url") {
-		originURLStr, err := config.ValidateUrl(c, compatibilityMode)
+		originURL, err := config.ValidateUrl(c, allowURLFromArgs)
 		if err != nil {
 			return nil, errors.Wrap(err, "Error validating origin URL")
 		}
-		originURL, err := url.Parse(originURLStr)
-		if err != nil {
-			return nil, errors.Wrap(err, "couldn't parse origin URL")
-		}
 		return &localService{URL: originURL, RootURL: originURL}, nil
 	}
 	if c.IsSet("unix-socket") {

ingress/origin_service.go

@@ -245,7 +245,7 @@ type statusCode struct {
 func newStatusCode(status int) statusCode {
 	resp := &http.Response{
 		StatusCode: status,
-		Status:     http.StatusText(status),
+		Status:     fmt.Sprintf("%d %s", status, http.StatusText(status)),
 		Body:       new(NopReadCloser),
 	}
 	return statusCode{resp: resp}