mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-10-06 10:57:51 +00:00
d1752a7af7
This is a significant overhaul! Quite a lot of things changed here. Basically: Prior to this, every request for data from etcd implies requesting the current live data from etcd, and then unpacking it or extracting it in some way. This is expensive, which is why we have a cache in front of it. We don't need to do that! We can use a Watch, which notifies us of changes as changes happen. However, there's some challenges and difficulties along the way, and there's a couple of other changes which are included here because it's a pain to try to separate them out. 1. We require a logger to be provided to create our internal Etcd wrapper. We then use that logger, instead of `fmt.Printf`. This makes debugging messages work better, and also diagnostics, and so on. 2. The internal client that we are reusing can enter a failed state after a leader election, in which case we have to recreate the client to have a working client. We add a new internal-use method, `retryClient`, which wraps a function which takes an etcd client and returns an error, and checks for leader-election type errors and retries creating the client when they happen. That last bit has not been successfully tested because it's actually really hard to trigger this now. (Because it was related in part to the amount of etcd traffic we were producing, which is reduced.) 3. The general swap over from looking things up to unpacking things as they come in, then returning those already-unpacked things when we get requests. With this change, *many tests will fail*. That is addressed by a separate commit which addresses the secondary problem, which is that some of our test harness code was relying on the assumption that if any node in a cluster thinks the cluster is up, every node will. That was usually true when we were doing everything as expensive fully-synchronized cluster checks, but becomes significantly less reliably true in real-world cases where nodes are also going down sometimes, or nodes are going up and down unexpectedly.
107 lines
2.2 KiB
Go
107 lines
2.2 KiB
Go
// Copyright 2017 Pilosa Corp.
|
|
//
|
|
// 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.
|
|
|
|
package etcd
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/pilosa/pilosa/v2/disco"
|
|
"github.com/pilosa/pilosa/v2/logger"
|
|
"github.com/pilosa/pilosa/v2/testhook"
|
|
"go.etcd.io/etcd/embed"
|
|
"go.etcd.io/etcd/etcdserver/api/v3client"
|
|
)
|
|
|
|
const initVal = "test"
|
|
const newVal = "newValue"
|
|
|
|
func TestLeasedKv(t *testing.T) {
|
|
cfg := embed.NewConfig()
|
|
|
|
dir, err := testhook.TempDir(t, "leasedkv-*")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer os.RemoveAll(dir)
|
|
|
|
cfg.Dir = dir
|
|
etcd, err := embed.StartEtcd(cfg)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cli := v3client.New(etcd.Server)
|
|
defer func() {
|
|
etcd.Close()
|
|
<-etcd.Server.StopNotify()
|
|
cli.Close()
|
|
}()
|
|
wrapper := &Etcd{e: etcd, cli: cli, logger: logger.NewLogfLogger(t)}
|
|
|
|
lkv := newLeasedKV(wrapper, "/test", 1)
|
|
|
|
ctx := context.Background()
|
|
|
|
err = lkv.Start(initVal)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
v, err := lkv.Get(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if v != initVal {
|
|
t.Fatal("obtained value is not the same as expected. Obtained:", v, "Expected:", initVal)
|
|
}
|
|
|
|
err = lkv.Set(ctx, "otherValue")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
err = lkv.Set(ctx, "3")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
err = lkv.Set(ctx, newVal)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
v, err = lkv.Get(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if v != newVal {
|
|
t.Fatal("obtained value is not the same as expected. Obtained:", v, "Expected:", newVal)
|
|
}
|
|
|
|
time.Sleep(5 * time.Second)
|
|
|
|
lkv.Stop()
|
|
|
|
// we need to wait to force the lease expiration
|
|
time.Sleep(5 * time.Second)
|
|
|
|
_, err = lkv.Get(ctx)
|
|
if err == nil || !errors.Is(err, disco.ErrNoResults) {
|
|
t.Fatal("expected error:", disco.ErrNoResults, "obtained:", err)
|
|
}
|
|
}
|