utils

package
v1.2.9 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 24, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

Package utils provides utility functions and types for the Lumen SDK.

The utils package contains common utilities used throughout the SDK:

  • Structured error handling with LumenError
  • Logging configuration and initialization
  • Retry mechanisms with exponential backoff
  • Circuit breaker for fault tolerance
  • Health monitoring utilities

Structured Errors

Use LumenError for consistent error handling: 

if node == nil {
    return utils.NodeNotFoundError(nodeID, map[string]interface{}{
        "available_nodes": nodeCount,
    })
}

Check error types programmatically: 

if utils.HasErrorCode(err, utils.ErrCodeTimeout) {
    // Handle timeout specifically
    return retry(operation)
}

Logging

Initialize structured logging: 

logCfg := &config.LoggingConfig{
    Level:  "info",
    Format: "json",
    Output: "stdout",
}
utils.InitLogger(logCfg)

// Use the logger
utils.Logger.Info("Operation completed",
    zap.String("operation", "inference"),
    zap.Duration("duration", elapsed))

Retry Logic

Add resilience with automatic retries: 

retryConfig := utils.DefaultRetryConfig()
err := utils.Retry(ctx, retryConfig, func(ctx context.Context) error {
    result, err := client.Infer(ctx, request)
    if err != nil {
        return err
    }
    return processResult(result)
})

The retry mechanism automatically determines if errors are retryable:

  • Timeout errors: retried
  • Network errors: retried
  • Validation errors: not retried
  • Authorization errors: not retried

Circuit Breaker

Protect against cascading failures: 

cb := utils.NewCircuitBreaker("ml-node-1", 5, 30*time.Second)
err := cb.Execute(func() error {
    return client.Infer(ctx, request)
})
if err != nil {
    if cb.GetState() == utils.CircuitOpen {
        log.Println("Circuit breaker open, service unavailable")
    }
}

Error Aggregation

Collect multiple errors in batch operations: 

aggr := utils.NewErrorAggregator()
for _, node := range nodes {
    if err := node.Connect(); err != nil {
        aggr.Add(err)
    }
}
if aggr.HasErrors() {
    log.Printf("Connection errors: %v", aggr.Error())
}

Role in Project

The utils package provides foundational utilities that enhance the SDK's reliability, observability, and error handling. These utilities are used throughout the codebase to ensure consistent behavior and robust operation.

Index

Constants

This section is empty.

Variables

View Source 
var Logger *zap.Logger

Logger is the global structured logger instance for the Lumen SDK.

This is a zap.Logger configured based on the logging configuration. Use this for performance-critical logging with structured fields.

View Source 
var Sugar *zap.SugaredLogger

Sugar is the global sugared logger for more convenient logging.

SugaredLogger provides a more ergonomic API with printf-style formatting at the cost of minor performance overhead. Use for non-critical paths.

Functions

func Backoff 

func Backoff(attempt int, base time.Duration, multiplier float64, max time.Duration) time.Duration

Backoff 计算退避时间

func HasErrorCode 

func HasErrorCode(err error, code ErrorCode) bool

HasErrorCode 检查是否包含特定错误码

func InitLogger 

func InitLogger(cfg *config.LoggingConfig)

InitLogger initializes the global logger with the specified configuration.

This function sets up structured logging with configurable:

  • Log level (debug, info, warn, error, fatal)
  • Output format (json, text/console)
  • Output destination (stdout, stderr, file)
  • File rotation (when using file output)

The function initializes both Logger (structured) and Sugar (convenient) loggers. It should be called once during application startup.

Parameters:

  • cfg: Logging configuration specifying level, format, and output

Role in project: Configures the centralized logging system used throughout the SDK. Proper logging is essential for debugging, monitoring, and operational visibility.

Example: 

// Initialize with custom config
logCfg := &config.LoggingConfig{
    Level:  "debug",
    Format: "json",
    Output: "stdout",
}
utils.InitLogger(logCfg)

// Use the logger
utils.Logger.Info("Client started",
    zap.String("version", "1.0.0"),
    zap.Int("nodes", 3))

// Or use sugar for convenience
utils.Sugar.Infof("Client started: version=%s, nodes=%d", "1.0.0", 3)

func IsLumenError 

func IsLumenError(err error) bool

IsLumenError 检查是否为Lumen错误

func IsRetryable 

func IsRetryable(err error) bool

IsRetryable 检查错误是否可重试

