Real Time Configs with NATS

2025-11-05

One pattern I’ve used previously was setting up real time configs for services using NATS KV. A KV watcher would monitor the config value(s) and then in real time update values in a service. I guess you could think of this as service level feature flags. I had thought about writing a small Go package for this but it’s not enough to warrant a package so I figured I’d just write a small post about it.

type WatchFunc func(key string, val []byte) error
type ConfigWatcherOpts struct {
  Bucket        string
  Name          string
  Keys          map[string]WatchFunc
  Conn          *nats.Conn
  WatchInterval int64
}
type ConfigWatcher struct {
  kv            jetstream.KeyValue
  watcher       jetstream.KeyWatcher
  keys          map[string]WatchFunc
  watchInterval int64
}

type WatchFunc func(key string, val []byte) error
type ConfigWatcherOpts struct {
  Bucket        string
  Name          string
  Keys          map[string]WatchFunc
  Conn          *nats.Conn
  WatchInterval int64
}
type ConfigWatcher struct {
  kv            jetstream.KeyValue
  watcher       jetstream.KeyWatcher
  keys          map[string]WatchFunc
  watchInterval int64
}

Let’s setup a function to create our watcher and map everything we need to our Flag struct. I’m using the IgnoreDeletesIgnoreDeletes functional option here because I don’t necessarily care about a key being deleted, but there could be use cases where a service should terminate if the key with the config value disappears.

func New(opts ConfigWatcherOpts) (ConfigWatcher, error) {
  ctx := context.TODO()
  js, err := jetstream.New(opts.Conn)
  if err != nil {
    return ConfigWatcher{}, err
  }
  kv, err := js.KeyValue(ctx, opts.Bucket)
  if err != nil {
    return ConfigWatcher{}, err
  }
  w, err := kv.Watch(ctx, opts.Name, jetstream.IgnoreDeletes())
  if err != nil {
    return ConfigWatcher{}, err
  }
  return ConfigWatcher{
    watcher:       w,
    keys:          opts.Keys,
    watchInterval: opts.WatchInterval,
  }, nil
}

func New(opts ConfigWatcherOpts) (ConfigWatcher, error) {
  ctx := context.TODO()
  js, err := jetstream.New(opts.Conn)
  if err != nil {
    return ConfigWatcher{}, err
  }
  kv, err := js.KeyValue(ctx, opts.Bucket)
  if err != nil {
    return ConfigWatcher{}, err
  }
  w, err := kv.Watch(ctx, opts.Name, jetstream.IgnoreDeletes())
  if err != nil {
    return ConfigWatcher{}, err
  }
  return ConfigWatcher{
    watcher:       w,
    keys:          opts.Keys,
    watchInterval: opts.WatchInterval,
  }, nil
}

Now that we have our builder, let’s add a method to watch our configs. This doesn’t do much other than call Updates()Updates() on our watcher and range over them. Since watchers can accept NATS wildcard characters, this is where we filter the subjects we care about, something like services.serviceA.logger.levelservices.serviceA.logger.level. If the KeysKeys field contains a subject defined in our map, the key and value are passed to that WatchFuncWatchFunc where it can do something with that key/value.

We also take in a channel for errors that we can send in case one of our WatchFuncs encounter an error.

func (c *ConfigWatcher) Watch(errChan chan<- error) error {
  for val := range c.watcher.Updates() {
    if val == nil {
      continue
    }
    wf, ok := c.Keys[val.Key()]
    if !ok {
      continue
    }
    if err := wf(val.Key(), val.Value()); err != nil {
      errChan <- err
    }
  }
  time.Sleep(time.Duration(c.watchInterval) * time.Second)
  return nil
}

func (c *ConfigWatcher) Watch(errChan chan<- error) error {
  for val := range c.watcher.Updates() {
    if val == nil {
      continue
    }
    wf, ok := c.Keys[val.Key()]
    if !ok {
      continue
    }
    if err := wf(val.Key(), val.Value()); err != nil {
      errChan <- err
    }
  }
  time.Sleep(time.Duration(c.watchInterval) * time.Second)
  return nil
}

Let’s put together our watcher functions. There are two examples here. Both wrap an anonymous function that matches the WatchFuncWatchFunc type. One uses a logger that I wrote directly and the other uses an interface just to make things a bit easier if we decide to refactor things into their own packages later on.

The LogLevelLogLevel function will watch for changes to our log level key and apply changes to our logger real time. The FooFuncFooFunc is an example of setting a value in a struct in real time.

