blob: 53d93b7b4ba93b3317f0f48a00b3660fef640444 [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 Bakerce767002019-10-23 13:30:24 -070025 "github.com/opencord/voltha-lib-go/v2/pkg/db/kvstore"
26 "github.com/opencord/voltha-lib-go/v2/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()})
Stephane Barbarie42b97a42019-10-23 11:05:04 -0400165
166 // Remove reference from cache
167 GetRevCache().Delete(latestRev.GetName())
168
169 // Remove reference from parent
170 parent := pr.GetBranch().Node.GetRoot()
171 parent.GetBranch(NONE).Latest.ChildDropByName(latestRev.GetName())
172
Scott Baker2c1c4822019-10-16 11:02:41 -0700173 break StopWatchLoop
174
175 case kvstore.PUT:
176 log.Debugw("update-in-memory", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
177 if latestRev.getVersion() >= event.Version {
178 log.Debugw("skipping-matching-or-older-revision", log.Fields{
179 "watch": latestRev.GetName(),
180 "watch-version": event.Version,
181 "latest-version": latestRev.getVersion(),
182 })
183 continue
184 } else {
185 log.Debugw("watch-revision-is-newer", log.Fields{
186 "watch": latestRev.GetName(),
187 "watch-version": event.Version,
188 "latest-version": latestRev.getVersion(),
189 })
190 }
191
192 data := reflect.New(reflect.TypeOf(latestRev.GetData()).Elem())
193
194 if err := proto.Unmarshal(event.Value.([]byte), data.Interface().(proto.Message)); err != nil {
195 log.Errorw("failed-to-unmarshal-watch-data", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "error": err})
196 } else {
197 log.Debugw("un-marshaled-watch-data", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "data": data.Interface()})
198
199 var pathLock string
200 var blobs map[string]*kvstore.KVPair
201
202 // The watch reported new persistence data.
203 // Construct an object that will be used to update the memory
204 blobs = make(map[string]*kvstore.KVPair)
205 key, _ := kvstore.ToString(event.Key)
206 blobs[key] = &kvstore.KVPair{
207 Key: key,
208 Value: event.Value,
209 Session: "",
210 Lease: 0,
211 Version: event.Version,
212 }
213
214 if latestRev.GetNode().GetProxy() != nil {
215 //
216 // If a proxy exists for this revision, use it to lock access to the path
217 // and prevent simultaneous updates to the object in memory
218 //
219
220 //If the proxy already has a request in progress, then there is no need to process the watch
221 if latestRev.GetNode().GetProxy().GetOperation() != PROXY_NONE {
222 log.Debugw("operation-in-progress", log.Fields{
223 "key": latestRev.GetHash(),
224 "path": latestRev.GetNode().GetProxy().getFullPath(),
225 "operation": latestRev.GetNode().GetProxy().operation.String(),
226 })
227 continue
228 }
229
230 pathLock, _ = latestRev.GetNode().GetProxy().parseForControlledPath(latestRev.GetNode().GetProxy().getFullPath())
231
232 // Reserve the path to prevent others to modify while we reload from persistence
233 latestRev.GetNode().GetProxy().GetRoot().KvStore.Client.Reserve(pathLock+"_", uuid.New().String(), ReservationTTL)
234 latestRev.GetNode().GetProxy().SetOperation(PROXY_WATCH)
235
236 // Load changes and apply to memory
237 latestRev.LoadFromPersistence(context.Background(), latestRev.GetName(), "", blobs)
238
239 // Release path
240 latestRev.GetNode().GetProxy().GetRoot().KvStore.Client.ReleaseReservation(pathLock + "_")
241
242 } else {
243 // This block should be reached only if coming from a non-proxied request
244 log.Debugw("revision-with-no-proxy", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName()})
245
246 // Load changes and apply to memory
247 latestRev.LoadFromPersistence(context.Background(), latestRev.GetName(), "", blobs)
248 }
249 }
250
251 default:
252 log.Debugw("unhandled-event", log.Fields{"key": latestRev.GetHash(), "watch": latestRev.GetName(), "type": event.EventType})
253 }
254 }
255 }
256
257 Watches().Cache.Delete(pr.GetName() + "-" + pr.GetHash())
258
259 log.Debugw("exiting-watch", log.Fields{"key": pr.GetHash(), "watch": pr.GetName()})
260}
261
262// UpdateData modifies the information in the data model and saves it in the persistent storage
263func (pr *PersistedRevision) UpdateData(ctx context.Context, data interface{}, branch *Branch) Revision {
264 log.Debugw("updating-persisted-data", log.Fields{"hash": pr.GetHash()})
265
266 newNPR := pr.Revision.UpdateData(ctx, data, branch)
267
268 newPR := &PersistedRevision{
269 Revision: newNPR,
270 Compress: pr.Compress,
271 kvStore: pr.kvStore,
272 events: pr.events,
273 Version: pr.getVersion(),
274 isWatched: pr.isWatched,
275 }
276
277 if newPR.GetHash() != pr.GetHash() {
278 newPR.isStored = false
279 pr.Drop(branch.Txid, false)
280 pr.Drop(branch.Txid, false)
281 } else {
282 newPR.isStored = true
283 }
284
285 return newPR
286}
287
288// UpdateChildren modifies the children of a revision and of a specific component and saves it in the persistent storage
289func (pr *PersistedRevision) UpdateChildren(ctx context.Context, name string, children []Revision, branch *Branch) Revision {
290 log.Debugw("updating-persisted-children", log.Fields{"hash": pr.GetHash()})
291
292 newNPR := pr.Revision.UpdateChildren(ctx, name, children, branch)
293
294 newPR := &PersistedRevision{
295 Revision: newNPR,
296 Compress: pr.Compress,
297 kvStore: pr.kvStore,
298 events: pr.events,
299 Version: pr.getVersion(),
300 isWatched: pr.isWatched,
301 }
302
303 if newPR.GetHash() != pr.GetHash() {
304 newPR.isStored = false
305 pr.Drop(branch.Txid, false)
306 } else {
307 newPR.isStored = true
308 }
309
310 return newPR
311}
312
313// UpdateAllChildren modifies the children for all components of a revision and saves it in the peristent storage
314func (pr *PersistedRevision) UpdateAllChildren(children map[string][]Revision, branch *Branch) Revision {
315 log.Debugw("updating-all-persisted-children", log.Fields{"hash": pr.GetHash()})
316
317 newNPR := pr.Revision.UpdateAllChildren(children, branch)
318
319 newPR := &PersistedRevision{
320 Revision: newNPR,
321 Compress: pr.Compress,
322 kvStore: pr.kvStore,
323 events: pr.events,
324 Version: pr.getVersion(),
325 isWatched: pr.isWatched,
326 }
327
328 if newPR.GetHash() != pr.GetHash() {
329 newPR.isStored = false
330 pr.Drop(branch.Txid, false)
331 } else {
332 newPR.isStored = true
333 }
334
335 return newPR
336}
337
338// Drop takes care of eliminating a revision hash that is no longer needed
339// and its associated config when required
340func (pr *PersistedRevision) Drop(txid string, includeConfig bool) {
341 pr.Revision.Drop(txid, includeConfig)
342}
343
344// Drop takes care of eliminating a revision hash that is no longer needed
345// and its associated config when required
346func (pr *PersistedRevision) StorageDrop(txid string, includeConfig bool) {
347 log.Debugw("dropping-revision", log.Fields{"txid": txid, "hash": pr.GetHash(), "config-hash": pr.GetConfig().Hash})
348
349 pr.mutex.Lock()
350 defer pr.mutex.Unlock()
351 if pr.kvStore != nil && txid == "" {
352 if pr.isStored {
353 if pr.isWatched {
354 pr.kvStore.DeleteWatch(pr.GetName(), pr.events)
355 pr.isWatched = false
356 }
357
358 if err := pr.kvStore.Delete(pr.GetName()); err != nil {
359 log.Errorw("failed-to-remove-revision", log.Fields{"hash": pr.GetHash(), "error": err.Error()})
360 } else {
361 pr.isStored = false
362 }
363 }
364
365 } else {
366 if includeConfig {
367 log.Debugw("attempted-to-remove-transacted-revision-config", log.Fields{"hash": pr.GetConfig().Hash, "txid": txid})
368 }
369 log.Debugw("attempted-to-remove-transacted-revision", log.Fields{"hash": pr.GetHash(), "txid": txid})
370 }
371
372 pr.Revision.Drop(txid, includeConfig)
373}
374
375// verifyPersistedEntry validates if the provided data is available or not in memory and applies updates as required
376func (pr *PersistedRevision) verifyPersistedEntry(ctx context.Context, data interface{}, typeName string, keyName string,
377 keyValue string, txid string, version int64) (response Revision) {
378 // Parent which holds the current node entry
379 parent := pr.GetBranch().Node.GetRoot()
380
381 // Get a copy of the parent's children
382 children := make([]Revision, len(parent.GetBranch(NONE).Latest.GetChildren(typeName)))
383 copy(children, parent.GetBranch(NONE).Latest.GetChildren(typeName))
384
385 // Verify if a child with the provided key value can be found
386 if childIdx, childRev := pr.GetNode().findRevByKey(children, keyName, keyValue); childRev != nil {
387 // A child matching the provided key exists in memory
388 // Verify if the data differs from what was retrieved from persistence
389 // Also check if we are treating a newer revision of the data or not
390 if childRev.GetData().(proto.Message).String() != data.(proto.Message).String() && childRev.getVersion() < version {
391 log.Debugw("revision-data-is-different", log.Fields{
Stephane Barbarie42b97a42019-10-23 11:05:04 -0400392 "key": childRev.GetHash(),
393 "name": childRev.GetName(),
394 "data": childRev.GetData(),
395 "in-memory-version": childRev.getVersion(),
396 "persisted-version": version,
Scott Baker2c1c4822019-10-16 11:02:41 -0700397 })
398
399 //
400 // Data has changed; replace the child entry and update the parent revision
401 //
402
403 // BEGIN Lock child -- prevent any incoming changes
404 childRev.GetBranch().LatestLock.Lock()
405
406 // Update child
407 updatedChildRev := childRev.UpdateData(ctx, data, childRev.GetBranch())
408
409 updatedChildRev.GetNode().SetProxy(childRev.GetNode().GetProxy())
410 updatedChildRev.SetupWatch(updatedChildRev.GetName())
411 updatedChildRev.SetLastUpdate()
412 updatedChildRev.(*PersistedRevision).setVersion(version)
413
414 // Update cache
415 GetRevCache().Set(updatedChildRev.GetName(), updatedChildRev)
416 childRev.Drop(txid, false)
417
418 childRev.GetBranch().LatestLock.Unlock()
419 // END lock child
420
421 // Update child entry
422 children[childIdx] = updatedChildRev
423
424 // BEGIN lock parent -- Update parent
425 parent.GetBranch(NONE).LatestLock.Lock()
426
427 updatedRev := parent.GetBranch(NONE).GetLatest().UpdateChildren(ctx, typeName, children, parent.GetBranch(NONE))
428 parent.GetBranch(NONE).Node.makeLatest(parent.GetBranch(NONE), updatedRev, nil)
429
430 parent.GetBranch(NONE).LatestLock.Unlock()
431 // END lock parent
432
433 // Drop the previous child revision
434 parent.GetBranch(NONE).Latest.ChildDrop(typeName, childRev.GetHash())
435
436 if updatedChildRev != nil {
437 log.Debugw("verify-persisted-entry--adding-child", log.Fields{
438 "key": updatedChildRev.GetHash(),
439 "name": updatedChildRev.GetName(),
440 "data": updatedChildRev.GetData(),
441 })
442 response = updatedChildRev
443 }
444 } else {
445 if childRev != nil {
446 log.Debugw("keeping-revision-data", log.Fields{
Stephane Barbarie42b97a42019-10-23 11:05:04 -0400447 "key": childRev.GetHash(),
448 "name": childRev.GetName(),
449 "data": childRev.GetData(),
450 "in-memory-version": childRev.getVersion(),
451 "persistence-version": version,
Scott Baker2c1c4822019-10-16 11:02:41 -0700452 })
453
454 // Update timestamp to reflect when it was last read and to reset tracked timeout
455 childRev.SetLastUpdate()
456 if childRev.getVersion() < version {
457 childRev.(*PersistedRevision).setVersion(version)
458 }
459 GetRevCache().Set(childRev.GetName(), childRev)
460 response = childRev
461 }
462 }
463
464 } else {
465 // There is no available child with that key value.
466 // Create a new child and update the parent revision.
467 log.Debugw("no-such-revision-entry", log.Fields{
Stephane Barbarie42b97a42019-10-23 11:05:04 -0400468 "key": keyValue,
469 "name": typeName,
470 "data": data,
471 "version": version,
Scott Baker2c1c4822019-10-16 11:02:41 -0700472 })
473
474 // BEGIN child lock
475 pr.GetBranch().LatestLock.Lock()
476
477 // Construct a new child node with the retrieved persistence data
478 childRev = pr.GetBranch().Node.MakeNode(data, txid).Latest(txid)
479
480 // We need to start watching this entry for future changes
481 childRev.SetName(typeName + "/" + keyValue)
482 childRev.SetupWatch(childRev.GetName())
483 childRev.(*PersistedRevision).setVersion(version)
484
485 // Add entry to cache
486 GetRevCache().Set(childRev.GetName(), childRev)
487
488 pr.GetBranch().LatestLock.Unlock()
489 // END child lock
490
491 //
492 // Add the child to the parent revision
493 //
494
495 // BEGIN parent lock
496 parent.GetBranch(NONE).LatestLock.Lock()
497 children = append(children, childRev)
498 updatedRev := parent.GetBranch(NONE).GetLatest().UpdateChildren(ctx, typeName, children, parent.GetBranch(NONE))
499 updatedRev.GetNode().SetProxy(parent.GetBranch(NONE).Node.GetProxy())
500 parent.GetBranch(NONE).Node.makeLatest(parent.GetBranch(NONE), updatedRev, nil)
501 parent.GetBranch(NONE).LatestLock.Unlock()
502 // END parent lock
503
504 // Child entry is valid and can be included in the response object
505 if childRev != nil {
506 log.Debugw("adding-revision-to-response", log.Fields{
507 "key": childRev.GetHash(),
508 "name": childRev.GetName(),
509 "data": childRev.GetData(),
510 })
511 response = childRev
512 }
513 }
514
515 return response
516}
517
518// LoadFromPersistence retrieves data from kv store at the specified location and refreshes the memory
519// by adding missing entries, updating changed entries and ignoring unchanged ones
520func (pr *PersistedRevision) LoadFromPersistence(ctx context.Context, path string, txid string, blobs map[string]*kvstore.KVPair) []Revision {
521 pr.mutex.Lock()
522 defer pr.mutex.Unlock()
523
524 log.Debugw("loading-from-persistence", log.Fields{"path": path, "txid": txid})
525
526 var response []Revision
527
528 for strings.HasPrefix(path, "/") {
529 path = path[1:]
530 }
531
532 if pr.kvStore != nil && path != "" {
533 if blobs == nil || len(blobs) == 0 {
534 log.Debugw("retrieve-from-kv", log.Fields{"path": path, "txid": txid})
535 blobs, _ = pr.kvStore.List(path)
536 }
537
538 partition := strings.SplitN(path, "/", 2)
539 name := partition[0]
540
541 var nodeType interface{}
542 if len(partition) < 2 {
543 path = ""
544 nodeType = pr.GetBranch().Node.Type
545 } else {
546 path = partition[1]
547 nodeType = pr.GetBranch().Node.GetRoot().Type
548 }
549
550 field := ChildrenFields(nodeType)[name]
551
552 if field != nil && field.IsContainer {
553 log.Debugw("parsing-data-blobs", log.Fields{
554 "path": path,
555 "name": name,
556 "size": len(blobs),
557 })
558
559 for _, blob := range blobs {
560 output := blob.Value.([]byte)
561
562 data := reflect.New(field.ClassType.Elem())
563
564 if err := proto.Unmarshal(output, data.Interface().(proto.Message)); err != nil {
565 log.Errorw("failed-to-unmarshal", log.Fields{
566 "path": path,
567 "txid": txid,
568 "error": err,
569 })
570 } else if path == "" {
571 if field.Key != "" {
572 log.Debugw("no-path-with-container-key", log.Fields{
573 "path": path,
574 "txid": txid,
575 "data": data.Interface(),
576 })
577
578 // Retrieve the key identifier value from the data structure
579 // based on the field's key attribute
580 _, key := GetAttributeValue(data.Interface(), field.Key, 0)
581
582 if entry := pr.verifyPersistedEntry(ctx, data.Interface(), name, field.Key, key.String(), txid, blob.Version); entry != nil {
583 response = append(response, entry)
584 }
585 } else {
586 log.Debugw("path-with-no-container-key", log.Fields{
587 "path": path,
588 "txid": txid,
589 "data": data.Interface(),
590 })
591 }
592
593 } else if field.Key != "" {
594 log.Debugw("path-with-container-key", log.Fields{
595 "path": path,
596 "txid": txid,
597 "data": data.Interface(),
598 })
599 // The request is for a specific entry/id
600 partition := strings.SplitN(path, "/", 2)
601 key := partition[0]
602 if len(partition) < 2 {
603 path = ""
604 } else {
605 path = partition[1]
606 }
607 keyValue := field.KeyFromStr(key)
608
609 if entry := pr.verifyPersistedEntry(ctx, data.Interface(), name, field.Key, keyValue.(string), txid, blob.Version); entry != nil {
610 response = append(response, entry)
611 }
612 }
613 }
614
615 log.Debugw("no-more-data-blobs", log.Fields{"path": path, "name": name})
616 } else {
617 log.Debugw("cannot-process-field", log.Fields{
618 "type": pr.GetBranch().Node.Type,
619 "name": name,
620 })
621 }
622 }
623
624 return response
625}