aboutsummaryrefslogtreecommitdiff
path: root/weed/command
diff options
context:
space:
mode:
Diffstat (limited to 'weed/command')
-rw-r--r--weed/command/backup.go11
-rw-r--r--weed/command/benchmark.go14
-rw-r--r--weed/command/filer.go6
-rw-r--r--weed/command/filer_copy.go49
-rw-r--r--weed/command/filer_replication.go1
-rw-r--r--weed/command/master.go7
-rw-r--r--weed/command/mount_std.go7
-rw-r--r--weed/command/s3.go6
-rw-r--r--weed/command/scaffold.go26
-rw-r--r--weed/command/server.go8
-rw-r--r--weed/command/upload.go11
-rw-r--r--weed/command/volume.go7
12 files changed, 122 insertions, 31 deletions
diff --git a/weed/command/backup.go b/weed/command/backup.go
index 0641f2e5d..86391f9c4 100644
--- a/weed/command/backup.go
+++ b/weed/command/backup.go
@@ -2,6 +2,9 @@ package command
import (
"fmt"
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/chrislusf/seaweedfs/weed/server"
+ "github.com/spf13/viper"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/storage"
@@ -46,6 +49,10 @@ var cmdBackup = &Command{
}
func runBackup(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+ grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "client")
+
if *s.volumeId == -1 {
return false
}
@@ -59,7 +66,7 @@ func runBackup(cmd *Command, args []string) bool {
}
volumeServer := lookup.Locations[0].Url
- stats, err := operation.GetVolumeSyncStatus(volumeServer, uint32(vid))
+ stats, err := operation.GetVolumeSyncStatus(volumeServer, grpcDialOption, uint32(vid))
if err != nil {
fmt.Printf("Error get volume %d status: %v\n", vid, err)
return true
@@ -81,7 +88,7 @@ func runBackup(cmd *Command, args []string) bool {
return true
}
- if err := v.Synchronize(volumeServer); err != nil {
+ if err := v.Synchronize(volumeServer, grpcDialOption); err != nil {
fmt.Printf("Error synchronizing volume %d: %v\n", vid, err)
return true
}
diff --git a/weed/command/benchmark.go b/weed/command/benchmark.go
index 93359b243..44601e567 100644
--- a/weed/command/benchmark.go
+++ b/weed/command/benchmark.go
@@ -4,6 +4,9 @@ import (
"bufio"
"context"
"fmt"
+ "github.com/chrislusf/seaweedfs/weed/server"
+ "github.com/spf13/viper"
+ "google.golang.org/grpc"
"io"
"math"
"math/rand"
@@ -35,6 +38,7 @@ type BenchmarkOptions struct {
collection *string
cpuprofile *string
maxCpu *int
+ grpcDialOption grpc.DialOption
}
var (
@@ -101,6 +105,10 @@ var (
)
func runBenchmark(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+ b.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client")
+
fmt.Printf("This is SeaweedFS version %s %s %s\n", util.VERSION, runtime.GOOS, runtime.GOARCH)
if *b.maxCpu < 1 {
*b.maxCpu = runtime.NumCPU()
@@ -115,7 +123,7 @@ func runBenchmark(cmd *Command, args []string) bool {
defer pprof.StopCPUProfile()
}
- masterClient = wdclient.NewMasterClient(context.Background(), "benchmark", strings.Split(*b.masters, ","))
+ masterClient = wdclient.NewMasterClient(context.Background(), b.grpcDialOption, "client", strings.Split(*b.masters, ","))
go masterClient.KeepConnectedToMaster()
masterClient.WaitUntilConnected()
@@ -223,12 +231,12 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
Count: 1,
Collection: *b.collection,
}
- if assignResult, err := operation.Assign(masterClient.GetMaster(), ar); err == nil {
+ if assignResult, err := operation.Assign(masterClient.GetMaster(), b.grpcDialOption, ar); err == nil {
fp.Server, fp.Fid, fp.Collection = assignResult.Url, assignResult.Fid, *b.collection
if !isSecure && assignResult.Auth != "" {
isSecure = true
}
- if _, err := fp.Upload(0, masterClient.GetMaster(), assignResult.Auth); err == nil {
+ if _, err := fp.Upload(0, masterClient.GetMaster(), assignResult.Auth, b.grpcDialOption); err == nil {
if random.Intn(100) < *b.deletePercentage {
s.total++
delayedDeleteChan <- &delayedFile{time.Now().Add(time.Second), fp}
diff --git a/weed/command/filer.go b/weed/command/filer.go
index a07a67471..478b7d6bf 100644
--- a/weed/command/filer.go
+++ b/weed/command/filer.go
@@ -1,6 +1,8 @@
package command
import (
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/spf13/viper"
"net/http"
"strconv"
"strings"
@@ -75,6 +77,8 @@ var cmdFiler = &Command{
func runFiler(cmd *Command, args []string) bool {
+ weed_server.LoadConfiguration("security", false)
+
f.startFiler()
return true
@@ -141,7 +145,7 @@ func (fo *FilerOptions) startFiler() {
if err != nil {
glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
}
- grpcS := util.NewGrpcServer()
+ grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "filer"))
filer_pb.RegisterSeaweedFilerServer(grpcS, fs)
reflection.Register(grpcS)
go grpcS.Serve(grpcL)
diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go
index 39d83c31e..650757442 100644
--- a/weed/command/filer_copy.go
+++ b/weed/command/filer_copy.go
@@ -2,6 +2,10 @@ package command
import (
"fmt"
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/chrislusf/seaweedfs/weed/server"
+ "github.com/spf13/viper"
+ "google.golang.org/grpc"
"io/ioutil"
"net/url"
"os"
@@ -23,13 +27,14 @@ var (
)
type CopyOptions struct {
- filerGrpcPort *int
- master *string
- include *string
- replication *string
- collection *string
- ttl *string
- maxMB *int
+ filerGrpcPort *int
+ master *string
+ include *string
+ replication *string
+ collection *string
+ ttl *string
+ maxMB *int
+ grpcDialOption grpc.DialOption
}
func init() {
@@ -61,6 +66,9 @@ var cmdCopy = &Command{
}
func runCopy(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+
if len(args) <= 1 {
return false
}
@@ -95,16 +103,17 @@ func runCopy(cmd *Command, args []string) bool {
}
filerGrpcAddress := fmt.Sprintf("%s:%d", filerUrl.Hostname(), filerGrpcPort)
+ copy.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client")
for _, fileOrDir := range fileOrDirs {
- if !doEachCopy(fileOrDir, filerUrl.Host, filerGrpcAddress, urlPath) {
+ if !doEachCopy(fileOrDir, filerUrl.Host, filerGrpcAddress, copy.grpcDialOption, urlPath) {
return false
}
}
return true
}
-func doEachCopy(fileOrDir string, filerAddress, filerGrpcAddress string, path string) bool {
+func doEachCopy(fileOrDir string, filerAddress, filerGrpcAddress string, grpcDialOption grpc.DialOption, path string) bool {
f, err := os.Open(fileOrDir)
if err != nil {
fmt.Printf("Failed to open file %s: %v\n", fileOrDir, err)
@@ -122,7 +131,7 @@ func doEachCopy(fileOrDir string, filerAddress, filerGrpcAddress string, path st
if mode.IsDir() {
files, _ := ioutil.ReadDir(fileOrDir)
for _, subFileOrDir := range files {
- if !doEachCopy(fileOrDir+"/"+subFileOrDir.Name(), filerAddress, filerGrpcAddress, path+fi.Name()+"/") {
+ if !doEachCopy(fileOrDir+"/"+subFileOrDir.Name(), filerAddress, filerGrpcAddress, grpcDialOption, path+fi.Name()+"/") {
return false
}
}
@@ -144,13 +153,13 @@ func doEachCopy(fileOrDir string, filerAddress, filerGrpcAddress string, path st
}
if chunkCount == 1 {
- return uploadFileAsOne(filerAddress, filerGrpcAddress, path, f, fi)
+ return uploadFileAsOne(filerAddress, filerGrpcAddress, grpcDialOption, path, f, fi)
}
- return uploadFileInChunks(filerAddress, filerGrpcAddress, path, f, fi, chunkCount, chunkSize)
+ return uploadFileInChunks(filerAddress, filerGrpcAddress, grpcDialOption, path, f, fi, chunkCount, chunkSize)
}
-func uploadFileAsOne(filerAddress, filerGrpcAddress string, urlFolder string, f *os.File, fi os.FileInfo) bool {
+func uploadFileAsOne(filerAddress, filerGrpcAddress string, grpcDialOption grpc.DialOption, urlFolder string, f *os.File, fi os.FileInfo) bool {
// upload the file content
fileName := filepath.Base(f.Name())
@@ -161,7 +170,7 @@ func uploadFileAsOne(filerAddress, filerGrpcAddress string, urlFolder string, f
if fi.Size() > 0 {
// assign a volume
- assignResult, err := operation.Assign(*copy.master, &operation.VolumeAssignRequest{
+ assignResult, err := operation.Assign(*copy.master, grpcDialOption, &operation.VolumeAssignRequest{
Count: 1,
Replication: *copy.replication,
Collection: *copy.collection,
@@ -195,7 +204,7 @@ func uploadFileAsOne(filerAddress, filerGrpcAddress string, urlFolder string, f
fmt.Printf("copied %s => http://%s%s%s\n", fileName, filerAddress, urlFolder, fileName)
}
- if err := withFilerClient(filerGrpcAddress, func(client filer_pb.SeaweedFilerClient) error {
+ if err := withFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.CreateEntryRequest{
Directory: urlFolder,
Entry: &filer_pb.Entry{
@@ -228,7 +237,7 @@ func uploadFileAsOne(filerAddress, filerGrpcAddress string, urlFolder string, f
return true
}
-func uploadFileInChunks(filerAddress, filerGrpcAddress string, urlFolder string, f *os.File, fi os.FileInfo, chunkCount int, chunkSize int64) bool {
+func uploadFileInChunks(filerAddress, filerGrpcAddress string, grpcDialOption grpc.DialOption, urlFolder string, f *os.File, fi os.FileInfo, chunkCount int, chunkSize int64) bool {
fileName := filepath.Base(f.Name())
mimeType := detectMimeType(f)
@@ -238,7 +247,7 @@ func uploadFileInChunks(filerAddress, filerGrpcAddress string, urlFolder string,
for i := int64(0); i < int64(chunkCount); i++ {
// assign a volume
- assignResult, err := operation.Assign(*copy.master, &operation.VolumeAssignRequest{
+ assignResult, err := operation.Assign(*copy.master, grpcDialOption, &operation.VolumeAssignRequest{
Count: 1,
Replication: *copy.replication,
Collection: *copy.collection,
@@ -272,7 +281,7 @@ func uploadFileInChunks(filerAddress, filerGrpcAddress string, urlFolder string,
fmt.Printf("uploaded %s-%d to %s [%d,%d)\n", fileName, i+1, targetUrl, i*chunkSize, i*chunkSize+int64(uploadResult.Size))
}
- if err := withFilerClient(filerGrpcAddress, func(client filer_pb.SeaweedFilerClient) error {
+ if err := withFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.CreateEntryRequest{
Directory: urlFolder,
Entry: &filer_pb.Entry{
@@ -323,9 +332,9 @@ func detectMimeType(f *os.File) string {
return mimeType
}
-func withFilerClient(filerAddress string, fn func(filer_pb.SeaweedFilerClient) error) error {
+func withFilerClient(filerAddress string, grpcDialOption grpc.DialOption, fn func(filer_pb.SeaweedFilerClient) error) error {
- grpcConnection, err := util.GrpcDial(filerAddress)
+ grpcConnection, err := util.GrpcDial(filerAddress, grpcDialOption)
if err != nil {
return fmt.Errorf("fail to dial %s: %v", filerAddress, err)
}
diff --git a/weed/command/filer_replication.go b/weed/command/filer_replication.go
index c24f63bf0..c9afbdc8a 100644
--- a/weed/command/filer_replication.go
+++ b/weed/command/filer_replication.go
@@ -35,6 +35,7 @@ var cmdFilerReplicate = &Command{
func runFilerReplicate(cmd *Command, args []string) bool {
+ weed_server.LoadConfiguration("security", false)
weed_server.LoadConfiguration("replication", true)
weed_server.LoadConfiguration("notification", true)
config := viper.GetViper()
diff --git a/weed/command/master.go b/weed/command/master.go
index 6f1373aa2..5b45c9627 100644
--- a/weed/command/master.go
+++ b/weed/command/master.go
@@ -1,6 +1,8 @@
package command
import (
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/spf13/viper"
"net/http"
"os"
"runtime"
@@ -54,6 +56,9 @@ var (
)
func runMaster(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+
if *mMaxCpu < 1 {
*mMaxCpu = runtime.NumCPU()
}
@@ -104,7 +109,7 @@ func runMaster(cmd *Command, args []string) bool {
glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
}
// Create your protocol servers.
- grpcS := util.NewGrpcServer()
+ grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "master"))
master_pb.RegisterSeaweedServer(grpcS, ms)
reflection.Register(grpcS)
diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go
index 2937b9ef1..3e4249bfc 100644
--- a/weed/command/mount_std.go
+++ b/weed/command/mount_std.go
@@ -4,6 +4,9 @@ package command
import (
"fmt"
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/chrislusf/seaweedfs/weed/server"
+ "github.com/spf13/viper"
"os"
"os/user"
"runtime"
@@ -19,6 +22,9 @@ import (
)
func runMount(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+
fmt.Printf("This is SeaweedFS version %s %s %s\n", util.VERSION, runtime.GOOS, runtime.GOARCH)
if *mountOptions.dir == "" {
fmt.Printf("Please specify the mount directory via \"-dir\"")
@@ -91,6 +97,7 @@ func runMount(cmd *Command, args []string) bool {
err = fs.Serve(c, filesys.NewSeaweedFileSystem(&filesys.Option{
FilerGrpcAddress: filerGrpcAddress,
+ GrpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "client"),
FilerMountRootPath: mountRoot,
Collection: *mountOptions.collection,
Replication: *mountOptions.replication,
diff --git a/weed/command/s3.go b/weed/command/s3.go
index 16a9490ff..a54ddd2f7 100644
--- a/weed/command/s3.go
+++ b/weed/command/s3.go
@@ -1,6 +1,9 @@
package command
import (
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/chrislusf/seaweedfs/weed/server"
+ "github.com/spf13/viper"
"net/http"
"time"
@@ -46,6 +49,8 @@ var cmdS3 = &Command{
func runS3(cmd *Command, args []string) bool {
+ weed_server.LoadConfiguration("security", false)
+
filerGrpcAddress, err := parseFilerGrpcAddress(*s3options.filer, *s3options.filerGrpcPort)
if err != nil {
glog.Fatal(err)
@@ -59,6 +64,7 @@ func runS3(cmd *Command, args []string) bool {
FilerGrpcAddress: filerGrpcAddress,
DomainName: *s3options.domainName,
BucketsPath: *s3options.filerBucketsPath,
+ GrpcDialOption: security.LoadClientTLS(viper.Sub("grpc"), "client"),
})
if s3ApiServer_err != nil {
glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err)
diff --git a/weed/command/scaffold.go b/weed/command/scaffold.go
index 22300d3ba..e8608e9dd 100644
--- a/weed/command/scaffold.go
+++ b/weed/command/scaffold.go
@@ -250,8 +250,34 @@ directory = "/" # destination directory
# /etc/seaweedfs/security.toml
# this file is read by master, volume server, and filer
+# the jwt signing key is read by master and volume server
+# a jwt expires in 10 seconds
[jwt.signing]
key = ""
+# volume server also uses grpc that should be secured.
+
+# all grpc tls authentications are mutual
+[grpc]
+ca = ""
+
+[grpc.volume]
+cert = ""
+key = ""
+
+[grpc.master]
+cert = ""
+key = ""
+
+[grpc.filer]
+cert = ""
+key = ""
+
+# use this for any place needs a grpc client
+# i.e., "weed backup|benchmark|filer.copy|filer.replicate|mount|s3|upload"
+[grpc.client]
+cert = ""
+key = ""
+
`
)
diff --git a/weed/command/server.go b/weed/command/server.go
index 2dd506772..a9415d068 100644
--- a/weed/command/server.go
+++ b/weed/command/server.go
@@ -1,6 +1,8 @@
package command
import (
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/spf13/viper"
"net/http"
"os"
"runtime"
@@ -95,6 +97,9 @@ func init() {
}
func runServer(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+
if *serverOptions.cpuprofile != "" {
f, err := os.Create(*serverOptions.cpuprofile)
if err != nil {
@@ -188,7 +193,8 @@ func runServer(cmd *Command, args []string) bool {
glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
}
// Create your protocol servers.
- grpcS := util.NewGrpcServer()
+ glog.V(0).Infof("grpc config %+v", viper.Sub("grpc"))
+ grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "master"))
master_pb.RegisterSeaweedServer(grpcS, ms)
reflection.Register(grpcS)
diff --git a/weed/command/upload.go b/weed/command/upload.go
index df2cb9892..80fc635c1 100644
--- a/weed/command/upload.go
+++ b/weed/command/upload.go
@@ -3,6 +3,9 @@ package command
import (
"encoding/json"
"fmt"
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/chrislusf/seaweedfs/weed/server"
+ "github.com/spf13/viper"
"os"
"path/filepath"
@@ -57,6 +60,10 @@ var cmdUpload = &Command{
}
func runUpload(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+ grpcDialOption := security.LoadClientTLS(viper.Sub("grpc"), "client")
+
if len(args) == 0 {
if *upload.dir == "" {
return false
@@ -73,7 +80,7 @@ func runUpload(cmd *Command, args []string) bool {
if e != nil {
return e
}
- results, e := operation.SubmitFiles(*upload.master, parts,
+ results, e := operation.SubmitFiles(*upload.master, grpcDialOption, parts,
*upload.replication, *upload.collection, *upload.dataCenter,
*upload.ttl, *upload.maxMB)
bytes, _ := json.Marshal(results)
@@ -92,7 +99,7 @@ func runUpload(cmd *Command, args []string) bool {
if e != nil {
fmt.Println(e.Error())
}
- results, _ := operation.SubmitFiles(*upload.master, parts,
+ results, _ := operation.SubmitFiles(*upload.master, grpcDialOption, parts,
*upload.replication, *upload.collection, *upload.dataCenter,
*upload.ttl, *upload.maxMB)
bytes, _ := json.Marshal(results)
diff --git a/weed/command/volume.go b/weed/command/volume.go
index 27a075b5b..32ec7819b 100644
--- a/weed/command/volume.go
+++ b/weed/command/volume.go
@@ -1,6 +1,8 @@
package command
import (
+ "github.com/chrislusf/seaweedfs/weed/security"
+ "github.com/spf13/viper"
"net/http"
"os"
"runtime"
@@ -78,6 +80,9 @@ var (
)
func runVolume(cmd *Command, args []string) bool {
+
+ weed_server.LoadConfiguration("security", false)
+
if *v.maxCpu < 1 {
*v.maxCpu = runtime.NumCPU()
}
@@ -185,7 +190,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
if err != nil {
glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
}
- grpcS := util.NewGrpcServer()
+ grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "volume"))
volume_server_pb.RegisterVolumeServerServer(grpcS, volumeServer)
reflection.Register(grpcS)
go grpcS.Serve(grpcL)
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 15:12:28 +0000