mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-08-17 20:56:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Integrated metric manager into the gateway

Browse Source
This commit is contained in:
jonaustin09
2024-05-28 12:50:50 -07:00
committed by Ben McClelland
parent 3408470d7b
commit be6f9a86cd
14 changed files with 451 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/versitygw/main.go
+21 -1
View File
@@ -27,6 +27,7 @@ import (
"github.com/urfave/cli/v2"
"github.com/versity/versitygw/auth"
"github.com/versity/versitygw/backend"
"github.com/versity/versitygw/metrics"
"github.com/versity/versitygw/s3api"
"github.com/versity/versitygw/s3api/middlewares"
"github.com/versity/versitygw/s3event"
@@ -62,6 +63,7 @@ var (
iamCacheDisable bool
iamCacheTTL int
iamCachePrune int
statsdServers string
)
var (
@@ -398,6 +400,13 @@ func initFlags() []cli.Flag {
EnvVars: []string{"VGW_READ_ONLY"},
Destination: &readonly,
},
&cli.StringFlag{
Name: "metrics-statsd-servers",
Usage: "StatsD server urls comma separated. e.g. 'statsd.example1.com:8125, statsd.example2.com:8125'",
EnvVars: []string{"VGW_METRICS_STATSD_SERVERS"},
Aliases: []string{"mss"},
Destination: &statsdServers,
},
}
}
@@ -508,6 +517,13 @@ func runGateway(ctx context.Context, be backend.Backend) error {
return fmt.Errorf("setup logger: %w", err)
}
metricsManager, err := metrics.NewManager(ctx, metrics.Config{
StatsdServers: statsdServers,
})
if err != nil {
return fmt.Errorf("init metrics manager: %w", err)
}
evSender, err := s3event.InitEventSender(&s3event.EventConfig{
KafkaURL: kafkaURL,
KafkaTopic: kafkaTopic,
@@ -524,7 +540,7 @@ func runGateway(ctx context.Context, be backend.Backend) error {
srv, err := s3api.New(app, be, middlewares.RootUserConfig{
Access: rootUserAccess,
Secret: rootUserSecret,
}, port, region, iam, logger, evSender, opts...)
}, port, region, iam, logger, evSender, metricsManager, opts...)
if err != nil {
return fmt.Errorf("init gateway: %v", err)
}
@@ -587,5 +603,9 @@ Loop:
}
}
if metricsManager != nil {
metricsManager.Close()
}
return saveErr
}
metrics/metrics.go
+43 -8
View File
@@ -18,6 +18,7 @@ import (
"context"
"fmt"
"os"
"strings"
"sync"
)
@@ -46,11 +47,14 @@ type Manager struct {
}
type Config struct {
StatsdServers []string
StatsdServers string
}
// NewManager initializes metrics plugins and returns a new metrics manager
func NewManager(ctx context.Context, conf Config) (*Manager, error) {
if len(conf.StatsdServers) == 0 {
return nil, nil
}
hostname, err := os.Hostname()
if err != nil {
return nil, fmt.Errorf("failed to get hostname: %w", err)
@@ -65,7 +69,9 @@ func NewManager(ctx context.Context, conf Config) (*Manager, error) {
ctx: ctx,
}
for _, server := range conf.StatsdServers {
statsdServers := strings.Split(conf.StatsdServers, ",")
for _, server := range statsdServers {
statsd, err := NewStatsd(server, hostname)
if err != nil {
return nil, err
@@ -81,13 +87,42 @@ func NewManager(ctx context.Context, conf Config) (*Manager, error) {
return mgr, nil
}
func (m *Manager) Send(err error, action string, objSize int64, objCount int64) {
// In case of Authentication failures, url parsing ...
if action == "" {
action = "s3:UnDetected"
}
if err != nil {
m.Increment(action, "failed_count")
}
m.Increment(action, "success_count")
switch action {
case "s3:PutObject":
m.Add(action, "bytes_written", objSize)
m.Increment(action, "object_created_count")
case "s3:CompleteMultipartUpload":
m.Increment(action, "object_created_count")
case "s3:UploadPart":
m.Add(action, "bytes_written", objSize)
case "s3:GetObject":
m.Add(action, "bytes_read", objSize)
case "s3:DeleteObject":
m.Increment(action, "object_removed_count")
case "s3:DeleteObjects":
m.Add(action, "object_removed_count", objCount)
}
//TODO: Handle UploadPartCopy case
//TODO: Handle CopyObject case
}
// Increment increments the key by one
func (m *Manager) Increment(module, key string, tags ...Tag) {
m.Add(module, key, 1, tags...)
}
// Add adds value to key
func (m *Manager) Add(module, key string, value int, tags ...Tag) {
func (m *Manager) Add(module, key string, value int64, tags ...Tag) {
if m.ctx.Err() != nil {
return
}
@@ -107,7 +142,7 @@ func (m *Manager) Add(module, key string, value int, tags ...Tag) {
}
// Gauge sets key to value
func (m *Manager) Gauge(module, key string, value int, tags ...Tag) {
func (m *Manager) Gauge(module, key string, value int64, tags ...Tag) {
if m.ctx.Err() != nil {
return
}
@@ -126,7 +161,7 @@ func (m *Manager) Gauge(module, key string, value int, tags ...Tag) {
}
}
// Close closes metrics channels, waits for data to cmplete, closes all plugins
// Close closes metrics channels, waits for data to complete, closes all plugins
func (m *Manager) Close() {
// drain the datapoint channels
close(m.addDataChan)
@@ -141,8 +176,8 @@ func (m *Manager) Close() {
// publisher is the interface for interacting with the metrics plugins
type publisher interface {
Add(module, key string, value int, tags ...Tag)
Gauge(module, key string, value int, tags ...Tag)
Add(module, key string, value int64, tags ...Tag)
Gauge(module, key string, value int64, tags ...Tag)
Close()
}
@@ -167,6 +202,6 @@ func (m *Manager) gaugeForwarder(gaugeChan <-chan datapoint) {
type datapoint struct {
module string
key string
value int
value int64
tags []Tag
}
metrics/statsd.go
+3 -3
View File
@@ -45,16 +45,16 @@ func (s *Statsd) Close() {
}
// Add adds value to key
func (s *Statsd) Add(module, key string, value int, tags ...Tag) {
func (s *Statsd) Add(module, key string, value int64, tags ...Tag) {
stags := make([]statsd.Tag, len(tags))
for i, t := range tags {
stags[i] = statsd.StringTag(t.Key, t.Value)
}
s.c.Incr(fmt.Sprintf("%v.%v", module, key), int64(value), stags...)
s.c.Incr(fmt.Sprintf("%v.%v", module, key), value, stags...)
}
// Gauge sets key to value
func (s *Statsd) Gauge(module, key string, value int, tags ...Tag) {
func (s *Statsd) Gauge(module, key string, value int64, tags ...Tag) {
stags := make([]statsd.Tag, len(tags))
for i, t := range tags {
stags[i] = statsd.StringTag(t.Key, t.Value)
s3api/admin-server.go
+2 -2
View File
@@ -46,10 +46,10 @@ func NewAdminServer(app *fiber.App, be backend.Backend, root middlewares.RootUse
// Logging middlewares
app.Use(logger.New())
app.Use(middlewares.DecodeURL(nil))
app.Use(middlewares.DecodeURL(nil, nil))
// Authentication middlewares
app.Use(middlewares.VerifyV4Signature(root, iam, nil, region, false))
app.Use(middlewares.VerifyV4Signature(root, iam, nil, nil, region, false))
app.Use(middlewares.VerifyMD5Body(nil))
server.router.Init(app, be, iam)
s3api/controllers/base.go
+329 -163
View File
File diff suppressed because it is too large Load Diff
s3api/controllers/base_test.go
+1 -1
View File
@@ -77,7 +77,7 @@ func TestNew(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := New(tt.args.be, tt.args.iam, nil, nil, false, false)
got := New(tt.args.be, tt.args.iam, nil, nil, nil, false, false)
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("New() = %v, want %v", got, tt.want)
}
s3api/middlewares/authentication.go
+17 -16
View File
@@ -25,6 +25,7 @@ import (
"github.com/gofiber/fiber/v2"
"github.com/versity/versitygw/auth"
"github.com/versity/versitygw/metrics"
"github.com/versity/versitygw/s3api/controllers"
"github.com/versity/versitygw/s3api/utils"
"github.com/versity/versitygw/s3err"
@@ -40,7 +41,7 @@ type RootUserConfig struct {
Secret string
}
func VerifyV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.AuditLogger, region string, debug bool) fiber.Handler {
func VerifyV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.AuditLogger, mm *metrics.Manager, region string, debug bool) fiber.Handler {
acct := accounts{root: root, iam: iam}
return func(ctx *fiber.Ctx) error {
@@ -54,16 +55,16 @@ func VerifyV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.Au
ctx.Locals("startTime", time.Now())
authorization := ctx.Get("Authorization")
if authorization == "" {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrAuthHeaderEmpty), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrAuthHeaderEmpty), logger, mm)
}
authData, err := utils.ParseAuthorization(authorization)
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
if authData.Algorithm != "AWS4-HMAC-SHA256" {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrSignatureVersionNotSupported), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrSignatureVersionNotSupported), logger, mm)
}
if authData.Region != region {
@@ -71,40 +72,40 @@ func VerifyV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.Au
Code: "SignatureDoesNotMatch",
Description: fmt.Sprintf("Credential should be scoped to a valid Region, not %v", authData.Region),
HTTPStatusCode: http.StatusForbidden,
}, logger)
}, logger, mm)
}
ctx.Locals("isRoot", authData.Access == root.Access)
account, err := acct.getAccount(authData.Access)
if err == auth.ErrNoSuchUser {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidAccessKeyID), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidAccessKeyID), logger, mm)
}
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
ctx.Locals("account", account)
// Check X-Amz-Date header
date := ctx.Get("X-Amz-Date")
if date == "" {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrMissingDateHeader), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrMissingDateHeader), logger, mm)
}
// Parse the date and check the date validity
tdate, err := time.Parse(iso8601Format, date)
if err != nil {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrMalformedDate), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrMalformedDate), logger, mm)
}
if date[:8] != authData.Date {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrSignatureDateDoesNotMatch), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrSignatureDateDoesNotMatch), logger, mm)
}
// Validate the dates difference
err = utils.ValidateDate(tdate)
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
if utils.IsBigDataAction(ctx) {
@@ -125,7 +126,7 @@ func VerifyV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.Au
// Compare the calculated hash with the hash provided
if hashPayload != hexPayload {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrContentSHA256Mismatch), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrContentSHA256Mismatch), logger, mm)
}
}
@@ -134,13 +135,13 @@ func VerifyV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.Au
if contentLengthStr != "" {
contentLength, err = strconv.ParseInt(contentLengthStr, 10, 64)
if err != nil {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidRequest), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidRequest), logger, mm)
}
}
err = utils.CheckValidSignature(ctx, authData, account.Secret, hashPayload, tdate, contentLength, debug)
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
return ctx.Next()
@@ -164,6 +165,6 @@ func (a accounts) getAccount(access string) (auth.Account, error) {
return a.iam.GetUserAccount(access)
}
func sendResponse(ctx *fiber.Ctx, err error, logger s3log.AuditLogger) error {
return controllers.SendResponse(ctx, err, &controllers.MetaOpts{Logger: logger})
func sendResponse(ctx *fiber.Ctx, err error, logger s3log.AuditLogger, mm *metrics.Manager) error {
return controllers.SendResponse(ctx, err, &controllers.MetaOpts{Logger: logger, MetricsMng: mm})
}
s3api/middlewares/chunk.go
+4 -3
View File
@@ -20,13 +20,14 @@ import (
"github.com/gofiber/fiber/v2"
"github.com/versity/versitygw/auth"
"github.com/versity/versitygw/metrics"
"github.com/versity/versitygw/s3api/utils"
"github.com/versity/versitygw/s3log"
)
// ProcessChunkedBody initializes the chunked upload stream if the
// request appears to be a chunked upload
func ProcessChunkedBody(root RootUserConfig, iam auth.IAMService, logger s3log.AuditLogger, region string) fiber.Handler {
func ProcessChunkedBody(root RootUserConfig, iam auth.IAMService, logger s3log.AuditLogger, mm *metrics.Manager, region string) fiber.Handler {
return func(ctx *fiber.Ctx) error {
decodedLength := ctx.Get("X-Amz-Decoded-Content-Length")
if decodedLength == "" {
@@ -36,7 +37,7 @@ func ProcessChunkedBody(root RootUserConfig, iam auth.IAMService, logger s3log.A
authData, err := utils.ParseAuthorization(ctx.Get("Authorization"))
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
acct := ctx.Locals("account").(auth.Account)
@@ -51,7 +52,7 @@ func ProcessChunkedBody(root RootUserConfig, iam auth.IAMService, logger s3log.A
return cr
})
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
return ctx.Next()
}
s3api/middlewares/presign-auth.go
+6 -5
View File
@@ -20,12 +20,13 @@ import (
"github.com/gofiber/fiber/v2"
"github.com/versity/versitygw/auth"
"github.com/versity/versitygw/metrics"
"github.com/versity/versitygw/s3api/utils"
"github.com/versity/versitygw/s3err"
"github.com/versity/versitygw/s3log"
)
func VerifyPresignedV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.AuditLogger, region string, debug bool) fiber.Handler {
func VerifyPresignedV4Signature(root RootUserConfig, iam auth.IAMService, logger s3log.AuditLogger, mm *metrics.Manager, region string, debug bool) fiber.Handler {
acct := accounts{root: root, iam: iam}
return func(ctx *fiber.Ctx) error {
@@ -38,16 +39,16 @@ func VerifyPresignedV4Signature(root RootUserConfig, iam auth.IAMService, logger
authData, err := utils.ParsePresignedURIParts(ctx)
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
ctx.Locals("isRoot", authData.Access == root.Access)
account, err := acct.getAccount(authData.Access)
if err == auth.ErrNoSuchUser {
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidAccessKeyID), logger)
return sendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidAccessKeyID), logger, mm)
}
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
ctx.Locals("account", account)
@@ -61,7 +62,7 @@ func VerifyPresignedV4Signature(root RootUserConfig, iam auth.IAMService, logger
err = utils.CheckPresignedSignature(ctx, authData, account.Secret, debug)
if err != nil {
return sendResponse(ctx, err, logger)
return sendResponse(ctx, err, logger, mm)
}
return ctx.Next()
s3api/middlewares/url-decoder.go
+3 -2
View File
@@ -18,17 +18,18 @@ import (
"net/url"
"github.com/gofiber/fiber/v2"
"github.com/versity/versitygw/metrics"
"github.com/versity/versitygw/s3api/controllers"
"github.com/versity/versitygw/s3err"
"github.com/versity/versitygw/s3log"
)
func DecodeURL(logger s3log.AuditLogger) fiber.Handler {
func DecodeURL(logger s3log.AuditLogger, mm *metrics.Manager) fiber.Handler {
return func(ctx *fiber.Ctx) error {
reqURL := ctx.Request().URI().String()
decoded, err := url.Parse(reqURL)
if err != nil {
return controllers.SendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidURI), &controllers.MetaOpts{Logger: logger})
return controllers.SendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidURI), &controllers.MetaOpts{Logger: logger, MetricsMng: mm})
}
ctx.Path(decoded.Path)
return ctx.Next()
s3api/router.go
+3 -2
View File
@@ -18,6 +18,7 @@ import (
"github.com/gofiber/fiber/v2"
"github.com/versity/versitygw/auth"
"github.com/versity/versitygw/backend"
"github.com/versity/versitygw/metrics"
"github.com/versity/versitygw/s3api/controllers"
"github.com/versity/versitygw/s3event"
"github.com/versity/versitygw/s3log"
@@ -27,8 +28,8 @@ type S3ApiRouter struct {
WithAdmSrv bool
}
func (sa *S3ApiRouter) Init(app *fiber.App, be backend.Backend, iam auth.IAMService, logger s3log.AuditLogger, evs s3event.S3EventSender, debug bool, readonly bool) {
s3ApiController := controllers.New(be, iam, logger, evs, debug, readonly)
func (sa *S3ApiRouter) Init(app *fiber.App, be backend.Backend, iam auth.IAMService, logger s3log.AuditLogger, evs s3event.S3EventSender, mm *metrics.Manager, debug bool, readonly bool) {
s3ApiController := controllers.New(be, iam, logger, evs, mm, debug, readonly)
if sa.WithAdmSrv {
adminController := controllers.NewAdminController(iam, be)
s3api/router_test.go
+1 -1
View File
@@ -45,7 +45,7 @@ func TestS3ApiRouter_Init(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
tt.sa.Init(tt.args.app, tt.args.be, tt.args.iam, nil, nil, false, false)
tt.sa.Init(tt.args.app, tt.args.be, tt.args.iam, nil, nil, nil, false, false)
})
}
}
s3api/server.go
+17 -6
View File
@@ -22,6 +22,7 @@ import (
"github.com/gofiber/fiber/v2/middleware/logger"
"github.com/versity/versitygw/auth"
"github.com/versity/versitygw/backend"
"github.com/versity/versitygw/metrics"
"github.com/versity/versitygw/s3api/middlewares"
"github.com/versity/versitygw/s3event"
"github.com/versity/versitygw/s3log"
@@ -39,7 +40,17 @@ type S3ApiServer struct {
health string
}
func New(app *fiber.App, be backend.Backend, root middlewares.RootUserConfig, port, region string, iam auth.IAMService, l s3log.AuditLogger, evs s3event.S3EventSender, opts ...Option) (*S3ApiServer, error) {
func New(
app *fiber.App,
be backend.Backend,
root middlewares.RootUserConfig,
port, region string,
iam auth.IAMService,
l s3log.AuditLogger,
evs s3event.S3EventSender,
mm *metrics.Manager,
opts ...Option,
) (*S3ApiServer, error) {
server := &S3ApiServer{
app: app,
backend: be,
@@ -61,17 +72,17 @@ func New(app *fiber.App, be backend.Backend, root middlewares.RootUserConfig, po
return ctx.SendStatus(http.StatusOK)
})
}
app.Use(middlewares.DecodeURL(l))
app.Use(middlewares.DecodeURL(l, mm))
app.Use(middlewares.RequestLogger(server.debug))
// Authentication middlewares
app.Use(middlewares.VerifyPresignedV4Signature(root, iam, l, region, server.debug))
app.Use(middlewares.VerifyV4Signature(root, iam, l, region, server.debug))
app.Use(middlewares.ProcessChunkedBody(root, iam, l, region))
app.Use(middlewares.VerifyPresignedV4Signature(root, iam, l, mm, region, server.debug))
app.Use(middlewares.VerifyV4Signature(root, iam, l, mm, region, server.debug))
app.Use(middlewares.ProcessChunkedBody(root, iam, l, mm, region))
app.Use(middlewares.VerifyMD5Body(l))
app.Use(middlewares.AclParser(be, l, server.readonly))
server.router.Init(app, be, iam, l, evs, server.debug, server.readonly)
server.router.Init(app, be, iam, l, evs, mm, server.debug, server.readonly)
return server, nil
}
s3api/server_test.go
+1 -1
View File
@@ -64,7 +64,7 @@ func TestNew(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
gotS3ApiServer, err := New(tt.args.app, tt.args.be, tt.args.root,
tt.args.port, "us-east-1", &auth.IAMServiceInternal{}, nil, nil)
tt.args.port, "us-east-1", &auth.IAMServiceInternal{}, nil, nil, nil)
if (err != nil) != tt.wantErr {
t.Errorf("New() error = %v, wantErr %v", err, tt.wantErr)
return