etcdregistry

README

ETCD Registry

A Go library for service registration and discovery using etcd. This library provides a hierarchical system of service groups and nodes, with support for common configuration and label inheritance.

Features

• Service groups with shared configuration
• Node registration with TTL (Time To Live)
• Common labels inheritance from group to nodes
• Automatic registration renewal
• Service discovery at group and node level
• Authentication support per group
• Configurable timeouts
• Error handling and automatic retries
• Graceful cleanup on context cancellation
• Automatic cleanup of empty groups

Installation

go get go.lumeweb.com/etcd-registry

Usage

Creating a Registry Instance

import (
    "time"
    etcdregistry "go.lumeweb.com/etcd-registry"
)

registry, err := etcdregistry.NewEtcdRegistry(
    []string{"localhost:2379"},  // etcd endpoints
    "/services",                 // base path
    "username",                  // username (optional)
    "password",                  // password (optional)
    10*time.Second,             // default timeout
    3,                          // max retries
)
if err != nil {
    log.Fatal(err)
}

Creating or Joining a Service Group

// Create/join a service group
group, err := registry.CreateOrJoinServiceGroup(ctx, "my-service")
if err != nil {
    log.Fatal(err)
}

// Configure the group
spec := types.ServiceGroupSpec{
    Username: "group-user",        // Auth credentials for this group
    Password: "group-pass",
    CommonLabels: map[string]string{
        "environment": "production",
        "region": "us-east",
    },
}

err = group.Configure(spec)
if err != nil {
    log.Fatal(err)
}

Registering a Node

node := types.Node{
    ID:           "node1",
    ExporterType: "node_exporter",
    Port:         9100,
    MetricsPath:  "/metrics",
    Labels: map[string]string{
        "instance": "i-1234567",  // Node-specific labels
    },
}

// Register with 30-second TTL
done, errChan, err := group.RegisterNode(ctx, node, 30*time.Second)
if err != nil {
    log.Fatal(err)
}

// Wait for initial registration
select {
case <-done:
    log.Println("Node registered successfully")
case err := <-errChan:
    log.Printf("Registration failed: %v", err)
}

Service Discovery

// List all service groups
groups, err := registry.GetServiceGroups(ctx)
if err != nil {
    log.Fatal(err)
}

// Get group configuration
group, err := registry.GetServiceGroup(ctx, "my-service")
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Group: %s, Common Labels: %v\n", group.Name, group.Spec.CommonLabels)

// Get nodes in a group
nodes, err := group.GetNodes(ctx)
if err != nil {
    log.Fatal(err)
}

for _, node := range nodes {
    // Node labels include both group common labels and node-specific labels
    fmt.Printf("Node: %s, Type: %s, Labels: %v\n", 
               node.ID, node.ExporterType, node.Labels)
}

Watch API

The library provides three watch methods for monitoring changes:

// Watch all services and nodes
events, err := registry.WatchServices(ctx)
if err != nil {
    log.Fatal(err)
}

// Watch a specific service group
events, err := registry.WatchGroup(ctx, "my-service")
if err != nil {
    log.Fatal(err)
}

// Watch nodes in a specific group
events, err := registry.WatchGroupNodes(ctx, "my-service")
if err != nil {
    log.Fatal(err)
}

// Handle events
for event := range events {
    switch event.Type {
    case types.EventTypeCreate:
        fmt.Printf("Created: group=%s node=%v\n", event.GroupName, event.Node)
    case types.EventTypeUpdate:
        fmt.Printf("Updated: group=%s node=%v\n", event.GroupName, event.Node)
    case types.EventTypeDelete:
        fmt.Printf("Deleted: group=%s node=%v\n", event.GroupName, event.Node)
    }
}

Each watch method returns a channel that receives WatchEvent structs containing:

• Type: The event type (create/update/delete)
• GroupName: The name of the affected service group
• Node: The affected node (if applicable)

The watch will continue until the context is cancelled or an error occurs.

Configuration

Registry Configuration

The registry can be configured with the following parameters:

• etcdEndpoints: List of etcd server endpoints (required)
• etcdBasePath: Base path for all etcd operations (required)
• etcdUsername: Username for etcd authentication (optional)
• etcdPassword: Password for etcd authentication (optional)
• defaultTimeout: Default timeout for etcd operations (required)
• maxRetries: Maximum number of retry attempts for operations (required)

