Add NATS publisher support to reminder #4829
Conversation
| _, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{ | ||
| Name: c.cfg.Prefix, | ||
| Subjects: []string{c.cfg.Prefix + ".>"}, | ||
| }) | ||
| return err |
There was a problem hiding this comment.
Newer jetstream API, cleaner code, context support. https://github.com/nats-io/nats.go/blob/main/jetstream/README.md
There was a problem hiding this comment.
👍 if this works; I think when I started looking, this wasn't cleaned up / operable with CloudEvents yet.
Vyom-Yadav
force-pushed
the
add-nats-publisher-to-reminder
branch
from
3d399bf to
748cb2c
Compare
There was a problem hiding this comment.
There's a separate fix going in for the one test failure.
In general, I think you could have separated the changes to internal/events/nats from the changes to reminder, which makes it easier to roll back or hold one change or the other if there needs to be discussion. But I'm willing to approve this and then iterate more going forward, modulo the comments, particularly about re-using eventer.New, rather than duplicating the setup code at this point.
| @@ -153,7 +150,7 @@ func instantiateDriver( | |||
| return eventersql.BuildPostgreSQLDriver(ctx, cfg) | |||
| case constants.NATSDriver: | |||
| zerolog.Ctx(ctx).Info().Msg("Using NATS driver") | |||
| return nats.BuildNatsChannelDriver(cfg) | |||
| return nats.BuildNatsChannelDriver(&cfg.Nats) | |||
There was a problem hiding this comment.
All the other constructors take the entire cfg, so that's the model I followed here.
There was a problem hiding this comment.
This has become a tedious problem to refactor. I have to change the signature of BuildNatsChannelDriver as cloudEventsNatsAdapter is not externally exposed.
Either I change the signature of every driver function to take in not serverconfig, but only the relevant config (in order to follow a pattern), which again isn't pretty as it does not truly achieve separation from serverconfig as functions like BuildPostgreSQLDriver would still rely on components purely in serverconfig (like cfg.SQLPubSub.AckDeadline) which isn't there in reminderconfig. Either it would result in breaking the pattern or some form of duplication of the instantiation code.
I see how this can slowly become just duplication of all instantiation code in order to keep minder and reminder separate, but I'm torn between keeping them separate (as minder is the main engine and reminder is a completely different service) v/s keeping them in the same code for ease of maintainability (and for de duplication - which again strikes the thought that this duplication would not have been there if reminder was in another language).
What do you think about it?
There was a problem hiding this comment.
I think it's fine to re-use the implementation of the construction code in reminder, even as it runs as a separate service. (For example, most Kubernetes controllers use the same informers library and construction client to create their connection to the K8s apiserver.)
So, my suggestion would be to embed a serverconfig.EventConfig into the ReminderConfig, rather than attempting to mirror only selected options into the reminder config. This will mean that it's possible to have something meaningless like the Go channel in reminder; you can avoid that by adding a check in the top-level ReminderConfig.Validate that the event configuration isn't using the go module.
There was a problem hiding this comment.
(This pattern will also give you access to the flag-driven eventer, and future improvements in flagging that I'm planning -- using a central flag server, rather than repeating the flag file for each binary.)
| func (r *reminder) getMessagePublisher(ctx context.Context) (message.Publisher, common.DriverCloser, error) { | ||
| switch r.cfg.EventConfig.Driver { | ||
| case constants.NATSDriver: | ||
| return r.setupNATSPublisher(ctx) | ||
| case constants.SQLDriver: | ||
| return r.setupSQLPublisher(ctx) | ||
| default: | ||
| return nil, nil, fmt.Errorf("unknown publisher type: %s", r.cfg.EventConfig.Driver) | ||
| } | ||
| } |
There was a problem hiding this comment.
I'm nervous about having this diverge from internal/events/events.go. Can we use an eventer.Interface from eventer.New() here?
(I need to do some refactoring in that package to move eventer/interface into eventer, but I'll wait until this code is landed to reduce the amount of conflicts.)
There was a problem hiding this comment.
I'm nervous about having this diverge from internal/events/events.go
internal/events/events.go is all about serverconfig, it would diverge from it as this is reminderconfig. I don't think we should use the interfaces.Interface:
// Interface is a combination of the Publisher, Registrar, and Service interfaces.
// This is handy when spawning the eventer in a single function for easy setup.
type Interface interface {
Publisher
Registrar
Service
}as reminder only needs publishing capability, I'd be happy to implement interfaces.Publisher for reminder.
There was a problem hiding this comment.
I think it's fine to use the interfaces.Publisher in the reminder code, but it feels like it might be easiest if the reminder code expected an interfaces.Publisher (which is satisfied by the current eventing implementations), but the construction code used NewEventer to construct the interfaces.Publisher.
|
|
||
| // NatsConfig is the configuration when using NATS as the event driver | ||
| type NatsConfig struct { | ||
| // URL is the URL for the NATS server | ||
| URL string `mapstructure:"url" default:"nats://localhost:4222"` | ||
| // Prefix is the prefix for the NATS subjects to subscribe to | ||
| Prefix string `mapstructure:"prefix" default:"minder"` | ||
| // Queue is the name of the queue group to join when consuming messages | ||
| // queue groups allow multiple process to round-robin process messages. | ||
| Queue string `mapstructure:"queue" default:"minder"` | ||
| } |
There was a problem hiding this comment.
If we want to move this up to common (rather than just having a split between client and server configuration), let's put it in its own file. It feels kind of peculiar to have this in common while having the rest of the event configuration still in server, so I'd argue for moving the code back, particularly in the context of having smaller, more focused PRs.
There was a problem hiding this comment.
If we want to move this up to common (rather than just having a split between client and server configuration), let's put it in its own file
That still breaks the pattern as all the structs in common don't have their own file.
so I'd argue for moving the code back,
To duplicate vs 'have common code'. Do you want me to duplicate the NatsConfig code in that case for reminderconfig
There was a problem hiding this comment.
I think there are a couple options here:
- The way that structs are all piled into
common.gotoday is probably a historical mistake. I'm not asking you fix that mistake in this PR (because we should be making small, targeted changes), but we shouldn't make that mistake worse. By putting this innatsoreventingfiles, it will make the future expansion more clear. - I think it's fine to have
pkg/config/reminderdepend onpkg/config/server, as long as we keep the graph acyclic. In particular, if we have common server-side configuration for e.g. flags and messaging, it feels like it would make sense to have those be shared between the two. Given that you can reasonably run aminderserver withoutreminder, but the opposite is not very useful, I'd argue for havingpkg/config/reminderimport parts of the config frompkg/config/server. In the future, if we split the Minder server into e.g.rpc_serverandevent_processor, I'd suggest that both would want to use parts of the currentServerConfig, but would probably each have their own top-level struct for configuration (possibly in a separate module, or possibly in thepkg/config/servermodule).
I'm more concerned about leaking the dependencies on e.g. NATS into the minder client, which is statically linked and downloaded by many more users than the server. Right now, this take that dependency, but I could see add a helper to create a NATS connection to this library in the future.
pkg/config/reminder/events.go
Outdated
There was a problem hiding this comment.
What do you think of moving the reminder configuration into the same package as server, and calling them all "server-side components`? It feels like it would allow for more sharing between the two implementations.
There was a problem hiding this comment.
My initial thoughts with reminder were that it'll be a separate micro service (remember when we had the talk of maybe let's write it in python). I agree with it all being server side components, but I'm not sure about moving that to server config (I wanted to keep the dependency tree separate, as they'd be packaged as separate binaries):
$ go list -f '{{range .Imports}}{{.}} # These are just imports and not deps
{{end}}' ./pkg/config/server
context
crypto/rsa
errors
fmt
github.com/go-playground/validator/v10
github.com/golang-jwt/jwt/v4
github.com/mindersec/minder/internal/auth/jwt
github.com/mindersec/minder/internal/util
github.com/mindersec/minder/pkg/config
github.com/rs/zerolog
github.com/rs/zerolog/log
github.com/spf13/pflag
github.com/spf13/viper
golang.org/x/oauth2
golang.org/x/oauth2/clientcredentials
io
net/http
net/url
os
path/filepath
strings
time
$ go list -f '{{range .Imports}}{{.}}
{{end}}' ./pkg/config/reminder
fmt
github.com/mindersec/minder/internal/util
github.com/mindersec/minder/pkg/config
github.com/rs/zerolog
github.com/spf13/pflag
github.com/spf13/viper
os
strings
time
It feels like it would allow for more sharing between the two implementations.
I agree with that. Currently, there is some duplication regarding how we manage components, which I try to tackle by moving things to a common place (which doesn't feel ideal in some situations). But, I'm of the opinion to try to keep things separate.
| _, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{ | ||
| Name: c.cfg.Prefix, | ||
| Subjects: []string{c.cfg.Prefix + ".>"}, | ||
| }) | ||
| return err |
There was a problem hiding this comment.
👍 if this works; I think when I started looking, this wasn't cleaned up / operable with CloudEvents yet.
|
This PR needs additional information before we can continue. It is now marked as stale because it has been open for 30 days with no activity. Please provide the necessary details to continue or it will be closed in 30 days. |
Signed-off-by: Vyom Yadav <[email protected]>
Signed-off-by: Vyom Yadav <[email protected]>
Vyom-Yadav
force-pushed
the
add-nats-publisher-to-reminder
branch
from
748cb2c to
8b9ebfb
Compare
|
Sorry for the delay on this @evankanderson. Please ignore the failing CI, that's not the main point. I'm confused between how to refactor this properly, because there isn't any easy and good way imo. In all approaches, we either have some duplication or break some pattern (unless we refactor a lot of code in serverconfig and internal/events/events.go). Three problems right now:
|
|
|
||
| // NatsConfig is the configuration when using NATS as the event driver | ||
| type NatsConfig struct { | ||
| // URL is the URL for the NATS server | ||
| URL string `mapstructure:"url" default:"nats://localhost:4222"` | ||
| // Prefix is the prefix for the NATS subjects to subscribe to | ||
| Prefix string `mapstructure:"prefix" default:"minder"` | ||
| // Queue is the name of the queue group to join when consuming messages | ||
| // queue groups allow multiple process to round-robin process messages. | ||
| Queue string `mapstructure:"queue" default:"minder"` | ||
| } |
There was a problem hiding this comment.
I think there are a couple options here:
- The way that structs are all piled into
common.gotoday is probably a historical mistake. I'm not asking you fix that mistake in this PR (because we should be making small, targeted changes), but we shouldn't make that mistake worse. By putting this innatsoreventingfiles, it will make the future expansion more clear. - I think it's fine to have
pkg/config/reminderdepend onpkg/config/server, as long as we keep the graph acyclic. In particular, if we have common server-side configuration for e.g. flags and messaging, it feels like it would make sense to have those be shared between the two. Given that you can reasonably run aminderserver withoutreminder, but the opposite is not very useful, I'd argue for havingpkg/config/reminderimport parts of the config frompkg/config/server. In the future, if we split the Minder server into e.g.rpc_serverandevent_processor, I'd suggest that both would want to use parts of the currentServerConfig, but would probably each have their own top-level struct for configuration (possibly in a separate module, or possibly in thepkg/config/servermodule).
I'm more concerned about leaking the dependencies on e.g. NATS into the minder client, which is statically linked and downloaded by many more users than the server. Right now, this take that dependency, but I could see add a helper to create a NATS connection to this library in the future.
| func (r *reminder) getMessagePublisher(ctx context.Context) (message.Publisher, common.DriverCloser, error) { | ||
| switch r.cfg.EventConfig.Driver { | ||
| case constants.NATSDriver: | ||
| return r.setupNATSPublisher(ctx) | ||
| case constants.SQLDriver: | ||
| return r.setupSQLPublisher(ctx) | ||
| default: | ||
| return nil, nil, fmt.Errorf("unknown publisher type: %s", r.cfg.EventConfig.Driver) | ||
| } | ||
| } |
There was a problem hiding this comment.
I think it's fine to use the interfaces.Publisher in the reminder code, but it feels like it might be easiest if the reminder code expected an interfaces.Publisher (which is satisfied by the current eventing implementations), but the construction code used NewEventer to construct the interfaces.Publisher.
| @@ -153,7 +150,7 @@ func instantiateDriver( | |||
| return eventersql.BuildPostgreSQLDriver(ctx, cfg) | |||
| case constants.NATSDriver: | |||
| zerolog.Ctx(ctx).Info().Msg("Using NATS driver") | |||
| return nats.BuildNatsChannelDriver(cfg) | |||
| return nats.BuildNatsChannelDriver(&cfg.Nats) | |||
There was a problem hiding this comment.
I think it's fine to re-use the implementation of the construction code in reminder, even as it runs as a separate service. (For example, most Kubernetes controllers use the same informers library and construction client to create their connection to the K8s apiserver.)
So, my suggestion would be to embed a serverconfig.EventConfig into the ReminderConfig, rather than attempting to mirror only selected options into the reminder config. This will mean that it's possible to have something meaningless like the Go channel in reminder; you can avoid that by adding a check in the top-level ReminderConfig.Validate that the event configuration isn't using the go module.
| @@ -153,7 +150,7 @@ func instantiateDriver( | |||
| return eventersql.BuildPostgreSQLDriver(ctx, cfg) | |||
| case constants.NATSDriver: | |||
| zerolog.Ctx(ctx).Info().Msg("Using NATS driver") | |||
| return nats.BuildNatsChannelDriver(cfg) | |||
| return nats.BuildNatsChannelDriver(&cfg.Nats) | |||
There was a problem hiding this comment.
(This pattern will also give you access to the flag-driven eventer, and future improvements in flagging that I'm planning -- using a central flag server, rather than repeating the flag file for each binary.)
Summary
Provide a brief overview of the changes and the issue being addressed.
Explain the rationale and any background necessary for understanding the changes.
List dependencies required by this change, if any.
Add NATS as an option for
reminderto publish events.Change Type
Mark the type of change your PR introduces:
Testing
Outline how the changes were tested, including steps to reproduce and any relevant configurations.
Attach screenshots if helpful.
Tested locally, working fine (
nats stream viewoutput):Review Checklist: