cubefs

1
// Copyright 2022 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package auditlog
16

17
import (
18
	"bytes"
19
	"encoding/json"
20
	"encoding/xml"
21
	"fmt"
22
	"io"
23
	"io/ioutil"
24
	"math/rand"
25
	"net/http"
26
	"net/http/httptest"
27
	"os"
28
	"strconv"
29
	"strings"
30
	"testing"
31
	"time"
32

33
	"github.com/stretchr/testify/require"
34

35
	"github.com/cubefs/cubefs/blobstore/common/rpc"
36
	"github.com/cubefs/cubefs/blobstore/common/trace"
37
)
38

39
type testRespData struct {
40
	Result string `json:"result"`
41
}
42

43
type testErrorRespData struct {
44
	XMLName xml.Name `xml:"Error"`
45
	Code    string   `xml:"Code"`
46
	Message string   `xml:"Message"`
47
}
48

49
func initServer(t *testing.T, name string, cfg Config) (server *httptest.Server, tmpDir string, lc LogCloser) {
50
	tracer := trace.NewTracer(name)
51
	trace.SetGlobalTracer(tracer)
52

53
	tmpDir = fmt.Sprintf("%s/%s-auditlog-%s%s", os.TempDir(), name, strconv.FormatInt(time.Now().Unix(), 10), strconv.Itoa(rand.Intn(100000)))
54
	err := os.Mkdir(tmpDir, 0o755)
55
	require.NoError(t, err)
56

57
	cfg.LogDir = tmpDir
58
	cfg.MetricConfig = PrometheusConfig{Idc: name}
59
	cfg.Filters = []FilterConfig{{Must: Conditions{"term": {"method": "GET"}}}}
60

61
	var ah rpc.ProgressHandler
62
	ah, lc, err = Open(name, &cfg)
63
	require.NoError(t, err)
64
	require.NotNil(t, ah)
65
	require.NotNil(t, lc)
66

67
	bussinessHandler := func(w http.ResponseWriter, req *http.Request) {
68
		_, err := ioutil.ReadAll(req.Body)
69
		require.NoError(t, err)
70
		w.Header().Set("testh1", "testh1value")
71
		w.Header().Add("testh1", "testh1value2")
72
		w.Header().Set("Content-Type", rpc.MIMEJSON)
73
		w.Header()["ETag"] = []string{"etag value"}
74

75
		extraHeader := ExtraHeader(w)
76
		extraHeader.Set("extra-header1", "header1 value")
77
		extraHeader.Set("Extra-header2", "header2 value")
78
		extraHeader.Add("Extra-header2", "header2 value2")
79

80
		data, err := json.Marshal(testRespData{Result: "success"})
81
		require.NoError(t, err)
82
		w.Write(data)
83

84
		rw := w.(*responseWriter)
85
		if !rw.no2xxBody && rw.bodyLimit > 0 {
86
			require.Equal(t, len(data), rw.n)
87
			require.Equal(t, data, rw.body[:rw.n])
88
		} else {
89
			require.Equal(t, 0, rw.n)
90
		}
91
	}
92

93
	streamHandler := func(w http.ResponseWriter, req *http.Request) {
94
		size := int64(64 * 1024)
95
		buffer := make([]byte, size)
96
		for range [1024]struct{}{} {
97
			_, err := io.CopyN(w, bytes.NewReader(buffer), size)
98
			require.NoError(t, err)
99
		}
100

101
		span := trace.SpanFromContextSafe(req.Context())
102
		span.SetTag("response", "readfrom")
103
		span.AppendRPCTrackLog([]string{"stream"})
104

105
		rw := w.(*responseWriter)
106
		require.Equal(t, 0, rw.n)
107
		require.Equal(t, 1024*size, rw.bodyWritten)
108
	}
109

110
	errorResponseHandler := func(w http.ResponseWriter, req *http.Request) {
111
		errResp := testErrorRespData{
112
			Code:    "ErrorTestResponseCode",
113
			Message: "error test response message",
114
		}
115
		data, err := xml.Marshal(errResp)
116
		require.NoError(t, err)
117
		w.Header().Set("Content-Type", rpc.MIMEXML)
118
		w.Header().Set("Content-Length", strconv.Itoa(len(data)))
119
		w.WriteHeader(http.StatusBadRequest)
120
		w.Write(data)
121
	}
122

123
	entryHandler := func(w http.ResponseWriter, req *http.Request) {
124
		switch req.URL.Path {
125
		case "/":
126
			ah.Handler(w, req, bussinessHandler)
127
		case "/stream":
128
			ah.Handler(w, req, streamHandler)
129
		case "/error-response":
130
			ah.Handler(w, req, errorResponseHandler)
131
		default:
132
			w.WriteHeader(http.StatusNotImplemented)
133
		}
134
	}
135

136
	server = httptest.NewServer(http.HandlerFunc(entryHandler))
137
	return server, tmpDir, lc
138
}
139

