blob: c644e148841a1b7426f61ce769cf7ba5c5b51162 [file] [log] [blame]
Scott Baker2c1c4822019-10-16 11:02:41 -07001/*
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 model
18
19import (
20 "bytes"
21 "compress/gzip"
22 "context"
23 "github.com/golang/protobuf/proto"
24 "github.com/google/uuid"
Scott Baker2c1c4822019-10-16 11:02:41 -070025 "github.com/opencord/voltha-lib-go/pkg/db/kvstore"
Scott Bakere73f91e2019-10-17 12:58:11 -070026 "github.com/opencord/voltha-lib-go/pkg/log"
Scott Baker2c1c4822019-10-16 11:02:41 -070027 "reflect"
28 "strings"
29 "sync"
30)
31
32// PersistedRevision holds information of revision meant to be saved in a persistent storage
33type PersistedRevision struct {
34 Revision
35 Compress bool
36
37 events chan *kvstore.Event
38 kvStore *Backend
39 mutex sync.RWMutex
40 versionMutex sync.RWMutex
41 Version int64
42 isStored bool
43 isWatched bool
44}
45
46type watchCache struct {
47 Cache sync.Map
48}
49
50var watchCacheInstance *watchCache
51var watchCacheOne sync.Once
52
53func Watches() *watchCache {
54 watchCacheOne.Do(func() {
55 watchCacheInstance = &watchCache{Cache: sync.Map{}}
56 })
57 return watchCacheInstance
58}
59
60// NewPersistedRevision creates a new instance of a PersistentRevision structure
61func NewPersistedRevision(branch *Branch, data interface{}, children map[string][]Revision) Revision {
62 pr := &PersistedRevision{}
63 pr.kvStore = branch.Node.GetRoot().KvStore
64 pr.Version = 1
65 pr.Revision = NewNonPersistedRevision(nil, branch, data, children)
66 return pr
67}
68
69func (pr *PersistedRevision) getVersion() int64 {
70 pr.versionMutex.RLock()
71 defer pr.versionMutex.RUnlock()
72 return pr.Version
73}
74
75func (pr *PersistedRevision) setVersion(version int64) {
76 pr.versionMutex.Lock()
77 defer pr.versionMutex.Unlock()
78 pr.Version = version
79}
80
81// Finalize is responsible of saving the revision in the persistent storage
82func (pr *PersistedRevision) Finalize(skipOnExist bool) {
83 pr.store(skipOnExist)
84}
85
86func (pr *PersistedRevision) store(skipOnExist bool) {
87 if pr.GetBranch().Txid != "" {
88 return
89 }
90
91 log.Debugw("ready-to-store-revision", log.Fields{"hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetData()})
92
93 // clone the revision data to avoid any race conditions with processes
94 // accessing the same data
95 cloned := proto.Clone(pr.GetConfig().Data.(proto.Message))
96
97 if blob, err := proto.Marshal(cloned); err != nil {
98 log.Errorw("problem-to-marshal", log.Fields{"error": err, "hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetData()})
99 } else {
100 if pr.Compress {
101 var b bytes.Buffer
102 w := gzip.NewWriter(&b)
103 w.Write(blob)
104 w.Close()
105 blob = b.Bytes()
106 }
107
108 GetRevCache().Set(pr.GetName(), pr)
109 if err := pr.kvStore.Put(pr.GetName(), blob); err != nil {
110 log.Warnw("problem-storing-revision", log.Fields{"error": err, "hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetConfig().Data})
111 } else {
112 log.Debugw("storing-revision", log.Fields{"hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetConfig().Data, "version": pr.getVersion()})
113 pr.isStored = true
114 }
115 }
116}
117
118func (pr *PersistedRevision) SetupWatch(key string) {
119 if key == "" {
120 log.Debugw("ignoring-watch", log.Fields{"key": key, "revision-hash": pr.GetHash()})
121 return
122 }
123
124 if _, exists := Watches().Cache.LoadOrStore(key+"-"+pr.GetHash(), struct{}{}); exists {
125 return
126 }
127
128 if pr.events == nil {
129 pr.events = make(chan *kvstore.Event)
130
131 log.Debugw("setting-watch-channel", log.Fields{"key": key, "revision-hash": pr.GetHash()})
132
133 pr.SetName(key)
134 pr.events = pr.kvStore.CreateWatch(key)
135 }
136
137 if !pr.isWatched {
138 pr.isWatched = true
139
140 log.Debugw("setting-watch-routine", log.Fields{"key": key, "revision-hash": pr.GetHash()})
141
142 // Start watching
143 go pr.startWatching()
144 }
145}
146
147func (pr *PersistedRevision) startWatching() {
148 log.Debugw("starting-watch", log.Fields{"key": pr.GetHash(), "watch": pr.GetName()})
149
150StopWatchLoop:
151 for {
152 latestRev := pr.GetBranch().GetLatest()
153
154 select {
155 case event, ok := <-pr.events:
156 if !ok {
157 log.Errorw("event-channel-failure: stopping watch loop", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
158 break StopWatchLoop
159 }
160 log.Debugw("received-event", log.Fields{"type": event.EventType, "watch": latestRev.GetName()})
161
162 switch event.EventType {
163 case kvstore.DELETE:
164 log.Debugw("delete-from-memory", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
165 pr.Revision.Drop("", true)
166 break StopWatchLoop
167
168 case kvstore.PUT:
169 log.Debugw("update-in-memory", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
170 if latestRev.getVersion() >= event.Version {
171 log.Debugw("skipping-matching-or-older-revision", log.Fields{
172 "watch": latestRev.GetName(),
173 "watch-version": event.Version,
174 "latest-version": latestRev.getVersion(),
175 })
176 continue
177 } else {
178 log.Debugw("watch-revision-is-newer", log.Fields{
179 "watch": latestRev.GetName(),
180 "watch-version": event.Version,
181 "latest-version": latestRev.getVersion(),
182 })
183 }
184
185 data := reflect.New(reflect.TypeOf(latestRev.GetData()).Elem())
186
187 if err := proto.Unmarshal(event.Value.([]byte), data.Interface().(proto.Message)); err != nil {
188 log.Errorw("failed-to-unmarshal-watch-data", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "error": err})
189 } else {
190 log.Debugw("un-marshaled-watch-data", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "data": data.Interface()})
191
192 var pathLock string
193 var blobs map[string]*kvstore.KVPair
194
195 // The watch reported new persistence data.
196 // Construct an object that will be used to update the memory
197 blobs = make(map[string]*kvstore.KVPair)
198 key, _ := kvstore.ToString(event.Key)
199 blobs[key] = &kvstore.KVPair{
200 Key: key,
201 Value: event.Value,
202 Session: "",
203 Lease: 0,
204 Version: event.Version,
205 }
206
207 if latestRev.GetNode().GetProxy() != nil {
208 //
209 // If a proxy exists for this revision, use it to lock access to the path
210 // and prevent simultaneous updates to the object in memory
211 //
212
213 //If the proxy already has a request in progress, then there is no need to process the watch
214 if latestRev.GetNode().GetProxy().GetOperation() != PROXY_NONE {
215 log.Debugw("operation-in-progress", log.Fields{
216 "key": latestRev.GetHash(),
217 "path": latestRev.GetNode().GetProxy().getFullPath(),
218 "operation": latestRev.GetNode().GetProxy().operation.String(),
219 })
220 continue
221 }
222
223 pathLock, _ = latestRev.GetNode().GetProxy().parseForControlledPath(latestRev.GetNode().GetProxy().getFullPath())
224
225 // Reserve the path to prevent others to modify while we reload from persistence
226 latestRev.GetNode().GetProxy().GetRoot().KvStore.Client.Reserve(pathLock+"_", uuid.New().String(), ReservationTTL)
227 latestRev.GetNode().GetProxy().SetOperation(PROXY_WATCH)
228
229 // Load changes and apply to memory
230 latestRev.LoadFromPersistence(context.Background(), latestRev.GetName(), "", blobs)
231
232 // Release path
233 latestRev.GetNode().GetProxy().GetRoot().KvStore.Client.ReleaseReservation(pathLock + "_")
234
235 } else {
236 // This block should be reached only if coming from a non-proxied request
237 log.Debugw("revision-with-no-proxy", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
238
239 // Load changes and apply to memory
240 latestRev.LoadFromPersistence(context.Background(), latestRev.GetName(), "", blobs)
241 }
242 }
243
244 default:
245 log.Debugw("unhandled-event", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "type": event.EventType})
246 }
247 }
248 }
249
250 Watches().Cache.Delete(pr.GetName() + "-" + pr.GetHash())
251
252 log.Debugw("exiting-watch", log.Fields{"key": pr.GetHash(), "watch": pr.GetName()})
253}
254
255// UpdateData modifies the information in the data model and saves it in the persistent storage
256func (pr *PersistedRevision) UpdateData(ctx context.Context, data interface{}, branch *Branch) Revision {
257 log.Debugw("updating-persisted-data", log.Fields{"hash": pr.GetHash()})
258
259 newNPR := pr.Revision.UpdateData(ctx, data, branch)
260
261 newPR := &PersistedRevision{
262 Revision: newNPR,
263 Compress: pr.Compress,
264 kvStore: pr.kvStore,
265 events: pr.events,
266 Version: pr.getVersion(),
267 isWatched: pr.isWatched,
268 }
269
270 if newPR.GetHash() != pr.GetHash() {
271 newPR.isStored = false
272 pr.Drop(branch.Txid, false)
273 pr.Drop(branch.Txid, false)
274 } else {
275 newPR.isStored = true
276 }
277
278 return newPR
279}
280
281// UpdateChildren modifies the children of a revision and of a specific component and saves it in the persistent storage
282func (pr *PersistedRevision) UpdateChildren(ctx context.Context, name string, children []Revision, branch *Branch) Revision {
283 log.Debugw("updating-persisted-children", log.Fields{"hash": pr.GetHash()})
284
285 newNPR := pr.Revision.UpdateChildren(ctx, name, children, branch)
286
287 newPR := &PersistedRevision{
288 Revision: newNPR,
289 Compress: pr.Compress,
290 kvStore: pr.kvStore,
291 events: pr.events,
292 Version: pr.getVersion(),
293 isWatched: pr.isWatched,
294 }
295
296 if newPR.GetHash() != pr.GetHash() {
297 newPR.isStored = false
298 pr.Drop(branch.Txid, false)
299 } else {
300 newPR.isStored = true
301 }
302
303 return newPR
304}
305
306// UpdateAllChildren modifies the children for all components of a revision and saves it in the peristent storage
307func (pr *PersistedRevision) UpdateAllChildren(children map[string][]Revision, branch *Branch) Revision {
308 log.Debugw("updating-all-persisted-children", log.Fields{"hash": pr.GetHash()})
309
310 newNPR := pr.Revision.UpdateAllChildren(children, branch)
311
312 newPR := &PersistedRevision{
313 Revision: newNPR,
314 Compress: pr.Compress,
315 kvStore: pr.kvStore,
316 events: pr.events,
317 Version: pr.getVersion(),
318 isWatched: pr.isWatched,
319 }
320
321 if newPR.GetHash() != pr.GetHash() {
322 newPR.isStored = false
323 pr.Drop(branch.Txid, false)
324 } else {
325 newPR.isStored = true
326 }
327
328 return newPR
329}
330
331// Drop takes care of eliminating a revision hash that is no longer needed
332// and its associated config when required
333func (pr *PersistedRevision) Drop(txid string, includeConfig bool) {
334 pr.Revision.Drop(txid, includeConfig)
335}
336
337// Drop takes care of eliminating a revision hash that is no longer needed
338// and its associated config when required
339func (pr *PersistedRevision) StorageDrop(txid string, includeConfig bool) {
340 log.Debugw("dropping-revision", log.Fields{"txid": txid, "hash": pr.GetHash(), "config-hash": pr.GetConfig().Hash})
341
342 pr.mutex.Lock()
343 defer pr.mutex.Unlock()
344 if pr.kvStore != nil && txid == "" {
345 if pr.isStored {
346 if pr.isWatched {
347 pr.kvStore.DeleteWatch(pr.GetName(), pr.events)
348 pr.isWatched = false
349 }
350
351 if err := pr.kvStore.Delete(pr.GetName()); err != nil {
352 log.Errorw("failed-to-remove-revision", log.Fields{"hash": pr.GetHash(), "error": err.Error()})
353 } else {
354 pr.isStored = false
355 }
356 }
357
358 } else {
359 if includeConfig {
360 log.Debugw("attempted-to-remove-transacted-revision-config", log.Fields{"hash": pr.GetConfig().Hash, "txid": txid})
361 }
362 log.Debugw("attempted-to-remove-transacted-revision", log.Fields{"hash": pr.GetHash(), "txid": txid})
363 }
364
365 pr.Revision.Drop(txid, includeConfig)
366}
367
368// verifyPersistedEntry validates if the provided data is available or not in memory and applies updates as required
369func (pr *PersistedRevision) verifyPersistedEntry(ctx context.Context, data interface{}, typeName string, keyName string,
370 keyValue string, txid string, version int64) (response Revision) {
371 // Parent which holds the current node entry
372 parent := pr.GetBranch().Node.GetRoot()
373
374 // Get a copy of the parent's children
375 children := make([]Revision, len(parent.GetBranch(NONE).Latest.GetChildren(typeName)))
376 copy(children, parent.GetBranch(NONE).Latest.GetChildren(typeName))
377
378 // Verify if a child with the provided key value can be found
379 if childIdx, childRev := pr.GetNode().findRevByKey(children, keyName, keyValue); childRev != nil {
380 // A child matching the provided key exists in memory
381 // Verify if the data differs from what was retrieved from persistence
382 // Also check if we are treating a newer revision of the data or not
383 if childRev.GetData().(proto.Message).String() != data.(proto.Message).String() && childRev.getVersion() < version {
384 log.Debugw("revision-data-is-different", log.Fields{
385 "key": childRev.GetHash(),
386 "name": childRev.GetName(),
387 "data": childRev.GetData(),
388 "version": childRev.getVersion(),
389 })
390
391 //
392 // Data has changed; replace the child entry and update the parent revision
393 //
394
395 // BEGIN Lock child -- prevent any incoming changes
396 childRev.GetBranch().LatestLock.Lock()
397
398 // Update child
399 updatedChildRev := childRev.UpdateData(ctx, data, childRev.GetBranch())
400
401 updatedChildRev.GetNode().SetProxy(childRev.GetNode().GetProxy())
402 updatedChildRev.SetupWatch(updatedChildRev.GetName())
403 updatedChildRev.SetLastUpdate()
404 updatedChildRev.(*PersistedRevision).setVersion(version)
405
406 // Update cache
407 GetRevCache().Set(updatedChildRev.GetName(), updatedChildRev)
408 childRev.Drop(txid, false)
409
410 childRev.GetBranch().LatestLock.Unlock()
411 // END lock child
412
413 // Update child entry
414 children[childIdx] = updatedChildRev
415
416 // BEGIN lock parent -- Update parent
417 parent.GetBranch(NONE).LatestLock.Lock()
418
419 updatedRev := parent.GetBranch(NONE).GetLatest().UpdateChildren(ctx, typeName, children, parent.GetBranch(NONE))
420 parent.GetBranch(NONE).Node.makeLatest(parent.GetBranch(NONE), updatedRev, nil)
421
422 parent.GetBranch(NONE).LatestLock.Unlock()
423 // END lock parent
424
425 // Drop the previous child revision
426 parent.GetBranch(NONE).Latest.ChildDrop(typeName, childRev.GetHash())
427
428 if updatedChildRev != nil {
429 log.Debugw("verify-persisted-entry--adding-child", log.Fields{
430 "key": updatedChildRev.GetHash(),
431 "name": updatedChildRev.GetName(),
432 "data": updatedChildRev.GetData(),
433 })
434 response = updatedChildRev
435 }
436 } else {
437 if childRev != nil {
438 log.Debugw("keeping-revision-data", log.Fields{
439 "key": childRev.GetHash(),
440 "name": childRev.GetName(),
441 "data": childRev.GetData(),
442 })
443
444 // Update timestamp to reflect when it was last read and to reset tracked timeout
445 childRev.SetLastUpdate()
446 if childRev.getVersion() < version {
447 childRev.(*PersistedRevision).setVersion(version)
448 }
449 GetRevCache().Set(childRev.GetName(), childRev)
450 response = childRev
451 }
452 }
453
454 } else {
455 // There is no available child with that key value.
456 // Create a new child and update the parent revision.
457 log.Debugw("no-such-revision-entry", log.Fields{
458 "key": keyValue,
459 "name": typeName,
460 "data": data,
461 })
462
463 // BEGIN child lock
464 pr.GetBranch().LatestLock.Lock()
465
466 // Construct a new child node with the retrieved persistence data
467 childRev = pr.GetBranch().Node.MakeNode(data, txid).Latest(txid)
468
469 // We need to start watching this entry for future changes
470 childRev.SetName(typeName + "/" + keyValue)
471 childRev.SetupWatch(childRev.GetName())
472 childRev.(*PersistedRevision).setVersion(version)
473
474 // Add entry to cache
475 GetRevCache().Set(childRev.GetName(), childRev)
476
477 pr.GetBranch().LatestLock.Unlock()
478 // END child lock
479
480 //
481 // Add the child to the parent revision
482 //
483
484 // BEGIN parent lock
485 parent.GetBranch(NONE).LatestLock.Lock()
486 children = append(children, childRev)
487 updatedRev := parent.GetBranch(NONE).GetLatest().UpdateChildren(ctx, typeName, children, parent.GetBranch(NONE))
488 updatedRev.GetNode().SetProxy(parent.GetBranch(NONE).Node.GetProxy())
489 parent.GetBranch(NONE).Node.makeLatest(parent.GetBranch(NONE), updatedRev, nil)
490 parent.GetBranch(NONE).LatestLock.Unlock()
491 // END parent lock
492
493 // Child entry is valid and can be included in the response object
494 if childRev != nil {
495 log.Debugw("adding-revision-to-response", log.Fields{
496 "key": childRev.GetHash(),
497 "name": childRev.GetName(),
498 "data": childRev.GetData(),
499 })
500 response = childRev
501 }
502 }
503
504 return response
505}
506
507// LoadFromPersistence retrieves data from kv store at the specified location and refreshes the memory
508// by adding missing entries, updating changed entries and ignoring unchanged ones
509func (pr *PersistedRevision) LoadFromPersistence(ctx context.Context, path string, txid string, blobs map[string]*kvstore.KVPair) []Revision {
510 pr.mutex.Lock()
511 defer pr.mutex.Unlock()
512
513 log.Debugw("loading-from-persistence", log.Fields{"path": path, "txid": txid})
514
515 var response []Revision
516
517 for strings.HasPrefix(path, "/") {
518 path = path[1:]
519 }
520
521 if pr.kvStore != nil && path != "" {
522 if blobs == nil || len(blobs) == 0 {
523 log.Debugw("retrieve-from-kv", log.Fields{"path": path, "txid": txid})
524 blobs, _ = pr.kvStore.List(path)
525 }
526
527 partition := strings.SplitN(path, "/", 2)
528 name := partition[0]
529
530 var nodeType interface{}
531 if len(partition) < 2 {
532 path = ""
533 nodeType = pr.GetBranch().Node.Type
534 } else {
535 path = partition[1]
536 nodeType = pr.GetBranch().Node.GetRoot().Type
537 }
538
539 field := ChildrenFields(nodeType)[name]
540
541 if field != nil && field.IsContainer {
542 log.Debugw("parsing-data-blobs", log.Fields{
543 "path": path,
544 "name": name,
545 "size": len(blobs),
546 })
547
548 for _, blob := range blobs {
549 output := blob.Value.([]byte)
550
551 data := reflect.New(field.ClassType.Elem())
552
553 if err := proto.Unmarshal(output, data.Interface().(proto.Message)); err != nil {
554 log.Errorw("failed-to-unmarshal", log.Fields{
555 "path": path,
556 "txid": txid,
557 "error": err,
558 })
559 } else if path == "" {
560 if field.Key != "" {
561 log.Debugw("no-path-with-container-key", log.Fields{
562 "path": path,
563 "txid": txid,
564 "data": data.Interface(),
565 })
566
567 // Retrieve the key identifier value from the data structure
568 // based on the field's key attribute
569 _, key := GetAttributeValue(data.Interface(), field.Key, 0)
570
571 if entry := pr.verifyPersistedEntry(ctx, data.Interface(), name, field.Key, key.String(), txid, blob.Version); entry != nil {
572 response = append(response, entry)
573 }
574 } else {
575 log.Debugw("path-with-no-container-key", log.Fields{
576 "path": path,
577 "txid": txid,
578 "data": data.Interface(),
579 })
580 }
581
582 } else if field.Key != "" {
583 log.Debugw("path-with-container-key", log.Fields{
584 "path": path,
585 "txid": txid,
586 "data": data.Interface(),
587 })
588 // The request is for a specific entry/id
589 partition := strings.SplitN(path, "/", 2)
590 key := partition[0]
591 if len(partition) < 2 {
592 path = ""
593 } else {
594 path = partition[1]
595 }
596 keyValue := field.KeyFromStr(key)
597
598 if entry := pr.verifyPersistedEntry(ctx, data.Interface(), name, field.Key, keyValue.(string), txid, blob.Version); entry != nil {
599 response = append(response, entry)
600 }
601 }
602 }
603
604 log.Debugw("no-more-data-blobs", log.Fields{"path": path, "name": name})
605 } else {
606 log.Debugw("cannot-process-field", log.Fields{
607 "type": pr.GetBranch().Node.Type,
608 "name": name,
609 })
610 }
611 }
612
613 return response
614}