Service Group Configuration

Each service group can have its own configuration:

• Username: Authentication username for the group
• Password: Authentication password for the group
• CommonLabels: Labels that are automatically applied to all nodes in the group

Node Configuration

Nodes require the following configuration:

• ID: Unique identifier for the node
• ExporterType: Type of metrics exporter (e.g., "node_exporter")
• Port: Port number where metrics are exposed
• MetricsPath: HTTP path where metrics are available
• Labels: Node-specific labels (merged with group's common labels)

Features in Detail

Service Groups

Service groups provide:

• Logical grouping of related nodes
• Shared authentication credentials
• Common labels inherited by all nodes
• Automatic cleanup when empty
• Name validation and sanitization

Automatic Registration Renewal

The library automatically maintains node registrations by:

• Refreshing the TTL periodically (at half the TTL interval)
• Handling connection failures with exponential backoff
• Retrying failed registrations automatically
• Inheriting and merging group labels

Error Handling

The library provides comprehensive error handling:

• Connection failures with automatic reconnection
• Authentication errors at both registry and group level
• Timeout handling with configurable retries
• Context cancellation
• Label validation
• Group name validation

Cleanup

The library ensures proper cleanup by:

• Closing connections properly
• Handling context cancellation
• Removing stale registrations
• Automatically cleaning up empty groups
• Proper lease management

Example

Here's a complete example showing how to use the library:

package main

import (
    "context"
    "log"
    "time"
    etcdregistry "go.lumeweb.com/etcd-registry"
)

func main() {
    // Create registry
    registry, err := etcdregistry.NewEtcdRegistry(
        []string{"localhost:2379"},
        "/services",
        "",
        "",
        10*time.Second,
    )
    if err != nil {
        log.Fatal(err)
    }

    // Create a node
    node := etcdregistry.Node{
        Name: "api-server-1",
        Info: map[string]string{
            "version": "1.0.0",
            "address": "localhost:8080",
        },
    }

    // Create context with cancellation
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Register node
    done, errChan, err := registry.RegisterNode(ctx, "api", node, 30*time.Second)
    if err != nil {
        log.Fatal(err)
    }

    // Wait for registration
    select {
    case <-done:
        log.Println("Node registered successfully")
    case err := <-errChan:
        log.Printf("Registration failed: %v", err)
        return
    }

    // Get all nodes
    nodes, err := registry.GetServiceNodes("api")
    if err != nil {
        log.Fatal(err)
    }

    for _, n := range nodes {
        log.Printf("Found node: %s with info: %v\n", n.Name, n.Info)
    }

    // Keep the program running
    <-ctx.Done()
}

Testing

The library includes comprehensive tests. Run them using:

go test -v

Requirements

• Go 1.23 or higher
• etcd 3.5 or higher

License

This project is licensed under the terms of the LICENSE file included in the repository.

Documentation

Overview

Package etcdregistry provides a library for registering and retrieving service nodes from etcd.

Index

• type EtcdRegistry
  • func NewEtcdRegistry(etcdEndpoints []string, etcdBasePath string, etcdUsername string, ...) (*EtcdRegistry, error)
  • func (r *EtcdRegistry) Close() error
  • func (r *EtcdRegistry) CreateOrJoinServiceGroup(ctx context.Context, groupName string) (*types.ServiceGroup, error)
  • func (r *EtcdRegistry) DeleteNode(ctx context.Context, groupName string, node types.Node) error
  • func (r *EtcdRegistry) GetClient() *clientv3.Client
  • func (r *EtcdRegistry) GetEtcdBasePath() string
  • func (r *EtcdRegistry) GetNodes(ctx context.Context, servicePath string) ([]types.Node, error)
  • func (r *EtcdRegistry) GetServiceGroup(ctx context.Context, groupName string) (*types.ServiceGroup, error)
  • func (r *EtcdRegistry) GetServiceGroups(ctx context.Context) ([]string, error)
  • func (r *EtcdRegistry) NodePath(groupName string, node types.Node) string
  • func (r *EtcdRegistry) RegisterNodeWithRetry(ctx context.Context, groupName string, node types.Node, ttl time.Duration, ...)
  • func (r *EtcdRegistry) ServicePath(groupName string) string
  • func (r *EtcdRegistry) UpdateServiceGroupConfig(ctx context.Context, groupName string, ...) error
  • func (r *EtcdRegistry) WatchGroup(ctx context.Context, groupName string) (<-chan types.WatchEvent, error)
  • func (r *EtcdRegistry) WatchGroupNodes(ctx context.Context, groupName string) (<-chan types.WatchEvent, error)
  • func (r *EtcdRegistry) WatchServices(ctx context.Context) (<-chan types.WatchEvent, error)

Types

type EtcdRegistry

type EtcdRegistry struct {
	// contains filtered or unexported fields
}

func NewEtcdRegistry

func NewEtcdRegistry(etcdEndpoints []string, etcdBasePath string, etcdUsername string, etcdPassword string, defaultTimeout time.Duration, maxRetries int) (*EtcdRegistry, error)

NewEtcdRegistry creates a new EtcdRegistry instance.

Args:

etcdEndpoints ([]string): The endpoints for the etcd cluster.
etcdBasePath (string): The base path for all etcd operations.
etcdUsername (string): The username for etcd authentication.
etcdPassword (string): The password for etcd authentication.
defaultTimeout (time.Duration): The default timeout for etcd operations.

Returns:

*EtcdRegistry: A new EtcdRegistry instance.
error: An error if the instance could not be created.

func (*EtcdRegistry) Close

func (r *EtcdRegistry) Close() error

Close gracefully shuts down the registry and cleans up resources

func (*EtcdRegistry) CreateOrJoinServiceGroup

func (r *EtcdRegistry) CreateOrJoinServiceGroup(ctx context.Context, groupName string) (*types.ServiceGroup, error)

CreateOrJoinServiceGroup ensures a group exists and returns a ServiceGroup instance

func (*EtcdRegistry) DeleteNode

func (r *EtcdRegistry) DeleteNode(ctx context.Context, groupName string, node types.Node) error

DeleteNode removes a node from etcd

func (*EtcdRegistry) GetClient

func (r *EtcdRegistry) GetClient() *clientv3.Client

GetClient returns the etcd client instance

func (*EtcdRegistry) GetEtcdBasePath

func (r *EtcdRegistry) GetEtcdBasePath() string

GetEtcdBasePath returns the base path for etcd operations

func (*EtcdRegistry) GetNodes

func (r *EtcdRegistry) GetNodes(ctx context.Context, servicePath string) ([]types.Node, error)

GetNodes returns all nodes at the given path

func (*EtcdRegistry) GetServiceGroup

func (r *EtcdRegistry) GetServiceGroup(ctx context.Context, groupName string) (*types.ServiceGroup, error)

GetServiceGroup retrieves group configuration

func (*EtcdRegistry) GetServiceGroups

func (r *EtcdRegistry) GetServiceGroups(ctx context.Context) ([]string, error)

GetServiceGroups returns a list of all service group names under the base path

func (*EtcdRegistry) NodePath

func (r *EtcdRegistry) NodePath(groupName string, node types.Node) string

NodePath returns the path for a specific exporter node

func (*EtcdRegistry) RegisterNodeWithRetry

func (r *EtcdRegistry) RegisterNodeWithRetry(ctx context.Context, groupName string, node types.Node, ttl time.Duration, done chan<- struct{}, errChan chan error)

RegisterNodeWithRetry handles the node registration retry loop

func (*EtcdRegistry) ServicePath

func (r *EtcdRegistry) ServicePath(groupName string) string

ServicePath returns the path for a service group's nodes

func (*EtcdRegistry) UpdateServiceGroupConfig

func (r *EtcdRegistry) UpdateServiceGroupConfig(ctx context.Context, groupName string, updateFn func(*types.ServiceGroup) error) error

UpdateServiceGroupConfig updates group configuration using CAS operations

func (*EtcdRegistry) WatchGroup

func (r *EtcdRegistry) WatchGroup(ctx context.Context, groupName string) (<-chan types.WatchEvent, error)

WatchGroup watches for changes in a specific service group

func (*EtcdRegistry) WatchGroupNodes

func (r *EtcdRegistry) WatchGroupNodes(ctx context.Context, groupName string) (<-chan types.WatchEvent, error)

WatchGroupNodes watches for node changes in a specific service group

func (*EtcdRegistry) WatchServices

func (r *EtcdRegistry) WatchServices(ctx context.Context) (<-chan types.WatchEvent, error)

WatchServices watches for all service and node changes