Move cri/streaming to k8s.io/kubelet staging repository

Container runtimes like CRI-O and containerd reuse the code by copying
it from Kubernetes. To have a single source of truth for the streaming
server we now move the already isolated implementation to the
k8s.io/kubelet staging repository. This way runtimes can re-use the code
without copying the parts.

Signed-off-by: Sascha Grunert <sgrunert@redhat.com>

Kubernetes-commit: db9fcfeed29b860d8dd7188bc1903c4709977890

go.mod

@@ -5,17 +5,25 @@ module k8s.io/kubelet
 go 1.20
 
 require (
+	github.com/emicklei/go-restful/v3 v3.9.0
 	github.com/gogo/protobuf v1.3.2
+	github.com/stretchr/testify v1.8.2
 	google.golang.org/grpc v1.54.0
-	k8s.io/api v0.0.0-20230527221710-63505c4e796f
+	k8s.io/api v0.0.0
-	k8s.io/apimachinery v0.0.0-20230602181533-57a927096db7
+	k8s.io/apimachinery v0.0.0
-	k8s.io/component-base v0.0.0-20230602182702-3c5fd0e42366
+	k8s.io/apiserver v0.0.0
+	k8s.io/client-go v0.0.0
+	k8s.io/component-base v0.0.0
+	k8s.io/cri-api v0.0.0
+	k8s.io/klog/v2 v2.100.1
+	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2
 )
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logr/logr v1.2.4 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
@@ -23,8 +31,10 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2 // indirect
+	github.com/moby/spdystream v0.2.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.14.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.37.0 // indirect
@@ -32,22 +42,29 @@ require (
 	github.com/spf13/cobra v1.7.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	golang.org/x/net v0.9.0 // indirect
+	golang.org/x/oauth2 v0.6.0 // indirect
 	golang.org/x/sys v0.8.0 // indirect
+	golang.org/x/term v0.7.0 // indirect
 	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/time v0.3.0 // indirect
+	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
-	k8s.io/klog/v2 v2.100.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
 )
 
 replace (
-	k8s.io/api => k8s.io/api v0.0.0-20230527221710-63505c4e796f
+	k8s.io/api => ../api
-	k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20230602181533-57a927096db7
+	k8s.io/apimachinery => ../apimachinery
-	k8s.io/client-go => k8s.io/client-go v0.0.0-20230602182034-fe81cbee2bbd
+	k8s.io/apiserver => ../apiserver
-	k8s.io/component-base => k8s.io/component-base v0.0.0-20230602182702-3c5fd0e42366
+	k8s.io/client-go => ../client-go
+	k8s.io/component-base => ../component-base
+	k8s.io/cri-api => ../cri-api
+	k8s.io/kms => ../kms
+	k8s.io/kubelet => ../kubelet
 )

go.sum

@@ -38,6 +38,7 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
+github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -58,6 +59,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46t
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/emicklei/go-restful/v3 v3.9.0 h1:XwGDlfxEnQZzuopoqxwSEllNcCOM9DhhFyhFIIGKwxE=
+github.com/emicklei/go-restful/v3 v3.9.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -76,7 +79,11 @@ github.com/go-logfmt/logfmt v0.5.1/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KE
 github.com/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
 github.com/go-logr/logr v1.2.4/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-openapi/jsonpointer v0.19.6 h1:eCs3fxoIi3Wh6vtgmLTOjdhSpiqphQ+DaPn38N2ZdrE=
+github.com/go-openapi/jsonreference v0.20.2 h1:3sVjiK66+uXK/6oQ8xgcRKcFgQ5KXa2KvnJRumpMGbE=
+github.com/go-openapi/swag v0.22.3 h1:yMBqmnQ0gyZvEb/+KzuWZOXgllrXT4SADYbvDaXHv/g=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
@@ -111,6 +118,7 @@ github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg
 github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
+github.com/google/gnostic-models v0.6.8 h1:yo/ABAfM5IMRsS1VnXjTBvUb61tFIHozhlYvRgGre9I=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -134,14 +142,18 @@ github.com/google/pprof v0.0.0-20200212024743-f11f1df84d12/go.mod h1:ZgVRPoUq/hf
 github.com/google/pprof v0.0.0-20200229191704-1ebb73c60ed3/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/pprof v0.0.0-20200430221834-fc25d7d30c6d/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
+github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJYCmNdQXq6neHJOYx3V6jnqNEec=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
@@ -162,9 +174,12 @@ github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2 h1:hAHbPm5IJGijwng3PWk09JkG9WeqChjprR5s9bBZ+OM=
 github.com/matttproud/golang_protobuf_extensions v1.0.2/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/moby/spdystream v0.2.0 h1:cjW1zVyyoiM0T7b6UoySUFqzXMoqRckQtXwGPiBhOM8=
+github.com/moby/spdystream v0.2.0/go.mod h1:f7i0iNDQJ059oMTcWxx8MA/zKFIuD/lY+0GqbN2Wy8c=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -172,8 +187,10 @@ github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lN
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/onsi/ginkgo/v2 v2.9.4 h1:xR7vG4IXt5RWx6FfIjyAtsoMAtnc3C/rFXBBd2AjZwE=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -217,10 +234,15 @@ github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
+github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -307,6 +329,8 @@ golang.org/x/oauth2 v0.0.0-20191202225959-858c2ad4c8b6/go.mod h1:gOpvHmFTYa4Iltr
 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20210514164344-f6687ab2804c/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
 golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b/go.mod h1:DAh4E804XQdzx2j+YRIaUnCqCV2RuMz24cGBJ5QYIrc=
+golang.org/x/oauth2 v0.6.0 h1:Lh8GPgSKBfWSwFvtuWOfeI3aAAnbXTSutYxJiOJFgIw=
+golang.org/x/oauth2 v0.6.0/go.mod h1:ycmewcwgD4Rpr3eZJLSB4Kyyljb3qDh40vJ8STE5HKw=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -360,6 +384,8 @@ golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.7.0 h1:BEvjmm5fURWqcfbSKTdpkDXYBrUS1c0m8agp14W48vQ=
+golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -372,6 +398,8 @@ golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
+golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
@@ -414,6 +442,7 @@ golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc
 golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/tools v0.8.0 h1:vSDcovVPld282ceKgDimkRSC8kpaH1dgyc9UMzlt84Y=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@@ -440,6 +469,8 @@ google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7
 google.golang.org/appengine v1.6.1/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0=
 google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/appengine v1.6.6/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
+google.golang.org/appengine v1.6.7 h1:FZR1q0exgwxzPzp/aF+VccGrSfxfPpkBqjIIEq3ru6c=
+google.golang.org/appengine v1.6.7/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/genproto v0.0.0-20190307195333-5fe7a883aa19/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
 google.golang.org/genproto v0.0.0-20190418145605-e7d98fc518a7/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
@@ -515,6 +546,7 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
@@ -524,14 +556,9 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-k8s.io/api v0.0.0-20230527221710-63505c4e796f h1:5n05Z8OLrrVvkQF6z0OgUK3yGzRm7GY1bUwhUBp3g5g=
-k8s.io/api v0.0.0-20230527221710-63505c4e796f/go.mod h1:iclI1239YG5Z5YyOH/tKJi2vGhQ0k0dl+1xSvFaR6Dk=
-k8s.io/apimachinery v0.0.0-20230602181533-57a927096db7 h1:3z/+ZDgkV79IDqhoCLSY/eYs7icxxpqVPMP6k+pGsB4=
-k8s.io/apimachinery v0.0.0-20230602181533-57a927096db7/go.mod h1:yLwDAEEdvM6sn/hdHLSdNPaP+w2iEDBApMp2VMR2Q14=
-k8s.io/component-base v0.0.0-20230602182702-3c5fd0e42366 h1:ZediZKSfr7U82lfcJFKyRTtcObKX74n/ssFDQmgTEos=
-k8s.io/component-base v0.0.0-20230602182702-3c5fd0e42366/go.mod h1:CrtyOMvTdChfEUa3V05XRFdFt/VKsF2UC3/alVZ7NA4=
 k8s.io/klog/v2 v2.100.1 h1:7WCHKK6K8fNhTqfBhISHQ97KrnJNFZMcQvKp7gP/tmg=
 k8s.io/klog/v2 v2.100.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
+k8s.io/kube-openapi v0.0.0-20230601164746-7562a1006961 h1:pqRVJGQJz6oeZby8qmPKXYIBjyrcv7EHCe/33UkZMYA=
 k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 h1:qY1Ad8PODbnymg2pRbkyMT/ylpTrCM8P2RJ0yroCyIk=
 k8s.io/utils v0.0.0-20230406110748-d93618cff8a2/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=

pkg/cri/streaming/.import-restrictions

@@ -0,0 +1,5 @@
+rules:
+  # prevent exposing internal api in streaming packages
+  - selectorRegexp: k8s[.]io/kubernetes
+    allowedPrefixes:
+      - k8s.io/kubernetes/pkg/kubelet/cri

pkg/cri/streaming/errors.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2016 The Kubernetes 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 streaming
+
+import (
+	"net/http"
+	"strconv"
+
+	"google.golang.org/grpc/codes"
+	grpcstatus "google.golang.org/grpc/status"
+)
+
+// NewErrorTooManyInFlight creates an error for exceeding the maximum number of in-flight requests.
+func NewErrorTooManyInFlight() error {
+	return grpcstatus.Error(codes.ResourceExhausted, "maximum number of in-flight requests exceeded")
+}
+
+// WriteError translates a CRI streaming error into an appropriate HTTP response.
+func WriteError(err error, w http.ResponseWriter) error {
+	s, _ := grpcstatus.FromError(err)
+	var status int
+	switch s.Code() {
+	case codes.NotFound:
+		status = http.StatusNotFound
+	case codes.ResourceExhausted:
+		// We only expect to hit this if there is a DoS, so we just wait the full TTL.
+		// If this is ever hit in steady-state operations, consider increasing the maxInFlight requests,
+		// or plumbing through the time to next expiration.
+		w.Header().Set("Retry-After", strconv.Itoa(int(cacheTTL.Seconds())))
+		status = http.StatusTooManyRequests
+	default:
+		status = http.StatusInternalServerError
+	}
+	w.WriteHeader(status)
+	_, writeErr := w.Write([]byte(err.Error()))
+	return writeErr
+}

pkg/cri/streaming/portforward/constants.go

