seaweedfs/weed/command/s3.go
Amir H. Yeganemehr f0d3c874b6
Unix Socket listener for S3 server (#4621)
Add localSocket option to s3 server
2023-06-26 16:22:45 -07:00

340 lines
11 KiB
Go

package command
import (
"context"
"crypto/tls"
"fmt"
"net"
"net/http"
"os"
"runtime"
"time"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
"google.golang.org/grpc/credentials/tls/certprovider"
"google.golang.org/grpc/credentials/tls/certprovider/pemfile"
"google.golang.org/grpc/reflection"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
"github.com/seaweedfs/seaweedfs/weed/security"
"github.com/gorilla/mux"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/s3api"
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/util"
)
var (
s3StandaloneOptions S3Options
)
type S3Options struct {
filer *string
bindIp *string
port *int
portHttps *int
portGrpc *int
config *string
domainName *string
tlsPrivateKey *string
tlsCertificate *string
metricsHttpPort *int
allowEmptyFolder *bool
allowDeleteBucketNotEmpty *bool
auditLogConfig *string
localFilerSocket *string
dataCenter *string
localSocket *string
certProvider certprovider.Provider
}
func init() {
cmdS3.Run = runS3 // break init cycle
s3StandaloneOptions.filer = cmdS3.Flag.String("filer", "localhost:8888", "filer server address")
s3StandaloneOptions.bindIp = cmdS3.Flag.String("ip.bind", "", "ip address to bind to. Default to localhost.")
s3StandaloneOptions.port = cmdS3.Flag.Int("port", 8333, "s3 server http listen port")
s3StandaloneOptions.portHttps = cmdS3.Flag.Int("port.https", 0, "s3 server https listen port")
s3StandaloneOptions.portGrpc = cmdS3.Flag.Int("port.grpc", 0, "s3 server grpc listen port")
s3StandaloneOptions.domainName = cmdS3.Flag.String("domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}")
s3StandaloneOptions.dataCenter = cmdS3.Flag.String("dataCenter", "", "prefer to read and write to volumes in this data center")
s3StandaloneOptions.config = cmdS3.Flag.String("config", "", "path to the config file")
s3StandaloneOptions.auditLogConfig = cmdS3.Flag.String("auditLogConfig", "", "path to the audit log config file")
s3StandaloneOptions.tlsPrivateKey = cmdS3.Flag.String("key.file", "", "path to the TLS private key file")
s3StandaloneOptions.tlsCertificate = cmdS3.Flag.String("cert.file", "", "path to the TLS certificate file")
s3StandaloneOptions.metricsHttpPort = cmdS3.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
s3StandaloneOptions.allowEmptyFolder = cmdS3.Flag.Bool("allowEmptyFolder", true, "allow empty folders")
s3StandaloneOptions.allowDeleteBucketNotEmpty = cmdS3.Flag.Bool("allowDeleteBucketNotEmpty", true, "allow recursive deleting all entries along with bucket")
s3StandaloneOptions.localFilerSocket = cmdS3.Flag.String("localFilerSocket", "", "local filer socket path")
s3StandaloneOptions.localSocket = cmdS3.Flag.String("localSocket", "", "default to /tmp/seaweedfs-s3-<port>.sock")
}
var cmdS3 = &Command{
UsageLine: "s3 [-port=8333] [-filer=<ip:port>] [-config=</path/to/config.json>]",
Short: "start a s3 API compatible server that is backed by a filer",
Long: `start a s3 API compatible server that is backed by a filer.
By default, you can use any access key and secret key to access the S3 APIs.
To enable credential based access, create a config.json file similar to this:
{
"identities": [
{
"name": "anonymous",
"actions": [
"Read"
]
},
{
"name": "some_admin_user",
"credentials": [
{
"accessKey": "some_access_key1",
"secretKey": "some_secret_key1"
}
],
"actions": [
"Admin",
"Read",
"List",
"Tagging",
"Write"
]
},
{
"name": "some_read_only_user",
"credentials": [
{
"accessKey": "some_access_key2",
"secretKey": "some_secret_key2"
}
],
"actions": [
"Read"
]
},
{
"name": "some_normal_user",
"credentials": [
{
"accessKey": "some_access_key3",
"secretKey": "some_secret_key3"
}
],
"actions": [
"Read",
"List",
"Tagging",
"Write"
]
},
{
"name": "user_limited_to_bucket1",
"credentials": [
{
"accessKey": "some_access_key4",
"secretKey": "some_secret_key4"
}
],
"actions": [
"Read:bucket1",
"List:bucket1",
"Tagging:bucket1",
"Write:bucket1"
]
}
]
}
`,
}
func runS3(cmd *Command, args []string) bool {
util.LoadConfiguration("security", false)
go stats_collect.StartMetricsServer(*s3StandaloneOptions.bindIp, *s3StandaloneOptions.metricsHttpPort)
return s3StandaloneOptions.startS3Server()
}
// GetCertificateWithUpdate Auto refreshing TSL certificate
func (S3opt *S3Options) GetCertificateWithUpdate(*tls.ClientHelloInfo) (*tls.Certificate, error) {
certs, err := S3opt.certProvider.KeyMaterial(context.Background())
return &certs.Certs[0], err
}
func (s3opt *S3Options) startS3Server() bool {
filerAddress := pb.ServerAddress(*s3opt.filer)
filerBucketsPath := "/buckets"
filerGroup := ""
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
// metrics read from the filer
var metricsAddress string
var metricsIntervalSec int
for {
err := pb.WithGrpcFilerClient(false, 0, filerAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return fmt.Errorf("get filer %s configuration: %v", filerAddress, err)
}
filerBucketsPath = resp.DirBuckets
filerGroup = resp.FilerGroup
metricsAddress, metricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSec)
glog.V(0).Infof("S3 read filer buckets dir: %s", filerBucketsPath)
return nil
})
if err != nil {
glog.V(0).Infof("wait to connect to filer %s grpc address %s", *s3opt.filer, filerAddress.ToGrpcAddress())
time.Sleep(time.Second)
} else {
glog.V(0).Infof("connected to filer %s grpc address %s", *s3opt.filer, filerAddress.ToGrpcAddress())
break
}
}
go stats_collect.LoopPushingMetric("s3", stats_collect.SourceName(uint32(*s3opt.port)), metricsAddress, metricsIntervalSec)
router := mux.NewRouter().SkipClean(true)
var localFilerSocket string
if s3opt.localFilerSocket != nil {
localFilerSocket = *s3opt.localFilerSocket
}
s3ApiServer, s3ApiServer_err := s3api.NewS3ApiServer(router, &s3api.S3ApiServerOption{
Filer: filerAddress,
Port: *s3opt.port,
Config: *s3opt.config,
DomainName: *s3opt.domainName,
BucketsPath: filerBucketsPath,
GrpcDialOption: grpcDialOption,
AllowEmptyFolder: *s3opt.allowEmptyFolder,
AllowDeleteBucketNotEmpty: *s3opt.allowDeleteBucketNotEmpty,
LocalFilerSocket: localFilerSocket,
DataCenter: *s3opt.dataCenter,
FilerGroup: filerGroup,
})
if s3ApiServer_err != nil {
glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err)
}
httpS := &http.Server{Handler: router}
if *s3opt.portGrpc == 0 {
*s3opt.portGrpc = 10000 + *s3opt.port
}
if *s3opt.bindIp == "" {
*s3opt.bindIp = "localhost"
}
if runtime.GOOS != "windows" {
localSocket := *s3opt.localSocket
if localSocket == "" {
localSocket = fmt.Sprintf("/tmp/seaweedfs-s3-%d.sock", *s3opt.port)
}
if err := os.Remove(localSocket); err != nil && !os.IsNotExist(err) {
glog.Fatalf("Failed to remove %s, error: %s", localSocket, err.Error())
}
go func() {
// start on local unix socket
s3SocketListener, err := net.Listen("unix", localSocket)
if err != nil {
glog.Fatalf("Failed to listen on %s: %v", localSocket, err)
}
httpS.Serve(s3SocketListener)
}()
}
listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.port)
s3ApiListener, s3ApiLocalListener, err := util.NewIpAndLocalListeners(*s3opt.bindIp, *s3opt.port, time.Duration(10)*time.Second)
if err != nil {
glog.Fatalf("S3 API Server listener on %s error: %v", listenAddress, err)
}
if len(*s3opt.auditLogConfig) > 0 {
s3err.InitAuditLog(*s3opt.auditLogConfig)
if s3err.Logger != nil {
defer s3err.Logger.Close()
}
}
// starting grpc server
grpcPort := *s3opt.portGrpc
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*s3opt.bindIp, grpcPort, 0)
if err != nil {
glog.Fatalf("s3 failed to listen on grpc port %d: %v", grpcPort, err)
}
grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.s3"))
s3_pb.RegisterSeaweedS3Server(grpcS, s3ApiServer)
reflection.Register(grpcS)
if grpcLocalL != nil {
go grpcS.Serve(grpcLocalL)
}
go grpcS.Serve(grpcL)
if *s3opt.tlsPrivateKey != "" {
pemfileOptions := pemfile.Options{
CertFile: *s3opt.tlsCertificate,
KeyFile: *s3opt.tlsPrivateKey,
RefreshDuration: security.CredRefreshingInterval,
}
if s3opt.certProvider, err = pemfile.NewProvider(pemfileOptions); err != nil {
glog.Fatalf("pemfile.NewProvider(%v) failed: %v", pemfileOptions, err)
}
httpS.TLSConfig = &tls.Config{GetCertificate: s3opt.GetCertificateWithUpdate}
if *s3opt.portHttps == 0 {
glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", util.Version(), *s3opt.port)
if s3ApiLocalListener != nil {
go func() {
if err = httpS.ServeTLS(s3ApiLocalListener, "", ""); err != nil {
glog.Fatalf("S3 API Server Fail to serve: %v", err)
}
}()
}
if err = httpS.ServeTLS(s3ApiListener, "", ""); err != nil {
glog.Fatalf("S3 API Server Fail to serve: %v", err)
}
} else {
glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", util.Version(), *s3opt.portHttps)
s3ApiListenerHttps, s3ApiLocalListenerHttps, _ := util.NewIpAndLocalListeners(
*s3opt.bindIp, *s3opt.portHttps, time.Duration(10)*time.Second)
if s3ApiLocalListenerHttps != nil {
go func() {
if err = httpS.ServeTLS(s3ApiLocalListenerHttps, "", ""); err != nil {
glog.Fatalf("S3 API Server Fail to serve: %v", err)
}
}()
}
go func() {
if err = httpS.ServeTLS(s3ApiListenerHttps, "", ""); err != nil {
glog.Fatalf("S3 API Server Fail to serve: %v", err)
}
}()
}
}
if *s3opt.tlsPrivateKey == "" || *s3opt.portHttps > 0 {
glog.V(0).Infof("Start Seaweed S3 API Server %s at http port %d", util.Version(), *s3opt.port)
if s3ApiLocalListener != nil {
go func() {
if err = httpS.Serve(s3ApiLocalListener); err != nil {
glog.Fatalf("S3 API Server Fail to serve: %v", err)
}
}()
}
if err = httpS.Serve(s3ApiListener); err != nil {
glog.Fatalf("S3 API Server Fail to serve: %v", err)
}
}
return true
}