github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-29 14:37:00 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #93877 from bjrara/test-metrics

Browse Source 
Bugfix missing metrics and test cases to ensure terminated request metrics is not missing
This commit is contained in:
Kubernetes Prow Robot 2020-11-02 19:14:14 -08:00 committed by GitHub
commit fe654797b5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 370 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
staging/src/k8s.io/apiserver/pkg/server/filters/BUILD
View File

@ -13,19 +13,28 @@ go_test(
"cors_test.go",
"goaway_test.go",
"maxinflight_test.go",
"priority-and-fairness_test.go",
"timeout_test.go",
],
embed = [":go_default_library"],
deps = [
"//staging/src/k8s.io/api/flowcontrol/v1alpha1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/diff:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/apis/flowcontrol/bootstrap:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/authentication/user:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/endpoints/filters:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/endpoints/metrics:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/mux:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/flowcontrol:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics:go_default_library",
"//staging/src/k8s.io/component-base/metrics/legacyregistry:go_default_library",
"//vendor/golang.org/x/net/http2:go_default_library",
],
)

11
staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness.go
View File

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"net/http"
"sync"
"sync/atomic"
fcv1a1 "k8s.io/api/flowcontrol/v1alpha1"
@ -57,6 +58,9 @@ var waitingMark = &requestWatermark{
mutatingObserver: fcmetrics.ReadWriteConcurrencyObserverPairGenerator.Generate(1, 1, []string{epmetrics.MutatingKind}).RequestsWaiting,
}
// apfStartOnce is used to avoid sharing one-time mutex with maxinflight handler
var apfStartOnce sync.Once
var atomicMutatingExecuting, atomicReadOnlyExecuting int32
var atomicMutatingWaiting, atomicReadOnlyWaiting int32
@ -73,6 +77,8 @@ func WithPriorityAndFairness(
}
startOnce.Do(func() {
startRecordingUsage(watermark)
})
apfStartOnce.Do(func() {
startRecordingUsage(waitingMark)
})
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@ -139,6 +145,11 @@ func WithPriorityAndFairness(
}
}, execute)
if !served {
if isMutatingRequest {
epmetrics.DroppedRequests.WithLabelValues(epmetrics.MutatingKind).Inc()
} else {
epmetrics.DroppedRequests.WithLabelValues(epmetrics.ReadOnlyKind).Inc()
}
epmetrics.RecordRequestTermination(r, requestInfo, epmetrics.APIServerComponent, http.StatusTooManyRequests)
tooManyRequests(r, w)
}

350
staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness_test.go Normal file
View File

