Supports gRPC as a protocol (without TLS)

cmd/activator/main.go

@@ -25,7 +25,9 @@ import (
 	"github.com/knative/serving/pkg/activator"
 	clientset "github.com/knative/serving/pkg/client/clientset/versioned"
 	"github.com/knative/serving/pkg/controller"
+	h2cutil "github.com/knative/serving/pkg/h2c"
 	"github.com/knative/serving/pkg/signals"
+	"github.com/knative/serving/third_party/h2c"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 )
@@ -45,7 +47,13 @@ type activationHandler struct {
 type retryRoundTripper struct{}
 
 func (rrt retryRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
-	transport := http.DefaultTransport
+	var transport http.RoundTripper
+
+	transport = http.DefaultTransport
+	if r.ProtoMajor == 2 {
+		transport = h2cutil.NewTransport()
+	}
+
 	resp, err := transport.RoundTrip(r)
 	// TODO: Activator should retry with backoff.
 	// https://github.com/knative/serving/issues/1229
@@ -79,9 +87,11 @@ func (a *activationHandler) handler(w http.ResponseWriter, r *http.Request) {
 	}
 	proxy := httputil.NewSingleHostReverseProxy(target)
 	proxy.Transport = retryRoundTripper{}
+
 	// TODO: Clear the host to avoid 404's.
 	// https://github.com/elafros/elafros/issues/964
 	r.Host = ""
+
 	proxy.ServeHTTP(w, r)
 }
 
@@ -114,5 +124,5 @@ func main() {
 	}()
 
 	http.HandleFunc("/", ah.handler)
-	http.ListenAndServe(":8080", nil)
+	h2c.ListenAndServe(":8080", nil)
 }

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)
 }

config/400-activator-service.yaml

@@ -15,7 +15,7 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: activator-service
+  name: activator-http-service
   namespace: knative-serving-system
   labels:
     app: activator
@@ -28,3 +28,20 @@ spec:
     port: 80
     targetPort: 8080
   type: NodePort
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: activator-grpc-service
+  namespace: knative-serving-system
+  labels:
+    app: activator
+spec:
+  selector:
+    app: activator
+  ports:
+  - name: grpc
+    protocol: TCP
+    port: 80
+    targetPort: 8080
+  type: NodePort

docs/spec/spec.md

@@ -181,6 +181,8 @@ spec:
         livenessProbe: ...  # Optional
         readinessProbe: ...  # Optional
 
+      # +optional protcol (http or grpc). Defaults to http.
+      protocol: ...
       # +optional concurrency strategy.  Defaults to Multi.
       concurrencyModel: ...
       # +optional. max time the instance is allowed for responding to a request
@@ -256,6 +258,10 @@ spec:
   # scaling to/from 0.
   servingState: Active | Reserve | Retired
 
+  # The protocol used for this Revision. This will define the Istio ingress spec
+  # and service spec required for this Revision.
+  protocol: grpc | http
+
   # Some function or server frameworks or application code may be written to
   # expect that each request will be granted a single-tenant process to run
   # (i.e. that the request code is run single-threaded).
@@ -331,6 +337,7 @@ spec:  # One of "runLatest" or "pinned"
         - ...
         livenessProbe: ...  # Optional
         readinessProbe: ...  # Optional
+      protocol: ...
       concurrencyModel: ...
       timeoutSeconds: ...
       serviceAccountName: ...  # Name of the service account the code should run as
@@ -353,6 +360,7 @@ spec:  # One of "runLatest" or "pinned"
         - ...
         livenessProbe: ...  # Optional
         readinessProbe: ...  # Optional
+      protocol: ...
       concurrencyModel: ...
       timeoutSeconds: ...
       serviceAccountName: ...  # Name of the service account the code should run as

pkg/apis/serving/v1alpha1/revision_types.go

@@ -90,6 +90,17 @@ const (
 	RevisionRequestConcurrencyModelMulti RevisionRequestConcurrencyModelType = "Multi"
 )
 
