streamingconfig

README

Streaming configuration

Streamingconfig is a MongoDB Change Streams powered, hot-reloadable configuration for Go!

This library enables real-time configuration updates in your distributed Go services, eliminating the need for restarts.

Benefits:

• Dynamic Feature Flags: Activate or deactivate features on the fly.
• Zero-Downtime Config Reloads: Update configurations without service interruptions.

Ideal for:

• Kubernetes Deployments: Ensures all pods receive configuration changes seamlessly.

A streamingconfig is a simple Go struct that:

• contains json and default field tags;
• implements a simple single interface:

type Config interface {
    Update(new Config) error
}

Design

• Auditable: All configuration changes are tracked for historical reference. This comes at the cost of increased storage consumption, as each change creates a new document containing all configuration fields.
• Immutable: Configurations are versioned and immutable. Every update creates a new version, preserving the history.
• Eventually Consistent: Configuration changes eventually replicate to other local repositories. There may be a slight delay.
• Dynamic Defaults: Default values are not stored and can be modified during deployment of new configuration versions.
• Fast Local Retrieval: Getting configuration data locally is fast as it's retrieved from memory, not requiring remote queries.
• Input validation: User-provided configuration changes validation through the Update method.

Usage

checkout the example folder for a more real-world scenario.

As a library user, you will have to:

1. Define a configuration with json field tags (and optionally with default field tags);
2. Make sure that your configuration type implements the streamingconfig.Config interface;

   NOTE: Within the Update method you can implement configuration validation see example below.

3. Instantiate and start the repository and use it;

package main

import (
	"errors"
	config "github.com/rbroggi/streamingconfig"
)

type conf struct {
	Name    string   `json:"name" default:"john"`
	Age     int      `json:"age"`
}

func (c *conf) Update(new config.Config) error {
	newCfg, ok := new.(*conf)
	if !ok {
		return errors.New("wrong configuration")
	}
	c.Name = newCfg.Name
	c.Age = newCfg.Age
	return c.validate()
}

// validation should not disallow zero-values as the `Update` 
// method is called on the struct without it's default values.
func (c *conf) validate() error {
	if c.Age < 0 {
		return errors.New("age must not be negative")
	}
	return nil
}

func main() {
	repo, err := config.NewWatchedRepo[*conf](
		config.Args{
			Logger: getLogger(),
			DB:     getDatabase(),
		})
	if err != nil {
		log.Fatal(err)
	}
	ctx, cnl := context.WithCancel(context.Background())
	done, err := repo.Start(ctx)
	if err != nil {
		log.Fatal(err)
	}
	// use repo
	cnl()
	<-done
}

Test

make dependencies_up
make tests

Run example server

make dependencies_up
make example

Optionally, you can also start a second server to check that the changes happening in one server will be reflected in the other:

HTTP_PORT=8081 make example

Getting latest configuration request

curl -X GET --location "http://localhost:8080/configs/latest"

Changing latest configuration request

curl -X PUT --location "http://localhost:8080/configs/latest" \
    -H "user-id: mark" \
    -d '{
  "name": "betty",
  "age": 35
}'

Listing multiple versions

curl -X GET --location "http://localhost:8080/configs?fromVersion=0&toVersion=21"

Documentation

Overview

Package streamingconfig provides functionality for using dynamic configuration functionalities.

Index

• Variables
• func WithCollectionName[T Config](collectionName string) func(repo *WatchedRepo[T])
• func WithNowFn[T Config](nf func() time.Time) func(*WatchedRepo[T])
• func WithSkipIndexOperations[T Config]() func(*WatchedRepo[T])
• type Args
• type Config
• type ListConfigDatesQuery
• type ListVersionedConfigsQuery
• type UpdateConfigCmd
• type Versioned
• type WatchedRepo
  • func NewWatchedRepo[T Config](args Args, opts ...func(*WatchedRepo[T])) (*WatchedRepo[T], error)
  • func (s *WatchedRepo[T]) GetConfig() (T, error)
  • func (s *WatchedRepo[T]) GetLatestVersion() (*Versioned[T], error)
  • func (s *WatchedRepo[T]) ListVersionedConfigs(ctx context.Context, query ListVersionedConfigsQuery) ([]*Versioned[T], error)
  • func (s *WatchedRepo[T]) ListVersionedConfigsByDate(ctx context.Context, query ListConfigDatesQuery) ([]*Versioned[T], error)
  • func (s *WatchedRepo[T]) Start(ctx context.Context) (<-chan struct{}, error)
  • func (s *WatchedRepo[T]) UpdateConfig(ctx context.Context, cmd UpdateConfigCmd[T]) (*Versioned[T], error)