func NewRetryableError 

func NewRetryableError(err error, shouldRetry bool) error

NewRetryableError wraps an error with retry eligibility information.

Use this to explicitly mark errors as retryable or non-retryable when the automatic detection isn't sufficient for your use case.

Parameters:

  • err: The underlying error
  • shouldRetry: Whether this error should trigger a retry

Returns:

  • error: Wrapped error implementing RetryableError interface

Example: 

if networkErr != nil {
    // Mark as retryable
    return utils.NewRetryableError(networkErr, true)
}
if validationErr != nil {
    // Mark as non-retryable
    return utils.NewRetryableError(validationErr, false)
}

func Recover 

func Recover() error

Recover 恢复panic并转换为错误

func Retry 

func Retry(ctx context.Context, config *RetryConfig, fn RetryFunc) error

Retry executes a function with automatic retry on transient failures.

This function implements exponential backoff retry logic with configurable parameters. It automatically determines if errors are retryable based on:

  • RetryableError interface implementation
  • Lumen error codes (timeout, unavailable, connection failed)
  • Common network error patterns

Non-retryable errors (validation, authorization, etc.) fail immediately without retry.

Parameters:

  • ctx: Context for cancellation (respects context timeout/cancellation)
  • config: Retry configuration (attempts, backoff, etc.)
  • fn: The function to execute with retry logic

Returns:

  • error: The last error encountered, or nil if successful

Role in project: Adds resilience to ML inference operations by handling transient network and service failures automatically. Essential for production reliability.

Example: 

retryConfig := utils.DefaultRetryConfig()
err := utils.Retry(ctx, retryConfig, func(ctx context.Context) error {
    result, err := client.Infer(ctx, request)
    if err != nil {
        return err
    }
    // Process result
    return nil
})
if err != nil {
    log.Printf("Failed after retries: %v", err)
}

func RetryWithCallback 

func RetryWithCallback(ctx context.Context, config *RetryConfig, fn RetryFunc, callback RetryCallback) error

func SafeExecute 

func SafeExecute(fn func() error) (err error)

SafeExecute 安全执行函数，捕获panic

Types

type CircuitBreaker 

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

CircuitBreaker implements the circuit breaker pattern for fault tolerance.

The circuit breaker prevents cascading failures by:

  • Tracking consecutive failures
  • Opening the circuit after threshold failures (rejecting requests immediately)
  • Allowing test requests after a reset timeout (half-open state)
  • Closing the circuit when requests succeed again

States:

  • Closed: Normal operation, all requests go through
  • Open: Circuit tripped, requests fail fast without execution
  • Half-Open: Testing if service recovered, limited requests allowed

Role in project: Protects the system from repeatedly calling failing services, allowing them time to recover and preventing resource exhaustion.

Example: 

cb := utils.NewCircuitBreaker("ml-node-1", 5, 30*time.Second)
err := cb.Execute(func() error {
    return client.Infer(ctx, request)
})
if err != nil {
    if cb.GetState() == utils.CircuitOpen {
        log.Println("Circuit breaker open, service unavailable")
    }
}

func NewCircuitBreaker 

func NewCircuitBreaker(name string, maxFailures int, resetTime time.Duration) *CircuitBreaker

NewCircuitBreaker 创建断路器

func (*CircuitBreaker) Execute 

func (cb *CircuitBreaker) Execute(fn func() error) error

Execute 执行函数

func (*CircuitBreaker) GetState 

func (cb *CircuitBreaker) GetState() CircuitState

GetState 获取断路器状态

func (*CircuitBreaker) Reset 

func (cb *CircuitBreaker) Reset()

Reset 重置断路器

type CircuitState 

type CircuitState int
const (
	CircuitClosed CircuitState = iota
	CircuitOpen
	CircuitHalfOpen
)

type CompositeHealthChecker 

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

CompositeHealthChecker 复合健康检查器

func NewCompositeHealthChecker 

func NewCompositeHealthChecker(name string, checkers ...HealthChecker) *CompositeHealthChecker

NewCompositeHealthChecker 创建复合健康检查器

func (*CompositeHealthChecker) Check 

func (h *CompositeHealthChecker) Check(ctx context.Context) *HealthCheckResult

Check 执行复合健康检查

func (*CompositeHealthChecker) Name 

func (h *CompositeHealthChecker) Name() string

Name 返回检查器名称

type ErrorAggregator 

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

ErrorAggregator collects multiple errors for batch operations.