type ConfigSetter interface {
  Set(string, []byte) error
}
func LogLevel(logger *logr.Logger) configs.WatchFunc {
  return func(key string, value []byte) error {
    level := string(value)
    switch level {
    case "info":
      logger.Level = logr.InfoLevel
    case "error":
      logger.Level = logr.ErrorLevel
    case "debug":
      logger.Level = logr.DebugLevel
    }
    logger.Infof("set log level to %s ", string(value))
    return nil
  }
}
func FooFunc(c ConfigSetter) configs.WatchFunc {
  return func(key string, value []byte) error {
    return c.Set(key, value)
  }
}

type ConfigSetter interface {
  Set(string, []byte) error
}
func LogLevel(logger *logr.Logger) configs.WatchFunc {
  return func(key string, value []byte) error {
    level := string(value)
    switch level {
    case "info":
      logger.Level = logr.InfoLevel
    case "error":
      logger.Level = logr.ErrorLevel
    case "debug":
      logger.Level = logr.DebugLevel
    }
    logger.Infof("set log level to %s ", string(value))
    return nil
  }
}
func FooFunc(c ConfigSetter) configs.WatchFunc {
  return func(key string, value []byte) error {
    return c.Set(key, value)
  }
}

Here is our example type that we can send to FooFuncFooFunc. For this example, we aren’t doing anything other than setting the value.

type Foo struct {
  Flag string
}
func (f *Foo) Set(key string, value []byte) error {
  f.Flag = string(value)
  return nil
}

type Foo struct {
  Flag string
}
func (f *Foo) Set(key string, value []byte) error {
  f.Flag = string(value)
  return nil
}

Let’s finally build our main function.

We first build our logger and instantiate our FooFoo object. We then set up our NATS connection so we can pass it to our NewNew function. Then we define the keys that we want to watch and the WatchFuncWatchFunc that goes with that key. These are our lower level filter subjects for this service. Next we call NewNew with our options to build our config watcher. Once the watcher is build, we create a channel for any errors encountered by the watcher functions. Since the watcher is going to run in a goroutine, these errors are passed back to the next goroutine to handle. Finally we run an infinite loop to demonstrate our real time changes.

func main() {
  logger := logr.NewLogger()
  foo := Foo{
    Flag: "initial",
  }
  nc, err := nats.Connect(nats.DefaultURL)
  if err != nil {
    log.Fatal(err)
  }
  keys := map[string]configs.WatchFunc{
    "services.logger.log_level": LogLevel(logger),
    "services.foo":              FooFunc(&foo),
  }
  ff, err := configs.New(configs.ConfigWatcherOpts{
    Bucket: "configs",
    Name:   "services.>",
    Keys:   keys,
    Conn:   nc,
  })
  if err != nil {
    log.Fatal(err)
  }
  errChan := make(chan error)
  go ff.Watch(errChan)
  fmt.Println("service started")
  go func() {
    err := <-errChan
    if err != nil {
      os.Exit(1)
    }
  }()
  for {
    logger.Info("info")
    logger.Debug("debug")
    logger.Infof("%+v\n", foo)
    time.Sleep(2 * time.Second)
  }
}

func main() {
  logger := logr.NewLogger()
  foo := Foo{
    Flag: "initial",
  }
  nc, err := nats.Connect(nats.DefaultURL)
  if err != nil {
    log.Fatal(err)
  }
  keys := map[string]configs.WatchFunc{
    "services.logger.log_level": LogLevel(logger),
    "services.foo":              FooFunc(&foo),
  }
  ff, err := configs.New(configs.ConfigWatcherOpts{
    Bucket: "configs",
    Name:   "services.>",
    Keys:   keys,
    Conn:   nc,
  })
  if err != nil {
    log.Fatal(err)
  }
  errChan := make(chan error)
  go ff.Watch(errChan)
  fmt.Println("service started")
  go func() {
    err := <-errChan
    if err != nil {
      os.Exit(1)
    }
  }()
  for {
    logger.Info("info")
    logger.Debug("debug")
    logger.Infof("%+v\n", foo)
    time.Sleep(2 * time.Second)
  }
}

As the service runs, you can see our intial values are set for our log level and our Foo flag. As we update the KVs on the right side, the watcher receives the change and then updates our logger to use the debug logging level and then updates the value set in the Foo struct.

All of the code here is in this example-repo