object/ceph: fix ceph range get and get no-exist (#3855)

pkg/object/ceph.go

@@ -98,6 +98,9 @@ type cephReader struct {
 }
 
 func (r *cephReader) Read(buf []byte) (n int, err error) {
+ if r.limit == 0 {
+ return 0, io.EOF
+ }
  if r.limit > 0 && int64(len(buf)) > r.limit {
  buf = buf[:r.limit]
  }
@@ -121,6 +124,9 @@ func (r *cephReader) Close() error {
 }
 
 func (c *ceph) Get(key string, off, limit int64) (io.ReadCloser, error) {
+ if _, err := c.Head(key); err != nil {
+ return nil, err
+ }
  ctx, err := c.newContext()
  if err != nil {
  return nil, err
@@ -135,14 +141,18 @@ var cephPool = sync.Pool{
 }
 
 func (c *ceph) Put(key string, in io.Reader) error {
+ // ceph default osd_max_object_size = 128M
  return c.do(func(ctx *rados.IOContext) error {
  if b, ok := in.(*bytes.Reader); ok {
  v := reflect.ValueOf(b)
  data := v.Elem().Field(0).Bytes()
  if len(data) == 0 {
  return errors.New("ceph: can't put empty file")
  }
- return ctx.WriteFull(key, data)
+ // If the data exceeds 90M, ceph will report an error: 'rados: ret=-90, Message too long'
+ if len(data) < 85<<20 {
+ return ctx.WriteFull(key, data)
+ }
  }
  buf := cephPool.Get().([]byte)
  defer cephPool.Put(buf)