[#5] Request metrics per user #5

Merged
KirillovDenis merged 4 commits from feature/billing_metrics into master 2023-01-25 08:02:24 +00:00
7 changed files with 676 additions and 314 deletions

View file

@ -4,6 +4,9 @@ This document outlines major changes between releases.
## [Unreleased]
### Added
- Billing metrics (TrueCloudLab#5)
### Changed
- Update neo-go to v0.101.0 (#14)
- Update viper to v1.15.0 (#14)

View file

@ -42,6 +42,10 @@ func transformToS3Error(err error) error {
return errors.GetAPIError(errors.ErrInternalError)
}
func (h *handler) ResolveBucket(ctx context.Context, bucket string) (*data.BucketInfo, error) {
return h.obj.GetBucketInfo(ctx, bucket)
}
func (h *handler) getBucketAndCheckOwner(r *http.Request, bucket string, header ...string) (*data.BucketInfo, error) {
bktInfo, err := h.obj.GetBucketInfo(r.Context(), bucket)
if err != nil {

473
api/metrics.go Normal file
View file

@ -0,0 +1,473 @@
package api
import (
"context"
"io"
"net/http"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/TrueCloudLab/frostfs-s3-gw/creds/accessbox"
"github.com/TrueCloudLab/frostfs-sdk-go/bearer"
"github.com/prometheus/client_golang/prometheus"
)
type TrafficType int
const (
UnknownTraffic TrafficType = iota
INTraffic TrafficType = iota
OUTTraffic TrafficType = iota
)
func (t TrafficType) String() string {
switch t {
case 1:
return "IN"
case 2:
return "OUT"
default:
return "Unknown"
}
}
type RequestType int
const (
UNKNOWNRequest RequestType = iota
HEADRequest RequestType = iota
PUTRequest RequestType = iota
LISTRequest RequestType = iota
GETRequest RequestType = iota
DELETERequest RequestType = iota
)
func (t RequestType) String() string {
switch t {
case 1:
return "HEAD"
case 2:
return "PUT"
case 3:
return "LIST"
case 4:
return "GET"
case 5:
return "DELETE"
default:
return "Unknown"
}
}
func RequestTypeFromAPI(api string) RequestType {
switch api {
case "Options", "HeadObject", "HeadBucket":
return HEADRequest
case "CreateMultipartUpload", "UploadPartCopy", "UploadPart", "CompleteMultipartUpload",
"PutObjectACL", "PutObjectTagging", "CopyObject", "PutObjectRetention", "PutObjectLegalHold",
"PutObject", "PutBucketCors", "PutBucketACL", "PutBucketLifecycle", "PutBucketEncryption",
"PutBucketPolicy", "PutBucketObjectLockConfig", "PutBucketTagging", "PutBucketVersioning",
"PutBucketNotification", "CreateBucket", "PostObject":
return PUTRequest
case "ListObjectParts", "ListMultipartUploads", "ListObjectsV2M", "ListObjectsV2", "ListBucketVersions",
"ListObjectsV1", "ListBuckets":
return LISTRequest
case "GetObjectACL", "GetObjectTagging", "SelectObjectContent", "GetObjectRetention", "getobjectlegalhold",
"GetObjectAttributes", "GetObject", "GetBucketLocation", "GetBucketPolicy",
"GetBucketLifecycle", "GetBucketEncryption", "GetBucketCors", "GetBucketACL",
"GetBucketWebsite", "GetBucketAccelerate", "GetBucketRequestPayment", "GetBucketLogging",
"GetBucketReplication", "GetBucketTagging", "GetBucketObjectLockConfig",
"GetBucketVersioning", "GetBucketNotification", "ListenBucketNotification":
return GETRequest
case "AbortMultipartUpload", "DeleteObjectTagging", "DeleteObject", "DeleteBucketCors",
"DeleteBucketWebsite", "DeleteBucketTagging", "DeleteMultipleObjects", "DeleteBucketPolicy",
"DeleteBucketLifecycle", "DeleteBucketEncryption", "DeleteBucket":
return DELETERequest
default:
return UNKNOWNRequest
}
}
type OperationList [6]int
type (
// HTTPAPIStats holds statistics information about
// the API given in the requests.
HTTPAPIStats struct {
apiStats map[string]int
sync.RWMutex
}
UsersAPIStats struct {
users map[string]*userAPIStats
sync.RWMutex
}
bucketKey struct {
name string
cid string
}
bucketStat struct {
Operations OperationList
InTraffic uint64
OutTraffic uint64
}
userAPIStats struct {
buckets map[bucketKey]bucketStat
user string
}
UserBucketInfo struct {
User string
Bucket string
ContainerID string
}
UserMetricsInfo struct {
UserBucketInfo
Operation RequestType
Requests int
}
UserTrafficMetricsInfo struct {
UserBucketInfo
Type TrafficType
Value uint64
}
UserMetrics struct {
Requests []UserMetricsInfo
Traffic []UserTrafficMetricsInfo
}
// HTTPStats holds statistics information about
// HTTP requests made by all clients.
HTTPStats struct {
usersS3Requests UsersAPIStats
currentS3Requests HTTPAPIStats
totalS3Requests HTTPAPIStats
totalS3Errors HTTPAPIStats
totalInputBytes uint64
totalOutputBytes uint64
}
readCounter struct {
io.ReadCloser
countBytes uint64
}
writeCounter struct {
http.ResponseWriter
countBytes uint64
}
responseWrapper struct {
sync.Once
http.ResponseWriter
statusCode int
startTime time.Time
}
)
const systemPath = "/system"
var (
httpStatsMetric = new(HTTPStats)
httpRequestsDuration = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Name: "frostfs_s3_request_seconds",
Help: "Time taken by requests served by current FrostFS S3 Gate instance",
Buckets: []float64{.05, .1, .25, .5, 1, 2.5, 5, 10},
},
[]string{"api"},
)
)
// Collects HTTP metrics for FrostFS S3 Gate in Prometheus specific format
// and sends to the given channel.
func collectHTTPMetrics(ch chan<- prometheus.Metric) {
for api, value := range httpStatsMetric.currentS3Requests.Load() {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "requests", "current"),
"Total number of running s3 requests in current FrostFS S3 Gate instance",
[]string{"api"}, nil),
prometheus.CounterValue,
float64(value),
api,
)
}
for api, value := range httpStatsMetric.totalS3Requests.Load() {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "requests", "total"),
"Total number of s3 requests in current FrostFS S3 Gate instance",
[]string{"api"}, nil),
prometheus.CounterValue,
float64(value),
api,
)
}
for api, value := range httpStatsMetric.totalS3Errors.Load() {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "errors", "total"),
"Total number of s3 errors in current FrostFS S3 Gate instance",
[]string{"api"}, nil),
prometheus.CounterValue,
float64(value),
api,
)
}
}
func collectUserMetrics(ch chan<- prometheus.Metric) {
userMetrics := httpStatsMetric.usersS3Requests.DumpMetrics()
for _, value := range userMetrics.Requests {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "user_requests", "count"),
"",
[]string{"user", "bucket", "cid", "operation"}, nil),
prometheus.CounterValue,
float64(value.Requests),
value.User,
value.Bucket,
value.ContainerID,
value.Operation.String(),
)
}
for _, value := range userMetrics.Traffic {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "user_traffic", "bytes"),
"",
[]string{"user", "bucket", "cid", "type"}, nil),
prometheus.CounterValue,
float64(value.Value),
value.User,
value.Bucket,
value.ContainerID,
value.Type.String(),
)
}
}
// CIDResolveFunc is a func to resolve CID in Stats handler.
type CIDResolveFunc func(ctx context.Context, reqInfo *ReqInfo) (cnrID string)
// Stats is a handler that update metrics.
func Stats(f http.HandlerFunc, resolveCID CIDResolveFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
reqInfo := GetReqInfo(r.Context())
httpStatsMetric.currentS3Requests.Inc(reqInfo.API)
defer httpStatsMetric.currentS3Requests.Dec(reqInfo.API)
in := &readCounter{ReadCloser: r.Body}
out := &writeCounter{ResponseWriter: w}
r.Body = in
statsWriter := &responseWrapper{
ResponseWriter: out,
startTime: time.Now(),
}
f(statsWriter, r)
// Time duration in secs since the call started.
// We don't need to do nanosecond precision here
// simply for the fact that it is not human-readable.
durationSecs := time.Since(statsWriter.startTime).Seconds()
user := resolveUser(r.Context())
cnrID := resolveCID(r.Context(), reqInfo)
httpStatsMetric.usersS3Requests.Update(user, reqInfo.BucketName, cnrID, RequestTypeFromAPI(reqInfo.API), in.countBytes, out.countBytes)
code := statsWriter.statusCode
// A successful request has a 2xx response code
successReq := code >= http.StatusOK && code < http.StatusMultipleChoices
if !strings.HasSuffix(r.URL.Path, systemPath) {
httpStatsMetric.totalS3Requests.Inc(reqInfo.API)
if !successReq && code != 0 {
httpStatsMetric.totalS3Errors.Inc(reqInfo.API)
}
}
if r.Method == http.MethodGet {
// Increment the prometheus http request response histogram with appropriate label
httpRequestsDuration.With(prometheus.Labels{"api": reqInfo.API}).Observe(durationSecs)
}
atomic.AddUint64(&httpStatsMetric.totalInputBytes, in.countBytes)
atomic.AddUint64(&httpStatsMetric.totalOutputBytes, out.countBytes)
}
}
func resolveUser(ctx context.Context) string {
user := "anon"
if bd, ok := ctx.Value(BoxData).(*accessbox.Box); ok && bd != nil && bd.Gate != nil && bd.Gate.BearerToken != nil {
user = bearer.ResolveIssuer(*bd.Gate.BearerToken).String()
}
return user
}
// Inc increments the api stats counter.
func (stats *HTTPAPIStats) Inc(api string) {
if stats == nil {
return
}
stats.Lock()
defer stats.Unlock()
if stats.apiStats == nil {
stats.apiStats = make(map[string]int)
}
stats.apiStats[api]++
}
// Dec increments the api stats counter.
func (stats *HTTPAPIStats) Dec(api string) {
if stats == nil {
return
}
stats.Lock()
defer stats.Unlock()
if val, ok := stats.apiStats[api]; ok && val > 0 {
stats.apiStats[api]--
}
}
// Load returns the recorded stats.
func (stats *HTTPAPIStats) Load() map[string]int {
stats.Lock()
defer stats.Unlock()
var apiStats = make(map[string]int, len(stats.apiStats))
for k, v := range stats.apiStats {
apiStats[k] = v
}
return apiStats
}
func (u *UsersAPIStats) Update(user, bucket, cnrID string, reqType RequestType, in, out uint64) {
u.Lock()
defer u.Unlock()
usersStat := u.users[user]
if usersStat == nil {
if u.users == nil {
u.users = make(map[string]*userAPIStats)
}
usersStat = &userAPIStats{
buckets: make(map[bucketKey]bucketStat, 1),
user: user,
}
u.users[user] = usersStat
}
key := bucketKey{
name: bucket,
cid: cnrID,
}
bktStat := usersStat.buckets[key]
bktStat.Operations[reqType]++
bktStat.InTraffic += in
bktStat.OutTraffic += out
usersStat.buckets[key] = bktStat
}
func (u *UsersAPIStats) DumpMetrics() UserMetrics {
u.Lock()
defer u.Unlock()
result := UserMetrics{
Requests: make([]UserMetricsInfo, 0, len(u.users)),
Traffic: make([]UserTrafficMetricsInfo, 0, len(u.users)),
}
for user, userStat := range u.users {
for key, bktStat := range userStat.buckets {
userBktInfo := UserBucketInfo{
User: user,
Bucket: key.name,
ContainerID: key.cid,
}
if bktStat.InTraffic != 0 {
result.Traffic = append(result.Traffic, UserTrafficMetricsInfo{
UserBucketInfo: userBktInfo,
Type: INTraffic,
Value: bktStat.InTraffic,
})
}
if bktStat.OutTraffic != 0 {
result.Traffic = append(result.Traffic, UserTrafficMetricsInfo{
UserBucketInfo: userBktInfo,
Type: OUTTraffic,
Value: bktStat.OutTraffic,
})
}
for op, val := range bktStat.Operations {
if val != 0 {
result.Requests = append(result.Requests, UserMetricsInfo{
UserBucketInfo: userBktInfo,
Operation: RequestType(op),
Requests: val,
})
}
}
}
}
u.users = make(map[string]*userAPIStats)
return result
}
func (st *HTTPStats) getInputBytes() uint64 {
return atomic.LoadUint64(&st.totalInputBytes)
}
func (st *HTTPStats) getOutputBytes() uint64 {
return atomic.LoadUint64(&st.totalOutputBytes)
}
// WriteHeader -- writes http status code.
func (w *responseWrapper) WriteHeader(code int) {
w.Do(func() {
w.statusCode = code
w.ResponseWriter.WriteHeader(code)
})
}
// Flush -- calls the underlying Flush.
func (w *responseWrapper) Flush() {
if f, ok := w.ResponseWriter.(http.Flusher); ok {
f.Flush()
}
}
func (w *writeCounter) Write(p []byte) (int, error) {
n, err := w.ResponseWriter.Write(p)
atomic.AddUint64(&w.countBytes, uint64(n))
return n, err
}
func (r *readCounter) Read(p []byte) (int, error) {
n, err := r.ReadCloser.Read(p)
atomic.AddUint64(&r.countBytes, uint64(n))
return n, err
}

View file

@ -1,229 +0,0 @@
package metrics
import (
"io"
"net/http"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/prometheus/client_golang/prometheus"
)
type (
// HTTPAPIStats holds statistics information about
// the API given in the requests.
HTTPAPIStats struct {
apiStats map[string]int
sync.RWMutex
}
// HTTPStats holds statistics information about
// HTTP requests made by all clients.
HTTPStats struct {
currentS3Requests HTTPAPIStats
totalS3Requests HTTPAPIStats
totalS3Errors HTTPAPIStats
totalInputBytes uint64
totalOutputBytes uint64
}
readCounter struct {
io.ReadCloser
countBytes uint64
}
writeCounter struct {
http.ResponseWriter
countBytes uint64
}
responseWrapper struct {
sync.Once
http.ResponseWriter
statusCode int
startTime time.Time
}
)
const systemPath = "/system"
var (
httpStatsMetric = new(HTTPStats)
httpRequestsDuration = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Name: "frostfs_s3_request_seconds",
Help: "Time taken by requests served by current FrostFS S3 Gate instance",
Buckets: []float64{.05, .1, .25, .5, 1, 2.5, 5, 10},
},
[]string{"api"},
)
)
// Collects HTTP metrics for FrostFS S3 Gate in Prometheus specific format
// and sends to the given channel.
func collectHTTPMetrics(ch chan<- prometheus.Metric) {
for api, value := range httpStatsMetric.currentS3Requests.Load() {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "requests", "current"),
"Total number of running s3 requests in current FrostFS S3 Gate instance",
[]string{"api"}, nil),
prometheus.CounterValue,
float64(value),
api,
)
}
for api, value := range httpStatsMetric.totalS3Requests.Load() {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "requests", "total"),
"Total number of s3 requests in current FrostFS S3 Gate instance",
[]string{"api"}, nil),
prometheus.CounterValue,
float64(value),
api,
)
}
for api, value := range httpStatsMetric.totalS3Errors.Load() {
ch <- prometheus.MustNewConstMetric(
prometheus.NewDesc(
prometheus.BuildFQName("frostfs_s3", "errors", "total"),
"Total number of s3 errors in current FrostFS S3 Gate instance",
[]string{"api"}, nil),
prometheus.CounterValue,
float64(value),
api,
)
}
}
// APIStats wraps http handler for api with basic statistics collection.
func APIStats(api string, f http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
httpStatsMetric.currentS3Requests.Inc(api)
defer httpStatsMetric.currentS3Requests.Dec(api)
in := &readCounter{ReadCloser: r.Body}
out := &writeCounter{ResponseWriter: w}
r.Body = in
statsWriter := &responseWrapper{
ResponseWriter: out,
startTime: time.Now(),
}
f.ServeHTTP(statsWriter, r)
// Time duration in secs since the call started.
// We don't need to do nanosecond precision here
// simply for the fact that it is not human readable.
durationSecs := time.Since(statsWriter.startTime).Seconds()
httpStatsMetric.updateStats(api, statsWriter, r, durationSecs)
atomic.AddUint64(&httpStatsMetric.totalInputBytes, in.countBytes)
atomic.AddUint64(&httpStatsMetric.totalOutputBytes, out.countBytes)
}
}
// Inc increments the api stats counter.
func (stats *HTTPAPIStats) Inc(api string) {
if stats == nil {
return
}
stats.Lock()
defer stats.Unlock()
if stats.apiStats == nil {
stats.apiStats = make(map[string]int)
}
stats.apiStats[api]++
}
// Dec increments the api stats counter.
func (stats *HTTPAPIStats) Dec(api string) {
if stats == nil {
return
}
stats.Lock()
defer stats.Unlock()
if val, ok := stats.apiStats[api]; ok && val > 0 {
stats.apiStats[api]--
}
}
// Load returns the recorded stats.
func (stats *HTTPAPIStats) Load() map[string]int {
stats.Lock()
defer stats.Unlock()
var apiStats = make(map[string]int, len(stats.apiStats))
for k, v := range stats.apiStats {
apiStats[k] = v
}
return apiStats
}
func (st *HTTPStats) getInputBytes() uint64 {
return atomic.LoadUint64(&st.totalInputBytes)
}
func (st *HTTPStats) getOutputBytes() uint64 {
return atomic.LoadUint64(&st.totalOutputBytes)
}
// Update statistics from http request and response data.
func (st *HTTPStats) updateStats(api string, w http.ResponseWriter, r *http.Request, durationSecs float64) {
var code int
if res, ok := w.(*responseWrapper); ok {
code = res.statusCode
}
// A successful request has a 2xx response code
successReq := code >= http.StatusOK && code < http.StatusMultipleChoices
if !strings.HasSuffix(r.URL.Path, systemPath) {
st.totalS3Requests.Inc(api)
if !successReq && code != 0 {
st.totalS3Errors.Inc(api)
}
}
if r.Method == http.MethodGet {
// Increment the prometheus http request response histogram with appropriate label
httpRequestsDuration.With(prometheus.Labels{"api": api}).Observe(durationSecs)
}
}
// WriteHeader -- writes http status code.
func (w *responseWrapper) WriteHeader(code int) {
w.Do(func() {
w.statusCode = code
w.ResponseWriter.WriteHeader(code)
})
}
// Flush -- calls the underlying Flush.
func (w *responseWrapper) Flush() {
if f, ok := w.ResponseWriter.(http.Flusher); ok {
f.Flush()
}
}
func (w *writeCounter) Write(p []byte) (int, error) {
n, err := w.ResponseWriter.Write(p)
atomic.AddUint64(&w.countBytes, uint64(n))
return n, err
}
func (r *readCounter) Read(p []byte) (int, error) {
n, err := r.ReadCloser.Read(p)
atomic.AddUint64(&r.countBytes, uint64(n))
return n, err
}