Use this when performing operations on multiple items where you want to:

  • Continue processing despite individual failures
  • Report all errors at once rather than failing on first error
  • Track partial success/failure in batch operations

Role in project: Enables robust error handling in batch operations like initializing multiple connections or processing multiple inference requests.

Example: 

aggr := utils.NewErrorAggregator()
for _, node := range nodes {
    if err := node.Connect(); err != nil {
        aggr.Add(err)
    }
}
if aggr.HasErrors() {
    log.Printf("Connection errors: %v", aggr.Error())
}

func NewErrorAggregator 

func NewErrorAggregator() *ErrorAggregator

NewErrorAggregator 创建错误聚合器

func (*ErrorAggregator) Add 

func (ea *ErrorAggregator) Add(err error)

Add 添加错误

func (*ErrorAggregator) Error 

func (ea *ErrorAggregator) Error() string

Error 返回聚合错误信息

func (*ErrorAggregator) GetErrors 

func (ea *ErrorAggregator) GetErrors() []error

GetErrors 获取所有错误

func (*ErrorAggregator) HasErrors 

func (ea *ErrorAggregator) HasErrors() bool

HasErrors 是否有错误

func (*ErrorAggregator) Reset 

func (ea *ErrorAggregator) Reset()

Reset 重置错误聚合器

func (*ErrorAggregator) ToError 

func (ea *ErrorAggregator) ToError() error

ToError 转换为error接口

type ErrorCode 

type ErrorCode string

ErrorCode represents a standardized error code for the Lumen SDK.

Error codes provide a machine-readable way to identify error types, enabling clients to handle errors programmatically with appropriate retry logic, user messages, and recovery strategies.

Role in project: Standardizes error handling across the SDK, enabling consistent error reporting and intelligent error recovery. 

const (
	// General error codes applicable across all components
	ErrCodeInternal     ErrorCode = "INTERNAL" // Internal server/SDK error
	ErrCodeInvalid      ErrorCode = "INVALID"
	ErrCodeTimeout      ErrorCode = "TIMEOUT"
	ErrCodeUnavailable  ErrorCode = "UNAVAILABLE"
	ErrCodeNotFound     ErrorCode = "NOT_FOUND"
	ErrCodeUnauthorized ErrorCode = "UNAUTHORIZED"
	ErrCodeForbidden    ErrorCode = "FORBIDDEN"

	// Lumen特定错误码
	ErrCodeNodeNotFound       ErrorCode = "NODE_NOT_FOUND"
	ErrCodeServiceUnavailable ErrorCode = "SERVICE_UNAVAILABLE"
	ErrCodeTaskUnsupported    ErrorCode = "TASK_UNSUPPORTED"
	ErrCodecMismatch          ErrorCode = "CODEC_MISMATCH"
	ErrCodeDiscoveryFailed    ErrorCode = "DISCOVERY_FAILED"
	ErrCodeConnectionFailed   ErrorCode = "CONNECTION_FAILED"
	ErrCodeRequestFailed      ErrorCode = "REQUEST_FAILED"
	ErrCodeResponseFailed     ErrorCode = "RESPONSE_FAILED"
)

type GRPCHealthChecker 

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

GRPCHealthChecker gRPC健康检查器

func NewGRPCHealthChecker 

func NewGRPCHealthChecker(client pb.InferenceClient, name string, timeout time.Duration) *GRPCHealthChecker

NewGRPCHealthChecker 创建gRPC健康检查器

func (*GRPCHealthChecker) Check 

func (h *GRPCHealthChecker) Check(ctx context.Context) *HealthCheckResult

Check 执行健康检查

func (*GRPCHealthChecker) Name 

func (h *GRPCHealthChecker) Name() string

Name 返回检查器名称

type HTTPHealthChecker 

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

HTTPHealthChecker HTTP健康检查器

func NewHTTPHealthChecker 

func NewHTTPHealthChecker(url, name string, timeout time.Duration) *HTTPHealthChecker

NewHTTPHealthChecker 创建HTTP健康检查器

func (*HTTPHealthChecker) Check 

func (h *HTTPHealthChecker) Check(ctx context.Context) *HealthCheckResult

Check 执行健康检查

func (*HTTPHealthChecker) Name 

func (h *HTTPHealthChecker) Name() string

Name 返回检查器名称

type HealthCheckResult 

