forked from TrueCloudLab/frostfs-http-gw
257 lines
7.5 KiB
Go
257 lines
7.5 KiB
Go
package handler
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/url"
|
|
"strings"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/cache"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/handler/middleware"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/resolver"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/response"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/tree"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
|
|
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
|
|
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/pool"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
|
|
"github.com/valyala/fasthttp"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type Config interface {
|
|
DefaultTimestamp() bool
|
|
ZipCompression() bool
|
|
ClientCut() bool
|
|
BufferMaxSizeForPut() uint64
|
|
NamespaceHeader() string
|
|
}
|
|
|
|
type Handler struct {
|
|
log *zap.Logger
|
|
pool *pool.Pool
|
|
ownerID *user.ID
|
|
config Config
|
|
containerResolver *resolver.ContainerResolver
|
|
tree *tree.Tree
|
|
cache *cache.BucketCache
|
|
}
|
|
|
|
func New(params *utils.AppParams, config Config, tree *tree.Tree) *Handler {
|
|
return &Handler{
|
|
log: params.Logger,
|
|
pool: params.Pool,
|
|
ownerID: params.Owner,
|
|
config: config,
|
|
containerResolver: params.Resolver,
|
|
tree: tree,
|
|
cache: params.Cache,
|
|
}
|
|
}
|
|
|
|
// byAddress is a wrapper for function (e.g. request.headObject, request.receiveFile) that
|
|
// prepares request and object address to it.
|
|
func (h *Handler) byAddress(c *fasthttp.RequestCtx, f func(context.Context, request, oid.Address)) {
|
|
var (
|
|
idCnr, _ = c.UserValue("cid").(string)
|
|
idObj, _ = c.UserValue("oid").(string)
|
|
log = h.log.With(zap.String("cid", idCnr), zap.String("oid", idObj))
|
|
)
|
|
|
|
ctx := utils.GetContextFromRequest(c)
|
|
|
|
bktInfo, err := h.getBucketInfo(ctx, idCnr, log)
|
|
if err != nil {
|
|
logAndSendBucketError(c, log, err)
|
|
return
|
|
}
|
|
|
|
objID := new(oid.ID)
|
|
if err = objID.DecodeString(idObj); err != nil {
|
|
log.Error(logs.WrongObjectID, zap.Error(err))
|
|
response.Error(c, "wrong object id", fasthttp.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
var addr oid.Address
|
|
addr.SetContainer(bktInfo.CID)
|
|
addr.SetObject(*objID)
|
|
|
|
f(ctx, *h.newRequest(c, log), addr)
|
|
}
|
|
|
|
// byObjectName is a wrapper for function (e.g. request.headObject, request.receiveFile) that
|
|
// prepares request and object address to it.
|
|
func (h *Handler) byObjectName(req *fasthttp.RequestCtx, f func(context.Context, request, oid.Address)) {
|
|
var (
|
|
bucketname = req.UserValue("cid").(string)
|
|
key = req.UserValue("oid").(string)
|
|
log = h.log.With(zap.String("bucketname", bucketname), zap.String("key", key))
|
|
)
|
|
|
|
ctx := utils.GetContextFromRequest(req)
|
|
|
|
bktInfo, err := h.getBucketInfo(ctx, bucketname, log)
|
|
if err != nil {
|
|
logAndSendBucketError(req, log, err)
|
|
return
|
|
}
|
|
|
|
foundOid, err := h.tree.GetLatestVersion(ctx, &bktInfo.CID, key)
|
|
if err != nil {
|
|
if errors.Is(err, tree.ErrNodeAccessDenied) {
|
|
response.Error(req, "Access Denied", fasthttp.StatusForbidden)
|
|
return
|
|
}
|
|
log.Error(logs.GetLatestObjectVersion, zap.Error(err))
|
|
|
|
response.Error(req, "object wasn't found", fasthttp.StatusNotFound)
|
|
return
|
|
}
|
|
if foundOid.DeleteMarker {
|
|
log.Error(logs.ObjectWasDeleted)
|
|
response.Error(req, "object deleted", fasthttp.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
var addr oid.Address
|
|
addr.SetContainer(bktInfo.CID)
|
|
addr.SetObject(foundOid.OID)
|
|
|
|
f(ctx, *h.newRequest(req, log), addr)
|
|
}
|
|
|
|
// byAttribute is a wrapper similar to byAddress.
|
|
func (h *Handler) byAttribute(c *fasthttp.RequestCtx, f func(context.Context, request, oid.Address)) {
|
|
scid, _ := c.UserValue("cid").(string)
|
|
key, _ := c.UserValue("attr_key").(string)
|
|
val, _ := c.UserValue("attr_val").(string)
|
|
|
|
key, err := url.QueryUnescape(key)
|
|
if err != nil {
|
|
h.log.Error(logs.FailedToUnescapeQuery, zap.String("cid", scid), zap.String("attr_key", key), zap.Uint64("id", c.ID()), zap.Error(err))
|
|
response.Error(c, "could not unescape attr_key: "+err.Error(), fasthttp.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
val, err = url.QueryUnescape(val)
|
|
if err != nil {
|
|
h.log.Error(logs.FailedToUnescapeQuery, zap.String("cid", scid), zap.String("attr_val", val), zap.Uint64("id", c.ID()), zap.Error(err))
|
|
response.Error(c, "could not unescape attr_val: "+err.Error(), fasthttp.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
log := h.log.With(zap.String("cid", scid), zap.String("attr_key", key), zap.String("attr_val", val))
|
|
|
|
ctx := utils.GetContextFromRequest(c)
|
|
|
|
bktInfo, err := h.getBucketInfo(ctx, scid, log)
|
|
if err != nil {
|
|
logAndSendBucketError(c, log, err)
|
|
return
|
|
}
|
|
|
|
res, err := h.search(ctx, &bktInfo.CID, key, val, object.MatchStringEqual)
|
|
if err != nil {
|
|
log.Error(logs.CouldNotSearchForObjects, zap.Error(err))
|
|
response.Error(c, "could not search for objects: "+err.Error(), fasthttp.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
defer res.Close()
|
|
|
|
buf := make([]oid.ID, 1)
|
|
|
|
n, err := res.Read(buf)
|
|
if n == 0 {
|
|
if errors.Is(err, io.EOF) {
|
|
log.Error(logs.ObjectNotFound, zap.Error(err))
|
|
response.Error(c, "object not found", fasthttp.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
log.Error(logs.ReadObjectListFailed, zap.Error(err))
|
|
response.Error(c, "read object list failed: "+err.Error(), fasthttp.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
var addrObj oid.Address
|
|
addrObj.SetContainer(bktInfo.CID)
|
|
addrObj.SetObject(buf[0])
|
|
|
|
f(ctx, *h.newRequest(c, log), addrObj)
|
|
}
|
|
|
|
// resolveContainer decode container id, if it's not a valid container id
|
|
// then trey to resolve name using provided resolver.
|
|
func (h *Handler) resolveContainer(ctx context.Context, containerID string) (*cid.ID, error) {
|
|
cnrID := new(cid.ID)
|
|
err := cnrID.DecodeString(containerID)
|
|
if err != nil {
|
|
cnrID, err = h.containerResolver.Resolve(ctx, containerID)
|
|
if err != nil && strings.Contains(err.Error(), "not found") {
|
|
err = fmt.Errorf("%w: %s", new(apistatus.ContainerNotFound), err.Error())
|
|
}
|
|
}
|
|
return cnrID, err
|
|
}
|
|
|
|
func (h *Handler) getBucketInfo(ctx context.Context, containerName string, log *zap.Logger) (*data.BucketInfo, error) {
|
|
ns, err := middleware.GetNamespace(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if bktInfo := h.cache.Get(ns, containerName); bktInfo != nil {
|
|
return bktInfo, nil
|
|
}
|
|
|
|
cnrID, err := h.resolveContainer(ctx, containerName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
bktInfo, err := h.readContainer(ctx, *cnrID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err = h.cache.Put(bktInfo); err != nil {
|
|
log.Warn(logs.CouldntPutBucketIntoCache,
|
|
zap.String("bucket name", bktInfo.Name),
|
|
zap.Stringer("bucket cid", bktInfo.CID),
|
|
zap.Error(err))
|
|
}
|
|
|
|
return bktInfo, nil
|
|
}
|
|
|
|
func (h *Handler) readContainer(ctx context.Context, cnrID cid.ID) (*data.BucketInfo, error) {
|
|
prm := pool.PrmContainerGet{ContainerID: cnrID}
|
|
res, err := h.pool.GetContainer(ctx, prm)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("get frostfs container '%s': %w", cnrID.String(), err)
|
|
}
|
|
|
|
bktInfo := &data.BucketInfo{
|
|
CID: cnrID,
|
|
Name: cnrID.EncodeToString(),
|
|
}
|
|
|
|
if domain := container.ReadDomain(res); domain.Name() != "" {
|
|
bktInfo.Name = domain.Name()
|
|
bktInfo.Zone = domain.Zone()
|
|
}
|
|
|
|
bktInfo.HomomorphicHashDisabled = container.IsHomomorphicHashingDisabled(res)
|
|
|
|
return bktInfo, err
|
|
}
|