2015-09-15 23:54:11 +03:00
|
|
|
// Copyright 2015 CoreOS, Inc.
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
|
|
|
|
2015-09-09 19:49:28 +03:00
|
|
|
package storage
|
|
|
|
|
|
|
|
import (
|
|
|
|
"log"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
2016-01-05 21:16:50 +03:00
|
|
|
"github.com/coreos/etcd/lease"
|
2016-01-06 06:45:18 +03:00
|
|
|
"github.com/coreos/etcd/storage/backend"
|
2015-09-09 19:49:28 +03:00
|
|
|
"github.com/coreos/etcd/storage/storagepb"
|
|
|
|
)
|
|
|
|
|
2015-11-03 04:43:41 +03:00
|
|
|
const (
|
|
|
|
// chanBufLen is the length of the buffered chan
|
|
|
|
// for sending out watched events.
|
|
|
|
// TODO: find a good buf value. 1024 is just a random one that
|
|
|
|
// seems to be reasonable.
|
|
|
|
chanBufLen = 1024
|
|
|
|
)
|
|
|
|
|
|
|
|
type watchable interface {
|
2016-02-26 19:55:28 +03:00
|
|
|
watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse) (*watcher, cancelFunc)
|
2016-03-03 09:34:54 +03:00
|
|
|
progress(w *watcher)
|
2016-01-06 08:33:50 +03:00
|
|
|
rev() int64
|
2015-11-03 04:43:41 +03:00
|
|
|
}
|
|
|
|
|
2015-09-09 19:49:28 +03:00
|
|
|
type watchableStore struct {
|
|
|
|
mu sync.Mutex
|
|
|
|
|
2015-10-21 02:10:52 +03:00
|
|
|
*store
|
2015-09-09 19:49:28 +03:00
|
|
|
|
2016-01-03 07:20:22 +03:00
|
|
|
// contains all unsynced watchers that needs to sync with events that have happened
|
2016-02-26 19:55:28 +03:00
|
|
|
unsynced watcherGroup
|
2015-10-21 22:06:27 +03:00
|
|
|
|
2016-01-03 07:20:22 +03:00
|
|
|
// contains all synced watchers that are in sync with the progress of the store.
|
|
|
|
// The key of the map is the key that the watcher watches on.
|
2016-02-26 19:55:28 +03:00
|
|
|
synced watcherGroup
|
2015-09-09 19:49:28 +03:00
|
|
|
|
|
|
|
stopc chan struct{}
|
|
|
|
wg sync.WaitGroup
|
|
|
|
}
|
|
|
|
|
2016-01-05 03:17:55 +03:00
|
|
|
// cancelFunc updates unsynced and synced maps when running
|
|
|
|
// cancel operations.
|
|
|
|
type cancelFunc func()
|
|
|
|
|
2016-01-08 21:06:33 +03:00
|
|
|
func newWatchableStore(b backend.Backend, le lease.Lessor) *watchableStore {
|
2015-09-09 19:49:28 +03:00
|
|
|
s := &watchableStore{
|
2016-01-08 21:06:33 +03:00
|
|
|
store: NewStore(b, le),
|
2016-02-26 19:55:28 +03:00
|
|
|
unsynced: newWatcherGroup(),
|
|
|
|
synced: newWatcherGroup(),
|
2015-10-21 22:06:27 +03:00
|
|
|
stopc: make(chan struct{}),
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-01-22 12:40:19 +03:00
|
|
|
if s.le != nil {
|
|
|
|
// use this store as the deleter so revokes trigger watch events
|
|
|
|
s.le.SetRangeDeleter(s)
|
|
|
|
}
|
2015-09-09 19:49:28 +03:00
|
|
|
s.wg.Add(1)
|
2016-01-03 07:20:22 +03:00
|
|
|
go s.syncWatchersLoop()
|
2015-09-09 19:49:28 +03:00
|
|
|
return s
|
|
|
|
}
|
|
|
|
|
2016-01-05 21:16:50 +03:00
|
|
|
func (s *watchableStore) Put(key, value []byte, lease lease.LeaseID) (rev int64) {
|
2015-09-09 19:49:28 +03:00
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
2016-01-05 02:12:59 +03:00
|
|
|
rev = s.store.Put(key, value, lease)
|
2016-02-01 02:56:31 +03:00
|
|
|
changes := s.store.getChanges()
|
|
|
|
if len(changes) != 1 {
|
|
|
|
log.Panicf("unexpected len(changes) != 1 after put")
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-02-01 02:56:31 +03:00
|
|
|
|
2016-01-01 06:26:59 +03:00
|
|
|
ev := storagepb.Event{
|
2015-09-09 19:49:28 +03:00
|
|
|
Type: storagepb.PUT,
|
2016-02-01 02:56:31 +03:00
|
|
|
Kv: &changes[0],
|
2016-01-01 06:26:59 +03:00
|
|
|
}
|
2016-02-03 10:02:15 +03:00
|
|
|
s.notify(rev, []storagepb.Event{ev})
|
2015-09-09 19:49:28 +03:00
|
|
|
return rev
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *watchableStore) DeleteRange(key, end []byte) (n, rev int64) {
|
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
2015-10-21 02:10:52 +03:00
|
|
|
n, rev = s.store.DeleteRange(key, end)
|
2016-02-01 02:56:31 +03:00
|
|
|
changes := s.store.getChanges()
|
|
|
|
|
|
|
|
if len(changes) != int(n) {
|
|
|
|
log.Panicf("unexpected len(changes) != n after deleteRange")
|
|
|
|
}
|
|
|
|
|
|
|
|
if n == 0 {
|
|
|
|
return n, rev
|
|
|
|
}
|
|
|
|
|
|
|
|
evs := make([]storagepb.Event, n)
|
|
|
|
for i, change := range changes {
|
2016-01-01 06:26:59 +03:00
|
|
|
evs[i] = storagepb.Event{
|
2015-09-09 19:49:28 +03:00
|
|
|
Type: storagepb.DELETE,
|
2016-02-01 02:56:31 +03:00
|
|
|
Kv: &change}
|
2016-02-17 13:49:04 +03:00
|
|
|
evs[i].Kv.ModRevision = rev
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-02-03 10:02:15 +03:00
|
|
|
s.notify(rev, evs)
|
2015-09-09 19:49:28 +03:00
|
|
|
return n, rev
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *watchableStore) TxnBegin() int64 {
|
|
|
|
s.mu.Lock()
|
2015-10-21 02:10:52 +03:00
|
|
|
return s.store.TxnBegin()
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (s *watchableStore) TxnEnd(txnID int64) error {
|
2015-10-21 02:10:52 +03:00
|
|
|
err := s.store.TxnEnd(txnID)
|
2015-09-09 19:49:28 +03:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2016-02-01 02:56:31 +03:00
|
|
|
changes := s.getChanges()
|
|
|
|
if len(changes) == 0 {
|
|
|
|
s.mu.Unlock()
|
|
|
|
return nil
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-01-01 06:26:59 +03:00
|
|
|
|
2016-02-17 13:49:04 +03:00
|
|
|
rev := s.store.Rev()
|
2016-02-01 02:56:31 +03:00
|
|
|
evs := make([]storagepb.Event, len(changes))
|
|
|
|
for i, change := range changes {
|
|
|
|
switch change.Value {
|
|
|
|
case nil:
|
|
|
|
evs[i] = storagepb.Event{
|
|
|
|
Type: storagepb.DELETE,
|
|
|
|
Kv: &changes[i]}
|
2016-02-17 13:49:04 +03:00
|
|
|
evs[i].Kv.ModRevision = rev
|
2016-02-01 02:56:31 +03:00
|
|
|
default:
|
|
|
|
evs[i] = storagepb.Event{
|
|
|
|
Type: storagepb.PUT,
|
|
|
|
Kv: &changes[i]}
|
2016-01-01 06:26:59 +03:00
|
|
|
}
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-01-01 06:26:59 +03:00
|
|
|
|
2016-02-17 13:49:04 +03:00
|
|
|
s.notify(rev, evs)
|
2015-09-09 19:49:28 +03:00
|
|
|
s.mu.Unlock()
|
2016-02-01 02:56:31 +03:00
|
|
|
|
2015-09-09 19:49:28 +03:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *watchableStore) Close() error {
|
|
|
|
close(s.stopc)
|
|
|
|
s.wg.Wait()
|
2015-10-21 02:10:52 +03:00
|
|
|
return s.store.Close()
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
|
|
|
|
2016-01-03 02:59:32 +03:00
|
|
|
func (s *watchableStore) NewWatchStream() WatchStream {
|
|
|
|
watchStreamGauge.Inc()
|
|
|
|
return &watchStream{
|
2015-11-03 04:43:41 +03:00
|
|
|
watchable: s,
|
2016-01-03 23:48:58 +03:00
|
|
|
ch: make(chan WatchResponse, chanBufLen),
|
2016-01-05 06:50:17 +03:00
|
|
|
cancels: make(map[WatchID]cancelFunc),
|
2016-03-03 09:34:54 +03:00
|
|
|
watchers: make(map[WatchID]*watcher),
|
2015-11-03 04:43:41 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-02-26 19:55:28 +03:00
|
|
|
func (s *watchableStore) watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse) (*watcher, cancelFunc) {
|
2015-09-09 19:49:28 +03:00
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
2016-01-03 07:20:22 +03:00
|
|
|
wa := &watcher{
|
2016-02-26 19:55:28 +03:00
|
|
|
key: key,
|
|
|
|
end: end,
|
|
|
|
cur: startRev,
|
|
|
|
id: id,
|
|
|
|
ch: ch,
|
2015-11-03 04:43:41 +03:00
|
|
|
}
|
|
|
|
|
2016-02-17 14:17:37 +03:00
|
|
|
s.store.mu.Lock()
|
|
|
|
synced := startRev > s.store.currentRev.main || startRev == 0
|
|
|
|
if synced {
|
|
|
|
wa.cur = s.store.currentRev.main + 1
|
|
|
|
}
|
|
|
|
s.store.mu.Unlock()
|
|
|
|
if synced {
|
|
|
|
if startRev > wa.cur {
|
|
|
|
panic("can't watch past sync revision")
|
|
|
|
}
|
2016-02-03 06:15:46 +03:00
|
|
|
s.synced.add(wa)
|
2015-09-09 19:49:28 +03:00
|
|
|
} else {
|
2016-01-03 07:20:22 +03:00
|
|
|
slowWatcherGauge.Inc()
|
2016-02-03 07:06:53 +03:00
|
|
|
s.unsynced.add(wa)
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-01-03 07:20:22 +03:00
|
|
|
watcherGauge.Inc()
|
2015-09-09 19:49:28 +03:00
|
|
|
|
2016-01-05 03:17:55 +03:00
|
|
|
cancel := cancelFunc(func() {
|
2015-09-09 19:49:28 +03:00
|
|
|
s.mu.Lock()
|
2015-09-23 09:25:07 +03:00
|
|
|
defer s.mu.Unlock()
|
2016-02-03 07:06:53 +03:00
|
|
|
// remove references of the watcher
|
|
|
|
if s.unsynced.delete(wa) {
|
2016-01-03 07:20:22 +03:00
|
|
|
slowWatcherGauge.Dec()
|
|
|
|
watcherGauge.Dec()
|
2015-10-21 22:06:27 +03:00
|
|
|
return
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
|
|
|
|
2016-02-03 06:28:42 +03:00
|
|
|
if s.synced.delete(wa) {
|
|
|
|
watcherGauge.Dec()
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
|
|
|
// If we cannot find it, it should have finished watch.
|
|
|
|
})
|
|
|
|
|
|
|
|
return wa, cancel
|
|
|
|
}
|
|
|
|
|
2016-02-01 08:42:39 +03:00
|
|
|
// syncWatchersLoop syncs the watcher in the unsynced map every 100ms.
|
2016-01-03 07:20:22 +03:00
|
|
|
func (s *watchableStore) syncWatchersLoop() {
|
2015-09-09 19:49:28 +03:00
|
|
|
defer s.wg.Done()
|
|
|
|
|
|
|
|
for {
|
|
|
|
s.mu.Lock()
|
2016-01-03 07:20:22 +03:00
|
|
|
s.syncWatchers()
|
2015-09-09 19:49:28 +03:00
|
|
|
s.mu.Unlock()
|
|
|
|
|
|
|
|
select {
|
|
|
|
case <-time.After(100 * time.Millisecond):
|
|
|
|
case <-s.stopc:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-01-03 07:20:22 +03:00
|
|
|
// syncWatchers periodically syncs unsynced watchers by: Iterate all unsynced
|
|
|
|
// watchers to get the minimum revision within its range, skipping the
|
|
|
|
// watcher if its current revision is behind the compact revision of the
|
2015-12-22 20:30:40 +03:00
|
|
|
// store. And use this minimum revision to get all key-value pairs. Then send
|
2016-01-03 07:20:22 +03:00
|
|
|
// those events to watchers.
|
|
|
|
func (s *watchableStore) syncWatchers() {
|
2015-12-22 20:30:40 +03:00
|
|
|
s.store.mu.Lock()
|
|
|
|
defer s.store.mu.Unlock()
|
|
|
|
|
2016-02-26 19:55:28 +03:00
|
|
|
if s.unsynced.size() == 0 {
|
2015-12-22 20:30:40 +03:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-01-03 07:20:22 +03:00
|
|
|
// in order to find key-value pairs from unsynced watchers, we need to
|
2015-12-22 20:30:40 +03:00
|
|
|
// find min revision index, and these revisions can be used to
|
|
|
|
// query the backend store of key-value pairs
|
2016-02-17 13:49:04 +03:00
|
|
|
curRev := s.store.currentRev.main
|
2016-02-26 19:55:28 +03:00
|
|
|
compactionRev := s.store.compactMainRev
|
|
|
|
minRev := s.unsynced.scanMinRev(curRev, compactionRev)
|
2016-02-17 13:49:04 +03:00
|
|
|
minBytes, maxBytes := newRevBytes(), newRevBytes()
|
|
|
|
revToBytes(revision{main: minRev}, minBytes)
|
|
|
|
revToBytes(revision{main: curRev + 1}, maxBytes)
|
|
|
|
|
|
|
|
// UnsafeRange returns keys and values. And in boltdb, keys are revisions.
|
|
|
|
// values are actual key-value pairs in backend.
|
|
|
|
tx := s.store.b.BatchTx()
|
|
|
|
tx.Lock()
|
|
|
|
revs, vs := tx.UnsafeRange(keyBucketName, minBytes, maxBytes, 0)
|
2016-02-26 19:55:28 +03:00
|
|
|
evs := kvsToEvents(&s.unsynced, revs, vs)
|
2016-02-17 13:49:04 +03:00
|
|
|
tx.Unlock()
|
2015-12-22 20:30:40 +03:00
|
|
|
|
2016-03-03 09:34:54 +03:00
|
|
|
wb := newWatcherBatch(&s.unsynced, evs)
|
|
|
|
|
|
|
|
for w, eb := range wb {
|
2016-02-17 13:49:04 +03:00
|
|
|
select {
|
|
|
|
// s.store.Rev also uses Lock, so just return directly
|
2016-02-18 15:12:40 +03:00
|
|
|
case w.ch <- WatchResponse{WatchID: w.id, Events: eb.evs, Revision: s.store.currentRev.main}:
|
|
|
|
pendingEventsGauge.Add(float64(len(eb.evs)))
|
2016-02-17 13:49:04 +03:00
|
|
|
default:
|
|
|
|
// TODO: handle the full unsynced watchers.
|
|
|
|
// continue to process other watchers for now, the full ones
|
|
|
|
// will be processed next time and hopefully it will not be full.
|
|
|
|
continue
|
|
|
|
}
|
2016-02-18 15:12:40 +03:00
|
|
|
if eb.moreRev != 0 {
|
|
|
|
w.cur = eb.moreRev
|
|
|
|
continue
|
|
|
|
}
|
2016-02-17 13:49:04 +03:00
|
|
|
w.cur = curRev
|
|
|
|
s.synced.add(w)
|
|
|
|
s.unsynced.delete(w)
|
|
|
|
}
|
|
|
|
|
2016-03-03 09:34:54 +03:00
|
|
|
// bring all un-notified watchers to synced.
|
|
|
|
for w := range s.unsynced.watchers {
|
|
|
|
if !wb.contains(w) {
|
|
|
|
w.cur = curRev
|
|
|
|
s.synced.add(w)
|
|
|
|
s.unsynced.delete(w)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-02-26 19:55:28 +03:00
|
|
|
slowWatcherGauge.Set(float64(s.unsynced.size()))
|
2016-02-17 13:49:04 +03:00
|
|
|
}
|
2016-01-01 06:26:59 +03:00
|
|
|
|
2016-02-17 13:49:04 +03:00
|
|
|
// kvsToEvents gets all events for the watchers from all key-value pairs
|
2016-02-26 19:55:28 +03:00
|
|
|
func kvsToEvents(wg *watcherGroup, revs, vals [][]byte) (evs []storagepb.Event) {
|
2016-02-17 13:49:04 +03:00
|
|
|
for i, v := range vals {
|
2015-12-22 20:30:40 +03:00
|
|
|
var kv storagepb.KeyValue
|
|
|
|
if err := kv.Unmarshal(v); err != nil {
|
|
|
|
log.Panicf("storage: cannot unmarshal event: %v", err)
|
|
|
|
}
|
|
|
|
|
2016-02-26 19:55:28 +03:00
|
|
|
if !wg.contains(string(kv.Key)) {
|
2015-09-09 19:49:28 +03:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2016-02-17 13:49:04 +03:00
|
|
|
ty := storagepb.PUT
|
|
|
|
if isTombstone(revs[i]) {
|
|
|
|
ty = storagepb.DELETE
|
|
|
|
// patch in mod revision so watchers won't skip
|
|
|
|
kv.ModRevision = bytesToRev(revs[i]).main
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-02-17 13:49:04 +03:00
|
|
|
evs = append(evs, storagepb.Event{Kv: &kv, Type: ty})
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
2016-02-17 13:49:04 +03:00
|
|
|
return evs
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// notify notifies the fact that given event at the given rev just happened to
|
2016-01-03 07:20:22 +03:00
|
|
|
// watchers that watch on the key of the event.
|
2016-01-01 06:26:59 +03:00
|
|
|
func (s *watchableStore) notify(rev int64, evs []storagepb.Event) {
|
2016-02-26 19:55:28 +03:00
|
|
|
for w, eb := range newWatcherBatch(&s.synced, evs) {
|
|
|
|
if eb.revs != 1 {
|
|
|
|
panic("unexpected multiple revisions in notification")
|
|
|
|
}
|
|
|
|
select {
|
|
|
|
case w.ch <- WatchResponse{WatchID: w.id, Events: eb.evs, Revision: s.Rev()}:
|
|
|
|
pendingEventsGauge.Add(float64(len(eb.evs)))
|
|
|
|
default:
|
|
|
|
// move slow watcher to unsynced
|
|
|
|
w.cur = rev
|
|
|
|
s.unsynced.add(w)
|
|
|
|
s.synced.delete(w)
|
|
|
|
slowWatcherGauge.Inc()
|
2015-09-09 19:49:28 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-01-06 08:33:50 +03:00
|
|
|
func (s *watchableStore) rev() int64 { return s.store.Rev() }
|
|
|
|
|
2016-03-03 09:34:54 +03:00
|
|
|
func (s *watchableStore) progress(w *watcher) {
|
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
|
|
|
if _, ok := s.synced.watchers[w]; ok {
|
|
|
|
select {
|
|
|
|
case w.ch <- WatchResponse{WatchID: w.id, Revision: s.rev()}:
|
|
|
|
default:
|
|
|
|
// If the ch is full, this watcher is receiving events.
|
|
|
|
// We do not need to send progress at all.
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-01-03 07:20:22 +03:00
|
|
|
type watcher struct {
|
|
|
|
// the watcher key
|
2015-11-03 04:43:41 +03:00
|
|
|
key []byte
|
2016-02-26 19:55:28 +03:00
|
|
|
// end indicates the end of the range to watch.
|
|
|
|
// If end is set, the watcher is on a range.
|
|
|
|
end []byte
|
2016-01-03 07:20:22 +03:00
|
|
|
// cur is the current watcher revision.
|
2015-11-03 04:43:41 +03:00
|
|
|
// If cur is behind the current revision of the KV,
|
2016-01-03 07:20:22 +03:00
|
|
|
// watcher is unsynced and needs to catch up.
|
2015-11-03 04:43:41 +03:00
|
|
|
cur int64
|
2016-01-05 06:50:17 +03:00
|
|
|
id WatchID
|
2015-11-03 04:43:41 +03:00
|
|
|
|
2016-01-03 23:48:58 +03:00
|
|
|
// a chan to send out the watch response.
|
2016-01-03 07:20:22 +03:00
|
|
|
// The chan might be shared with other watchers.
|
2016-01-03 23:48:58 +03:00
|
|
|
ch chan<- WatchResponse
|
2015-11-03 04:43:41 +03:00
|
|
|
}
|