Variables

var (
	// ErrNotStarted is returned when a user attempts to use the store before calling the `Start` method.
	ErrNotStarted            = errors.New("config store not started - call Start() before using it")
	ErrConfigurationNotFound = errors.New("configuration not found")
	// ErrConcurrentUpdate signals that multiple repositories are attempting to change the configuration concurrently.
	ErrConcurrentUpdate = errors.New("configuration concurrently being updated by someone-else")
	// ErrTypeMustBePointer by the constructor of the repo if the provided type is not a pointer type.
	ErrTypeMustBePointer = errors.New("configuration type argument must be pointer")
)

Functions

func WithCollectionName

func WithCollectionName[T Config](collectionName string) func(repo *WatchedRepo[T])

func WithNowFn

func WithNowFn[T Config](nf func() time.Time) func(*WatchedRepo[T])

func WithSkipIndexOperations

func WithSkipIndexOperations[T Config]() func(*WatchedRepo[T])

Types

type Args

type Args struct {
	Logger *slog.Logger
	DB     *mongo.Database
}

type Config

type Config interface {
	Update(new Config) error
}

Config is a representation of the app configuration. It is wrapped into a generic configuration which includes additional fields for the sake of auditability (version, updated_at and updated_by).

type ListConfigDatesQuery

type ListConfigDatesQuery struct {
	// From is the time from which retrieve the configs (inclusive)
	From time.Time
	// To is the time until which retrieve the configs (exclusive)
	To time.Time
}

ListConfigDatesQuery provide query parameters for listing configurations by dates.

type ListVersionedConfigsQuery

type ListVersionedConfigsQuery struct {
	// FromVersion version from which retrieve the configs (inclusive)
	FromVersion uint32
	// ToVersion version until which retrieve the configs (exclusive)
	ToVersion uint32
}

ListVersionedConfigsQuery provide query parameters for listing configurations by version.

type UpdateConfigCmd

type UpdateConfigCmd[T Config] struct {
	By     string
	Config T
}

type Versioned

type Versioned[T Config] struct {
	// Version is used as unique ID of the config collection.
	Version uint64 `json:"version" bson:"_id"`
	// UpdatedBy author of the config update
	UpdatedBy string `json:"updated_by" bson:"updated_by"`
	// CreatedAt time of the last config update.
	CreatedAt time.Time `json:"created_at" bson:"created_at"`
	// Config embeds the application-specific configuration.
	Config T `json:"config" bson:"app_config"`
}

Versioned encapsulates a version of the configuration and adds some auditing information on top of the configuration.

type WatchedRepo

type WatchedRepo[T Config] struct {
	// contains filtered or unexported fields
}

func NewWatchedRepo

func NewWatchedRepo[T Config](
	args Args,
	opts ...func(*WatchedRepo[T]),
) (*WatchedRepo[T], error)

func (*WatchedRepo[T]) GetConfig

func (s *WatchedRepo[T]) GetConfig() (T, error)

GetConfig gets the current user-defined configuration with defaults applied to it.

func (*WatchedRepo[T]) GetLatestVersion

func (s *WatchedRepo[T]) GetLatestVersion() (*Versioned[T], error)

GetLatestVersion returns the latest version of the user-provided configuration along with auditing data.

func (*WatchedRepo[T]) ListVersionedConfigs

func (s *WatchedRepo[T]) ListVersionedConfigs(
	ctx context.Context,
	query ListVersionedConfigsQuery,
) ([]*Versioned[T], error)

ListVersionedConfigs returns a list of the user-provided configuration versions along with auditing data.

func (*WatchedRepo[T]) ListVersionedConfigsByDate

func (s *WatchedRepo[T]) ListVersionedConfigsByDate(
	ctx context.Context,
	query ListConfigDatesQuery,
) ([]*Versioned[T], error)

ListVersionedConfigsByDate returns a list of the user-provided configuration versions along with auditing data.

func (*WatchedRepo[T]) Start

func (s *WatchedRepo[T]) Start(ctx context.Context) (<-chan struct{}, error)

Start is a non-blocking call that starts the store and watches for updates on the persisted configurations.

For graceful shutdown, cancel the input context and wait for the returned channel to be closed.

func (*WatchedRepo[T]) UpdateConfig

func (s *WatchedRepo[T]) UpdateConfig(ctx context.Context, cmd UpdateConfigCmd[T]) (*Versioned[T], error)

UpdateConfig retrieves the latest configuration, modifies it by calling the underlying `Update` method and creates a new updated version.