redis

package
v0.9.1 Latest Latest
Warning

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

 Go to latest Published: Sep 4, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ParseRedisOptions 

func ParseRedisOptions(url string) (*redis.Options, error)

ParseRedisOptions returns *redis.Options derived from the given URL.

Types

type ConnectionOptions 

type ConnectionOptions struct {
	URL string
}

ConnectionOptions holds the Redis connection configuration.

func NewConnectionOptions 

func NewConnectionOptions() *ConnectionOptions

func (*ConnectionOptions) AddFlags 

func (o *ConnectionOptions) AddFlags(fs *pflag.FlagSet)

type GatingMode 

type GatingMode string
const (
	GatingModeBlocking    GatingMode = "blocking"
	GatingModeClassifying GatingMode = "classifying"
)

type PubSubConfig added in v0.9.1 

type PubSubConfig struct {
	URL             string        `json:"url,omitempty"`
	RetryQueueName  string        `json:"retry_queue_name,omitempty"`
	ResultQueueName string        `json:"result_queue_name,omitempty"`
	EnableTracing   bool          `json:"enable_tracing,omitempty"`
	Queues          []QueueConfig `json:"queues"`
}

PubSubConfig is the transport config for the Redis pub/sub flow. It is parsed from JSON provided via --transport-config or --transport-config-file.

func LoadPubSubConfig added in v0.9.1 

func LoadPubSubConfig(data []byte) (*PubSubConfig, error)

LoadPubSubConfig parses, applies env overrides/defaults, and validates a PubSubConfig.

func (*PubSubConfig) ApplyDefaults added in v0.9.1 

func (c *PubSubConfig) ApplyDefaults()

func (*PubSubConfig) ApplyEnvOverrides added in v0.9.1 

func (c *PubSubConfig) ApplyEnvOverrides()

ApplyEnvOverrides seeds URL from REDIS_URL only when it is not already set in the config. An explicit url in --transport-config therefore wins over the environment; REDIS_URL is a fallback default, not an override.

func (*PubSubConfig) Validate added in v0.9.1 

func (c *PubSubConfig) Validate() error

type PubSubFlowOptions 

type PubSubFlowOptions struct {
	IGWBaseURL         string
	RequestPathURL     string
	InferenceObjective string
	RequestQueueName   string
	RetryQueueName     string
	ResultQueueName    string
	QueuesConfig       string
	QueuesConfigFile   string
}

PubSubFlowOptions holds CLI flags for the Redis pub/sub flow.

func NewPubSubFlowOptions 

func NewPubSubFlowOptions() *PubSubFlowOptions

func (*PubSubFlowOptions) AddFlags 

func (o *PubSubFlowOptions) AddFlags(fs *pflag.FlagSet)

func (*PubSubFlowOptions) HasQueueConfig 

func (o *PubSubFlowOptions) HasQueueConfig() bool

HasQueueConfig reports whether any multi-queue configuration is set.

type QueueConfig 

type QueueConfig struct {
	QueueName          string            `json:"queue_name"`
	WorkerPoolID       string            `json:"worker_pool_id"`
	InferenceObjective string            `json:"inference_objective"`
	RequestPathURL     string            `json:"request_path_url"`
	IGWBaseURL         string            `json:"igw_base_url"`
	Labels             map[string]string `json:"labels,omitempty"`
}

type QueueReconfigureResult added in v0.9.1 

type QueueReconfigureResult struct {
	Added   []pipeline.RequestChannel
	Removed []pipeline.RequestChannel
}

QueueReconfigureResult describes what one ReconfigureQueues call changed, in the terms the merge policy layer understands: channels that appeared (to be added to the fan-in) and channels that were closed (removed from it; closing the source channel is how a merge policy forgets a queue).

type QuotaMode 

type QuotaMode string
const (
	QuotaModeRateLimit   QuotaMode = "rate-limit"
	QuotaModeConcurrency QuotaMode = "concurrency"
)

type RedisDispatchGate 

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

RedisDispatchGate implements pipeline.DispatchGate by reading the budget from a Redis key. This allows external systems to dynamically control the dispatch rate. If the key does not exist or is invalid, it defaults to full capacity (1.0).

func NewRedisDispatchGate 

func NewRedisDispatchGate(client *goredis.Client, budgetKey string) *RedisDispatchGate

NewRedisDispatchGate creates a new RedisDispatchGate that reads budget from the given Redis client and budget key.

func (*RedisDispatchGate) Apply 

func (g *RedisDispatchGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)

Apply implements pipeline.Gate.

func (*RedisDispatchGate) Budget 

func (g *RedisDispatchGate) Budget(ctx context.Context) float64

Budget reads the dispatch budget from Redis. Returns a value in [0.0, 1.0]. Defaults to 1.0 (full capacity) when the key is absent or unparsable.

type RedisMQFlow 

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

func NewRedisMQFlow 

func NewRedisMQFlow(cfg PubSubConfig, workerPools []pipeline.WorkerPoolConfig) (*RedisMQFlow, error)

