cagrr

README

Cassandra Repair Scheduler

Anti-entropy Cassandra cluster tool. It uses repair service written in Java.

Prerequisites

You need cajrr up and running.

Run tests:

make integration   # Make a "hole" and check existence

make check         # Only check, no write/restart cycle

Repair your cluster:

cagrr -k keyspace

Analyze your logs in Kibana interface available at:

http://172.16.237.50:5601

Check your metrics in Grafana interface available at:

http://172.16.237.30:3000

##Troubleshooting

1. Elasticsearch didn't start with vm.max_map_count error:

   Run on your host machine:

   sudo sysctl -w vm.max_map_count=262144
   

Documentation

Index

• type Closer
• type Cluster
  • func (c *Cluster) RegulateWith(r Regulator) Scheduler
  • func (c *Cluster) RunRepair(repair *Repair) error
  • func (c *Cluster) Schedule()
  • func (c *Cluster) TrackIn(t Tracker) Scheduler
  • func (c *Cluster) Until(done chan bool) Scheduler
• type ClusterStats
• type Config
  • func ReadConfiguration(filename string) (*Config, error)
• type DB
  • func NewConsulDb() DB
  • func NewRedisDb(addr, password string, db int) DB
• type DurationQueue
  • func NewQueue(size int) DurationQueue
• type Fragment
• type Keyspace
  • func (k *Keyspace) SetTables(tables []*Table)
  • func (k *Keyspace) SetTotal(total int)
  • func (k *Keyspace) Tables() []*Table
  • func (k *Keyspace) Total() int
• type Logger
  • func NewLogger(verb, filename string) Logger
• type Regulator
  • func NewRegulator(size int) Regulator
• type Repair
• type RepairStats
• type RepairStatus
• type Scheduler
• type Server
  • func NewServer(tracker Tracker) Server
• type Table
  • func (t *Table) Repairs() []*Repair
  • func (t *Table) SetRepairs(repairs []*Repair)
  • func (t *Table) SetTotal(total int)
  • func (t *Table) Total() int
• type Token
• type TokenSet
• type Track
  • func (t *Track) CheckCompletion()
  • func (t *Track) Complete(duration time.Duration) (int, int, time.Duration, float32, time.Duration, time.Duration)
  • func (t *Track) IsNew() bool
  • func (t *Track) IsRepaired(threshold time.Duration) bool
  • func (t *Track) IsSpoiled(threshold time.Duration) bool
  • func (t *Track) Restart()
  • func (t *Track) Skip()
  • func (t *Track) Start(total int)
• type Tracker
  • func NewTracker(db DB, r Regulator) Tracker
• type ValueReader
• type ValueWriter

Types

type Closer

type Closer interface {
	Close()
}

Closer closes DB connection

type Cluster

type Cluster struct {
	ID        int
	Name      string      `yaml:"name"`
	Interval  string      `yaml:"interval"`
	Keyspaces []*Keyspace `yaml:"keyspaces"`
	Host      string
	Port      int
	// contains filtered or unexported fields
}

Cluster contains configuration of cluster item

func (*Cluster) RegulateWith

func (c *Cluster) RegulateWith(r Regulator) Scheduler

RegulateWith given rate limiter

func (*Cluster) RunRepair

func (c *Cluster) RunRepair(repair *Repair) error

RunRepair runs fragment repair

func (*Cluster) Schedule

func (c *Cluster) Schedule()

Schedule cluster repair

func (*Cluster) TrackIn

func (c *Cluster) TrackIn(t Tracker) Scheduler

TrackIn given tracker

func (*Cluster) Until

func (c *Cluster) Until(done chan bool) Scheduler

Until sets chan for done event

type ClusterStats

type ClusterStats struct {
	Cluster            string
	ClusterDuration    time.Duration
	LastClusterSuccess time.Time
}

ClusterStats for logging

type Config

type Config struct {
	BufferLength int        `yaml:"buffer"`
	Clusters     []*Cluster `yaml:"clusters"`
}

Config is a configuration file struct

func ReadConfiguration

func ReadConfiguration(filename string) (*Config, error)

ReadConfiguration parses yaml configuration file

type DB

type DB interface {
	CreateKey(keys ...string) string
	ValueReader
	ValueWriter
	Closer
}

DB implements DB interface

func NewConsulDb

func NewConsulDb() DB

NewConsulDb connects to DB

func NewRedisDb

func NewRedisDb(addr, password string, db int) DB

NewRedisDb connects to DB

type DurationQueue

type DurationQueue interface {
	Push(time.Duration)
	Pop() time.Duration
	Len() int
	Average() time.Duration
}

DurationQueue is a fixed size FIFO queue of durations

func NewQueue

func NewQueue(size int) DurationQueue

NewQueue returns a new queue with the given initial size.

type Fragment

type Fragment struct {
	ID       int `json:"id"`
	Endpoint string
	Start    string
	End      string
}

Fragment of Token range for repair

type Keyspace

type Keyspace struct {
	Name string `yaml:"name"`
	// contains filtered or unexported fields
}

Keyspace contains keyspace repair schedule description

func (*Keyspace) SetTables

func (k *Keyspace) SetTables(tables []*Table)

SetTables to keyspace

func (*Keyspace) SetTotal

func (k *Keyspace) SetTotal(total int)

SetTotal to keyspace

func (*Keyspace) Tables

func (k *Keyspace) Tables() []*Table

