etcd: Simple Key-Value with Raft

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# etcd uses flat keys, but conventions create logical hierarchy
# Keys are byte strings; these examples use human-readable paths
 
# Service registry pattern
/services/user-service/instances/10.0.0.1:8080
/services/user-service/instances/10.0.0.2:8080
/services/user-service/instances/10.0.0.3:8080
/services/payment-service/instances/10.0.1.1:8080
/services/payment-service/instances/10.0.1.2:8080
 
# Configuration pattern
/config/database/connection-string
/config/database/pool-size
/config/features/new-checkout-enabled
/config/features/dark-mode-enabled
 
# Leader election pattern
/election/cluster-leader/candidate-001
/election/cluster-leader/candidate-002
 
# Distributed lock pattern  
/locks/payment-processing/lock
 
# Prefix query: get all user-service instances
etcdctl get /services/user-service/instances/ --prefix
 
# Range query: get all config between keys
etcdctl get /config/database/a /config/database/z

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# Put a key and see the revision
$ etcdctl put /mykey "value1"
OK
 
$ etcdctl get /mykey -w json | jq
{
  "header": { "revision": 42 },
  "kvs": [{
    "key": "L215a2V5",           # base64 encoded
    "create_revision": 42,       # Revision when created
    "mod_revision": 42,          # Revision when last modified  
    "version": 1,                # Number of modifications
    "value": "dmFsdWUx"          # base64 encoded
  }]
}
 
# Update the key
$ etcdctl put /mykey "value2"
 
$ etcdctl get /mykey -w json | jq '.kvs[0] | {mod_revision, version}'
{
  "mod_revision": 43,
  "version": 2
}
 
# Get historical value at revision 42
$ etcdctl get /mykey --rev=42
/mykey
value1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# Serializable read (may be stale, but fast)
$ etcdctl get /mykey --consistency=s
 
# Linearizable read (always up-to-date, may be slower)
$ etcdctl get /mykey --consistency=l
 
# Read at specific revision
$ etcdctl get /mykey --rev=42
 
# In Go client
resp, err := client.Get(ctx, "/mykey", 
    clientv3.WithSerializable())  // Don't require leader confirmation
 
resp, err := client.Get(ctx, "/mykey")  
    // Default is linearizable when using leader-only endpoints

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# Create a lease with 60-second TTL
$ etcdctl lease grant 60
lease 694d7c8c9b6c6b0a granted with TTL(60s)
 
# Attach keys to the lease
$ etcdctl put /services/myapp/instance-1 "10.0.0.1:8080" --lease=694d7c8c9b6c6b0a
$ etcdctl put /services/myapp/instance-1/health "OK" --lease=694d7c8c9b6c6b0a
 
# Both keys share the same lease
# If we stop keeping the lease alive, both disappear after 60s
 
# Keep the lease alive (run this continuously in your app)
$ etcdctl lease keep-alive 694d7c8c9b6c6b0a
lease 694d7c8c9b6c6b0a keepalived with TTL(60)
lease 694d7c8c9b6c6b0a keepalived with TTL(60)
...
 
# Revoke the lease manually (both keys deleted immediately)
$ etcdctl lease revoke 694d7c8c9b6c6b0a
lease 694d7c8c9b6c6b0a revoked
 
# Check remaining TTL
$ etcdctl lease timetolive 694d7c8c9b6c6b0a
lease 694d7c8c9b6c6b0a granted with TTL(60s), remaining(45s)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

package main
 
import (
    "context"
    "log"
    clientv3 "go.etcd.io/etcd/client/v3"
)
 
func registerService(client *clientv3.Client) error {
    ctx := context.Background()
    
    // Step 1: Grant a lease with 30-second TTL
    lease, err := client.Grant(ctx, 30)
    if err != nil {
        return err
    }
    
    // Step 2: Create registration key with lease
    _, err = client.Put(ctx, 
        "/services/myapp/instances/"+hostname,
        "10.0.0.1:8080",
        clientv3.WithLease(lease.ID),
    )
    if err != nil {
        return err
    }
    
    // Step 3: Keep the lease alive
    // KeepAlive returns a channel that sends keep-alive responses
    keepAliveChan, err := client.KeepAlive(ctx, lease.ID)
    if err != nil {
        return err
    }
    
    // Process keep-alive responses in background
    go func() {
        for {
            select {
            case resp := <-keepAliveChan:
                if resp == nil {
                    // Lease expired or revoked
                    log.Println("Lease expired, re-registering...")
                    registerService(client)  // Re-register
                    return
                }
                // Successfully kept alive
                log.Printf("Lease %x kept alive, TTL: %d
", 
                    resp.ID, resp.TTL)
            }
        }
    }()
    
    return nil
}
 