type HealthCheckResult struct {
	Status    HealthStatus           `json:"status"`
	Message   string                 `json:"message"`
	Timestamp time.Time              `json:"timestamp"`
	Details   map[string]interface{} `json:"details,omitempty"`
	Duration  time.Duration          `json:"duration"`
}

HealthCheckResult 健康检查结果

type HealthChecker 

type HealthChecker interface {
	Check(ctx context.Context) *HealthCheckResult
	Name() string
}

HealthChecker 健康检查器接口

type HealthMonitor 

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

HealthMonitor 健康监控器

func NewHealthMonitor 

func NewHealthMonitor(interval time.Duration) *HealthMonitor

NewHealthMonitor 创建健康监控器

func (*HealthMonitor) AddChecker 

func (hm *HealthMonitor) AddChecker(checker HealthChecker)

AddChecker 添加健康检查器

func (*HealthMonitor) CheckAll 

func (hm *HealthMonitor) CheckAll(ctx context.Context) map[string]*HealthCheckResult

CheckAll 执行所有健康检查

func (*HealthMonitor) GetAllResults 

func (hm *HealthMonitor) GetAllResults() map[string]*HealthCheckResult

GetAllResults 获取所有检查器的最新结果

func (*HealthMonitor) GetHealthyCheckers 

func (hm *HealthMonitor) GetHealthyCheckers() []string

GetHealthyCheckers 获取所有健康的检查器

func (*HealthMonitor) GetResult 

func (hm *HealthMonitor) GetResult(name string) (*HealthCheckResult, bool)

GetResult 获取特定检查器的最新结果

func (*HealthMonitor) GetUnhealthyCheckers 

func (hm *HealthMonitor) GetUnhealthyCheckers() []string

GetUnhealthyCheckers 获取所有不健康的检查器

func (*HealthMonitor) IsRunning 

func (hm *HealthMonitor) IsRunning() bool

IsRunning 检查是否正在运行

func (*HealthMonitor) RemoveChecker 

func (hm *HealthMonitor) RemoveChecker(name string)

RemoveChecker 移除健康检查器

func (*HealthMonitor) Start 

func (hm *HealthMonitor) Start(ctx context.Context)

Start 启动定期健康检查

func (*HealthMonitor) Stop 

func (hm *HealthMonitor) Stop()

Stop 停止健康监控

type HealthStatus 

type HealthStatus string

HealthStatus 健康状态 

const (
	StatusHealthy   HealthStatus = "healthy"
	StatusUnhealthy HealthStatus = "unhealthy"
	StatusUnknown   HealthStatus = "unknown"
)

type LumenError 

type LumenError struct {
	Code    ErrorCode   `json:"code"`
	Message string      `json:"message"`
	Details interface{} `json:"details,omitempty"`
	Cause   error       `json:"-"`
}

LumenError represents a structured error from the Lumen SDK.

This error type provides:

  • Standardized error codes for programmatic handling
  • Human-readable error messages
  • Optional structured details (can be logged or returned to API clients)
  • Error wrapping/chaining support via Cause

Role in project: Provides structured, actionable error information throughout the SDK. Essential for debugging, monitoring, and building resilient applications.

Example: 

err := utils.NodeNotFoundError("node-123", map[string]interface{}{
    "requested_at": time.Now(),
    "available_nodes": 5,
})
if utils.HasErrorCode(err, utils.ErrCodeNodeNotFound) {
    // Handle node not found specifically
}

func CodecMismatchError 

func CodecMismatchError(expected, actual string, details ...interface{}) *LumenError

func ConnectionFailedError 

func ConnectionFailedError(target string, details ...interface{}) *LumenError

func DiscoveryFailedError 

func DiscoveryFailedError(message string, details ...interface{}) *LumenError

func ForbiddenError 

func ForbiddenError(message string, details ...interface{}) *LumenError

func GetLumenError 

func GetLumenError(err error) (*LumenError, bool)

GetLumenError 获取Lumen错误

func InternalError 

func InternalError(message string, details ...interface{}) *LumenError

InternalError creates an internal error indicating an unexpected condition.

Use for unexpected errors, programming errors, or unhandled edge cases. These typically indicate bugs or misconfigurations.

Example: 

if node == nil {
    return utils.InternalError("node should never be nil at this point")
}

func InvalidError 

func InvalidError(message string, details ...interface{}) *LumenError

func NewLumenError 

func NewLumenError(code ErrorCode, message string, details ...interface{}) *LumenError