Tables of keyspace

func (*Keyspace) Total

func (k *Keyspace) Total() int

Total repairs in keyspace

type Logger

type Logger interface {
	WithError(err error) Logger
	WithFields(str interface{}) Logger
	Debug(message interface{}) Logger
	Error(message interface{}) Logger
	Fatal(message interface{}) Logger
	Warn(message interface{}) Logger
	Info(message interface{}) Logger
}

Logger logs messages

func NewLogger

func NewLogger(verb, filename string) Logger

NewLogger cretes an implementation of Logger

type Regulator

type Regulator interface {
	LimitRateTo(key string, duration time.Duration) time.Duration
	Limit(key string)
	Rate(key string) time.Duration
}

Regulator moderates the process

func NewRegulator

func NewRegulator(size int) Regulator

NewRegulator initializes new stability service object

type Repair

type Repair struct {
	ID       int    `json:"id"`
	Cluster  string `json:"cluster"`
	Keyspace string `json:"keyspace"`
	Table    string `json:"table"`
	Endpoint string `json:"endpoint"`
	Start    string `json:"start"`
	End      string `json:"end"`
}

Repair object

type RepairStats

type RepairStats struct {
	Cluster            string
	Keyspace           string
	Table              string
	ID                 int
	Duration           time.Duration
	Rate               time.Duration
	TableTotal         int
	TableCompleted     int
	TablePercent       float32
	TableDuration      time.Duration
	TableAverage       time.Duration
	TableEstimate      time.Duration
	KeyspaceTotal      int
	KeyspaceCompleted  int
	KeyspacePercent    float32
	KeyspaceDuration   time.Duration
	KeyspaceAverage    time.Duration
	KeyspaceEstimate   time.Duration
	ClusterTotal       int
	ClusterCompleted   int
	ClusterPercent     float32
	ClusterDuration    time.Duration
	ClusterAverage     time.Duration
	ClusterEstimate    time.Duration
	LastClusterSuccess time.Time
}

RepairStats for logging

type RepairStatus

type RepairStatus struct {
	Repair  Repair
	Message string
	Type    string
}

RepairStatus keeps status of repair

type Scheduler

type Scheduler interface {
	RegulateWith(Regulator) Scheduler
	Schedule()
	TrackIn(Tracker) Scheduler
	Until(chan bool) Scheduler
}

Scheduler creates jobs in time

type Server

type Server interface {
	ServeAt(callback string) Server
}

Server serves repair handlers

func NewServer

func NewServer(tracker Tracker) Server

NewServer initializes loops for scheduling repair jobs

type Table

type Table struct {
	Name   string  `yaml:"name"`
	Size   int64   `yaml:"size"`
	Slices int     `yaml:"slices"`
	Weight float32 `yaml:"weight"`
	// contains filtered or unexported fields
}

Table contains column families to repair

func (*Table) Repairs

func (t *Table) Repairs() []*Repair

Repairs of table

func (*Table) SetRepairs

func (t *Table) SetRepairs(repairs []*Repair)

SetRepairs to table

func (*Table) SetTotal

func (t *Table) SetTotal(total int)

SetTotal to table

func (*Table) Total

func (t *Table) Total() int

Total repairs in table

type Token

type Token struct {
	ID     string `json:"id"`
	Ranges []Fragment
}

Token represents cassandra key range

type TokenSet

type TokenSet []Token

TokenSet is a set of Token

type Track

type Track struct {
	Completed bool
	Count     int
	Total     int
	Percent   float32
	Duration  time.Duration
	Average   time.Duration
	Estimate  time.Duration
	Rate      time.Duration
	Finished  time.Time
	Started   time.Time
}

Track of repair item

func (*Track) CheckCompletion

func (t *Track) CheckCompletion()

CheckCompletion of repair

func (*Track) Complete

func (t *Track) Complete(duration time.Duration) (int, int, time.Duration, float32, time.Duration, time.Duration)

Complete repair fragment

func (*Track) IsNew

func (t *Track) IsNew() bool

IsNew object

func (*Track) IsRepaired

func (t *Track) IsRepaired(threshold time.Duration) bool

IsRepaired is check for repair completeness

func (*Track) IsSpoiled

func (t *Track) IsSpoiled(threshold time.Duration) bool

IsSpoiled checks that repair stinks

func (*Track) Restart

func (t *Track) Restart()

Restart track

func (*Track) Skip

func (t *Track) Skip()

Skip track

func (*Track) Start

func (t *Track) Start(total int)

Start track

type Tracker

type Tracker interface {
	Complete(cluster, keyspace, table string, repair int) *RepairStats
	IsCompleted(cluster, keyspace, table string, repair int, threshold time.Duration) bool
	Restart(cluster, keyspace, table string, repair int)
	Skip(cluster, keyspace, table string, repair int)
	Start(cluster, keyspace, table string, repair int)
	StartTable(cluster, keyspace, table string, total int)
	StartKeyspace(cluster, keyspace string, total int)
	StartCluster(cluster string, total int)
}

Tracker keeps progress of repair

func NewTracker

func NewTracker(db DB, r Regulator) Tracker

NewTracker created new progress tracker

type ValueReader

type ValueReader interface {
	ReadValue(string, string) []byte
}

ValueReader reads position data from DB

type ValueWriter

type ValueWriter interface {
	WriteValue(string, string, []byte) error
}

ValueWriter writes position to DB