@ -0,0 +1,350 @@
/*
Copyright 2016 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package filters
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
fctypesv1a1 "k8s.io/api/flowcontrol/v1alpha1"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apiserver/pkg/apis/flowcontrol/bootstrap"
"k8s.io/apiserver/pkg/authentication/user"
apifilters "k8s.io/apiserver/pkg/endpoints/filters"
epmetrics "k8s.io/apiserver/pkg/endpoints/metrics"
apirequest "k8s.io/apiserver/pkg/endpoints/request"
"k8s.io/apiserver/pkg/server/mux"
utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
fq "k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing"
fcmetrics "k8s.io/apiserver/pkg/util/flowcontrol/metrics"
"k8s.io/component-base/metrics/legacyregistry"
)
type mockDecision int
const (
decisionNoQueuingExecute mockDecision = iota
decisionQueuingExecute
decisionCancelWait
decisionReject
decisionSkipFilter
)
type fakeApfFilter struct {
mockDecision mockDecision
postEnqueue func()
postDequeue func()
}
func (t fakeApfFilter) MaintainObservations(stopCh <-chan struct{}) {
}
func (t fakeApfFilter) Handle(ctx context.Context,
requestDigest utilflowcontrol.RequestDigest,
noteFn func(fs *fctypesv1a1.FlowSchema, pl *fctypesv1a1.PriorityLevelConfiguration),
queueNoteFn fq.QueueNoteFn,
execFn func(),
) {
if t.mockDecision == decisionSkipFilter {
panic("Handle should not be invoked")
}
noteFn(bootstrap.SuggestedFlowSchemaGlobalDefault, bootstrap.SuggestedPriorityLevelConfigurationGlobalDefault)
switch t.mockDecision {
case decisionNoQueuingExecute:
execFn()
case decisionQueuingExecute:
queueNoteFn(true)
t.postEnqueue()
queueNoteFn(false)
t.postDequeue()
execFn()
case decisionCancelWait:
queueNoteFn(true)
t.postEnqueue()
queueNoteFn(false)
t.postDequeue()
case decisionReject:
return
}
}
func (t fakeApfFilter) Run(stopCh <-chan struct{}) error {
return nil
}
func (t fakeApfFilter) Install(c *mux.PathRecorderMux) {
}
func newApfServerWithSingleRequest(decision mockDecision, t *testing.T) *httptest.Server {
onExecuteFunc := func() {
if decision == decisionCancelWait {
t.Errorf("execute should not be invoked")
}
// atomicReadOnlyExecuting can be either 0 or 1 as we test one request at a time.
if decision != decisionSkipFilter && atomicReadOnlyExecuting != 1 {
t.Errorf("Wanted %d requests executing, got %d", 1, atomicReadOnlyExecuting)
}
}
postExecuteFunc := func() {}
// atomicReadOnlyWaiting can be either 0 or 1 as we test one request at a time.
postEnqueueFunc := func() {
if atomicReadOnlyWaiting != 1 {
t.Errorf("Wanted %d requests in queue, got %d", 1, atomicReadOnlyWaiting)
}
}
postDequeueFunc := func() {
if atomicReadOnlyWaiting != 0 {
t.Errorf("Wanted %d requests in queue, got %d", 0, atomicReadOnlyWaiting)
}
}
return newApfServerWithHooks(decision, onExecuteFunc, postExecuteFunc, postEnqueueFunc, postDequeueFunc, t)
}
func newApfServerWithHooks(decision mockDecision, onExecute, postExecute, postEnqueue, postDequeue func(), t *testing.T) *httptest.Server {
requestInfoFactory := &apirequest.RequestInfoFactory{APIPrefixes: sets.NewString("apis", "api"), GrouplessAPIPrefixes: sets.NewString("api")}
longRunningRequestCheck := BasicLongRunningRequestCheck(sets.NewString("watch"), sets.NewString("proxy"))
apfHandler := WithPriorityAndFairness(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
onExecute()
}), longRunningRequestCheck, fakeApfFilter{
mockDecision: decision,
postEnqueue: postEnqueue,
postDequeue: postDequeue,
})
handler := apifilters.WithRequestInfo(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
r = r.WithContext(apirequest.WithUser(r.Context(), &user.DefaultInfo{
Groups: []string{user.AllUnauthenticated},
}))
apfHandler.ServeHTTP(w, r)
postExecute()
if atomicReadOnlyExecuting != 0 {
t.Errorf("Wanted %d requests executing, got %d", 0, atomicReadOnlyExecuting)
}
}), requestInfoFactory)
apfServer := httptest.NewServer(handler)
return apfServer
}
func TestApfSkipLongRunningRequest(t *testing.T) {
epmetrics.Register()
server := newApfServerWithSingleRequest(decisionSkipFilter, t)
defer server.Close()
// send a watch request to test skipping long running request
if err := expectHTTPGet(fmt.Sprintf("%s/api/v1/namespaces?watch=true", server.URL), http.StatusOK); err != nil {
// request should not be rejected
t.Error(err)
}
}
func TestApfRejectRequest(t *testing.T) {
epmetrics.Register()
server := newApfServerWithSingleRequest(decisionReject, t)
defer server.Close()
if err := expectHTTPGet(fmt.Sprintf("%s/api/v1/namespaces/default", server.URL), http.StatusTooManyRequests); err != nil {
t.Error(err)
}
checkForExpectedMetricsWithRetry(t, []string{
"apiserver_request_terminations_total",
"apiserver_dropped_requests_total",
})
}
func TestApfExemptRequest(t *testing.T) {
epmetrics.Register()
fcmetrics.Register()
// Wait for at least one sampling window to pass since creation of metrics.ReadWriteConcurrencyObserverPairGenerator,
// so that an observation will cause some data to go into the Prometheus metrics.
time.Sleep(time.Millisecond * 50)
server := newApfServerWithSingleRequest(decisionNoQueuingExecute, t)
defer server.Close()
if err := expectHTTPGet(fmt.Sprintf("%s/api/v1/namespaces/default", server.URL), http.StatusOK); err != nil {
t.Error(err)
}
checkForExpectedMetricsWithRetry(t, []string{
"apiserver_current_inflight_requests",
"apiserver_flowcontrol_read_vs_write_request_count_watermarks",
"apiserver_flowcontrol_read_vs_write_request_count_samples",
})
}
func TestApfExecuteRequest(t *testing.T) {
epmetrics.Register()
fcmetrics.Register()
// Wait for at least one sampling window to pass since creation of metrics.ReadWriteConcurrencyObserverPairGenerator,
// so that an observation will cause some data to go into the Prometheus metrics.
time.Sleep(time.Millisecond * 50)
server := newApfServerWithSingleRequest(decisionQueuingExecute, t)
defer server.Close()
if err := expectHTTPGet(fmt.Sprintf("%s/api/v1/namespaces/default", server.URL), http.StatusOK); err != nil {
t.Error(err)
}
checkForExpectedMetricsWithRetry(t, []string{
"apiserver_current_inflight_requests",
"apiserver_current_inqueue_requests",
"apiserver_flowcontrol_read_vs_write_request_count_watermarks",
"apiserver_flowcontrol_read_vs_write_request_count_samples",
})
}
func TestApfExecuteMultipleRequests(t *testing.T) {
epmetrics.Register()
fcmetrics.Register()
// Wait for at least one sampling window to pass since creation of metrics.ReadWriteConcurrencyObserverPairGenerator,
// so that an observation will cause some data to go into the Prometheus metrics.
time.Sleep(time.Millisecond * 50)
concurrentRequests := 5
var preStartExecute, postStartExecute, preEnqueue, postEnqueue, preDequeue, postDequeue, finishExecute sync.WaitGroup
for _, wg := range []*sync.WaitGroup{&preStartExecute, &postStartExecute, &preEnqueue, &postEnqueue, &preDequeue, &postDequeue, &finishExecute} {
wg.Add(concurrentRequests)
}
onExecuteFunc := func() {
preStartExecute.Done()
preStartExecute.Wait()
if int(atomicReadOnlyExecuting) != concurrentRequests {
t.Errorf("Wanted %d requests executing, got %d", concurrentRequests, atomicReadOnlyExecuting)
}
postStartExecute.Done()
postStartExecute.Wait()
}
postEnqueueFunc := func() {
preEnqueue.Done()
preEnqueue.Wait()
if int(atomicReadOnlyWaiting) != concurrentRequests {
t.Errorf("Wanted %d requests in queue, got %d", 1, atomicReadOnlyWaiting)
}
postEnqueue.Done()
postEnqueue.Wait()
}
postDequeueFunc := func() {
preDequeue.Done()
preDequeue.Wait()
if atomicReadOnlyWaiting != 0 {
t.Errorf("Wanted %d requests in queue, got %d", 0, atomicReadOnlyWaiting)
}
postDequeue.Done()
postDequeue.Wait()
}
postExecuteFunc := func() {
finishExecute.Done()
finishExecute.Wait()
}
server := newApfServerWithHooks(decisionQueuingExecute, onExecuteFunc, postExecuteFunc, postEnqueueFunc, postDequeueFunc, t)
defer server.Close()
for i := 0; i < concurrentRequests; i++ {
var err error
go func() {
err = expectHTTPGet(fmt.Sprintf("%s/api/v1/namespaces/default", server.URL), http.StatusOK)
}()
if err != nil {
t.Error(err)
}
}
checkForExpectedMetricsWithRetry(t, []string{
"apiserver_current_inflight_requests",
"apiserver_current_inqueue_requests",
"apiserver_flowcontrol_read_vs_write_request_count_watermarks",
"apiserver_flowcontrol_read_vs_write_request_count_samples",
})
}
func TestApfCancelWaitRequest(t *testing.T) {
epmetrics.Register()
server := newApfServerWithSingleRequest(decisionCancelWait, t)
defer server.Close()
if err := expectHTTPGet(fmt.Sprintf("%s/api/v1/namespaces/default", server.URL), http.StatusTooManyRequests); err != nil {
t.Error(err)
}
checkForExpectedMetricsWithRetry(t, []string{
"apiserver_current_inflight_requests",
"apiserver_request_terminations_total",
"apiserver_dropped_requests_total",
})
}
// wait async metrics to be collected
func checkForExpectedMetricsWithRetry(t *testing.T, expectedMetrics []string) {
maxRetries := 5
var checkErrors []error
for i := 0; i < maxRetries; i++ {
t.Logf("Check for expected metrics with retry %d", i)
metricsFamily, err := legacyregistry.DefaultGatherer.Gather()
if err != nil {
t.Fatalf("Failed to gather metrics %v", err)
}
metrics := map[string]interface{}{}
for _, mf := range metricsFamily {
metrics[*mf.Name] = mf
}
checkErrors = checkForExpectedMetrics(expectedMetrics, metrics)
if checkErrors == nil {
return
}
time.Sleep(1 * time.Second)
}
for _, checkError := range checkErrors {
t.Error(checkError)
}
}
func checkForExpectedMetrics(expectedMetrics []string, metrics map[string]interface{}) []error {
var errs []error
for _, metricName := range expectedMetrics {
if _, ok := metrics[metricName]; !ok {
if !ok {
errs = append(errs, errors.New("Scraped metrics did not include expected metric "+metricName))
}
}
}
return errs
}