NewRedisMQFlow builds a Redis pub/sub flow from a parsed PubSubConfig. The config is expected to have had ApplyDefaults applied (LoadPubSubConfig does this); workerPools resolves the named pool each queue routes to.

func (*RedisMQFlow) Characteristics 

func (r *RedisMQFlow) Characteristics() pipeline.Characteristics

func (*RedisMQFlow) HealthCheck 

func (r *RedisMQFlow) HealthCheck(ctx context.Context) error

func (*RedisMQFlow) RequestChannels 

func (r *RedisMQFlow) RequestChannels() []pipeline.RequestChannel

func (*RedisMQFlow) ResultChannel 

func (r *RedisMQFlow) ResultChannel() chan api.ResultMessage

func (*RedisMQFlow) RetryChannel 

func (r *RedisMQFlow) RetryChannel() chan pipeline.RetryMessage

func (*RedisMQFlow) Shutdown 

func (r *RedisMQFlow) Shutdown()

func (*RedisMQFlow) Start 

func (r *RedisMQFlow) Start(ctx context.Context)

func (*RedisMQFlow) StopConsuming 

func (r *RedisMQFlow) StopConsuming()

type RedisQuotaGate 

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

func NewRedisQuotaGate 

func NewRedisQuotaGate(client *redis.Client, attribute string, mode QuotaMode, limit int, window time.Duration, prefix string) *RedisQuotaGate

func (*RedisQuotaGate) Apply 

func (g *RedisQuotaGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)

Apply implements pipeline.Gate.

func (*RedisQuotaGate) Budget 

func (g *RedisQuotaGate) Budget(ctx context.Context) float64

Budget implements api.DispatchGate. For quota gates, we return 1.0 (open) because the actual gating happens at the message level via Acquire.

func (*RedisQuotaGate) WithGatingMode 

func (g *RedisQuotaGate) WithGatingMode(mode GatingMode) *RedisQuotaGate

type RedisSortedSetFlow 

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

func NewRedisSortedSetFlow 

func NewRedisSortedSetFlow(cfg SortedSetConfig, workerPools []pipeline.WorkerPoolConfig, gateFactory pipeline.GateFactory) (*RedisSortedSetFlow, error)

func (*RedisSortedSetFlow) CancellationChecker 

func (r *RedisSortedSetFlow) CancellationChecker() api.CancellationChecker

func (*RedisSortedSetFlow) Characteristics 

func (r *RedisSortedSetFlow) Characteristics() pipeline.Characteristics

func (*RedisSortedSetFlow) HealthCheck 

func (r *RedisSortedSetFlow) HealthCheck(ctx context.Context) error

func (*RedisSortedSetFlow) QueueBacklog 

func (r *RedisSortedSetFlow) QueueBacklog(ctx context.Context) ([]pipeline.QueueBacklogStat, error)

func (*RedisSortedSetFlow) ReconfigureQueues added in v0.9.1 

func (r *RedisSortedSetFlow) ReconfigureQueues(queues []SortedSetQueueConfig, beforeCommit func([]pipeline.RequestChannel) error) (QueueReconfigureResult, error)

ReconfigureQueues swaps the live queue set against the registry in one atomic step:

  • A queue left unchanged across configs keeps its channel, its gate and its consume worker untouched.
  • A new or modified queue is validated and built before anything live changes; the fresh channel and worker start inside the registry swap, so a new queue is consumable the moment the function returns.
  • A removed or replaced queue's worker is cancelled; after it exits, the flow itself closes the queue's channel. Closing the source channel is how merge policies unregister a queue, keeping the policy's merged channel (and the inference workers reading it) untouched.

Neither Redis backlog of removed queues nor in-flight messages are lost: a worker cancelled mid-send re-enqueues the message into the queue before exiting. Empty queue slices are legal — the flow idles and can still accept queues later. Any validation error leaves the previous, last-good configuration untouched. beforeCommit runs after all preparation and the final stopped check, but before any live state changes. Once it succeeds, the remaining commit path cannot fail.

func (*RedisSortedSetFlow) RequestChannels 

func (r *RedisSortedSetFlow) RequestChannels() []pipeline.RequestChannel

func (*RedisSortedSetFlow) ResultChannel 

func (r *RedisSortedSetFlow) ResultChannel() chan api.ResultMessage

func (*RedisSortedSetFlow) RetryChannel 

func (r *RedisSortedSetFlow) RetryChannel() chan pipeline.RetryMessage

func (*RedisSortedSetFlow) Shutdown 

func (r *RedisSortedSetFlow) Shutdown()

func (*RedisSortedSetFlow) Start 

func (r *RedisSortedSetFlow) Start(ctx context.Context)

func (*RedisSortedSetFlow) StopConsuming 

func (r *RedisSortedSetFlow) StopConsuming()

func (*RedisSortedSetFlow) StopHeartbeatForTest added in v0.9.1 

func (r *RedisSortedSetFlow) StopHeartbeatForTest()

type RequestChannelData 

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

