forked from hashicorp/consul
/
session_endpoint.go
127 lines (113 loc) · 3.4 KB
/
session_endpoint.go
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
package consul
import (
"fmt"
"time"
"github.com/armon/go-metrics"
"github.com/hashicorp/consul/consul/structs"
)
// Session endpoint is used to manipulate sessions for KV
type Session struct {
srv *Server
}
// Apply is used to apply a modifying request to the data store. This should
// only be used for operations that modify the data
func (s *Session) Apply(args *structs.SessionRequest, reply *string) error {
if done, err := s.srv.forward("Session.Apply", args, args, reply); done {
return err
}
defer metrics.MeasureSince([]string{"consul", "session", "apply"}, time.Now())
// Verify the args
if args.Session.ID == "" && args.Op == structs.SessionDestroy {
return fmt.Errorf("Must provide ID")
}
if args.Session.Node == "" && args.Op == structs.SessionCreate {
return fmt.Errorf("Must provide Node")
}
// If this is a create, we must generate the Session ID. This must
// be done prior to appending to the raft log, because the ID is not
// deterministic. Once the entry is in the log, the state update MUST
// be deterministic or the followers will not converge.
if args.Op == structs.SessionCreate {
// Generate a new session ID, verify uniqueness
state := s.srv.fsm.State()
for {
args.Session.ID = generateUUID()
_, sess, err := state.SessionGet(args.Session.ID)
if err != nil {
s.srv.logger.Printf("[ERR] consul.session: Session lookup failed: %v", err)
return err
}
if sess == nil {
break
}
}
}
// Apply the update
resp, err := s.srv.raftApply(structs.SessionRequestType, args)
if err != nil {
s.srv.logger.Printf("[ERR] consul.session: Apply failed: %v", err)
return err
}
if respErr, ok := resp.(error); ok {
return respErr
}
// Check if the return type is a string
if respString, ok := resp.(string); ok {
*reply = respString
}
return nil
}
// Get is used to retrieve a single session
func (s *Session) Get(args *structs.SessionSpecificRequest,
reply *structs.IndexedSessions) error {
if done, err := s.srv.forward("Session.Get", args, args, reply); done {
return err
}
// Get the local state
state := s.srv.fsm.State()
return s.srv.blockingRPC(&args.QueryOptions,
&reply.QueryMeta,
state.QueryTables("SessionGet"),
func() error {
index, session, err := state.SessionGet(args.Session)
reply.Index = index
if session != nil {
reply.Sessions = structs.Sessions{session}
}
return err
})
}
// List is used to list all the active sessions
func (s *Session) List(args *structs.DCSpecificRequest,
reply *structs.IndexedSessions) error {
if done, err := s.srv.forward("Session.List", args, args, reply); done {
return err
}
// Get the local state
state := s.srv.fsm.State()
return s.srv.blockingRPC(&args.QueryOptions,
&reply.QueryMeta,
state.QueryTables("SessionList"),
func() error {
var err error
reply.Index, reply.Sessions, err = state.SessionList()
return err
})
}
// NodeSessions is used to get all the sessions for a particular node
func (s *Session) NodeSessions(args *structs.NodeSpecificRequest,
reply *structs.IndexedSessions) error {
if done, err := s.srv.forward("Session.NodeSessions", args, args, reply); done {
return err
}
// Get the local state
state := s.srv.fsm.State()
return s.srv.blockingRPC(&args.QueryOptions,
&reply.QueryMeta,
state.QueryTables("NodeSessions"),
func() error {
var err error
reply.Index, reply.Sessions, err = state.NodeSessions(args.Node)
return err
})
}