|
| 1 | +package ctlstore |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "errors" |
| 6 | + "fmt" |
| 7 | + "github.com/segmentio/ctlstore/pkg/errs" |
| 8 | + "github.com/segmentio/ctlstore/pkg/globalstats" |
| 9 | + "github.com/segmentio/ctlstore/pkg/ldb" |
| 10 | + "github.com/segmentio/events/v2" |
| 11 | + "github.com/segmentio/stats/v4" |
| 12 | + "path" |
| 13 | + "strconv" |
| 14 | + "sync/atomic" |
| 15 | + "time" |
| 16 | +) |
| 17 | + |
| 18 | +// LDBRotatingReader reads data from multiple LDBs on a rotating schedule. |
| 19 | +// The main benefit is relieving read pressure on a particular ldb file when it becomes inactive, |
| 20 | +// allowing sqlite maintenance |
| 21 | +type LDBRotatingReader struct { |
| 22 | + active int32 |
| 23 | + dbs []*LDBReader |
| 24 | + schedule []int8 |
| 25 | + now func() time.Time |
| 26 | + tickerInterval time.Duration |
| 27 | +} |
| 28 | + |
| 29 | +// RotationPeriod how many minutes each reader is active for before rotating to the next |
| 30 | +type RotationPeriod int |
| 31 | + |
| 32 | +const ( |
| 33 | + // Every30 rotate on 30 minute mark in an hour |
| 34 | + Every30 RotationPeriod = 30 |
| 35 | + // Every20 rotate on 20 minute marks in an hour |
| 36 | + Every20 RotationPeriod = 20 |
| 37 | + // Every15 rotate on 15 minute marks in an hour |
| 38 | + Every15 RotationPeriod = 15 |
| 39 | + // Every10 rotate on 10 minute marks in an hour |
| 40 | + Every10 RotationPeriod = 10 |
| 41 | + // Every6 rotate on 6 minute marks in an hour |
| 42 | + Every6 RotationPeriod = 6 |
| 43 | + |
| 44 | + // for simpler migration, the first ldb retains the original name |
| 45 | + defaultPath = DefaultCtlstorePath + ldb.DefaultLDBFilename |
| 46 | + ldbFormat = DefaultCtlstorePath + "ldb-%d.db" |
| 47 | +) |
| 48 | + |
| 49 | +func defaultPaths(count int) []string { |
| 50 | + paths := []string{defaultPath} |
| 51 | + for i := 1; i < count; i++ { |
| 52 | + paths = append(paths, fmt.Sprintf(ldbFormat, i+1)) |
| 53 | + } |
| 54 | + return paths |
| 55 | +} |
| 56 | + |
| 57 | +// RotatingReader creates a new reader that rotates which ldb it reads from on a rotation period with the default location in /var/spool/ctlstore |
| 58 | +func RotatingReader(ctx context.Context, minutesPerRotation RotationPeriod, ldbsCount int) (RowRetriever, error) { |
| 59 | + return CustomerRotatingReader(ctx, minutesPerRotation, defaultPaths(ldbsCount)...) |
| 60 | +} |
| 61 | + |
| 62 | +// CustomerRotatingReader creates a new reader that rotates which ldb it reads from on a rotation period |
| 63 | +func CustomerRotatingReader(ctx context.Context, minutesPerRotation RotationPeriod, ldbPaths ...string) (RowRetriever, error) { |
| 64 | + r, err := rotatingReader(minutesPerRotation, ldbPaths...) |
| 65 | + if err != nil { |
| 66 | + return nil, err |
| 67 | + } |
| 68 | + r.setActive() |
| 69 | + go r.rotate(ctx) |
| 70 | + return r, nil |
| 71 | +} |
| 72 | + |
| 73 | +func rotatingReader(minutesPerRotation RotationPeriod, ldbPaths ...string) (*LDBRotatingReader, error) { |
| 74 | + if len(ldbPaths) < 2 { |
| 75 | + return nil, errors.New("RotatingReader requires more than 1 ldb") |
| 76 | + } |
| 77 | + if !isValid(minutesPerRotation) { |
| 78 | + return nil, errors.New(fmt.Sprintf("invalid rotation period: %v", minutesPerRotation)) |
| 79 | + } |
| 80 | + if len(ldbPaths) > 60/int(minutesPerRotation) { |
| 81 | + return nil, errors.New("cannot have more ldbs than rotations per hour") |
| 82 | + } |
| 83 | + var r LDBRotatingReader |
| 84 | + for _, p := range ldbPaths { |
| 85 | + events.Log("Opening ldb %s for reading", p) |
| 86 | + reader, err := newLDBReader(p) |
| 87 | + if err != nil { |
| 88 | + return nil, err |
| 89 | + } |
| 90 | + r.dbs = append(r.dbs, reader) |
| 91 | + } |
| 92 | + r.schedule = make([]int8, 60) |
| 93 | + idx := 0 |
| 94 | + for i := 1; i < 61; i++ { |
| 95 | + r.schedule[i-1] = int8(idx % len(ldbPaths)) |
| 96 | + if i%int(minutesPerRotation) == 0 { |
| 97 | + idx++ |
| 98 | + } |
| 99 | + } |
| 100 | + return &r, nil |
| 101 | +} |
| 102 | + |
| 103 | +func (r *LDBRotatingReader) setActive() { |
| 104 | + if r.now == nil { |
| 105 | + r.now = time.Now |
| 106 | + } |
| 107 | + atomic.StoreInt32(&r.active, int32(r.schedule[r.now().Minute()])) |
| 108 | +} |
| 109 | + |
| 110 | +// GetRowsByKeyPrefix delegates to the active LDBReader |
| 111 | +func (r *LDBRotatingReader) GetRowsByKeyPrefix(ctx context.Context, familyName string, tableName string, key ...interface{}) (*Rows, error) { |
| 112 | + return r.dbs[atomic.LoadInt32(&r.active)].GetRowsByKeyPrefix(ctx, familyName, tableName, key...) |
| 113 | +} |
| 114 | + |
| 115 | +// GetRowByKey delegates to the active LDBReader |
| 116 | +func (r *LDBRotatingReader) GetRowByKey(ctx context.Context, out interface{}, familyName string, tableName string, key ...interface{}) (found bool, err error) { |
| 117 | + return r.dbs[atomic.LoadInt32(&r.active)].GetRowByKey(ctx, out, familyName, tableName, key...) |
| 118 | +} |
| 119 | + |
| 120 | +// rotate by default checks every 1 minute if the active db has changed according to schedule |
| 121 | +func (r *LDBRotatingReader) rotate(ctx context.Context) { |
| 122 | + if r.tickerInterval == 0 { |
| 123 | + r.tickerInterval = 1 * time.Minute |
| 124 | + } |
| 125 | + ticker := time.NewTicker(r.tickerInterval) |
| 126 | + |
| 127 | + for { |
| 128 | + select { |
| 129 | + case <-ctx.Done(): |
| 130 | + return |
| 131 | + case <-ticker.C: |
| 132 | + next := r.schedule[r.now().Minute()] |
| 133 | + last := atomic.LoadInt32(&r.active) |
| 134 | + |
| 135 | + // move the next to active and close and reopen the last one |
| 136 | + if int32(next) != last { |
| 137 | + atomic.StoreInt32(&r.active, int32(next)) |
| 138 | + stats.Incr("rotating_reader.rotate") |
| 139 | + globalstats.Set("rotating_reader.active", next) |
| 140 | + err := r.dbs[last].Close() |
| 141 | + if err != nil { |
| 142 | + events.Log("failed to close LDBReader for %s on rotation: %{error}v", r.dbs[last].path, err) |
| 143 | + errs.Incr("rotating_reader.closing_ldbreader", stats.T("id", strconv.Itoa(int(last)))) |
| 144 | + return |
| 145 | + } |
| 146 | + |
| 147 | + reader, err := newLDBReader(r.dbs[last].path) |
| 148 | + if err != nil { |
| 149 | + events.Log("failed to open LDBReader for %s on rotation: %{error}v", r.dbs[last].path, err) |
| 150 | + errs.Incr("rotating_reader.opening_ldbreader", |
| 151 | + stats.T("id", strconv.Itoa(int(last))), |
| 152 | + stats.T("path", path.Base(r.dbs[last].path))) |
| 153 | + return |
| 154 | + } |
| 155 | + r.dbs[last] = reader |
| 156 | + |
| 157 | + } |
| 158 | + } |
| 159 | + } |
| 160 | +} |
| 161 | + |
| 162 | +func isValid(rf RotationPeriod) bool { |
| 163 | + switch rf { |
| 164 | + case Every6, Every10, Every15, Every20, Every30: |
| 165 | + return true |
| 166 | + } |
| 167 | + return false |
| 168 | +} |
0 commit comments