minio/cmd/http-tracer.go

201 lines
6.0 KiB
Go
Raw Normal View History

// Copyright (c) 2015-2021 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"context"
"net"
"net/http"
"reflect"
"regexp"
"runtime"
"strconv"
"strings"
"time"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/handlers"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/mcontext"
)
var ldapPwdRegex = regexp.MustCompile("(^.*?)LDAPPassword=([^&]*?)(&(.*?))?$")
// redact LDAP password if part of string
func redactLDAPPwd(s string) string {
parts := ldapPwdRegex.FindStringSubmatch(s)
2023-03-30 11:03:48 -04:00
if len(parts) > 3 {
return parts[1] + "LDAPPassword=*REDACTED*" + parts[3]
}
return s
}
// getOpName sanitizes the operation name for mc
func getOpName(name string) (op string) {
op = strings.TrimPrefix(name, "github.com/minio/minio/cmd.")
op = strings.TrimSuffix(op, "Handler-fm")
op = strings.Replace(op, "objectAPIHandlers", "s3", 1)
op = strings.Replace(op, "adminAPIHandlers", "admin", 1)
op = strings.Replace(op, "(*storageRESTServer)", "storageR", 1)
op = strings.Replace(op, "(*peerRESTServer)", "peer", 1)
op = strings.Replace(op, "(*lockRESTServer)", "lockR", 1)
op = strings.Replace(op, "(*stsAPIHandlers)", "sts", 1)
perf: websocket grid connectivity for all internode communication (#18461) This PR adds a WebSocket grid feature that allows servers to communicate via a single two-way connection. There are two request types: * Single requests, which are `[]byte => ([]byte, error)`. This is for efficient small roundtrips with small payloads. * Streaming requests which are `[]byte, chan []byte => chan []byte (and error)`, which allows for different combinations of full two-way streams with an initial payload. Only a single stream is created between two machines - and there is, as such, no server/client relation since both sides can initiate and handle requests. Which server initiates the request is decided deterministically on the server names. Requests are made through a mux client and server, which handles message passing, congestion, cancelation, timeouts, etc. If a connection is lost, all requests are canceled, and the calling server will try to reconnect. Registered handlers can operate directly on byte slices or use a higher-level generics abstraction. There is no versioning of handlers/clients, and incompatible changes should be handled by adding new handlers. The request path can be changed to a new one for any protocol changes. First, all servers create a "Manager." The manager must know its address as well as all remote addresses. This will manage all connections. To get a connection to any remote, ask the manager to provide it given the remote address using. ``` func (m *Manager) Connection(host string) *Connection ``` All serverside handlers must also be registered on the manager. This will make sure that all incoming requests are served. The number of in-flight requests and responses must also be given for streaming requests. The "Connection" returned manages the mux-clients. Requests issued to the connection will be sent to the remote. * `func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error)` performs a single request and returns the result. Any deadline provided on the request is forwarded to the server, and canceling the context will make the function return at once. * `func (c *Connection) NewStream(ctx context.Context, h HandlerID, payload []byte) (st *Stream, err error)` will initiate a remote call and send the initial payload. ```Go // A Stream is a two-way stream. // All responses *must* be read by the caller. // If the call is canceled through the context, //The appropriate error will be returned. type Stream struct { // Responses from the remote server. // Channel will be closed after an error or when the remote closes. // All responses *must* be read by the caller until either an error is returned or the channel is closed. // Canceling the context will cause the context cancellation error to be returned. Responses <-chan Response // Requests sent to the server. // If the handler is defined with 0 incoming capacity this will be nil. // Channel *must* be closed to signal the end of the stream. // If the request context is canceled, the stream will no longer process requests. Requests chan<- []byte } type Response struct { Msg []byte Err error } ``` There are generic versions of the server/client handlers that allow the use of type safe implementations for data types that support msgpack marshal/unmarshal.
2023-11-20 20:09:35 -05:00
op = strings.Replace(op, "(*peerS3Server)", "s3", 1)
op = strings.Replace(op, "ClusterCheckHandler", "health.Cluster", 1)
op = strings.Replace(op, "ClusterReadCheckHandler", "health.ClusterRead", 1)
op = strings.Replace(op, "LivenessCheckHandler", "health.Liveness", 1)
op = strings.Replace(op, "ReadinessCheckHandler", "health.Readiness", 1)
op = strings.Replace(op, "-fm", "", 1)
return op
}
// If trace is enabled, execute the request if it is traced by other handlers
// otherwise, generate a trace event with request information but no response.
func httpTracerMiddleware(h http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Setup a http request response recorder - this is needed for
// http stats requests and audit if enabled.
respRecorder := xhttp.NewResponseRecorder(w)
// Setup a http request body recorder
reqRecorder := &xhttp.RequestRecorder{Reader: r.Body}
r.Body = reqRecorder
// Create tracing data structure and associate it to the request context
tc := mcontext.TraceCtxt{
AmzReqID: w.Header().Get(xhttp.AmzRequestID),
RequestRecorder: reqRecorder,
ResponseRecorder: respRecorder,
}
r = r.WithContext(context.WithValue(r.Context(), mcontext.ContextTraceKey, &tc))
reqStartTime := time.Now().UTC()
h.ServeHTTP(respRecorder, r)
reqEndTime := time.Now().UTC()
if globalTrace.NumSubscribers(madmin.TraceS3|madmin.TraceInternal) == 0 {
// no subscribers nothing to trace.
return
}
2022-07-05 17:45:49 -04:00
tt := madmin.TraceInternal
if strings.HasPrefix(tc.FuncName, "s3.") {
2022-07-05 17:45:49 -04:00
tt = madmin.TraceS3
}
// Calculate input body size with headers
reqHeaders := r.Header.Clone()
reqHeaders.Set("Host", r.Host)
if len(r.TransferEncoding) == 0 {
reqHeaders.Set("Content-Length", strconv.Itoa(int(r.ContentLength)))
} else {
reqHeaders.Set("Transfer-Encoding", strings.Join(r.TransferEncoding, ","))
}
inputBytes := reqRecorder.Size()
for k, v := range reqHeaders {
inputBytes += len(k) + len(v)
}
// Calculate node name
nodeName := r.Host
if globalIsDistErasure {
nodeName = globalLocalNodeName
}
if host, port, err := net.SplitHostPort(nodeName); err == nil {
if port == "443" || port == "80" {
nodeName = host
}
}
// Calculate reqPath
reqPath := r.URL.RawPath
if reqPath == "" {
reqPath = r.URL.Path
}
// Calculate function name
funcName := tc.FuncName
if funcName == "" {
funcName = "<unknown>"
}
t := madmin.TraceInfo{
2022-07-05 17:45:49 -04:00
TraceType: tt,
FuncName: funcName,
NodeName: nodeName,
Time: reqStartTime,
2022-07-05 17:45:49 -04:00
Duration: reqEndTime.Sub(respRecorder.StartTime),
Path: reqPath,
Bytes: int64(inputBytes + respRecorder.Size()),
2022-07-05 17:45:49 -04:00
HTTP: &madmin.TraceHTTPStats{
ReqInfo: madmin.TraceRequestInfo{
Time: reqStartTime,
Proto: r.Proto,
Method: r.Method,
RawQuery: redactLDAPPwd(r.URL.RawQuery),
Client: handlers.GetSourceIP(r),
Headers: reqHeaders,
Path: reqPath,
Body: reqRecorder.Data(),
},
RespInfo: madmin.TraceResponseInfo{
Time: reqEndTime,
Headers: respRecorder.Header().Clone(),
StatusCode: respRecorder.StatusCode,
Body: respRecorder.Body(),
},
CallStats: madmin.TraceCallStats{
Latency: reqEndTime.Sub(respRecorder.StartTime),
InputBytes: inputBytes,
OutputBytes: respRecorder.Size(),
TimeToFirstByte: respRecorder.TTFB(),
2022-07-05 17:45:49 -04:00
},
},
}
globalTrace.Publish(t)
})
}
func httpTrace(f http.HandlerFunc, logBody bool) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
tc, ok := r.Context().Value(mcontext.ContextTraceKey).(*mcontext.TraceCtxt)
if !ok {
// Tracing is not enabled for this request
f.ServeHTTP(w, r)
return
}
tc.FuncName = getOpName(runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name())
tc.RequestRecorder.LogBody = logBody
tc.ResponseRecorder.LogAllBody = logBody
tc.ResponseRecorder.LogErrBody = true
f.ServeHTTP(w, r)
}
}
func httpTraceAll(f http.HandlerFunc) http.HandlerFunc {
return httpTrace(f, true)
}
func httpTraceHdrs(f http.HandlerFunc) http.HandlerFunc {
return httpTrace(f, false)
}