Etcdv3 · watch operation implementation and relevant key instructions

Based on the previous encapsulation , Here we explain watch Related definitions and operations .

In response to this demand , Here's an example ：
A There will be different operations as needed , and B Need to monitor A Different operations change to deal with different things , At this time, if it is etcd Then the problem can be easily solved .

Be more practical ： Such as in business , A privilege function of the user will be stopped after the user owes money , Then the stopped party needs to monitor when the arrearage is , Once in arrears , Then give the privilege to stop the operation .

Catalog

Code implementation

Operation trigger source

watch monitor

Detail analysis

Important summary

---

 

Code implementation

func InitV3Client() (cli *clientv3.Client, err error) {
    cli, err = clientv3.New(clientv3.Config{
        Endpoints:   []string{"192.168.31.103:32379"},
        DialTimeout: 5 * time.Second,
        Username:    "root",
        Password:    "555",
    })

    return
}

Operation trigger source

var (
    watchPrefix = "/a/b"
)

func Source(cli *clientv3.Client) {
    var err error
    for i := 0; ; i++ {
        fmt.Println("i=", i)
        if i%3 == 0 { //  Simulate triggering a condition , Then it starts to trigger watch
            key := fmt.Sprintf("%s/%v", watchPrefix, i)
            if i%2 == 0 { //  If the number is even DELETE operation （ pure delete Comment out the following else）
                fmt.Printf(" Triggered the condition , Get ready DELETE: %v\n", key)
                _, err = cli.Delete(context.Background(), key)
                if err != nil {
                    fmt.Printf("DELETE  Failure ： %v\n", err)
                    continue
                }

                fmt.Printf("DELETE %v success\n", key)
            } else {
                //fmt.Printf(" Triggered the condition , Get ready PUT: %v\n", key)
                //_, err = cli.Put(context.Background(), key, strconv.Itoa(i))
                //if err != nil {
                //    fmt.Printf("PUT  Failure ： %v\n", err)
                //    continue
                //}
                //
                //fmt.Printf("PUT %v success\n", key)
            }
        }

        time.Sleep(time.Second * 1)
    }
}

watch monitor

func WatchSource(cli *clientv3.Client) {
    wCh := cli.Watch(context.Background(), watchPrefix, clientv3.WithPrefix())
    for {
        select {
        case data := <-wCh:
            for i, event := range data.Events { //  There are several events at a time , Here is a few cycles 
                fmt.Printf("receive %v event, %v\n", i, event.Type) // event.Type Represents the event type ,PUT still DELETE
                switch event.Type {
                case clientv3.EventTypePut:
                    // do your sth
                case clientv3.EventTypeDelete:
                    // do your sth
                }

                getKey, getValue := getKeyAndVal(event)
                fmt.Printf(" The signal has been monitored , %v=%v  the  %v  operation .  Next, inform the operator to handle it accordingly \n", getKey, getValue, event.Type)

                //  Actual business processing ...
            }
        }
    }
}

func getKeyAndVal(event *clientv3.Event) (getKey, getValue string) {
    fmt.Printf("event.PrevKv==nil result  %v, event.Kv==nil result  %v\n", event.PrevKv == nil, event.Kv == nil)
    fmt.Printf("event.PrevKv= %+v\n", event.PrevKv)
    fmt.Printf("event.Kv= %+v\n", event.Kv)
    if event.PrevKv != nil {
        getKey, getValue = string(event.PrevKv.Key), string(event.PrevKv.Value)
    }

    if event.Type == mvccpb.PUT {
        if event.Kv != nil {
            getKey, getValue = string(event.Kv.Key), string(event.Kv.Value)
        }
    }

    return
}

start-up

func main() {
    cli, err := client.InitV3Client()
    if err != nil {
        fmt.Println(err)
        return
    }
    
    //  Verification mode 1, Manually operate at the terminal , Here are the monitoring results 
    watch.WatchSource(cli)  
    
    //  Verification mode 2, All run automatically in code 
    //go watch.WatchSource(cli) 
    //watch.Source(cli)
}

good , One watch The process is finished .

Detail analysis

Next, based on the above code , Let's summarize some details , Such as when the signal can be received , When can I get the value of the last version after receiving the signal ？

We do it in a way 1 Do validation .

PUT A new value to see ：

[email protected]:~# docker exec etcd etcdctl put /a/b/3 3 --user=a:123
OK

 result ：

receive 0 event, PUT
event.PrevKv==nil result  true, event.Kv==nil result  false
event.PrevKv= <nil>
event.Kv= key:"/a/b/3" create_revision:762 mod_revision:762 version:1 value:"3" 
 The signal has been monitored , /a/b/3=3  the  PUT  operation .  Next, inform the operator to handle it accordingly 

