aboutsummaryrefslogtreecommitdiff
path: root/weed/command
diff options
context:
space:
mode:
Diffstat (limited to 'weed/command')
-rw-r--r--weed/command/command.go1
-rw-r--r--weed/command/fix.go12
-rw-r--r--weed/command/mq_broker.go39
-rw-r--r--weed/command/mq_kafka_gateway.go143
-rw-r--r--weed/command/scaffold/security.toml5
-rw-r--r--weed/command/server.go1
-rw-r--r--weed/command/sql.go3
7 files changed, 194 insertions, 10 deletions
diff --git a/weed/command/command.go b/weed/command/command.go
index b1c8df5b7..e4695a199 100644
--- a/weed/command/command.go
+++ b/weed/command/command.go
@@ -35,6 +35,7 @@ var Commands = []*Command{
cmdMount,
cmdMqAgent,
cmdMqBroker,
+ cmdMqKafkaGateway,
cmdDB,
cmdS3,
cmdScaffold,
diff --git a/weed/command/fix.go b/weed/command/fix.go
index 2b7b425f3..34dee3732 100644
--- a/weed/command/fix.go
+++ b/weed/command/fix.go
@@ -162,6 +162,18 @@ func doFixOneVolume(basepath string, baseFileName string, collection string, vol
defer nm.Close()
defer nmDeleted.Close()
+ // Validate volumeId range before converting to uint32
+ if volumeId < 0 || volumeId > 0xFFFFFFFF {
+ err := fmt.Errorf("volume ID out of range: %d", volumeId)
+ if *fixIgnoreError {
+ glog.Error(err)
+ return
+ } else {
+ glog.Fatal(err)
+ }
+ }
+ // lgtm[go/incorrect-integer-conversion]
+ // Safe conversion: volumeId has been validated to be in range [0, 0xFFFFFFFF] above
vid := needle.VolumeId(volumeId)
scanner := &VolumeFileScanner4Fix{
nm: nm,
diff --git a/weed/command/mq_broker.go b/weed/command/mq_broker.go
index ac7deac2c..8ea7f96a4 100644
--- a/weed/command/mq_broker.go
+++ b/weed/command/mq_broker.go
@@ -1,6 +1,10 @@
package command
import (
+ "fmt"
+ "net/http"
+ _ "net/http/pprof"
+
"google.golang.org/grpc/reflection"
"github.com/seaweedfs/seaweedfs/weed/util/grace"
@@ -18,15 +22,17 @@ var (
)
type MessageQueueBrokerOptions struct {
- masters map[string]pb.ServerAddress
- mastersString *string
- filerGroup *string
- ip *string
- port *int
- dataCenter *string
- rack *string
- cpuprofile *string
- memprofile *string
+ masters map[string]pb.ServerAddress
+ mastersString *string
+ filerGroup *string
+ ip *string
+ port *int
+ pprofPort *int
+ dataCenter *string
+ rack *string
+ cpuprofile *string
+ memprofile *string
+ logFlushInterval *int
}
func init() {
@@ -35,10 +41,12 @@ func init() {
mqBrokerStandaloneOptions.filerGroup = cmdMqBroker.Flag.String("filerGroup", "", "share metadata with other filers in the same filerGroup")
mqBrokerStandaloneOptions.ip = cmdMqBroker.Flag.String("ip", util.DetectedHostAddress(), "broker host address")
mqBrokerStandaloneOptions.port = cmdMqBroker.Flag.Int("port", 17777, "broker gRPC listen port")
+ mqBrokerStandaloneOptions.pprofPort = cmdMqBroker.Flag.Int("port.pprof", 0, "HTTP profiling port (0 to disable)")
mqBrokerStandaloneOptions.dataCenter = cmdMqBroker.Flag.String("dataCenter", "", "prefer to read and write to volumes in this data center")
mqBrokerStandaloneOptions.rack = cmdMqBroker.Flag.String("rack", "", "prefer to write to volumes in this rack")
mqBrokerStandaloneOptions.cpuprofile = cmdMqBroker.Flag.String("cpuprofile", "", "cpu profile output file")
mqBrokerStandaloneOptions.memprofile = cmdMqBroker.Flag.String("memprofile", "", "memory profile output file")
+ mqBrokerStandaloneOptions.logFlushInterval = cmdMqBroker.Flag.Int("logFlushInterval", 5, "log buffer flush interval in seconds")
}
var cmdMqBroker = &Command{
@@ -77,6 +85,7 @@ func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool {
MaxMB: 0,
Ip: *mqBrokerOpt.ip,
Port: *mqBrokerOpt.port,
+ LogFlushInterval: *mqBrokerOpt.logFlushInterval,
}, grpcDialOption)
if err != nil {
glog.Fatalf("failed to create new message broker for queue server: %v", err)
@@ -106,6 +115,18 @@ func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool {
}()
}
+ // Start HTTP profiling server if enabled
+ if mqBrokerOpt.pprofPort != nil && *mqBrokerOpt.pprofPort > 0 {
+ go func() {
+ pprofAddr := fmt.Sprintf(":%d", *mqBrokerOpt.pprofPort)
+ glog.V(0).Infof("MQ Broker pprof server listening on %s", pprofAddr)
+ glog.V(0).Infof("Access profiling at: http://localhost:%d/debug/pprof/", *mqBrokerOpt.pprofPort)
+ if err := http.ListenAndServe(pprofAddr, nil); err != nil {
+ glog.Errorf("pprof server error: %v", err)
+ }
+ }()
+ }
+
glog.V(0).Infof("MQ Broker listening on %s:%d", *mqBrokerOpt.ip, *mqBrokerOpt.port)
grpcS.Serve(grpcL)
diff --git a/weed/command/mq_kafka_gateway.go b/weed/command/mq_kafka_gateway.go
new file mode 100644
index 000000000..614f03e9c
--- /dev/null
+++ b/weed/command/mq_kafka_gateway.go
@@ -0,0 +1,143 @@
+package command
+
+import (
+ "fmt"
+ "net/http"
+ _ "net/http/pprof"
+ "os"
+
+ "github.com/seaweedfs/seaweedfs/weed/glog"
+ "github.com/seaweedfs/seaweedfs/weed/mq/kafka/gateway"
+ "github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+var (
+ mqKafkaGatewayOptions mqKafkaGatewayOpts
+)
+
+type mqKafkaGatewayOpts struct {
+ ip *string
+ ipBind *string
+ port *int
+ pprofPort *int
+ master *string
+ filerGroup *string
+ schemaRegistryURL *string
+ defaultPartitions *int
+}
+
+func init() {
+ cmdMqKafkaGateway.Run = runMqKafkaGateway
+ mqKafkaGatewayOptions.ip = cmdMqKafkaGateway.Flag.String("ip", util.DetectedHostAddress(), "Kafka gateway advertised host address")
+ mqKafkaGatewayOptions.ipBind = cmdMqKafkaGateway.Flag.String("ip.bind", "", "Kafka gateway bind address (default: same as -ip)")
+ mqKafkaGatewayOptions.port = cmdMqKafkaGateway.Flag.Int("port", 9092, "Kafka gateway listen port")
+ mqKafkaGatewayOptions.pprofPort = cmdMqKafkaGateway.Flag.Int("port.pprof", 0, "HTTP profiling port (0 to disable)")
+ mqKafkaGatewayOptions.master = cmdMqKafkaGateway.Flag.String("master", "localhost:9333", "comma-separated SeaweedFS master servers")
+ mqKafkaGatewayOptions.filerGroup = cmdMqKafkaGateway.Flag.String("filerGroup", "", "filer group name")
+ mqKafkaGatewayOptions.schemaRegistryURL = cmdMqKafkaGateway.Flag.String("schema-registry-url", "", "Schema Registry URL (required for schema management)")
+ mqKafkaGatewayOptions.defaultPartitions = cmdMqKafkaGateway.Flag.Int("default-partitions", 4, "Default number of partitions for auto-created topics")
+}
+
+var cmdMqKafkaGateway = &Command{
+ UsageLine: "mq.kafka.gateway [-ip=<host>] [-ip.bind=<bind_addr>] [-port=9092] [-master=<master_servers>] [-filerGroup=<group>] [-default-partitions=4] -schema-registry-url=<url>",
+ Short: "start a Kafka wire-protocol gateway for SeaweedMQ with schema management",
+ Long: `Start a Kafka wire-protocol gateway translating Kafka client requests to SeaweedMQ.
+
+Connects to SeaweedFS master servers to discover available brokers and integrates with
+Schema Registry for schema-aware topic management.
+
+Options:
+ -ip Advertised host address that clients should connect to (default: auto-detected)
+ -ip.bind Bind address for the gateway to listen on (default: same as -ip)
+ Use 0.0.0.0 to bind to all interfaces while advertising specific IP
+ -port Listen port (default: 9092)
+ -default-partitions Default number of partitions for auto-created topics (default: 4)
+ -schema-registry-url Schema Registry URL (REQUIRED for schema management)
+
+Examples:
+ weed mq.kafka.gateway -port=9092 -master=localhost:9333 -schema-registry-url=http://localhost:8081
+ weed mq.kafka.gateway -ip=gateway1 -port=9092 -master=master1:9333,master2:9333 -schema-registry-url=http://schema-registry:8081
+ weed mq.kafka.gateway -ip=external.host.com -ip.bind=0.0.0.0 -master=localhost:9333 -schema-registry-url=http://schema-registry:8081
+
+This is experimental and currently supports a minimal subset for development.
+`,
+}
+
+func runMqKafkaGateway(cmd *Command, args []string) bool {
+ // Validate required options
+ if *mqKafkaGatewayOptions.master == "" {
+ glog.Fatalf("SeaweedFS master address is required (-master)")
+ return false
+ }
+
+ // Schema Registry URL is required for schema management
+ if *mqKafkaGatewayOptions.schemaRegistryURL == "" {
+ glog.Fatalf("Schema Registry URL is required (-schema-registry-url)")
+ return false
+ }
+
+ // Determine bind address - default to advertised IP if not specified
+ bindIP := *mqKafkaGatewayOptions.ipBind
+ if bindIP == "" {
+ bindIP = *mqKafkaGatewayOptions.ip
+ }
+
+ // Construct listen address from bind IP and port
+ listenAddr := fmt.Sprintf("%s:%d", bindIP, *mqKafkaGatewayOptions.port)
+
+ // Set advertised host for Kafka protocol handler
+ if err := os.Setenv("KAFKA_ADVERTISED_HOST", *mqKafkaGatewayOptions.ip); err != nil {
+ glog.Warningf("Failed to set KAFKA_ADVERTISED_HOST environment variable: %v", err)
+ }
+
+ srv := gateway.NewServer(gateway.Options{
+ Listen: listenAddr,
+ Masters: *mqKafkaGatewayOptions.master,
+ FilerGroup: *mqKafkaGatewayOptions.filerGroup,
+ SchemaRegistryURL: *mqKafkaGatewayOptions.schemaRegistryURL,
+ DefaultPartitions: int32(*mqKafkaGatewayOptions.defaultPartitions),
+ })
+
+ glog.Warningf("EXPERIMENTAL FEATURE: MQ Kafka Gateway is experimental and should NOT be used in production environments. It currently supports only a minimal subset of Kafka protocol for development purposes.")
+
+ // Show bind vs advertised addresses for clarity
+ if bindIP != *mqKafkaGatewayOptions.ip {
+ glog.V(0).Infof("Starting MQ Kafka Gateway: binding to %s, advertising %s:%d to clients",
+ listenAddr, *mqKafkaGatewayOptions.ip, *mqKafkaGatewayOptions.port)
+ } else {
+ glog.V(0).Infof("Starting MQ Kafka Gateway on %s", listenAddr)
+ }
+ glog.V(0).Infof("Using SeaweedMQ brokers from masters: %s", *mqKafkaGatewayOptions.master)
+
+ // Start HTTP profiling server if enabled
+ if *mqKafkaGatewayOptions.pprofPort > 0 {
+ go func() {
+ pprofAddr := fmt.Sprintf(":%d", *mqKafkaGatewayOptions.pprofPort)
+ glog.V(0).Infof("Kafka Gateway pprof server listening on %s", pprofAddr)
+ glog.V(0).Infof("Access profiling at: http://localhost:%d/debug/pprof/", *mqKafkaGatewayOptions.pprofPort)
+ if err := http.ListenAndServe(pprofAddr, nil); err != nil {
+ glog.Errorf("pprof server error: %v", err)
+ }
+ }()
+ }
+
+ if err := srv.Start(); err != nil {
+ glog.Fatalf("mq kafka gateway start: %v", err)
+ return false
+ }
+
+ // Set up graceful shutdown
+ defer func() {
+ glog.V(0).Infof("Shutting down MQ Kafka Gateway...")
+ if err := srv.Close(); err != nil {
+ glog.Errorf("mq kafka gateway close: %v", err)
+ }
+ }()
+
+ // Serve blocks until closed
+ if err := srv.Wait(); err != nil {
+ glog.Errorf("mq kafka gateway wait: %v", err)
+ return false
+ }
+ return true
+}
diff --git a/weed/command/scaffold/security.toml b/weed/command/scaffold/security.toml
index bc95ecf2e..10f472d81 100644
--- a/weed/command/scaffold/security.toml
+++ b/weed/command/scaffold/security.toml
@@ -104,6 +104,11 @@ cert = ""
key = ""
allowed_commonNames = "" # comma-separated SSL certificate common names
+[grpc.mq]
+cert = ""
+key = ""
+allowed_commonNames = "" # comma-separated SSL certificate common names
+
# use this for any place needs a grpc client
# i.e., "weed backup|benchmark|filer.copy|filer.replicate|mount|s3|upload"
[grpc.client]
diff --git a/weed/command/server.go b/weed/command/server.go
index 0ad126dbb..8f7267d3e 100644
--- a/weed/command/server.go
+++ b/weed/command/server.go
@@ -192,6 +192,7 @@ func init() {
webdavOptions.filerRootPath = cmdServer.Flag.String("webdav.filer.path", "/", "use this remote path from filer server")
mqBrokerOptions.port = cmdServer.Flag.Int("mq.broker.port", 17777, "message queue broker gRPC listen port")
+ mqBrokerOptions.logFlushInterval = cmdServer.Flag.Int("mq.broker.logFlushInterval", 5, "log buffer flush interval in seconds")
mqAgentServerOptions.brokersString = cmdServer.Flag.String("mq.agent.brokers", "localhost:17777", "comma-separated message queue brokers")
mqAgentServerOptions.port = cmdServer.Flag.Int("mq.agent.port", 16777, "message queue agent gRPC listen port")
diff --git a/weed/command/sql.go b/weed/command/sql.go
index adc2ad52b..682c8e46d 100644
--- a/weed/command/sql.go
+++ b/weed/command/sql.go
@@ -408,7 +408,8 @@ func executeAndDisplay(ctx *SQLContext, query string, showTiming bool) bool {
}
// Show execution time for interactive/table mode
- if showTiming && ctx.outputFormat == OutputTable {
+ // Only show timing if there are columns or if result is truly empty
+ if showTiming && ctx.outputFormat == OutputTable && (len(result.Columns) > 0 || len(result.Rows) == 0) {
elapsed := time.Since(startTime)
fmt.Printf("\n(%d rows in set, %.3f sec)\n\n", len(result.Rows), elapsed.Seconds())
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 04:59:29 +0000