File: session.go

package info (click to toggle)
consul 1.5.2+dfsg1-12
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 41,656 kB
  • sloc: sh: 2,887; makefile: 127; ruby: 75; cpp: 58; xml: 36
file content (224 lines) | stat: -rw-r--r-- 5,516 bytes parent folder | download | duplicates (7)
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
package api

import (
	"errors"
	"fmt"
	"time"
)

const (
	// SessionBehaviorRelease is the default behavior and causes
	// all associated locks to be released on session invalidation.
	SessionBehaviorRelease = "release"

	// SessionBehaviorDelete is new in Consul 0.5 and changes the
	// behavior to delete all associated locks on session invalidation.
	// It can be used in a way similar to Ephemeral Nodes in ZooKeeper.
	SessionBehaviorDelete = "delete"
)

var ErrSessionExpired = errors.New("session expired")

// SessionEntry represents a session in consul
type SessionEntry struct {
	CreateIndex uint64
	ID          string
	Name        string
	Node        string
	Checks      []string
	LockDelay   time.Duration
	Behavior    string
	TTL         string
}

// Session can be used to query the Session endpoints
type Session struct {
	c *Client
}

// Session returns a handle to the session endpoints
func (c *Client) Session() *Session {
	return &Session{c}
}

// CreateNoChecks is like Create but is used specifically to create
// a session with no associated health checks.
func (s *Session) CreateNoChecks(se *SessionEntry, q *WriteOptions) (string, *WriteMeta, error) {
	body := make(map[string]interface{})
	body["Checks"] = []string{}
	if se != nil {
		if se.Name != "" {
			body["Name"] = se.Name
		}
		if se.Node != "" {
			body["Node"] = se.Node
		}
		if se.LockDelay != 0 {
			body["LockDelay"] = durToMsec(se.LockDelay)
		}
		if se.Behavior != "" {
			body["Behavior"] = se.Behavior
		}
		if se.TTL != "" {
			body["TTL"] = se.TTL
		}
	}
	return s.create(body, q)

}

// Create makes a new session. Providing a session entry can
// customize the session. It can also be nil to use defaults.
func (s *Session) Create(se *SessionEntry, q *WriteOptions) (string, *WriteMeta, error) {
	var obj interface{}
	if se != nil {
		body := make(map[string]interface{})
		obj = body
		if se.Name != "" {
			body["Name"] = se.Name
		}
		if se.Node != "" {
			body["Node"] = se.Node
		}
		if se.LockDelay != 0 {
			body["LockDelay"] = durToMsec(se.LockDelay)
		}
		if len(se.Checks) > 0 {
			body["Checks"] = se.Checks
		}
		if se.Behavior != "" {
			body["Behavior"] = se.Behavior
		}
		if se.TTL != "" {
			body["TTL"] = se.TTL
		}
	}
	return s.create(obj, q)
}

func (s *Session) create(obj interface{}, q *WriteOptions) (string, *WriteMeta, error) {
	var out struct{ ID string }
	wm, err := s.c.write("/v1/session/create", obj, &out, q)
	if err != nil {
		return "", nil, err
	}
	return out.ID, wm, nil
}

// Destroy invalidates a given session
func (s *Session) Destroy(id string, q *WriteOptions) (*WriteMeta, error) {
	wm, err := s.c.write("/v1/session/destroy/"+id, nil, nil, q)
	if err != nil {
		return nil, err
	}
	return wm, nil
}

// Renew renews the TTL on a given session
func (s *Session) Renew(id string, q *WriteOptions) (*SessionEntry, *WriteMeta, error) {
	r := s.c.newRequest("PUT", "/v1/session/renew/"+id)
	r.setWriteOptions(q)
	rtt, resp, err := s.c.doRequest(r)
	if err != nil {
		return nil, nil, err
	}
	defer resp.Body.Close()

	wm := &WriteMeta{RequestTime: rtt}

	if resp.StatusCode == 404 {
		return nil, wm, nil
	} else if resp.StatusCode != 200 {
		return nil, nil, fmt.Errorf("Unexpected response code: %d", resp.StatusCode)
	}

	var entries []*SessionEntry
	if err := decodeBody(resp, &entries); err != nil {
		return nil, nil, fmt.Errorf("Failed to read response: %v", err)
	}
	if len(entries) > 0 {
		return entries[0], wm, nil
	}
	return nil, wm, nil
}

// RenewPeriodic is used to periodically invoke Session.Renew on a
// session until a doneCh is closed. This is meant to be used in a long running
// goroutine to ensure a session stays valid.
func (s *Session) RenewPeriodic(initialTTL string, id string, q *WriteOptions, doneCh <-chan struct{}) error {
	ctx := q.Context()

	ttl, err := time.ParseDuration(initialTTL)
	if err != nil {
		return err
	}

	waitDur := ttl / 2
	lastRenewTime := time.Now()
	var lastErr error
	for {
		if time.Since(lastRenewTime) > ttl {
			return lastErr
		}
		select {
		case <-time.After(waitDur):
			entry, _, err := s.Renew(id, q)
			if err != nil {
				waitDur = time.Second
				lastErr = err
				continue
			}
			if entry == nil {
				return ErrSessionExpired
			}

			// Handle the server updating the TTL
			ttl, _ = time.ParseDuration(entry.TTL)
			waitDur = ttl / 2
			lastRenewTime = time.Now()

		case <-doneCh:
			// Attempt a session destroy
			s.Destroy(id, q)
			return nil

		case <-ctx.Done():
			// Bail immediately since attempting the destroy would
			// use the canceled context in q, which would just bail.
			return ctx.Err()
		}
	}
}

// Info looks up a single session
func (s *Session) Info(id string, q *QueryOptions) (*SessionEntry, *QueryMeta, error) {
	var entries []*SessionEntry
	qm, err := s.c.query("/v1/session/info/"+id, &entries, q)
	if err != nil {
		return nil, nil, err
	}
	if len(entries) > 0 {
		return entries[0], qm, nil
	}
	return nil, qm, nil
}

// List gets sessions for a node
func (s *Session) Node(node string, q *QueryOptions) ([]*SessionEntry, *QueryMeta, error) {
	var entries []*SessionEntry
	qm, err := s.c.query("/v1/session/node/"+node, &entries, q)
	if err != nil {
		return nil, nil, err
	}
	return entries, qm, nil
}

// List gets all active sessions
func (s *Session) List(q *QueryOptions) ([]*SessionEntry, *QueryMeta, error) {
	var entries []*SessionEntry
	qm, err := s.c.query("/v1/session/list", &entries, q)
	if err != nil {
		return nil, nil, err
	}
	return entries, qm, nil
}