Again put /a/b/3 3 Try it once ：

receive 0 event, PUT
event.PrevKv==nil result  true, event.Kv==nil result  false
event.PrevKv= <nil>
event.Kv= key:"/a/b/3" create_revision:762 mod_revision:763 version:2 value:"3" 
 The signal has been monitored , /a/b/3=3  the  PUT  operation .  Next, inform the operator to handle it accordingly 

You can see version、mod_revision Worth doing +1 operation ,PrevKv Not getting value .

that key unchanged ,val Change ？

[email protected]:~# docker exec etcd etcdctl put /a/b/3 13 --user=a:123
OK

 Look at the results ：
receive 0 event, PUT
event.PrevKv==nil result  true, event.Kv==nil result  false
event.PrevKv= <nil>
event.Kv= key:"/a/b/3" create_revision:762 mod_revision:764 version:3 value:"13" 
 The signal has been monitored , /a/b/3=13  the  PUT  operation .  Next, inform the operator to handle it accordingly 

so event.PrevKv Still worthless .

Delete an existing data ：

[email protected]:~# docker exec etcd etcdctl del /a/b/3  --user=a:123
1

 Look at the results ：

receive 0 event, DELETE
event.PrevKv==nil result  true, event.Kv==nil result  false
event.PrevKv= <nil>
event.Kv= key:"/a/b/3" mod_revision:767 
 The signal has been monitored , /a/b/3=  the  DELETE  operation .  Next, inform the operator to handle it accordingly 

Delete a nonexistent data ：

[email protected]:~# docker exec etcd etcdctl del /a/b/5  --user=a:123
0

receive 0 event, DELETE
event.PrevKv==nil result  true, event.Kv==nil result  false
event.PrevKv= <nil>
event.Kv= key:"/a/b/3" mod_revision:767 
 The signal has been monitored , /a/b/3=  the  DELETE  operation .  Next, inform the operator to handle it accordingly 

PrevKv It's still empty .

Don't ,PrevKv Is it always worthless ？ What's the point of having this ？ Don't worry , Let's make a small change ：

wCh := cli.Watch(context.Background(), watchPrefix, clientv3.WithPrefix(),clientv3.WithPrevKV())

Repeat the above process for verification ：

PUT A new value ：

[email protected]:~# docker exec etcd etcdctl put /a/b/3 2 --user=a:123
OK

receive 0 event, PUT
event.PrevKv==nil result  true, event.Kv==nil result  false
event.PrevKv= <nil>
event.Kv= key:"/a/b/3" create_revision:806 mod_revision:806 version:1 value:"2" 
 The signal has been monitored , /a/b/3=2  the  PUT  operation .  Next, inform the operator to handle it accordingly 

You can see PrevKv It's empty .

Once again ：

receive 0 event, PUT
event.PrevKv==nil result  false, event.Kv==nil result  false
event.PrevKv= key:"/a/b/3" create_revision:799 mod_revision:800 version:2 value:"2" 
event.Kv= key:"/a/b/3" create_revision:799 mod_revision:801 version:3 value:"2" 
 The signal has been monitored , /a/b/3=2  the  PUT  operation .  Next, inform the operator to handle it accordingly 

Change it value :

[email protected]:~# docker exec etcd etcdctl put /a/b/3 3 --user=a:123
OK

receive 0 event, PUT
event.PrevKv==nil result  false, event.Kv==nil result  false
event.PrevKv= key:"/a/b/3" create_revision:799 mod_revision:801 version:3 value:"2" 
event.Kv= key:"/a/b/3" create_revision:799 mod_revision:802 version:4 value:"3" 
 The signal has been monitored , /a/b/3=3  the  PUT  operation .  Next, inform the operator to handle it accordingly 

Important summary

1, about PUT operation
put Over time key After again put, Equivalent to update operation , here PrevKv Valuable 、 Can receive any monitoring signal ;
Yes key For the first time PUT when PrevKv It's empty 、 Can receive any monitoring signal ;

2, about DELETE operation
The key When it exists delete operation , here PrevKvs Valuable , Can receive any monitoring signal ;
if delete There is no the key, Then no monitoring signal is received ;

In fact, there are many operations that can be carried out , It depends on what you need , Another example is that we add clientv3.WithPrefix(), This means that the operation will be treated as including prefix , That is to say, the listener listens to /a/b Prefixed key The change of , without clientv3.WithPrefix(), Delete or PUT One /a/b/5 Such a value is doomed to failure 、 Will not receive any monitoring signal .