type SortedSetConfig added in v0.9.1 

type SortedSetConfig struct {
	URL string `json:"url,omitempty"`
	// RetryQueueName is the sorted set holding backoff retries, scored by
	// retry-due time. Retries re-enter their request queue (with the original
	// deadline score) only once due, so backoff is actually enforced.
	RetryQueueName  string `json:"retry_queue_name,omitempty"`
	ResultQueueName string `json:"result_queue_name,omitempty"`
	PollIntervalMs  int    `json:"poll_interval_ms,omitempty"`
	BatchSize       int    `json:"batch_size,omitempty"`
	EnableTracing   bool   `json:"enable_tracing,omitempty"`
	// ClaimLeaseTTLSeconds bounds how long one consumer may hold a dequeued
	// request before survivors treat it as abandoned and redeliver it.
	// Should exceed the longest possible inference time for the queues served.
	ClaimLeaseTTLSeconds int64 `json:"claim_lease_ttl_seconds,omitempty"`
	// ClaimReclaimIntervalMs is how often expired claims are scanned for
	// redelivery.
	ClaimReclaimIntervalMs int64                  `json:"claim_reclaim_interval_ms,omitempty"`
	Queues                 []SortedSetQueueConfig `json:"queues"`
}

SortedSetConfig is the transport config for the Redis sorted-set flow. It is parsed from JSON provided via --transport-config or --transport-config-file.

func LoadSortedSetConfig added in v0.9.1 

func LoadSortedSetConfig(data []byte) (*SortedSetConfig, error)

LoadSortedSetConfig parses, applies env overrides/defaults, and validates a SortedSetConfig.

func LoadSortedSetConfigAllowEmptyQueues added in v0.9.1 

func LoadSortedSetConfigAllowEmptyQueues(data []byte) (*SortedSetConfig, error)

LoadSortedSetConfigAllowEmptyQueues is used by queue hot reload, where an operator may temporarily drain every queue. Startup still requires a queue.

func (*SortedSetConfig) ApplyDefaults added in v0.9.1 

func (c *SortedSetConfig) ApplyDefaults()

func (*SortedSetConfig) ApplyEnvOverrides added in v0.9.1 

func (c *SortedSetConfig) ApplyEnvOverrides()

ApplyEnvOverrides seeds URL from REDIS_URL only when it is not already set in the config. An explicit url in --transport-config therefore wins over the environment; REDIS_URL is a fallback default, not an override.

func (*SortedSetConfig) Validate added in v0.9.1 

func (c *SortedSetConfig) Validate() error

type SortedSetFlowOptions 

type SortedSetFlowOptions struct {
	IGWBaseURL         string
	RequestPathURL     string
	InferenceObjective string
	RequestQueueName   string
	ResultQueueName    string
	QueuesConfig       string
	QueuesConfigFile   string
	PollIntervalMs     int
	BatchSize          int
	GateType           string
	GateParamsJSON     string
}

SortedSetFlowOptions holds CLI flags for the Redis sorted-set flow.

func NewSortedSetFlowOptions 

func NewSortedSetFlowOptions() *SortedSetFlowOptions

func (*SortedSetFlowOptions) AddFlags 

func (o *SortedSetFlowOptions) AddFlags(fs *pflag.FlagSet)

func (*SortedSetFlowOptions) HasQueueConfig 

func (o *SortedSetFlowOptions) HasQueueConfig() bool

HasQueueConfig reports whether any multi-queue configuration is set.

type SortedSetQueueConfig added in v0.9.1 

type SortedSetQueueConfig struct {
	ID              string `json:"id,omitempty"`
	QueueName       string `json:"queue_name,omitempty"`
	ResultQueueName string `json:"result_queue_name,omitempty"`
	// ResultTTLSeconds, when > 0, sets an expiry on the result destination
	// each time results are pushed. Used for per-request result keys
	// (frontend enqueue mode) so unfetched results are cleaned up. Queues
	// without it behave as before (no expiry).
	ResultTTLSeconds   int64  `json:"result_ttl_seconds,omitempty"`
	WorkerPoolID       string `json:"worker_pool_id"`
	InferenceObjective string `json:"inference_objective"`
	RequestPathURL     string `json:"request_path_url"`
	IGWBaseURL         string `json:"igw_base_url"`
	pipeline.GateConfig
	Labels map[string]string `json:"labels,omitempty"`
}

SortedSetQueueConfig defines a single queue entry in the sorted-set transport config.

type SortedSetQueueReconfigurer added in v0.9.1 

type SortedSetQueueReconfigurer interface {
	ReconfigureQueues(queues []SortedSetQueueConfig, beforeCommit func([]pipeline.RequestChannel) error) (QueueReconfigureResult, error)
}

SortedSetQueueReconfigurer is the optional capability of a Redis sorted-set flow that supports replacing its queue set at runtime, e.g. after a hot reload of a queues configuration file. beforeCommit receives every new or replacement channel after preparation but before live state is changed. If it returns an error, the old registry and workers are untouched.

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.