updates & unit test

cmd/activator/main.go

@@ -25,7 +25,6 @@ import (
  "net/http"
  "os"
  "strconv"
- "strings"
  "sync"
  "time"
 
@@ -238,6 +237,8 @@ func main() {
  // NOTE: MetricHandler is being used as the outermost handler of the meaty bits. We're not interested in measuring
  // the healthchecks or probes.
  ah = activatorhandler.NewMetricHandler(env.PodName, ah)
+ // We need the context handler to run first so ctx gets the revision info.
+ ah = activatorhandler.WrapActivatorHandlerWithFullDuplex(ah, logger)
  ah = activatorhandler.NewContextHandler(ctx, ah, configStore)
 
  // Network probe handlers.
@@ -248,8 +249,6 @@ func main() {
  hc := newHealthCheck(sigCtx, logger, statSink)
  ah = &activatorhandler.HealthHandler{HealthCheck: hc, NextHandler: ah, Logger: logger}
 
- ah = wrapActivatorHandlerWithFullDuplex(ah, logger)
-
  profilingHandler := profiling.NewHandler(logger, false)
  // Watch the logging config map and dynamically update logging levels.
  configMapWatcher.Watch(pkglogging.ConfigMapName(), pkglogging.UpdateLevelFromConfigMap(logger, atomicLevel, component))
@@ -341,16 +340,3 @@ func flush(logger *zap.SugaredLogger) {
  os.Stderr.Sync()
  metrics.FlushExporter()
 }
-
-func wrapActivatorHandlerWithFullDuplex(h http.Handler, logger *zap.SugaredLogger) http.HandlerFunc {
- return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- revEnableHTTP1FullDuplex := strings.EqualFold(activatorhandler.RevAnnotation(r.Context(), apiconfig.AllowHTTPFullDuplexFeatureKey), "Enabled")
- if revEnableHTTP1FullDuplex {
- rc := http.NewResponseController(w)
- if err := rc.EnableFullDuplex(); err != nil {
- logger.Errorw("Unable to enable full duplex", zap.Error(err))
- }
- }
- h.ServeHTTP(w, r)
- })
-}

pkg/activator/handler/handler.go

@@ -37,6 +37,7 @@ import (
  "knative.dev/pkg/tracing/propagation/tracecontextb3"
  "knative.dev/serving/pkg/activator"
  activatorconfig "knative.dev/serving/pkg/activator/config"
+ apiconfig "knative.dev/serving/pkg/apis/config"
  pkghttp "knative.dev/serving/pkg/http"
  "knative.dev/serving/pkg/networking"
  "knative.dev/serving/pkg/queue"
@@ -150,3 +151,16 @@ func useSecurePort(target string) string {
  target = strings.Split(target, ":")[0]
  return target + ":" + strconv.Itoa(networking.BackendHTTPSPort)
 }
+
+func WrapActivatorHandlerWithFullDuplex(h http.Handler, logger *zap.SugaredLogger) http.HandlerFunc {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ revEnableHTTP1FullDuplex := strings.EqualFold(RevAnnotation(r.Context(), apiconfig.AllowHTTPFullDuplexFeatureKey), "Enabled")
+ if revEnableHTTP1FullDuplex {
+ rc := http.NewResponseController(w)
+ if err := rc.EnableFullDuplex(); err != nil {
+ logger.Errorw("Unable to enable full duplex", zap.Error(err))
+ }
+ }
+ h.ServeHTTP(w, r)
+ })
+}

pkg/activator/handler/main_test.go

@@ -18,16 +18,22 @@ package handler
 
 import (
  "bytes"
+ "fmt"
  "io"
+ "io/ioutil"
+ "log"
  "net/http"
  "net/http/httptest"
+ "net/url"
+ "sync"
  "testing"
 
  netprobe "knative.dev/networking/pkg/http/probe"
  "knative.dev/pkg/logging"
  pkgnet "knative.dev/pkg/network"
  rtesting "knative.dev/pkg/reconciler/testing"
  "knative.dev/serving/pkg/activator"
+ apiconfig "knative.dev/serving/pkg/apis/config"
  asmetrics "knative.dev/serving/pkg/autoscaler/metrics"
  pkghttp "knative.dev/serving/pkg/http"
 )
@@ -114,3 +120,152 @@ func BenchmarkHandlerChain(b *testing.B) {
  })
  })
 }