NewLumenError creates a new LumenError with the specified code and message.

Parameters:

  • code: Standardized error code (e.g., ErrCodeTimeout)
  • message: Human-readable error description
  • details: Optional structured details (first element used if provided)

Returns:

  • *LumenError: New error instance

Example: 

err := utils.NewLumenError(
    utils.ErrCodeTimeout,
    "inference request timed out",
    map[string]interface{}{"timeout": "30s", "node": "node-1"},
)

func NodeNotFoundError 

func NodeNotFoundError(nodeID string, details ...interface{}) *LumenError

NodeNotFoundError creates an error indicating a requested ML node was not found.

This error typically occurs when:

  • The node ID doesn't exist in the discovered nodes
  • The node has been removed or gone offline
  • Service discovery hasn't found any nodes yet

Parameters:

  • nodeID: The ID of the node that was not found
  • details: Optional additional context

Example: 

node, exists := discovery.GetNode(nodeID)
if !exists {
    return utils.NodeNotFoundError(nodeID, map[string]interface{}{
        "available_nodes": discovery.GetNodeCount(),
    })
}

func NotFoundError 

func NotFoundError(message string, details ...interface{}) *LumenError

func RequestFailedError 

func RequestFailedError(message string, details ...interface{}) *LumenError

func ResponseFailedError 

func ResponseFailedError(message string, details ...interface{}) *LumenError

func ServiceUnavailableError 

func ServiceUnavailableError(service string, details ...interface{}) *LumenError

func TaskUnsupportedError 

func TaskUnsupportedError(task string, details ...interface{}) *LumenError

func TimeoutError 

func TimeoutError(message string, details ...interface{}) *LumenError

func UnauthorizedError 

func UnauthorizedError(message string, details ...interface{}) *LumenError

func UnavailableError 

func UnavailableError(message string, details ...interface{}) *LumenError

func Wrap 

func Wrap(err error, code ErrorCode, message string, details ...interface{}) *LumenError

Wrap wraps an existing error with additional context and a Lumen error code.

This function creates a new LumenError that preserves the original error as the cause, enabling error chain unwrapping with errors.Unwrap().

Parameters:

  • err: The original error to wrap
  • code: Lumen error code for categorization
  • message: Additional context message
  • details: Optional structured details

Returns:

  • *LumenError: Wrapped error with Lumen context

Example: 

_, err := conn.Dial(address)
if err != nil {
    return utils.Wrap(err, utils.ErrCodeConnectionFailed,
        "failed to connect to ML node",
        map[string]string{"address": address})
}

func (*LumenError) Error 

func (e *LumenError) Error() string

Error 实现error接口

func (*LumenError) Unwrap 

func (e *LumenError) Unwrap() error

Unwrap 支持错误链

type RetryCallback 

type RetryCallback func(attempt int, err error)

RetryWithCallback 带回调的重试执行

type RetryConfig 

type RetryConfig struct {
	Enabled     bool          `json:"enabled"`
	MaxAttempts int           `json:"max_attempts"`
	Backoff     time.Duration `json:"backoff"`
	MaxBackoff  time.Duration `json:"max_backoff"`
	Multiplier  float64       `json:"multiplier"`
}

RetryConfig defines configuration for retry behavior with exponential backoff.

Retry mechanisms help handle transient failures in distributed systems like network timeouts, temporary service unavailability, and connection issues.

Role in project: Provides resilience against temporary failures in ML inference requests, making the SDK more robust in production environments.

Example: 

retryConfig := &utils.RetryConfig{
    Enabled:     true,
    MaxAttempts: 3,
    Backoff:     100 * time.Millisecond,
    MaxBackoff:  5 * time.Second,
    Multiplier:  2.0,
}

func DefaultRetryConfig 

func DefaultRetryConfig() *RetryConfig

DefaultRetryConfig 默认重试配置

type RetryFunc 

type RetryFunc func(ctx context.Context) error

RetryFunc is the function signature for operations that can be retried.

Functions of this type should be idempotent or handle their own state to ensure correct behavior when executed multiple times.

type RetryableError 

type RetryableError interface {
	ShouldRetry() bool
}

RetryableError is an interface for errors that indicate whether they should be retried.

Implement this interface on custom error types to control retry behavior. The SDK automatically determines retry eligibility for common error types.

Role in project: Enables intelligent retry decisions based on error semantics, preventing unnecessary retries for non-transient errors (e.g., validation errors).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.