140
func initNoContentLengthServer(t *testing.T) (server *httptest.Server, tmpDir string, lc LogCloser) {
141
	moduleName := "TESTNOCONTENTLENGTHMOULE"
142
	tracer := trace.NewTracer(moduleName)
143
	trace.SetGlobalTracer(tracer)
144

145
	tmpDir = os.TempDir() + "/test-NoContentLength-log" + strconv.FormatInt(time.Now().Unix(), 10) + strconv.Itoa(rand.Intn(100000))
146
	err := os.Mkdir(tmpDir, 0o755)
147
	require.NoError(t, err)
148

149
	var ah rpc.ProgressHandler
150
	ah, lc, err = Open(moduleName, &Config{
151
		LogDir: tmpDir, ChunkBits: 29,
152
		MetricConfig: PrometheusConfig{
153
			Idc: moduleName,
154
		},
155
	})
156
	require.NoError(t, err)
157
	require.NotNil(t, ah)
158
	require.NotNil(t, lc)
159

160
	noContentLengthHandler := func(w http.ResponseWriter, req *http.Request) {
161
		buffered, err := ioutil.ReadAll(req.Body)
162
		require.NoError(t, err)
163
		bodySize := req.Body.(*reqBodyReadCloser).bodyRead
164
		readSting := string(buffered[:bodySize])
165
		w.Header().Set("Content-Type", rpc.MIMEJSON)
166
		w.WriteHeader(http.StatusOK)
167
		data, err := json.Marshal(testRespData{Result: readSting})
168
		require.NoError(t, err)
169
		w.Write(data)
170
	}
171
	entryHandler := func(w http.ResponseWriter, req *http.Request) {
172
		ah.Handler(w, req, noContentLengthHandler)
173
	}
174

175
	server = httptest.NewServer(http.HandlerFunc(entryHandler))
176
	return server, tmpDir, lc
177
}
178

179
func TestOpen(t *testing.T) {
180
	cfg := Config{
181
		Filters: []FilterConfig{{Must: Conditions{"term": {"method": "GET"}}}},
182
	}
183
	server, tmpDir, lc := initServer(t, "testOpen", cfg)
184
	defer func() {
185
		server.Close()
186
		os.RemoveAll(tmpDir)
187
		lc.Close()
188
	}()
189

190
	url := server.URL
191
	client := http.DefaultClient
192

193
	req, err := http.NewRequest(http.MethodGet, url, nil)
194
	require.NoError(t, err)
195
	resp, err := client.Do(req)
196
	require.NoError(t, err)
197
	require.Equal(t, http.StatusOK, resp.StatusCode)
198
	b, err := ioutil.ReadAll(resp.Body)
199
	require.NoError(t, err)
200
	respData := &testRespData{}
201
	err = json.Unmarshal(b, respData)
202
	require.NoError(t, err)
203
	require.Equal(t, "success", respData.Result)
204
	resp.Body.Close()
205

206
	open, err := os.Open(tmpDir)
207
	require.NoError(t, err)
208
	dirEntries, err := open.ReadDir(-1)
209
	require.NoError(t, err)
210
	require.Equal(t, 0, len(dirEntries))
211
	require.NoError(t, open.Close())
212

213
	for _, method := range []string{http.MethodPost, http.MethodDelete, http.MethodPut} {
214
		req, err := http.NewRequest(method, url, nil)
215
		require.NoError(t, err)
216
		resp, err := client.Do(req)
217
		require.NoError(t, err)
218
		require.Equal(t, http.StatusOK, resp.StatusCode)
219
		b, err := ioutil.ReadAll(resp.Body)
220
		require.NoError(t, err)
221
		respData := &testRespData{}
222
		err = json.Unmarshal(b, respData)
223
		require.NoError(t, err)
224
		require.Equal(t, "success", respData.Result)
225
		resp.Body.Close()
226
	}
227

228
	open, err = os.Open(tmpDir)
229
	require.NoError(t, err)
230
	dirEntries, err = open.ReadDir(-1)
231
	require.NoError(t, err)
232
	require.Greater(t, len(dirEntries), 0)
233
}
234

235
func TestNoContentLength(t *testing.T) {
236
	server, tmpDir, lc := initNoContentLengthServer(t)
237
	defer func() {
238
		server.Close()
239
		os.RemoveAll(tmpDir)
240
		lc.Close()
241
	}()
242

243
	url := server.URL
244
	client := http.DefaultClient
245

246
	body := strings.NewReader("{\"test1\":\"test1value\"}")
247
	req, err := http.NewRequest(http.MethodPost, url, body)
248
	require.NoError(t, err)
249
	resp, err := client.Do(req)
250
	require.NoError(t, err)
251
	require.Equal(t, http.StatusOK, resp.StatusCode)
252
	b, err := ioutil.ReadAll(resp.Body)
253
	require.NoError(t, err)
254
	respData := &testRespData{}
255
	err = json.Unmarshal(b, respData)
256
	require.NoError(t, err)
257
	require.Equal(t, "{\"test1\":\"test1value\"}", respData.Result)
258
	resp.Body.Close()
259
}
260

