Updates queue container to support h2c

cmd/queue/main.go

@@ -32,14 +32,15 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/knative/serving/pkg"
-
 	"github.com/knative/serving/cmd/util"
+	"github.com/knative/serving/pkg"
 	"github.com/knative/serving/pkg/apis/serving/v1alpha1"
 	"github.com/knative/serving/pkg/autoscaler"
+	h2cutil "github.com/knative/serving/pkg/h2c"
 	"github.com/knative/serving/pkg/logging"
 	"github.com/knative/serving/pkg/logging/logkey"
 	"github.com/knative/serving/pkg/queue"
+	"github.com/knative/serving/third_party/h2c"
 	"go.uber.org/zap"
 
 	"github.com/gorilla/websocket"
@@ -66,18 +67,21 @@ const (
 )
 
 var (
-	podName                  string
-	elaNamespace             string
-	elaConfiguration         string
-	elaRevision              string
-	elaAutoscaler            string
-	elaAutoscalerPort        string
-	statChan                 = make(chan *autoscaler.Stat, statReportingQueueLength)
-	reqChan                  = make(chan queue.ReqEvent, requestCountingQueueLength)
-	kubeClient               *kubernetes.Clientset
-	statSink                 *websocket.Conn
-	proxy                    *httputil.ReverseProxy
-	logger                   *zap.SugaredLogger
+	podName           string
+	elaNamespace      string
+	elaConfiguration  string
+	elaRevision       string
+	elaAutoscaler     string
+	elaAutoscalerPort string
+	statChan          = make(chan *autoscaler.Stat, statReportingQueueLength)
+	reqChan           = make(chan queue.ReqEvent, requestCountingQueueLength)
+	kubeClient        *kubernetes.Clientset
+	statSink          *websocket.Conn
+	logger            *zap.SugaredLogger
+
+	h2cProxy  *httputil.ReverseProxy
+	httpProxy *httputil.ReverseProxy
+
 	concurrencyQuantumOfTime = flag.Duration("concurrencyQuantumOfTime", 100*time.Millisecond, "")
 	concurrencyModel         = flag.String("concurrencyModel", string(v1alpha1.RevisionRequestConcurrencyModelMulti), "")
 	singleConcurrencyBreaker = queue.NewBreaker(singleConcurrencyQueueDepth, 1)
@@ -141,18 +145,29 @@ func statReporter() {
 	}
 }
 
+func proxyForRequest(req *http.Request) *httputil.ReverseProxy {
+	if req.ProtoMajor == 2 {
+		return h2cProxy
+	}
+
+	return httpProxy
+}
+
 func isProbe(r *http.Request) bool {
 	// Since K8s 1.8, prober requests have
 	//   User-Agent = "kube-probe/{major-version}.{minor-version}".
 	return strings.HasPrefix(r.Header.Get("User-Agent"), "kube-probe/")
 }
 
 func handler(w http.ResponseWriter, r *http.Request) {
+	proxy := proxyForRequest(r)
+
 	if isProbe(r) {
 		// Do not count health checks for concurrency metrics
 		proxy.ServeHTTP(w, r)
 		return
 	}
+
 	// Metrics for autoscaling
 	reqChan <- queue.ReqIn
 	defer func() {
@@ -253,7 +268,10 @@ func main() {
 	if err != nil {
 		logger.Fatal("Failed to parse localhost url", zap.Error(err))
 	}
-	proxy = httputil.NewSingleHostReverseProxy(target)
+
+	httpProxy = httputil.NewSingleHostReverseProxy(target)
+	h2cProxy = httputil.NewSingleHostReverseProxy(target)
+	h2cProxy.Transport = h2cutil.NewTransport()
 
 	logger.Infof("Queue container is starting, concurrencyModel: %s", *concurrencyModel)
 	config, err := rest.InClusterConfig()
@@ -281,10 +299,15 @@ func main() {
 		}
 	}()
 
-	server := &http.Server{
-		Addr: fmt.Sprintf(":%d", queue.RequestQueuePort), Handler: nil}
 	adminServer := &http.Server{
-		Addr: fmt.Sprintf(":%d", queue.RequestQueueAdminPort), Handler: nil}
+		Addr:    fmt.Sprintf(":%d", queue.RequestQueueAdminPort),
+		Handler: nil,
+	}
+
+	h2cServer := h2c.Server{Server: &http.Server{
+		Addr:    fmt.Sprintf(":%d", queue.RequestQueuePort),
+		Handler: http.HandlerFunc(handler),
+	}}
 
 	// Add a SIGTERM handler to gracefully shutdown the servers during
 	// pod termination.
@@ -294,11 +317,12 @@ func main() {
 		<-sigTermChan
 		// Calling server.Shutdown() allows pending requests to
 		// complete, while no new work is accepted.
-		server.Shutdown(context.Background())
+
+		h2cServer.Shutdown(context.Background())
 		adminServer.Shutdown(context.Background())
 		os.Exit(0)
 	}()
-	http.HandleFunc("/", handler)
-	go server.ListenAndServe()
+
+	go h2cServer.ListenAndServe()
 	setupAdminHandlers(adminServer)
 }

pkg/h2c/transport.go

@@ -0,0 +1,21 @@
+package h2c
+
+import (
+	"crypto/tls"
+	"net"
+	"net/http"
+
+	"golang.org/x/net/http2"
+)
+
+// NewTransport will reroute all https traffic to http. This is
+// to explicitly allow h2c (http2 without TLS) transport.
+// See https://github.com/golang/go/issues/14141 for more details.
+func NewTransport() http.RoundTripper {
+	return &http2.Transport{
+		AllowHTTP: true,
+		DialTLS: func(netw, addr string, cfg *tls.Config) (net.Conn, error) {
+			return net.Dial(netw, addr)
+		},
+	}
+}