+type RevisionProtocolType string
+
+const (
+	// RevisionProtocolHTTP specifies that the container will
+	// exclusively receive HTTP traffic.
+	RevisionProtocolHTTP RevisionProtocolType = "http"
+	// RevisionProtocolGRPC specifies that the container will
+	// exclusively receive gRPC traffic.
+	RevisionProtocolGRPC RevisionProtocolType = "grpc"
+)
+
 // RevisionSpec holds the desired state of the Revision (from the client).
 type RevisionSpec struct {
 	// TODO: Generation does not work correctly with CRD. They are scrubbed
@@ -112,6 +123,10 @@ type RevisionSpec struct {
 	// +optional
 	ConcurrencyModel RevisionRequestConcurrencyModelType `json:"concurrencyModel,omitempty"`
 
+	// Protocol specifies the exclusive network protocol
+	// for the Revision. Defaults to HTTP.
+	Protocol RevisionProtocolType `json:"protocol,omitempty"`
+
 	// ServiceAccountName holds the name of the Kubernetes service account
 	// as which the underlying K8s resources should be run. If unspecified
 	// this will default to the "default" service account for the namespace

pkg/controller/names.go

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"context"
+	"fmt"
 
 	"go.uber.org/zap"
 
@@ -80,8 +81,8 @@ func GetServiceRouteName(u *v1alpha1.Service) string {
 	return u.Name
 }
 
-func GetServingK8SActivatorServiceName() string {
-	return "activator-service"
+func GetServingK8SActivatorServiceName(protocol v1alpha1.RevisionProtocolType) string {
+	return fmt.Sprintf("activator-%s-service", protocol)
 }
 
 func GetRevisionHeaderName() string {

pkg/controller/revision/revision_test.go

@@ -26,6 +26,7 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -88,6 +89,7 @@ func getTestRevision() *v1alpha1.Revision {
 			UID: "test-rev-uid",
 		},
 		Spec: v1alpha1.RevisionSpec{
+			Protocol: v1alpha1.RevisionProtocolHTTP,
 			// corev1.Container has a lot of setting.  We try to pass many
 			// of them here to verify that we pass through the settings to
 			// derived objects.
@@ -634,6 +636,32 @@ func TestCreateRevCreatesStuff(t *testing.T) {
 	}
 }
 
+func TestCreateRevisionWithGRPCProtocol(t *testing.T) {
+	kubeClient, _, elaClient, controller, _, _, elaInformer, _ := newTestController(t)
+
+	config := getTestConfiguration()
+	config.Spec.RevisionTemplate.Spec.Protocol = v1alpha1.RevisionProtocolGRPC
+
+	rev := getTestRevision()
+	rev.Spec.Protocol = v1alpha1.RevisionProtocolGRPC
+
+	createRevision(elaClient, elaInformer, controller, rev)
+
+	expectedServiceName := fmt.Sprintf("%s-service", rev.Name)
+	service, err := kubeClient.CoreV1().Services(testNamespace).Get(expectedServiceName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Couldn't get revision service: %v", err)
+	}
+
+	for _, port := range service.Spec.Ports {
+		servicePortName := port.Name
+		if strings.HasPrefix(servicePortName, string(v1alpha1.RevisionProtocolHTTP)) {
+			t.Errorf("Revision service port name does not match protocol: expected %v got %v.",
+				v1alpha1.RevisionProtocolHTTP, servicePortName)
+		}
+	}
+}
+
 type errorResolver struct {
 	error string
 }

pkg/controller/revision/service.go

@@ -27,7 +27,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
-var httpServicePortName = "http"
 var servicePort = 80
 
 // MakeRevisionK8sService creates a Service that targets all pods with the same
@@ -44,7 +43,7 @@ func MakeRevisionK8sService(rev *v1alpha1.Revision) *corev1.Service {
 		Spec: corev1.ServiceSpec{
 			Ports: []corev1.ServicePort{
 				{
-					Name:       httpServicePortName,
+					Name:       string(rev.Spec.Protocol),
 					Port:       int32(servicePort),
 					TargetPort: intstr.IntOrString{Type: intstr.String, StrVal: queue.RequestQueuePortName},
 				},