Skip to content

Commit 9e59765

Browse files
k8s-ci-robotk8s-ci-robot
authored
Merge pull request kubernetes#126999 from pohly/log-client-go-rest-body
client-go/rest: contextual logging of request/response
2 parents d589a3f + 57f9b7c commit 9e59765

File tree

2 files changed

+129
-22
lines changed

2 files changed

+129
-22
lines changed

‎staging/src/k8s.io/client-go/rest/request.go‎

Lines changed: 40 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -450,11 +450,9 @@ func (r *Request) Body(obj interface{}) *Request {
450450
r.err = err
451451
return r
452452
}
453-
glogBody("Request Body", data)
454453
r.body = nil
455454
r.bodyBytes = data
456455
case []byte:
457-
glogBody("Request Body", t)
458456
r.body = nil
459457
r.bodyBytes = t
460458
case io.Reader:
@@ -475,7 +473,6 @@ func (r *Request) Body(obj interface{}) *Request {
475473
r.err = err
476474
return r
477475
}
478-
glogBody("Request Body", data)
479476
r.body = nil
480477
r.bodyBytes = data
481478
r.SetHeader("Content-Type", r.c.content.ContentType)
@@ -704,6 +701,10 @@ func (b *throttledLogger) Infof(message string, args ...interface{}) {
704701
// Watch attempts to begin watching the requested location.
705702
// Returns a watch.Interface, or an error.
706703
func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
704+
if r.body == nil {
705+
logBody(ctx, 2, "Request Body", r.bodyBytes)
706+
}
707+
707708
// We specifically don't want to rate limit watches, so we
708709
// don't use r.rateLimiter here.
709710
if r.err != nil {
@@ -752,7 +753,7 @@ func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
752753
// the server must have sent us an error in 'err'
753754
return true, nil
754755
}
755-
result := r.transformResponse(resp, req)
756+
result := r.transformResponse(ctx, resp, req)
756757
if err := result.Error(); err != nil {
757758
return true, err
758759
}
@@ -845,6 +846,10 @@ func (r WatchListResult) Into(obj runtime.Object) error {
845846
// Check the documentation https://kubernetes.io/docs/reference/using-api/api-concepts/#streaming-lists
846847
// to see what parameters are currently required.
847848
func (r *Request) WatchList(ctx context.Context) WatchListResult {
849+
if r.body == nil {
850+
logBody(ctx, 2, "Request Body", r.bodyBytes)
851+
}
852+
848853
if !clientfeatures.FeatureGates().Enabled(clientfeatures.WatchListClient) {
849854
return WatchListResult{err: fmt.Errorf("%q feature gate is not enabled", clientfeatures.WatchListClient)}
850855
}
@@ -969,6 +974,10 @@ func sanitize(req *Request, resp *http.Response, err error) (string, string) {
969974
// Any non-2xx http status code causes an error. If we get a non-2xx code, we try to convert the body into an APIStatus object.
970975
// If we can, we return that as an error. Otherwise, we create an error that lists the http status and the content of the response.
971976
func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
977+
if r.body == nil {
978+
logBody(ctx, 2, "Request Body", r.bodyBytes)
979+
}
980+
972981
if r.err != nil {
973982
return nil, r.err
974983
}
@@ -1012,7 +1021,7 @@ func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
10121021
if retry.IsNextRetry(ctx, r, req, resp, err, neverRetryError) {
10131022
return false, nil
10141023
}
1015-
result := r.transformResponse(resp, req)
1024+
result := r.transformResponse(ctx, resp, req)
10161025
if err := result.Error(); err != nil {
10171026
return true, err
10181027
}
@@ -1199,9 +1208,13 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
11991208
// - If the server responds with a status: *errors.StatusError or *errors.UnexpectedObjectError
12001209
// - http.Client.Do errors are returned directly.
12011210
func (r *Request) Do(ctx context.Context) Result {
1211+
if r.body == nil {
1212+
logBody(ctx, 2, "Request Body", r.bodyBytes)
1213+
}
1214+
12021215
var result Result
12031216
err := r.request(ctx, func(req *http.Request, resp *http.Response) {
1204-
result = r.transformResponse(resp, req)
1217+
result = r.transformResponse(ctx, resp, req)
12051218
})
12061219
if err != nil {
12071220
return Result{err: err}
@@ -1214,10 +1227,14 @@ func (r *Request) Do(ctx context.Context) Result {
12141227

12151228
// DoRaw executes the request but does not process the response body.
12161229
func (r *Request) DoRaw(ctx context.Context) ([]byte, error) {
1230+
if r.body == nil {
1231+
logBody(ctx, 2, "Request Body", r.bodyBytes)
1232+
}
1233+
12171234
var result Result
12181235
err := r.request(ctx, func(req *http.Request, resp *http.Response) {
12191236
result.body, result.err = io.ReadAll(resp.Body)
1220-
glogBody("Response Body", result.body)
1237+
logBody(ctx, 2, "Response Body", result.body)
12211238
if resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent {
12221239
result.err = r.transformUnstructuredResponseError(resp, req, result.body)
12231240
}
@@ -1232,7 +1249,7 @@ func (r *Request) DoRaw(ctx context.Context) ([]byte, error) {
12321249
}
12331250

12341251
// transformResponse converts an API response into a structured API object
1235-
func (r *Request) transformResponse(resp *http.Response, req *http.Request) Result {
1252+
func (r *Request) transformResponse(ctx context.Context, resp *http.Response, req *http.Request) Result {
12361253
var body []byte
12371254
if resp.Body != nil {
12381255
data, err := io.ReadAll(resp.Body)
@@ -1261,7 +1278,8 @@ func (r *Request) transformResponse(resp *http.Response, req *http.Request) Resu
12611278
}
12621279
}
12631280

1264-
glogBody("Response Body", body)
1281+
// Call depth is tricky. This one is okay for Do and DoRaw.
1282+
logBody(ctx, 7, "Response Body", body)
12651283

12661284
// verify the content type is accurate
12671285
var decoder runtime.Decoder
@@ -1321,14 +1339,14 @@ func (r *Request) transformResponse(resp *http.Response, req *http.Request) Resu
13211339
}
13221340

13231341
// truncateBody decides if the body should be truncated, based on the glog Verbosity.
1324-
func truncateBody(body string) string {
1342+
func truncateBody(logger klog.Logger, body string) string {
13251343
max := 0
13261344
switch {
1327-
case bool(klog.V(10).Enabled()):
1345+
case bool(logger.V(10).Enabled()):
13281346
return body
1329-
case bool(klog.V(9).Enabled()):
1347+
case bool(logger.V(9).Enabled()):
13301348
max = 10240
1331-
case bool(klog.V(8).Enabled()):
1349+
case bool(logger.V(8).Enabled()):
13321350
max = 1024
13331351
}
13341352

@@ -1339,17 +1357,21 @@ func truncateBody(body string) string {
13391357
return body[:max] + fmt.Sprintf(" [truncated %d chars]", len(body)-max)
13401358
}
13411359

1342-
// glogBody logs a body output that could be either JSON or protobuf. It explicitly guards against
1360+
// logBody logs a body output that could be either JSON or protobuf. It explicitly guards against
13431361
// allocating a new string for the body output unless necessary. Uses a simple heuristic to determine
13441362
// whether the body is printable.
1345-
func glogBody(prefix string, body []byte) {
1346-
if klogV := klog.V(8); klogV.Enabled() {
1363+
//
1364+
// It needs to be called by all functions which send or receive the data.
1365+
func logBody(ctx context.Context, callDepth int, prefix string, body []byte) {
1366+
logger := klog.FromContext(ctx)
1367+
if loggerV := logger.V(8); loggerV.Enabled() {
1368+
loggerV := loggerV.WithCallDepth(callDepth)
13471369
if bytes.IndexFunc(body, func(r rune) bool {
13481370
return r < 0x0a
13491371
}) != -1 {
1350-
klogV.Infof("%s:\n%s", prefix, truncateBody(hex.Dump(body)))
1372+
loggerV.Info(prefix, "body", truncateBody(logger, hex.Dump(body)))
13511373
} else {
1352-
klogV.Infof("%s: %s", prefix, truncateBody(string(body)))
1374+
loggerV.Info(prefix, "body", truncateBody(logger, string(body)))
13531375
}
13541376
}
13551377
}

‎staging/src/k8s.io/client-go/rest/request_test.go‎

Lines changed: 89 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -28,14 +28,20 @@ import (
2828
"net/http/httptest"
2929
"net/url"
3030
"os"
31+
"path"
3132
"reflect"
33+
"regexp"
34+
goruntime "runtime"
3235
"strings"
3336
"sync"
3437
"sync/atomic"
3538
"syscall"
3639
"testing"
3740
"time"
3841

42+
"github.com/stretchr/testify/assert"
43+
"github.com/stretchr/testify/require"
44+
3945
"github.com/google/go-cmp/cmp"
4046
v1 "k8s.io/api/core/v1"
4147
apiequality "k8s.io/apimachinery/pkg/api/equality"
@@ -54,6 +60,7 @@ import (
5460
"k8s.io/client-go/util/flowcontrol"
5561
utiltesting "k8s.io/client-go/util/testing"
5662
"k8s.io/klog/v2"
63+
"k8s.io/klog/v2/ktesting"
5764
testingclock "k8s.io/utils/clock/testing"
5865
)
5966

@@ -553,6 +560,7 @@ func TestURLTemplate(t *testing.T) {
553560
}
554561

555562
func TestTransformResponse(t *testing.T) {
563+
_, ctx := ktesting.NewTestContext(t)
556564
invalid := []byte("aaaaa")
557565
uri, _ := url.Parse("http://localhost")
558566
testCases := []struct {
@@ -601,7 +609,7 @@ func TestTransformResponse(t *testing.T) {
601609
if test.Response.Body == nil {
602610
test.Response.Body = io.NopCloser(bytes.NewReader([]byte{}))
603611
}
604-
result := r.transformResponse(test.Response, &http.Request{})
612+
result := r.transformResponse(ctx, test.Response, &http.Request{})
605613
response, created, err := result.body, result.statusCode == http.StatusCreated, result.err
606614
hasErr := err != nil
607615
if hasErr != test.Error {
@@ -652,6 +660,7 @@ func (r *renegotiator) StreamDecoder(contentType string, params map[string]strin
652660
}
653661

654662
func TestTransformResponseNegotiate(t *testing.T) {
663+
_, ctx := ktesting.NewTestContext(t)
655664
invalid := []byte("aaaaa")
656665
uri, _ := url.Parse("http://localhost")
657666
testCases := []struct {
@@ -765,7 +774,7 @@ func TestTransformResponseNegotiate(t *testing.T) {
765774
if test.Response.Body == nil {
766775
test.Response.Body = io.NopCloser(bytes.NewReader([]byte{}))
767776
}
768-
result := r.transformResponse(test.Response, &http.Request{})
777+
result := r.transformResponse(ctx, test.Response, &http.Request{})
769778
_, err := result.body, result.err
770779
hasErr := err != nil
771780
if hasErr != test.Error {
@@ -890,14 +899,15 @@ func TestTransformUnstructuredError(t *testing.T) {
890899

891900
for _, testCase := range testCases {
892901
t.Run("", func(t *testing.T) {
902+
_, ctx := ktesting.NewTestContext(t)
893903
r := &Request{
894904
c: &RESTClient{
895905
content: defaultContentConfig(),
896906
},
897907
resourceName: testCase.Name,
898908
resource: testCase.Resource,
899909
}
900-
result := r.transformResponse(testCase.Res, testCase.Req)
910+
result := r.transformResponse(ctx, testCase.Res, testCase.Req)
901911
err := result.err
902912
if !testCase.ErrFn(err) {
903913
t.Fatalf("unexpected error: %v", err)
@@ -2331,7 +2341,7 @@ func TestTruncateBody(t *testing.T) {
23312341
l := flag.Lookup("v").Value.(flag.Getter).Get().(klog.Level)
23322342
for _, test := range tests {
23332343
flag.Set("v", test.level)
2334-
got := truncateBody(test.body)
2344+
got := truncateBody(klog.Background(), test.body)
23352345
if got != test.want {
23362346
t.Errorf("truncateBody(%v) = %v, want %v", test.body, got, test.want)
23372347
}
@@ -4051,3 +4061,78 @@ func TestRequestConcurrencyWithRetry(t *testing.T) {
40514061
t.Errorf("Expected attempts: %d, but got: %d", expected, attempts)
40524062
}
40534063
}
4064+
4065+
func TestRequestLogging(t *testing.T) {
4066+
testcases := map[string]struct {
4067+
v int
4068+
body any
4069+
expectedOutput string
4070+
}{
4071+
"no-output": {
4072+
v: 7,
4073+
body: []byte("ping"),
4074+
},
4075+
"output": {
4076+
v: 8,
4077+
body: []byte("ping"),
4078+
expectedOutput: `<location>] "Request Body" logger="TestLogger" body="ping"
4079+
<location>] "Response Body" logger="TestLogger" body="pong"
4080+
`,
4081+
},
4082+
"io-reader": {
4083+
v: 8,
4084+
body: strings.NewReader("ping"),
4085+
// Cannot log the request body!
4086+
expectedOutput: `<location>] "Response Body" logger="TestLogger" body="pong"
4087+
`,
4088+
},
4089+
"truncate": {
4090+
v: 8,
4091+
body: []byte(strings.Repeat("a", 2000)),
4092+
expectedOutput: fmt.Sprintf(`<location>] "Request Body" logger="TestLogger" body="%s [truncated 976 chars]"
4093+
<location>] "Response Body" logger="TestLogger" body="pong"
4094+
`, strings.Repeat("a", 1024)),
4095+
},
4096+
}
4097+
4098+
for name, tc := range testcases {
4099+
t.Run(name, func(t *testing.T) {
4100+
state := klog.CaptureState()
4101+
defer state.Restore()
4102+
4103+
var buffer bytes.Buffer
4104+
klog.SetOutput(&buffer)
4105+
klog.LogToStderr(false)
4106+
var fs flag.FlagSet
4107+
klog.InitFlags(&fs)
4108+
require.NoError(t, fs.Set("v", fmt.Sprintf("%d", tc.v)), "set verbosity")
4109+
4110+
client := clientForFunc(func(req *http.Request) (*http.Response, error) {
4111+
return &http.Response{
4112+
StatusCode: http.StatusOK,
4113+
Body: io.NopCloser(strings.NewReader("pong")),
4114+
}, nil
4115+
})
4116+
4117+
req := NewRequestWithClient(nil, "", defaultContentConfig(), client).
4118+
Body(tc.body)
4119+
4120+
logger := klog.Background()
4121+
logger = klog.LoggerWithName(logger, "TestLogger")
4122+
ctx := klog.NewContext(context.Background(), logger)
4123+
4124+
_, file, line, _ := goruntime.Caller(0)
4125+
result := req.Do(ctx)
4126+
require.NoError(t, result.Error(), "request.Do")
4127+
4128+
// Compare log output:
4129+
// - strip date/time/pid from each line (fixed length header)
4130+
// - replace <location> with the actual call location
4131+
state.Restore()
4132+
expectedOutput := strings.ReplaceAll(tc.expectedOutput, "<location>", fmt.Sprintf("%s:%d", path.Base(file), line+1))
4133+
actualOutput := buffer.String()
4134+
actualOutput = regexp.MustCompile(`(?m)^.{30}`).ReplaceAllString(actualOutput, "")
4135+
assert.Equal(t, expectedOutput, actualOutput)
4136+
})
4137+
}
4138+
}

0 commit comments

Comments
 (0)