HTTP2 and gRPC support

cmd/activator/main.go

@@ -184,14 +184,22 @@ func main() {
 		logger.Fatalw("Failed to start configuration manager", zap.Error(err))
 	}
 
-	srv := h2c.NewServer(":8080", ah)
+	http1Srv := h2c.NewServer(":8080", ah)
 	go func() {
-		if err := srv.ListenAndServe(); err != nil {
+		if err := http1Srv.ListenAndServe(); err != nil {
+			logger.Errorw("Error running HTTP server", zap.Error(err))
+		}
+	}()
+
+	h2cSrv := h2c.NewServer(":8081", ah)
+	go func() {
+		if err := h2cSrv.ListenAndServe(); err != nil {
 			logger.Errorw("Error running HTTP server", zap.Error(err))
 		}
 	}()
 
 	<-stopCh
 	a.Shutdown()
-	srv.Shutdown(context.TODO())
+	http1Srv.Shutdown(context.TODO())
+	h2cSrv.Shutdown(context.TODO())
 }

cmd/queue/main.go

@@ -189,6 +189,7 @@ func handler(w http.ResponseWriter, r *http.Request) {
 	} else {
 		proxy.ServeHTTP(w, r)
 	}
+
 }
 
 // Sets up /health and /quitquitquit endpoints.

config/400-activator-service.yaml

@@ -29,6 +29,10 @@ spec:
     port: 80
     targetPort: 8080
     nodePort: # empty removing existing value to avoid upgrade error # TODO delete after v0.4
+  - name: http2
+    protocol: TCP
+    port: 81
+    targetPort: 8081
   - name: metrics
     protocol: TCP
     port: 9090

config/activator.yaml

@@ -41,8 +41,10 @@ spec:
         # and substituted here.
         image: github.com/knative/serving/cmd/activator
         ports:
-        - name: activator-port
+        - name: http1-port
           containerPort: 8080
+        - name: h2c-port
+          containerPort: 8081
         - name: metrics-port
           containerPort: 9090
         args:

pkg/activator/activator.go

@@ -16,6 +16,8 @@ limitations under the License.
 
 package activator
 
