diff --git a/app.go b/app.go index 2e53a13..52991ff 100644 --- a/app.go +++ b/app.go @@ -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) { diff --git a/downloader/download.go b/downloader/download.go index 1272777..0b77969 100644 --- a/downloader/download.go +++ b/downloader/download.go @@ -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 }