261
func TestNoLogBody(t *testing.T) {
262
	cfg := Config{
263
		No2xxBody: true,
264
	}
265
	server, tmpDir, lc := initServer(t, "testNoLogBody", cfg)
266
	defer func() {
267
		server.Close()
268
		os.RemoveAll(tmpDir)
269
		lc.Close()
270
	}()
271

272
	url := server.URL
273
	client := http.DefaultClient
274

275
	req, err := http.NewRequest(http.MethodGet, url, nil)
276
	require.NoError(t, err)
277
	resp, err := client.Do(req)
278
	require.NoError(t, err)
279
	require.Equal(t, http.StatusOK, resp.StatusCode)
280
	b, err := ioutil.ReadAll(resp.Body)
281
	require.NoError(t, err)
282
	respData := &testRespData{}
283
	err = json.Unmarshal(b, respData)
284
	require.NoError(t, err)
285
	require.Equal(t, "success", respData.Result)
286
	resp.Body.Close()
287

288
	req, err = http.NewRequest(http.MethodPut, url+"/error-response", nil)
289
	require.NoError(t, err)
290
	resp, err = client.Do(req)
291
	require.NoError(t, err)
292
	require.Equal(t, http.StatusBadRequest, resp.StatusCode)
293
	b, err = ioutil.ReadAll(resp.Body)
294
	require.NoError(t, err)
295
	respData2 := &testErrorRespData{}
296
	err = xml.Unmarshal(b, respData2)
297
	require.NoError(t, err)
298
	require.Equal(t, "ErrorTestResponseCode", respData2.Code)
299
	resp.Body.Close()
300

301
	open, err := os.Open(tmpDir)
302
	require.NoError(t, err)
303
	dirEntries, err := open.ReadDir(-1)
304
	require.NoError(t, err)
305
	require.Greater(t, len(dirEntries), 0)
306
	require.NoError(t, open.Close())
307
}
308

309
func TestResponseReadFrom(t *testing.T) {
310
	server, tmpDir, lc := initServer(t, "testResponseReadFrom", Config{LogFormat: LogFormatJSON})
311
	defer func() {
312
		server.Close()
313
		os.RemoveAll(tmpDir)
314
		lc.Close()
315
	}()
316

317
	req, err := http.NewRequest(http.MethodGet, server.URL+"/stream", nil)
318
	require.NoError(t, err)
319
	resp, err := http.DefaultClient.Do(req)
320
	require.NoError(t, err)
321
	defer resp.Body.Close()
322
	require.Equal(t, http.StatusOK, resp.StatusCode)
323
	_, err = io.Copy(io.Discard, resp.Body)
324
	require.NoError(t, err)
325
}
326

327
func TestBodylimit(t *testing.T) {
328
	for _, limit := range []struct {
329
		actual, expected int
330
	}{
331
		{-100, 0},
332
		{-1, 0},
333
		{0, defaultReadBodyBuffLength},
334
		{10, 10},
335
		{defaultReadBodyBuffLength, defaultReadBodyBuffLength},
336
		{maxReadBodyBuffLength, maxReadBodyBuffLength},
337
		{maxReadBodyBuffLength + 1, maxReadBodyBuffLength},
338
	} {
339
		cfg := Config{BodyLimit: limit.actual}
340
		cfg.MetricConfig.Idc = fmt.Sprint(limit)
341
		_, lc, err := Open("name", &cfg)
342
		require.NoError(t, err)
343
		require.NoError(t, lc.Close())
344
		require.Equal(t, limit.expected, cfg.BodyLimit)
345
	}
346
}
347

348
func Benchmark_RowParser(b *testing.B) {
349
	line := strings.Join([]string{
350
		"REQ", "BENCH", "16866434380042975", "POST", "/bench/mark/test",
351
		`{"Host":"127.0.0.1:9500","IP":"10.10.10.10","RawQuery":"size=1751\u0026hashes=4","X-Crc-Encoded":"1"}`, "200",
352
		`{"Blobstore-Tracer-Traceid":"0e34ac5020793b24","Content-Length":"195","Content-Type":"application/json",` +
353
			`"Trace-Log":["PROXY","a_0_r_19_w_2","ACCESS:22"],"Trace-Tags":["http.method:POST"],"X-Ack-Crc-Encoded":"1"}`,
354
		"199", "22348",
355
	}, "\t")
356
	buff := []byte(line)
357
	sender := NewPrometheusSender(PrometheusConfig{Idc: "Benchmark_RowParser" + strconv.Itoa(rand.Intn(100000))})
358
	b.ResetTimer()
359
	for ii := 0; ii < b.N; ii++ {
360
		sender.Send(buff)
361
	}
362
}
363