mirror of
https://github.com/kubernetes/client-go.git
synced 2026-06-20 16:55:32 +00:00
Compare commits
1 Commits
kubernetes
...
v0.25.1-rc
| Author | SHA1 | Date | |
|---|---|---|---|
|
d383142160 |
20
go.mod
20
go.mod
@@ -13,21 +13,21 @@ require (
|
||||
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
|
||||
github.com/golang/protobuf v1.5.2
|
||||
github.com/google/gnostic v0.5.7-v3refs
|
||||
github.com/google/go-cmp v0.5.8
|
||||
github.com/google/go-cmp v0.5.6
|
||||
github.com/google/gofuzz v1.1.0
|
||||
github.com/google/uuid v1.1.2
|
||||
github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7
|
||||
github.com/imdario/mergo v0.3.6
|
||||
github.com/peterbourgon/diskv v2.0.1+incompatible
|
||||
github.com/spf13/pflag v1.0.5
|
||||
github.com/stretchr/testify v1.8.0
|
||||
golang.org/x/net v0.8.0
|
||||
github.com/stretchr/testify v1.7.0
|
||||
golang.org/x/net v0.0.0-20220722155237-a158d28d115b
|
||||
golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8
|
||||
golang.org/x/term v0.6.0
|
||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
|
||||
golang.org/x/time v0.0.0-20220210224613-90d013bbcef8
|
||||
google.golang.org/protobuf v1.28.0
|
||||
k8s.io/api v0.25.10
|
||||
k8s.io/apimachinery v0.25.10
|
||||
k8s.io/api v0.25.1-rc.0
|
||||
k8s.io/apimachinery v0.25.1-rc.0
|
||||
k8s.io/klog/v2 v2.70.1
|
||||
k8s.io/kube-openapi v0.0.0-20220803162953-67bda5d908f1
|
||||
k8s.io/utils v0.0.0-20220728103510-ee6ede2d64ed
|
||||
@@ -60,8 +60,8 @@ require (
|
||||
github.com/pkg/errors v0.9.1 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd // indirect
|
||||
golang.org/x/sys v0.6.0 // indirect
|
||||
golang.org/x/text v0.8.0 // indirect
|
||||
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f // indirect
|
||||
golang.org/x/text v0.3.7 // indirect
|
||||
google.golang.org/appengine v1.6.7 // indirect
|
||||
gopkg.in/inf.v0 v0.9.1 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
@@ -70,6 +70,6 @@ require (
|
||||
)
|
||||
|
||||
replace (
|
||||
k8s.io/api => k8s.io/api v0.25.10
|
||||
k8s.io/apimachinery => k8s.io/apimachinery v0.25.10
|
||||
k8s.io/api => k8s.io/api v0.25.1-rc.0
|
||||
k8s.io/apimachinery => k8s.io/apimachinery v0.25.1-rc.0
|
||||
)
|
||||
|
||||
38
go.sum
38
go.sum
@@ -169,9 +169,8 @@ github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
|
||||
github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
|
||||
github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
|
||||
github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
|
||||
github.com/google/gofuzz v1.1.0 h1:Hsa8mG0dQ46ij8Sl2AYJDUv1oA9/d6Vk+3LG99Oe02g=
|
||||
github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
|
||||
@@ -238,8 +237,8 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
|
||||
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
|
||||
github.com/onsi/ginkgo/v2 v2.1.6 h1:Fx2POJZfKRQcM1pH49qSZiYeu319wji004qX+GDovrU=
|
||||
github.com/onsi/gomega v1.20.1 h1:PA/3qinGoukvymdIDV8pii6tiZgC8kbmJO6Z5+b002Q=
|
||||
github.com/onsi/ginkgo/v2 v2.1.4 h1:GNapqRSid3zijZ9H77KrgVG4/8KqiyRsxcSxe+7ApXY=
|
||||
github.com/onsi/gomega v1.19.0 h1:4ieX6qQjPP/BfC3mpsAtIGGlxTWPeA3Inl/7DtXw1tw=
|
||||
github.com/peterbourgon/diskv v2.0.1+incompatible h1:UBdAOUP5p4RWqPBg048CAvpKN+vxiaj6gdUUzhl4XmI=
|
||||
github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
@@ -255,14 +254,12 @@ 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/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag1KpM8ahLw8=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
|
||||
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.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
|
||||
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
|
||||
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
|
||||
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
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=
|
||||
@@ -357,8 +354,8 @@ golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLd
|
||||
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
|
||||
golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
|
||||
golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
|
||||
golang.org/x/net v0.0.0-20220722155237-a158d28d115b h1:PxfKdU9lEEDYjdIzOtC4qFWgkU2rGHdKlKowJSMN9h0=
|
||||
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
|
||||
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
|
||||
golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
|
||||
golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
|
||||
@@ -433,11 +430,11 @@ golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBc
|
||||
golang.org/x/sys v0.0.0-20210806184541-e5e7981a1069/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f h1:v4INt8xihDGvnrfjMDVXGxw9wrfxYyCjk0KbXjhR55s=
|
||||
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/term v0.6.0 h1:clScbb1cHjoCkyRbWwBEUZ5H/tIFu5TAXIqaZD0Gcjw=
|
||||
golang.org/x/term v0.6.0/go.mod h1:m6U89DPEgQRMq3DNkDClhWw02AUbt2daBVO4cn4Hv9U=
|
||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 h1:JGgROgKl9N8DuW20oFS5gxc+lE67/N3FcwmBPMe7ArY=
|
||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
|
||||
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=
|
||||
@@ -446,8 +443,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68=
|
||||
golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
|
||||
golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
|
||||
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
|
||||
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=
|
||||
@@ -509,6 +506,7 @@ golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
|
||||
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=
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
|
||||
google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
|
||||
@@ -667,10 +665,10 @@ 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.25.10 h1:YfcmWMKDnWpzKV2byP+fu0v00yNTS4+cqw4g0ndUsJA=
|
||||
k8s.io/api v0.25.10/go.mod h1:7inWacs1rgsi5uLOONfUmo4on+tVkkuJZNsMLouGAhA=
|
||||
k8s.io/apimachinery v0.25.10 h1:uvPXar0BVg9g2R5a5kTjMuHCjLxC5LiAclSrLOP8Q20=
|
||||
k8s.io/apimachinery v0.25.10/go.mod h1:PJ+6cm50BMETqCCJx1RXQIXaq937SUdAq2vVKCGDZXU=
|
||||
k8s.io/api v0.25.1-rc.0 h1:xzf2pEPM9qtnWBPug6T2UFFGrGcHJ6RrpFsvherz8sg=
|
||||
k8s.io/api v0.25.1-rc.0/go.mod h1:YIpLmti+R/WJ+xTEzpxoG8lWs8IEuXi8uU1m6ROjjy0=
|
||||
k8s.io/apimachinery v0.25.1-rc.0 h1:hktbPYsiFe7tannlmz3LJIHzswKxNgkd2KUEnqF8Ctw=
|
||||
k8s.io/apimachinery v0.25.1-rc.0/go.mod h1:qMx9eAk0sZQGsXGu86fab8tZdffHbwUfsvzqKn4mfB0=
|
||||
k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
|
||||
k8s.io/klog/v2 v2.70.1 h1:7aaoSdahviPmR+XkS7FyxlkkXs6tHISSG03RxleQAVQ=
|
||||
k8s.io/klog/v2 v2.70.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
|
||||
|
||||
@@ -199,18 +199,14 @@ func newAuthenticator(c *cache, isTerminalFunc func(int) bool, config *api.ExecC
|
||||
now: time.Now,
|
||||
environ: os.Environ,
|
||||
|
||||
connTracker: connTracker,
|
||||
defaultDialer: defaultDialer,
|
||||
connTracker: connTracker,
|
||||
}
|
||||
|
||||
for _, env := range config.Env {
|
||||
a.env = append(a.env, env.Name+"="+env.Value)
|
||||
}
|
||||
|
||||
// these functions are made comparable and stored in the cache so that repeated clientset
|
||||
// construction with the same rest.Config results in a single TLS cache and Authenticator
|
||||
a.getCert = &transport.GetCertHolder{GetCert: a.cert}
|
||||
a.dial = &transport.DialHolder{Dial: defaultDialer.DialContext}
|
||||
|
||||
return c.put(key, a), nil
|
||||
}
|
||||
|
||||
@@ -265,6 +261,8 @@ type Authenticator struct {
|
||||
now func() time.Time
|
||||
environ func() []string
|
||||
|
||||
// defaultDialer is used for clients which don't specify a custom dialer
|
||||
defaultDialer *connrotation.Dialer
|
||||
// connTracker tracks all connections opened that we need to close when rotating a client certificate
|
||||
connTracker *connrotation.ConnectionTracker
|
||||
|
||||
@@ -275,12 +273,6 @@ type Authenticator struct {
|
||||
mu sync.Mutex
|
||||
cachedCreds *credentials
|
||||
exp time.Time
|
||||
|
||||
// getCert makes Authenticator.cert comparable to support TLS config caching
|
||||
getCert *transport.GetCertHolder
|
||||
// dial is used for clients which do not specify a custom dialer
|
||||
// it is comparable to support TLS config caching
|
||||
dial *transport.DialHolder
|
||||
}
|
||||
|
||||
type credentials struct {
|
||||
@@ -308,20 +300,18 @@ func (a *Authenticator) UpdateTransportConfig(c *transport.Config) error {
|
||||
if c.HasCertCallback() {
|
||||
return errors.New("can't add TLS certificate callback: transport.Config.TLS.GetCert already set")
|
||||
}
|
||||
c.TLS.GetCert = a.getCert.GetCert
|
||||
c.TLS.GetCertHolder = a.getCert // comparable for TLS config caching
|
||||
c.TLS.GetCert = a.cert
|
||||
|
||||
var d *connrotation.Dialer
|
||||
if c.Dial != nil {
|
||||
// if c has a custom dialer, we have to wrap it
|
||||
// TLS config caching is not supported for this config
|
||||
d := connrotation.NewDialerWithTracker(c.Dial, a.connTracker)
|
||||
c.Dial = d.DialContext
|
||||
c.DialHolder = nil
|
||||
d = connrotation.NewDialerWithTracker(c.Dial, a.connTracker)
|
||||
} else {
|
||||
c.Dial = a.dial.Dial
|
||||
c.DialHolder = a.dial // comparable for TLS config caching
|
||||
d = a.defaultDialer
|
||||
}
|
||||
|
||||
c.Dial = d.DialContext
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,106 +0,0 @@
|
||||
/*
|
||||
Copyright 2022 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 exec_test // separate package to prevent circular import
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
utilnet "k8s.io/apimachinery/pkg/util/net"
|
||||
clientset "k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/rest"
|
||||
clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
|
||||
)
|
||||
|
||||
// TestExecTLSCache asserts the semantics of the TLS cache when exec auth is used.
|
||||
//
|
||||
// In particular, when:
|
||||
// - multiple identical rest configs exist as distinct objects, and
|
||||
// - these rest configs use exec auth, and
|
||||
// - these rest configs are used to create distinct clientsets, then
|
||||
//
|
||||
// the underlying TLS config is shared between those clientsets.
|
||||
func TestExecTLSCache(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
|
||||
config1 := &rest.Config{
|
||||
Host: "https://localhost",
|
||||
ExecProvider: &clientcmdapi.ExecConfig{
|
||||
Command: "./testdata/test-plugin.sh",
|
||||
APIVersion: "client.authentication.k8s.io/v1",
|
||||
InteractiveMode: clientcmdapi.IfAvailableExecInteractiveMode,
|
||||
},
|
||||
}
|
||||
client1 := clientset.NewForConfigOrDie(config1)
|
||||
|
||||
config2 := &rest.Config{
|
||||
Host: "https://localhost",
|
||||
ExecProvider: &clientcmdapi.ExecConfig{
|
||||
Command: "./testdata/test-plugin.sh",
|
||||
APIVersion: "client.authentication.k8s.io/v1",
|
||||
InteractiveMode: clientcmdapi.IfAvailableExecInteractiveMode,
|
||||
},
|
||||
}
|
||||
client2 := clientset.NewForConfigOrDie(config2)
|
||||
|
||||
config3 := &rest.Config{
|
||||
Host: "https://localhost",
|
||||
ExecProvider: &clientcmdapi.ExecConfig{
|
||||
Command: "./testdata/test-plugin.sh",
|
||||
Args: []string{"make this exec auth different"},
|
||||
APIVersion: "client.authentication.k8s.io/v1",
|
||||
InteractiveMode: clientcmdapi.IfAvailableExecInteractiveMode,
|
||||
},
|
||||
}
|
||||
client3 := clientset.NewForConfigOrDie(config3)
|
||||
|
||||
_, _ = client1.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
|
||||
_, _ = client2.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
|
||||
_, _ = client3.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
|
||||
|
||||
rt1 := client1.RESTClient().(*rest.RESTClient).Client.Transport
|
||||
rt2 := client2.RESTClient().(*rest.RESTClient).Client.Transport
|
||||
rt3 := client3.RESTClient().(*rest.RESTClient).Client.Transport
|
||||
|
||||
tlsConfig1, err := utilnet.TLSClientConfig(rt1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tlsConfig2, err := utilnet.TLSClientConfig(rt2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tlsConfig3, err := utilnet.TLSClientConfig(rt3)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if tlsConfig1 == nil || tlsConfig2 == nil || tlsConfig3 == nil {
|
||||
t.Fatal("expected non-nil TLS configs")
|
||||
}
|
||||
|
||||
if tlsConfig1 != tlsConfig2 {
|
||||
t.Fatal("expected the same TLS config for matching exec config via rest config")
|
||||
}
|
||||
|
||||
if tlsConfig1 == tlsConfig3 {
|
||||
t.Fatal("expected different TLS config for non-matching exec config via rest config")
|
||||
}
|
||||
}
|
||||
121
rest/request.go
121
rest/request.go
@@ -34,7 +34,6 @@ import (
|
||||
"time"
|
||||
|
||||
"golang.org/x/net/http2"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
@@ -117,11 +116,8 @@ type Request struct {
|
||||
subresource string
|
||||
|
||||
// output
|
||||
err error
|
||||
|
||||
// only one of body / bodyBytes may be set. requests using body are not retriable.
|
||||
body io.Reader
|
||||
bodyBytes []byte
|
||||
err error
|
||||
body io.Reader
|
||||
|
||||
retryFn requestRetryFunc
|
||||
}
|
||||
@@ -447,15 +443,12 @@ func (r *Request) Body(obj interface{}) *Request {
|
||||
return r
|
||||
}
|
||||
glogBody("Request Body", data)
|
||||
r.body = nil
|
||||
r.bodyBytes = data
|
||||
r.body = bytes.NewReader(data)
|
||||
case []byte:
|
||||
glogBody("Request Body", t)
|
||||
r.body = nil
|
||||
r.bodyBytes = t
|
||||
r.body = bytes.NewReader(t)
|
||||
case io.Reader:
|
||||
r.body = t
|
||||
r.bodyBytes = nil
|
||||
case runtime.Object:
|
||||
// callers may pass typed interface pointers, therefore we must check nil with reflection
|
||||
if reflect.ValueOf(t).IsNil() {
|
||||
@@ -472,8 +465,7 @@ func (r *Request) Body(obj interface{}) *Request {
|
||||
return r
|
||||
}
|
||||
glogBody("Request Body", data)
|
||||
r.body = nil
|
||||
r.bodyBytes = data
|
||||
r.body = bytes.NewReader(data)
|
||||
r.SetHeader("Content-Type", r.c.content.ContentType)
|
||||
default:
|
||||
r.err = fmt.Errorf("unknown type used for body: %+v", obj)
|
||||
@@ -516,87 +508,6 @@ func (r *Request) URL() *url.URL {
|
||||
return finalURL
|
||||
}
|
||||
|
||||
// finalURLTemplate is similar to URL(), but will make all specific parameter values equal
|
||||
// - instead of name or namespace, "{name}" and "{namespace}" will be used, and all query
|
||||
// parameters will be reset. This creates a copy of the url so as not to change the
|
||||
// underlying object.
|
||||
func (r Request) finalURLTemplate() url.URL {
|
||||
newParams := url.Values{}
|
||||
v := []string{"{value}"}
|
||||
for k := range r.params {
|
||||
newParams[k] = v
|
||||
}
|
||||
r.params = newParams
|
||||
u := r.URL()
|
||||
if u == nil {
|
||||
return url.URL{}
|
||||
}
|
||||
|
||||
segments := strings.Split(u.Path, "/")
|
||||
groupIndex := 0
|
||||
index := 0
|
||||
trimmedBasePath := ""
|
||||
if r.c.base != nil && strings.Contains(u.Path, r.c.base.Path) {
|
||||
p := strings.TrimPrefix(u.Path, r.c.base.Path)
|
||||
if !strings.HasPrefix(p, "/") {
|
||||
p = "/" + p
|
||||
}
|
||||
// store the base path that we have trimmed so we can append it
|
||||
// before returning the URL
|
||||
trimmedBasePath = r.c.base.Path
|
||||
segments = strings.Split(p, "/")
|
||||
groupIndex = 1
|
||||
}
|
||||
if len(segments) <= 2 {
|
||||
return *u
|
||||
}
|
||||
|
||||
const CoreGroupPrefix = "api"
|
||||
const NamedGroupPrefix = "apis"
|
||||
isCoreGroup := segments[groupIndex] == CoreGroupPrefix
|
||||
isNamedGroup := segments[groupIndex] == NamedGroupPrefix
|
||||
if isCoreGroup {
|
||||
// checking the case of core group with /api/v1/... format
|
||||
index = groupIndex + 2
|
||||
} else if isNamedGroup {
|
||||
// checking the case of named group with /apis/apps/v1/... format
|
||||
index = groupIndex + 3
|
||||
} else {
|
||||
// this should not happen that the only two possibilities are /api... and /apis..., just want to put an
|
||||
// outlet here in case more API groups are added in future if ever possible:
|
||||
// https://kubernetes.io/docs/concepts/overview/kubernetes-api/#api-groups
|
||||
// if a wrong API groups name is encountered, return the {prefix} for url.Path
|
||||
u.Path = "/{prefix}"
|
||||
u.RawQuery = ""
|
||||
return *u
|
||||
}
|
||||
// switch segLength := len(segments) - index; segLength {
|
||||
switch {
|
||||
// case len(segments) - index == 1:
|
||||
// resource (with no name) do nothing
|
||||
case len(segments)-index == 2:
|
||||
// /$RESOURCE/$NAME: replace $NAME with {name}
|
||||
segments[index+1] = "{name}"
|
||||
case len(segments)-index == 3:
|
||||
if segments[index+2] == "finalize" || segments[index+2] == "status" {
|
||||
// /$RESOURCE/$NAME/$SUBRESOURCE: replace $NAME with {name}
|
||||
segments[index+1] = "{name}"
|
||||
} else {
|
||||
// /namespace/$NAMESPACE/$RESOURCE: replace $NAMESPACE with {namespace}
|
||||
segments[index+1] = "{namespace}"
|
||||
}
|
||||
case len(segments)-index >= 4:
|
||||
segments[index+1] = "{namespace}"
|
||||
// /namespace/$NAMESPACE/$RESOURCE/$NAME: replace $NAMESPACE with {namespace}, $NAME with {name}
|
||||
if segments[index+3] != "finalize" && segments[index+3] != "status" {
|
||||
// /$RESOURCE/$NAME/$SUBRESOURCE: replace $NAME with {name}
|
||||
segments[index+3] = "{name}"
|
||||
}
|
||||
}
|
||||
u.Path = path.Join(trimmedBasePath, path.Join(segments...))
|
||||
return *u
|
||||
}
|
||||
|
||||
func (r *Request) tryThrottleWithInfo(ctx context.Context, retryInfo string) error {
|
||||
if r.rateLimiter == nil {
|
||||
return nil
|
||||
@@ -626,7 +537,7 @@ func (r *Request) tryThrottleWithInfo(ctx context.Context, retryInfo string) err
|
||||
// but we use a throttled logger to prevent spamming.
|
||||
globalThrottledLogger.Infof("%s", message)
|
||||
}
|
||||
metrics.RateLimiterLatency.Observe(ctx, r.verb, r.finalURLTemplate(), latency)
|
||||
metrics.RateLimiterLatency.Observe(ctx, r.verb, *r.URL(), latency)
|
||||
|
||||
return err
|
||||
}
|
||||
@@ -833,7 +744,9 @@ func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if r.body != nil {
|
||||
req.Body = ioutil.NopCloser(r.body)
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
updateURLMetrics(ctx, r, resp, err)
|
||||
retry.After(ctx, r, resp, err)
|
||||
@@ -895,20 +808,8 @@ func (r *Request) requestPreflightCheck() error {
|
||||
}
|
||||
|
||||
func (r *Request) newHTTPRequest(ctx context.Context) (*http.Request, error) {
|
||||
var body io.Reader
|
||||
switch {
|
||||
case r.body != nil && r.bodyBytes != nil:
|
||||
return nil, fmt.Errorf("cannot set both body and bodyBytes")
|
||||
case r.body != nil:
|
||||
body = r.body
|
||||
case r.bodyBytes != nil:
|
||||
// Create a new reader specifically for this request.
|
||||
// Giving each request a dedicated reader allows retries to avoid races resetting the request body.
|
||||
body = bytes.NewReader(r.bodyBytes)
|
||||
}
|
||||
|
||||
url := r.URL().String()
|
||||
req, err := http.NewRequest(r.verb, url, body)
|
||||
req, err := http.NewRequest(r.verb, url, r.body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -925,7 +826,7 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
|
||||
// Metrics for total request latency
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
metrics.RequestLatency.Observe(ctx, r.verb, r.finalURLTemplate(), time.Since(start))
|
||||
metrics.RequestLatency.Observe(ctx, r.verb, *r.URL(), time.Since(start))
|
||||
}()
|
||||
|
||||
if r.err != nil {
|
||||
|
||||
@@ -338,206 +338,6 @@ func TestResultIntoWithNoBodyReturnsErr(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestURLTemplate(t *testing.T) {
|
||||
uri, _ := url.Parse("http://localhost/some/base/url/path")
|
||||
uriSingleSlash, _ := url.Parse("http://localhost/")
|
||||
testCases := []struct {
|
||||
Request *Request
|
||||
ExpectedFullURL string
|
||||
ExpectedFinalURL string
|
||||
}{
|
||||
{
|
||||
// non dynamic client
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("POST").
|
||||
Prefix("api", "v1").Resource("r1").Namespace("ns").Name("nm").Param("p0", "v0"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/api/v1/namespaces/ns/r1/nm?p0=v0",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/api/v1/namespaces/%7Bnamespace%7D/r1/%7Bname%7D?p0=%7Bvalue%7D",
|
||||
},
|
||||
{
|
||||
// non dynamic client with wrong api group
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("POST").
|
||||
Prefix("pre1", "v1").Resource("r1").Namespace("ns").Name("nm").Param("p0", "v0"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/pre1/v1/namespaces/ns/r1/nm?p0=v0",
|
||||
ExpectedFinalURL: "http://localhost/%7Bprefix%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with core group + namespace + resourceResource (with name)
|
||||
// /api/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/api/v1/namespaces/ns/r1/name1"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/api/v1/namespaces/ns/r1/name1",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/api/v1/namespaces/%7Bnamespace%7D/r1/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + namespace + resourceResource (with name)
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/g1/v1/namespaces/ns/r1/name1"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/g1/v1/namespaces/ns/r1/name1",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/g1/v1/namespaces/%7Bnamespace%7D/r1/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with core group + namespace + resourceResource (with NO name)
|
||||
// /api/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/api/v1/namespaces/ns/r1"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/api/v1/namespaces/ns/r1",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/api/v1/namespaces/%7Bnamespace%7D/r1",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + namespace + resourceResource (with NO name)
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/g1/v1/namespaces/ns/r1"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/g1/v1/namespaces/ns/r1",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/g1/v1/namespaces/%7Bnamespace%7D/r1",
|
||||
},
|
||||
{
|
||||
// dynamic client with core group + resourceResource (with name)
|
||||
// /api/$RESOURCEVERSION/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/api/v1/r1/name1"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/api/v1/r1/name1",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/api/v1/r1/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + resourceResource (with name)
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/g1/v1/r1/name1"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/g1/v1/r1/name1",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/g1/v1/r1/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + namespace + resourceResource (with name) + subresource
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME/$SUBRESOURCE
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces/namespaces/namespaces/finalize"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces/namespaces/namespaces/finalize",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bnamespace%7D/namespaces/%7Bname%7D/finalize",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + namespace + resourceResource (with name)
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces/namespaces/namespaces"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces/namespaces/namespaces",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bnamespace%7D/namespaces/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + namespace + resourceResource (with NO name) + subresource
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%SUBRESOURCE
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces/namespaces/finalize"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces/namespaces/finalize",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bnamespace%7D/namespaces/finalize",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + namespace + resourceResource (with NO name) + subresource
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%SUBRESOURCE
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces/namespaces/status"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces/namespaces/status",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bnamespace%7D/namespaces/status",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + namespace + resourceResource (with no name)
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces/namespaces"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces/namespaces",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bnamespace%7D/namespaces",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + resourceResource (with name) + subresource
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces/finalize"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces/finalize",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bname%7D/finalize",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + resourceResource (with name) + subresource
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces/status"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces/status",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bname%7D/status",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + resourceResource (with name)
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces/namespaces"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/namespaces",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with named group + resourceResource (with no name)
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/apis/namespaces/namespaces/namespaces"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/apis/namespaces/namespaces/namespaces",
|
||||
},
|
||||
{
|
||||
// dynamic client with wrong api group + namespace + resourceResource (with name) + subresource
|
||||
// /apis/$NAMEDGROUPNAME/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME/$SUBRESOURCE
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/pre1/namespaces/namespaces/namespaces/namespaces/namespaces/namespaces/finalize"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/pre1/namespaces/namespaces/namespaces/namespaces/namespaces/namespaces/finalize",
|
||||
ExpectedFinalURL: "http://localhost/%7Bprefix%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with core group + namespace + resourceResource (with name) where baseURL is a single /
|
||||
// /api/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uriSingleSlash, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/api/v1/namespaces/ns/r2/name1"),
|
||||
ExpectedFullURL: "http://localhost/api/v1/namespaces/ns/r2/name1",
|
||||
ExpectedFinalURL: "http://localhost/api/v1/namespaces/%7Bnamespace%7D/r2/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client with core group + namespace + resourceResource (with name) where baseURL is 'some/base/url/path'
|
||||
// /api/$RESOURCEVERSION/namespaces/$NAMESPACE/$RESOURCE/%NAME
|
||||
Request: NewRequestWithClient(uri, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/api/v1/namespaces/ns/r3/name1"),
|
||||
ExpectedFullURL: "http://localhost/some/base/url/path/api/v1/namespaces/ns/r3/name1",
|
||||
ExpectedFinalURL: "http://localhost/some/base/url/path/api/v1/namespaces/%7Bnamespace%7D/r3/%7Bname%7D",
|
||||
},
|
||||
{
|
||||
// dynamic client where baseURL is a single /
|
||||
// /
|
||||
Request: NewRequestWithClient(uriSingleSlash, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/"),
|
||||
ExpectedFullURL: "http://localhost/",
|
||||
ExpectedFinalURL: "http://localhost/",
|
||||
},
|
||||
{
|
||||
// dynamic client where baseURL is a single /
|
||||
// /version
|
||||
Request: NewRequestWithClient(uriSingleSlash, "", ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "test"}}, nil).Verb("DELETE").
|
||||
Prefix("/version"),
|
||||
ExpectedFullURL: "http://localhost/version",
|
||||
ExpectedFinalURL: "http://localhost/version",
|
||||
},
|
||||
}
|
||||
for i, testCase := range testCases {
|
||||
r := testCase.Request
|
||||
full := r.URL()
|
||||
if full.String() != testCase.ExpectedFullURL {
|
||||
t.Errorf("%d: unexpected initial URL: %s %s", i, full, testCase.ExpectedFullURL)
|
||||
}
|
||||
actualURL := r.finalURLTemplate()
|
||||
actual := actualURL.String()
|
||||
if actual != testCase.ExpectedFinalURL {
|
||||
t.Errorf("%d: unexpected URL template: %s %s", i, actual, testCase.ExpectedFinalURL)
|
||||
}
|
||||
if r.URL().String() != full.String() {
|
||||
t.Errorf("%d, creating URL template changed request: %s -> %s", i, full.String(), r.URL().String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTransformResponse(t *testing.T) {
|
||||
invalid := []byte("aaaaa")
|
||||
uri, _ := url.Parse("http://localhost")
|
||||
@@ -1123,6 +923,42 @@ func TestRequestWatch(t *testing.T) {
|
||||
},
|
||||
Empty: true,
|
||||
},
|
||||
{
|
||||
name: "max retries 1, server returns a retry-after response, request body seek error",
|
||||
Request: &Request{
|
||||
body: &readSeeker{err: io.EOF},
|
||||
c: &RESTClient{
|
||||
base: &url.URL{},
|
||||
},
|
||||
},
|
||||
maxRetries: 1,
|
||||
attemptsExpected: 1,
|
||||
serverReturns: []responseErr{
|
||||
{response: retryAfterResponse(), err: nil},
|
||||
},
|
||||
Err: true,
|
||||
ErrFn: func(err error) bool {
|
||||
return !apierrors.IsInternalError(err) && strings.Contains(err.Error(), "failed to reset the request body while retrying a request: EOF")
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "max retries 1, server returns a retryable error, request body seek error",
|
||||
Request: &Request{
|
||||
body: &readSeeker{err: io.EOF},
|
||||
c: &RESTClient{
|
||||
base: &url.URL{},
|
||||
},
|
||||
},
|
||||
maxRetries: 1,
|
||||
attemptsExpected: 1,
|
||||
serverReturns: []responseErr{
|
||||
{response: nil, err: io.EOF},
|
||||
},
|
||||
Err: true,
|
||||
ErrFn: func(err error) bool {
|
||||
return !apierrors.IsInternalError(err)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "max retries 2, server always returns a response with Retry-After header",
|
||||
Request: &Request{
|
||||
@@ -1284,7 +1120,7 @@ func TestRequestStream(t *testing.T) {
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "max retries 1, server returns a retry-after response, non-bytes request, no retry",
|
||||
name: "max retries 1, server returns a retry-after response, request body seek error",
|
||||
Request: &Request{
|
||||
body: &readSeeker{err: io.EOF},
|
||||
c: &RESTClient{
|
||||
@@ -1297,6 +1133,9 @@ func TestRequestStream(t *testing.T) {
|
||||
{response: retryAfterResponse(), err: nil},
|
||||
},
|
||||
Err: true,
|
||||
ErrFn: func(err error) bool {
|
||||
return !apierrors.IsInternalError(err) && strings.Contains(err.Error(), "failed to reset the request body while retrying a request: EOF")
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "max retries 2, server always returns a response with Retry-After header",
|
||||
@@ -1978,24 +1817,20 @@ func TestBody(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
req, err := r.newHTTPRequest(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if req.Body == nil {
|
||||
if r.body == nil {
|
||||
if len(tt.expected) != 0 {
|
||||
t.Errorf("%d: req.Body = %q; want %q", i, req.Body, tt.expected)
|
||||
t.Errorf("%d: r.body = %q; want %q", i, r.body, tt.expected)
|
||||
}
|
||||
continue
|
||||
}
|
||||
buf := make([]byte, len(tt.expected))
|
||||
if _, err := req.Body.Read(buf); err != nil {
|
||||
t.Errorf("%d: req.Body.Read error: %v", i, err)
|
||||
if _, err := r.body.Read(buf); err != nil {
|
||||
t.Errorf("%d: r.body.Read error: %v", i, err)
|
||||
continue
|
||||
}
|
||||
body := string(buf)
|
||||
if body != tt.expected {
|
||||
t.Errorf("%d: req.Body = %q; want %q", i, body, tt.expected)
|
||||
t.Errorf("%d: r.body = %q; want %q", i, body, tt.expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2606,7 +2441,6 @@ func TestRequestWithRetry(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
body io.Reader
|
||||
bodyBytes []byte
|
||||
serverReturns responseErr
|
||||
errExpected error
|
||||
errContains string
|
||||
@@ -2614,53 +2448,53 @@ func TestRequestWithRetry(t *testing.T) {
|
||||
roundTripInvokedExpected int
|
||||
}{
|
||||
{
|
||||
name: "server returns retry-after response, no request body, retry goes ahead",
|
||||
bodyBytes: nil,
|
||||
name: "server returns retry-after response, request body is not io.Seeker, retry goes ahead",
|
||||
body: ioutil.NopCloser(bytes.NewReader([]byte{})),
|
||||
serverReturns: responseErr{response: retryAfterResponse(), err: nil},
|
||||
errExpected: nil,
|
||||
transformFuncInvokedExpected: 1,
|
||||
roundTripInvokedExpected: 2,
|
||||
},
|
||||
{
|
||||
name: "server returns retry-after response, bytes request body, retry goes ahead",
|
||||
bodyBytes: []byte{},
|
||||
name: "server returns retry-after response, request body Seek returns error, retry aborted",
|
||||
body: &readSeeker{err: io.EOF},
|
||||
serverReturns: responseErr{response: retryAfterResponse(), err: nil},
|
||||
errExpected: nil,
|
||||
transformFuncInvokedExpected: 0,
|
||||
roundTripInvokedExpected: 1,
|
||||
},
|
||||
{
|
||||
name: "server returns retry-after response, request body Seek returns no error, retry goes ahead",
|
||||
body: &readSeeker{err: nil},
|
||||
serverReturns: responseErr{response: retryAfterResponse(), err: nil},
|
||||
errExpected: nil,
|
||||
transformFuncInvokedExpected: 1,
|
||||
roundTripInvokedExpected: 2,
|
||||
},
|
||||
{
|
||||
name: "server returns retry-after response, opaque request body, retry aborted",
|
||||
body: &readSeeker{},
|
||||
serverReturns: responseErr{response: retryAfterResponse(), err: nil},
|
||||
errExpected: nil,
|
||||
transformFuncInvokedExpected: 1,
|
||||
name: "server returns retryable err, request body is not io.Seek, retry goes ahead",
|
||||
body: ioutil.NopCloser(bytes.NewReader([]byte{})),
|
||||
serverReturns: responseErr{response: nil, err: io.ErrUnexpectedEOF},
|
||||
errExpected: io.ErrUnexpectedEOF,
|
||||
transformFuncInvokedExpected: 0,
|
||||
roundTripInvokedExpected: 2,
|
||||
},
|
||||
{
|
||||
name: "server returns retryable err, request body Seek returns error, retry aborted",
|
||||
body: &readSeeker{err: io.EOF},
|
||||
serverReturns: responseErr{response: nil, err: io.ErrUnexpectedEOF},
|
||||
errContains: "failed to reset the request body while retrying a request: EOF",
|
||||
transformFuncInvokedExpected: 0,
|
||||
roundTripInvokedExpected: 1,
|
||||
},
|
||||
{
|
||||
name: "server returns retryable err, no request body, retry goes ahead",
|
||||
bodyBytes: nil,
|
||||
name: "server returns retryable err, request body Seek returns no err, retry goes ahead",
|
||||
body: &readSeeker{err: nil},
|
||||
serverReturns: responseErr{response: nil, err: io.ErrUnexpectedEOF},
|
||||
errExpected: io.ErrUnexpectedEOF,
|
||||
transformFuncInvokedExpected: 0,
|
||||
roundTripInvokedExpected: 2,
|
||||
},
|
||||
{
|
||||
name: "server returns retryable err, bytes request body, retry goes ahead",
|
||||
bodyBytes: []byte{},
|
||||
serverReturns: responseErr{response: nil, err: io.ErrUnexpectedEOF},
|
||||
errExpected: io.ErrUnexpectedEOF,
|
||||
transformFuncInvokedExpected: 0,
|
||||
roundTripInvokedExpected: 2,
|
||||
},
|
||||
{
|
||||
name: "server returns retryable err, opaque request body, retry aborted",
|
||||
body: &readSeeker{},
|
||||
serverReturns: responseErr{response: nil, err: io.ErrUnexpectedEOF},
|
||||
errExpected: io.ErrUnexpectedEOF,
|
||||
transformFuncInvokedExpected: 0,
|
||||
roundTripInvokedExpected: 1,
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
@@ -2831,8 +2665,7 @@ func testRequestWithRetry(t *testing.T, key string, doFunc func(ctx context.Cont
|
||||
tests := []struct {
|
||||
name string
|
||||
verb string
|
||||
body io.Reader
|
||||
bodyBytes []byte
|
||||
body func() io.Reader
|
||||
maxRetries int
|
||||
serverReturns []responseErr
|
||||
|
||||
@@ -2842,7 +2675,7 @@ func testRequestWithRetry(t *testing.T, key string, doFunc func(ctx context.Cont
|
||||
{
|
||||
name: "server always returns retry-after response",
|
||||
verb: "GET",
|
||||
bodyBytes: []byte{},
|
||||
body: func() io.Reader { return bytes.NewReader([]byte{}) },
|
||||
maxRetries: 2,
|
||||
serverReturns: []responseErr{
|
||||
{response: retryAfterResponse(), err: nil},
|
||||
@@ -2870,7 +2703,7 @@ func testRequestWithRetry(t *testing.T, key string, doFunc func(ctx context.Cont
|
||||
{
|
||||
name: "server always returns retryable error",
|
||||
verb: "GET",
|
||||
bodyBytes: []byte{},
|
||||
body: func() io.Reader { return bytes.NewReader([]byte{}) },
|
||||
maxRetries: 2,
|
||||
serverReturns: []responseErr{
|
||||
{response: nil, err: io.EOF},
|
||||
@@ -2899,7 +2732,7 @@ func testRequestWithRetry(t *testing.T, key string, doFunc func(ctx context.Cont
|
||||
{
|
||||
name: "server returns success on the final retry",
|
||||
verb: "GET",
|
||||
bodyBytes: []byte{},
|
||||
body: func() io.Reader { return bytes.NewReader([]byte{}) },
|
||||
maxRetries: 2,
|
||||
serverReturns: []responseErr{
|
||||
{response: retryAfterResponse(), err: nil},
|
||||
@@ -2946,10 +2779,13 @@ func testRequestWithRetry(t *testing.T, key string, doFunc func(ctx context.Cont
|
||||
return resp, test.serverReturns[attempts].err
|
||||
})
|
||||
|
||||
reqCountGot := newCount()
|
||||
reqRecorder := newReadTracker(reqCountGot)
|
||||
reqRecorder.delegated = test.body()
|
||||
|
||||
req := &Request{
|
||||
verb: test.verb,
|
||||
body: test.body,
|
||||
bodyBytes: test.bodyBytes,
|
||||
verb: test.verb,
|
||||
body: reqRecorder,
|
||||
c: &RESTClient{
|
||||
content: defaultContentConfig(),
|
||||
Client: client,
|
||||
@@ -2969,6 +2805,9 @@ func testRequestWithRetry(t *testing.T, key string, doFunc func(ctx context.Cont
|
||||
t.Errorf("Expected retries: %d, but got: %d", expected.attempts, attempts)
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(expected.reqCount.seeks, reqCountGot.seeks) {
|
||||
t.Errorf("Expected request body to have seek invocation: %v, but got: %v", expected.reqCount.seeks, reqCountGot.seeks)
|
||||
}
|
||||
if expected.respCount.closes != respCountGot.getCloseCount() {
|
||||
t.Errorf("Expected response body Close to be invoked %d times, but got: %d", expected.respCount.closes, respCountGot.getCloseCount())
|
||||
}
|
||||
@@ -3225,8 +3064,8 @@ func testRetryWithRateLimiterBackoffAndMetrics(t *testing.T, key string, doFunc
|
||||
t.Fatalf("Wrong test setup - did not find expected for: %s", key)
|
||||
}
|
||||
req := &Request{
|
||||
verb: "GET",
|
||||
bodyBytes: []byte{},
|
||||
verb: "GET",
|
||||
body: bytes.NewReader([]byte{}),
|
||||
c: &RESTClient{
|
||||
base: base,
|
||||
content: defaultContentConfig(),
|
||||
@@ -3361,8 +3200,8 @@ func testWithRetryInvokeOrder(t *testing.T, key string, doFunc func(ctx context.
|
||||
t.Fatalf("Wrong test setup - did not find expected for: %s", key)
|
||||
}
|
||||
req := &Request{
|
||||
verb: "GET",
|
||||
bodyBytes: []byte{},
|
||||
verb: "GET",
|
||||
body: bytes.NewReader([]byte{}),
|
||||
c: &RESTClient{
|
||||
base: base,
|
||||
content: defaultContentConfig(),
|
||||
@@ -3536,8 +3375,8 @@ func testWithWrapPreviousError(t *testing.T, doFunc func(ctx context.Context, r
|
||||
t.Fatalf("Failed to create new HTTP request - %v", err)
|
||||
}
|
||||
req := &Request{
|
||||
verb: "GET",
|
||||
bodyBytes: []byte{},
|
||||
verb: "GET",
|
||||
body: bytes.NewReader([]byte{}),
|
||||
c: &RESTClient{
|
||||
base: base,
|
||||
content: defaultContentConfig(),
|
||||
@@ -3772,3 +3611,104 @@ func TestTransportConcurrency(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: see if we can consolidate the other trackers into one.
|
||||
type requestBodyTracker struct {
|
||||
io.ReadSeeker
|
||||
f func(string)
|
||||
}
|
||||
|
||||
func (t *requestBodyTracker) Read(p []byte) (int, error) {
|
||||
t.f("Request.Body.Read")
|
||||
return t.ReadSeeker.Read(p)
|
||||
}
|
||||
|
||||
func (t *requestBodyTracker) Seek(offset int64, whence int) (int64, error) {
|
||||
t.f("Request.Body.Seek")
|
||||
return t.ReadSeeker.Seek(offset, whence)
|
||||
}
|
||||
|
||||
type responseBodyTracker struct {
|
||||
io.ReadCloser
|
||||
f func(string)
|
||||
}
|
||||
|
||||
func (t *responseBodyTracker) Read(p []byte) (int, error) {
|
||||
t.f("Response.Body.Read")
|
||||
return t.ReadCloser.Read(p)
|
||||
}
|
||||
|
||||
func (t *responseBodyTracker) Close() error {
|
||||
t.f("Response.Body.Close")
|
||||
return t.ReadCloser.Close()
|
||||
}
|
||||
|
||||
type recorder struct {
|
||||
order []string
|
||||
}
|
||||
|
||||
func (r *recorder) record(call string) {
|
||||
r.order = append(r.order, call)
|
||||
}
|
||||
|
||||
func TestRequestBodyResetOrder(t *testing.T) {
|
||||
recorder := &recorder{}
|
||||
respBodyTracker := &responseBodyTracker{
|
||||
ReadCloser: nil, // the server will fill it
|
||||
f: recorder.record,
|
||||
}
|
||||
|
||||
var attempts int
|
||||
client := clientForFunc(func(req *http.Request) (*http.Response, error) {
|
||||
defer func() {
|
||||
attempts++
|
||||
}()
|
||||
|
||||
// read the request body.
|
||||
ioutil.ReadAll(req.Body)
|
||||
|
||||
// first attempt, we send a retry-after
|
||||
if attempts == 0 {
|
||||
resp := retryAfterResponse()
|
||||
respBodyTracker.ReadCloser = ioutil.NopCloser(bytes.NewReader([]byte{}))
|
||||
resp.Body = respBodyTracker
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
return &http.Response{StatusCode: http.StatusOK}, nil
|
||||
})
|
||||
|
||||
reqBodyTracker := &requestBodyTracker{
|
||||
ReadSeeker: bytes.NewReader([]byte{}), // empty body ensures one Read operation at most.
|
||||
f: recorder.record,
|
||||
}
|
||||
req := &Request{
|
||||
verb: "POST",
|
||||
body: reqBodyTracker,
|
||||
c: &RESTClient{
|
||||
content: defaultContentConfig(),
|
||||
Client: client,
|
||||
},
|
||||
backoff: &noSleepBackOff{},
|
||||
maxRetries: 1,
|
||||
retryFn: defaultRequestRetryFn,
|
||||
}
|
||||
|
||||
req.Do(context.Background())
|
||||
|
||||
expected := []string{
|
||||
// 1st attempt: the server handler reads the request body
|
||||
"Request.Body.Read",
|
||||
// the server sends a retry-after, client reads the
|
||||
// response body, and closes it
|
||||
"Response.Body.Read",
|
||||
"Response.Body.Close",
|
||||
// client retry logic seeks to the beginning of the request body
|
||||
"Request.Body.Seek",
|
||||
// 2nd attempt: the server reads the request body
|
||||
"Request.Body.Read",
|
||||
}
|
||||
if !reflect.DeepEqual(expected, recorder.order) {
|
||||
t.Errorf("Expected invocation request and response body operations for retry do not match: %s", cmp.Diff(expected, recorder.order))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -154,11 +154,6 @@ func (r *withRetry) IsNextRetry(ctx context.Context, restReq *Request, httpReq *
|
||||
return false
|
||||
}
|
||||
|
||||
if restReq.body != nil {
|
||||
// we have an opaque reader, we can't safely reset it
|
||||
return false
|
||||
}
|
||||
|
||||
r.attempts++
|
||||
r.retryAfter = &RetryAfter{Attempt: r.attempts}
|
||||
if r.attempts > r.maxRetries {
|
||||
@@ -215,6 +210,18 @@ func (r *withRetry) Before(ctx context.Context, request *Request) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// At this point we've made atleast one attempt, post which the response
|
||||
// body should have been fully read and closed in order for it to be safe
|
||||
// to reset the request body before we reconnect, in order for us to reuse
|
||||
// the same TCP connection.
|
||||
if seeker, ok := request.body.(io.Seeker); ok && request.body != nil {
|
||||
if _, err := seeker.Seek(0, io.SeekStart); err != nil {
|
||||
err = fmt.Errorf("failed to reset the request body while retrying a request: %v", err)
|
||||
r.trackPreviousError(err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// if we are here, we have made attempt(s) at least once before.
|
||||
if request.backoff != nil {
|
||||
delay := request.backoff.CalculateBackoff(url)
|
||||
|
||||
@@ -17,6 +17,7 @@ limitations under the License.
|
||||
package rest
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -211,7 +212,7 @@ func TestIsNextRetry(t *testing.T) {
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
restReq := &Request{
|
||||
bodyBytes: []byte{},
|
||||
body: bytes.NewReader([]byte{}),
|
||||
c: &RESTClient{
|
||||
base: &url.URL{},
|
||||
},
|
||||
|
||||
22
tools/cache/controller.go
vendored
22
tools/cache/controller.go
vendored
@@ -353,6 +353,17 @@ func NewIndexerInformer(
|
||||
return clientState, newInformer(lw, objType, resyncPeriod, h, clientState, nil)
|
||||
}
|
||||
|
||||
// TransformFunc allows for transforming an object before it will be processed
|
||||
// and put into the controller cache and before the corresponding handlers will
|
||||
// be called on it.
|
||||
// TransformFunc (similarly to ResourceEventHandler functions) should be able
|
||||
// to correctly handle the tombstone of type cache.DeletedFinalStateUnknown
|
||||
//
|
||||
// The most common usage pattern is to clean-up some parts of the object to
|
||||
// reduce component memory usage if a given component doesn't care about them.
|
||||
// given controller doesn't care for them
|
||||
type TransformFunc func(interface{}) (interface{}, error)
|
||||
|
||||
// NewTransformingInformer returns a Store and a controller for populating
|
||||
// the store while also providing event notifications. You should only used
|
||||
// the returned Store for Get/List operations; Add/Modify/Deletes will cause
|
||||
@@ -400,11 +411,19 @@ func processDeltas(
|
||||
// Object which receives event notifications from the given deltas
|
||||
handler ResourceEventHandler,
|
||||
clientState Store,
|
||||
transformer TransformFunc,
|
||||
deltas Deltas,
|
||||
) error {
|
||||
// from oldest to newest
|
||||
for _, d := range deltas {
|
||||
obj := d.Object
|
||||
if transformer != nil {
|
||||
var err error
|
||||
obj, err = transformer(obj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
switch d.Type {
|
||||
case Sync, Replaced, Added, Updated:
|
||||
@@ -456,7 +475,6 @@ func newInformer(
|
||||
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
|
||||
KnownObjects: clientState,
|
||||
EmitDeltaTypeReplaced: true,
|
||||
Transformer: transformer,
|
||||
})
|
||||
|
||||
cfg := &Config{
|
||||
@@ -468,7 +486,7 @@ func newInformer(
|
||||
|
||||
Process: func(obj interface{}) error {
|
||||
if deltas, ok := obj.(Deltas); ok {
|
||||
return processDeltas(h, clientState, deltas)
|
||||
return processDeltas(h, clientState, transformer, deltas)
|
||||
}
|
||||
return errors.New("object given as Process argument is not Deltas")
|
||||
},
|
||||
|
||||
4
tools/cache/controller_test.go
vendored
4
tools/cache/controller_test.go
vendored
@@ -23,7 +23,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
v1 "k8s.io/api/core/v1"
|
||||
"k8s.io/api/core/v1"
|
||||
apiequality "k8s.io/apimachinery/pkg/api/equality"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
@@ -32,7 +32,7 @@ import (
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
fcache "k8s.io/client-go/tools/cache/testing"
|
||||
|
||||
fuzz "github.com/google/gofuzz"
|
||||
"github.com/google/gofuzz"
|
||||
)
|
||||
|
||||
func Example() {
|
||||
|
||||
135
tools/cache/delta_fifo.go
vendored
135
tools/cache/delta_fifo.go
vendored
@@ -51,10 +51,6 @@ type DeltaFIFOOptions struct {
|
||||
// When true, `Replaced` events will be sent for items passed to a Replace() call.
|
||||
// When false, `Sync` events will be sent instead.
|
||||
EmitDeltaTypeReplaced bool
|
||||
|
||||
// If set, will be called for objects before enqueueing them. Please
|
||||
// see the comment on TransformFunc for details.
|
||||
Transformer TransformFunc
|
||||
}
|
||||
|
||||
// DeltaFIFO is like FIFO, but differs in two ways. One is that the
|
||||
@@ -133,32 +129,8 @@ type DeltaFIFO struct {
|
||||
// emitDeltaTypeReplaced is whether to emit the Replaced or Sync
|
||||
// DeltaType when Replace() is called (to preserve backwards compat).
|
||||
emitDeltaTypeReplaced bool
|
||||
|
||||
// Called with every object if non-nil.
|
||||
transformer TransformFunc
|
||||
}
|
||||
|
||||
// TransformFunc allows for transforming an object before it will be processed.
|
||||
// TransformFunc (similarly to ResourceEventHandler functions) should be able
|
||||
// to correctly handle the tombstone of type cache.DeletedFinalStateUnknown.
|
||||
//
|
||||
// New in v1.27: In such cases, the contained object will already have gone
|
||||
// through the transform object separately (when it was added / updated prior
|
||||
// to the delete), so the TransformFunc can likely safely ignore such objects
|
||||
// (i.e., just return the input object).
|
||||
//
|
||||
// The most common usage pattern is to clean-up some parts of the object to
|
||||
// reduce component memory usage if a given component doesn't care about them.
|
||||
//
|
||||
// New in v1.27: unless the object is a DeletedFinalStateUnknown, TransformFunc
|
||||
// sees the object before any other actor, and it is now safe to mutate the
|
||||
// object in place instead of making a copy.
|
||||
//
|
||||
// Note that TransformFunc is called while inserting objects into the
|
||||
// notification queue and is therefore extremely performance sensitive; please
|
||||
// do not do anything that will take a long time.
|
||||
type TransformFunc func(interface{}) (interface{}, error)
|
||||
|
||||
// DeltaType is the type of a change (addition, deletion, etc)
|
||||
type DeltaType string
|
||||
|
||||
@@ -255,7 +227,6 @@ func NewDeltaFIFOWithOptions(opts DeltaFIFOOptions) *DeltaFIFO {
|
||||
knownObjects: opts.KnownObjects,
|
||||
|
||||
emitDeltaTypeReplaced: opts.EmitDeltaTypeReplaced,
|
||||
transformer: opts.Transformer,
|
||||
}
|
||||
f.cond.L = &f.lock
|
||||
return f
|
||||
@@ -440,21 +411,6 @@ func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) err
|
||||
if err != nil {
|
||||
return KeyError{obj, err}
|
||||
}
|
||||
|
||||
// Every object comes through this code path once, so this is a good
|
||||
// place to call the transform func. If obj is a
|
||||
// DeletedFinalStateUnknown tombstone, then the containted inner object
|
||||
// will already have gone through the transformer, but we document that
|
||||
// this can happen. In cases involving Replace(), such an object can
|
||||
// come through multiple times.
|
||||
if f.transformer != nil {
|
||||
var err error
|
||||
obj, err = f.transformer(obj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
oldDeltas := f.items[id]
|
||||
newDeltas := append(oldDeltas, Delta{actionType, obj})
|
||||
newDeltas = dedupDeltas(newDeltas)
|
||||
@@ -610,11 +566,12 @@ func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
|
||||
// using the Sync or Replace DeltaType and then (2) it does some deletions.
|
||||
// In particular: for every pre-existing key K that is not the key of
|
||||
// an object in `list` there is the effect of
|
||||
// `Delete(DeletedFinalStateUnknown{K, O})` where O is the latest known
|
||||
// object of K. The pre-existing keys are those in the union set of the keys in
|
||||
// `f.items` and `f.knownObjects` (if not nil). The last known object for key K is
|
||||
// the one present in the last delta in `f.items`. If there is no delta for K
|
||||
// in `f.items`, it is the object in `f.knownObjects`
|
||||
// `Delete(DeletedFinalStateUnknown{K, O})` where O is current object
|
||||
// of K. If `f.knownObjects == nil` then the pre-existing keys are
|
||||
// those in `f.items` and the current object of K is the `.Newest()`
|
||||
// of the Deltas associated with K. Otherwise the pre-existing keys
|
||||
// are those listed by `f.knownObjects` and the current object of K is
|
||||
// what `f.knownObjects.GetByKey(K)` returns.
|
||||
func (f *DeltaFIFO) Replace(list []interface{}, _ string) error {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
@@ -638,54 +595,56 @@ func (f *DeltaFIFO) Replace(list []interface{}, _ string) error {
|
||||
}
|
||||
}
|
||||
|
||||
// Do deletion detection against objects in the queue
|
||||
queuedDeletions := 0
|
||||
for k, oldItem := range f.items {
|
||||
if keys.Has(k) {
|
||||
continue
|
||||
}
|
||||
// Delete pre-existing items not in the new list.
|
||||
// This could happen if watch deletion event was missed while
|
||||
// disconnected from apiserver.
|
||||
var deletedObj interface{}
|
||||
if n := oldItem.Newest(); n != nil {
|
||||
deletedObj = n.Object
|
||||
|
||||
// if the previous object is a DeletedFinalStateUnknown, we have to extract the actual Object
|
||||
if d, ok := deletedObj.(DeletedFinalStateUnknown); ok {
|
||||
deletedObj = d.Obj
|
||||
}
|
||||
}
|
||||
queuedDeletions++
|
||||
if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if f.knownObjects != nil {
|
||||
// Detect deletions for objects not present in the queue, but present in KnownObjects
|
||||
knownKeys := f.knownObjects.ListKeys()
|
||||
for _, k := range knownKeys {
|
||||
if f.knownObjects == nil {
|
||||
// Do deletion detection against our own list.
|
||||
queuedDeletions := 0
|
||||
for k, oldItem := range f.items {
|
||||
if keys.Has(k) {
|
||||
continue
|
||||
}
|
||||
if len(f.items[k]) > 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
deletedObj, exists, err := f.knownObjects.GetByKey(k)
|
||||
if err != nil {
|
||||
deletedObj = nil
|
||||
klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k)
|
||||
} else if !exists {
|
||||
deletedObj = nil
|
||||
klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k)
|
||||
// Delete pre-existing items not in the new list.
|
||||
// This could happen if watch deletion event was missed while
|
||||
// disconnected from apiserver.
|
||||
var deletedObj interface{}
|
||||
if n := oldItem.Newest(); n != nil {
|
||||
deletedObj = n.Object
|
||||
}
|
||||
queuedDeletions++
|
||||
if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if !f.populated {
|
||||
f.populated = true
|
||||
// While there shouldn't be any queued deletions in the initial
|
||||
// population of the queue, it's better to be on the safe side.
|
||||
f.initialPopulationCount = keys.Len() + queuedDeletions
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Detect deletions not already in the queue.
|
||||
knownKeys := f.knownObjects.ListKeys()
|
||||
queuedDeletions := 0
|
||||
for _, k := range knownKeys {
|
||||
if keys.Has(k) {
|
||||
continue
|
||||
}
|
||||
|
||||
deletedObj, exists, err := f.knownObjects.GetByKey(k)
|
||||
if err != nil {
|
||||
deletedObj = nil
|
||||
klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k)
|
||||
} else if !exists {
|
||||
deletedObj = nil
|
||||
klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k)
|
||||
}
|
||||
queuedDeletions++
|
||||
if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if !f.populated {
|
||||
|
||||
269
tools/cache/delta_fifo_test.go
vendored
269
tools/cache/delta_fifo_test.go
vendored
@@ -121,130 +121,6 @@ func TestDeltaFIFO_replaceWithDeleteDeltaIn(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) {
|
||||
obj := mkFifoObj("foo", 2)
|
||||
objV2 := mkFifoObj("foo", 3)
|
||||
table := []struct {
|
||||
name string
|
||||
operations func(f *DeltaFIFO)
|
||||
expectedDeltas Deltas
|
||||
}{
|
||||
{
|
||||
name: "Added object should be deleted on Replace",
|
||||
operations: func(f *DeltaFIFO) {
|
||||
f.Add(obj)
|
||||
f.Replace([]interface{}{}, "0")
|
||||
},
|
||||
expectedDeltas: Deltas{
|
||||
{Added, obj},
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Replaced object should have only a single Delete",
|
||||
operations: func(f *DeltaFIFO) {
|
||||
f.emitDeltaTypeReplaced = true
|
||||
f.Add(obj)
|
||||
f.Replace([]interface{}{obj}, "0")
|
||||
f.Replace([]interface{}{}, "0")
|
||||
},
|
||||
expectedDeltas: Deltas{
|
||||
{Added, obj},
|
||||
{Replaced, obj},
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Deleted object should have only a single Delete",
|
||||
operations: func(f *DeltaFIFO) {
|
||||
f.Add(obj)
|
||||
f.Delete(obj)
|
||||
f.Replace([]interface{}{}, "0")
|
||||
},
|
||||
expectedDeltas: Deltas{
|
||||
{Added, obj},
|
||||
{Deleted, obj},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Synced objects should have a single delete",
|
||||
operations: func(f *DeltaFIFO) {
|
||||
f.Add(obj)
|
||||
f.Replace([]interface{}{obj}, "0")
|
||||
f.Replace([]interface{}{obj}, "0")
|
||||
f.Replace([]interface{}{}, "0")
|
||||
},
|
||||
expectedDeltas: Deltas{
|
||||
{Added, obj},
|
||||
{Sync, obj},
|
||||
{Sync, obj},
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Added objects should have a single delete on multiple Replaces",
|
||||
operations: func(f *DeltaFIFO) {
|
||||
f.Add(obj)
|
||||
f.Replace([]interface{}{}, "0")
|
||||
f.Replace([]interface{}{}, "1")
|
||||
},
|
||||
expectedDeltas: Deltas{
|
||||
{Added, obj},
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: obj}},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "Added and deleted and added object should be deleted",
|
||||
operations: func(f *DeltaFIFO) {
|
||||
f.Add(obj)
|
||||
f.Delete(obj)
|
||||
f.Add(objV2)
|
||||
f.Replace([]interface{}{}, "0")
|
||||
},
|
||||
expectedDeltas: Deltas{
|
||||
{Added, obj},
|
||||
{Deleted, obj},
|
||||
{Added, objV2},
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: objV2}},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range table {
|
||||
tt := tt
|
||||
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Test with a DeltaFIFO with a backing KnownObjects
|
||||
fWithKnownObjects := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
|
||||
KeyFunction: testFifoObjectKeyFunc,
|
||||
KnownObjects: literalListerGetter(func() []testFifoObject {
|
||||
return []testFifoObject{}
|
||||
}),
|
||||
})
|
||||
tt.operations(fWithKnownObjects)
|
||||
actualDeltasWithKnownObjects := Pop(fWithKnownObjects)
|
||||
if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithKnownObjects) {
|
||||
t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithKnownObjects)
|
||||
}
|
||||
if len(fWithKnownObjects.items) != 0 {
|
||||
t.Errorf("expected no extra deltas (empty map), got %#v", fWithKnownObjects.items)
|
||||
}
|
||||
|
||||
// Test with a DeltaFIFO without a backing KnownObjects
|
||||
fWithoutKnownObjects := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
|
||||
KeyFunction: testFifoObjectKeyFunc,
|
||||
})
|
||||
tt.operations(fWithoutKnownObjects)
|
||||
actualDeltasWithoutKnownObjects := Pop(fWithoutKnownObjects)
|
||||
if !reflect.DeepEqual(tt.expectedDeltas, actualDeltasWithoutKnownObjects) {
|
||||
t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualDeltasWithoutKnownObjects)
|
||||
}
|
||||
if len(fWithoutKnownObjects.items) != 0 {
|
||||
t.Errorf("expected no extra deltas (empty map), got %#v", fWithoutKnownObjects.items)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeltaFIFO_requeueOnPop(t *testing.T) {
|
||||
f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
|
||||
|
||||
@@ -327,88 +203,6 @@ func TestDeltaFIFO_addUpdate(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
type rvAndXfrm struct {
|
||||
rv int
|
||||
xfrm int
|
||||
}
|
||||
|
||||
func TestDeltaFIFO_transformer(t *testing.T) {
|
||||
mk := func(name string, rv int) testFifoObject {
|
||||
return mkFifoObj(name, &rvAndXfrm{rv, 0})
|
||||
}
|
||||
xfrm := TransformFunc(func(obj interface{}) (interface{}, error) {
|
||||
switch v := obj.(type) {
|
||||
case testFifoObject:
|
||||
v.val.(*rvAndXfrm).xfrm++
|
||||
case DeletedFinalStateUnknown:
|
||||
if x := v.Obj.(testFifoObject).val.(*rvAndXfrm).xfrm; x != 1 {
|
||||
return nil, fmt.Errorf("object has been transformed wrong number of times: %#v", obj)
|
||||
}
|
||||
default:
|
||||
return nil, fmt.Errorf("unexpected object: %#v", obj)
|
||||
}
|
||||
return obj, nil
|
||||
})
|
||||
|
||||
must := func(err error) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
|
||||
KeyFunction: testFifoObjectKeyFunc,
|
||||
Transformer: xfrm,
|
||||
})
|
||||
must(f.Add(mk("foo", 10)))
|
||||
must(f.Add(mk("bar", 11)))
|
||||
must(f.Update(mk("foo", 12)))
|
||||
must(f.Delete(mk("foo", 15)))
|
||||
must(f.Replace([]interface{}{}, ""))
|
||||
must(f.Add(mk("bar", 16)))
|
||||
must(f.Replace([]interface{}{}, ""))
|
||||
|
||||
// Should be empty
|
||||
if e, a := []string{"foo", "bar"}, f.ListKeys(); !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("Expected %+v, got %+v", e, a)
|
||||
}
|
||||
|
||||
for i := 0; i < 2; i++ {
|
||||
obj, err := f.Pop(func(o interface{}) error { return nil })
|
||||
if err != nil {
|
||||
t.Fatalf("got nothing on try %v?", i)
|
||||
}
|
||||
obj = obj.(Deltas).Newest().Object
|
||||
switch v := obj.(type) {
|
||||
case testFifoObject:
|
||||
if v.name != "foo" {
|
||||
t.Errorf("expected regular deletion of foo, got %q", v.name)
|
||||
}
|
||||
rx := v.val.(*rvAndXfrm)
|
||||
if rx.rv != 15 {
|
||||
t.Errorf("expected last message, got %#v", obj)
|
||||
}
|
||||
if rx.xfrm != 1 {
|
||||
t.Errorf("obj %v transformed wrong number of times.", obj)
|
||||
}
|
||||
case DeletedFinalStateUnknown:
|
||||
tf := v.Obj.(testFifoObject)
|
||||
rx := tf.val.(*rvAndXfrm)
|
||||
if tf.name != "bar" {
|
||||
t.Errorf("expected tombstone deletion of bar, got %q", tf.name)
|
||||
}
|
||||
if rx.rv != 16 {
|
||||
t.Errorf("expected last message, got %#v", obj)
|
||||
}
|
||||
if rx.xfrm != 1 {
|
||||
t.Errorf("tombstoned obj %v transformed wrong number of times.", obj)
|
||||
}
|
||||
default:
|
||||
t.Errorf("unknown item %#v", obj)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeltaFIFO_enqueueingNoLister(t *testing.T) {
|
||||
f := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
|
||||
f.Add(mkFifoObj("foo", 10))
|
||||
@@ -577,7 +371,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) {
|
||||
|
||||
expectedList = []Deltas{
|
||||
{{Added, mkFifoObj("baz", 10)},
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 10)}}},
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
|
||||
{{Sync, mkFifoObj("foo", 5)}},
|
||||
// Since "bar" didn't have a delete event and wasn't in the Replace list
|
||||
// it should get a tombstone key with the right Obj.
|
||||
@@ -591,67 +385,6 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// Now try deleting and recreating the object in the queue, then delete it by a Replace call
|
||||
f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{
|
||||
KeyFunction: testFifoObjectKeyFunc,
|
||||
KnownObjects: literalListerGetter(func() []testFifoObject {
|
||||
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
|
||||
}),
|
||||
})
|
||||
f.Delete(mkFifoObj("bar", 6))
|
||||
f.Add(mkFifoObj("bar", 100))
|
||||
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
|
||||
|
||||
expectedList = []Deltas{
|
||||
{
|
||||
{Deleted, mkFifoObj("bar", 6)},
|
||||
{Added, mkFifoObj("bar", 100)},
|
||||
// Since "bar" has a newer object in the queue than in the state,
|
||||
// it should get a tombstone key with the latest object from the queue
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}},
|
||||
},
|
||||
{{Sync, mkFifoObj("foo", 5)}},
|
||||
{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
|
||||
}
|
||||
|
||||
for _, expected := range expectedList {
|
||||
cur := Pop(f).(Deltas)
|
||||
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("Expected %#v, got %#v", e, a)
|
||||
}
|
||||
}
|
||||
|
||||
// Now try syncing it first to ensure the delete use the latest version
|
||||
f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{
|
||||
KeyFunction: testFifoObjectKeyFunc,
|
||||
KnownObjects: literalListerGetter(func() []testFifoObject {
|
||||
return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)}
|
||||
}),
|
||||
})
|
||||
f.Replace([]interface{}{mkFifoObj("bar", 100), mkFifoObj("foo", 5)}, "0")
|
||||
f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0")
|
||||
|
||||
expectedList = []Deltas{
|
||||
{
|
||||
{Sync, mkFifoObj("bar", 100)},
|
||||
// Since "bar" didn't have a delete event and wasn't in the Replace list
|
||||
// it should get a tombstone key with the right Obj.
|
||||
{Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}},
|
||||
},
|
||||
{
|
||||
{Sync, mkFifoObj("foo", 5)},
|
||||
{Sync, mkFifoObj("foo", 5)},
|
||||
},
|
||||
{{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}},
|
||||
}
|
||||
|
||||
for _, expected := range expectedList {
|
||||
cur := Pop(f).(Deltas)
|
||||
if e, a := expected, cur; !reflect.DeepEqual(e, a) {
|
||||
t.Errorf("Expected %#v, got %#v", e, a)
|
||||
}
|
||||
}
|
||||
|
||||
// Now try starting without an explicit KeyListerGetter
|
||||
f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc})
|
||||
f.Add(mkFifoObj("baz", 10))
|
||||
|
||||
8
tools/cache/shared_informer.go
vendored
8
tools/cache/shared_informer.go
vendored
@@ -190,7 +190,10 @@ type SharedInformer interface {
|
||||
//
|
||||
// Must be set before starting the informer.
|
||||
//
|
||||
// Please see the comment on TransformFunc for more details.
|
||||
// Note: Since the object given to the handler may be already shared with
|
||||
// other goroutines, it is advisable to copy the object being
|
||||
// transform before mutating it at all and returning the copy to prevent
|
||||
// data races.
|
||||
SetTransform(handler TransformFunc) error
|
||||
}
|
||||
|
||||
@@ -401,7 +404,6 @@ func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
|
||||
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
|
||||
KnownObjects: s.indexer,
|
||||
EmitDeltaTypeReplaced: true,
|
||||
Transformer: s.transform,
|
||||
})
|
||||
|
||||
cfg := &Config{
|
||||
@@ -566,7 +568,7 @@ func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
|
||||
defer s.blockDeltas.Unlock()
|
||||
|
||||
if deltas, ok := obj.(Deltas); ok {
|
||||
return processDeltas(s, s.indexer, deltas)
|
||||
return processDeltas(s, s.indexer, s.transform, deltas)
|
||||
}
|
||||
return errors.New("object given as Process argument is not Deltas")
|
||||
}
|
||||
|
||||
5
tools/cache/shared_informer_test.go
vendored
5
tools/cache/shared_informer_test.go
vendored
@@ -371,8 +371,9 @@ func TestSharedInformerTransformer(t *testing.T) {
|
||||
name := pod.GetName()
|
||||
|
||||
if upper := strings.ToUpper(name); upper != name {
|
||||
pod.SetName(upper)
|
||||
return pod, nil
|
||||
copied := pod.DeepCopyObject().(*v1.Pod)
|
||||
copied.SetName(upper)
|
||||
return copied, nil
|
||||
}
|
||||
}
|
||||
return obj, nil
|
||||
|
||||
@@ -17,7 +17,6 @@ limitations under the License.
|
||||
package transport
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
@@ -56,9 +55,6 @@ type tlsCacheKey struct {
|
||||
serverName string
|
||||
nextProtos string
|
||||
disableCompression bool
|
||||
// these functions are wrapped to allow them to be used as map keys
|
||||
getCert *GetCertHolder
|
||||
dial *DialHolder
|
||||
}
|
||||
|
||||
func (t tlsCacheKey) String() string {
|
||||
@@ -66,8 +62,7 @@ func (t tlsCacheKey) String() string {
|
||||
if len(t.keyData) > 0 {
|
||||
keyText = "<redacted>"
|
||||
}
|
||||
return fmt.Sprintf("insecure:%v, caData:%#v, certData:%#v, keyData:%s, serverName:%s, disableCompression:%t, getCert:%p, dial:%p",
|
||||
t.insecure, t.caData, t.certData, keyText, t.serverName, t.disableCompression, t.getCert, t.dial)
|
||||
return fmt.Sprintf("insecure:%v, caData:%#v, certData:%#v, keyData:%s, serverName:%s, disableCompression:%t", t.insecure, t.caData, t.certData, keyText, t.serverName, t.disableCompression)
|
||||
}
|
||||
|
||||
func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
|
||||
@@ -97,10 +92,8 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
|
||||
return http.DefaultTransport, nil
|
||||
}
|
||||
|
||||
var dial func(ctx context.Context, network, address string) (net.Conn, error)
|
||||
if config.Dial != nil {
|
||||
dial = config.Dial
|
||||
} else {
|
||||
dial := config.Dial
|
||||
if dial == nil {
|
||||
dial = (&net.Dialer{
|
||||
Timeout: 30 * time.Second,
|
||||
KeepAlive: 30 * time.Second,
|
||||
@@ -109,7 +102,7 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
|
||||
|
||||
// If we use are reloading files, we need to handle certificate rotation properly
|
||||
// TODO(jackkleeman): We can also add rotation here when config.HasCertCallback() is true
|
||||
if config.TLS.ReloadTLSFiles && tlsConfig != nil && tlsConfig.GetClientCertificate != nil {
|
||||
if config.TLS.ReloadTLSFiles {
|
||||
dynamicCertDialer := certRotatingDialer(tlsConfig.GetClientCertificate, dial)
|
||||
tlsConfig.GetClientCertificate = dynamicCertDialer.GetClientCertificate
|
||||
dial = dynamicCertDialer.connDialer.DialContext
|
||||
@@ -145,18 +138,10 @@ func tlsConfigKey(c *Config) (tlsCacheKey, bool, error) {
|
||||
return tlsCacheKey{}, false, err
|
||||
}
|
||||
|
||||
if c.Proxy != nil {
|
||||
if c.TLS.GetCert != nil || c.Dial != nil || c.Proxy != nil {
|
||||
// cannot determine equality for functions
|
||||
return tlsCacheKey{}, false, nil
|
||||
}
|
||||
if c.Dial != nil && c.DialHolder == nil {
|
||||
// cannot determine equality for dial function that doesn't have non-nil DialHolder set as well
|
||||
return tlsCacheKey{}, false, nil
|
||||
}
|
||||
if c.TLS.GetCert != nil && c.TLS.GetCertHolder == nil {
|
||||
// cannot determine equality for getCert function that doesn't have non-nil GetCertHolder set as well
|
||||
return tlsCacheKey{}, false, nil
|
||||
}
|
||||
|
||||
k := tlsCacheKey{
|
||||
insecure: c.TLS.Insecure,
|
||||
@@ -164,8 +149,6 @@ func tlsConfigKey(c *Config) (tlsCacheKey, bool, error) {
|
||||
serverName: c.TLS.ServerName,
|
||||
nextProtos: strings.Join(c.TLS.NextProtos, ","),
|
||||
disableCompression: c.DisableCompression,
|
||||
getCert: c.TLS.GetCertHolder,
|
||||
dial: c.DialHolder,
|
||||
}
|
||||
|
||||
if c.TLS.ReloadTLSFiles {
|
||||
|
||||
@@ -21,7 +21,6 @@ import (
|
||||
"crypto/tls"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"testing"
|
||||
)
|
||||
|
||||
@@ -59,24 +58,16 @@ func TestTLSConfigKey(t *testing.T) {
|
||||
t.Errorf("Expected identical cache keys for %q and %q, got:\n\t%s\n\t%s", nameA, nameB, keyA, keyB)
|
||||
continue
|
||||
}
|
||||
if keyA != (tlsCacheKey{}) {
|
||||
t.Errorf("Expected empty cache keys for %q and %q, got:\n\t%s\n\t%s", nameA, nameB, keyA, keyB)
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Make sure config fields that affect the tls config affect the cache key
|
||||
dialer := net.Dialer{}
|
||||
getCert := func() (*tls.Certificate, error) { return nil, nil }
|
||||
getCertHolder := &GetCertHolder{GetCert: getCert}
|
||||
uniqueConfigurations := map[string]*Config{
|
||||
"proxy": {Proxy: func(request *http.Request) (*url.URL, error) { return nil, nil }},
|
||||
"no tls": {},
|
||||
"dialer": {Dial: dialer.DialContext},
|
||||
"dialer2": {Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil }},
|
||||
"dialer3": {Dial: dialer.DialContext, DialHolder: &DialHolder{Dial: dialer.DialContext}},
|
||||
"dialer4": {Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil }, DialHolder: &DialHolder{Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil }}},
|
||||
"insecure": {TLS: TLSConfig{Insecure: true}},
|
||||
"cadata 1": {TLS: TLSConfig{CAData: []byte{1}}},
|
||||
"cadata 2": {TLS: TLSConfig{CAData: []byte{2}}},
|
||||
@@ -137,13 +128,6 @@ func TestTLSConfigKey(t *testing.T) {
|
||||
GetCert: func() (*tls.Certificate, error) { return nil, nil },
|
||||
},
|
||||
},
|
||||
"getCert3": {
|
||||
TLS: TLSConfig{
|
||||
KeyData: []byte{1},
|
||||
GetCert: getCert,
|
||||
GetCertHolder: getCertHolder,
|
||||
},
|
||||
},
|
||||
"getCert1, key 2": {
|
||||
TLS: TLSConfig{
|
||||
KeyData: []byte{2},
|
||||
|
||||
@@ -68,11 +68,7 @@ type Config struct {
|
||||
WrapTransport WrapperFunc
|
||||
|
||||
// Dial specifies the dial function for creating unencrypted TCP connections.
|
||||
// If specified, this transport will be non-cacheable unless DialHolder is also set.
|
||||
Dial func(ctx context.Context, network, address string) (net.Conn, error)
|
||||
// DialHolder can be populated to make transport configs cacheable.
|
||||
// If specified, DialHolder.Dial must be equal to Dial.
|
||||
DialHolder *DialHolder
|
||||
|
||||
// Proxy is the proxy func to be used for all requests made by this
|
||||
// transport. If Proxy is nil, http.ProxyFromEnvironment is used. If Proxy
|
||||
@@ -82,11 +78,6 @@ type Config struct {
|
||||
Proxy func(*http.Request) (*url.URL, error)
|
||||
}
|
||||
|
||||
// DialHolder is used to make the wrapped function comparable so that it can be used as a map key.
|
||||
type DialHolder struct {
|
||||
Dial func(ctx context.Context, network, address string) (net.Conn, error)
|
||||
}
|
||||
|
||||
// ImpersonationConfig has all the available impersonation options
|
||||
type ImpersonationConfig struct {
|
||||
// UserName matches user.Info.GetName()
|
||||
@@ -152,15 +143,5 @@ type TLSConfig struct {
|
||||
// To use only http/1.1, set to ["http/1.1"].
|
||||
NextProtos []string
|
||||
|
||||
// Callback that returns a TLS client certificate. CertData, CertFile, KeyData and KeyFile supercede this field.
|
||||
// If specified, this transport is non-cacheable unless CertHolder is populated.
|
||||
GetCert func() (*tls.Certificate, error)
|
||||
// CertHolder can be populated to make transport configs that set GetCert cacheable.
|
||||
// If set, CertHolder.GetCert must be equal to GetCert.
|
||||
GetCertHolder *GetCertHolder
|
||||
}
|
||||
|
||||
// GetCertHolder is used to make the wrapped function comparable so that it can be used as a map key.
|
||||
type GetCertHolder struct {
|
||||
GetCert func() (*tls.Certificate, error)
|
||||
GetCert func() (*tls.Certificate, error) // Callback that returns a TLS client certificate. CertData, CertFile, KeyData and KeyFile supercede this field.
|
||||
}
|
||||
|
||||
@@ -24,7 +24,6 @@ import (
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -40,10 +39,6 @@ func New(config *Config) (http.RoundTripper, error) {
|
||||
return nil, fmt.Errorf("using a custom transport with TLS certificate options or the insecure flag is not allowed")
|
||||
}
|
||||
|
||||
if !isValidHolders(config) {
|
||||
return nil, fmt.Errorf("misconfigured holder for dialer or cert callback")
|
||||
}
|
||||
|
||||
var (
|
||||
rt http.RoundTripper
|
||||
err error
|
||||
@@ -61,26 +56,6 @@ func New(config *Config) (http.RoundTripper, error) {
|
||||
return HTTPWrappersForConfig(config, rt)
|
||||
}
|
||||
|
||||
func isValidHolders(config *Config) bool {
|
||||
if config.TLS.GetCertHolder != nil {
|
||||
if config.TLS.GetCertHolder.GetCert == nil ||
|
||||
config.TLS.GetCert == nil ||
|
||||
reflect.ValueOf(config.TLS.GetCertHolder.GetCert).Pointer() != reflect.ValueOf(config.TLS.GetCert).Pointer() {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
if config.DialHolder != nil {
|
||||
if config.DialHolder.Dial == nil ||
|
||||
config.Dial == nil ||
|
||||
reflect.ValueOf(config.DialHolder.Dial).Pointer() != reflect.ValueOf(config.Dial).Pointer() {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// TLSConfigFor returns a tls.Config that will provide the transport level security defined
|
||||
// by the provided Config. Will return nil if no transport level security is requested.
|
||||
func TLSConfigFor(c *Config) (*tls.Config, error) {
|
||||
|
||||
@@ -21,7 +21,6 @@ import (
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"testing"
|
||||
)
|
||||
@@ -95,13 +94,6 @@ stR0Yiw0buV6DL/moUO0HIM9Bjh96HJp+LxiIS6UCdIhMPp5HoQa
|
||||
)
|
||||
|
||||
func TestNew(t *testing.T) {
|
||||
globalGetCert := &GetCertHolder{
|
||||
GetCert: func() (*tls.Certificate, error) { return nil, nil },
|
||||
}
|
||||
globalDial := &DialHolder{
|
||||
Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil },
|
||||
}
|
||||
|
||||
testCases := map[string]struct {
|
||||
Config *Config
|
||||
Err bool
|
||||
@@ -263,144 +255,6 @@ func TestNew(t *testing.T) {
|
||||
},
|
||||
},
|
||||
},
|
||||
"nil holders and nil regular": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: nil,
|
||||
GetCertHolder: nil,
|
||||
},
|
||||
Dial: nil,
|
||||
DialHolder: nil,
|
||||
},
|
||||
Err: false,
|
||||
TLS: false,
|
||||
TLSCert: false,
|
||||
TLSErr: false,
|
||||
Default: true,
|
||||
Insecure: false,
|
||||
DefaultRoots: false,
|
||||
},
|
||||
"nil holders and non-nil regular get cert": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: func() (*tls.Certificate, error) { return nil, nil },
|
||||
GetCertHolder: nil,
|
||||
},
|
||||
Dial: nil,
|
||||
DialHolder: nil,
|
||||
},
|
||||
Err: false,
|
||||
TLS: true,
|
||||
TLSCert: true,
|
||||
TLSErr: false,
|
||||
Default: false,
|
||||
Insecure: false,
|
||||
DefaultRoots: true,
|
||||
},
|
||||
"nil holders and non-nil regular dial": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: nil,
|
||||
GetCertHolder: nil,
|
||||
},
|
||||
Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil },
|
||||
DialHolder: nil,
|
||||
},
|
||||
Err: false,
|
||||
TLS: true,
|
||||
TLSCert: false,
|
||||
TLSErr: false,
|
||||
Default: false,
|
||||
Insecure: false,
|
||||
DefaultRoots: true,
|
||||
},
|
||||
"non-nil dial holder and nil regular": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: nil,
|
||||
GetCertHolder: nil,
|
||||
},
|
||||
Dial: nil,
|
||||
DialHolder: &DialHolder{},
|
||||
},
|
||||
Err: true,
|
||||
},
|
||||
"non-nil cert holder and nil regular": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: nil,
|
||||
GetCertHolder: &GetCertHolder{},
|
||||
},
|
||||
Dial: nil,
|
||||
DialHolder: nil,
|
||||
},
|
||||
Err: true,
|
||||
},
|
||||
"non-nil dial holder and non-nil regular": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: nil,
|
||||
GetCertHolder: nil,
|
||||
},
|
||||
Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil },
|
||||
DialHolder: &DialHolder{},
|
||||
},
|
||||
Err: true,
|
||||
},
|
||||
"non-nil cert holder and non-nil regular": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: func() (*tls.Certificate, error) { return nil, nil },
|
||||
GetCertHolder: &GetCertHolder{},
|
||||
},
|
||||
Dial: nil,
|
||||
DialHolder: nil,
|
||||
},
|
||||
Err: true,
|
||||
},
|
||||
"non-nil dial holder+internal and non-nil regular": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: nil,
|
||||
GetCertHolder: nil,
|
||||
},
|
||||
Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil },
|
||||
DialHolder: &DialHolder{
|
||||
Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil },
|
||||
},
|
||||
},
|
||||
Err: true,
|
||||
},
|
||||
"non-nil cert holder+internal and non-nil regular": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: func() (*tls.Certificate, error) { return nil, nil },
|
||||
GetCertHolder: &GetCertHolder{
|
||||
GetCert: func() (*tls.Certificate, error) { return nil, nil },
|
||||
},
|
||||
},
|
||||
Dial: nil,
|
||||
DialHolder: nil,
|
||||
},
|
||||
Err: true,
|
||||
},
|
||||
"non-nil holders+internal and non-nil regular with correct address": {
|
||||
Config: &Config{
|
||||
TLS: TLSConfig{
|
||||
GetCert: globalGetCert.GetCert,
|
||||
GetCertHolder: globalGetCert,
|
||||
},
|
||||
Dial: globalDial.Dial,
|
||||
DialHolder: globalDial,
|
||||
},
|
||||
Err: false,
|
||||
TLS: true,
|
||||
TLSCert: true,
|
||||
TLSErr: false,
|
||||
Default: false,
|
||||
Insecure: false,
|
||||
DefaultRoots: true,
|
||||
},
|
||||
}
|
||||
for k, testCase := range testCases {
|
||||
t.Run(k, func(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user