+
+func TestProxyWithChainHandler(t *testing.T) {
+ ctx, cancel, _ := rtesting.SetupFakeContextWithCancel(t)
+ rev := revision(testNamespace, testRevName)
+ rev.Annotations = map[string]string{apiconfig.AllowHTTPFullDuplexFeatureKey: "Enabled"}
+ t.Cleanup(cancel)
+
+ logger := logging.FromContext(ctx)
+ configStore := setupConfigStore(t, logger)
+ revisionInformer(ctx, rev)
+
+ // Buffer equal to the activator.
+ statCh := make(chan []asmetrics.StatMessage)
+ concurrencyReporter := NewConcurrencyReporter(ctx, activatorPodName, statCh)
+ go concurrencyReporter.Run(ctx.Done())
+
+ // Just read and ignore all stat messages.
+ go func() {
+ for {
+ select {
+ case <-statCh:
+ case <-ctx.Done():
+ return
+ }
+ }
+ }()
+
+ // The server responding with the sent body.
+ echoServer := httptest.NewServer(http.HandlerFunc(
+ func(w http.ResponseWriter, req *http.Request) {
+ body, err := ioutil.ReadAll(req.Body)
+ if err != nil {
+ log.Printf("error reading body: %v", err)
+ http.Error(w, fmt.Sprintf("error reading body: %v", err), http.StatusInternalServerError)
+ return
+ }
+
+ if _, err := w.Write(body); err != nil {
+ log.Printf("error writing body: %v", err)
+ }
+ },
+ ))
+ defer echoServer.Close()
+
+ // The server proxying requests to the echo server.
+ echoURL, err := url.Parse(echoServer.URL)
+ if err != nil {
+ t.Fatalf("Failed to parse echo URL: %v", err)
+ }
+
+ proxy := pkghttp.NewHeaderPruningReverseProxy(echoURL.Host, "", []string{}, false)
+ proxy.FlushInterval = 0
+ proxyWithMiddleware := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ proxy.ServeHTTP(w, r)
+ })
+ var ah http.Handler
+ ah = concurrencyReporter.Handler(proxyWithMiddleware)
+ ah = NewTracingHandler(ah)
+ ah, _ = pkghttp.NewRequestLogHandler(ah, io.Discard, "", nil, false)
+ ah = NewMetricHandler(activatorPodName, ah)
+ ah = WrapActivatorHandlerWithFullDuplex(ah, logger)
+ ah = NewContextHandler(ctx, ah, configStore)
+ ah = &ProbeHandler{NextHandler: ah}
+ ah = netprobe.NewHandler(ah)
+ ah = &HealthHandler{HealthCheck: func() error { return nil }, NextHandler: ah, Logger: logger}
+
+ bodySize := 32 * 1024
+ parallelism := 32
+
+ proxyServer := httptest.NewServer(ah)
+
+ defer proxyServer.Close()
+
+ transport := http.DefaultTransport.(*http.Transport).Clone()
+
+ // Turning on this will hide the issue
+ transport.DisableKeepAlives = false
+ c := &http.Client{
+ Transport: transport,
+ }
+
+ body := make([]byte, bodySize)
+ for i := 0; i < cap(body); i++ {
+ body[i] = 42
+ }
+
+ for i := 0; i < 10; i++ {
+ var wg sync.WaitGroup
+ wg.Add(parallelism)
+ for i := 0; i < parallelism; i++ {
+ go func(i int) {
+ defer wg.Done()
+
+ for i := 0; i < 1000; i++ {
+ if err := send(c, proxyServer.URL, body, "test-host"); err != nil {
+ t.Errorf("error during request: %v", err)
+ }
+ }
+ }(i)
+ }
+
+ wg.Wait()
+ }
+
+}
+
+func send(client *http.Client, url string, body []byte, rHost string) error {
+ r := bytes.NewBuffer(body)
+ req, err := http.NewRequest("POST", url, r)
+
+ if rHost != "" {
+ req.Host = rHost
+ }
+
+ req.Header.Set(activator.RevisionHeaderNamespace, testNamespace)
+ req.Header.Set(activator.RevisionHeaderName, testRevName)
+
+ if err != nil {
+ return fmt.Errorf("failed to create request: %w", err)
+ }
+
+ resp, err := client.Do(req)
+ if err != nil {
+ return fmt.Errorf("failed to execute request: %w", err)
+ }
+ defer resp.Body.Close()
+
+ bd := io.Reader(resp.Body)
+
+ rec, err := ioutil.ReadAll(bd)
+
+ if err != nil {
+ return fmt.Errorf("failed to read body: %w", err)
+ }
+
+ if _, err = io.Copy(io.Discard, resp.Body); err != nil {
+ return fmt.Errorf("failed to discard body: %w", err)
+ }
+
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+ }
+
+ if len(rec) != len(body) {
+ return fmt.Errorf("unexpected body length: %d", len(rec))
+ }
+
+ return nil
+}

pkg/http/handler/timeout.go

@@ -163,13 +163,14 @@ type timeoutWriter struct {
  lastWriteTime time.Time
 }
 
+var _ http.Flusher = (*timeoutWriter)(nil)
+var _ http.ResponseWriter = (*timeoutWriter)(nil)
+
+// Unwrap returns the underlying writer
 func (tw *timeoutWriter) Unwrap() http.ResponseWriter {
  return tw.w
 }
 
-var _ http.Flusher = (*timeoutWriter)(nil)
-var _ http.ResponseWriter = (*timeoutWriter)(nil)
-
 func (tw *timeoutWriter) Flush() {
  // The inner handler of timeoutHandler can call Flush at any time including after
  // timeoutHandler.ServeHTTP has returned. Forwarding this call to the inner

pkg/http/response_recorder.go

@@ -53,6 +53,7 @@ func NewResponseRecorder(w http.ResponseWriter, responseCode int) *ResponseRecor
  }
 }
 
+// Unwrap returns the underlying writer
 func (rr *ResponseRecorder) Unwrap() http.ResponseWriter {
  return rr.writer
 }

pkg/queue/sharedmain/handlers.go

@@ -71,7 +71,7 @@ func mainHandler(
  if metricsSupported {
  composedHandler = requestAppMetricsHandler(logger, composedHandler, breaker, env)
  }
- composedHandler = queue.ProxyHandler(breaker, stats, false, composedHandler)
+ composedHandler = queue.ProxyHandler(breaker, stats, tracingEnabled, composedHandler)
  composedHandler = queue.ForwardedShimHandler(composedHandler)
  composedHandler = handler.NewTimeoutHandler(composedHandler, "request timeout", func(r *http.Request) (time.Duration, time.Duration, time.Duration) {
  return timeout, responseStartTimeout, idleTimeout