mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
add jwt for writes
This commit is contained in:
parent
7103c1ab7e
commit
157c0f7c01
|
@ -4,13 +4,14 @@ import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||||
"github.com/chrislusf/seaweedfs/weed/operation"
|
"github.com/chrislusf/seaweedfs/weed/operation"
|
||||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||||
"sync"
|
"github.com/chrislusf/seaweedfs/weed/security"
|
||||||
)
|
)
|
||||||
|
|
||||||
type ContinuousDirtyPages struct {
|
type ContinuousDirtyPages struct {
|
||||||
|
@ -164,6 +165,7 @@ func (pages *ContinuousDirtyPages) saveExistingPagesToStorage(ctx context.Contex
|
||||||
func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte, offset int64) (*filer_pb.FileChunk, error) {
|
func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte, offset int64) (*filer_pb.FileChunk, error) {
|
||||||
|
|
||||||
var fileId, host string
|
var fileId, host string
|
||||||
|
var auth security.EncodedJwt
|
||||||
|
|
||||||
if err := pages.f.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
if err := pages.f.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
||||||
|
|
||||||
|
@ -181,7 +183,7 @@ func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
fileId, host = resp.FileId, resp.Url
|
fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
|
@ -190,7 +192,7 @@ func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte
|
||||||
|
|
||||||
fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
|
fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
|
||||||
bufReader := bytes.NewReader(buf)
|
bufReader := bytes.NewReader(buf)
|
||||||
uploadResult, err := operation.Upload(fileUrl, pages.f.Name, bufReader, false, "application/octet-stream", nil, "")
|
uploadResult, err := operation.Upload(fileUrl, pages.f.Name, bufReader, false, "application/octet-stream", nil, auth)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
glog.V(0).Infof("upload data %v to %s: %v", pages.f.Name, fileUrl, err)
|
glog.V(0).Infof("upload data %v to %s: %v", pages.f.Name, fileUrl, err)
|
||||||
return nil, fmt.Errorf("upload data: %v", err)
|
return nil, fmt.Errorf("upload data: %v", err)
|
||||||
|
|
|
@ -9,6 +9,7 @@ import (
|
||||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||||
"github.com/chrislusf/seaweedfs/weed/operation"
|
"github.com/chrislusf/seaweedfs/weed/operation"
|
||||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||||
|
"github.com/chrislusf/seaweedfs/weed/security"
|
||||||
"github.com/chrislusf/seaweedfs/weed/util"
|
"github.com/chrislusf/seaweedfs/weed/util"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -59,6 +60,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk) (fileId stri
|
||||||
defer readCloser.Close()
|
defer readCloser.Close()
|
||||||
|
|
||||||
var host string
|
var host string
|
||||||
|
var auth security.EncodedJwt
|
||||||
|
|
||||||
if err := fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
if err := fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
||||||
|
|
||||||
|
@ -76,7 +78,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk) (fileId stri
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
fileId, host = resp.FileId, resp.Url
|
fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
|
@ -88,7 +90,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk) (fileId stri
|
||||||
glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)
|
glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)
|
||||||
|
|
||||||
uploadResult, err := operation.Upload(fileUrl, filename, readCloser,
|
uploadResult, err := operation.Upload(fileUrl, filename, readCloser,
|
||||||
"gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, "")
|
"gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, auth)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
glog.V(0).Infof("upload data %v to %s: %v", filename, fileUrl, err)
|
glog.V(0).Infof("upload data %v to %s: %v", filename, fileUrl, err)
|
||||||
return "", fmt.Errorf("upload data: %v", err)
|
return "", fmt.Errorf("upload data: %v", err)
|
||||||
|
|
Loading…
Reference in a new issue