blob: 1e23a0f08f68c6dc622120a7a3eebbe4e060bb74 [file] [log] [blame]
divyadesai81bb7ba2020-03-11 11:45:23 +00001/*
2 * Copyright 2018-present Open Networking Foundation
3
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17package db
18
19import (
20 "context"
21 "errors"
22 "fmt"
divyadesai81bb7ba2020-03-11 11:45:23 +000023 "time"
24
25 "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
26 "github.com/opencord/voltha-lib-go/v3/pkg/log"
27 "google.golang.org/grpc/codes"
28 "google.golang.org/grpc/status"
29)
30
31const (
32 // Default Minimal Interval for posting alive state of backend kvstore on Liveness Channel
33 DefaultLivenessChannelInterval = time.Second * 30
34)
35
36// Backend structure holds details for accessing the kv store
37type Backend struct {
divyadesai81bb7ba2020-03-11 11:45:23 +000038 Client kvstore.Client
39 StoreType string
Neha Sharma87d43d72020-04-08 14:10:40 +000040 Timeout time.Duration
Neha Sharma318a1292020-05-14 19:53:26 +000041 Address string
divyadesai81bb7ba2020-03-11 11:45:23 +000042 PathPrefix string
43 alive bool // Is this backend connection alive?
44 liveness chan bool // channel to post alive state
45 LivenessChannelInterval time.Duration // regularly push alive state beyond this interval
46 lastLivenessTime time.Time // Instant of last alive state push
47}
48
49// NewBackend creates a new instance of a Backend structure
Neha Sharma318a1292020-05-14 19:53:26 +000050func NewBackend(storeType string, address string, timeout time.Duration, pathPrefix string) *Backend {
divyadesai81bb7ba2020-03-11 11:45:23 +000051 var err error
52
53 b := &Backend{
54 StoreType: storeType,
Neha Sharma318a1292020-05-14 19:53:26 +000055 Address: address,
divyadesai81bb7ba2020-03-11 11:45:23 +000056 Timeout: timeout,
57 LivenessChannelInterval: DefaultLivenessChannelInterval,
58 PathPrefix: pathPrefix,
59 alive: false, // connection considered down at start
60 }
61
divyadesai81bb7ba2020-03-11 11:45:23 +000062 if b.Client, err = b.newClient(address, timeout); err != nil {
63 logger.Errorw("failed-to-create-kv-client",
64 log.Fields{
Neha Sharma318a1292020-05-14 19:53:26 +000065 "type": storeType, "address": address,
divyadesai81bb7ba2020-03-11 11:45:23 +000066 "timeout": timeout, "prefix": pathPrefix,
67 "error": err.Error(),
68 })
69 }
70
71 return b
72}
73
Neha Sharma87d43d72020-04-08 14:10:40 +000074func (b *Backend) newClient(address string, timeout time.Duration) (kvstore.Client, error) {
divyadesai81bb7ba2020-03-11 11:45:23 +000075 switch b.StoreType {
76 case "consul":
77 return kvstore.NewConsulClient(address, timeout)
78 case "etcd":
Rohan Agrawal00d3a412020-04-22 10:51:39 +000079 return kvstore.NewEtcdClient(address, timeout, log.WarnLevel)
divyadesai81bb7ba2020-03-11 11:45:23 +000080 }
81 return nil, errors.New("unsupported-kv-store")
82}
83
84func (b *Backend) makePath(key string) string {
85 path := fmt.Sprintf("%s/%s", b.PathPrefix, key)
86 return path
87}
88
89func (b *Backend) updateLiveness(alive bool) {
90 // Periodically push stream of liveness data to the channel,
91 // so that in a live state, the core does not timeout and
92 // send a forced liveness message. Push alive state if the
93 // last push to channel was beyond livenessChannelInterval
94 if b.liveness != nil {
95
96 if b.alive != alive {
97 logger.Debug("update-liveness-channel-reason-change")
98 b.liveness <- alive
99 b.lastLivenessTime = time.Now()
100 } else if time.Since(b.lastLivenessTime) > b.LivenessChannelInterval {
101 logger.Debug("update-liveness-channel-reason-interval")
102 b.liveness <- alive
103 b.lastLivenessTime = time.Now()
104 }
105 }
106
107 // Emit log message only for alive state change
108 if b.alive != alive {
109 logger.Debugw("change-kvstore-alive-status", log.Fields{"alive": alive})
110 b.alive = alive
111 }
112}
113
114// Perform a dummy Key Lookup on kvstore to test Connection Liveness and
115// post on Liveness channel
116func (b *Backend) PerformLivenessCheck(ctx context.Context) bool {
117 alive := b.Client.IsConnectionUp(ctx)
118 logger.Debugw("kvstore-liveness-check-result", log.Fields{"alive": alive})
119
120 b.updateLiveness(alive)
121 return alive
122}
123
124// Enable the liveness monitor channel. This channel will report
125// a "true" or "false" on every kvstore operation which indicates whether
126// or not the connection is still Live. This channel is then picked up
127// by the service (i.e. rw_core / ro_core) to update readiness status
128// and/or take other actions.
129func (b *Backend) EnableLivenessChannel() chan bool {
130 logger.Debug("enable-kvstore-liveness-channel")
131
132 if b.liveness == nil {
133 logger.Debug("create-kvstore-liveness-channel")
134
135 // Channel size of 10 to avoid any possibility of blocking in Load conditions
136 b.liveness = make(chan bool, 10)
137
138 // Post initial alive state
139 b.liveness <- b.alive
140 b.lastLivenessTime = time.Now()
141 }
142
143 return b.liveness
144}
145
146// Extract Alive status of Kvstore based on type of error
147func (b *Backend) isErrorIndicatingAliveKvstore(err error) bool {
148 // Alive unless observed an error indicating so
149 alive := true
150
151 if err != nil {
152
153 // timeout indicates kvstore not reachable/alive
154 if err == context.DeadlineExceeded {
155 alive = false
156 }
157
158 // Need to analyze client-specific errors based on backend type
159 if b.StoreType == "etcd" {
160
161 // For etcd backend, consider not-alive only for errors indicating
162 // timedout request or unavailable/corrupted cluster. For all remaining
163 // error codes listed in https://godoc.org/google.golang.org/grpc/codes#Code,
164 // we would not infer a not-alive backend because such a error may also
165 // occur due to bad client requests or sequence of operations
166 switch status.Code(err) {
167 case codes.DeadlineExceeded:
168 fallthrough
169 case codes.Unavailable:
170 fallthrough
171 case codes.DataLoss:
172 alive = false
173 }
174
175 //} else {
176 // TODO: Implement for consul backend; would it be needed ever?
177 }
178 }
179
180 return alive
181}
182
183// List retrieves one or more items that match the specified key
184func (b *Backend) List(ctx context.Context, key string) (map[string]*kvstore.KVPair, error) {
divyadesai81bb7ba2020-03-11 11:45:23 +0000185 formattedPath := b.makePath(key)
186 logger.Debugw("listing-key", log.Fields{"key": key, "path": formattedPath})
187
188 pair, err := b.Client.List(ctx, formattedPath)
189
190 b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
191
192 return pair, err
193}
194
195// Get retrieves an item that matches the specified key
196func (b *Backend) Get(ctx context.Context, key string) (*kvstore.KVPair, error) {
divyadesai81bb7ba2020-03-11 11:45:23 +0000197 formattedPath := b.makePath(key)
198 logger.Debugw("getting-key", log.Fields{"key": key, "path": formattedPath})
199
200 pair, err := b.Client.Get(ctx, formattedPath)
201
202 b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
203
204 return pair, err
205}
206
207// Put stores an item value under the specifed key
208func (b *Backend) Put(ctx context.Context, key string, value interface{}) error {
divyadesai81bb7ba2020-03-11 11:45:23 +0000209 formattedPath := b.makePath(key)
Rohan Agrawal00d3a412020-04-22 10:51:39 +0000210 logger.Debugw("putting-key", log.Fields{"key": key, "path": formattedPath})
divyadesai81bb7ba2020-03-11 11:45:23 +0000211
212 err := b.Client.Put(ctx, formattedPath, value)
213
214 b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
215
216 return err
217}
218
219// Delete removes an item under the specified key
220func (b *Backend) Delete(ctx context.Context, key string) error {
divyadesai81bb7ba2020-03-11 11:45:23 +0000221 formattedPath := b.makePath(key)
222 logger.Debugw("deleting-key", log.Fields{"key": key, "path": formattedPath})
223
224 err := b.Client.Delete(ctx, formattedPath)
225
226 b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
227
228 return err
229}
230
231// CreateWatch starts watching events for the specified key
232func (b *Backend) CreateWatch(ctx context.Context, key string, withPrefix bool) chan *kvstore.Event {
divyadesai81bb7ba2020-03-11 11:45:23 +0000233 formattedPath := b.makePath(key)
234 logger.Debugw("creating-key-watch", log.Fields{"key": key, "path": formattedPath})
235
236 return b.Client.Watch(ctx, formattedPath, withPrefix)
237}
238
239// DeleteWatch stops watching events for the specified key
240func (b *Backend) DeleteWatch(key string, ch chan *kvstore.Event) {
divyadesai81bb7ba2020-03-11 11:45:23 +0000241 formattedPath := b.makePath(key)
242 logger.Debugw("deleting-key-watch", log.Fields{"key": key, "path": formattedPath})
243
244 b.Client.CloseWatch(formattedPath, ch)
245}