-
Notifications
You must be signed in to change notification settings - Fork 93
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Working, tested, but unused pubsub system (#1205)
## Which problem is this PR solving? - In order to implement a gossip protocol, we need a functional pubsub system. I tried the go-redis library instead of redigo and it was pretty simple to use. ## Short description of the changes - Create a standardized pubsub interface - Implement go-redis version of it - Implement local version of it - Write tests - Write benchmarks - Implement startstop - Set up based on our redis peer config (I have not used all of the pool size parameters, since they don't seem to matter anymore) On my local machine (without going through an external network), it runs at about 20K messages per second, with average latency of about 200uS and max latency of about 5mS. On CI, the first test had avg latency of 6mS. I also tried implementing a version using the reuidis library; see comments in the code. Basically, for pubsub it was no faster and the implementation was a bit ugly so I did not include it.
- Loading branch information
Showing
6 changed files
with
577 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package pubsub | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/facebookgo/startstop" | ||
| ) | ||
|
|
||
| // general usage: | ||
| // pubsub := pubsub.NewXXXPubSub() | ||
| // pubsub.Start() | ||
| // defer pubsub.Stop() | ||
| // ctx := context.Background() | ||
| // pubsub.Publish(ctx, "topic", "message") | ||
| // sub := pubsub.Subscribe(ctx, "topic") | ||
| // for msg := range sub.Channel() { | ||
| // fmt.Println(msg) | ||
| // } | ||
| // sub.Close() // optional | ||
| // pubsub.Close() | ||
|
|
||
| type PubSub interface { | ||
| // Publish sends a message to all subscribers of the specified topic. | ||
| Publish(ctx context.Context, topic, message string) error | ||
| // Subscribe returns a Subscription that will receive all messages published to the specified topic. | ||
| // There is no unsubscribe method; close the subscription to stop receiving messages. | ||
| Subscribe(ctx context.Context, topic string) Subscription | ||
| // Close shuts down all topics and the pubsub connection. | ||
| Close() | ||
|
|
||
| // we want to embed startstop.Starter and startstop.Stopper so that we | ||
| // can participate in injection | ||
| startstop.Starter | ||
| startstop.Stopper | ||
| } | ||
|
|
||
| type Subscription interface { | ||
| // Channel returns the channel that will receive all messages published to the topic. | ||
| Channel() <-chan string | ||
| // Close stops the subscription and closes the channel. Calling this is optional; | ||
| // the topic will be closed when the pubsub connection is closed. | ||
| Close() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| package pubsub | ||
|
|
||
| import ( | ||
| "context" | ||
| "strings" | ||
| "sync" | ||
|
|
||
| "github.com/honeycombio/refinery/config" | ||
| "github.com/honeycombio/refinery/logger" | ||
| "github.com/redis/go-redis/v9" | ||
| ) | ||
|
|
||
| // Notes for the future: we implemented a Redis-based PubSub system using 3 | ||
| // different libraries: go-redis, redigo, and rueidis. All three implementations | ||
| // perform similarly, but go-redis is definitely the easiest to use for PubSub. | ||
| // The rueidis library is probably the fastest for high-performance Redis use | ||
| // when you want Redis to be a database or cache, and it has some nice features | ||
| // like automatic pipelining, but it's pretty low-level and the documentation is | ||
| // poor. Redigo is feeling pretty old at this point. | ||
|
|
||
| // GoRedisPubSub is a PubSub implementation that uses Redis as the message broker | ||
| // and the go-redis library to interact with Redis. | ||
| type GoRedisPubSub struct { | ||
| Config config.Config `inject:""` | ||
| Logger logger.Logger `inject:""` | ||
| client redis.UniversalClient | ||
| subs []*GoRedisSubscription | ||
| mut sync.RWMutex | ||
| } | ||
|
|
||
| // Ensure that GoRedisPubSub implements PubSub | ||
| var _ PubSub = (*GoRedisPubSub)(nil) | ||
|
|
||
| type GoRedisSubscription struct { | ||
| topic string | ||
| pubsub *redis.PubSub | ||
| ch chan string | ||
| done chan struct{} | ||
| once sync.Once | ||
| } | ||
|
|
||
| // Ensure that GoRedisSubscription implements Subscription | ||
| var _ Subscription = (*GoRedisSubscription)(nil) | ||
|
|
||
| func (ps *GoRedisPubSub) Start() error { | ||
| options := &redis.UniversalOptions{} | ||
| authcode := "" | ||
|
|
||
| if ps.Config != nil { | ||
| host, err := ps.Config.GetRedisHost() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| username, err := ps.Config.GetRedisUsername() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| pw, err := ps.Config.GetRedisPassword() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| authcode, err = ps.Config.GetRedisAuthCode() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // we may have multiple hosts, separated by commas, so split them up and | ||
| // use them as the addrs for the client (if there are multiples, it will | ||
| // create a cluster client) | ||
| hosts := strings.Split(host, ",") | ||
| options.Addrs = hosts | ||
| options.Username = username | ||
| options.Password = pw | ||
| options.DB = ps.Config.GetRedisDatabase() | ||
| } | ||
| client := redis.NewUniversalClient(options) | ||
|
|
||
| // if an authcode was provided, use it to authenticate the connection | ||
| if authcode != "" { | ||
| pipe := client.Pipeline() | ||
| pipe.Auth(context.Background(), authcode) | ||
| if _, err := pipe.Exec(context.Background()); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| ps.client = client | ||
| ps.subs = make([]*GoRedisSubscription, 0) | ||
| return nil | ||
| } | ||
|
|
||
| func (ps *GoRedisPubSub) Stop() error { | ||
| ps.Close() | ||
| return nil | ||
| } | ||
|
|
||
| func (ps *GoRedisPubSub) Close() { | ||
| ps.mut.Lock() | ||
| for _, sub := range ps.subs { | ||
| sub.Close() | ||
| } | ||
| ps.subs = nil | ||
| ps.mut.Unlock() | ||
| ps.client.Close() | ||
| } | ||
|
|
||
| func (ps *GoRedisPubSub) Publish(ctx context.Context, topic, message string) error { | ||
| return ps.client.Publish(ctx, topic, message).Err() | ||
| } | ||
|
|
||
| func (ps *GoRedisPubSub) Subscribe(ctx context.Context, topic string) Subscription { | ||
| sub := &GoRedisSubscription{ | ||
| topic: topic, | ||
| pubsub: ps.client.Subscribe(ctx, topic), | ||
| ch: make(chan string, 100), | ||
| done: make(chan struct{}), | ||
| } | ||
| ps.mut.Lock() | ||
| ps.subs = append(ps.subs, sub) | ||
| ps.mut.Unlock() | ||
| go func() { | ||
| redisch := sub.pubsub.Channel() | ||
| for { | ||
| select { | ||
| case <-sub.done: | ||
| close(sub.ch) | ||
| return | ||
| case msg := <-redisch: | ||
| if msg == nil { | ||
| continue | ||
| } | ||
| select { | ||
| case sub.ch <- msg.Payload: | ||
| default: | ||
| ps.Logger.Warn().WithField("topic", topic).Logf("Dropping subscription message because channel is full") | ||
| } | ||
| } | ||
| } | ||
| }() | ||
| return sub | ||
| } | ||
|
|
||
| func (s *GoRedisSubscription) Channel() <-chan string { | ||
| return s.ch | ||
| } | ||
|
|
||
| func (s *GoRedisSubscription) Close() { | ||
| s.once.Do(func() { | ||
| s.pubsub.Close() | ||
| close(s.done) | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| package pubsub | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
|
|
||
| "github.com/honeycombio/refinery/config" | ||
| ) | ||
|
|
||
| // LocalPubSub is a PubSub implementation that uses local channels to send messages; it does | ||
| // not communicate with any external processes. | ||
| type LocalPubSub struct { | ||
| Config *config.Config `inject:""` | ||
| subs []*LocalSubscription | ||
| topics map[string]chan string | ||
| mut sync.RWMutex | ||
| } | ||
|
|
||
| // Ensure that LocalPubSub implements PubSub | ||
| var _ PubSub = (*LocalPubSub)(nil) | ||
|
|
||
| type LocalSubscription struct { | ||
| topic string | ||
| ch chan string | ||
| done chan struct{} | ||
| } | ||
|
|
||
| // Ensure that LocalSubscription implements Subscription | ||
| var _ Subscription = (*LocalSubscription)(nil) | ||
|
|
||
| // Start initializes the LocalPubSub | ||
| func (ps *LocalPubSub) Start() error { | ||
| ps.subs = make([]*LocalSubscription, 0) | ||
| ps.topics = make(map[string]chan string) | ||
| return nil | ||
| } | ||
|
|
||
| // Stop shuts down the LocalPubSub | ||
| func (ps *LocalPubSub) Stop() error { | ||
| ps.Close() | ||
| return nil | ||
| } | ||
|
|
||
| func (ps *LocalPubSub) Close() { | ||
| ps.mut.Lock() | ||
| defer ps.mut.Unlock() | ||
| for _, sub := range ps.subs { | ||
| sub.Close() | ||
| } | ||
| ps.subs = nil | ||
| } | ||
|
|
||
| func (ps *LocalPubSub) ensureTopic(topic string) chan string { | ||
| if _, ok := ps.topics[topic]; !ok { | ||
| ps.topics[topic] = make(chan string, 100) | ||
| } | ||
| return ps.topics[topic] | ||
| } | ||
|
|
||
| func (ps *LocalPubSub) Publish(ctx context.Context, topic, message string) error { | ||
| ps.mut.RLock() | ||
| ch := ps.ensureTopic(topic) | ||
| ps.mut.RUnlock() | ||
| select { | ||
| case ch <- message: | ||
| case <-ctx.Done(): | ||
| return ctx.Err() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (ps *LocalPubSub) Subscribe(ctx context.Context, topic string) Subscription { | ||
| ps.mut.Lock() | ||
| defer ps.mut.Unlock() | ||
| ch := ps.ensureTopic(topic) | ||
| sub := &LocalSubscription{ | ||
| topic: topic, | ||
| ch: ch, | ||
| done: make(chan struct{}), | ||
| } | ||
| ps.subs = append(ps.subs, sub) | ||
| go func() { | ||
| for { | ||
| select { | ||
| case <-sub.done: | ||
| close(ch) | ||
| return | ||
| case msg := <-ch: | ||
| sub.ch <- msg | ||
| } | ||
| } | ||
| }() | ||
| return sub | ||
| } | ||
|
|
||
| func (s *LocalSubscription) Channel() <-chan string { | ||
| return s.ch | ||
| } | ||
|
|
||
| func (s *LocalSubscription) Close() { | ||
| close(s.done) | ||
| } |
Oops, something went wrong.