From e6d8a339281b895fd8635c8daea04a4a908179d9 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Wed, 7 Aug 2024 13:51:53 +1000 Subject: [PATCH] experiment: Serf transport for go-ds-crdt --- experiment/broadcaster.go | 73 +++++++++++ experiment/explorer/explorer.go | 45 +++++++ experiment/logger.go | 73 +++++++++++ experiment/serf_crdt.go | 221 ++++++++++++++++++++++++++++++++ experiment/syncer.go | 97 ++++++++++++++ 5 files changed, 509 insertions(+) create mode 100644 experiment/broadcaster.go create mode 100644 experiment/explorer/explorer.go create mode 100644 experiment/logger.go create mode 100644 experiment/serf_crdt.go create mode 100644 experiment/syncer.go diff --git a/experiment/broadcaster.go b/experiment/broadcaster.go new file mode 100644 index 00000000..81acfa90 --- /dev/null +++ b/experiment/broadcaster.go @@ -0,0 +1,73 @@ +package main + +import ( + "context" + "fmt" + "github.com/hashicorp/serf/serf" + crdt "github.com/ipfs/go-ds-crdt" + "log/slog" + "time" +) + +// Implements the Broadcaster interface. +type SerfBroadcaster struct { + ctx context.Context + serf *serf.Serf + nextCh chan []byte +} + +// The broadcaster can be shut down by cancelling the given context. This must be done before Closing +// the crdt.Datastore, otherwise things may hang. +func NewSerfBroadcaster(ctx context.Context, serf *serf.Serf) *SerfBroadcaster { + return &SerfBroadcaster{ + ctx: ctx, + serf: serf, + nextCh: make(chan []byte), + } +} + +func (b *SerfBroadcaster) Broadcast(bytes []byte) error { + slog.Debug("Broadcasting head nodes to peers", "size", len(bytes)) + // Other peers are not allowed to coalesce this event by name as the payload may differ. + // TODO: decode CRDTBroadcast from bytes and embed the nodes content in the payload as well to save round trip. + if err := b.serf.UserEvent("heads", bytes, false); err != nil { + return fmt.Errorf("broadcast heads event: %w", err) + } + return nil +} + +func (b *SerfBroadcaster) Next() ([]byte, error) { + select { + case bytes := <-b.nextCh: // Blocks until a new heads event is received. + return bytes, nil + case <-b.ctx.Done(): + return nil, crdt.ErrNoMoreBroadcast + } +} + +func (b *SerfBroadcaster) HandleEvent(event serf.Event) { + select { + case <-b.ctx.Done(): + return + default: + } + + slog.Debug("Received event in broadcaster", "event", event) + userEvent, ok := event.(serf.UserEvent) + if !ok { + // Ignore non-user events. + return + } + if userEvent.Name != "heads" { + // Ignore non-heads user events. + return + } + start := time.Now() + select { + case b.nextCh <- userEvent.Payload: + slog.Debug("Handled heads event", "duration", time.Since(start)) + case <-b.ctx.Done(): + } +} + +var _ crdt.Broadcaster = (*SerfBroadcaster)(nil) diff --git a/experiment/explorer/explorer.go b/experiment/explorer/explorer.go new file mode 100644 index 00000000..3b8e85b6 --- /dev/null +++ b/experiment/explorer/explorer.go @@ -0,0 +1,45 @@ +package main + +import ( + "fmt" + "github.com/dgraph-io/badger/v3" + "log" + "time" +) + +func customTimeEncoder(t time.Time) string { + return t.Format("2006-01-02T15:04:05.000000Z07:00") +} + +func main() { + opts := badger.DefaultOptions("./store") + opts.Logger = nil // Disable the noisy Badger's default logger + + db, err := badger.Open(opts) + if err != nil { + log.Fatal(err) + } + defer db.Close() + + err = db.View(func(txn *badger.Txn) error { + opts := badger.DefaultIteratorOptions + opts.PrefetchSize = 10 + it := txn.NewIterator(opts) + defer it.Close() + for it.Rewind(); it.Valid(); it.Next() { + item := it.Item() + k := item.Key() + err := item.Value(func(v []byte) error { + fmt.Printf("key=%s, value=%s, raw value=%v\n", k, v, v) + return nil + }) + if err != nil { + return err + } + } + return nil + }) + if err != nil { + log.Fatal(err) + } +} diff --git a/experiment/logger.go b/experiment/logger.go new file mode 100644 index 00000000..9a8c6a3c --- /dev/null +++ b/experiment/logger.go @@ -0,0 +1,73 @@ +package main + +import ( + "fmt" + "github.com/ipfs/go-log/v2" + "log/slog" + "os" +) + +// ipfsLogger is an slog logger that implements the IPFS go-log StandardLogger interface. +type ipfsLogger struct { + log slog.Logger +} + +func newIPFSLogger(l *slog.Logger) *ipfsLogger { + return &ipfsLogger{log: *l} +} + +func (l *ipfsLogger) Debug(args ...interface{}) { + l.log.Debug(fmt.Sprint(args...)) +} + +func (l *ipfsLogger) Debugf(format string, args ...interface{}) { + l.log.Debug(fmt.Sprintf(format, args...)) +} + +func (l *ipfsLogger) Error(args ...interface{}) { + l.log.Error(fmt.Sprint(args...)) +} + +func (l *ipfsLogger) Errorf(format string, args ...interface{}) { + l.log.Error(fmt.Sprintf(format, args...)) +} + +func (l *ipfsLogger) Fatal(args ...interface{}) { + l.log.Error(fmt.Sprint(args...)) + os.Exit(1) +} + +func (l *ipfsLogger) Fatalf(format string, args ...interface{}) { + l.log.Error(fmt.Sprintf(format, args...)) + os.Exit(1) +} + +func (l *ipfsLogger) Info(args ...interface{}) { + l.log.Info(fmt.Sprint(args...)) +} + +func (l *ipfsLogger) Infof(format string, args ...interface{}) { + l.log.Info(fmt.Sprintf(format, args...)) +} + +func (l *ipfsLogger) Panic(args ...interface{}) { + msg := fmt.Sprint(args...) + l.log.Error(msg) + panic(msg) +} + +func (l *ipfsLogger) Panicf(format string, args ...interface{}) { + msg := fmt.Sprintf(format, args...) + l.log.Error(msg) + panic(msg) +} + +func (l *ipfsLogger) Warn(args ...interface{}) { + l.log.Warn(fmt.Sprint(args...)) +} + +func (l *ipfsLogger) Warnf(format string, args ...interface{}) { + l.log.Warn(fmt.Sprintf(format, args...)) +} + +var _ log.StandardLogger = (*ipfsLogger)(nil) diff --git a/experiment/serf_crdt.go b/experiment/serf_crdt.go new file mode 100644 index 00000000..d9eb5621 --- /dev/null +++ b/experiment/serf_crdt.go @@ -0,0 +1,221 @@ +package main + +import ( + "context" + "flag" + "fmt" + "github.com/hashicorp/memberlist" + "github.com/hashicorp/serf/cmd/serf/command/agent" + "github.com/hashicorp/serf/serf" + ds "github.com/ipfs/go-datastore" + badger "github.com/ipfs/go-ds-badger3" + crdt "github.com/ipfs/go-ds-crdt" + "github.com/lmittmann/tint" + "log/slog" + "net" + "os" + "os/signal" + "syscall" + "time" +) + +func createSerfAgentConfig(name, bindAddr, rpcAddr string) *agent.Config { + config := agent.DefaultConfig() + config.NodeName = name + config.BindAddr = bindAddr + config.RPCAddr = rpcAddr + + return config +} + +func createSerfAgent(config *agent.Config) (*agent.Agent, error) { + bindIP, bindPort, err := config.AddrParts(config.BindAddr) + if err != nil { + return nil, fmt.Errorf("invalid bind address: %w", err) + } + serfConfig := serf.DefaultConfig() + switch config.Profile { + case "lan": + serfConfig.MemberlistConfig = memberlist.DefaultLANConfig() + case "wan": + serfConfig.MemberlistConfig = memberlist.DefaultWANConfig() + case "local": + serfConfig.MemberlistConfig = memberlist.DefaultLocalConfig() + default: + return nil, fmt.Errorf("unknown profile: %s", config.Profile) + } + + serfConfig.MemberlistConfig.BindAddr = bindIP + serfConfig.MemberlistConfig.BindPort = bindPort + //serfConfig.MemberlistConfig.AdvertiseAddr = advertiseIP + //serfConfig.MemberlistConfig.AdvertisePort = advertisePort + //serfConfig.MemberlistConfig.SecretKey = encryptKey + serfConfig.NodeName = config.NodeName + serfConfig.Tags = config.Tags + serfConfig.SnapshotPath = config.SnapshotPath + serfConfig.ProtocolVersion = uint8(config.Protocol) + serfConfig.CoalescePeriod = 3 * time.Second + serfConfig.QuiescentPeriod = time.Second + serfConfig.QueryResponseSizeLimit = config.QueryResponseSizeLimit + serfConfig.QuerySizeLimit = config.QuerySizeLimit + serfConfig.UserEventSizeLimit = config.UserEventSizeLimit + serfConfig.UserCoalescePeriod = 3 * time.Second + serfConfig.UserQuiescentPeriod = time.Second + if config.ReconnectInterval != 0 { + serfConfig.ReconnectInterval = config.ReconnectInterval + } + if config.ReconnectTimeout != 0 { + serfConfig.ReconnectTimeout = config.ReconnectTimeout + } + if config.TombstoneTimeout != 0 { + serfConfig.TombstoneTimeout = config.TombstoneTimeout + } + serfConfig.EnableNameConflictResolution = !config.DisableNameResolution + if config.KeyringFile != "" { + serfConfig.KeyringFile = config.KeyringFile + } + serfConfig.RejoinAfterLeave = config.RejoinAfterLeave + if config.BroadcastTimeout != 0 { + serfConfig.BroadcastTimeout = config.BroadcastTimeout + } + + serfAgent, err := agent.Create(config, serfConfig, os.Stdout) + if err != nil { + return nil, fmt.Errorf("setup the Serf agent: %w", err) + } + return serfAgent, nil +} + +func startSerfAgent(config *agent.Config, serfAgent *agent.Agent) (*agent.AgentIPC, error) { + if err := serfAgent.Start(); err != nil { + return nil, fmt.Errorf("start Serf agent: %w", err) + } + + // Setup the RPC listener + rpcListener, err := net.Listen("tcp", config.RPCAddr) + if err != nil { + return nil, fmt.Errorf("start RPC listener: %w", err) + } + + // Start the IPC layer + logWriter := agent.NewLogWriter(512) + ipc := agent.NewAgentIPC(serfAgent, config.RPCAuthKey, rpcListener, os.Stdout, logWriter) + + slog.Info("Started Serf agent", + "name", config.NodeName, + "bind_addr", config.BindAddr, + "rpc_addr", config.RPCAddr, + "encrypted", serfAgent.Serf().EncryptionEnabled(), + "snapshot", config.SnapshotPath != "", + "profile", config.Profile, + "compression", config.EnableCompression, + ) + + return ipc, nil +} + +func main() { + name := flag.String("name", "", "Machine name") + bindAddr := flag.String("bind", "0.0.0.0:7946", "Bind address") + rpcAddr := flag.String("rpc-addr", "127.0.0.1:7373", "RPC address") + storeDir := flag.String("store-dir", "./store", "Store directory path") + runTick := flag.Bool("tick", false, "Periodically update /tick value") + flag.Parse() + + logger := slog.New(tint.NewHandler(os.Stdout, &tint.Options{ + AddSource: true, + Level: slog.LevelDebug, + TimeFormat: time.RFC3339Nano, + })) + slog.SetDefault(logger) + + config := createSerfAgentConfig(*name, *bindAddr, *rpcAddr) + serfAgent, err := createSerfAgent(config) + if err != nil { + panic(err) + } + ipc, err := startSerfAgent(config, serfAgent) + if err != nil { + panic(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + // Ideally, the broadcaster should be registered as an event handler before starting the agent. + // However, we need the agent.serf to be initialized which is done in agent.Start(). + broadcaster := NewSerfBroadcaster(ctx, serfAgent.Serf()) + serfAgent.RegisterEventHandler(broadcaster) + + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) + // A channel to signal that shutdown is done. + done := make(chan bool, 1) + + localStore, err := badger.NewDatastore(*storeDir, nil) // default options + if err != nil { + panic(err) + } + syncer := newDAGSyncer(localStore, ds.NewKey("/node")) + + opts := crdt.DefaultOptions() + opts.Logger = newIPFSLogger(logger) + store, err := crdt.New(localStore, ds.NewKey("/"), syncer, broadcaster, opts) + if err != nil { + panic(err) + } + + if *runTick { + ticker := time.NewTicker(10 * time.Second) + go func() { + for { + select { + case t := <-ticker.C: + err = store.Put(ctx, ds.NewKey("/tick"), []byte(t.String())) + if err != nil { + slog.Error("Put /tick", "error", err) + } + case <-ctx.Done(): + return + } + } + }() + } + + //err = store.Put(ctx, ds.NewKey("/test3"), []byte("hello3")) + //if err != nil { + // panic(err) + //} + v, err := store.Get(ctx, ds.NewKey("/test")) + if err != nil { + slog.Error("Get /test", "error", err) + } + slog.Info("Get /test", "value", string(v)) + + _ = store.PrintDAG() + + // Start a goroutine to handle signals. + go func() { + sig := <-sigs + slog.Info("Received signal, shutting down Serf agent", "signal", sig) + + cancel() + ipc.Shutdown() + if err = serfAgent.Leave(); err != nil { + slog.Error("Gracefully leave Serf cluster", "error", err) + } + if err = serfAgent.Shutdown(); err != nil { + slog.Error("Gracefully shutdown Serf agent", "error", err) + } + fmt.Println("closing crdt store") + if err = store.Close(); err != nil { + slog.Error("Close CRDT store", "error", err) + } + fmt.Println("closing local store") + if err = localStore.Close(); err != nil { + slog.Error("Close local store", "error", err) + } + + done <- true + }() + + <-done +} diff --git a/experiment/syncer.go b/experiment/syncer.go new file mode 100644 index 00000000..f685f510 --- /dev/null +++ b/experiment/syncer.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "errors" + "fmt" + "github.com/ipfs/boxo/datastore/dshelp" + dag "github.com/ipfs/boxo/ipld/merkledag" + "github.com/ipfs/go-cid" + ds "github.com/ipfs/go-datastore" + ipld "github.com/ipfs/go-ipld-format" + "log/slog" +) + +// Implements the DAGService interface. +// TODO: implement SessionDAGService to optimize node fetching. +// TOOD: persistentSerfDAG? +type dagSyncer struct { + // Persistent storage for the nodes. + store ds.Datastore + namespace ds.Key +} + +func newDAGSyncer(store ds.Datastore, namespace ds.Key) *dagSyncer { + return &dagSyncer{ + store: store, + namespace: namespace, + } +} + +func (d *dagSyncer) Get(ctx context.Context, cid cid.Cid) (ipld.Node, error) { + slog.Debug("Getting node", "cid", cid) + nodeBytes, err := d.store.Get(ctx, d.nodeKey(cid)) + if err != nil { + if errors.Is(err, ds.ErrNotFound) { + // TODO: try to retrieve the node from the peers. + return nil, ipld.ErrNotFound{Cid: cid} + } + return nil, fmt.Errorf("get node %s from local store: %w", cid, err) + } + protoNode, err := dag.DecodeProtobuf(nodeBytes) + if err != nil { + return nil, fmt.Errorf("decode node from protobuf: %w", err) + } + // CID is lazily computed from the node content. Ensure the node uses CIDv1. + if err = protoNode.SetCidBuilder(dag.V1CidPrefix()); err != nil { + return nil, fmt.Errorf("set CIDv1 on node: %w", err) + } + slog.Debug("Retrieved node from local store", "cid", protoNode.Cid()) + + return protoNode, nil +} + +func (d *dagSyncer) GetMany(ctx context.Context, cids []cid.Cid) <-chan *ipld.NodeOption { + //TODO implement me + panic("implement me") +} + +func (d *dagSyncer) Add(ctx context.Context, node ipld.Node) error { + slog.Debug("Adding node", "cid", node.Cid()) + protoNode, ok := node.(*dag.ProtoNode) + if !ok { + return fmt.Errorf("node is not a ProtoNode") + } + nodeBytes, err := protoNode.EncodeProtobuf(false) + if err != nil { + return fmt.Errorf("encode node to protobuf: %w", err) + } + + if err = d.store.Put(ctx, d.nodeKey(node.Cid()), nodeBytes); err != nil { + return fmt.Errorf("put node %s in local store: %w", node, err) + } + slog.Debug("Persisted node in local store", "cid", node.Cid(), "size", len(nodeBytes)) + + // TODO: Think about broadcasting the new node to the peers to not require each peer to query the node + // when a new CID is broadcasted. + + return nil +} + +func (d *dagSyncer) AddMany(ctx context.Context, nodes []ipld.Node) error { + panic("implement me") +} + +func (d *dagSyncer) Remove(ctx context.Context, cid cid.Cid) error { + panic("implement me") +} + +func (d *dagSyncer) RemoveMany(ctx context.Context, cids []cid.Cid) error { + panic("implement me") +} + +func (d *dagSyncer) nodeKey(cid cid.Cid) ds.Key { + return d.namespace.Child(dshelp.MultihashToDsKey(cid.Hash())) +} + +var _ ipld.DAGService = (*dagSyncer)(nil)