hh

Documentation

Overview

Package hh implements a hinted handoff for writes

Index

• Constants
• Variables
• func NewRateLimiter(limit int64) *limiter
• type Config
  • func NewConfig() Config
• type Processor
  • func NewProcessor(dir string, writer shardWriter, metastore metaStore, options ProcessorOptions) (*Processor, error)
  • func (p *Processor) Process() error
  • func (p *Processor) PurgeInactiveOlderThan(when time.Duration) error
  • func (p *Processor) PurgeOlderThan(when time.Duration) error
  • func (p *Processor) WriteShard(shardID, ownerID uint64, points []models.Point) error
• type ProcessorOptions
• type Service
  • func NewService(c Config, w shardWriter, m metaStore) *Service
  • func (s *Service) Close() error
  • func (s *Service) Open() error
  • func (s *Service) SetLogger(l *log.Logger)
  • func (s *Service) WriteShard(shardID, ownerID uint64, points []models.Point) error

Constants

const (
	// DefaultMaxSize is the default maximum size of all hinted handoff queues in bytes.
	DefaultMaxSize = 1024 * 1024 * 1024

	// DefaultMaxAge is the default maximum amount of time that a hinted handoff write
	// can stay in the queue.  After this time, the write will be purged.
	DefaultMaxAge = 7 * 24 * time.Hour

	// DefaultRetryRateLimit is the default rate that hinted handoffs will be retried.
	// The rate is in bytes per second and applies across all nodes when retried.   A
	// value of 0 disables the rate limit.
	DefaultRetryRateLimit = 0

	// DefaultRetryInterval is the default amount of time the system waits before
	// attempting to flush hinted handoff queues. With each failure of a hinted
	// handoff write, this retry interval increases exponentially until it reaches
	// the maximum
	DefaultRetryInterval = time.Second

	// DefaultRetryMaxInterval is the maximum the hinted handoff retry interval
	// will ever be.
	DefaultRetryMaxInterval = time.Minute
)

Variables

var (
	ErrNotOpen     = fmt.Errorf("queue not open")
	ErrQueueFull   = fmt.Errorf("queue is full")
	ErrSegmentFull = fmt.Errorf("segment is full")
)

var ErrHintedHandoffDisabled = fmt.Errorf("hinted handoff disabled")

Functions

func NewRateLimiter

func NewRateLimiter(limit int64) *limiter

NewRateLimiter returns a new limiter configured to restrict a process to the limit per second. limit is the maximum amount that can be used per second. The limit should be > 0. A limit <= 0, will not limit the processes.

Types

type Config

type Config struct {
	Enabled          bool          `toml:"enabled"`
	Dir              string        `toml:"dir"`
	MaxSize          int64         `toml:"max-size"`
	MaxAge           toml.Duration `toml:"max-age"`
	RetryRateLimit   int64         `toml:"retry-rate-limit"`
	RetryInterval    toml.Duration `toml:"retry-interval"`
	RetryMaxInterval toml.Duration `toml:"retry-max-interval"`
}

func NewConfig

func NewConfig() Config

type Processor

type Processor struct {
	Logger *log.Logger
	// contains filtered or unexported fields
}

func NewProcessor

func NewProcessor(dir string, writer shardWriter, metastore metaStore, options ProcessorOptions) (*Processor, error)

func (*Processor) Process

func (p *Processor) Process() error

func (*Processor) PurgeInactiveOlderThan

func (p *Processor) PurgeInactiveOlderThan(when time.Duration) error

func (*Processor) PurgeOlderThan

func (p *Processor) PurgeOlderThan(when time.Duration) error

func (*Processor) WriteShard

func (p *Processor) WriteShard(shardID, ownerID uint64, points []models.Point) error

WriteShard writes hinted-handoff data for the given shard and node. Since it may manipulate hinted-handoff queues, and be called concurrently, it takes a lock during queue access.

type ProcessorOptions

type ProcessorOptions struct {
	MaxSize        int64
	RetryRateLimit int64
}

type Service

type Service struct {
	Logger *log.Logger

	ShardWriter shardWriter

	HintedHandoff interface {
		WriteShard(shardID, ownerID uint64, points []models.Point) error
		Process() error
		PurgeOlderThan(when time.Duration) error
		PurgeInactiveOlderThan(when time.Duration) error
	}
	// contains filtered or unexported fields
}

func NewService

func NewService(c Config, w shardWriter, m metaStore) *Service

NewService returns a new instance of Service.

func (*Service) Close

func (s *Service) Close() error

func (*Service) Open

func (s *Service) Open() error

func (*Service) SetLogger

func (s *Service) SetLogger(l *log.Logger)

SetLogger sets the internal logger to the logger passed in.

func (*Service) WriteShard

func (s *Service) WriteShard(shardID, ownerID uint64, points []models.Point) error

WriteShard queues the points write for shardID to node ownerID to handoff queue