@@ -0,0 +1,24 @@
+/*
+Copyright 2015 The Kubernetes 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 portforward contains server-side logic for handling port forwarding requests.
+package portforward
+
+// ProtocolV1Name is the name of the subprotocol used for port forwarding.
+const ProtocolV1Name = "portforward.k8s.io"
+
+// SupportedProtocols are the supported port forwarding protocols.
+var SupportedProtocols = []string{ProtocolV1Name}

pkg/cri/streaming/portforward/httpstream.go

@@ -0,0 +1,315 @@
+/*
+Copyright 2016 The Kubernetes 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 portforward
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+
+	"k8s.io/klog/v2"
+)
+
+func handleHTTPStreams(req *http.Request, w http.ResponseWriter, portForwarder PortForwarder, podName string, uid types.UID, supportedPortForwardProtocols []string, idleTimeout, streamCreationTimeout time.Duration) error {
+	_, err := httpstream.Handshake(req, w, supportedPortForwardProtocols)
+	// negotiated protocol isn't currently used server side, but could be in the future
+	if err != nil {
+		// Handshake writes the error to the client
+		return err
+	}
+	streamChan := make(chan httpstream.Stream, 1)
+
+	klog.V(5).InfoS("Upgrading port forward response")
+	upgrader := spdy.NewResponseUpgrader()
+	conn := upgrader.UpgradeResponse(w, req, httpStreamReceived(streamChan))
+	if conn == nil {
+		return errors.New("unable to upgrade httpstream connection")
+	}
+	defer conn.Close()
+
+	klog.V(5).InfoS("Connection setting port forwarding streaming connection idle timeout", "connection", conn, "idleTimeout", idleTimeout)
+	conn.SetIdleTimeout(idleTimeout)
+
+	h := &httpStreamHandler{
+		conn:                  conn,
+		streamChan:            streamChan,
+		streamPairs:           make(map[string]*httpStreamPair),
+		streamCreationTimeout: streamCreationTimeout,
+		pod:                   podName,
+		uid:                   uid,
+		forwarder:             portForwarder,
+	}
+	h.run()
+
+	return nil
+}
+
+// httpStreamReceived is the httpstream.NewStreamHandler for port
+// forward streams. It checks each stream's port and stream type headers,
+// rejecting any streams that with missing or invalid values. Each valid
+// stream is sent to the streams channel.
+func httpStreamReceived(streams chan httpstream.Stream) func(httpstream.Stream, <-chan struct{}) error {
+	return func(stream httpstream.Stream, replySent <-chan struct{}) error {
+		// make sure it has a valid port header
+		portString := stream.Headers().Get(api.PortHeader)
+		if len(portString) == 0 {
+			return fmt.Errorf("%q header is required", api.PortHeader)
+		}
+		port, err := strconv.ParseUint(portString, 10, 16)
+		if err != nil {
+			return fmt.Errorf("unable to parse %q as a port: %v", portString, err)
+		}
+		if port < 1 {
+			return fmt.Errorf("port %q must be > 0", portString)
+		}
+
+		// make sure it has a valid stream type header
+		streamType := stream.Headers().Get(api.StreamType)
+		if len(streamType) == 0 {
+			return fmt.Errorf("%q header is required", api.StreamType)
+		}
+		if streamType != api.StreamTypeError && streamType != api.StreamTypeData {
+			return fmt.Errorf("invalid stream type %q", streamType)
+		}
+
+		streams <- stream
+		return nil
+	}
+}
+
+// httpStreamHandler is capable of processing multiple port forward
+// requests over a single httpstream.Connection.
+type httpStreamHandler struct {
+	conn                  httpstream.Connection
+	streamChan            chan httpstream.Stream
+	streamPairsLock       sync.RWMutex
+	streamPairs           map[string]*httpStreamPair
+	streamCreationTimeout time.Duration
+	pod                   string
+	uid                   types.UID
+	forwarder             PortForwarder
+}
+
+// getStreamPair returns a httpStreamPair for requestID. This creates a
+// new pair if one does not yet exist for the requestID. The returned bool is
+// true if the pair was created.
+func (h *httpStreamHandler) getStreamPair(requestID string) (*httpStreamPair, bool) {
+	h.streamPairsLock.Lock()
+	defer h.streamPairsLock.Unlock()
+
+	if p, ok := h.streamPairs[requestID]; ok {
+		klog.V(5).InfoS("Connection request found existing stream pair", "connection", h.conn, "request", requestID)
+		return p, false
+	}
+
+	klog.V(5).InfoS("Connection request creating new stream pair", "connection", h.conn, "request", requestID)
+
+	p := newPortForwardPair(requestID)
+	h.streamPairs[requestID] = p
+
+	return p, true
+}
+
+// monitorStreamPair waits for the pair to receive both its error and data
+// streams, or for the timeout to expire (whichever happens first), and then
+// removes the pair.
+func (h *httpStreamHandler) monitorStreamPair(p *httpStreamPair, timeout <-chan time.Time) {
+	select {
+	case <-timeout:
+		err := fmt.Errorf("(conn=%v, request=%s) timed out waiting for streams", h.conn, p.requestID)
+		utilruntime.HandleError(err)
+		p.printError(err.Error())
+	case <-p.complete:
+		klog.V(5).InfoS("Connection request successfully received error and data streams", "connection", h.conn, "request", p.requestID)
+	}
+	h.removeStreamPair(p.requestID)
+}
+
+// hasStreamPair returns a bool indicating if a stream pair for requestID
+// exists.
+func (h *httpStreamHandler) hasStreamPair(requestID string) bool {
+	h.streamPairsLock.RLock()
+	defer h.streamPairsLock.RUnlock()
+
+	_, ok := h.streamPairs[requestID]
+	return ok
+}
+
+// removeStreamPair removes the stream pair identified by requestID from streamPairs.
+func (h *httpStreamHandler) removeStreamPair(requestID string) {
+	h.streamPairsLock.Lock()
+	defer h.streamPairsLock.Unlock()
+
+	if h.conn != nil {
+		pair := h.streamPairs[requestID]
+		h.conn.RemoveStreams(pair.dataStream, pair.errorStream)
+	}
+	delete(h.streamPairs, requestID)
+}
+
+// requestID returns the request id for stream.
+func (h *httpStreamHandler) requestID(stream httpstream.Stream) string {
+	requestID := stream.Headers().Get(api.PortForwardRequestIDHeader)
+	if len(requestID) == 0 {
+		klog.V(5).InfoS("Connection stream received without requestID header", "connection", h.conn)
+		// If we get here, it's because the connection came from an older client
+		// that isn't generating the request id header
+		// (https://github.com/kubernetes/kubernetes/blob/843134885e7e0b360eb5441e85b1410a8b1a7a0c/pkg/client/unversioned/portforward/portforward.go#L258-L287)
+		//
+		// This is a best-effort attempt at supporting older clients.
+		//
+		// When there aren't concurrent new forwarded connections, each connection
+		// will have a pair of streams (data, error), and the stream IDs will be
+		// consecutive odd numbers, e.g. 1 and 3 for the first connection. Convert
+		// the stream ID into a pseudo-request id by taking the stream type and
+		// using id = stream.Identifier() when the stream type is error,
+		// and id = stream.Identifier() - 2 when it's data.
+		//
+		// NOTE: this only works when there are not concurrent new streams from
+		// multiple forwarded connections; it's a best-effort attempt at supporting
+		// old clients that don't generate request ids.  If there are concurrent
+		// new connections, it's possible that 1 connection gets streams whose IDs
+		// are not consecutive (e.g. 5 and 9 instead of 5 and 7).
+		streamType := stream.Headers().Get(api.StreamType)
+		switch streamType {
+		case api.StreamTypeError:
+			requestID = strconv.Itoa(int(stream.Identifier()))
+		case api.StreamTypeData:
+			requestID = strconv.Itoa(int(stream.Identifier()) - 2)
+		}
+
+		klog.V(5).InfoS("Connection automatically assigning request ID from stream type and stream ID", "connection", h.conn, "request", requestID, "streamType", streamType, "stream", stream.Identifier())
+	}
+	return requestID
+}
+
+// run is the main loop for the httpStreamHandler. It processes new
+// streams, invoking portForward for each complete stream pair. The loop exits
+// when the httpstream.Connection is closed.
+func (h *httpStreamHandler) run() {
+	klog.V(5).InfoS("Connection waiting for port forward streams", "connection", h.conn)
+Loop:
+	for {
+		select {
+		case <-h.conn.CloseChan():
+			klog.V(5).InfoS("Connection upgraded connection closed", "connection", h.conn)
+			break Loop
+		case stream := <-h.streamChan:
+			requestID := h.requestID(stream)
+			streamType := stream.Headers().Get(api.StreamType)
+			klog.V(5).InfoS("Connection request received new type of stream", "connection", h.conn, "request", requestID, "streamType", streamType)
+
+			p, created := h.getStreamPair(requestID)
+			if created {
+				go h.monitorStreamPair(p, time.After(h.streamCreationTimeout))
+			}
+			if complete, err := p.add(stream); err != nil {
+				msg := fmt.Sprintf("error processing stream for request %s: %v", requestID, err)
+				utilruntime.HandleError(errors.New(msg))
+				p.printError(msg)
+			} else if complete {
+				go h.portForward(p)
+			}
+		}
+	}
+}
+
+// portForward invokes the httpStreamHandler's forwarder.PortForward
+// function for the given stream pair.
+func (h *httpStreamHandler) portForward(p *httpStreamPair) {
+	ctx := context.Background()
+	defer p.dataStream.Close()
+	defer p.errorStream.Close()
+
+	portString := p.dataStream.Headers().Get(api.PortHeader)
+	port, _ := strconv.ParseInt(portString, 10, 32)
+
+	klog.V(5).InfoS("Connection request invoking forwarder.PortForward for port", "connection", h.conn, "request", p.requestID, "port", portString)
+	err := h.forwarder.PortForward(ctx, h.pod, h.uid, int32(port), p.dataStream)
+	klog.V(5).InfoS("Connection request done invoking forwarder.PortForward for port", "connection", h.conn, "request", p.requestID, "port", portString)
+
+	if err != nil {
+		msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", port, h.pod, h.uid, err)
+		utilruntime.HandleError(msg)
+		fmt.Fprint(p.errorStream, msg.Error())
+	}
+}
+
+// httpStreamPair represents the error and data streams for a port
+// forwarding request.
+type httpStreamPair struct {
+	lock        sync.RWMutex
+	requestID   string
+	dataStream  httpstream.Stream
+	errorStream httpstream.Stream
+	complete    chan struct{}
+}
+
+// newPortForwardPair creates a new httpStreamPair.
+func newPortForwardPair(requestID string) *httpStreamPair {
+	return &httpStreamPair{
+		requestID: requestID,
+		complete:  make(chan struct{}),
+	}
+}
+
+// add adds the stream to the httpStreamPair. If the pair already
+// contains a stream for the new stream's type, an error is returned. add
+// returns true if both the data and error streams for this pair have been
+// received.
+func (p *httpStreamPair) add(stream httpstream.Stream) (bool, error) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	switch stream.Headers().Get(api.StreamType) {
+	case api.StreamTypeError:
+		if p.errorStream != nil {
+			return false, errors.New("error stream already assigned")
+		}
+		p.errorStream = stream
+	case api.StreamTypeData:
+		if p.dataStream != nil {
+			return false, errors.New("data stream already assigned")
+		}
+		p.dataStream = stream
+	}
+
+	complete := p.errorStream != nil && p.dataStream != nil
+	if complete {
+		close(p.complete)
+	}
+	return complete, nil
+}
+
+// printError writes s to p.errorStream if p.errorStream has been set.
+func (p *httpStreamPair) printError(s string) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	if p.errorStream != nil {
+		fmt.Fprint(p.errorStream, s)
+	}
+}

pkg/cri/streaming/portforward/httpstream_test.go

@@ -0,0 +1,267 @@
+/*
+Copyright 2016 The Kubernetes 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 portforward
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+)
+
+func TestHTTPStreamReceived(t *testing.T) {
+	tests := map[string]struct {
+		port          string
+		streamType    string
+		expectedError string
+	}{
+		"missing port": {
+			expectedError: `"port" header is required`,
+		},
+		"unable to parse port": {
+			port:          "abc",
+			expectedError: `unable to parse "abc" as a port: strconv.ParseUint: parsing "abc": invalid syntax`,
+		},
+		"negative port": {
+			port:          "-1",
+			expectedError: `unable to parse "-1" as a port: strconv.ParseUint: parsing "-1": invalid syntax`,
+		},
+		"missing stream type": {
+			port:          "80",
+			expectedError: `"streamType" header is required`,
+		},
+		"valid port with error stream": {
+			port:       "80",
+			streamType: "error",
+		},
+		"valid port with data stream": {
+			port:       "80",
+			streamType: "data",
+		},
+		"invalid stream type": {
+			port:          "80",
+			streamType:    "foo",
+			expectedError: `invalid stream type "foo"`,
+		},
+	}
+	for name, test := range tests {
+		streams := make(chan httpstream.Stream, 1)
+		f := httpStreamReceived(streams)
+		stream := newFakeHTTPStream()
+		if len(test.port) > 0 {
+			stream.headers.Set("port", test.port)
+		}
+		if len(test.streamType) > 0 {
+			stream.headers.Set("streamType", test.streamType)
+		}
+		replySent := make(chan struct{})
+		err := f(stream, replySent)
+		close(replySent)
+		if len(test.expectedError) > 0 {
+			if err == nil {
+				t.Errorf("%s: expected err=%q, but it was nil", name, test.expectedError)
+			}
+			if e, a := test.expectedError, err.Error(); e != a {
+				t.Errorf("%s: expected err=%q, got %q", name, e, a)
+			}
+			continue
+		}
+		if err != nil {
+			t.Errorf("%s: unexpected error %v", name, err)
+			continue
+		}
+		if s := <-streams; s != stream {
+			t.Errorf("%s: expected stream %#v, got %#v", name, stream, s)
+		}
+	}
+}
+
+type fakeConn struct {
+	removeStreamsCalled bool
+}
+
+func (*fakeConn) CreateStream(headers http.Header) (httpstream.Stream, error) { return nil, nil }
+func (*fakeConn) Close() error                                                { return nil }
+func (*fakeConn) CloseChan() <-chan bool                                      { return nil }
+func (*fakeConn) SetIdleTimeout(timeout time.Duration)                        {}
+func (f *fakeConn) RemoveStreams(streams ...httpstream.Stream)                { f.removeStreamsCalled = true }
+
+func TestGetStreamPair(t *testing.T) {
+	timeout := make(chan time.Time)
+
+	conn := &fakeConn{}
+	h := &httpStreamHandler{
+		streamPairs: make(map[string]*httpStreamPair),
+		conn:        conn,
+	}
+
+	// test adding a new entry
+	p, created := h.getStreamPair("1")
+	if p == nil {
+		t.Fatalf("unexpected nil pair")
+	}
+	if !created {
+		t.Fatal("expected created=true")
+	}
+	if p.dataStream != nil {
+		t.Errorf("unexpected non-nil data stream")
+	}
+	if p.errorStream != nil {
+		t.Errorf("unexpected non-nil error stream")
+	}
+
+	// start the monitor for this pair
+	monitorDone := make(chan struct{})
+	go func() {
+		h.monitorStreamPair(p, timeout)
+		close(monitorDone)
+	}()
+
+	if !h.hasStreamPair("1") {
+		t.Fatal("This should still be true")
+	}
+
+	// make sure we can retrieve an existing entry
+	p2, created := h.getStreamPair("1")
+	if created {
+		t.Fatal("expected created=false")
+	}
+	if p != p2 {
+		t.Fatalf("retrieving an existing pair: expected %#v, got %#v", p, p2)
+	}
+
+	// removed via complete
+	dataStream := newFakeHTTPStream()
+	dataStream.headers.Set(api.StreamType, api.StreamTypeData)
+	complete, err := p.add(dataStream)
+	if err != nil {
+		t.Fatalf("unexpected error adding data stream to pair: %v", err)
+	}
+	if complete {
+		t.Fatalf("unexpected complete")
+	}
+
+	errorStream := newFakeHTTPStream()
+	errorStream.headers.Set(api.StreamType, api.StreamTypeError)
+	complete, err = p.add(errorStream)
+	if err != nil {
+		t.Fatalf("unexpected error adding error stream to pair: %v", err)
+	}
+	if !complete {
+		t.Fatal("unexpected incomplete")
+	}
+
+	// make sure monitorStreamPair completed
+	<-monitorDone
+
+	if !conn.removeStreamsCalled {
+		t.Fatalf("connection remove stream not called")
+	}
+	conn.removeStreamsCalled = false
+
+	// make sure the pair was removed
+	if h.hasStreamPair("1") {
+		t.Fatal("expected removal of pair after both data and error streams received")
+	}
+
+	// removed via timeout
+	p, created = h.getStreamPair("2")
+	if !created {
+		t.Fatal("expected created=true")
+	}
+	if p == nil {
+		t.Fatal("expected p not to be nil")
+	}
+
+	monitorDone = make(chan struct{})
+	go func() {
+		h.monitorStreamPair(p, timeout)
+		close(monitorDone)
+	}()
+	// cause the timeout
+	close(timeout)
+	// make sure monitorStreamPair completed
+	<-monitorDone
+	if h.hasStreamPair("2") {
+		t.Fatal("expected stream pair to be removed")
+	}
+	if !conn.removeStreamsCalled {
+		t.Fatalf("connection remove stream not called")
+	}
+}
+
+func TestRequestID(t *testing.T) {
+	h := &httpStreamHandler{}
+
+	s := newFakeHTTPStream()
+	s.headers.Set(api.StreamType, api.StreamTypeError)
+	s.id = 1
+	if e, a := "1", h.requestID(s); e != a {
+		t.Errorf("expected %q, got %q", e, a)
+	}
+
+	s.headers.Set(api.StreamType, api.StreamTypeData)
+	s.id = 3
+	if e, a := "1", h.requestID(s); e != a {
+		t.Errorf("expected %q, got %q", e, a)
+	}
+
+	s.id = 7
+	s.headers.Set(api.PortForwardRequestIDHeader, "2")
+	if e, a := "2", h.requestID(s); e != a {
+		t.Errorf("expected %q, got %q", e, a)
+	}
+}
+
+type fakeHTTPStream struct {
+	headers http.Header
+	id      uint32
+}
+
+func newFakeHTTPStream() *fakeHTTPStream {
+	return &fakeHTTPStream{
+		headers: make(http.Header),
+	}
+}
+
+var _ httpstream.Stream = &fakeHTTPStream{}
+
+func (s *fakeHTTPStream) Read(data []byte) (int, error) {
+	return 0, nil
+}
+
+func (s *fakeHTTPStream) Write(data []byte) (int, error) {
+	return 0, nil
+}
+
+func (s *fakeHTTPStream) Close() error {
+	return nil
+}
+
+func (s *fakeHTTPStream) Reset() error {
+	return nil
+}
+
+func (s *fakeHTTPStream) Headers() http.Header {
+	return s.headers
+}
+
+func (s *fakeHTTPStream) Identifier() uint32 {
+	return s.id
+}

pkg/cri/streaming/portforward/portforward.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2016 The Kubernetes 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 portforward
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/util/wsstream"
+)
+
+// PortForwarder knows how to forward content from a data stream to/from a port
+// in a pod.
+type PortForwarder interface {
+	// PortForwarder copies data between a data stream and a port in a pod.
+	PortForward(ctx context.Context, name string, uid types.UID, port int32, stream io.ReadWriteCloser) error
+}
+
+// ServePortForward handles a port forwarding request.  A single request is
+// kept alive as long as the client is still alive and the connection has not
+// been timed out due to idleness. This function handles multiple forwarded
+// connections; i.e., multiple `curl http://localhost:8888/` requests will be
+// handled by a single invocation of ServePortForward.
+func ServePortForward(w http.ResponseWriter, req *http.Request, portForwarder PortForwarder, podName string, uid types.UID, portForwardOptions *V4Options, idleTimeout time.Duration, streamCreationTimeout time.Duration, supportedProtocols []string) {
+	var err error
+	if wsstream.IsWebSocketRequest(req) {
+		err = handleWebSocketStreams(req, w, portForwarder, podName, uid, portForwardOptions, supportedProtocols, idleTimeout, streamCreationTimeout)
+	} else {
+		err = handleHTTPStreams(req, w, portForwarder, podName, uid, supportedProtocols, idleTimeout, streamCreationTimeout)
+	}
+
+	if err != nil {
+		runtime.HandleError(err)
+		return
+	}
+}

pkg/cri/streaming/portforward/websocket.go

@@ -0,0 +1,199 @@
+/*
+Copyright 2016 The Kubernetes 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 portforward
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"k8s.io/klog/v2"
+
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/endpoints/responsewriter"
+	"k8s.io/apiserver/pkg/util/wsstream"
+)
+
+const (
+	dataChannel = iota
+	errorChannel
+
+	v4BinaryWebsocketProtocol = "v4." + wsstream.ChannelWebSocketProtocol
+	v4Base64WebsocketProtocol = "v4." + wsstream.Base64ChannelWebSocketProtocol
+)
+
+// V4Options contains details about which streams are required for port
+// forwarding.
+// All fields included in V4Options need to be expressed explicitly in the
+// CRI (k8s.io/cri-api/pkg/apis/{version}/api.proto) PortForwardRequest.
+type V4Options struct {
+	Ports []int32
+}
+
+// NewV4Options creates a new options from the Request.
+func NewV4Options(req *http.Request) (*V4Options, error) {
+	if !wsstream.IsWebSocketRequest(req) {
+		return &V4Options{}, nil
+	}
+
+	portStrings := req.URL.Query()[api.PortHeader]
+	if len(portStrings) == 0 {
+		return nil, fmt.Errorf("query parameter %q is required", api.PortHeader)
+	}
+
+	ports := make([]int32, 0, len(portStrings))
+	for _, portString := range portStrings {
+		if len(portString) == 0 {
+			return nil, fmt.Errorf("query parameter %q cannot be empty", api.PortHeader)
+		}
+		for _, p := range strings.Split(portString, ",") {
+			port, err := strconv.ParseUint(p, 10, 16)
+			if err != nil {
+				return nil, fmt.Errorf("unable to parse %q as a port: %v", portString, err)
+			}
+			if port < 1 {
+				return nil, fmt.Errorf("port %q must be > 0", portString)
+			}
+			ports = append(ports, int32(port))
+		}
+	}
+
+	return &V4Options{
+		Ports: ports,
+	}, nil
+}
+
+// BuildV4Options returns a V4Options based on the given information.
+func BuildV4Options(ports []int32) (*V4Options, error) {
+	return &V4Options{Ports: ports}, nil
+}
+
+// handleWebSocketStreams handles requests to forward ports to a pod via
+// a PortForwarder. A pair of streams are created per port (DATA n,
+// ERROR n+1). The associated port is written to each stream as a unsigned 16
+// bit integer in little endian format.
+func handleWebSocketStreams(req *http.Request, w http.ResponseWriter, portForwarder PortForwarder, podName string, uid types.UID, opts *V4Options, supportedPortForwardProtocols []string, idleTimeout, streamCreationTimeout time.Duration) error {
+	channels := make([]wsstream.ChannelType, 0, len(opts.Ports)*2)
+	for i := 0; i < len(opts.Ports); i++ {
+		channels = append(channels, wsstream.ReadWriteChannel, wsstream.WriteChannel)
+	}
+	conn := wsstream.NewConn(map[string]wsstream.ChannelProtocolConfig{
+		"": {
+			Binary:   true,
+			Channels: channels,
+		},
+		v4BinaryWebsocketProtocol: {
+			Binary:   true,
+			Channels: channels,
+		},
+		v4Base64WebsocketProtocol: {
+			Binary:   false,
+			Channels: channels,
+		},
+	})
+	conn.SetIdleTimeout(idleTimeout)
+	_, streams, err := conn.Open(responsewriter.GetOriginal(w), req)
+	if err != nil {
+		err = fmt.Errorf("unable to upgrade websocket connection: %v", err)
+		return err
+	}
+	defer conn.Close()
+	streamPairs := make([]*websocketStreamPair, len(opts.Ports))
+	for i := range streamPairs {
+		streamPair := websocketStreamPair{
+			port:        opts.Ports[i],
+			dataStream:  streams[i*2+dataChannel],
+			errorStream: streams[i*2+errorChannel],
+		}
+		streamPairs[i] = &streamPair
+
+		portBytes := make([]byte, 2)
+		// port is always positive so conversion is allowable
+		binary.LittleEndian.PutUint16(portBytes, uint16(streamPair.port))
+		streamPair.dataStream.Write(portBytes)
+		streamPair.errorStream.Write(portBytes)
+	}
+	h := &websocketStreamHandler{
+		conn:        conn,
+		streamPairs: streamPairs,
+		pod:         podName,
+		uid:         uid,
+		forwarder:   portForwarder,
+	}
+	h.run()
+
+	return nil
+}
+
+// websocketStreamPair represents the error and data streams for a port
+// forwarding request.
+type websocketStreamPair struct {
+	port        int32
+	dataStream  io.ReadWriteCloser
+	errorStream io.WriteCloser
+}
+
+// websocketStreamHandler is capable of processing a single port forward
+// request over a websocket connection
+type websocketStreamHandler struct {
+	conn        *wsstream.Conn
+	streamPairs []*websocketStreamPair
+	pod         string
+	uid         types.UID
+	forwarder   PortForwarder
+}
+
+// run invokes the websocketStreamHandler's forwarder.PortForward
+// function for the given stream pair.
+func (h *websocketStreamHandler) run() {
+	wg := sync.WaitGroup{}
+	wg.Add(len(h.streamPairs))
+
+	for _, pair := range h.streamPairs {
+		p := pair
+		go func() {
+			defer wg.Done()
+			h.portForward(p)
+		}()
+	}
+
+	wg.Wait()
+}
+
+func (h *websocketStreamHandler) portForward(p *websocketStreamPair) {
+	ctx := context.Background()
+	defer p.dataStream.Close()
+	defer p.errorStream.Close()
+
+	klog.V(5).InfoS("Connection invoking forwarder.PortForward for port", "connection", h.conn, "port", p.port)
+	err := h.forwarder.PortForward(ctx, h.pod, h.uid, p.port, p.dataStream)
+	klog.V(5).InfoS("Connection done invoking forwarder.PortForward for port", "connection", h.conn, "port", p.port)
+
+	if err != nil {
+		msg := fmt.Errorf("error forwarding port %d to pod %s, uid %v: %v", p.port, h.pod, h.uid, err)
+		runtime.HandleError(msg)
+		fmt.Fprint(p.errorStream, msg.Error())
+	}
+}

pkg/cri/streaming/portforward/websocket_test.go

@@ -0,0 +1,101 @@
+/*
+Copyright 2016 The Kubernetes 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 portforward
+
+import (
+	"net/http"
+	"reflect"
+	"testing"
+)
+
+func TestV4Options(t *testing.T) {
+	tests := map[string]struct {
+		url           string
+		websocket     bool
+		expectedOpts  *V4Options
+		expectedError string
+	}{
+		"non-ws request": {
+			url:          "http://example.com",
+			expectedOpts: &V4Options{},
+		},
+		"missing port": {
+			url:           "http://example.com",
+			websocket:     true,
+			expectedError: `query parameter "port" is required`,
+		},
+		"unable to parse port": {
+			url:           "http://example.com?port=abc",
+			websocket:     true,
+			expectedError: `unable to parse "abc" as a port: strconv.ParseUint: parsing "abc": invalid syntax`,
+		},
+		"negative port": {
+			url:           "http://example.com?port=-1",
+			websocket:     true,
+			expectedError: `unable to parse "-1" as a port: strconv.ParseUint: parsing "-1": invalid syntax`,
+		},
+		"one port": {
+			url:       "http://example.com?port=80",
+			websocket: true,
+			expectedOpts: &V4Options{
+				Ports: []int32{80},
+			},
+		},
+		"multiple ports": {
+			url:       "http://example.com?port=80,90,100",
+			websocket: true,
+			expectedOpts: &V4Options{
+				Ports: []int32{80, 90, 100},
+			},
+		},
+		"multiple port": {
+			url:       "http://example.com?port=80&port=90",
+			websocket: true,
+			expectedOpts: &V4Options{
+				Ports: []int32{80, 90},
+			},
+		},
+	}
+	for name, test := range tests {
+		req, err := http.NewRequest(http.MethodGet, test.url, nil)
+		if err != nil {
+			t.Errorf("%s: invalid url %q err=%q", name, test.url, err)
+			continue
+		}
+		if test.websocket {
+			req.Header.Set("Connection", "Upgrade")
+			req.Header.Set("Upgrade", "websocket")
+		}
+		opts, err := NewV4Options(req)
+		if len(test.expectedError) > 0 {
+			if err == nil {
+				t.Errorf("%s: expected err=%q, but it was nil", name, test.expectedError)
+			}
+			if e, a := test.expectedError, err.Error(); e != a {
+				t.Errorf("%s: expected err=%q, got %q", name, e, a)
+			}
+			continue
+		}
+		if err != nil {
+			t.Errorf("%s: unexpected error %v", name, err)
+			continue
+		}
+		if !reflect.DeepEqual(test.expectedOpts, opts) {
+			t.Errorf("%s: expected options %#v, got %#v", name, test.expectedOpts, err)
+		}
+	}
+}

pkg/cri/streaming/remotecommand/attach.go

@@ -0,0 +1,60 @@
+/*
+Copyright 2016 The Kubernetes 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 remotecommand
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+// Attacher knows how to attach to a running container in a pod.
+type Attacher interface {
+	// AttachContainer attaches to the running container in the pod, copying data between in/out/err
+	// and the container's stdin/stdout/stderr.
+	AttachContainer(ctx context.Context, name string, uid types.UID, container string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error
+}
+
+// ServeAttach handles requests to attach to a container. After creating/receiving the required
+// streams, it delegates the actual attaching to attacher.
+func ServeAttach(w http.ResponseWriter, req *http.Request, attacher Attacher, podName string, uid types.UID, container string, streamOpts *Options, idleTimeout, streamCreationTimeout time.Duration, supportedProtocols []string) {
+	ctx, ok := createStreams(req, w, streamOpts, supportedProtocols, idleTimeout, streamCreationTimeout)
+	if !ok {
+		// error is handled by createStreams
+		return
+	}
+	defer ctx.conn.Close()
+
+	err := attacher.AttachContainer(req.Context(), podName, uid, container, ctx.stdinStream, ctx.stdoutStream, ctx.stderrStream, ctx.tty, ctx.resizeChan)
+	if err != nil {
+		err = fmt.Errorf("error attaching to container: %v", err)
+		runtime.HandleError(err)
+		ctx.writeStatus(apierrors.NewInternalError(err))
+	} else {
+		ctx.writeStatus(&apierrors.StatusError{ErrStatus: metav1.Status{
+			Status: metav1.StatusSuccess,
+		}})
+	}
+}

pkg/cri/streaming/remotecommand/doc.go

@@ -0,0 +1,18 @@
+/*
+Copyright 2016 The Kubernetes 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 remotecommand contains functions related to executing commands in and attaching to pods.
+package remotecommand

pkg/cri/streaming/remotecommand/exec.go

@@ -0,0 +1,80 @@
+/*
+Copyright 2016 The Kubernetes 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 remotecommand
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/remotecommand"
+	utilexec "k8s.io/utils/exec"
+)
+
+// Executor knows how to execute a command in a container in a pod.
+type Executor interface {
+	// ExecInContainer executes a command in a container in the pod, copying data
+	// between in/out/err and the container's stdin/stdout/stderr.
+	ExecInContainer(ctx context.Context, name string, uid types.UID, container string, cmd []string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error
+}
+
+// ServeExec handles requests to execute a command in a container. After
+// creating/receiving the required streams, it delegates the actual execution
+// to the executor.
+func ServeExec(w http.ResponseWriter, req *http.Request, executor Executor, podName string, uid types.UID, container string, cmd []string, streamOpts *Options, idleTimeout, streamCreationTimeout time.Duration, supportedProtocols []string) {
+	ctx, ok := createStreams(req, w, streamOpts, supportedProtocols, idleTimeout, streamCreationTimeout)
+	if !ok {
+		// error is handled by createStreams
+		return
+	}
+	defer ctx.conn.Close()
+
+	err := executor.ExecInContainer(req.Context(), podName, uid, container, cmd, ctx.stdinStream, ctx.stdoutStream, ctx.stderrStream, ctx.tty, ctx.resizeChan, 0)
+	if err != nil {
+		if exitErr, ok := err.(utilexec.ExitError); ok && exitErr.Exited() {
+			rc := exitErr.ExitStatus()
+			ctx.writeStatus(&apierrors.StatusError{ErrStatus: metav1.Status{
+				Status: metav1.StatusFailure,
+				Reason: remotecommandconsts.NonZeroExitCodeReason,
+				Details: &metav1.StatusDetails{
+					Causes: []metav1.StatusCause{
+						{
+							Type:    remotecommandconsts.ExitCodeCauseType,
+							Message: fmt.Sprintf("%d", rc),
+						},
+					},
+				},
+				Message: fmt.Sprintf("command terminated with non-zero exit code: %v", exitErr),
+			}})
+		} else {
+			err = fmt.Errorf("error executing command in container: %v", err)
+			runtime.HandleError(err)
+			ctx.writeStatus(apierrors.NewInternalError(err))
+		}
+	} else {
+		ctx.writeStatus(&apierrors.StatusError{ErrStatus: metav1.Status{
+			Status: metav1.StatusSuccess,
+		}})
+	}
+}

pkg/cri/streaming/remotecommand/httpstream.go

@@ -0,0 +1,447 @@
+/*
+Copyright 2016 The Kubernetes 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 remotecommand
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	api "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
+	remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/util/wsstream"
+	"k8s.io/client-go/tools/remotecommand"
+
+	"k8s.io/klog/v2"
+)
+
+// Options contains details about which streams are required for
+// remote command execution.
+type Options struct {
+	Stdin  bool
+	Stdout bool
+	Stderr bool
+	TTY    bool
+}
+
+// NewOptions creates a new Options from the Request.
+func NewOptions(req *http.Request) (*Options, error) {
+	tty := req.FormValue(api.ExecTTYParam) == "1"
+	stdin := req.FormValue(api.ExecStdinParam) == "1"
+	stdout := req.FormValue(api.ExecStdoutParam) == "1"
+	stderr := req.FormValue(api.ExecStderrParam) == "1"
+	if tty && stderr {
+		// TODO: make this an error before we reach this method
+		klog.V(4).InfoS("Access to exec with tty and stderr is not supported, bypassing stderr")
+		stderr = false
+	}
+
+	if !stdin && !stdout && !stderr {
+		return nil, fmt.Errorf("you must specify at least 1 of stdin, stdout, stderr")
+	}
+
+	return &Options{
+		Stdin:  stdin,
+		Stdout: stdout,
+		Stderr: stderr,
+		TTY:    tty,
+	}, nil
+}
+
+// connectionContext contains the connection and streams used when
+// forwarding an attach or execute session into a container.
+type connectionContext struct {
+	conn         io.Closer
+	stdinStream  io.ReadCloser
+	stdoutStream io.WriteCloser
+	stderrStream io.WriteCloser
+	writeStatus  func(status *apierrors.StatusError) error
+	resizeStream io.ReadCloser
+	resizeChan   chan remotecommand.TerminalSize
+	tty          bool
+}
+
+// streamAndReply holds both a Stream and a channel that is closed when the stream's reply frame is
+// enqueued. Consumers can wait for replySent to be closed prior to proceeding, to ensure that the
+// replyFrame is enqueued before the connection's goaway frame is sent (e.g. if a stream was
+// received and right after, the connection gets closed).
+type streamAndReply struct {
+	httpstream.Stream
+	replySent <-chan struct{}
+}
+
+// waitStreamReply waits until either replySent or stop is closed. If replySent is closed, it sends
+// an empty struct to the notify channel.
+func waitStreamReply(replySent <-chan struct{}, notify chan<- struct{}, stop <-chan struct{}) {
+	select {
+	case <-replySent:
+		notify <- struct{}{}
+	case <-stop:
+	}
+}
+
+func createStreams(req *http.Request, w http.ResponseWriter, opts *Options, supportedStreamProtocols []string, idleTimeout, streamCreationTimeout time.Duration) (*connectionContext, bool) {
+	var ctx *connectionContext
+	var ok bool
+	if wsstream.IsWebSocketRequest(req) {
+		ctx, ok = createWebSocketStreams(req, w, opts, idleTimeout)
+	} else {
+		ctx, ok = createHTTPStreamStreams(req, w, opts, supportedStreamProtocols, idleTimeout, streamCreationTimeout)
+	}
+	if !ok {
+		return nil, false
+	}
+
+	if ctx.resizeStream != nil {
+		ctx.resizeChan = make(chan remotecommand.TerminalSize)
+		go handleResizeEvents(ctx.resizeStream, ctx.resizeChan)
+	}
+
+	return ctx, true
+}
+
+func createHTTPStreamStreams(req *http.Request, w http.ResponseWriter, opts *Options, supportedStreamProtocols []string, idleTimeout, streamCreationTimeout time.Duration) (*connectionContext, bool) {
+	protocol, err := httpstream.Handshake(req, w, supportedStreamProtocols)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return nil, false
+	}
+
+	streamCh := make(chan streamAndReply)
+
+	upgrader := spdy.NewResponseUpgrader()
+	conn := upgrader.UpgradeResponse(w, req, func(stream httpstream.Stream, replySent <-chan struct{}) error {
+		streamCh <- streamAndReply{Stream: stream, replySent: replySent}
+		return nil
+	})
+	// from this point on, we can no longer call methods on response
+	if conn == nil {
+		// The upgrader is responsible for notifying the client of any errors that
+		// occurred during upgrading. All we can do is return here at this point
+		// if we weren't successful in upgrading.
+		return nil, false
+	}
+
+	conn.SetIdleTimeout(idleTimeout)
+
+	var handler protocolHandler
+	switch protocol {
+	case remotecommandconsts.StreamProtocolV4Name:
+		handler = &v4ProtocolHandler{}
+	case remotecommandconsts.StreamProtocolV3Name:
+		handler = &v3ProtocolHandler{}
+	case remotecommandconsts.StreamProtocolV2Name:
+		handler = &v2ProtocolHandler{}
+	case "":
+		klog.V(4).InfoS("Client did not request protocol negotiation. Falling back", "protocol", remotecommandconsts.StreamProtocolV1Name)
+		fallthrough
+	case remotecommandconsts.StreamProtocolV1Name:
+		handler = &v1ProtocolHandler{}
+	}
+
+	// count the streams client asked for, starting with 1
+	expectedStreams := 1
+	if opts.Stdin {
+		expectedStreams++
+	}
+	if opts.Stdout {
+		expectedStreams++
+	}
+	if opts.Stderr {
+		expectedStreams++
+	}
+	if opts.TTY && handler.supportsTerminalResizing() {
+		expectedStreams++
+	}
+
+	expired := time.NewTimer(streamCreationTimeout)
+	defer expired.Stop()
+
+	ctx, err := handler.waitForStreams(streamCh, expectedStreams, expired.C)
+	if err != nil {
+		runtime.HandleError(err)
+		return nil, false
+	}
+
+	ctx.conn = conn
+	ctx.tty = opts.TTY
+
+	return ctx, true
+}
+
+type protocolHandler interface {
+	// waitForStreams waits for the expected streams or a timeout, returning a
+	// remoteCommandContext if all the streams were received, or an error if not.
+	waitForStreams(streams <-chan streamAndReply, expectedStreams int, expired <-chan time.Time) (*connectionContext, error)
+	// supportsTerminalResizing returns true if the protocol handler supports terminal resizing
+	supportsTerminalResizing() bool
+}
+
+// v4ProtocolHandler implements the V4 protocol version for streaming command execution. It only differs
+// in from v3 in the error stream format using an json-marshaled metav1.Status which carries
+// the process' exit code.
+type v4ProtocolHandler struct{}
+
+func (*v4ProtocolHandler) waitForStreams(streams <-chan streamAndReply, expectedStreams int, expired <-chan time.Time) (*connectionContext, error) {
+	ctx := &connectionContext{}
+	receivedStreams := 0
+	replyChan := make(chan struct{})
+	stop := make(chan struct{})
+	defer close(stop)
+WaitForStreams:
+	for {
+		select {
+		case stream := <-streams:
+			streamType := stream.Headers().Get(api.StreamType)
+			switch streamType {
+			case api.StreamTypeError:
+				ctx.writeStatus = v4WriteStatusFunc(stream) // write json errors
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdin:
+				ctx.stdinStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdout:
+				ctx.stdoutStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStderr:
+				ctx.stderrStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeResize:
+				ctx.resizeStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			default:
+				runtime.HandleError(fmt.Errorf("unexpected stream type: %q", streamType))
+			}
+		case <-replyChan:
+			receivedStreams++
+			if receivedStreams == expectedStreams {
+				break WaitForStreams
+			}
+		case <-expired:
+			// TODO find a way to return the error to the user. Maybe use a separate
+			// stream to report errors?
+			return nil, errors.New("timed out waiting for client to create streams")
+		}
+	}
+
+	return ctx, nil
+}
+
+// supportsTerminalResizing returns true because v4ProtocolHandler supports it
+func (*v4ProtocolHandler) supportsTerminalResizing() bool { return true }
+
+// v3ProtocolHandler implements the V3 protocol version for streaming command execution.
+type v3ProtocolHandler struct{}
+
+func (*v3ProtocolHandler) waitForStreams(streams <-chan streamAndReply, expectedStreams int, expired <-chan time.Time) (*connectionContext, error) {
+	ctx := &connectionContext{}
+	receivedStreams := 0
+	replyChan := make(chan struct{})
+	stop := make(chan struct{})
+	defer close(stop)
+WaitForStreams:
+	for {
+		select {
+		case stream := <-streams:
+			streamType := stream.Headers().Get(api.StreamType)
+			switch streamType {
+			case api.StreamTypeError:
+				ctx.writeStatus = v1WriteStatusFunc(stream)
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdin:
+				ctx.stdinStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdout:
+				ctx.stdoutStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStderr:
+				ctx.stderrStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeResize:
+				ctx.resizeStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			default:
+				runtime.HandleError(fmt.Errorf("unexpected stream type: %q", streamType))
+			}
+		case <-replyChan:
+			receivedStreams++
+			if receivedStreams == expectedStreams {
+				break WaitForStreams
+			}
+		case <-expired:
+			// TODO find a way to return the error to the user. Maybe use a separate
+			// stream to report errors?
+			return nil, errors.New("timed out waiting for client to create streams")
+		}
+	}
+
+	return ctx, nil
+}
+
+// supportsTerminalResizing returns true because v3ProtocolHandler supports it
+func (*v3ProtocolHandler) supportsTerminalResizing() bool { return true }
+
+// v2ProtocolHandler implements the V2 protocol version for streaming command execution.
+type v2ProtocolHandler struct{}
+
+func (*v2ProtocolHandler) waitForStreams(streams <-chan streamAndReply, expectedStreams int, expired <-chan time.Time) (*connectionContext, error) {
+	ctx := &connectionContext{}
+	receivedStreams := 0
+	replyChan := make(chan struct{})
+	stop := make(chan struct{})
+	defer close(stop)
+WaitForStreams:
+	for {
+		select {
+		case stream := <-streams:
+			streamType := stream.Headers().Get(api.StreamType)
+			switch streamType {
+			case api.StreamTypeError:
+				ctx.writeStatus = v1WriteStatusFunc(stream)
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdin:
+				ctx.stdinStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdout:
+				ctx.stdoutStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStderr:
+				ctx.stderrStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			default:
+				runtime.HandleError(fmt.Errorf("unexpected stream type: %q", streamType))
+			}
+		case <-replyChan:
+			receivedStreams++
+			if receivedStreams == expectedStreams {
+				break WaitForStreams
+			}
+		case <-expired:
+			// TODO find a way to return the error to the user. Maybe use a separate
+			// stream to report errors?
+			return nil, errors.New("timed out waiting for client to create streams")
+		}
+	}
+
+	return ctx, nil
+}
+
+// supportsTerminalResizing returns false because v2ProtocolHandler doesn't support it.
+func (*v2ProtocolHandler) supportsTerminalResizing() bool { return false }
+
+// v1ProtocolHandler implements the V1 protocol version for streaming command execution.
+type v1ProtocolHandler struct{}
+
+func (*v1ProtocolHandler) waitForStreams(streams <-chan streamAndReply, expectedStreams int, expired <-chan time.Time) (*connectionContext, error) {
+	ctx := &connectionContext{}
+	receivedStreams := 0
+	replyChan := make(chan struct{})
+	stop := make(chan struct{})
+	defer close(stop)
+WaitForStreams:
+	for {
+		select {
+		case stream := <-streams:
+			streamType := stream.Headers().Get(api.StreamType)
+			switch streamType {
+			case api.StreamTypeError:
+				ctx.writeStatus = v1WriteStatusFunc(stream)
+
+				// This defer statement shouldn't be here, but due to previous refactoring, it ended up in
+				// here. This is what 1.0.x kubelets do, so we're retaining that behavior. This is fixed in
+				// the v2ProtocolHandler.
+				defer stream.Reset()
+
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdin:
+				ctx.stdinStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStdout:
+				ctx.stdoutStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			case api.StreamTypeStderr:
+				ctx.stderrStream = stream
+				go waitStreamReply(stream.replySent, replyChan, stop)
+			default:
+				runtime.HandleError(fmt.Errorf("unexpected stream type: %q", streamType))
+			}
+		case <-replyChan:
+			receivedStreams++
+			if receivedStreams == expectedStreams {
+				break WaitForStreams
+			}
+		case <-expired:
+			// TODO find a way to return the error to the user. Maybe use a separate
+			// stream to report errors?
+			return nil, errors.New("timed out waiting for client to create streams")
+		}
+	}
+
+	if ctx.stdinStream != nil {
+		ctx.stdinStream.Close()
+	}
+
+	return ctx, nil
+}
+
+// supportsTerminalResizing returns false because v1ProtocolHandler doesn't support it.
+func (*v1ProtocolHandler) supportsTerminalResizing() bool { return false }
+
+func handleResizeEvents(stream io.Reader, channel chan<- remotecommand.TerminalSize) {
+	defer runtime.HandleCrash()
+	defer close(channel)
+
+	decoder := json.NewDecoder(stream)
+	for {
+		size := remotecommand.TerminalSize{}
+		if err := decoder.Decode(&size); err != nil {
+			break
+		}
+		channel <- size
+	}
+}
+
+func v1WriteStatusFunc(stream io.Writer) func(status *apierrors.StatusError) error {
+	return func(status *apierrors.StatusError) error {
+		if status.Status().Status == metav1.StatusSuccess {
+			return nil // send error messages
+		}
+		_, err := stream.Write([]byte(status.Error()))
+		return err
+	}
+}
+
+// v4WriteStatusFunc returns a WriteStatusFunc that marshals a given api Status
+// as json in the error channel.
+func v4WriteStatusFunc(stream io.Writer) func(status *apierrors.StatusError) error {
+	return func(status *apierrors.StatusError) error {
+		bs, err := json.Marshal(status.Status())
+		if err != nil {
+			return err
+		}
+		_, err = stream.Write(bs)
+		return err
+	}
+}

pkg/cri/streaming/remotecommand/websocket.go

@@ -0,0 +1,132 @@
+/*
+Copyright 2016 The Kubernetes 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 remotecommand
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/endpoints/responsewriter"
+	"k8s.io/apiserver/pkg/util/wsstream"
+)
+
+const (
+	stdinChannel = iota
+	stdoutChannel
+	stderrChannel
+	errorChannel
+	resizeChannel
+
+	preV4BinaryWebsocketProtocol = wsstream.ChannelWebSocketProtocol
+	preV4Base64WebsocketProtocol = wsstream.Base64ChannelWebSocketProtocol
+	v4BinaryWebsocketProtocol    = "v4." + wsstream.ChannelWebSocketProtocol
+	v4Base64WebsocketProtocol    = "v4." + wsstream.Base64ChannelWebSocketProtocol
+)
+
+// createChannels returns the standard channel types for a shell connection (STDIN 0, STDOUT 1, STDERR 2)
+// along with the approximate duplex value. It also creates the error (3) and resize (4) channels.
+func createChannels(opts *Options) []wsstream.ChannelType {
+	// open the requested channels, and always open the error channel
+	channels := make([]wsstream.ChannelType, 5)
+	channels[stdinChannel] = readChannel(opts.Stdin)
+	channels[stdoutChannel] = writeChannel(opts.Stdout)
+	channels[stderrChannel] = writeChannel(opts.Stderr)
+	channels[errorChannel] = wsstream.WriteChannel
+	channels[resizeChannel] = wsstream.ReadChannel
+	return channels
+}
+
+// readChannel returns wsstream.ReadChannel if real is true, or wsstream.IgnoreChannel.
+func readChannel(real bool) wsstream.ChannelType {
+	if real {
+		return wsstream.ReadChannel
+	}
+	return wsstream.IgnoreChannel
+}
+
+// writeChannel returns wsstream.WriteChannel if real is true, or wsstream.IgnoreChannel.
+func writeChannel(real bool) wsstream.ChannelType {
+	if real {
+		return wsstream.WriteChannel
+	}
+	return wsstream.IgnoreChannel
+}
+
+// createWebSocketStreams returns a connectionContext containing the websocket connection and
+// streams needed to perform an exec or an attach.
+func createWebSocketStreams(req *http.Request, w http.ResponseWriter, opts *Options, idleTimeout time.Duration) (*connectionContext, bool) {
+	channels := createChannels(opts)
+	conn := wsstream.NewConn(map[string]wsstream.ChannelProtocolConfig{
+		"": {
+			Binary:   true,
+			Channels: channels,
+		},
+		preV4BinaryWebsocketProtocol: {
+			Binary:   true,
+			Channels: channels,
+		},
+		preV4Base64WebsocketProtocol: {
+			Binary:   false,
+			Channels: channels,
+		},
+		v4BinaryWebsocketProtocol: {
+			Binary:   true,
+			Channels: channels,
+		},
+		v4Base64WebsocketProtocol: {
+			Binary:   false,
+			Channels: channels,
+		},
+	})
+	conn.SetIdleTimeout(idleTimeout)
+	negotiatedProtocol, streams, err := conn.Open(responsewriter.GetOriginal(w), req)
+	if err != nil {
+		runtime.HandleError(fmt.Errorf("unable to upgrade websocket connection: %v", err))
+		return nil, false
+	}
+
+	// Send an empty message to the lowest writable channel to notify the client the connection is established
+	// TODO: make generic to SPDY and WebSockets and do it outside of this method?
+	switch {
+	case opts.Stdout:
+		streams[stdoutChannel].Write([]byte{})
+	case opts.Stderr:
+		streams[stderrChannel].Write([]byte{})
+	default:
+		streams[errorChannel].Write([]byte{})
+	}
+
+	ctx := &connectionContext{
+		conn:         conn,
+		stdinStream:  streams[stdinChannel],
+		stdoutStream: streams[stdoutChannel],
+		stderrStream: streams[stderrChannel],
+		tty:          opts.TTY,
+		resizeStream: streams[resizeChannel],
+	}
+
+	switch negotiatedProtocol {
+	case v4BinaryWebsocketProtocol, v4Base64WebsocketProtocol:
+		ctx.writeStatus = v4WriteStatusFunc(streams[errorChannel])
+	default:
+		ctx.writeStatus = v1WriteStatusFunc(streams[errorChannel])
+	}
+
+	return ctx, true
+}

pkg/cri/streaming/request_cache.go

@@ -0,0 +1,146 @@
+/*
+Copyright 2016 The Kubernetes 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 streaming
+
+import (
+	"container/list"
+	"crypto/rand"
+	"encoding/base64"
+	"fmt"
+	"math"
+	"sync"
+	"time"
+
+	"k8s.io/utils/clock"
+)
+
+var (
+	// cacheTTL is the timeout after which tokens become invalid.
+	cacheTTL = 1 * time.Minute
+	// maxInFlight is the maximum number of in-flight requests to allow.
+	maxInFlight = 1000
+	// tokenLen is the length of the random base64 encoded token identifying the request.
+	tokenLen = 8
+)
+
+// requestCache caches streaming (exec/attach/port-forward) requests and generates a single-use
+// random token for their retrieval. The requestCache is used for building streaming URLs without
+// the need to encode every request parameter in the URL.
+type requestCache struct {
+	// clock is used to obtain the current time
+	clock clock.Clock
+
+	// tokens maps the generate token to the request for fast retrieval.
+	tokens map[string]*list.Element
+	// ll maintains an age-ordered request list for faster garbage collection of expired requests.
+	ll *list.List
+
+	lock sync.Mutex
+}
+
+// Type representing an *ExecRequest, *AttachRequest, or *PortForwardRequest.
+type request interface{}
+
+type cacheEntry struct {
+	token      string `datapolicy:"token"`
+	req        request
+	expireTime time.Time
+}
+
+func newRequestCache() *requestCache {
+	return &requestCache{
+		clock:  clock.RealClock{},
+		ll:     list.New(),
+		tokens: make(map[string]*list.Element),
+	}
+}
+
+// Insert the given request into the cache and returns the token used for fetching it out.
+func (c *requestCache) Insert(req request) (token string, err error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	// Remove expired entries.
+	c.gc()
+	// If the cache is full, reject the request.
+	if c.ll.Len() == maxInFlight {
+		return "", NewErrorTooManyInFlight()
+	}
+	token, err = c.uniqueToken()
+	if err != nil {
+		return "", err
+	}
+	ele := c.ll.PushFront(&cacheEntry{token, req, c.clock.Now().Add(cacheTTL)})
+
+	c.tokens[token] = ele
+	return token, nil
+}
+
+// Consume the token (remove it from the cache) and return the cached request, if found.
+func (c *requestCache) Consume(token string) (req request, found bool) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	ele, ok := c.tokens[token]
+	if !ok {
+		return nil, false
+	}
+	c.ll.Remove(ele)
+	delete(c.tokens, token)
+
+	entry := ele.Value.(*cacheEntry)
+	if c.clock.Now().After(entry.expireTime) {
+		// Entry already expired.
+		return nil, false
+	}
+	return entry.req, true
+}
+
+// uniqueToken generates a random URL-safe token and ensures uniqueness.
+func (c *requestCache) uniqueToken() (string, error) {
+	const maxTries = 10
+	// Number of bytes to be tokenLen when base64 encoded.
+	tokenSize := math.Ceil(float64(tokenLen) * 6 / 8)
+	rawToken := make([]byte, int(tokenSize))
+	for i := 0; i < maxTries; i++ {
+		if _, err := rand.Read(rawToken); err != nil {
+			return "", err
+		}
+		encoded := base64.RawURLEncoding.EncodeToString(rawToken)
+		token := encoded[:tokenLen]
+		// If it's unique, return it. Otherwise retry.
+		if _, exists := c.tokens[encoded]; !exists {
+			return token, nil
+		}
+	}
+	return "", fmt.Errorf("failed to generate unique token")
+}
+
+// Must be write-locked prior to calling.
+func (c *requestCache) gc() {
+	now := c.clock.Now()
+	for c.ll.Len() > 0 {
+		oldest := c.ll.Back()
+		entry := oldest.Value.(*cacheEntry)
+		if !now.After(entry.expireTime) {
+			return
+		}
+
+		// Oldest value is expired; remove it.
+		c.ll.Remove(oldest)
+		delete(c.tokens, entry.token)
+	}
+}

pkg/cri/streaming/request_cache_test.go

@@ -0,0 +1,222 @@
+/*
+Copyright 2016 The Kubernetes 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 streaming
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	testingclock "k8s.io/utils/clock/testing"
+)
+
+func TestInsert(t *testing.T) {
+	c, _ := newTestCache()
+
+	// Insert normal
+	oldestTok, err := c.Insert(nextRequest())
+	require.NoError(t, err)
+	assert.Len(t, oldestTok, tokenLen)
+	assertCacheSize(t, c, 1)
+
+	// Insert until full
+	for i := 0; i < maxInFlight-2; i++ {
+		tok, err := c.Insert(nextRequest())
+		require.NoError(t, err)
+		assert.Len(t, tok, tokenLen)
+	}
+	assertCacheSize(t, c, maxInFlight-1)
+
+	newestReq := nextRequest()
+	newestTok, err := c.Insert(newestReq)
+	require.NoError(t, err)
+	assert.Len(t, newestTok, tokenLen)
+	assertCacheSize(t, c, maxInFlight)
+	require.Contains(t, c.tokens, oldestTok, "oldest request should still be cached")
+
+	// Consume newest token.
+	req, ok := c.Consume(newestTok)
+	assert.True(t, ok, "newest request should still be cached")
+	assert.Equal(t, newestReq, req)
+	require.Contains(t, c.tokens, oldestTok, "oldest request should still be cached")
+
+	// Insert again (still full)
+	tok, err := c.Insert(nextRequest())
+	require.NoError(t, err)
+	assert.Len(t, tok, tokenLen)
+	assertCacheSize(t, c, maxInFlight)
+
+	// Insert again (should evict)
+	_, err = c.Insert(nextRequest())
+	assert.Error(t, err, "should reject further requests")
+	recorder := httptest.NewRecorder()
+	require.NoError(t, WriteError(err, recorder))
+	errResponse := recorder.Result()
+	assert.Equal(t, errResponse.StatusCode, http.StatusTooManyRequests)
+	assert.Equal(t, strconv.Itoa(int(cacheTTL.Seconds())), errResponse.Header.Get("Retry-After"))
+
+	assertCacheSize(t, c, maxInFlight)
+	_, ok = c.Consume(oldestTok)
+	assert.True(t, ok, "oldest request should be valid")
+}
+
+func TestConsume(t *testing.T) {
+	c, clock := newTestCache()
+
+	{ // Insert & consume.
+		req := nextRequest()
+		tok, err := c.Insert(req)
+		require.NoError(t, err)
+		assertCacheSize(t, c, 1)
+
+		cachedReq, ok := c.Consume(tok)
+		assert.True(t, ok)
+		assert.Equal(t, req, cachedReq)
+		assertCacheSize(t, c, 0)
+	}
+
+	{ // Insert & consume out of order
+		req1 := nextRequest()
+		tok1, err := c.Insert(req1)
+		require.NoError(t, err)
+		assertCacheSize(t, c, 1)
+
+		req2 := nextRequest()
+		tok2, err := c.Insert(req2)
+		require.NoError(t, err)
+		assertCacheSize(t, c, 2)
+
+		cachedReq2, ok := c.Consume(tok2)
+		assert.True(t, ok)
+		assert.Equal(t, req2, cachedReq2)
+		assertCacheSize(t, c, 1)
+
+		cachedReq1, ok := c.Consume(tok1)
+		assert.True(t, ok)
+		assert.Equal(t, req1, cachedReq1)
+		assertCacheSize(t, c, 0)
+	}
+
+	{ // Consume a second time
+		req := nextRequest()
+		tok, err := c.Insert(req)
+		require.NoError(t, err)
+		assertCacheSize(t, c, 1)
+
+		cachedReq, ok := c.Consume(tok)
+		assert.True(t, ok)
+		assert.Equal(t, req, cachedReq)
+		assertCacheSize(t, c, 0)
+
+		_, ok = c.Consume(tok)
+		assert.False(t, ok)
+		assertCacheSize(t, c, 0)
+	}
+
+	{ // Consume without insert
+		_, ok := c.Consume("fooBAR")
+		assert.False(t, ok)
+		assertCacheSize(t, c, 0)
+	}
+
+	{ // Consume expired
+		tok, err := c.Insert(nextRequest())
+		require.NoError(t, err)
+		assertCacheSize(t, c, 1)
+
+		clock.Step(2 * cacheTTL)
+
+		_, ok := c.Consume(tok)
+		assert.False(t, ok)
+		assertCacheSize(t, c, 0)
+	}
+}
+
+func TestGC(t *testing.T) {
+	c, clock := newTestCache()
+
+	// When empty
+	c.gc()
+	assertCacheSize(t, c, 0)
+
+	tok1, err := c.Insert(nextRequest())
+	require.NoError(t, err)
+	assertCacheSize(t, c, 1)
+	clock.Step(10 * time.Second)
+	tok2, err := c.Insert(nextRequest())
+	require.NoError(t, err)
+	assertCacheSize(t, c, 2)
+
+	// expired: tok1, tok2
+	// non-expired: tok3, tok4
+	clock.Step(2 * cacheTTL)
+	tok3, err := c.Insert(nextRequest())
+	require.NoError(t, err)
+	assertCacheSize(t, c, 1)
+	clock.Step(10 * time.Second)
+	tok4, err := c.Insert(nextRequest())
+	require.NoError(t, err)
+	assertCacheSize(t, c, 2)
+
+	_, ok := c.Consume(tok1)
+	assert.False(t, ok)
+	_, ok = c.Consume(tok2)
+	assert.False(t, ok)
+	_, ok = c.Consume(tok3)
+	assert.True(t, ok)
+	_, ok = c.Consume(tok4)
+	assert.True(t, ok)
+
+	// When full, nothing is expired.
+	for i := 0; i < maxInFlight; i++ {
+		_, err := c.Insert(nextRequest())
+		require.NoError(t, err)
+	}
+	assertCacheSize(t, c, maxInFlight)
+
+	// When everything is expired
+	clock.Step(2 * cacheTTL)
+	_, err = c.Insert(nextRequest())
+	require.NoError(t, err)
+	assertCacheSize(t, c, 1)
+}
+
+func newTestCache() (*requestCache, *testingclock.FakeClock) {
+	c := newRequestCache()
+	fakeClock := testingclock.NewFakeClock(time.Now())
+	c.clock = fakeClock
+	return c, fakeClock
+}
+
+func assertCacheSize(t *testing.T, cache *requestCache, expectedSize int) {
+	tokenLen := len(cache.tokens)
+	llLen := cache.ll.Len()
+	assert.Equal(t, tokenLen, llLen, "inconsistent cache size! len(tokens)=%d; len(ll)=%d", tokenLen, llLen)
+	assert.Equal(t, expectedSize, tokenLen, "unexpected cache size!")
+}
+
+var requestUID = 0
+
+func nextRequest() interface{} {
+	requestUID++
+	return requestUID
+}

pkg/cri/streaming/server.go

@@ -0,0 +1,383 @@
+/*
+Copyright 2016 The Kubernetes 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 streaming
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	restful "github.com/emicklei/go-restful/v3"
+
+	"k8s.io/apimachinery/pkg/types"
+	remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
+	"k8s.io/client-go/tools/remotecommand"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
+	"k8s.io/kubelet/pkg/cri/streaming/portforward"
+	remotecommandserver "k8s.io/kubelet/pkg/cri/streaming/remotecommand"
+)
+
+// Server is the library interface to serve the stream requests.
+type Server interface {
+	http.Handler
+
+	// Get the serving URL for the requests.
+	// Requests must not be nil. Responses may be nil iff an error is returned.
+	GetExec(*runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error)
+	GetAttach(req *runtimeapi.AttachRequest) (*runtimeapi.AttachResponse, error)
+	GetPortForward(*runtimeapi.PortForwardRequest) (*runtimeapi.PortForwardResponse, error)
+
+	// Start the server.
+	// addr is the address to serve on (address:port) stayUp indicates whether the server should
+	// listen until Stop() is called, or automatically stop after all expected connections are
+	// closed. Calling Get{Exec,Attach,PortForward} increments the expected connection count.
+	// Function does not return until the server is stopped.
+	Start(stayUp bool) error
+	// Stop the server, and terminate any open connections.
+	Stop() error
+}
+
+// Runtime is the interface to execute the commands and provide the streams.
+type Runtime interface {
+	Exec(ctx context.Context, containerID string, cmd []string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error
+	Attach(ctx context.Context, containerID string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error
+	PortForward(ctx context.Context, podSandboxID string, port int32, stream io.ReadWriteCloser) error
+}
+
+// Config defines the options used for running the stream server.
+type Config struct {
+	// The host:port address the server will listen on.
+	Addr string
+	// The optional base URL for constructing streaming URLs. If empty, the baseURL will be
+	// constructed from the serve address.
+	// Note that for port "0", the URL port will be set to actual port in use.
+	BaseURL *url.URL
+
+	// How long to leave idle connections open for.
+	StreamIdleTimeout time.Duration
+	// How long to wait for clients to create streams. Only used for SPDY streaming.
+	StreamCreationTimeout time.Duration
+
+	// The streaming protocols the server supports (understands and permits).  See
+	// k8s.io/kubernetes/pkg/kubelet/server/remotecommand/constants.go for available protocols.
+	// Only used for SPDY streaming.
+	SupportedRemoteCommandProtocols []string
+
+	// The streaming protocols the server supports (understands and permits).  See
+	// k8s.io/kubernetes/pkg/kubelet/server/portforward/constants.go for available protocols.
+	// Only used for SPDY streaming.
+	SupportedPortForwardProtocols []string
+
+	// The config for serving over TLS. If nil, TLS will not be used.
+	TLSConfig *tls.Config
+}
+
+// DefaultConfig provides default values for server Config. The DefaultConfig is partial, so
+// some fields like Addr must still be provided.
+var DefaultConfig = Config{
+	StreamIdleTimeout:               4 * time.Hour,
+	StreamCreationTimeout:           remotecommandconsts.DefaultStreamCreationTimeout,
+	SupportedRemoteCommandProtocols: remotecommandconsts.SupportedStreamingProtocols,
+	SupportedPortForwardProtocols:   portforward.SupportedProtocols,
+}
+
+// NewServer creates a new Server for stream requests.
+// TODO(tallclair): Add auth(n/z) interface & handling.
+func NewServer(config Config, runtime Runtime) (Server, error) {
+	s := &server{
+		config:  config,
+		runtime: &criAdapter{runtime},
+		cache:   newRequestCache(),
+	}
+
+	if s.config.BaseURL == nil {
+		s.config.BaseURL = &url.URL{
+			Scheme: "http",
+			Host:   s.config.Addr,
+		}
+		if s.config.TLSConfig != nil {
+			s.config.BaseURL.Scheme = "https"
+		}
+	}
+
+	ws := &restful.WebService{}
+	endpoints := []struct {
+		path    string
+		handler restful.RouteFunction
+	}{
+		{"/exec/{token}", s.serveExec},
+		{"/attach/{token}", s.serveAttach},
+		{"/portforward/{token}", s.servePortForward},
+	}
+	// If serving relative to a base path, set that here.
+	pathPrefix := path.Dir(s.config.BaseURL.Path)
+	for _, e := range endpoints {
+		for _, method := range []string{"GET", "POST"} {
+			ws.Route(ws.
+				Method(method).
+				Path(path.Join(pathPrefix, e.path)).
+				To(e.handler))
+		}
+	}
+	handler := restful.NewContainer()
+	handler.Add(ws)
+	s.handler = handler
+	s.server = &http.Server{
+		Addr:      s.config.Addr,
+		Handler:   s.handler,
+		TLSConfig: s.config.TLSConfig,
+	}
+
+	return s, nil
+}
+
+type server struct {
+	config  Config
+	runtime *criAdapter
+	handler http.Handler
+	cache   *requestCache
+	server  *http.Server
+}
+
+func validateExecRequest(req *runtimeapi.ExecRequest) error {
+	if req.ContainerId == "" {
+		return status.Errorf(codes.InvalidArgument, "missing required container_id")
+	}
+	if req.Tty && req.Stderr {
+		// If TTY is set, stderr cannot be true because multiplexing is not
+		// supported.
+		return status.Errorf(codes.InvalidArgument, "tty and stderr cannot both be true")
+	}
+	if !req.Stdin && !req.Stdout && !req.Stderr {
+		return status.Errorf(codes.InvalidArgument, "one of stdin, stdout, or stderr must be set")
+	}
+	return nil
+}
+
+func (s *server) GetExec(req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
+	if err := validateExecRequest(req); err != nil {
+		return nil, err
+	}
+	token, err := s.cache.Insert(req)
+	if err != nil {
+		return nil, err
+	}
+	return &runtimeapi.ExecResponse{
+		Url: s.buildURL("exec", token),
+	}, nil
+}
+
+func validateAttachRequest(req *runtimeapi.AttachRequest) error {
+	if req.ContainerId == "" {
+		return status.Errorf(codes.InvalidArgument, "missing required container_id")
+	}
+	if req.Tty && req.Stderr {
+		// If TTY is set, stderr cannot be true because multiplexing is not
+		// supported.
+		return status.Errorf(codes.InvalidArgument, "tty and stderr cannot both be true")
+	}
+	if !req.Stdin && !req.Stdout && !req.Stderr {
+		return status.Errorf(codes.InvalidArgument, "one of stdin, stdout, and stderr must be set")
+	}
+	return nil
+}
+
+func (s *server) GetAttach(req *runtimeapi.AttachRequest) (*runtimeapi.AttachResponse, error) {
+	if err := validateAttachRequest(req); err != nil {
+		return nil, err
+	}
+	token, err := s.cache.Insert(req)
+	if err != nil {
+		return nil, err
+	}
+	return &runtimeapi.AttachResponse{
+		Url: s.buildURL("attach", token),
+	}, nil
+}
+
+func (s *server) GetPortForward(req *runtimeapi.PortForwardRequest) (*runtimeapi.PortForwardResponse, error) {
+	if req.PodSandboxId == "" {
+		return nil, status.Errorf(codes.InvalidArgument, "missing required pod_sandbox_id")
+	}
+	token, err := s.cache.Insert(req)
+	if err != nil {
+		return nil, err
+	}
+	return &runtimeapi.PortForwardResponse{
+		Url: s.buildURL("portforward", token),
+	}, nil
+}
+
+func (s *server) Start(stayUp bool) error {
+	if !stayUp {
+		// TODO(tallclair): Implement this.
+		return errors.New("stayUp=false is not yet implemented")
+	}
+
+	listener, err := net.Listen("tcp", s.config.Addr)
+	if err != nil {
+		return err
+	}
+	// Use the actual address as baseURL host. This handles the "0" port case.
+	s.config.BaseURL.Host = listener.Addr().String()
+	if s.config.TLSConfig != nil {
+		return s.server.ServeTLS(listener, "", "") // Use certs from TLSConfig.
+	}
+	return s.server.Serve(listener)
+}
+
+func (s *server) Stop() error {
+	return s.server.Close()
+}
+
+func (s *server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	s.handler.ServeHTTP(w, r)
+}
+
+func (s *server) buildURL(method, token string) string {
+	return s.config.BaseURL.ResolveReference(&url.URL{
+		Path: path.Join(method, token),
+	}).String()
+}
+
+func (s *server) serveExec(req *restful.Request, resp *restful.Response) {
+	token := req.PathParameter("token")
+	cachedRequest, ok := s.cache.Consume(token)
+	if !ok {
+		http.NotFound(resp.ResponseWriter, req.Request)
+		return
+	}
+	exec, ok := cachedRequest.(*runtimeapi.ExecRequest)
+	if !ok {
+		http.NotFound(resp.ResponseWriter, req.Request)
+		return
+	}
+
+	streamOpts := &remotecommandserver.Options{
+		Stdin:  exec.Stdin,
+		Stdout: exec.Stdout,
+		Stderr: exec.Stderr,
+		TTY:    exec.Tty,
+	}
+
+	remotecommandserver.ServeExec(
+		resp.ResponseWriter,
+		req.Request,
+		s.runtime,
+		"", // unused: podName
+		"", // unusued: podUID
+		exec.ContainerId,
+		exec.Cmd,
+		streamOpts,
+		s.config.StreamIdleTimeout,
+		s.config.StreamCreationTimeout,
+		s.config.SupportedRemoteCommandProtocols)
+}
+
+func (s *server) serveAttach(req *restful.Request, resp *restful.Response) {
+	token := req.PathParameter("token")
+	cachedRequest, ok := s.cache.Consume(token)
+	if !ok {
+		http.NotFound(resp.ResponseWriter, req.Request)
+		return
+	}
+	attach, ok := cachedRequest.(*runtimeapi.AttachRequest)
+	if !ok {
+		http.NotFound(resp.ResponseWriter, req.Request)
+		return
+	}
+
+	streamOpts := &remotecommandserver.Options{
+		Stdin:  attach.Stdin,
+		Stdout: attach.Stdout,
+		Stderr: attach.Stderr,
+		TTY:    attach.Tty,
+	}
+	remotecommandserver.ServeAttach(
+		resp.ResponseWriter,
+		req.Request,
+		s.runtime,
+		"", // unused: podName
+		"", // unusued: podUID
+		attach.ContainerId,
+		streamOpts,
+		s.config.StreamIdleTimeout,
+		s.config.StreamCreationTimeout,
+		s.config.SupportedRemoteCommandProtocols)
+}
+
+func (s *server) servePortForward(req *restful.Request, resp *restful.Response) {
+	token := req.PathParameter("token")
+	cachedRequest, ok := s.cache.Consume(token)
+	if !ok {
+		http.NotFound(resp.ResponseWriter, req.Request)
+		return
+	}
+	pf, ok := cachedRequest.(*runtimeapi.PortForwardRequest)
+	if !ok {
+		http.NotFound(resp.ResponseWriter, req.Request)
+		return
+	}
+
+	portForwardOptions, err := portforward.BuildV4Options(pf.Port)
+	if err != nil {
+		resp.WriteError(http.StatusBadRequest, err)
+		return
+	}
+
+	portforward.ServePortForward(
+		resp.ResponseWriter,
+		req.Request,
+		s.runtime,
+		pf.PodSandboxId,
+		"", // unused: podUID
+		portForwardOptions,
+		s.config.StreamIdleTimeout,
+		s.config.StreamCreationTimeout,
+		s.config.SupportedPortForwardProtocols)
+}
+
+// criAdapter wraps the Runtime functions to conform to the remotecommand interfaces.
+// The adapter binds the container ID to the container name argument, and the pod sandbox ID to the pod name.
+type criAdapter struct {
+	Runtime
+}
+
+var _ remotecommandserver.Executor = &criAdapter{}
+var _ remotecommandserver.Attacher = &criAdapter{}
+var _ portforward.PortForwarder = &criAdapter{}
+
+func (a *criAdapter) ExecInContainer(ctx context.Context, podName string, podUID types.UID, container string, cmd []string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error {
+	return a.Runtime.Exec(ctx, container, cmd, in, out, err, tty, resize)
+}
+
+func (a *criAdapter) AttachContainer(ctx context.Context, podName string, podUID types.UID, container string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error {
+	return a.Runtime.Attach(ctx, container, in, out, err, tty, resize)
+}
+
+func (a *criAdapter) PortForward(ctx context.Context, podName string, podUID types.UID, port int32, stream io.ReadWriteCloser) error {
+	return a.Runtime.PortForward(ctx, podName, port, stream)
+}

pkg/cri/streaming/server_test.go

@@ -0,0 +1,469 @@
+/*
+Copyright 2016 The Kubernetes 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 streaming
+
+import (
+	"context"
+	"crypto/tls"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	api "k8s.io/api/core/v1"
+	restclient "k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/remotecommand"
+	"k8s.io/client-go/transport/spdy"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
+	kubeletportforward "k8s.io/kubelet/pkg/cri/streaming/portforward"
+)
+
+const (
+	testAddr         = "localhost:12345"
+	testContainerID  = "container789"
+	testPodSandboxID = "pod0987"
+)
+
+func TestGetExec(t *testing.T) {
+	serv, err := NewServer(Config{
+		Addr: testAddr,
+	}, nil)
+	assert.NoError(t, err)
+
+	tlsServer, err := NewServer(Config{
+		Addr:      testAddr,
+		TLSConfig: &tls.Config{},
+	}, nil)
+	assert.NoError(t, err)
+
+	const pathPrefix = "cri/shim"
+	prefixServer, err := NewServer(Config{
+		Addr: testAddr,
+		BaseURL: &url.URL{
+			Scheme: "http",
+			Host:   testAddr,
+			Path:   "/" + pathPrefix + "/",
+		},
+	}, nil)
+	assert.NoError(t, err)
+
+	assertRequestToken := func(expectedReq *runtimeapi.ExecRequest, cache *requestCache, token string) {
+		req, ok := cache.Consume(token)
+		require.True(t, ok, "token %s not found!", token)
+		assert.Equal(t, expectedReq, req)
+	}
+	request := &runtimeapi.ExecRequest{
+		ContainerId: testContainerID,
+		Cmd:         []string{"echo", "foo"},
+		Tty:         true,
+		Stdin:       true,
+	}
+	{ // Non-TLS
+		resp, err := serv.GetExec(request)
+		assert.NoError(t, err)
+		expectedURL := "http://" + testAddr + "/exec/"
+		assert.Contains(t, resp.Url, expectedURL)
+		token := strings.TrimPrefix(resp.Url, expectedURL)
+		assertRequestToken(request, serv.(*server).cache, token)
+	}
+
+	{ // TLS
+		resp, err := tlsServer.GetExec(request)
+		assert.NoError(t, err)
+		expectedURL := "https://" + testAddr + "/exec/"
+		assert.Contains(t, resp.Url, expectedURL)
+		token := strings.TrimPrefix(resp.Url, expectedURL)
+		assertRequestToken(request, tlsServer.(*server).cache, token)
+	}
+
+	{ // Path prefix
+		resp, err := prefixServer.GetExec(request)
+		assert.NoError(t, err)
+		expectedURL := "http://" + testAddr + "/" + pathPrefix + "/exec/"
+		assert.Contains(t, resp.Url, expectedURL)
+		token := strings.TrimPrefix(resp.Url, expectedURL)
+		assertRequestToken(request, prefixServer.(*server).cache, token)
+	}
+}
+
+func TestValidateExecAttachRequest(t *testing.T) {
+	type config struct {
+		tty    bool
+		stdin  bool
+		stdout bool
+		stderr bool
+	}
+	for _, tc := range []struct {
+		desc      string
+		configs   []config
+		expectErr bool
+	}{
+		{
+			desc:      "at least one stream must be true",
+			expectErr: true,
+			configs: []config{
+				{false, false, false, false},
+				{true, false, false, false}},
+		},
+		{
+			desc:      "tty and stderr cannot both be true",
+			expectErr: true,
+			configs: []config{
+				{true, false, false, true},
+				{true, false, true, true},
+				{true, true, false, true},
+				{true, true, true, true},
+			},
+		},
+		{
+			desc:      "a valid config should pass",
+			expectErr: false,
+			configs: []config{
+				{false, false, false, true},
+				{false, false, true, false},
+				{false, false, true, true},
+				{false, true, false, false},
+				{false, true, false, true},
+				{false, true, true, false},
+				{false, true, true, true},
+				{true, false, true, false},
+				{true, true, false, false},
+				{true, true, true, false},
+			},
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			for _, c := range tc.configs {
+				// validate the exec request.
+				execReq := &runtimeapi.ExecRequest{
+					ContainerId: testContainerID,
+					Cmd:         []string{"date"},
+					Tty:         c.tty,
+					Stdin:       c.stdin,
+					Stdout:      c.stdout,
+					Stderr:      c.stderr,
+				}
+				err := validateExecRequest(execReq)
+				assert.Equal(t, tc.expectErr, err != nil, "config: %v,  err: %v", c, err)
+
+				// validate the attach request.
+				attachReq := &runtimeapi.AttachRequest{
+					ContainerId: testContainerID,
+					Tty:         c.tty,
+					Stdin:       c.stdin,
+					Stdout:      c.stdout,
+					Stderr:      c.stderr,
+				}
+				err = validateAttachRequest(attachReq)
+				assert.Equal(t, tc.expectErr, err != nil, "config: %v, err: %v", c, err)
+			}
+		})
+	}
+}
+
+func TestGetAttach(t *testing.T) {
+	serv, err := NewServer(Config{
+		Addr: testAddr,
+	}, nil)
+	require.NoError(t, err)
+
+	tlsServer, err := NewServer(Config{
+		Addr:      testAddr,
+		TLSConfig: &tls.Config{},
+	}, nil)
+	require.NoError(t, err)
+
+	assertRequestToken := func(expectedReq *runtimeapi.AttachRequest, cache *requestCache, token string) {
+		req, ok := cache.Consume(token)
+		require.True(t, ok, "token %s not found!", token)
+		assert.Equal(t, expectedReq, req)
+	}
+
+	request := &runtimeapi.AttachRequest{
+		ContainerId: testContainerID,
+		Stdin:       true,
+		Tty:         true,
+	}
+	{ // Non-TLS
+		resp, err := serv.GetAttach(request)
+		assert.NoError(t, err)
+		expectedURL := "http://" + testAddr + "/attach/"
+		assert.Contains(t, resp.Url, expectedURL)
+		token := strings.TrimPrefix(resp.Url, expectedURL)
+		assertRequestToken(request, serv.(*server).cache, token)
+	}
+
+	{ // TLS
+		resp, err := tlsServer.GetAttach(request)
+		assert.NoError(t, err)
+		expectedURL := "https://" + testAddr + "/attach/"
+		assert.Contains(t, resp.Url, expectedURL)
+		token := strings.TrimPrefix(resp.Url, expectedURL)
+		assertRequestToken(request, tlsServer.(*server).cache, token)
+	}
+}
+
+func TestGetPortForward(t *testing.T) {
+	podSandboxID := testPodSandboxID
+	request := &runtimeapi.PortForwardRequest{
+		PodSandboxId: podSandboxID,
+		Port:         []int32{1, 2, 3, 4},
+	}
+
+	{ // Non-TLS
+		serv, err := NewServer(Config{
+			Addr: testAddr,
+		}, nil)
+		assert.NoError(t, err)
+		resp, err := serv.GetPortForward(request)
+		assert.NoError(t, err)
+		expectedURL := "http://" + testAddr + "/portforward/"
+		assert.True(t, strings.HasPrefix(resp.Url, expectedURL))
+		token := strings.TrimPrefix(resp.Url, expectedURL)
+		req, ok := serv.(*server).cache.Consume(token)
+		require.True(t, ok, "token %s not found!", token)
+		assert.Equal(t, testPodSandboxID, req.(*runtimeapi.PortForwardRequest).PodSandboxId)
+	}
+
+	{ // TLS
+		tlsServer, err := NewServer(Config{
+			Addr:      testAddr,
+			TLSConfig: &tls.Config{},
+		}, nil)
+		assert.NoError(t, err)
+		resp, err := tlsServer.GetPortForward(request)
+		assert.NoError(t, err)
+		expectedURL := "https://" + testAddr + "/portforward/"
+		assert.True(t, strings.HasPrefix(resp.Url, expectedURL))
+		token := strings.TrimPrefix(resp.Url, expectedURL)
+		req, ok := tlsServer.(*server).cache.Consume(token)
+		require.True(t, ok, "token %s not found!", token)
+		assert.Equal(t, testPodSandboxID, req.(*runtimeapi.PortForwardRequest).PodSandboxId)
+	}
+}
+
+func TestServeExec(t *testing.T) {
+	runRemoteCommandTest(t, "exec")
+}
+
+func TestServeAttach(t *testing.T) {
+	runRemoteCommandTest(t, "attach")
+}
+
+func TestServePortForward(t *testing.T) {
+	s, testServer := startTestServer(t)
+	defer testServer.Close()
+
+	resp, err := s.GetPortForward(&runtimeapi.PortForwardRequest{
+		PodSandboxId: testPodSandboxID,
+	})
+	require.NoError(t, err)
+	reqURL, err := url.Parse(resp.Url)
+	require.NoError(t, err)
+
+	transport, upgrader, err := spdy.RoundTripperFor(&restclient.Config{})
+	require.NoError(t, err)
+	dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", reqURL)
+	streamConn, _, err := dialer.Dial(kubeletportforward.ProtocolV1Name)
+	require.NoError(t, err)
+	defer streamConn.Close()
+
+	// Create the streams.
+	headers := http.Header{}
+	// Error stream is required, but unused in this test.
+	headers.Set(api.StreamType, api.StreamTypeError)
+	headers.Set(api.PortHeader, strconv.Itoa(testPort))
+	_, err = streamConn.CreateStream(headers)
+	require.NoError(t, err)
+	// Setup the data stream.
+	headers.Set(api.StreamType, api.StreamTypeData)
+	headers.Set(api.PortHeader, strconv.Itoa(testPort))
+	stream, err := streamConn.CreateStream(headers)
+	require.NoError(t, err)
+
+	doClientStreams(t, "portforward", stream, stream, nil)
+}
+
+// Run the remote command test.
+// commandType is either "exec" or "attach".
+func runRemoteCommandTest(t *testing.T, commandType string) {
+	s, testServer := startTestServer(t)
+	defer testServer.Close()
+
+	var reqURL *url.URL
+	stdin, stdout, stderr := true, true, true
+	containerID := testContainerID
+	switch commandType {
+	case "exec":
+		resp, err := s.GetExec(&runtimeapi.ExecRequest{
+			ContainerId: containerID,
+			Cmd:         []string{"echo"},
+			Stdin:       stdin,
+			Stdout:      stdout,
+			Stderr:      stderr,
+		})
+		require.NoError(t, err)
+		reqURL, err = url.Parse(resp.Url)
+		require.NoError(t, err)
+	case "attach":
+		resp, err := s.GetAttach(&runtimeapi.AttachRequest{
+			ContainerId: containerID,
+			Stdin:       stdin,
+			Stdout:      stdout,
+			Stderr:      stderr,
+		})
+		require.NoError(t, err)
+		reqURL, err = url.Parse(resp.Url)
+		require.NoError(t, err)
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	stdinR, stdinW := io.Pipe()
+	stdoutR, stdoutW := io.Pipe()
+	stderrR, stderrW := io.Pipe()
+
+	go func() {
+		defer wg.Done()
+		exec, err := remotecommand.NewSPDYExecutor(&restclient.Config{}, "POST", reqURL)
+		require.NoError(t, err)
+
+		opts := remotecommand.StreamOptions{
+			Stdin:  stdinR,
+			Stdout: stdoutW,
+			Stderr: stderrW,
+			Tty:    false,
+		}
+		require.NoError(t, exec.StreamWithContext(context.Background(), opts))
+	}()
+
+	go func() {
+		defer wg.Done()
+		doClientStreams(t, commandType, stdinW, stdoutR, stderrR)
+	}()
+
+	wg.Wait()
+
+	// Repeat request with the same URL should be a 404.
+	resp, err := http.Get(reqURL.String())
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusNotFound, resp.StatusCode)
+}
+
+func startTestServer(t *testing.T) (Server, *httptest.Server) {
+	var s Server
+	testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		s.ServeHTTP(w, r)
+	}))
+	cleanup := true
+	defer func() {
+		if cleanup {
+			testServer.Close()
+		}
+	}()
+
+	testURL, err := url.Parse(testServer.URL)
+	require.NoError(t, err)
+
+	rt := newFakeRuntime(t)
+	config := DefaultConfig
+	config.BaseURL = testURL
+	s, err = NewServer(config, rt)
+	require.NoError(t, err)
+
+	cleanup = false // Caller must close the test server.
+	return s, testServer
+}
+
+const (
+	testInput  = "abcdefg"
+	testOutput = "fooBARbaz"
+	testErr    = "ERROR!!!"
+	testPort   = 12345
+)
+
+func newFakeRuntime(t *testing.T) *fakeRuntime {
+	return &fakeRuntime{
+		t: t,
+	}
+}
+
+type fakeRuntime struct {
+	t *testing.T
+}
+
+func (f *fakeRuntime) Exec(_ context.Context, containerID string, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error {
+	assert.Equal(f.t, testContainerID, containerID)
+	doServerStreams(f.t, "exec", stdin, stdout, stderr)
+	return nil
+}
+
+func (f *fakeRuntime) Attach(_ context.Context, containerID string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error {
+	assert.Equal(f.t, testContainerID, containerID)
+	doServerStreams(f.t, "attach", stdin, stdout, stderr)
+	return nil
+}
+
+func (f *fakeRuntime) PortForward(_ context.Context, podSandboxID string, port int32, stream io.ReadWriteCloser) error {
+	assert.Equal(f.t, testPodSandboxID, podSandboxID)
+	assert.EqualValues(f.t, testPort, port)
+	doServerStreams(f.t, "portforward", stream, stream, nil)
+	return nil
+}
+
+// Send & receive expected input/output. Must be the inverse of doClientStreams.
+// Function will block until the expected i/o is finished.
+func doServerStreams(t *testing.T, prefix string, stdin io.Reader, stdout, stderr io.Writer) {
+	if stderr != nil {
+		writeExpected(t, "server stderr", stderr, prefix+testErr)
+	}
+	readExpected(t, "server stdin", stdin, prefix+testInput)
+	writeExpected(t, "server stdout", stdout, prefix+testOutput)
+}
+
+// Send & receive expected input/output. Must be the inverse of doServerStreams.
+// Function will block until the expected i/o is finished.
+func doClientStreams(t *testing.T, prefix string, stdin io.Writer, stdout, stderr io.Reader) {
+	if stderr != nil {
+		readExpected(t, "client stderr", stderr, prefix+testErr)
+	}
+	writeExpected(t, "client stdin", stdin, prefix+testInput)
+	readExpected(t, "client stdout", stdout, prefix+testOutput)
+}
+
+// Read and verify the expected string from the stream.
+func readExpected(t *testing.T, streamName string, r io.Reader, expected string) {
+	result := make([]byte, len(expected))
+	_, err := io.ReadAtLeast(r, result, len(expected))
+	assert.NoError(t, err, "stream %s", streamName)
+	assert.Equal(t, expected, string(result), "stream %s", streamName)
+}
+
+// Write and verify success of the data over the stream.
+func writeExpected(t *testing.T, streamName string, w io.Writer, data string) {
+	n, err := io.WriteString(w, data)
+	assert.NoError(t, err, "stream %s", streamName)
+	assert.Equal(t, len(data), n, "stream %s", streamName)
+}