func discoverServices(client *clientv3.Client) ([]string, error) {
    ctx := context.Background()
    
    // Get all instances under the prefix
    resp, err := client.Get(ctx, 
        "/services/myapp/instances/",
        clientv3.WithPrefix(),
    )
    if err != nil {
        return nil, err
    }
    
    instances := make([]string, len(resp.Kvs))
    for i, kv := range resp.Kvs {
        instances[i] = string(kv.Value)
    }
    return instances, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

package main
 
import (
    "context"
    "fmt"
    clientv3 "go.etcd.io/etcd/client/v3"
)
 
func watchExamples(client *clientv3.Client) {
    ctx := context.Background()
    
    // Watch a single key
    watchChan := client.Watch(ctx, "/config/feature-flag")
    
    // Watch all keys with prefix
    watchChan = client.Watch(ctx, "/services/myapp/", 
        clientv3.WithPrefix())
    
    // Watch from a specific revision (useful after restart)
    watchChan = client.Watch(ctx, "/services/myapp/",
        clientv3.WithPrefix(),
        clientv3.WithRev(lastSeenRevision + 1))
    
    // Watch with filters (only deletions)
    watchChan = client.Watch(ctx, "/services/myapp/",
        clientv3.WithPrefix(),
        clientv3.WithFilterPut())  // Filter OUT puts, see only deletes
    
    // Process watch events
    for watchResp := range watchChan {
        // Check for errors
        if watchResp.Err() != nil {
            handleWatchError(watchResp.Err())
            continue
        }
        
        // Check if this is a progress notification
        if watchResp.IsProgressNotify() {
            // No actual events, just confirming liveness
            continue
        }
        
        // Process events
        for _, event := range watchResp.Events {
            switch event.Type {
            case clientv3.EventTypePut:
                fmt.Printf("PUT: %s -> %s (mod_rev: %d)
",
                    event.Kv.Key, event.Kv.Value, 
                    event.Kv.ModRevision)
                    
            case clientv3.EventTypeDelete:
                fmt.Printf("DELETE: %s (mod_rev: %d)
",
                    event.Kv.Key, event.Kv.ModRevision)
            }
        }
        
        // Save revision for restart recovery
        lastSeenRevision = watchResp.Header.Revision
    }
}
 
func handleWatchError(err error) {
    if err == context.Canceled {
        // Normal cancellation
        return
    }
    
    // Check if it's a compaction error
    if err.Error() == rpctypes.ErrCompacted.Error() {
        // History was compacted, need to resync from current state
        resyncFromCurrentState()
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

package main
 
import (
    "context"
    clientv3 "go.etcd.io/etcd/client/v3"
)
 
func transactionExamples(client *clientv3.Client) {
    ctx := context.Background()
    
    // Compare-And-Swap (CAS): Only set if current value matches
    _, err := client.Txn(ctx).
        If(clientv3.Compare(clientv3.Value("/config/version"), "=", "v1")).
        Then(clientv3.OpPut("/config/version", "v2")).
        Else(clientv3.OpGet("/config/version")).
        Commit()
    
    // Create-If-Not-Exists: Acquire lock only if key doesn't exist
    resp, err := client.Txn(ctx).
        If(clientv3.Compare(clientv3.CreateRevision("/locks/mylock"), "=", 0)).
        Then(clientv3.OpPut("/locks/mylock", myIdentity, 
             clientv3.WithLease(lease.ID))).
        Else(clientv3.OpGet("/locks/mylock")).
        Commit()
    
    if resp.Succeeded {
        fmt.Println("Lock acquired!")
    } else {
        holder := string(resp.Responses[0].GetResponseRange().Kvs[0].Value)
        fmt.Printf("Lock held by: %s
", holder)
    }
    
    // Multi-key atomic update
    _, err = client.Txn(ctx).
        If(
            clientv3.Compare(clientv3.ModRevision("/users/alice"), ">", 0),
            clientv3.Compare(clientv3.ModRevision("/users/bob"), ">", 0),
        ).
        Then(
            clientv3.OpPut("/users/alice", updatedAlice),
            clientv3.OpPut("/users/bob", updatedBob),
            clientv3.OpPut("/audit/log", "updated alice and bob"),
        ).
        Commit()
}
 
// Software Transactional Memory (STM) pattern for complex logic
func stmExample(client *clientv3.Client) {
    // Using the concurrency package for higher-level STM
    stm.NewSTM(client, func(stm concurrency.STM) error {
        // Read multiple keys
        balance1 := stm.Get("/accounts/alice/balance")
        balance2 := stm.Get("/accounts/bob/balance")
        
        // Perform logic
        amount := 100
        newBalance1 := parseBalance(balance1) - amount
        newBalance2 := parseBalance(balance2) + amount
        
        // Write multiple keys - all or nothing
        stm.Put("/accounts/alice/balance", formatBalance(newBalance1))
        stm.Put("/accounts/bob/balance", formatBalance(newBalance2))
        
        return nil  // Commit
    })
    // STM automatically retries on conflicts
}