alexvanin/frostfs-http-gw
1
0
Fork 0
forked from TrueCloudLab/frostfs-http-gw
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#76] Added HEAD methods

Browse source 
Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2021-07-09 16:28:39 +03:00
parent 2b8aa4914e
commit d856fdf4c4
2 changed files with 154 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app.go
View file

@ -200,8 +200,10 @@ func (a *app) Serve(ctx context.Context) {
r.POST("/upload/{cid}", a.logger(uploader.Upload))
a.log.Info("added path /upload/{cid}")
r.GET("/get/{cid}/{oid}", a.logger(downloader.DownloadByAddress))
r.HEAD("/get/{cid}/{oid}", a.logger(downloader.HeadByAddress))
a.log.Info("added path /get/{cid}/{oid}")
r.GET("/get_by_attribute/{cid}/{attr_key}/{attr_val:*}", a.logger(downloader.DownloadByAttribute))
r.HEAD("/get_by_attribute/{cid}/{attr_key}/{attr_val:*}", a.logger(downloader.HeadByAttribute))
a.log.Info("added path /get_by_attribute/{cid}/{attr_key}/{attr_val:*}")
// enable metrics
if a.cfg.GetBool(cmdMetrics) {

199
downloader/download.go
View file

@ -45,6 +45,10 @@ type (
}
)
var errObjectNotFound = errors.New("object not found")
const sizeToDetectType = 512
func newReader(data []byte, err error) *errReader {
return &errReader{data: data, err: err}
}
@ -112,7 +116,7 @@ func isValidValue(s string) bool {
return true
}
func (r *request) receiveFile(clnt client.Object, objectAddress *object.Address) {
func (r request) receiveFile(clnt client.Object, objectAddress *object.Address) {
var (
err error
dis = "inline"
@ -133,31 +137,9 @@ func (r *request) receiveFile(clnt client.Object, objectAddress *object.Address)
readDetector.Detect()
})
obj, err = clnt.GetObject(
r.RequestCtx,
options,
)
obj, err = clnt.GetObject(r.RequestCtx, options, bearerOpts(r.RequestCtx))
if err != nil {
r.log.Error(
"could not receive object",
zap.Stringer("elapsed", time.Since(start)),
zap.Error(err),
)
var (
msg = fmt.Sprintf("could not receive object: %v", err)
code = fasthttp.StatusBadRequest
cause = err
)
for unwrap := errors.Unwrap(err); unwrap != nil; unwrap = errors.Unwrap(cause) {
cause = unwrap
}
if st, ok := status.FromError(cause); ok && st != nil {
if st.Code() == codes.NotFound {
code = fasthttp.StatusNotFound
}
msg = st.Message()
}
r.Error(msg, code)
r.handleNeoFSErr(err, start)
return
}
if r.Request.URI().QueryArgs().GetBool("download") {
@ -209,6 +191,99 @@ func (r *request) receiveFile(clnt client.Object, objectAddress *object.Address)
r.Response.Header.Set("Content-Disposition", dis+"; filename="+path.Base(filename))
}
func bearerOpts(ctx context.Context) client.CallOption {
if tkn, err := tokens.LoadBearerToken(ctx); err == nil {
return client.WithBearer(tkn)
}
return client.WithBearer(nil)
}
func (r request) headObject(clnt client.Object, objectAddress *object.Address) {
var start = time.Now()
if err := tokens.StoreBearerToken(r.RequestCtx); err != nil {
r.log.Error("could not fetch and store bearer token", zap.Error(err))
r.Error("could not fetch and store bearer token", fasthttp.StatusBadRequest)
return
}
options := new(client.ObjectHeaderParams).WithAddress(objectAddress)
bearerOpt := bearerOpts(r.RequestCtx)
obj, err := clnt.GetObjectHeader(r.RequestCtx, options, bearerOpt)
if err != nil {
r.handleNeoFSErr(err, start)
return
}
r.Response.Header.Set("Content-Length", strconv.FormatUint(obj.PayloadSize(), 10))
var contentType string
for _, attr := range obj.Attributes() {
key := attr.Key()
val := attr.Value()
if !isValidToken(key) || !isValidValue(val) {
continue
}
r.Response.Header.Set("X-Attribute-"+key, val)
switch key {
case object.AttributeTimestamp:
value, err := strconv.ParseInt(val, 10, 64)
if err != nil {
r.log.Info("couldn't parse creation date",
zap.String("key", key),
zap.String("val", val),
zap.Error(err))
continue
}
r.Response.Header.Set("Last-Modified", time.Unix(value, 0).UTC().Format(http.TimeFormat))
case object.AttributeContentType:
contentType = val
}
}
r.Response.Header.Set("x-object-id", obj.ID().String())
r.Response.Header.Set("x-owner-id", obj.OwnerID().String())
r.Response.Header.Set("x-container-id", obj.ContainerID().String())
if len(contentType) == 0 {
objRange := object.NewRange()
objRange.SetOffset(0)
if sizeToDetectType < obj.PayloadSize() {
objRange.SetLength(sizeToDetectType)
} else {
objRange.SetLength(obj.PayloadSize())
}
ops := new(client.RangeDataParams).WithAddress(objectAddress).WithRange(objRange)
data, err := clnt.ObjectPayloadRangeData(r.RequestCtx, ops, bearerOpt)
if err != nil {
r.handleNeoFSErr(err, start)
return
}
contentType = http.DetectContentType(data)
}
r.SetContentType(contentType)
}
func (r *request) handleNeoFSErr(err error, start time.Time) {
r.log.Error(
"could not receive object",
zap.Stringer("elapsed", time.Since(start)),
zap.Error(err),
)
var (
msg = fmt.Sprintf("could not receive object: %v", err)
code = fasthttp.StatusBadRequest
cause = err
)
for unwrap := errors.Unwrap(err); unwrap != nil; unwrap = errors.Unwrap(cause) {
cause = unwrap
}
if st, ok := status.FromError(cause); ok && st != nil {
if st.Code() == codes.NotFound {
code = fasthttp.StatusNotFound
}
msg = st.Message()
}
r.Error(msg, code)
}
func (o objectIDs) Slice() []string {
res := make([]string, 0, len(o))
for _, oid := range o {
@ -242,53 +317,84 @@ func (d *Downloader) newRequest(ctx *fasthttp.RequestCtx, log *zap.Logger) *requ
// DownloadByAddress handles download requests using simple cid/oid format.
func (d *Downloader) DownloadByAddress(c *fasthttp.RequestCtx) {
d.byAddress(c, request.receiveFile)
}
// HeadByAddress handles head requests using simple cid/oid format.
func (d *Downloader) HeadByAddress(c *fasthttp.RequestCtx) {
d.byAddress(c, request.headObject)
}
// byAddress is wrapper for function (e.g. request.headObject, request.receiveFile) that
// prepares request and object address to it.
func (d *Downloader) byAddress(c *fasthttp.RequestCtx, f func(request, client.Object, *object.Address)) {
var (
err error
address = object.NewAddress()
cid, _ = c.UserValue("cid").(string)
oid, _ = c.UserValue("oid").(string)
val = strings.Join([]string{cid, oid}, "/")
log = d.log.With(zap.String("cid", cid), zap.String("oid", oid))
)
if err = address.Parse(val); err != nil {
if err := address.Parse(val); err != nil {
log.Error("wrong object address", zap.Error(err))
c.Error("wrong object address", fasthttp.StatusBadRequest)
return
}
d.newRequest(c, log).receiveFile(d.pool, address)
f(*d.newRequest(c, log), d.pool, address)
}
// DownloadByAttribute handles attribute-based download requests.
func (d *Downloader) DownloadByAttribute(c *fasthttp.RequestCtx) {
d.byAttribute(c, request.receiveFile)
}
// HeadByAttribute handles attribute-based head requests.
func (d *Downloader) HeadByAttribute(c *fasthttp.RequestCtx) {
d.byAttribute(c, request.headObject)
}
// byAttribute is wrapper similar to byAddress.
func (d *Downloader) byAttribute(c *fasthttp.RequestCtx, f func(request, client.Object, *object.Address)) {
var (
err error
scid, _ = c.UserValue("cid").(string)
key, _ = c.UserValue("attr_key").(string)
val, _ = c.UserValue("attr_val").(string)
log = d.log.With(zap.String("cid", scid), zap.String("attr_key", key), zap.String("attr_val", val))
ids []*object.ID
httpStatus = fasthttp.StatusBadRequest
scid, _ = c.UserValue("cid").(string)
key, _ = c.UserValue("attr_key").(string)
val, _ = c.UserValue("attr_val").(string)
log = d.log.With(zap.String("cid", scid), zap.String("attr_key", key), zap.String("attr_val", val))
)
cid := cid.New()
if err = cid.Parse(scid); err != nil {
containerID := cid.New()
if err := containerID.Parse(scid); err != nil {
log.Error("wrong container id", zap.Error(err))
c.Error("wrong container id", fasthttp.StatusBadRequest)
c.Error("wrong container id", httpStatus)
return
}
address, err := d.searchObject(c, log, containerID, key, val)
if err != nil {
log.Error("couldn't search object", zap.Error(err))
if errors.Is(err, errObjectNotFound) {
httpStatus = fasthttp.StatusNotFound
}
c.Error("couldn't search object", httpStatus)
return
}
f(*d.newRequest(c, log), d.pool, address)
}
func (d *Downloader) searchObject(c *fasthttp.RequestCtx, log *zap.Logger, cid *cid.ID, key, val string) (*object.Address, error) {
options := object.NewSearchFilters()
options.AddRootFilter()
options.AddFilter(key, val, object.MatchStringEqual)
sops := new(client.SearchObjectParams).WithContainerID(cid).WithSearchFilters(options)
if ids, err = d.pool.SearchObject(c, sops); err != nil {
log.Error("something went wrong", zap.Error(err))
c.Error("something went wrong", fasthttp.StatusBadRequest)
return
} else if len(ids) == 0 {
log.Debug("object not found")
c.Error("object not found", fasthttp.StatusNotFound)
return
ids, err := d.pool.SearchObject(c, sops)
if err != nil {
return nil, err
}
if len(ids) == 0 {
return nil, errObjectNotFound
}
if len(ids) > 1 {
log.Debug("found multiple objects",
@ -298,6 +404,5 @@ func (d *Downloader) DownloadByAttribute(c *fasthttp.RequestCtx) {
address := object.NewAddress()
address.SetContainerID(cid)
address.SetObjectID(ids[0])
d.newRequest(c, log).receiveFile(d.pool, address)
return address, nil
}