Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-29 16:30:44 +01:00
Code Issues Packages Projects Releases Wiki Activity
Files
18691e7b5b2b44442e72040b51d4e892a500e0d1
seaweedfs/weed/command/filer.go
T
Chris LuandDevin 18691e7b5b server: keep the first write on a fresh weed server from stalling (#12149) 
* master: end the post-mint allocation hold on the embedded reporter's clean inventory

DeferAllocation armed a warmup-length hold whenever the leader minted a
fresh topology id, and on weed server the hold ran the whole window: the
embedded volume server's join kept refreshing the quiet evidence clock, so
the two-pulse early release still cost the first write 10-15s on every
fresh start. weed mini already releases on its single pinned reporter's
first inventory, but pinning refuses outside volume servers, which weed
server must keep admitting.

Add a non-exclusive variant: the topology names one expected reporter
whose first clean full inventory lifts the hold, while other reporters
stay admitted and still feed prior-state evidence and the topology-id
freeze. A single-master weed server points it at its embedded volume
server — the one reporter guaranteed to arrive quickly; a veteran's
heartbeat carries prior state and keeps the window through the regular
checks, so a wiped -dir with surviving external volume servers still
trips the freeze on first report. Restarted clusters are untouched: a
recovered topology id means no mint and no hold.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: announce on the startup peer scan for embedded single-master setups

A fresh filer whose startup peer scan came back empty re-polled for the
full confirm window before registering on the master, so the lock ring
had no member and routed traffic such as S3 PUTs spun on "no lock server
found" for ~10s on every fresh weed server / weed mini start.

For setups that start with exactly one filer the scan is already final:
a single-master weed server and weed mini announce as soon as it
completes. A late external filer still joins and replays through the
normal path — the confirm window only ever caught peers arriving inside
it, and the residual risk mirrors the accepted volume-side trade-off: an
embedded filer announcing a few seconds earlier can own the lock ring
while a surviving external filer's backlog is still on its way.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: retry a failed heartbeat promptly until first registration

A heartbeat stream refused before the leader is known — the common race
when a volume server starts while an election is still running — always
fell back to sleeping out the full pulse, so a fresh volume's first
registration could wait ~5s behind a leader elected a second later.

The peer refresh already identifies the leader during the wait, so cap
the retry pause at a second once a leader is named; while no leader is
known the same short retry applies until the first stream is confirmed,
since an unregistered server has nothing else to do. Once the server was
registered the pulse backoff is unchanged, so a lost leader still paces
reconnects the way it did before.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: retry a failed heartbeat promptly until first registration (rust)

Mirror the Go heartbeat retry: cap the post-refresh wait at a second when
a leader is named, and keep the short retry until the first stream is
confirmed so a fresh volume registers promptly when it races an election.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* master: leave admitted reporters a confirm pulse when the embedded report ends the hold

The single-master weed server releases its post-mint allocation hold on
the embedded volume server's clean report, but outside volume servers
stay admitted and one may still carry prior cluster state — an inventory
from the embedded server alone cannot prove none is due. Clearing the
hold outright let a rebuilt master re-issue a volume id a reconnecting
veteran still owned.

Collapse the released hold to one heartbeat interval instead: late
reporters land their evidence or a topology-id freeze inside the pulse,
and a re-mint re-arms that pulse rather than the full warmup. The pinned
single-reporter release (weed mini) still drops the hold outright, since
no other reporter can be admitted.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: keep the empty-membership confirm on embedded single-master weed server

A single-master weed server still admits external filers, so an empty
startup peer scan is not final: after a master restart the existing
filers simply have not reconnected yet. Announcing on the scan let the
embedded filer take lock-ring ownership and serve writes that a peer's
replayed backlog could then overwrite.

Only weed mini keeps the skip, where the embedded filer is the only
member its dev cluster can have. The confirm window only delays the
master listing for a fresh store; the filer serves while it runs.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: tolerate an expired heartbeat retry deadline (rust)

refresh_leader_from_peers can consume the whole pulse budget, leaving
retry_at in the past; subtracting a later Instant panics and kills the
heartbeat task instead of retrying. Match the Go side, where
time.Until of a past deadline just sleeps zero.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: retry the seed list without a second pulse sleep (rust)

The error path already slept its retry delay; an extra full pulse after
losing the round delayed the next attempt past what Go's heartbeat()
does, which paces retries only inside the per-master attempt.

* master: keep a configured allocation hold shorter than the confirm pulse

Once the embedded report lands, the collapsed hold replaced the
configured deferral with a full pulse even when the operator set a
shorter one, including zero (disabled). Cap at the pulse instead.

* filer: end the empty-membership confirm early on a fresh-minted embedded master

On a fresh single-master weed server the announce gate still held the
filer for the full confirm window, so the master's lock ring stayed
empty and the first write retried 'no lock server found' for ~7s on
top of the collapsed allocation pulse. A fresh TopologyId mint proves
the master's persisted state was empty, making the empty peer scan
final; skip the rest of the confirm only then, and keep it whenever
the master recovers a prior id.

* filer: keep a short reconnect window on a fresh-minted embedded master

A fresh TopologyId mint proves the master's persisted state was empty,
not that no external filer survived its loss — one can still be
mid-reconnect on its keepalive cycle when the confirm starts. Ending
the empty-scan confirm on the first poll let the fresh filer register
and serve routed reads before that peer was found and backfilled, so
existing files could read as missing. The fresh-mint path now shortens
the confirm to a few keepalive cycles instead of skipping it: a
reconnecting filer still lands in time, while a genuinely fresh cluster
announces after ~3s rather than ~10s.

* volume: skip the heartbeat loop when no seed masters are configured

An empty seed list made the outer loop spin with no await between
passes and no path to observe the stop flag. A masterless volume server
has nothing to register with, so the heartbeat exits after a warning.

* volume: skip the heartbeat loop when no master addresses are given (rust)

Mirrors the Go heartbeat: an empty seed list spun the outer loop with
no await and no shutdown checks. Exit after a warning so library
callers get the same behavior as the binary path that validates the
list up front.

* filer: re-check the fresh-mint hint on each confirm pass

The topology id is minted after raft election, which can land after
the empty-scan confirm has already started — a one-time check at entry
missed the mint and held the full window. Evaluate the hint on every
poll and shorten the confirm from the first observed mint, still
capped by the normal confirm deadline.

* master: mark a fresh topology mint only when the volume bound is zero

Recovery restores the max volume id unconditionally but the topology id
only when present, so an upgraded or partially rebuilt master can mint a
new id while still holding prior volume state. Marking fresh there would
let the embedded filer shortcut its peer confirm on a cluster that is
not empty; only mark the mint fresh when no bound survived.

* filer: drop the fresh-mint shortcut once a peer has been observed

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-28 08:08:26 -07:00

680 lines
33 KiB
Go
Raw Blame History

package command
import (
"context"
"crypto/tls"
"crypto/x509"
"fmt"
"net/http"
"os"
"runtime"
"sort"
"strings"
"sync"
"time"
"github.com/spf13/viper"
"google.golang.org/grpc/reflection"
"github.com/seaweedfs/seaweedfs/weed/credential"
_ "github.com/seaweedfs/seaweedfs/weed/credential/filer_etc"
_ "github.com/seaweedfs/seaweedfs/weed/credential/memory"
_ "github.com/seaweedfs/seaweedfs/weed/credential/postgres"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/iam/integration"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
"github.com/seaweedfs/seaweedfs/weed/security"
weed_server "github.com/seaweedfs/seaweedfs/weed/server"
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/seaweedfs/seaweedfs/weed/util/grace"
"github.com/seaweedfs/seaweedfs/weed/util/version"
)
var (
f FilerOptions
filerStartS3 *bool
filerS3Options S3Options
filerStartWebDav *bool
filerWebDavOptions WebDavOption
filerStartIam *bool
filerIamOptions IamOptions
filerStartSftp *bool
filerSftpOptions SftpOptions
)
// allowUntrustedRemoteEndpointsUsage documents the flag shared by the filer and
// S3 gateway, whose remote-mount read paths dial the mounted endpoint directly.
const allowUntrustedRemoteEndpointsUsage = "if true, a read of a remote-only entry accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints."
// iamSelfServiceRequiresGrantUsage documents the flag shared by every command
// that runs the embedded IAM API.
const iamSelfServiceRequiresGrantUsage = "if true, an identity needs an explicit iam: grant to manage its own access keys (CreateAccessKey, DeleteAccessKey, UpdateAccessKey, ListAccessKeys, GetUser), as on AWS. Default allows it to every authenticated identity."
// allowBackslashInKeyUsage documents the S3 flag keeping '\' a literal
// object-key character instead of normalizing it to '/'. Repeated '//' in a
// key still collapses: a filer entry name cannot contain an empty segment.
const allowBackslashInKeyUsage = "treat backslash as a regular object-key character instead of converting it to '/' (aws-compatible; '//' still collapses)"
type FilerOptions struct {
masters *pb.ServerDiscovery
mastersString *string
ip *string
bindIp *string
listenOnLocalhost *bool
port *int
portGrpc *int
publicPort *int
idleTimeout *int
filerGroup *string
collection *string
defaultReplicaPlacement *string
disableDirListing *bool
maxMB *int
dirListingLimit *int
dataCenter *string
rack *string
enableNotification *bool
disableHttp *bool
cipher *bool
metricsHttpPort *int
metricsHttpIp *string
saveToFilerLimit *int
defaultLevelDbDirectory *string
concurrentUploadLimitMB *int
concurrentFileUploadLimit *int
debug *bool
debugPort *int
localSocket *string
showUIDirectoryDelete *bool
downloadMaxMBps *int
diskType *string
allowedOrigins *string
exposeDirectoryData *bool
tusBasePath *string
tusMaxSizeMB *int
tusSessionExpiry *time.Duration
s3ConfigFile *string // optional path to static S3 identity config
allowUntrustedRemoteEndpoints *bool
remoteCacheEvictThreshold *float64
// shutdownCtx, when non-nil, tells startFiler to gracefully shut down its
// HTTP/gRPC servers once the ctx is cancelled. Used by integration tests
// and by weed mini; nil for standalone weed filer.
shutdownCtx context.Context
// gracefulStopTimeout caps how long startFiler waits for gRPC graceful
// stop before forcing the server to stop. Zero means the default of 15s.
gracefulStopTimeout time.Duration
// skipEmptyMembershipConfirm forwards to FilerOption: a single-master
// weed mini's embedded filer announces on the startup scan instead of
// holding registration for the confirm window.
skipEmptyMembershipConfirm bool
// clusterFreshHint, when non-nil, reports that the embedded master minted
// a fresh TopologyId this boot — proof the peer scan's empty result is
// final rather than a settling master, so the announce gate can end the
// empty-membership confirm early. Used by weed server; nil elsewhere.
clusterFreshHint func() bool
}
func init() {
cmdFiler.Run = runFiler // break init cycle
f.mastersString = cmdFiler.Flag.String("master", "localhost:9333", "comma-separated master servers or a single DNS SRV record of at least 1 master server, prepended with dnssrv+")
f.filerGroup = cmdFiler.Flag.String("filerGroup", "", "share metadata with other filers in the same filerGroup")
f.collection = cmdFiler.Flag.String("collection", "", "all data will be stored in this default collection")
f.ip = cmdFiler.Flag.String("ip", util.DetectedHostAddress(), "filer server http listen ip address")
f.bindIp = cmdFiler.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
f.port = cmdFiler.Flag.Int("port", 8888, "filer server http listen port")
f.portGrpc = cmdFiler.Flag.Int("port.grpc", 0, "filer server grpc listen port")
f.publicPort = cmdFiler.Flag.Int("port.readonly", 0, "readonly port opened to public")
f.idleTimeout = cmdFiler.Flag.Int("idleTimeout", 10, "connection idle seconds")
f.defaultReplicaPlacement = cmdFiler.Flag.String("defaultReplicaPlacement", "", "default replication type. If not specified, use master setting.")
f.disableDirListing = cmdFiler.Flag.Bool("disableDirListing", false, "turn off directory listing")
f.maxMB = cmdFiler.Flag.Int("maxMB", 4, "split files larger than the limit")
f.dirListingLimit = cmdFiler.Flag.Int("dirListLimit", 100000, "limit sub dir listing size")
f.dataCenter = cmdFiler.Flag.String("dataCenter", "", "prefer to read and write to volumes in this data center")
f.rack = cmdFiler.Flag.String("rack", "", "prefer to write to volumes in this rack")
f.disableHttp = cmdFiler.Flag.Bool("disableHttp", false, "disable http request, only gRpc operations are allowed")
f.cipher = cmdFiler.Flag.Bool("encryptVolumeData", false, "encrypt data on volume servers")
f.metricsHttpPort = cmdFiler.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
f.metricsHttpIp = cmdFiler.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
f.saveToFilerLimit = cmdFiler.Flag.Int("saveToFilerLimit", 0, "files smaller than this limit will be saved in filer store")
f.defaultLevelDbDirectory = cmdFiler.Flag.String("defaultStoreDir", ".", "if filer.toml is empty, use an embedded filer store in the directory")
f.concurrentUploadLimitMB = cmdFiler.Flag.Int("concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited")
f.concurrentFileUploadLimit = cmdFiler.Flag.Int("concurrentFileUploadLimit", 0, "limit number of concurrent file uploads, 0 means unlimited")
f.debug = cmdFiler.Flag.Bool("debug", false, "serves runtime profiling data, e.g., http://localhost:<debug.port>/debug/pprof/goroutine?debug=2")
f.debugPort = cmdFiler.Flag.Int("debug.port", 6060, "http port for debugging")
f.localSocket = cmdFiler.Flag.String("localSocket", "", "default to /tmp/seaweedfs-filer-<port>.sock")
f.showUIDirectoryDelete = cmdFiler.Flag.Bool("ui.deleteDir", true, "enable filer UI show delete directory button")
f.downloadMaxMBps = cmdFiler.Flag.Int("downloadMaxMBps", 0, "download max speed for each download request, in MB per second")
f.diskType = cmdFiler.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
f.allowedOrigins = cmdFiler.Flag.String("allowedOrigins", "*", "comma separated list of allowed origins")
f.exposeDirectoryData = cmdFiler.Flag.Bool("exposeDirectoryData", true, "whether to return directory metadata and content in Filer UI")
f.tusBasePath = cmdFiler.Flag.String("tusBasePath", "/.tus", "TUS resumable upload endpoint base path (e.g., /.tus)")
f.tusMaxSizeMB = cmdFiler.Flag.Int("tusMaxSizeMB", 5*1024, "maximum TUS upload size in MB")
f.tusSessionExpiry = cmdFiler.Flag.Duration("tusSessionExpiry", 24*time.Hour, "incomplete TUS upload sessions are cleaned up after this duration, e.g. \"48h\", \"7h30m\"")
f.allowUntrustedRemoteEndpoints = cmdFiler.Flag.Bool("allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
f.remoteCacheEvictThreshold = cmdFiler.Flag.Float64("remoteCacheEvictThreshold", 0.9, "evict remote-cached objects (oldest first) when any volume disk exceeds this usage fraction; 0 disables")
// start s3 on filer
filerStartS3 = cmdFiler.Flag.Bool("s3", false, "whether to start S3 gateway")
filerS3Options.port = cmdFiler.Flag.Int("s3.port", 8333, "s3 server http listen port")
filerS3Options.portHttps = cmdFiler.Flag.Int("s3.port.https", 0, "s3 server https listen port")
filerS3Options.portGrpc = cmdFiler.Flag.Int("s3.port.grpc", 0, "s3 server grpc listen port")
filerS3Options.domainName = cmdFiler.Flag.String("s3.domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}")
filerS3Options.allowedOrigins = cmdFiler.Flag.String("s3.allowedOrigins", "*", "comma separated list of allowed origins")
filerS3Options.dataCenter = cmdFiler.Flag.String("s3.dataCenter", "", "prefer to read and write to volumes in this data center")
filerS3Options.tlsPrivateKey = cmdFiler.Flag.String("s3.key.file", "", "path to the TLS private key file")
filerS3Options.tlsCertificate = cmdFiler.Flag.String("s3.cert.file", "", "path to the TLS certificate file")
filerS3Options.config = cmdFiler.Flag.String("s3.config", "", "path to the config file")
filerS3Options.iamConfig = cmdFiler.Flag.String("s3.iam.config", "", "path to the advanced IAM config file")
filerS3Options.auditLogConfig = cmdFiler.Flag.String("s3.auditLogConfig", "", "path to the audit log config file")
filerS3Options.metricsHttpPort = cmdFiler.Flag.Int("s3.metricsPort", 0, "Prometheus metrics listen port")
filerS3Options.metricsHttpIp = cmdFiler.Flag.String("s3.metricsIp", "", "metrics listen ip. If empty, default to same as -s3.ip.bind option.")
cmdFiler.Flag.Bool("s3.allowEmptyFolder", true, "deprecated, ignored. Empty folder cleanup is now automatic.")
filerS3Options.allowDeleteBucketNotEmpty = cmdFiler.Flag.Bool("s3.allowDeleteBucketNotEmpty", false, "allow recursive deleting all entries along with bucket")
filerS3Options.autoCreateBucket = cmdFiler.Flag.Bool("s3.autoCreateBucket", true, "create the bucket on upload if it does not exist, for admin identities only")
filerS3Options.localSocket = cmdFiler.Flag.String("s3.localSocket", "", "default to /tmp/seaweedfs-s3-<port>.sock")
filerS3Options.tlsCACertificate = cmdFiler.Flag.String("s3.cacert.file", "", "path to the TLS CA certificate file")
filerS3Options.tlsVerifyClientCert = cmdFiler.Flag.Bool("s3.tlsVerifyClientCert", false, "whether to verify the client's certificate")
filerS3Options.bindIp = cmdFiler.Flag.String("s3.ip.bind", "", "ip address to bind to. If empty, default to same as -ip.bind option.")
filerS3Options.idleTimeout = cmdFiler.Flag.Int("s3.idleTimeout", 120, "connection idle seconds")
filerS3Options.concurrentUploadLimitMB = cmdFiler.Flag.Int("s3.concurrentUploadLimitMB", 0, "limit total concurrent upload size for S3, 0 means unlimited")
filerS3Options.concurrentFileUploadLimit = cmdFiler.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads for S3, 0 means unlimited")
filerS3Options.enableIam = cmdFiler.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same S3 port")
filerS3Options.cipher = cmdFiler.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads")
filerS3Options.iamReadOnly = cmdFiler.Flag.Bool("s3.iam.readOnly", true, "disable IAM write operations on this server")
filerS3Options.portIceberg = cmdFiler.Flag.Int("s3.port.iceberg", 8181, "Iceberg REST Catalog server listen port (0 to disable)")
filerS3Options.portLance = cmdFiler.Flag.Int("s3.port.lance", 9101, "Lance Namespace server listen port (0 to disable)")
filerS3Options.externalUrl = cmdFiler.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.")
filerS3Options.defaultFileMode = cmdFiler.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666")
filerS3Options.cacheSizeMB = cmdFiler.Flag.Int64("s3.cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)")
filerS3Options.readerCacheSizeMB = cmdFiler.Flag.Int64("s3.readerCacheSizeMB", 0, "memory budget in MiB for downloaded and in-flight reader buffers across all S3 GETs (0 means unlimited)")
filerS3Options.allowUntrustedRemoteEndpoints = cmdFiler.Flag.Bool("s3.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage)
filerS3Options.allowBackslashInKey = cmdFiler.Flag.Bool("s3.allowBackslashInKey", false, allowBackslashInKeyUsage)
filerS3Options.iamSelfServiceRequiresGrant = cmdFiler.Flag.Bool("s3.iam.selfServiceRequiresGrant", false, iamSelfServiceRequiresGrantUsage)
// start webdav on filer
filerStartWebDav = cmdFiler.Flag.Bool("webdav", false, "whether to start webdav gateway")
filerWebDavOptions.port = cmdFiler.Flag.Int("webdav.port", 7333, "webdav server http listen port")
filerWebDavOptions.collection = cmdFiler.Flag.String("webdav.collection", "", "collection to create the files")
filerWebDavOptions.replication = cmdFiler.Flag.String("webdav.replication", "", "replication to create the files")
filerWebDavOptions.disk = cmdFiler.Flag.String("webdav.disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
filerWebDavOptions.tlsPrivateKey = cmdFiler.Flag.String("webdav.key.file", "", "path to the TLS private key file")
filerWebDavOptions.tlsCertificate = cmdFiler.Flag.String("webdav.cert.file", "", "path to the TLS certificate file")
filerWebDavOptions.cacheDir = cmdFiler.Flag.String("webdav.cacheDir", os.TempDir(), "local cache directory for file chunks")
filerWebDavOptions.cacheSizeMB = cmdFiler.Flag.Int64("webdav.cacheCapacityMB", 0, "local cache capacity in MB")
filerWebDavOptions.maxMB = cmdFiler.Flag.Int("webdav.maxMB", 4, "split files larger than the limit")
filerWebDavOptions.filerRootPath = cmdFiler.Flag.String("webdav.filer.path", "/", "use this remote path from filer server")
filerWebDavOptions.idleTimeout = cmdFiler.Flag.Int("webdav.idleTimeout", 10, "connection idle seconds")
// start iam on filer
filerStartIam = cmdFiler.Flag.Bool("iam", false, "whether to start IAM service")
filerIamOptions.ip = cmdFiler.Flag.String("iam.ip", *f.ip, "iam server http listen ip address")
filerIamOptions.port = cmdFiler.Flag.Int("iam.port", 8111, "iam server http listen port")
filerIamOptions.idleTimeout = cmdFiler.Flag.Int("iam.idleTimeout", 10, "connection idle seconds")
filerIamOptions.selfServiceRequiresGrant = filerS3Options.iamSelfServiceRequiresGrant
filerStartSftp = cmdFiler.Flag.Bool("sftp", false, "whether to start the SFTP server")
filerSftpOptions.port = cmdFiler.Flag.Int("sftp.port", 2022, "SFTP server listen port")
filerSftpOptions.idleTimeout = cmdFiler.Flag.Int("sftp.idleTimeout", 10, "connection idle seconds")
filerSftpOptions.sshPrivateKey = cmdFiler.Flag.String("sftp.sshPrivateKey", "", "path to the SSH private key file for host authentication")
filerSftpOptions.hostKeysFolder = cmdFiler.Flag.String("sftp.hostKeysFolder", "", "path to folder containing SSH private key files for host authentication")
filerSftpOptions.authMethods = cmdFiler.Flag.String("sftp.authMethods", "password,publickey", "comma-separated list of allowed auth methods: password, publickey, certificate")
filerSftpOptions.maxAuthTries = cmdFiler.Flag.Int("sftp.maxAuthTries", 6, "maximum number of authentication attempts per connection")
filerSftpOptions.bannerMessage = cmdFiler.Flag.String("sftp.bannerMessage", "SeaweedFS SFTP Server - Unauthorized access is prohibited", "message displayed before authentication")
filerSftpOptions.loginGraceTime = cmdFiler.Flag.Duration("sftp.loginGraceTime", 2*time.Minute, "timeout for authentication")
filerSftpOptions.clientAliveInterval = cmdFiler.Flag.Duration("sftp.clientAliveInterval", 5*time.Second, "interval for sending keep-alive messages")
filerSftpOptions.clientAliveCountMax = cmdFiler.Flag.Int("sftp.clientAliveCountMax", 3, "maximum number of missed keep-alive messages before disconnecting")
filerSftpOptions.userStoreFile = cmdFiler.Flag.String("sftp.userStoreFile", "", "path to JSON file containing user credentials and permissions")
filerSftpOptions.trustedUserCAKeysFile = cmdFiler.Flag.String("sftp.trustedUserCAKeysFile", "", "path to a file with trusted user CA public keys (OpenSSH authorized_keys format); required when 'certificate' is in -sftp.authMethods")
filerSftpOptions.dataCenter = cmdFiler.Flag.String("sftp.dataCenter", "", "prefer to read and write to volumes in this data center")
filerSftpOptions.bindIp = cmdFiler.Flag.String("sftp.ip.bind", "", "ip address to bind to. If empty, default to same as -ip.bind option.")
filerSftpOptions.localSocket = cmdFiler.Flag.String("sftp.localSocket", "", "default to /tmp/seaweedfs-sftp-<port>.sock")
}
func filerLongDesc() string {
desc := `start a file server which accepts REST operation for any files.
//create or overwrite the file, the directories /path/to will be automatically created
POST /path/to/file
//get the file content
GET /path/to/file
//create or overwrite the file, the filename in the multipart request will be used
POST /path/to/
//return a json format subdirectory and files listing
GET /path/to/
The configuration file "filer.toml" is read from ".", "$HOME/.seaweedfs/", "/usr/local/etc/seaweedfs/", or "/etc/seaweedfs/", in that order.
If the "filer.toml" is not found, an embedded filer store will be created under "-defaultStoreDir".
The example filer.toml configuration file can be generated by "weed scaffold -config=filer"
Supported Filer Stores:
`
storeNames := make([]string, len(filer.Stores))
for i, store := range filer.Stores {
storeNames[i] = "\t" + store.GetName()
}
sort.Strings(storeNames)
storeList := strings.Join(storeNames, "\n")
return desc + storeList
}
var cmdFiler = &Command{
UsageLine: "filer -port=8888 -master=<ip:port>[,<ip:port>]*",
Short: "start a file server that points to a master server, or a list of master servers",
Long: filerLongDesc(),
}
func runFiler(cmd *Command, args []string) bool {
if *f.debug {
go http.ListenAndServe(fmt.Sprintf(":%d", *f.debugPort), nil)
}
*f.defaultLevelDbDirectory = util.ResolvePath(*f.defaultLevelDbDirectory)
filerS3Options.resolvePaths()
filerWebDavOptions.resolvePaths()
filerSftpOptions.resolvePaths()
util.LoadSecurityConfiguration()
// Share the S3 static identity config file with the filer regardless of
// whether the embedded S3 gateway runs on this node: the IAM gRPC service
// the admin UI and weed shell talk to is wired up unconditionally, and it
// needs the same identities the S3 server would load from -s3.config.
f.s3ConfigFile = filerS3Options.config
switch {
case *f.metricsHttpIp != "":
// noting to do, use f.metricsHttpIp
case *f.bindIp != "":
*f.metricsHttpIp = *f.bindIp
case *f.ip != "":
*f.metricsHttpIp = *f.ip
}
go stats_collect.StartMetricsServer(*f.metricsHttpIp, *f.metricsHttpPort)
// Embedded services keep their loopback companion listener only while
// -ip.bind is unset and the bind address falls back to -ip. This must be
// captured before startFiler fills *f.bindIp in place.
filerBindIpUnset := *f.bindIp == ""
filerAddress := pb.NewServerAddress(*f.ip, *f.port, *f.portGrpc).String()
startDelay := time.Duration(2)
if *filerStartS3 {
filerS3Options.filer = &filerAddress
filerS3Options.ip = f.ip
if *filerS3Options.bindIp == "" {
filerS3Options.listenOnLocalhost = &filerBindIpUnset
filerS3Options.bindIp = f.bindIp
}
filerS3Options.localFilerSocket = f.localSocket
// Set S3 metrics IP based on bind IP if not explicitly set
if *filerS3Options.metricsHttpIp == "" {
*filerS3Options.metricsHttpIp = *filerS3Options.bindIp
}
go func(delay time.Duration) {
time.Sleep(delay * time.Second)
filerS3Options.startS3Server()
}(startDelay)
startDelay++
}
if *filerStartWebDav {
filerWebDavOptions.filer = &filerAddress
filerWebDavOptions.ipBind = f.bindIp
if *filerWebDavOptions.disk == "" {
filerWebDavOptions.disk = f.diskType
}
go func(delay time.Duration) {
time.Sleep(delay * time.Second)
filerWebDavOptions.startWebDav()
}(startDelay)
startDelay++
}
if *filerStartIam {
filerIamOptions.filer = &filerAddress
filerIamOptions.masters = f.mastersString
filerIamOptions.listenOnLocalhost = &filerBindIpUnset
go func(delay time.Duration) {
time.Sleep(delay * time.Second)
filerIamOptions.startIamServer()
}(startDelay)
startDelay++
}
if *filerStartSftp {
filerSftpOptions.filer = &filerAddress
if *filerSftpOptions.bindIp == "" {
filerSftpOptions.listenOnLocalhost = &filerBindIpUnset
filerSftpOptions.bindIp = f.bindIp
}
go func(delay time.Duration) {
time.Sleep(delay * time.Second)
filerSftpOptions.startSftpServer()
}(startDelay)
}
f.masters = pb.ServerAddresses(*f.mastersString).ToServiceDiscovery()
f.startFiler()
return true
}
func (fo *FilerOptions) startFiler() {
defaultMux := http.NewServeMux()
publicVolumeMux := defaultMux
if *fo.publicPort != 0 {
publicVolumeMux = http.NewServeMux()
}
if *fo.portGrpc == 0 {
*fo.portGrpc = 10000 + *fo.port
}
// The loopback companion listener is only kept while -ip.bind is unset
// and the bind address falls back to -ip. An explicit -ip.bind pins each
// port to exactly that address.
listenOnLocalhost := *fo.bindIp == ""
if fo.listenOnLocalhost != nil {
listenOnLocalhost = *fo.listenOnLocalhost
}
idleTimeout := idleTimeoutSeconds(fo.idleTimeout)
if *fo.bindIp == "" {
*fo.bindIp = *fo.ip
}
util.SetOutboundLocalIP(*fo.bindIp)
if *fo.allowedOrigins == "" {
*fo.allowedOrigins = "*"
}
defaultLevelDbDirectory := *fo.defaultLevelDbDirectory + "/filerldb2"
filerAddress := pb.NewServerAddress(*fo.ip, *fo.port, *fo.portGrpc)
// Initialize credential manager for IAM gRPC service
var credentialManager *credential.CredentialManager
var err error
credentialManager, err = credential.NewCredentialManagerWithDefaults("")
if err != nil {
glog.Warningf("Failed to initialize credential manager: %v", err)
} else {
glog.V(0).Infof("Initialized credential manager: %s", credentialManager.GetStoreName())
}
// Load static S3 identities from config file if specified
if fo.s3ConfigFile != nil && *fo.s3ConfigFile != "" {
if credentialManager != nil {
if err := credentialManager.LoadS3ConfigFile(*fo.s3ConfigFile); err != nil {
glog.Warningf("Failed to load S3 config file for static identities: %v", err)
}
}
}
announce := make(chan struct{})
fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{
Masters: fo.masters,
AnnounceCh: announce,
SkipEmptyMembershipConfirm: fo.skipEmptyMembershipConfirm,
ClusterFreshHint: fo.clusterFreshHint,
FilerGroup: *fo.filerGroup,
Collection: *fo.collection,
DefaultReplication: *fo.defaultReplicaPlacement,
DisableDirListing: *fo.disableDirListing,
MaxMB: *fo.maxMB,
DirListingLimit: *fo.dirListingLimit,
DataCenter: *fo.dataCenter,
Rack: *fo.rack,
DefaultLevelDbDir: defaultLevelDbDirectory,
DisableHttp: *fo.disableHttp,
Host: filerAddress,
Cipher: *fo.cipher,
SaveToFilerLimit: int64(*fo.saveToFilerLimit),
ConcurrentUploadLimit: int64(*fo.concurrentUploadLimitMB) * 1024 * 1024,
ConcurrentFileUploadLimit: int64(*fo.concurrentFileUploadLimit),
ShowUIDirectoryDelete: *fo.showUIDirectoryDelete,
DownloadMaxBytesPs: int64(*fo.downloadMaxMBps) * 1024 * 1024,
DiskType: *fo.diskType,
AllowedOrigins: strings.Split(*fo.allowedOrigins, ","),
ExposeDirectoryData: *fo.exposeDirectoryData,
TusBasePath: *fo.tusBasePath,
TusMaxSize: int64(*fo.tusMaxSizeMB) * 1024 * 1024,
TusSessionExpiry: *fo.tusSessionExpiry,
CredentialManager: credentialManager,
AllowUntrustedRemoteEndpoints: *fo.allowUntrustedRemoteEndpoints,
RemoteCacheEvictThreshold: *fo.remoteCacheEvictThreshold,
})
if nfs_err != nil {
glog.Fatalf("Filer startup error: %v", nfs_err)
}
// Serve "//" and ".." paths at their cleaned form instead of letting the mux
// redirect: its Location is double-escaped, turning non-ASCII names into
// percent-encoded directory names when a client follows it (#11125).
defaultHandler := weed_server.CleanPathHandler(defaultMux)
var httpServers []*http.Server
if *fo.publicPort != 0 {
publicListeningAddress := util.JoinHostPort(*fo.bindIp, *fo.publicPort)
glog.V(0).Infoln("Start Seaweed filer server", version.Version(), "public at", publicListeningAddress)
publicListener, localPublicListener, e := util.NewIpAndLocalListeners(*fo.bindIp, *fo.publicPort, idleTimeout, listenOnLocalhost)
if e != nil {
glog.Fatalf("Filer server public listener error on port %d:%v", *fo.publicPort, e)
}
publicHandler := weed_server.CleanPathHandler(publicVolumeMux)
publicServer := newHttpServer(publicHandler, nil)
httpServers = append(httpServers, publicServer)
go func() {
if e := publicServer.Serve(publicListener); e != nil && e != http.ErrServerClosed {
glog.Fatalf("Volume server fail to serve public: %v", e)
}
}()
if localPublicListener != nil {
localPublicServer := newHttpServer(publicHandler, nil)
httpServers = append(httpServers, localPublicServer)
go func() {
if e := localPublicServer.Serve(localPublicListener); e != nil && e != http.ErrServerClosed {
glog.Errorf("Volume server fail to serve public: %v", e)
}
}()
}
}
glog.V(0).Infof("Start Seaweed Filer %s at %s:%d", version.Version(), *fo.ip, *fo.port)
filerListener, filerLocalListener, e := util.NewIpAndLocalListeners(
*fo.bindIp, *fo.port,
idleTimeout, listenOnLocalhost,
)
if e != nil {
glog.Fatalf("Filer listener error: %v", e)
}
// starting grpc server
grpcPort := *fo.portGrpc
grpcL, grpcLocalL, err := util.NewIpAndLocalListeners(*fo.bindIp, grpcPort, 0, listenOnLocalhost)
if err != nil {
glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
}
grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.filer"))
filer_pb.RegisterSeaweedFilerServer(grpcS, fs)
// Register the IAM gRPC service. Auth is opt-in: when
// jwt.filer_signing.key is configured the service requires a Bearer token
// signed with that key; otherwise it runs unauthenticated, matching the
// rest of the filer's gRPC surface. Operators who expose the filer gRPC
// port beyond a trusted network should set jwt.filer_signing.key on both
// the filer and the admin server.
if credentialManager != nil {
adminSigningKey := security.SigningKey(util.GetViper().GetString("jwt.filer_signing.key"))
iamGrpcServer := weed_server.NewIamGrpcServer(credentialManager, adminSigningKey)
// The OIDC provider and role RPCs write where S3 servers configured with
// filer-typed "oidcProviderStore" and "roleStore" read: this filer, at
// the stores' default base paths.
selfAddress := func() string { return string(filerAddress) }
if roleStore, err := integration.NewFilerRoleStore(nil, selfAddress); err != nil {
glog.Warningf("IAM gRPC: role RPCs disabled: %v", err)
} else {
iamGrpcServer.SetSTSStores(integration.NewFilerOIDCProviderStore(nil, selfAddress), roleStore)
}
iam_pb.RegisterSeaweedIdentityAccessManagementServer(grpcS, iamGrpcServer)
if len(adminSigningKey) == 0 {
glog.Warning("IAM gRPC service on filer is UNAUTHENTICATED: anyone who can reach this port can create users and policies, and its OIDC provider and role RPCs are refused; set jwt.filer_signing.key in security.toml to require an admin Bearer token")
} else {
glog.V(0).Info("Registered IAM gRPC service on filer (admin Bearer token required)")
}
}
reflection.Register(grpcS)
if grpcLocalL != nil {
go grpcS.Serve(grpcLocalL)
}
go grpcS.Serve(grpcL)
pb.ServeGrpcOnLocalSocket(grpcS, grpcPort)
// Membership in the master's filer list routes S3 and peer traffic here;
// announce once the gRPC port is serving and the metadata replay has
// caught up with the peers found at startup, so the filer does not take
// lock-ring ownership while a peer's older events can still land on top
// of its fresh writes. A cap bounds the wait.
go fs.AnnounceOnceCaughtUp(announce, time.Now())
// Helper to gracefully stop the gRPC server, waiting for active RPCs.
gracefulTimeout := fo.gracefulStopTimeout
if gracefulTimeout <= 0 {
gracefulTimeout = 15 * time.Second
}
stopGrpcServer := func() { gracefulStopGrpc(grpcS, gracefulTimeout) }
if runtime.GOOS != "windows" {
localSocket := *fo.localSocket
explicit := localSocket != ""
if !explicit {
localSocket = fmt.Sprintf("/tmp/seaweedfs-filer-%d.sock", *fo.port)
}
if filerSocketListener := pb.ListenOnLocalSocket(localSocket); filerSocketListener != nil {
filerSocketListener = &util.Listener{Listener: filerSocketListener, Timeout: idleTimeout}
socketServer := newHttpServer(defaultHandler, nil)
httpServers = append(httpServers, socketServer)
go socketServer.Serve(filerSocketListener)
}
}
if viper.GetString("https.filer.key") != "" {
certFile := viper.GetString("https.filer.cert")
keyFile := viper.GetString("https.filer.key")
caCertFile := viper.GetString("https.filer.ca")
disbaleTlsVerifyClientCert := viper.GetBool("https.filer.disable_tls_verify_client_cert")
getCert, certProvider, err := security.NewReloadingServerCertificate(certFile, keyFile)
if err != nil {
glog.Fatalf("Filer failed to load HTTPS certificate: %v", err)
}
defer certProvider.Close()
caCertPool := x509.NewCertPool()
if caCertFile != "" {
caCertFile, err := os.ReadFile(caCertFile)
if err != nil {
glog.Fatalf("error reading CA certificate: %v", err)
}
caCertPool.AppendCertsFromPEM(caCertFile)
}
clientAuth := tls.NoClientCert
if !disbaleTlsVerifyClientCert {
clientAuth = tls.RequireAndVerifyClientCert
}
tlsConfig := &tls.Config{
GetCertificate: getCert,
ClientAuth: clientAuth,
ClientCAs: caCertPool,
}
err = security.FixTlsConfig(util.GetViper(), tlsConfig)
if err != nil {
glog.Fatalf("Filer failed to fix TLS config: %v", err)
}
var localTLSServer *http.Server
if filerLocalListener != nil {
localTLSServer = newHttpServer(defaultHandler, tlsConfig)
httpServers = append(httpServers, localTLSServer)
go func() {
if err := localTLSServer.ServeTLS(filerLocalListener, "", ""); err != nil {
glog.Errorf("Filer Fail to serve: %v", err)
}
}()
}
httpS := newHttpServer(defaultHandler, tlsConfig)
httpServers = append(httpServers, httpS)
shutdown := newFilerShutdown(fs.LeaveLockRing, fs.StopSubscriptions, stopGrpcServer, fs.Shutdown, httpServers...)
grace.OnInterrupt(shutdown)
if fo.shutdownCtx != nil {
go func() {
<-fo.shutdownCtx.Done()
shutdown()
}()
}
if err := httpS.ServeTLS(filerListener, "", ""); err != nil && err != http.ErrServerClosed {
glog.Fatalf("Filer Fail to serve: %v", err)
}
// Serve returns when listeners close, before active requests finish.
// Join the same shutdown sequence instead of closing the filer here.
shutdown()
} else {
var localHTTPServer *http.Server
if filerLocalListener != nil {
localHTTPServer = newHttpServer(defaultHandler, nil)
httpServers = append(httpServers, localHTTPServer)
go func() {
if err := localHTTPServer.Serve(filerLocalListener); err != nil {
glog.Errorf("Filer Fail to serve: %v", err)
}
}()
}
httpS := newHttpServer(defaultHandler, nil)
httpServers = append(httpServers, httpS)
shutdown := newFilerShutdown(fs.LeaveLockRing, fs.StopSubscriptions, stopGrpcServer, fs.Shutdown, httpServers...)
grace.OnInterrupt(shutdown)
if fo.shutdownCtx != nil {
go func() {
<-fo.shutdownCtx.Done()
shutdown()
}()
}
if err := httpS.Serve(filerListener); err != nil && err != http.ErrServerClosed {
glog.Fatalf("Filer Fail to serve: %v", err)
}
// Serve returns when listeners close, before active requests finish.
// Join the same shutdown sequence instead of closing the filer here.
shutdown()
}
}
// newFilerShutdown joins shutdown callers while gRPC and HTTP drain concurrently.
// The filer leaves the lock ring first: peers and S3 gateways route keys to it
// until the ring changes, and would hit refused connections once gRPC stops.
// Then it ends its metadata subscriptions, so gRPC GracefulStop does not wait
// out its timeout on streams whose subscribers sit in this same process.
func newFilerShutdown(leaveLockRing, stopSubscriptions, stopGrpc, shutdownFiler func(), httpServers ...*http.Server) func() {
drain := newGracefulShutdown(stopGrpc, shutdownFiler, httpServers...)
return sync.OnceFunc(func() {
leaveLockRing()
stopSubscriptions()
drain()
})
}
Reference in new issue View Git Blame Copy Permalink