Add max duration timeout

cmd/queue/main.go

@@ -70,13 +70,14 @@ var (
 )
 
 type config struct {
-	ContainerConcurrency     int    `split_words:"true" required:"true"`
-	QueueServingPort         string `split_words:"true" required:"true"`
-	UserPort                 string `split_words:"true" required:"true"`
-	RevisionTimeoutSeconds   int    `split_words:"true" required:"true"`
-	ServingReadinessProbe    string `split_words:"true"` // optional
-	EnableProfiling          bool   `split_words:"true"` // optional
-	EnableHTTP2AutoDetection bool   `split_words:"true"` // optional
+	ContainerConcurrency              int    `split_words:"true" required:"true"`
+	QueueServingPort                  string `split_words:"true" required:"true"`
+	UserPort                          string `split_words:"true" required:"true"`
+	RevisionTimeoutSeconds            int    `split_words:"true" required:"true"`
+	RevisionMaxDurationTimeoutSeconds int    `split_words:"true" required:"true"`
+	ServingReadinessProbe             string `split_words:"true"` // optional
+	EnableProfiling                   bool   `split_words:"true"` // optional
+	EnableHTTP2AutoDetection          bool   `split_words:"true"` // optional
 
 	// Logging configuration
 	ServingLoggingConfig         string `split_words:"true" required:"true"`
@@ -280,6 +281,8 @@ func buildServer(ctx context.Context, env config, drainer *pkghandler.Drainer, p
 	// hardcoded to always disable idle timeout for now, will expose this later
 	var idleTimeout time.Duration
 
+	maxDurationTimeout := time.Duration(env.RevisionMaxDurationTimeoutSeconds) * time.Second
+
 	// Create queue handler chain.
 	// Note: innermost handlers are specified first, ie. the last handler in the chain will be executed first.
 	var composedHandler http.Handler = httpProxy
@@ -303,7 +306,7 @@ func buildServer(ctx context.Context, env config, drainer *pkghandler.Drainer, p
 	}
 	composedHandler = queue.ProxyHandler(breaker, stats, tracingEnabled, composedHandler)
 	composedHandler = queue.ForwardedShimHandler(composedHandler)
-	composedHandler = handler.NewTimeoutHandler(composedHandler, "request timeout", firstByteTimeout, idleTimeout)
+	composedHandler = handler.NewTimeoutHandler(composedHandler, "request timeout", firstByteTimeout, idleTimeout, maxDurationTimeout)
 
 	if metricsSupported {
 		composedHandler = requestMetricsHandler(logger, composedHandler, env)

config/core/300-resources/configuration.yaml

@@ -474,6 +474,10 @@ spec:
                               name:
                                 description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names TODO: Add other useful fields. apiVersion, kind, uid?'
                                 type: string
+                        maxDurationTimeoutSeconds:
+                          description: maxDurationTimeoutSeconds is the maximum duration in seconds a request will be allowed to stay open.
+                          type: integer
+                          format: int64
                         serviceAccountName:
                           description: 'ServiceAccountName is the name of the ServiceAccount to use to run this pod. More info: https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/'
                           type: string

config/core/300-resources/revision.yaml

@@ -453,6 +453,10 @@ spec:
                       name:
                         description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names TODO: Add other useful fields. apiVersion, kind, uid?'
                         type: string
+                maxDurationTimeoutSeconds:
+                  description: maxDurationTimeoutSeconds is the maximum duration in seconds a request will be allowed to stay open.
+                  type: integer
+                  format: int64
                 serviceAccountName:
                   description: 'ServiceAccountName is the name of the ServiceAccount to use to run this pod. More info: https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/'
                   type: string

config/core/300-resources/service.yaml

@@ -478,6 +478,10 @@ spec:
                               name:
                                 description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names TODO: Add other useful fields. apiVersion, kind, uid?'
                                 type: string
+                        maxDurationTimeoutSeconds:
+                          description: maxDurationTimeoutSeconds is the maximum duration in seconds a request will be allowed to stay open.
+                          type: integer
+                          format: int64
                         serviceAccountName:
                           description: 'ServiceAccountName is the name of the ServiceAccount to use to run this pod. More info: https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/'
                           type: string

docs/serving-api.md

@@ -920,6 +920,19 @@ layer will wait for a request delivered to a container to begin replying
 (send network traffic). If unspecified, a system default will be provided.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>maxDurationTimeoutSeconds</code><br/>
+<em>
+int64
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>maxDurationTimeoutSeconds is the maximum duration in seconds a request will be allowed
+to stay open.</p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -1382,6 +1395,19 @@ layer will wait for a request delivered to a container to begin replying
 (send network traffic). If unspecified, a system default will be provided.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>maxDurationTimeoutSeconds</code><br/>
+<em>
+int64
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>maxDurationTimeoutSeconds is the maximum duration in seconds a request will be allowed
+to stay open.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="serving.knative.dev/v1.RevisionStatus">RevisionStatus
@@ -1582,6 +1608,19 @@ layer will wait for a request delivered to a container to begin replying
 (send network traffic). If unspecified, a system default will be provided.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>maxDurationTimeoutSeconds</code><br/>
+<em>
+int64
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>maxDurationTimeoutSeconds is the maximum duration in seconds a request will be allowed
+to stay open.</p>
+</td>
+</tr>
 </table>
 </td>
 </tr>

pkg/apis/serving/v1/revision_types.go

@@ -88,6 +88,11 @@ type RevisionSpec struct {
 	// (send network traffic). If unspecified, a system default will be provided.
 	// +optional
 	TimeoutSeconds *int64 `json:"timeoutSeconds,omitempty"`
+
+	// maxDurationTimeoutSeconds is the maximum duration in seconds a request will be allowed
+	// to stay open.
+	// +optional
+	MaxDurationTimeoutSeconds *int64 `json:"maxDurationTimeoutSeconds,omitempty"`
 }
 
 const (

pkg/http/handler/timeout.go

@@ -30,11 +30,12 @@ import (
 )
 
 type timeoutHandler struct {
-	handler          http.Handler
-	firstByteTimeout time.Duration
-	idleTimeout      time.Duration
-	body             string
-	clock            clock.Clock
+	handler            http.Handler
+	firstByteTimeout   time.Duration
+	idleTimeout        time.Duration
+	maxDurationTimeout time.Duration
+	body               string
+	clock              clock.Clock
 }
 
 // NewTimeoutHandler returns a Handler that runs `h` with the
@@ -53,13 +54,14 @@ type timeoutHandler struct {
 // https://golang.org/pkg/net/http/#Handler.
 //
 // The implementation is largely inspired by http.TimeoutHandler.
-func NewTimeoutHandler(h http.Handler, msg string, firstByteTimeout time.Duration, idleTimeout time.Duration) http.Handler {
+func NewTimeoutHandler(h http.Handler, msg string, firstByteTimeout time.Duration, idleTimeout time.Duration, maxDurationTimeout time.Duration) http.Handler {
 	return &timeoutHandler{
-		handler:          h,
-		body:             msg,
-		firstByteTimeout: firstByteTimeout,
-		idleTimeout:      idleTimeout,
-		clock:            clock.RealClock{},
+		handler:            h,
+		body:               msg,
+		firstByteTimeout:   firstByteTimeout,
+		idleTimeout:        idleTimeout,
+		maxDurationTimeout: maxDurationTimeout,
+		clock:              clock.RealClock{},
 	}
 }
 
@@ -90,13 +92,32 @@ func (h *timeoutHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// the panic from h.handler.ServeHTTP if h.handler.ServeHTTP panics.
 	done := make(chan interface{})
 	tw := &timeoutWriter{w: w, clock: h.clock}
+
+	// make sure that when max duration time out expires
+	// curTime - requestTime >= timeout
+	tw.requestStartTime = tw.clock.Now()
+
+	var maxDurationTimeout clock.Timer
+	var maxDurationTimeoutDrained bool
+	if h.maxDurationTimeout > 0 {
+		maxDurationTimeout = getTimer(h.clock, h.maxDurationTimeout)
+		defer func() {
+			putTimer(maxDurationTimeout, maxDurationTimeoutDrained)
+		}()
+	}
+	var maxDurationTimeoutCh <-chan time.Time
+	if maxDurationTimeout != nil {
+		maxDurationTimeoutCh = maxDurationTimeout.C()
+	}
+
 	go func() {
 		defer func() {
 			defer close(done)
 			if p := recover(); p != nil {
 				done <- p
 			}
 		}()
+
 		h.handler.ServeHTTP(tw, r.WithContext(ctx))
 	}()
 
@@ -119,6 +140,12 @@ func (h *timeoutHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			idleTimeout.Reset(timeToNextTimeout)
+		case <-maxDurationTimeoutCh:
+			timedOut := tw.maxDurationTimeoutAndWriteError(h.body)
+			if timedOut {
+				maxDurationTimeoutDrained = true
+				return
+			}
 		}
 	}
 }
@@ -134,9 +161,10 @@ type timeoutWriter struct {
 	w     http.ResponseWriter
 	clock clock.PassiveClock
 
-	mu            sync.Mutex
-	timedOut      bool
-	lastWriteTime time.Time
+	mu               sync.Mutex
+	timedOut         bool
+	lastWriteTime    time.Time
+	requestStartTime time.Time
 }
 
 var _ http.Flusher = (*timeoutWriter)(nil)
@@ -222,6 +250,13 @@ func (tw *timeoutWriter) tryIdleTimeoutAndWriteError(curTime time.Time, idleTime
 	return false, idleTimeout - timeSinceLastWrite
 }
 
+func (tw *timeoutWriter) maxDurationTimeoutAndWriteError(msg string) bool {
+	tw.mu.Lock()
+	defer tw.mu.Unlock()
+	tw.timeoutAndWriteError(msg)
+	return true
+}
+
 func (tw *timeoutWriter) timeoutAndWriteError(msg string) {
 	tw.w.WriteHeader(http.StatusGatewayTimeout)
 	io.WriteString(tw.w, msg)