+import "github.com/knative/serving/pkg/apis/serving/v1alpha1"
+
 const (
 	// K8sServiceName is the name of the activator service
 	K8sServiceName = "activator-service"
@@ -25,6 +27,11 @@ const (
 	RevisionHeaderName string = "knative-serving-revision"
 	// RevisionHeaderNamespace is the header key for revision's namespace
 	RevisionHeaderNamespace string = "knative-serving-namespace"
+
+	// ServicePortHTTP1 is the port number for activating HTTP1 revisions
+	ServicePortHTTP1 int32 = 80
+	// ServicePortHTTP1 is the port number for activating H2C revisions
+	ServicePortH2C int32 = 81
 )
 
 // Activator provides an active endpoint for a revision or an error and
@@ -53,3 +60,13 @@ type ActivationResult struct {
 	ConfigurationName string
 	Error             error
 }
+
+// ServicePort returns the activator service port for the given Revision protocol.
+// Default is `ServicePortHTTP1`.
+func ServicePort(protocol v1alpha1.RevisionProtocolType) int32 {
+	if protocol == v1alpha1.RevisionProtocolH2C {
+		return ServicePortH2C
+	}
+
+	return ServicePortHTTP1
+}

pkg/activator/activator_test.go

@@ -0,0 +1,49 @@
+/*
+Copyright 2018 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package activator
+
+import (
+	"testing"
+
+	"github.com/knative/serving/pkg/apis/serving/v1alpha1"
+)
+
+func TestServicePort(t *testing.T) {
+	tests := []struct {
+		name     string
+		protocol v1alpha1.RevisionProtocolType
+		port     int32
+	}{{
+		name:     "h2c",
+		protocol: v1alpha1.RevisionProtocolH2C,
+		port:     ServicePortH2C,
+	}, {
+		name:     "http1",
+		protocol: v1alpha1.RevisionProtocolHTTP1,
+		port:     ServicePortHTTP1,
+	}}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			want := tt.port
+			got := ServicePort(tt.protocol)
+
+			if want != got {
+				t.Errorf("unexpected port. want %d, got %d", want, got)
+			}
+		})
+	}
+}

pkg/activator/handler/enforce_length_handler.go

@@ -15,6 +15,8 @@ package handler
 
 import (
 	"net/http"
+
+	"github.com/knative/serving/pkg/activator/util"
 )
 
 // EnforceMaxContentLengthHandler prevents uploads larger than `MaxContentLengthBytes`
@@ -24,10 +26,14 @@ type EnforceMaxContentLengthHandler struct {
 }
 
 func (h *EnforceMaxContentLengthHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// If we have a ContentLength, we can fail early
 	if r.ContentLength > h.MaxContentLengthBytes {
 		w.WriteHeader(http.StatusRequestEntityTooLarge)
 		return
 	}
 
+	// Enforce MaxContentLengthBytes
+	r.Body = util.LimitReadCloser(r.Body, h.MaxContentLengthBytes)
+
 	h.NextHandler.ServeHTTP(w, r)
 }

pkg/activator/handler/enforce_length_handler_test.go

@@ -13,51 +13,104 @@ limitations under the License.
 package handler
 
 import (
-	"bytes"
+	"io"
+	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 )
 
 func TestEnforceMaxContentLengthHandler(t *testing.T) {
 	payload := "SAMPLE PAYLOAD"
 
+	// httptest.NewRequest will set ContentLength for strings.NewReader
+	fixed := func(p string) io.Reader { return strings.NewReader(p) }
+	stream := func(p string) io.Reader { return ioutil.NopCloser(strings.NewReader(p)) }
+
 	examples := []struct {
 		label     string
 		maxUpload int
+		request   io.Reader
+		response  string
 		status    int
 	}{
 		{
-			label:     "under",
+			label:     "under with ContentLength",
 			maxUpload: len(payload) + 1,
+			request:   fixed(payload),
 			status:    http.StatusOK,
+			response:  payload,
 		},
 		{
-			label:     "equal",
+			label:     "equal with ContentLength",
 			maxUpload: len(payload),
+			request:   fixed(payload),
 			status:    http.StatusOK,
+			response:  payload,
 		},
 		{
-			label:     "over",
+			label:     "over with ContentLength",
 			maxUpload: len(payload) - 1,
+			request:   fixed(payload),
 			status:    http.StatusRequestEntityTooLarge,
+			response:  "",
+		},
+		{
+			label:     "under without ContentLength",
+			maxUpload: len(payload) + 1,
+			request:   stream(payload),
+			status:    http.StatusOK,
+			response:  payload,
+		},
+		{
+			label:     "equal without ContentLength",
+			maxUpload: len(payload),
+			request:   stream(payload),
+			status:    http.StatusOK,
+			response:  payload,
+		},
+		{
+			label:     "over without ContentLength",
+			maxUpload: len(payload) - 1,
+			request:   stream(payload),
+			status:    http.StatusOK,
+			response:  payload[0 : len(payload)-1],
 		},
 	}
 
 	for _, e := range examples {
 		t.Run(e.label, func(t *testing.T) {
+			// Return 200 and request body
 			baseHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 				w.WriteHeader(http.StatusOK)
+
+				if r.Body != nil {
+					body, err := ioutil.ReadAll(r.Body)
+					if err != nil {
+						t.Fatalf("Error reading request body: %v", err)
+					}
+					w.Write(body)
+				}
 			})
 			handler := EnforceMaxContentLengthHandler{NextHandler: baseHandler, MaxContentLengthBytes: int64(e.maxUpload)}
 
 			resp := httptest.NewRecorder()
-			req := httptest.NewRequest("POST", "http://example.com", bytes.NewBufferString(payload))
+			req := httptest.NewRequest("POST", "http://example.com", e.request)
 
 			handler.ServeHTTP(resp, req)
 
+			gotBody, err := ioutil.ReadAll(resp.Body)
+			if err != nil {
+				t.Fatalf("Error reading response body: %v", err)
+			}
+
+			if e.response != string(gotBody) {
+				t.Errorf("Unexpected response body. Want %q, got %q", e.response, gotBody)
+			}
+
 			if resp.Code != e.status {
-				t.Errorf("Unexpected response status for payload %q. Want %d, got %d", payload, e.status, resp.Code)
+				t.Errorf("Unexpected response status. Want %d, got %d", e.status, resp.Code)
 			}
 		})
 	}

pkg/activator/revision.go

@@ -123,7 +123,7 @@ func (r *revisionActivator) revisionEndpoint(revision *v1alpha1.Revision) (end E
 	// Search for the correct port in all the service ports.
 	port := int32(-1)
 	for _, p := range svc.Spec.Ports {
-		if p.Name == revisionresources.ServicePortName {
+		if p.Name == revisionresources.ServicePortName(revision) {
 			port = p.Port
 			break
 		}

pkg/activator/revision_test.go

@@ -289,6 +289,7 @@ func newRevisionBuilder(labels map[string]string) *revisionBuilder {
 			Spec: v1alpha1.RevisionSpec{
 				Container: corev1.Container{
 					Image: "gcr.io/repo/image",
+					Ports: []corev1.ContainerPort{{Name: "h2c"}},
 				},
 			},
 			Status: v1alpha1.RevisionStatus{
@@ -335,7 +336,7 @@ func newServiceBuilder() *serviceBuilder {
 			},
 			Spec: corev1.ServiceSpec{
 				Ports: []corev1.ServicePort{{
-					Name: revisionresources.ServicePortName,
+					Name: revisionresources.ServicePortNameH2C,
 					Port: v1alpha1.DefaultUserPort,
 				}, {
 					Name: "anotherport",

pkg/activator/util/io.go

@@ -0,0 +1,34 @@
+/*
+Copyright 2018 The Knative Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import "io"
+
+// LimitReadCloser returns a ReadCloser wrapped with an `io.LimitReader`
+func LimitReadCloser(rc io.ReadCloser, l int64) io.ReadCloser {
+	return &readCloser{io.LimitReader(rc, l), rc}
+}
+
+type readCloser struct {
+	Reader io.Reader
+	Closer io.Closer
+}
+
+func (lrc *readCloser) Read(b []byte) (int, error) {
+	return lrc.Reader.Read(b)
+}
+
+func (lrc *readCloser) Close() error {
+	return lrc.Closer.Close()
+}

pkg/activator/util/io_test.go

@@ -12,10 +12,15 @@ limitations under the License.
 */
 package util
 
-import "io"
+import (
+	"io"
+	"io/ioutil"
+	"strings"
+	"testing"
+)
 
 type spyReadCloser struct {
-	io.Reader
+	io.ReadCloser
 	Closed         bool
 	ReadAfterClose bool
 }
@@ -25,11 +30,34 @@ func (s *spyReadCloser) Read(b []byte) (n int, err error) {
 		s.ReadAfterClose = true
 	}
 
-	return s.Reader.Read(b)
+	return s.ReadCloser.Read(b)
 }
 
 func (s *spyReadCloser) Close() error {
 	s.Closed = true
 
-	return nil
+	return s.ReadCloser.Close()
+}
+
+func TestLimitReadCloser(t *testing.T) {
+	want := "test"
+	r := strings.NewReader(want + " foo")
+	rc := &spyReadCloser{ReadCloser: ioutil.NopCloser(r)}
+
+	lrc := LimitReadCloser(rc, int64(len(want)))
+
+	got, err := ioutil.ReadAll(lrc)
+	lrc.Close()
+
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	if string(got) != want {
+		t.Fatalf("Unexpected body. Want %q, got %q", want, got)
+	}
+
+	if !rc.Closed {
+		t.Fatalf("Expected ReadCloser to be closed.")
+	}
 }