forked from TrueCloudLab/frostfs-http-gw
Merge pull request #6 from nspcc-dev/refactoring_http_gate
Refactoring HTTP Gate
This commit is contained in:
commit
1b0acd4836
2 changed files with 118 additions and 41 deletions
4
app.go
4
app.go
|
@ -175,7 +175,9 @@ func (a *app) Serve(ctx context.Context) {
|
||||||
r.RedirectTrailingSlash = true
|
r.RedirectTrailingSlash = true
|
||||||
|
|
||||||
a.log.Info("enabled /get/{cid}/{oid}")
|
a.log.Info("enabled /get/{cid}/{oid}")
|
||||||
r.GET("/get/{cid}/{oid}", a.receiveFile)
|
r.GET("/get/{cid}/{oid}", a.byAddress)
|
||||||
|
a.log.Info("enabled /get_by_attribute/{cid}/{attr_key}/{attr_val}")
|
||||||
|
r.GET("/get_by_attribute/{cid}/{attr_key}/{attr_val}", a.byAttribute)
|
||||||
|
|
||||||
// attaching /-/(ready,healthy)
|
// attaching /-/(ready,healthy)
|
||||||
attachHealthy(r, a.pool.Status)
|
attachHealthy(r, a.pool.Status)
|
||||||
|
|
155
receive.go
155
receive.go
|
@ -10,6 +10,7 @@ import (
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
sdk "github.com/nspcc-dev/cdn-neofs-sdk"
|
sdk "github.com/nspcc-dev/cdn-neofs-sdk"
|
||||||
|
"github.com/nspcc-dev/neofs-api-go/pkg/container"
|
||||||
"github.com/nspcc-dev/neofs-api-go/pkg/object"
|
"github.com/nspcc-dev/neofs-api-go/pkg/object"
|
||||||
"github.com/pkg/errors"
|
"github.com/pkg/errors"
|
||||||
"github.com/valyala/fasthttp"
|
"github.com/valyala/fasthttp"
|
||||||
|
@ -18,12 +19,23 @@ import (
|
||||||
"google.golang.org/grpc/status"
|
"google.golang.org/grpc/status"
|
||||||
)
|
)
|
||||||
|
|
||||||
type detector struct {
|
type (
|
||||||
io.Writer
|
detector struct {
|
||||||
sync.Once
|
io.Writer
|
||||||
|
sync.Once
|
||||||
|
|
||||||
contentType string
|
contentType string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
request struct {
|
||||||
|
*fasthttp.RequestCtx
|
||||||
|
|
||||||
|
log *zap.Logger
|
||||||
|
obj sdk.ObjectClient
|
||||||
|
}
|
||||||
|
|
||||||
|
objectIDs []*object.ID
|
||||||
|
)
|
||||||
|
|
||||||
func newDetector(w io.Writer) *detector {
|
func newDetector(w io.Writer) *detector {
|
||||||
return &detector{Writer: w}
|
return &detector{Writer: w}
|
||||||
|
@ -37,34 +49,18 @@ func (d *detector) Write(data []byte) (int, error) {
|
||||||
return d.Writer.Write(data)
|
return d.Writer.Write(data)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *app) receiveFile(c *fasthttp.RequestCtx) {
|
func (r *request) receiveFile(address *object.Address) {
|
||||||
var (
|
var (
|
||||||
err error
|
err error
|
||||||
disp = "inline"
|
dis = "inline"
|
||||||
start = time.Now()
|
start = time.Now()
|
||||||
address = object.NewAddress()
|
|
||||||
sCID, _ = c.UserValue("cid").(string)
|
|
||||||
sOID, _ = c.UserValue("oid").(string)
|
|
||||||
value = strings.Join([]string{sCID, sOID}, "/")
|
|
||||||
|
|
||||||
filename string
|
filename string
|
||||||
)
|
)
|
||||||
|
|
||||||
log := a.log.With(
|
writer := newDetector(r.Response.BodyWriter())
|
||||||
// zap.String("node", con.Target()),
|
obj, err := r.obj.Get(r, address, sdk.WithGetWriter(writer))
|
||||||
zap.String("cid", sCID),
|
|
||||||
zap.String("oid", sOID))
|
|
||||||
|
|
||||||
if err = address.Parse(value); err != nil {
|
|
||||||
log.Error("wrong object address", zap.Error(err))
|
|
||||||
c.Error("wrong object address", fasthttp.StatusBadRequest)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
writer := newDetector(c.Response.BodyWriter())
|
|
||||||
obj, err := a.cli.Object().Get(c, address, sdk.WithGetWriter(writer))
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("could not receive object",
|
r.log.Error("could not receive object",
|
||||||
zap.Stringer("elapsed", time.Since(start)),
|
zap.Stringer("elapsed", time.Since(start)),
|
||||||
zap.Error(err))
|
zap.Error(err))
|
||||||
|
|
||||||
|
@ -81,24 +77,24 @@ func (a *app) receiveFile(c *fasthttp.RequestCtx) {
|
||||||
msg = st.Message()
|
msg = st.Message()
|
||||||
}
|
}
|
||||||
|
|
||||||
c.Error(msg, code)
|
r.Error(msg, code)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if c.Request.URI().QueryArgs().GetBool("download") {
|
if r.Request.URI().QueryArgs().GetBool("download") {
|
||||||
disp = "attachment"
|
dis = "attachment"
|
||||||
}
|
}
|
||||||
|
|
||||||
c.Response.Header.Set("Content-Length", strconv.FormatUint(obj.PayloadSize(), 10))
|
r.Response.Header.Set("Content-Length", strconv.FormatUint(obj.PayloadSize(), 10))
|
||||||
c.Response.Header.Set("x-object-id", obj.ID().String())
|
r.Response.Header.Set("x-object-id", obj.ID().String())
|
||||||
c.Response.Header.Set("x-owner-id", obj.OwnerID().String())
|
r.Response.Header.Set("x-owner-id", obj.OwnerID().String())
|
||||||
c.Response.Header.Set("x-container-id", obj.ContainerID().String())
|
r.Response.Header.Set("x-container-id", obj.ContainerID().String())
|
||||||
|
|
||||||
for _, attr := range obj.Attributes() {
|
for _, attr := range obj.Attributes() {
|
||||||
key := attr.Key()
|
key := attr.Key()
|
||||||
val := attr.Value()
|
val := attr.Value()
|
||||||
|
|
||||||
c.Response.Header.Set("x-"+key, val)
|
r.Response.Header.Set("x-"+key, val)
|
||||||
|
|
||||||
switch key {
|
switch key {
|
||||||
case object.AttributeFileName:
|
case object.AttributeFileName:
|
||||||
|
@ -106,19 +102,98 @@ func (a *app) receiveFile(c *fasthttp.RequestCtx) {
|
||||||
case object.AttributeTimestamp:
|
case object.AttributeTimestamp:
|
||||||
value, err := strconv.ParseInt(val, 10, 64)
|
value, err := strconv.ParseInt(val, 10, 64)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
a.log.Info("couldn't parse creation date",
|
r.log.Info("couldn't parse creation date",
|
||||||
zap.String("key", key),
|
zap.String("key", key),
|
||||||
zap.String("val", val),
|
zap.String("val", val),
|
||||||
zap.Error(err))
|
zap.Error(err))
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
c.Response.Header.Set("Last-Modified",
|
r.Response.Header.Set("Last-Modified",
|
||||||
time.Unix(value, 0).Format(time.RFC1123))
|
time.Unix(value, 0).Format(time.RFC1123))
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
c.SetContentType(writer.contentType)
|
r.SetContentType(writer.contentType)
|
||||||
c.Response.Header.Set("Content-Disposition", disp+"; filename="+path.Base(filename))
|
r.Response.Header.Set("Content-Disposition", dis+"; filename="+path.Base(filename))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (o objectIDs) Slice() []string {
|
||||||
|
res := make([]string, 0, len(o))
|
||||||
|
for _, oid := range o {
|
||||||
|
res = append(res, oid.String())
|
||||||
|
}
|
||||||
|
|
||||||
|
return res
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *app) request(ctx *fasthttp.RequestCtx, log *zap.Logger) *request {
|
||||||
|
return &request{
|
||||||
|
RequestCtx: ctx,
|
||||||
|
|
||||||
|
log: log,
|
||||||
|
obj: a.cli.Object(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *app) byAddress(c *fasthttp.RequestCtx) {
|
||||||
|
var (
|
||||||
|
err error
|
||||||
|
adr = object.NewAddress()
|
||||||
|
cid, _ = c.UserValue("cid").(string)
|
||||||
|
oid, _ = c.UserValue("oid").(string)
|
||||||
|
val = strings.Join([]string{cid, oid}, "/")
|
||||||
|
log = a.log.With(
|
||||||
|
zap.String("cid", cid),
|
||||||
|
zap.String("oid", oid))
|
||||||
|
)
|
||||||
|
|
||||||
|
if err = adr.Parse(val); err != nil {
|
||||||
|
log.Error("wrong object address", zap.Error(err))
|
||||||
|
c.Error("wrong object address", fasthttp.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
a.request(c, log).receiveFile(adr)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *app) byAttribute(c *fasthttp.RequestCtx) {
|
||||||
|
var (
|
||||||
|
err error
|
||||||
|
ids []*object.ID
|
||||||
|
cid = container.NewID()
|
||||||
|
adr = object.NewAddress()
|
||||||
|
sCID, _ = c.UserValue("cid").(string)
|
||||||
|
key, _ = c.UserValue("attr_key").(string)
|
||||||
|
val, _ = c.UserValue("attr_val").(string)
|
||||||
|
|
||||||
|
log = a.log.With(
|
||||||
|
zap.String("cid", sCID),
|
||||||
|
zap.String("attr_key", key),
|
||||||
|
zap.String("attr_val", val))
|
||||||
|
)
|
||||||
|
|
||||||
|
if err = cid.Parse(sCID); err != nil {
|
||||||
|
log.Error("wrong container id", zap.Error(err))
|
||||||
|
c.Error("wrong container id", fasthttp.StatusBadRequest)
|
||||||
|
return
|
||||||
|
} else if ids, err = a.cli.Object().Search(c, cid, sdk.SearchRootObjects(), sdk.SearchByAttribute(key, val)); err != nil {
|
||||||
|
log.Error("something went wrong", zap.Error(err))
|
||||||
|
c.Error("something went wrong", fasthttp.StatusBadRequest)
|
||||||
|
return
|
||||||
|
} else if len(ids) == 0 {
|
||||||
|
c.Error("not found", fasthttp.StatusNotFound)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(ids) > 1 {
|
||||||
|
log.Debug("found multiple objects",
|
||||||
|
zap.Strings("object_ids", objectIDs(ids).Slice()),
|
||||||
|
zap.Stringer("show_object_id", ids[0]))
|
||||||
|
}
|
||||||
|
|
||||||
|
adr.SetContainerID(cid)
|
||||||
|
adr.SetObjectID(ids[0])
|
||||||
|
a.request(c, log).receiveFile(adr)
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue