From a13526af7b1ee8c2263242c1f98a3b9fd7bab915 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Thu, 19 Jan 2023 23:26:10 +0000 Subject: [PATCH] chore(deps): bump k8s.io/controller-manager from 0.25.5 to 0.25.6 Bumps [k8s.io/controller-manager](https://github.com/kubernetes/controller-manager) from 0.25.5 to 0.25.6. - [Release notes](https://github.com/kubernetes/controller-manager/releases) - [Commits](https://github.com/kubernetes/controller-manager/compare/v0.25.5...v0.25.6) --- updated-dependencies: - dependency-name: k8s.io/controller-manager dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] --- go.mod | 12 +- go.sum | 24 +-- .../fairqueuing/queueset/queueset.go | 50 ++++-- vendor/k8s.io/client-go/rest/request.go | 36 +++- vendor/k8s.io/client-go/rest/with_retry.go | 17 +- vendor/modules.txt | 14 +- .../konnectivity-client/pkg/client/client.go | 119 +++++++++---- .../konnectivity-client/pkg/client/conn.go | 9 + .../pkg/client/metrics/metrics.go | 162 ++++++++++++++++++ .../pkg/common/metrics/metrics.go | 78 +++++++++ 10 files changed, 427 insertions(+), 94 deletions(-) create mode 100644 vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/metrics/metrics.go create mode 100644 vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics/metrics.go diff --git a/go.mod b/go.mod index b631af6061..af66081225 100644 --- a/go.mod +++ b/go.mod @@ -22,14 +22,14 @@ require ( golang.org/x/crypto v0.5.0 golang.org/x/sys v0.4.0 golang.org/x/text v0.6.0 - k8s.io/api v0.25.5 + k8s.io/api v0.25.6 k8s.io/apimachinery v0.25.6 - k8s.io/apiserver v0.25.5 - k8s.io/client-go v0.25.5 + k8s.io/apiserver v0.25.6 + k8s.io/client-go v0.25.6 k8s.io/cloud-provider v0.25.5 - k8s.io/component-base v0.25.5 + k8s.io/component-base v0.25.6 k8s.io/component-helpers v0.25.5 - k8s.io/controller-manager v0.25.5 + k8s.io/controller-manager v0.25.6 k8s.io/klog/v2 v2.80.0 k8s.io/kubelet v0.25.5 k8s.io/utils v0.0.0-20220728103510-ee6ede2d64ed @@ -124,7 +124,7 @@ require ( gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect k8s.io/kube-openapi v0.0.0-20220803162953-67bda5d908f1 // indirect - sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.33 // indirect + sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.35 // indirect sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 // indirect sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect ) diff --git a/go.sum b/go.sum index f7f2bac482..a5dadab281 100644 --- a/go.sum +++ b/go.sum @@ -766,22 +766,22 @@ 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.5 h1:mqyHf7aoaYMpdvO87mqpol+Qnsmo+y09S0PMIXwiZKo= -k8s.io/api v0.25.5/go.mod h1:RzplZX0Z8rV/WhSTfEvnyd91bBhBQTRWo85qBQwRmb8= +k8s.io/api v0.25.6 h1:LwDY2H6kD/3R8TekJYYaJWOdekNdXDO44eVpX6sNtJA= +k8s.io/api v0.25.6/go.mod h1:bVp01KUcl8VUHFBTJMOknWNo7XvR0cMbeTTuFg1zCUs= k8s.io/apimachinery v0.25.6 h1:r6KIF2AHwLqFfZ0LcOA3I11SF62YZK83dxj1fn14NOQ= k8s.io/apimachinery v0.25.6/go.mod h1:1S2i1QHkmxc8+EZCIxe/fX5hpldVXk4gvnJInMEb8D4= -k8s.io/apiserver v0.25.5 h1:oC6pd5Z/q8WKksJApbnnK0mODqIz/jKWw3Jk4QqIVXM= -k8s.io/apiserver v0.25.5/go.mod h1:iCRtEN+C0EsNBcbhmDJp41M7cspJM54VWoKyl62gGOU= -k8s.io/client-go v0.25.5 h1:7QWVK0Ph4bLn0UwotPTc2FTgm8shreQXyvXnnHDd8rE= -k8s.io/client-go v0.25.5/go.mod h1:bOeoaUUdpyz3WDFGo+Xm3nOQFh2KuYXRDwrvbAPtFQA= +k8s.io/apiserver v0.25.6 h1:32mn8HAlsEl1tpuiVmhAl0YCVkOugjybsJ6l6kf0c8k= +k8s.io/apiserver v0.25.6/go.mod h1:IEp2B2/FvQ8GmdspscUoUS0iFF/GGc6NVrJ/cTM4OaA= +k8s.io/client-go v0.25.6 h1:CHxACHi0DijmlYyUR7ooZoXnD5P8jYLgBHcxp775x/U= +k8s.io/client-go v0.25.6/go.mod h1:s9mMAGFYiH3Z66j7BESzu0GEradT9GQ2LjFf/YRrnyc= k8s.io/cloud-provider v0.25.5 h1:+EYx+3MVx3fMXVztq5Y2IshEVWaJvCEzH98Il8wTk0Q= k8s.io/cloud-provider v0.25.5/go.mod h1:aaPi/WFKX+yhQ6ctDFDEUkKnB9WfyAq5t+vqtXRaEeE= -k8s.io/component-base v0.25.5 h1:tVni0kgpceq71MDMBSixp8Y621YGvTS/1zq3RABgX9A= -k8s.io/component-base v0.25.5/go.mod h1:9J+e9uIUwUOG2x5q5+aaOR0b8QI5OIqwqPAbeODkYpc= +k8s.io/component-base v0.25.6 h1:v3ci6FbXFcxpjyQJaaLq0MgzT3vyFzwUDWtO+KRv9Bk= +k8s.io/component-base v0.25.6/go.mod h1:k7DfcfJ8cOI6A2xTCfU5LxsnXV+lWw1ME8cRCHzIh6o= k8s.io/component-helpers v0.25.5 h1:9lzSVcNmysgBIVFkdauELnDOrd3iJBBqilDeexpF7Wk= k8s.io/component-helpers v0.25.5/go.mod h1:K5wb7vW1QVzCRfKE6Fe0tLv4LOQLz/Yj51MSTLoQBOI= -k8s.io/controller-manager v0.25.5 h1:8NnInbCfet1YsI2ZOVKpRXPGsAzZKyf521SSAkucYSc= -k8s.io/controller-manager v0.25.5/go.mod h1:0T7CrJQUNRXTm7jmqZJVd5dm4BGThMxqw/g7RJ4GnQg= +k8s.io/controller-manager v0.25.6 h1:m6TOcxj20Na0mBg3d8596XdEsgdGwa0Fp36YsgS1G5o= +k8s.io/controller-manager v0.25.6/go.mod h1:NSq3oQUkOXxcj0RZCSxRBOXbEOQSZ1cYnVbaQb1LaQE= k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE= k8s.io/klog/v2 v2.80.0 h1:lyJt0TWMPaGoODa8B8bUuxgHS3W/m/bNr2cca3brA/g= k8s.io/klog/v2 v2.80.0/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0= @@ -794,8 +794,8 @@ k8s.io/utils v0.0.0-20220728103510-ee6ede2d64ed/go.mod h1:jPW/WVKK9YHAvNhRxK0md/ rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8= rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0= rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA= -sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.33 h1:LYqFq+6Cj2D0gFfrJvL7iElD4ET6ir3VDdhDdTK7rgc= -sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.33/go.mod h1:soWkSNf2tZC7aMibXEqVhCd73GOY5fJikn8qbdzemB0= +sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.35 h1:+xBL5uTc+BkPBwmMi3vYfUJjq+N3K+H6PXeETwf5cPI= +sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.35/go.mod h1:WxjusMwXlKzfAs4p9km6XJRndVt2FROgMVCE4cdohFo= sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 h1:iXTIw73aPyC+oRdyqqvVJuloN1p0AC/kzH07hu3NE+k= sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0= sigs.k8s.io/structured-merge-diff/v4 v4.2.3 h1:PRbqxJClWWYMNV1dhaG4NsibJbArud9kFxnAMREiWFE= diff --git a/vendor/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go b/vendor/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go index faa670bf17..684e76da64 100644 --- a/vendor/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go +++ b/vendor/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/queueset.go @@ -72,11 +72,19 @@ type queueSetCompleter struct { // described in this package's doc, and a pointer to one implements // the QueueSet interface. The fields listed before the lock // should not be changed; the fields listed after the -// lock must be accessed only while holding the lock. The methods of -// this type follow the naming convention that the suffix "Locked" -// means the caller must hold the lock; for a method whose name does -// not end in "Locked" either acquires the lock or does not care about -// locking. +// lock must be accessed only while holding the lock. +// +// The methods of this type follow the naming convention that the +// suffix "Locked" means the caller must hold the lock; for a method +// whose name does not end in "Locked" either acquires the lock or +// does not care about locking. +// +// The methods of this type also follow the convention that the suffix +// "ToBoundLocked" means that the caller may have to follow up with a +// call to `boundNextDispatchLocked`. This is so for a method that +// changes what request is oldest in a queue, because that change means +// that the anti-windup hack in boundNextDispatchLocked needs to be +// applied wrt the revised oldest request in the queue. type queueSet struct { clock eventclock.Interface estimatedServiceDuration time.Duration @@ -396,7 +404,9 @@ func (req *request) wait() (bool, bool) { // TODO(aaron-prindle) add metrics for this case klog.V(5).Infof("QS(%s): Ejecting request %#+v %#+v from its queue", qs.qCfg.Name, req.descr1, req.descr2) // remove the request from the queue as it has timed out + queue := req.queue if req.removeFromQueueLocked() != nil { + defer qs.boundNextDispatchLocked(queue) qs.totRequestsWaiting-- metrics.AddReject(req.ctx, qs.qCfg.Name, req.fsName, "cancelled") metrics.AddRequestsInQueues(req.ctx, qs.qCfg.Name, req.fsName, -1) @@ -521,7 +531,7 @@ func (qs *queueSet) timeoutOldRequestsAndRejectOrEnqueueLocked(ctx context.Conte queueIdx := qs.shuffleShardLocked(hashValue, descr1, descr2) queue := qs.queues[queueIdx] // The next step is the logic to reject requests that have been waiting too long - qs.removeTimedOutRequestsFromQueueLocked(queue, fsName) + qs.removeTimedOutRequestsFromQueueToBoundLocked(queue, fsName) // NOTE: currently timeout is only checked for each new request. This means that there can be // requests that are in the queue longer than the timeout if there are no new requests // We prefer the simplicity over the promptness, at least for now. @@ -543,7 +553,7 @@ func (qs *queueSet) timeoutOldRequestsAndRejectOrEnqueueLocked(ctx context.Conte queueNoteFn: queueNoteFn, workEstimate: qs.completeWorkEstimate(workEstimate), } - if ok := qs.rejectOrEnqueueLocked(req); !ok { + if ok := qs.rejectOrEnqueueToBoundLocked(req); !ok { return nil } metrics.ObserveQueueLength(ctx, qs.qCfg.Name, fsName, queue.requests.Length()) @@ -583,9 +593,9 @@ func (qs *queueSet) shuffleShardLocked(hashValue uint64, descr1, descr2 interfac return bestQueueIdx } -// removeTimedOutRequestsFromQueueLocked rejects old requests that have been enqueued +// removeTimedOutRequestsFromQueueToBoundLocked rejects old requests that have been enqueued // past the requestWaitLimit -func (qs *queueSet) removeTimedOutRequestsFromQueueLocked(queue *queue, fsName string) { +func (qs *queueSet) removeTimedOutRequestsFromQueueToBoundLocked(queue *queue, fsName string) { timeoutCount := 0 now := qs.clock.Now() reqs := queue.requests @@ -616,11 +626,11 @@ func (qs *queueSet) removeTimedOutRequestsFromQueueLocked(queue *queue, fsName s } } -// rejectOrEnqueueLocked rejects or enqueues the newly arrived +// rejectOrEnqueueToBoundLocked rejects or enqueues the newly arrived // request, which has been assigned to a queue. If up against the // queue length limit and the concurrency limit then returns false. // Otherwise enqueues and returns true. -func (qs *queueSet) rejectOrEnqueueLocked(request *request) bool { +func (qs *queueSet) rejectOrEnqueueToBoundLocked(request *request) bool { queue := request.queue curQueueLength := queue.requests.Length() // rejects the newly arrived request if resource criteria not met @@ -629,12 +639,12 @@ func (qs *queueSet) rejectOrEnqueueLocked(request *request) bool { return false } - qs.enqueueLocked(request) + qs.enqueueToBoundLocked(request) return true } // enqueues a request into its queue. -func (qs *queueSet) enqueueLocked(request *request) { +func (qs *queueSet) enqueueToBoundLocked(request *request) { queue := request.queue now := qs.clock.Now() if queue.requests.Length() == 0 && queue.requestsExecuting == 0 { @@ -693,7 +703,7 @@ func (qs *queueSet) dispatchSansQueueLocked(ctx context.Context, workEstimate *f // be false when either all queues are empty or the request at the head // of the next queue cannot be dispatched. func (qs *queueSet) dispatchLocked() bool { - queue, request := qs.findDispatchQueueLocked() + queue, request := qs.findDispatchQueueToBoundLocked() if queue == nil { return false } @@ -729,6 +739,11 @@ func (qs *queueSet) dispatchLocked() bool { request.workEstimate, queue.index, queue.nextDispatchR, queue.requests.Length(), queue.requestsExecuting, queue.seatsInUse, qs.totSeatsInUse) } // When a request is dequeued for service -> qs.virtualStart += G * width + if request.totalWork() > rDecrement/100 { // A single increment should never be so big + klog.Errorf("QS(%s) at t=%s R=%v: dispatching request %#+v %#+v with implausibly high work %v from queue %d with start R %v", + qs.qCfg.Name, request.startTime.Format(nsTimeFmt), qs.currentR, request.descr1, request.descr2, + request.workEstimate, queue.index, queue.nextDispatchR) + } queue.nextDispatchR += request.totalWork() return true } @@ -756,11 +771,12 @@ func (qs *queueSet) canAccommodateSeatsLocked(seats int) bool { return true } -// findDispatchQueueLocked examines the queues in round robin order and +// findDispatchQueueToBoundLocked examines the queues in round robin order and // returns the first one of those for which the virtual finish time of // the oldest waiting request is minimal, and also returns that request. -// Returns nils if the head of the selected queue can not be dispatched now. -func (qs *queueSet) findDispatchQueueLocked() (*queue, *request) { +// Returns nils if the head of the selected queue can not be dispatched now, +// in which case the caller does not need to follow up with`qs.boundNextDispatchLocked`. +func (qs *queueSet) findDispatchQueueToBoundLocked() (*queue, *request) { minVirtualFinish := fqrequest.MaxSeatSeconds sMin := fqrequest.MaxSeatSeconds dsMin := fqrequest.MaxSeatSeconds diff --git a/vendor/k8s.io/client-go/rest/request.go b/vendor/k8s.io/client-go/rest/request.go index acf3113615..a1f0a591d3 100644 --- a/vendor/k8s.io/client-go/rest/request.go +++ b/vendor/k8s.io/client-go/rest/request.go @@ -34,6 +34,7 @@ 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" @@ -116,8 +117,11 @@ type Request struct { subresource string // output - err error - body io.Reader + err error + + // only one of body / bodyBytes may be set. requests using body are not retriable. + body io.Reader + bodyBytes []byte retryFn requestRetryFunc } @@ -443,12 +447,15 @@ func (r *Request) Body(obj interface{}) *Request { return r } glogBody("Request Body", data) - r.body = bytes.NewReader(data) + r.body = nil + r.bodyBytes = data case []byte: glogBody("Request Body", t) - r.body = bytes.NewReader(t) + r.body = nil + r.bodyBytes = 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() { @@ -465,7 +472,8 @@ func (r *Request) Body(obj interface{}) *Request { return r } glogBody("Request Body", data) - r.body = bytes.NewReader(data) + r.body = nil + r.bodyBytes = data r.SetHeader("Content-Type", r.c.content.ContentType) default: r.err = fmt.Errorf("unknown type used for body: %+v", obj) @@ -825,9 +833,7 @@ 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) @@ -889,8 +895,20 @@ 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, r.body) + req, err := http.NewRequest(r.verb, url, body) if err != nil { return nil, err } diff --git a/vendor/k8s.io/client-go/rest/with_retry.go b/vendor/k8s.io/client-go/rest/with_retry.go index bdcc6f3a98..ab26947540 100644 --- a/vendor/k8s.io/client-go/rest/with_retry.go +++ b/vendor/k8s.io/client-go/rest/with_retry.go @@ -154,6 +154,11 @@ 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 { @@ -210,18 +215,6 @@ 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) diff --git a/vendor/modules.txt b/vendor/modules.txt index 8f44d405ea..cc67839659 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -641,7 +641,7 @@ gopkg.in/yaml.v2 # gopkg.in/yaml.v3 v3.0.1 ## explicit gopkg.in/yaml.v3 -# k8s.io/api v0.25.5 +# k8s.io/api v0.25.6 ## explicit; go 1.19 k8s.io/api/admission/v1 k8s.io/api/admission/v1beta1 @@ -745,7 +745,7 @@ k8s.io/apimachinery/pkg/version k8s.io/apimachinery/pkg/watch k8s.io/apimachinery/third_party/forked/golang/json k8s.io/apimachinery/third_party/forked/golang/reflect -# k8s.io/apiserver v0.25.5 +# k8s.io/apiserver v0.25.6 ## explicit; go 1.19 k8s.io/apiserver/pkg/admission k8s.io/apiserver/pkg/admission/configuration @@ -877,7 +877,7 @@ k8s.io/apiserver/plugin/pkg/audit/truncate k8s.io/apiserver/plugin/pkg/audit/webhook k8s.io/apiserver/plugin/pkg/authenticator/token/webhook k8s.io/apiserver/plugin/pkg/authorizer/webhook -# k8s.io/client-go v0.25.5 +# k8s.io/client-go v0.25.6 ## explicit; go 1.19 k8s.io/client-go/applyconfigurations/admissionregistration/v1 k8s.io/client-go/applyconfigurations/admissionregistration/v1beta1 @@ -1184,7 +1184,7 @@ k8s.io/cloud-provider/service/helpers k8s.io/cloud-provider/volume k8s.io/cloud-provider/volume/errors k8s.io/cloud-provider/volume/helpers -# k8s.io/component-base v0.25.5 +# k8s.io/component-base v0.25.6 ## explicit; go 1.19 k8s.io/component-base/cli/flag k8s.io/component-base/cli/globalflag @@ -1212,7 +1212,7 @@ k8s.io/component-base/version # k8s.io/component-helpers v0.25.5 ## explicit; go 1.19 k8s.io/component-helpers/node/util -# k8s.io/controller-manager v0.25.5 +# k8s.io/controller-manager v0.25.6 ## explicit; go 1.19 k8s.io/controller-manager/app k8s.io/controller-manager/config @@ -1275,9 +1275,11 @@ k8s.io/utils/path k8s.io/utils/pointer k8s.io/utils/strings/slices k8s.io/utils/trace -# sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.33 +# sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.35 ## explicit; go 1.17 sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client +sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/metrics +sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client # sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 ## explicit; go 1.18 diff --git a/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/client.go b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/client.go index c55070ae63..cb186cefc2 100644 --- a/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/client.go +++ b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/client.go @@ -29,6 +29,9 @@ import ( "google.golang.org/grpc" "k8s.io/klog/v2" + + "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/metrics" + commonmetrics "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics" "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client" ) @@ -131,6 +134,9 @@ type grpcTunnel struct { // closing should only be accessed through atomic methods. // TODO: switch this to an atomic.Bool once the client is exclusively buit with go1.19+ closing uint32 + + // Stores the current metrics.ClientConnectionStatus + prevStatus atomic.Value } type clientConn interface { @@ -139,6 +145,11 @@ type clientConn interface { var _ clientConn = &grpc.ClientConn{} +var ( + // Expose metrics for client to register. + Metrics = metrics.Metrics +) + // CreateSingleUseGrpcTunnel creates a Tunnel to dial to a remote server through a // gRPC based proxy service. // Currently, a single tunnel supports a single connection, and the tunnel is closed when the connection is terminated @@ -177,7 +188,7 @@ func CreateSingleUseGrpcTunnelWithContext(createCtx, tunnelCtx context.Context, } func newUnstartedTunnel(stream client.ProxyService_ProxyClient, c clientConn) *grpcTunnel { - return &grpcTunnel{ + t := grpcTunnel{ stream: stream, clientConn: c, pendingDial: pendingDialManager{pendingDials: make(map[int64]pendingDial)}, @@ -185,6 +196,36 @@ func newUnstartedTunnel(stream client.ProxyService_ProxyClient, c clientConn) *g readTimeoutSeconds: 10, done: make(chan struct{}), } + s := metrics.ClientConnectionStatusCreated + t.prevStatus.Store(s) + metrics.Metrics.GetClientConnectionsMetric().WithLabelValues(string(s)).Inc() + return &t +} + +func (t *grpcTunnel) updateMetric(status metrics.ClientConnectionStatus) { + select { + case <-t.Done(): + return + default: + } + + prevStatus := t.prevStatus.Swap(status).(metrics.ClientConnectionStatus) + + m := metrics.Metrics.GetClientConnectionsMetric() + m.WithLabelValues(string(prevStatus)).Dec() + m.WithLabelValues(string(status)).Inc() +} + +// closeMetric should be called exactly once to finalize client_connections metric. +func (t *grpcTunnel) closeMetric() { + select { + case <-t.Done(): + return + default: + } + prevStatus := t.prevStatus.Load().(metrics.ClientConnectionStatus) + + metrics.Metrics.GetClientConnectionsMetric().WithLabelValues(string(prevStatus)).Dec() } func (t *grpcTunnel) serve(tunnelCtx context.Context) { @@ -196,19 +237,29 @@ func (t *grpcTunnel) serve(tunnelCtx context.Context) { // close any channels remaining for these connections. t.conns.closeAll() + t.closeMetric() + close(t.done) }() for { pkt, err := t.stream.Recv() - if err == io.EOF || t.isClosing() { + if err == io.EOF { return } + const segment = commonmetrics.SegmentToClient + isClosing := t.isClosing() if err != nil || pkt == nil { - klog.ErrorS(err, "stream read failure") + if !isClosing { + klog.ErrorS(err, "stream read failure") + } + metrics.Metrics.ObserveStreamErrorNoPacket(segment, err) + return + } + metrics.Metrics.ObservePacket(segment, pkt.Type) + if isClosing { return } - klog.V(5).InfoS("[tracing] recv packet", "type", pkt.Type) switch pkt.Type { @@ -222,13 +273,19 @@ func (t *grpcTunnel) serve(tunnelCtx context.Context) { // 2. grpcTunnel.DialContext() returned early due to a dial timeout or the client canceling the context // // In either scenario, we should return here and close the tunnel as it is no longer needed. - klog.V(1).InfoS("DialResp not recognized; dropped", "connectionID", resp.ConnectID, "dialID", resp.Random) + kvs := []interface{}{"dialID", resp.Random, "connectID", resp.ConnectID} + if resp.Error != "" { + kvs = append(kvs, "error", resp.Error) + } + klog.V(1).InfoS("DialResp not recognized; dropped", kvs...) return } result := dialResult{connid: resp.ConnectID} if resp.Error != "" { - result.err = &dialFailure{resp.Error, DialFailureEndpoint} + result.err = &dialFailure{resp.Error, metrics.DialFailureEndpoint} + } else { + t.updateMetric(metrics.ClientConnectionStatusOk) } select { // try to send to the result channel @@ -263,7 +320,7 @@ func (t *grpcTunnel) serve(tunnelCtx context.Context) { klog.V(1).InfoS("DIAL_CLS after dial finished", "dialID", resp.Random) } else { result := dialResult{ - err: &dialFailure{"dial closed", DialFailureDialClosed}, + err: &dialFailure{"dial closed", metrics.DialFailureDialClosed}, } select { case pendingDial.resultCh <- result: @@ -316,6 +373,15 @@ func (t *grpcTunnel) serve(tunnelCtx context.Context) { // Dial connects to the address on the named network, similar to // what net.Dial does. The only supported protocol is tcp. func (t *grpcTunnel) DialContext(requestCtx context.Context, protocol, address string) (net.Conn, error) { + conn, err := t.dialContext(requestCtx, protocol, address) + if err != nil { + _, reason := GetDialFailureReason(err) + metrics.Metrics.ObserveDialFailure(reason) + } + return conn, err +} + +func (t *grpcTunnel) dialContext(requestCtx context.Context, protocol, address string) (net.Conn, error) { select { case <-t.done: return nil, errors.New("tunnel is closed") @@ -326,6 +392,8 @@ func (t *grpcTunnel) DialContext(requestCtx context.Context, protocol, address s return nil, errors.New("protocol not supported") } + t.updateMetric(metrics.ClientConnectionStatusDialing) + random := rand.Int63() /* #nosec G404 */ // This channel is closed once we're returning and no longer waiting on resultCh @@ -350,8 +418,11 @@ func (t *grpcTunnel) DialContext(requestCtx context.Context, protocol, address s } klog.V(5).InfoS("[tracing] send packet", "type", req.Type) + const segment = commonmetrics.SegmentFromClient + metrics.Metrics.ObservePacket(segment, req.Type) err := t.stream.Send(req) if err != nil { + metrics.Metrics.ObserveStreamError(segment, err, req.Type) return nil, err } @@ -375,14 +446,14 @@ func (t *grpcTunnel) DialContext(requestCtx context.Context, protocol, address s case <-time.After(30 * time.Second): klog.V(5).InfoS("Timed out waiting for DialResp", "dialID", random) go t.closeDial(random) - return nil, &dialFailure{"dial timeout, backstop", DialFailureTimeout} + return nil, &dialFailure{"dial timeout, backstop", metrics.DialFailureTimeout} case <-requestCtx.Done(): klog.V(5).InfoS("Context canceled waiting for DialResp", "ctxErr", requestCtx.Err(), "dialID", random) go t.closeDial(random) - return nil, &dialFailure{"dial timeout, context", DialFailureContext} + return nil, &dialFailure{"dial timeout, context", metrics.DialFailureContext} case <-t.done: klog.V(5).InfoS("Tunnel closed while waiting for DialResp", "dialID", random) - return nil, &dialFailure{"tunnel closed", DialFailureTunnelClosed} + return nil, &dialFailure{"tunnel closed", metrics.DialFailureTunnelClosed} } return c, nil @@ -402,7 +473,10 @@ func (t *grpcTunnel) closeDial(dialID int64) { }, }, } + const segment = commonmetrics.SegmentFromClient + metrics.Metrics.ObservePacket(segment, req.Type) if err := t.stream.Send(req); err != nil { + metrics.Metrics.ObserveStreamError(segment, err, req.Type) klog.V(5).InfoS("Failed to send DIAL_CLS", "err", err, "dialID", dialID) } t.closeTunnel() @@ -417,38 +491,19 @@ func (t *grpcTunnel) isClosing() bool { return atomic.LoadUint32(&t.closing) != 0 } -func GetDialFailureReason(err error) (isDialFailure bool, reason DialFailureReason) { +func GetDialFailureReason(err error) (isDialFailure bool, reason metrics.DialFailureReason) { var df *dialFailure if errors.As(err, &df) { return true, df.reason } - return false, DialFailureUnknown + return false, metrics.DialFailureUnknown } type dialFailure struct { msg string - reason DialFailureReason + reason metrics.DialFailureReason } func (df *dialFailure) Error() string { return df.msg } - -type DialFailureReason string - -const ( - DialFailureUnknown DialFailureReason = "unknown" - // DialFailureTimeout indicates the hard 30 second timeout was hit. - DialFailureTimeout DialFailureReason = "timeout" - // DialFailureContext indicates that the context was cancelled or reached it's deadline before - // the dial response was returned. - DialFailureContext DialFailureReason = "context" - // DialFailureEndpoint indicates that the konnectivity-agent was unable to reach the backend endpoint. - DialFailureEndpoint DialFailureReason = "endpoint" - // DialFailureDialClosed indicates that the client received a CloseDial response, indicating the - // connection was closed before the dial could complete. - DialFailureDialClosed DialFailureReason = "dialclosed" - // DialFailureTunnelClosed indicates that the client connection was closed before the dial could - // complete. - DialFailureTunnelClosed DialFailureReason = "tunnelclosed" -) diff --git a/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/conn.go b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/conn.go index f76b1e37a4..14384a62cb 100644 --- a/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/conn.go +++ b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/conn.go @@ -23,6 +23,9 @@ import ( "time" "k8s.io/klog/v2" + + "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/metrics" + commonmetrics "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics" "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client" ) @@ -62,8 +65,11 @@ func (c *conn) Write(data []byte) (n int, err error) { klog.V(5).InfoS("[tracing] send req", "type", req.Type) + const segment = commonmetrics.SegmentFromClient + metrics.Metrics.ObservePacket(segment, req.Type) err = c.stream.Send(req) if err != nil { + metrics.Metrics.ObserveStreamError(segment, err, req.Type) return 0, err } return len(data), err @@ -147,7 +153,10 @@ func (c *conn) Close() error { klog.V(5).InfoS("[tracing] send req", "type", req.Type) + const segment = commonmetrics.SegmentFromClient + metrics.Metrics.ObservePacket(segment, req.Type) if err := c.stream.Send(req); err != nil { + metrics.Metrics.ObserveStreamError(segment, err, req.Type) return err } diff --git a/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/metrics/metrics.go b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/metrics/metrics.go new file mode 100644 index 0000000000..03e9d94da8 --- /dev/null +++ b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/client/metrics/metrics.go @@ -0,0 +1,162 @@ +/* +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 metrics + +import ( + "sync" + + "github.com/prometheus/client_golang/prometheus" + + commonmetrics "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics" + "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client" +) + +const ( + Namespace = "konnectivity_network_proxy" + Subsystem = "client" +) + +var ( + // Metrics provides access to all client metrics. The client + // application is responsible for registering (via Metrics.RegisterMetrics). + Metrics = newMetrics() +) + +// ClientMetrics includes all the metrics of the konnectivity-client. +type ClientMetrics struct { + registerOnce sync.Once + streamPackets *prometheus.CounterVec + streamErrors *prometheus.CounterVec + dialFailures *prometheus.CounterVec + clientConns *prometheus.GaugeVec +} + +type DialFailureReason string + +const ( + DialFailureUnknown DialFailureReason = "unknown" + // DialFailureTimeout indicates the hard 30 second timeout was hit. + DialFailureTimeout DialFailureReason = "timeout" + // DialFailureContext indicates that the context was cancelled or reached it's deadline before + // the dial response was returned. + DialFailureContext DialFailureReason = "context" + // DialFailureEndpoint indicates that the konnectivity-agent was unable to reach the backend endpoint. + DialFailureEndpoint DialFailureReason = "endpoint" + // DialFailureDialClosed indicates that the client received a CloseDial response, indicating the + // connection was closed before the dial could complete. + DialFailureDialClosed DialFailureReason = "dialclosed" + // DialFailureTunnelClosed indicates that the client connection was closed before the dial could + // complete. + DialFailureTunnelClosed DialFailureReason = "tunnelclosed" +) + +type ClientConnectionStatus string + +const ( + // The connection is created but has not yet been dialed. + ClientConnectionStatusCreated ClientConnectionStatus = "created" + // The connection is pending dial response. + ClientConnectionStatusDialing ClientConnectionStatus = "dialing" + // The connection is established. + ClientConnectionStatusOk ClientConnectionStatus = "ok" + // The connection is closing. + ClientConnectionStatusClosing ClientConnectionStatus = "closing" +) + +func newMetrics() *ClientMetrics { + // The denominator (total dials started) for both + // dial_failure_total and dial_duration_seconds is the + // stream_packets_total (common metric), where segment is + // "from_client" and packet_type is "DIAL_REQ". + dialFailures := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: Subsystem, + Name: "dial_failure_total", + Help: "Number of dial failures observed, by reason (example: remote endpoint error)", + }, + []string{ + "reason", + }, + ) + clientConns := prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: Subsystem, + Name: "client_connections", + Help: "Number of open client connections, by status (Example: dialing)", + }, + []string{ + "status", + }, + ) + return &ClientMetrics{ + streamPackets: commonmetrics.MakeStreamPacketsTotalMetric(Namespace, Subsystem), + streamErrors: commonmetrics.MakeStreamErrorsTotalMetric(Namespace, Subsystem), + dialFailures: dialFailures, + clientConns: clientConns, + } +} + +// RegisterMetrics registers all metrics with the client application. +func (c *ClientMetrics) RegisterMetrics(r prometheus.Registerer) { + c.registerOnce.Do(func() { + r.MustRegister(c.streamPackets) + r.MustRegister(c.streamErrors) + r.MustRegister(c.dialFailures) + r.MustRegister(c.clientConns) + }) +} + +// LegacyRegisterMetrics registers all metrics via MustRegister func. +// TODO: remove this once https://github.com/kubernetes/kubernetes/pull/114293 is available. +func (c *ClientMetrics) LegacyRegisterMetrics(mustRegisterFn func(...prometheus.Collector)) { + c.registerOnce.Do(func() { + mustRegisterFn(c.streamPackets) + mustRegisterFn(c.streamErrors) + mustRegisterFn(c.dialFailures) + mustRegisterFn(c.clientConns) + }) +} + +// Reset resets the metrics. +func (c *ClientMetrics) Reset() { + c.streamPackets.Reset() + c.streamErrors.Reset() + c.dialFailures.Reset() + c.clientConns.Reset() +} + +func (c *ClientMetrics) ObserveDialFailure(reason DialFailureReason) { + c.dialFailures.WithLabelValues(string(reason)).Inc() +} + +func (c *ClientMetrics) GetClientConnectionsMetric() *prometheus.GaugeVec { + return c.clientConns +} + +func (c *ClientMetrics) ObservePacket(segment commonmetrics.Segment, packetType client.PacketType) { + commonmetrics.ObservePacket(c.streamPackets, segment, packetType) +} + +func (c *ClientMetrics) ObserveStreamErrorNoPacket(segment commonmetrics.Segment, err error) { + commonmetrics.ObserveStreamErrorNoPacket(c.streamErrors, segment, err) +} + +func (c *ClientMetrics) ObserveStreamError(segment commonmetrics.Segment, err error, packetType client.PacketType) { + commonmetrics.ObserveStreamError(c.streamErrors, segment, err, packetType) +} diff --git a/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics/metrics.go b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics/metrics.go new file mode 100644 index 0000000000..e8619f4720 --- /dev/null +++ b/vendor/sigs.k8s.io/apiserver-network-proxy/konnectivity-client/pkg/common/metrics/metrics.go @@ -0,0 +1,78 @@ +/* +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 metrics provides metric definitions and helpers used +// across konnectivity client, server, and agent. +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "google.golang.org/grpc/status" + + "sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client" +) + +// Segment identifies one of four tunnel segments (e.g. from server to agent). +type Segment string + +const ( + // SegmentFromClient indicates a packet from client to server. + SegmentFromClient Segment = "from_client" + // SegmentToClient indicates a packet from server to client. + SegmentToClient Segment = "to_client" + // SegmentFromAgent indicates a packet from agent to server. + SegmentFromAgent Segment = "from_agent" + // SegmentToAgent indicates a packet from server to agent. + SegmentToAgent Segment = "to_agent" +) + +func MakeStreamPacketsTotalMetric(namespace, subsystem string) *prometheus.CounterVec { + return prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "stream_packets_total", + Help: "Count of packets processed, by segment and packet type (example: from_client, DIAL_REQ)", + }, + []string{"segment", "packet_type"}, + ) +} + +func MakeStreamErrorsTotalMetric(namespace, subsystem string) *prometheus.CounterVec { + return prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "stream_errors_total", + Help: "Count of gRPC stream errors, by segment, grpc Code, packet type. (example: from_agent, Code.Unavailable, DIAL_RSP)", + }, + []string{"segment", "code", "packet_type"}, + ) +} + +func ObservePacket(m *prometheus.CounterVec, segment Segment, packetType client.PacketType) { + m.WithLabelValues(string(segment), packetType.String()).Inc() +} + +func ObserveStreamErrorNoPacket(m *prometheus.CounterVec, segment Segment, err error) { + code := status.Code(err) + m.WithLabelValues(string(segment), code.String(), "Unknown").Inc() +} + +func ObserveStreamError(m *prometheus.CounterVec, segment Segment, err error, packetType client.PacketType) { + code := status.Code(err) + m.WithLabelValues(string(segment), code.String(), packetType.String()).Inc() +}