View file

@ -1,4 +1,4 @@
package metrics
package api
import (
"github.com/TrueCloudLab/frostfs-s3-gw/internal/version"
@ -64,5 +64,6 @@ func (s *stats) Collect(ch chan<- prometheus.Metric) {
// connect collectors
collectHTTPMetrics(ch)
collectUserMetrics(ch)
collectNetworkMetrics(ch)
}

View file

@ -6,7 +6,7 @@ import (
"sync"
"github.com/TrueCloudLab/frostfs-s3-gw/api/auth"
"github.com/TrueCloudLab/frostfs-s3-gw/api/metrics"
"github.com/TrueCloudLab/frostfs-s3-gw/api/data"
"github.com/google/uuid"
"github.com/gorilla/mux"
"go.uber.org/zap"
@ -82,6 +82,8 @@ type (
AbortMultipartUploadHandler(http.ResponseWriter, *http.Request)
ListPartsHandler(w http.ResponseWriter, r *http.Request)
ListMultipartUploadsHandler(http.ResponseWriter, *http.Request)
ResolveBucket(ctx context.Context, bucket string) (*data.BucketInfo, error)
}
// mimeType represents various MIME types used in API responses.
@ -106,7 +108,7 @@ const (
MimeXML mimeType = "application/xml"
)
var _ = logErrorResponse
var _ = logSuccessResponse
func (lrw *logResponseWriter) WriteHeader(code int) {
lrw.Do(func() {
@ -145,7 +147,37 @@ func appendCORS(handler Handler) mux.MiddlewareFunc {
}
}
func logErrorResponse(l *zap.Logger) mux.MiddlewareFunc {
// BucketResolveFunc is a func to resolve bucket info by name.
type BucketResolveFunc func(ctx context.Context, bucket string) (*data.BucketInfo, error)
// metricsMiddleware wraps http handler for api with basic statistics collection.
func metricsMiddleware(log *zap.Logger, resolveBucket BucketResolveFunc) mux.MiddlewareFunc {
return func(h http.Handler) http.Handler {
return Stats(h.ServeHTTP, resolveCID(log, resolveBucket))
}
}
// resolveCID forms CIDResolveFunc using BucketResolveFunc.
func resolveCID(log *zap.Logger, resolveBucket BucketResolveFunc) CIDResolveFunc {
return func(ctx context.Context, reqInfo *ReqInfo) (cnrID string) {
if reqInfo.BucketName == "" || reqInfo.API == "CreateBucket" || reqInfo.API == "" {
return ""
}
bktInfo, err := resolveBucket(ctx, reqInfo.BucketName)
if err != nil {
log.Debug("failed to resolve CID",
zap.String("request_id", reqInfo.RequestID), zap.String("method", reqInfo.API),
zap.String("bucket", reqInfo.BucketName), zap.String("object", reqInfo.ObjectName),
zap.Error(err))
return ""
}
return bktInfo.CID.EncodeToString()
}
}
func logSuccessResponse(l *zap.Logger) mux.MiddlewareFunc {
return func(h http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
lw := &logResponseWriter{ResponseWriter: w}
@ -183,6 +215,30 @@ func GetRequestID(v interface{}) string {
}
}
func setErrorAPI(apiName string, h http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
ctx := SetReqInfo(r.Context(), &ReqInfo{API: apiName})
h.ServeHTTP(w, r.WithContext(ctx))
})
}
// attachErrorHandler set NotFoundHandler and MethodNotAllowedHandler for mux.Router.
func attachErrorHandler(api *mux.Router, log *zap.Logger, h Handler, center auth.Center) {
middlewares := []mux.MiddlewareFunc{
AuthMiddleware(log, center),
metricsMiddleware(log, h.ResolveBucket),
}
var errorHandler http.Handler = http.HandlerFunc(errorResponseHandler)
for i := len(middlewares) - 1; i >= 0; i-- {
errorHandler = middlewares[i](errorHandler)
}
// If none of the routes match, add default error handler routes
api.NotFoundHandler = setErrorAPI("NotFound", errorHandler)
api.MethodNotAllowedHandler = setErrorAPI("MethodNotAllowed", errorHandler)
}
// Attach adds S3 API handlers from h to r for domains with m client limit using
// center authentication and log logger.
func Attach(r *mux.Router, domains []string, m MaxClients, h Handler, center auth.Center, log *zap.Logger) {
@ -192,12 +248,16 @@ func Attach(r *mux.Router, domains []string, m MaxClients, h Handler, center aut
// -- prepare request
setRequestID,
// Attach user authentication for all S3 routes.
AuthMiddleware(log, center),
metricsMiddleware(log, h.ResolveBucket),
// -- logging error requests
logErrorResponse(log),
logSuccessResponse(log),
)
// Attach user authentication for all S3 routes.
AttachUserAuth(api, center, log)
attachErrorHandler(api, log, h, center)
buckets := make([]*mux.Router, 0, len(domains)+1)
buckets = append(buckets, api.PathPrefix("/{bucket}").Subrouter())
@ -213,277 +273,327 @@ func Attach(r *mux.Router, domains []string, m MaxClients, h Handler, center aut
// -- append CORS headers to a response for
appendCORS(h),
)
bucket.Methods(http.MethodOptions).HandlerFunc(m.Handle(metrics.APIStats("preflight", h.Preflight))).Name("Options")
bucket.Methods(http.MethodOptions).HandlerFunc(
m.Handle(h.Preflight)).
Name("Options")
bucket.Methods(http.MethodHead).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("headobject", h.HeadObjectHandler))).Name("HeadObject")
m.Handle(h.HeadObjectHandler)).
Name("HeadObject")
// CopyObjectPart
bucket.Methods(http.MethodPut).Path("/{object:.+}").Headers(hdrAmzCopySource, "").HandlerFunc(m.Handle(metrics.APIStats("uploadpartcopy", h.UploadPartCopy))).Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}").
bucket.Methods(http.MethodPut).Path("/{object:.+}").Headers(hdrAmzCopySource, "").HandlerFunc(
m.Handle(h.UploadPartCopy)).
Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}").
Name("UploadPartCopy")
// PutObjectPart
bucket.Methods(http.MethodPut).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("uploadpart", h.UploadPartHandler))).Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}").
m.Handle(h.UploadPartHandler)).
Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}").
Name("UploadPart")
// ListParts
bucket.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("listobjectparts", h.ListPartsHandler))).Queries("uploadId", "{uploadId:.*}").
m.Handle(h.ListPartsHandler)).
Queries("uploadId", "{uploadId:.*}").
Name("ListObjectParts")
// CompleteMultipartUpload
bucket.Methods(http.MethodPost).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("completemutipartupload", h.CompleteMultipartUploadHandler))).Queries("uploadId", "{uploadId:.*}").
m.Handle(h.CompleteMultipartUploadHandler)).
Queries("uploadId", "{uploadId:.*}").
Name("CompleteMultipartUpload")
// CreateMultipartUpload
bucket.Methods(http.MethodPost).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("createmultipartupload", h.CreateMultipartUploadHandler))).Queries("uploads", "").
m.Handle(h.CreateMultipartUploadHandler)).
Queries("uploads", "").
Name("CreateMultipartUpload")
// AbortMultipartUpload
bucket.Methods(http.MethodDelete).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("abortmultipartupload", h.AbortMultipartUploadHandler))).Queries("uploadId", "{uploadId:.*}").
m.Handle(h.AbortMultipartUploadHandler)).
Queries("uploadId", "{uploadId:.*}").
Name("AbortMultipartUpload")
// ListMultipartUploads
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("listmultipartuploads", h.ListMultipartUploadsHandler))).Queries("uploads", "").
m.Handle(h.ListMultipartUploadsHandler)).
Queries("uploads", "").
Name("ListMultipartUploads")
// GetObjectACL -- this is a dummy call.
bucket.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("getobjectacl", h.GetObjectACLHandler))).Queries("acl", "").
m.Handle(h.GetObjectACLHandler)).
Queries("acl", "").
Name("GetObjectACL")
// PutObjectACL -- this is a dummy call.
bucket.Methods(http.MethodPut).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("putobjectacl", h.PutObjectACLHandler))).Queries("acl", "").
m.Handle(h.PutObjectACLHandler)).
Queries("acl", "").
Name("PutObjectACL")
// GetObjectTagging
bucket.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("getobjecttagging", h.GetObjectTaggingHandler))).Queries("tagging", "").
m.Handle(h.GetObjectTaggingHandler)).
Queries("tagging", "").
Name("GetObjectTagging")
// PutObjectTagging
bucket.Methods(http.MethodPut).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("putobjecttagging", h.PutObjectTaggingHandler))).Queries("tagging", "").
m.Handle(h.PutObjectTaggingHandler)).
Queries("tagging", "").
Name("PutObjectTagging")
// DeleteObjectTagging
bucket.Methods(http.MethodDelete).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("deleteobjecttagging", h.DeleteObjectTaggingHandler))).Queries("tagging", "").
m.Handle(h.DeleteObjectTaggingHandler)).
Queries("tagging", "").
Name("DeleteObjectTagging")
// SelectObjectContent
bucket.Methods(http.MethodPost).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("selectobjectcontent", h.SelectObjectContentHandler))).Queries("select", "").Queries("select-type", "2").
m.Handle(h.SelectObjectContentHandler)).
Queries("select", "").Queries("select-type", "2").
Name("SelectObjectContent")
// GetObjectRetention
bucket.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("getobjectretention", h.GetObjectRetentionHandler))).Queries("retention", "").
m.Handle(h.GetObjectRetentionHandler)).
Queries("retention", "").
Name("GetObjectRetention")
// GetObjectLegalHold
bucket.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("getobjectlegalhold", h.GetObjectLegalHoldHandler))).Queries("legal-hold", "").
m.Handle(h.GetObjectLegalHoldHandler)).
Queries("legal-hold", "").
Name("GetObjectLegalHold")
// GetObjectAttributes
bucket.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("getobjectattributes", h.GetObjectAttributesHandler))).Queries("attributes", "").
m.Handle(h.GetObjectAttributesHandler)).
Queries("attributes", "").
Name("GetObjectAttributes")
// GetObject
bucket.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("getobject", h.GetObjectHandler))).
m.Handle(h.GetObjectHandler)).
Name("GetObject")
// CopyObject
bucket.Methods(http.MethodPut).Path("/{object:.+}").Headers(hdrAmzCopySource, "").HandlerFunc(m.Handle(metrics.APIStats("copyobject", h.CopyObjectHandler))).
bucket.Methods(http.MethodPut).Path("/{object:.+}").Headers(hdrAmzCopySource, "").HandlerFunc(
m.Handle(h.CopyObjectHandler)).
Name("CopyObject")
// PutObjectRetention
bucket.Methods(http.MethodPut).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("putobjectretention", h.PutObjectRetentionHandler))).Queries("retention", "").
m.Handle(h.PutObjectRetentionHandler)).
Queries("retention", "").
Name("PutObjectRetention")
// PutObjectLegalHold
bucket.Methods(http.MethodPut).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("putobjectlegalhold", h.PutObjectLegalHoldHandler))).Queries("legal-hold", "").
m.Handle(h.PutObjectLegalHoldHandler)).
Queries("legal-hold", "").
Name("PutObjectLegalHold")
// PutObject
bucket.Methods(http.MethodPut).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("putobject", h.PutObjectHandler))).
m.Handle(h.PutObjectHandler)).
Name("PutObject")
// DeleteObject
bucket.Methods(http.MethodDelete).Path("/{object:.+}").HandlerFunc(
m.Handle(metrics.APIStats("deleteobject", h.DeleteObjectHandler))).
m.Handle(h.DeleteObjectHandler)).
Name("DeleteObject")
// Bucket operations
// GetBucketLocation
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketlocation", h.GetBucketLocationHandler))).Queries("location", "").
m.Handle(h.GetBucketLocationHandler)).
Queries("location", "").
Name("GetBucketLocation")
// GetBucketPolicy
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketpolicy", h.GetBucketPolicyHandler))).Queries("policy", "").
m.Handle(h.GetBucketPolicyHandler)).
Queries("policy", "").
Name("GetBucketPolicy")
// GetBucketLifecycle
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketlifecycle", h.GetBucketLifecycleHandler))).Queries("lifecycle", "").
m.Handle(h.GetBucketLifecycleHandler)).
Queries("lifecycle", "").
Name("GetBucketLifecycle")
// GetBucketEncryption
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketencryption", h.GetBucketEncryptionHandler))).Queries("encryption", "").
m.Handle(h.GetBucketEncryptionHandler)).
Queries("encryption", "").
Name("GetBucketEncryption")
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketcors", h.GetBucketCorsHandler))).Queries("cors", "").
m.Handle(h.GetBucketCorsHandler)).
Queries("cors", "").
Name("GetBucketCors")
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketcors", h.PutBucketCorsHandler))).Queries("cors", "").
m.Handle(h.PutBucketCorsHandler)).
Queries("cors", "").
Name("PutBucketCors")
bucket.Methods(http.MethodDelete).HandlerFunc(
m.Handle(metrics.APIStats("deletebucketcors", h.DeleteBucketCorsHandler))).Queries("cors", "").
m.Handle(h.DeleteBucketCorsHandler)).
Queries("cors", "").
Name("DeleteBucketCors")
// Dummy Bucket Calls
// GetBucketACL -- this is a dummy call.
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketacl", h.GetBucketACLHandler))).Queries("acl", "").
m.Handle(h.GetBucketACLHandler)).
Queries("acl", "").
Name("GetBucketACL")
// PutBucketACL -- this is a dummy call.
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketacl", h.PutBucketACLHandler))).Queries("acl", "").
m.Handle(h.PutBucketACLHandler)).
Queries("acl", "").
Name("PutBucketACL")
// GetBucketWebsiteHandler -- this is a dummy call.
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketwebsite", h.GetBucketWebsiteHandler))).Queries("website", "").
m.Handle(h.GetBucketWebsiteHandler)).
Queries("website", "").
Name("GetBucketWebsite")
// GetBucketAccelerateHandler -- this is a dummy call.
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketaccelerate", h.GetBucketAccelerateHandler))).Queries("accelerate", "").
m.Handle(h.GetBucketAccelerateHandler)).
Queries("accelerate", "").
Name("GetBucketAccelerate")
// GetBucketRequestPaymentHandler -- this is a dummy call.
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketrequestpayment", h.GetBucketRequestPaymentHandler))).Queries("requestPayment", "").
m.Handle(h.GetBucketRequestPaymentHandler)).
Queries("requestPayment", "").
Name("GetBucketRequestPayment")
// GetBucketLoggingHandler -- this is a dummy call.
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketlogging", h.GetBucketLoggingHandler))).Queries("logging", "").
m.Handle(h.GetBucketLoggingHandler)).
Queries("logging", "").
Name("GetBucketLogging")
// GetBucketLifecycleHandler -- this is a dummy call.
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketlifecycle", h.GetBucketLifecycleHandler))).Queries("lifecycle", "").
Name("GetBucketLifecycle")
// GetBucketReplicationHandler -- this is a dummy call.
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketreplication", h.GetBucketReplicationHandler))).Queries("replication", "").
m.Handle(h.GetBucketReplicationHandler)).
Queries("replication", "").
Name("GetBucketReplication")
// GetBucketTaggingHandler
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbuckettagging", h.GetBucketTaggingHandler))).Queries("tagging", "").
m.Handle(h.GetBucketTaggingHandler)).
Queries("tagging", "").
Name("GetBucketTagging")
// DeleteBucketWebsiteHandler
bucket.Methods(http.MethodDelete).HandlerFunc(
m.Handle(metrics.APIStats("deletebucketwebsite", h.DeleteBucketWebsiteHandler))).Queries("website", "").
m.Handle(h.DeleteBucketWebsiteHandler)).
Queries("website", "").
Name("DeleteBucketWebsite")
// DeleteBucketTaggingHandler
bucket.Methods(http.MethodDelete).HandlerFunc(
m.Handle(metrics.APIStats("deletebuckettagging", h.DeleteBucketTaggingHandler))).Queries("tagging", "").
m.Handle(h.DeleteBucketTaggingHandler)).
Queries("tagging", "").
Name("DeleteBucketTagging")
// GetBucketObjectLockConfig
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketobjectlockconfiguration", h.GetBucketObjectLockConfigHandler))).Queries("object-lock", "").
m.Handle(h.GetBucketObjectLockConfigHandler)).
Queries("object-lock", "").
Name("GetBucketObjectLockConfig")
// GetBucketVersioning
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketversioning", h.GetBucketVersioningHandler))).Queries("versioning", "").
m.Handle(h.GetBucketVersioningHandler)).
Queries("versioning", "").
Name("GetBucketVersioning")
// GetBucketNotification
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("getbucketnotification", h.GetBucketNotificationHandler))).Queries("notification", "").
m.Handle(h.GetBucketNotificationHandler)).
Queries("notification", "").
Name("GetBucketNotification")
// ListenBucketNotification
bucket.Methods(http.MethodGet).HandlerFunc(metrics.APIStats("listenbucketnotification", h.ListenBucketNotificationHandler)).Queries("events", "{events:.*}").
bucket.Methods(http.MethodGet).HandlerFunc(h.ListenBucketNotificationHandler).
Queries("events", "{events:.*}").
Name("ListenBucketNotification")
// ListObjectsV2M
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("listobjectsv2M", h.ListObjectsV2MHandler))).Queries("list-type", "2", "metadata", "true").
m.Handle(h.ListObjectsV2MHandler)).
Queries("list-type", "2", "metadata", "true").
Name("ListObjectsV2M")
// ListObjectsV2
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("listobjectsv2", h.ListObjectsV2Handler))).Queries("list-type", "2").
m.Handle(h.ListObjectsV2Handler)).
Queries("list-type", "2").
Name("ListObjectsV2")
// ListBucketVersions
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("listbucketversions", h.ListBucketObjectVersionsHandler))).Queries("versions", "").
m.Handle(h.ListBucketObjectVersionsHandler)).
Queries("versions", "").
Name("ListBucketVersions")
// ListObjectsV1 (Legacy)
bucket.Methods(http.MethodGet).HandlerFunc(
m.Handle(metrics.APIStats("listobjectsv1", h.ListObjectsV1Handler))).
m.Handle(h.ListObjectsV1Handler)).
Name("ListObjectsV1")
// PutBucketLifecycle
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketlifecycle", h.PutBucketLifecycleHandler))).Queries("lifecycle", "").
m.Handle(h.PutBucketLifecycleHandler)).
Queries("lifecycle", "").
Name("PutBucketLifecycle")
// PutBucketEncryption
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketencryption", h.PutBucketEncryptionHandler))).Queries("encryption", "").
m.Handle(h.PutBucketEncryptionHandler)).
Queries("encryption", "").
Name("PutBucketEncryption")
// PutBucketPolicy
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketpolicy", h.PutBucketPolicyHandler))).Queries("policy", "").
m.Handle(h.PutBucketPolicyHandler)).
Queries("policy", "").
Name("PutBucketPolicy")
// PutBucketObjectLockConfig
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketobjectlockconfig", h.PutBucketObjectLockConfigHandler))).Queries("object-lock", "").
m.Handle(h.PutBucketObjectLockConfigHandler)).
Queries("object-lock", "").
Name("PutBucketObjectLockConfig")
// PutBucketTaggingHandler
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbuckettagging", h.PutBucketTaggingHandler))).Queries("tagging", "").
m.Handle(h.PutBucketTaggingHandler)).
Queries("tagging", "").
Name("PutBucketTagging")
// PutBucketVersioning
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketversioning", h.PutBucketVersioningHandler))).Queries("versioning", "").
m.Handle(h.PutBucketVersioningHandler)).
Queries("versioning", "").
Name("PutBucketVersioning")
// PutBucketNotification
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketnotification", h.PutBucketNotificationHandler))).Queries("notification", "").
m.Handle(h.PutBucketNotificationHandler)).
Queries("notification", "").
Name("PutBucketNotification")
// CreateBucket
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("createbucket", h.CreateBucketHandler))).
m.Handle(h.CreateBucketHandler)).
Name("CreateBucket")
// HeadBucket
bucket.Methods(http.MethodHead).HandlerFunc(
m.Handle(metrics.APIStats("headbucket", h.HeadBucketHandler))).
m.Handle(h.HeadBucketHandler)).
Name("HeadBucket")
// PostPolicy
bucket.Methods(http.MethodPost).HeadersRegexp(hdrContentType, "multipart/form-data*").HandlerFunc(
m.Handle(metrics.APIStats("postobject", h.PostObject))).
m.Handle(h.PostObject)).
Name("PostObject")
// DeleteMultipleObjects
bucket.Methods(http.MethodPost).HandlerFunc(
m.Handle(metrics.APIStats("deletemultipleobjects", h.DeleteMultipleObjectsHandler))).Queries("delete", "").
m.Handle(h.DeleteMultipleObjectsHandler)).
Queries("delete", "").
Name("DeleteMultipleObjects")
// DeleteBucketPolicy
bucket.Methods(http.MethodDelete).HandlerFunc(
m.Handle(metrics.APIStats("deletebucketpolicy", h.DeleteBucketPolicyHandler))).Queries("policy", "").
m.Handle(h.DeleteBucketPolicyHandler)).
Queries("policy", "").
Name("DeleteBucketPolicy")
// DeleteBucketLifecycle
bucket.Methods(http.MethodDelete).HandlerFunc(
m.Handle(metrics.APIStats("deletebucketlifecycle", h.DeleteBucketLifecycleHandler))).Queries("lifecycle", "").
m.Handle(h.DeleteBucketLifecycleHandler)).
Queries("lifecycle", "").
Name("DeleteBucketLifecycle")
// DeleteBucketEncryption
bucket.Methods(http.MethodDelete).HandlerFunc(
m.Handle(metrics.APIStats("deletebucketencryption", h.DeleteBucketEncryptionHandler))).Queries("encryption", "").
m.Handle(h.DeleteBucketEncryptionHandler)).
Queries("encryption", "").
Name("DeleteBucketEncryption")
// DeleteBucket
bucket.Methods(http.MethodDelete).HandlerFunc(
m.Handle(metrics.APIStats("deletebucket", h.DeleteBucketHandler))).
m.Handle(h.DeleteBucketHandler)).
Name("DeleteBucket")
}
// Root operation
// ListBuckets
api.Methods(http.MethodGet).Path(SlashSeparator).HandlerFunc(
m.Handle(metrics.APIStats("listbuckets", h.ListBucketsHandler))).
m.Handle(h.ListBucketsHandler)).
Name("ListBuckets")
// S3 browser with signature v4 adds '//' for ListBuckets request, so rather
// than failing with UnknownAPIRequest we simply handle it for now.
api.Methods(http.MethodGet).Path(SlashSeparator + SlashSeparator).HandlerFunc(
m.Handle(metrics.APIStats("listbuckets", h.ListBucketsHandler))).
m.Handle(h.ListBucketsHandler)).
Name("ListBuckets")
// If none of the routes match, add default error handler routes
api.NotFoundHandler = metrics.APIStats("notfound", errorResponseHandler)
api.MethodNotAllowedHandler = metrics.APIStats("methodnotallowed", errorResponseHandler)
}

View file

@ -19,9 +19,9 @@ var BoxData = KeyWrapper("__context_box_key")
// ClientTime is an ID used to store client time.Time in a context.
var ClientTime = KeyWrapper("__context_client_time")
// AttachUserAuth adds user authentication via center to router using log for logging.
func AttachUserAuth(router *mux.Router, center auth.Center, log *zap.Logger) {
router.Use(func(h http.Handler) http.Handler {
// AuthMiddleware adds user authentication via center to router using log for logging.
func AuthMiddleware(log *zap.Logger, center auth.Center) mux.MiddlewareFunc {
return func(h http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var ctx context.Context
box, err := center.Authenticate(r)
@ -46,5 +46,5 @@ func AttachUserAuth(router *mux.Router, center auth.Center, log *zap.Logger) {
h.ServeHTTP(w, r.WithContext(ctx))
})
})
}
}