mirror of
https://github.com/hanzoai/s3.git
synced 2026-11-03 03:07:41 +00:00
zooqueen
dd74d3e6d6
hanzoai/s3 was an AGPL-3.0 MinIO fork whose network-copyleft blocks embedding
and reselling as a managed service. Replace it wholesale with a fork of
SeaweedFS (github.com/seaweedfs/seaweedfs, Apache-2.0, Go) which ships a
built-in S3 gateway + Filer.
- License: AGPL-3.0 -> Apache-2.0 (upstream LICENSE kept verbatim, incl.
"Copyright 2025 Chris Lu"); NOTICE added crediting SeaweedFS. No AGPL/BUSL/
SSPL/GPL or enterprise ("ee") files remain.
- Module: go.mod path set to github.com/hanzoai/s3 with a self-replace
(replace github.com/seaweedfs/seaweedfs => ./) so the ~1800 internal Go files
keep their upstream import paths with zero churn. Internal identifiers and
import paths are intentionally NOT renamed.
- CLI rebrand: the binary is built as `s3` (was `weed`); root usage, per-command
help/usage/flag strings, shell autocomplete program name, and the product name
in CLI output now read "Hanzo S3" / `s3`. Users never see `weed`.
(e.g. `s3 server`, `s3 filer`, `s3 master`, `s3 volume`, `s3 s3`, `s3 mount`.)
- Build: `make` (or `go build -o s3 ./weed`) builds the `s3` binary green;
`cmd/weed-db` and `cmd/weed-sql` also build; `go vet ./weed/command/` clean.
- README/LLM.md rewritten for Hanzo S3; AGENTS.md/CLAUDE.md symlinks preserved.
- Dropped a 33MB prebuilt Mach-O `stress` binary upstream had committed; `.gitignore`
updated to ignore the `s3` build output.
107 lines
3.9 KiB
Go
107 lines
3.9 KiB
Go
package broker
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/logstore"
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/pub_balancer"
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
|
|
)
|
|
|
|
// AssignTopicPartitions Runs on the assigned broker, to execute the topic partition assignment
|
|
func (b *MessageQueueBroker) AssignTopicPartitions(c context.Context, request *mq_pb.AssignTopicPartitionsRequest) (*mq_pb.AssignTopicPartitionsResponse, error) {
|
|
ret := &mq_pb.AssignTopicPartitionsResponse{}
|
|
|
|
// drain existing topic partition subscriptions
|
|
for _, assignment := range request.BrokerPartitionAssignments {
|
|
t := topic.FromPbTopic(request.Topic)
|
|
partition := topic.FromPbPartition(assignment.Partition)
|
|
b.accessLock.Lock()
|
|
if request.IsDraining {
|
|
// TODO drain existing topic partition subscriptions
|
|
b.localTopicManager.RemoveLocalPartition(t, partition)
|
|
} else {
|
|
var localPartition *topic.LocalPartition
|
|
if localPartition = b.localTopicManager.GetLocalPartition(t, partition); localPartition == nil {
|
|
localPartition = topic.NewLocalPartition(partition, b.option.LogFlushInterval, b.genLogFlushFunc(t, partition), logstore.GenMergedReadFunc(b, t, partition))
|
|
|
|
// Initialize offset from existing data to ensure continuity on restart
|
|
b.initializePartitionOffsetFromExistingData(localPartition, t, partition)
|
|
|
|
b.localTopicManager.AddLocalPartition(t, localPartition)
|
|
} else {
|
|
}
|
|
}
|
|
b.accessLock.Unlock()
|
|
}
|
|
|
|
// if is leader, notify the followers to drain existing topic partition subscriptions
|
|
if request.IsLeader {
|
|
for _, brokerPartition := range request.BrokerPartitionAssignments {
|
|
if follower := brokerPartition.FollowerBroker; follower != "" {
|
|
err := pb.WithBrokerGrpcClient(false, follower, b.grpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
|
|
_, err := client.AssignTopicPartitions(context.Background(), request)
|
|
return err
|
|
})
|
|
if err != nil {
|
|
return ret, err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return ret, nil
|
|
}
|
|
|
|
// called by broker leader to drain existing partitions.
|
|
// new/updated partitions will be detected by broker from the filer
|
|
func (b *MessageQueueBroker) assignTopicPartitionsToBrokers(ctx context.Context, t *schema_pb.Topic, assignments []*mq_pb.BrokerPartitionAssignment, isAdd bool) error {
|
|
// notify the brokers to create the topic partitions in parallel
|
|
var wg sync.WaitGroup
|
|
for _, bpa := range assignments {
|
|
wg.Add(1)
|
|
go func(bpa *mq_pb.BrokerPartitionAssignment) {
|
|
defer wg.Done()
|
|
if doCreateErr := b.withBrokerClient(false, pb.ServerAddress(bpa.LeaderBroker), func(client mq_pb.SeaweedMessagingClient) error {
|
|
_, doCreateErr := client.AssignTopicPartitions(ctx, &mq_pb.AssignTopicPartitionsRequest{
|
|
Topic: t,
|
|
BrokerPartitionAssignments: []*mq_pb.BrokerPartitionAssignment{
|
|
{
|
|
Partition: bpa.Partition,
|
|
},
|
|
},
|
|
IsLeader: true,
|
|
IsDraining: !isAdd,
|
|
})
|
|
if doCreateErr != nil {
|
|
if !isAdd {
|
|
return fmt.Errorf("drain topic %s %v on %s: %v", t, bpa.LeaderBroker, bpa.Partition, doCreateErr)
|
|
} else {
|
|
return fmt.Errorf("create topic %s %v on %s: %v", t, bpa.LeaderBroker, bpa.Partition, doCreateErr)
|
|
}
|
|
}
|
|
brokerStats, found := b.PubBalancer.Brokers.Get(bpa.LeaderBroker)
|
|
if !found {
|
|
brokerStats = pub_balancer.NewBrokerStats()
|
|
if !b.PubBalancer.Brokers.SetIfAbsent(bpa.LeaderBroker, brokerStats) {
|
|
brokerStats, _ = b.PubBalancer.Brokers.Get(bpa.LeaderBroker)
|
|
}
|
|
}
|
|
brokerStats.RegisterAssignment(t, bpa.Partition, isAdd)
|
|
return nil
|
|
}); doCreateErr != nil {
|
|
glog.Errorf("create topic %s partition %+v on %s: %v", t, bpa.Partition, bpa.LeaderBroker, doCreateErr)
|
|
}
|
|
}(bpa)
|
|
}
|
|
wg.Wait()
|
|
|
|
return nil
|
|
}
|