| /* |
| * Copyright 2018-present Open Networking Foundation |
| |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package model |
| |
| import ( |
| "bytes" |
| "compress/gzip" |
| "context" |
| "reflect" |
| "strings" |
| "sync" |
| |
| "github.com/golang/protobuf/proto" |
| "github.com/opencord/voltha-lib-go/v3/pkg/db" |
| "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore" |
| "github.com/opencord/voltha-lib-go/v3/pkg/log" |
| ) |
| |
| // PersistedRevision holds information of revision meant to be saved in a persistent storage |
| type PersistedRevision struct { |
| Revision |
| Compress bool |
| |
| events chan *kvstore.Event |
| kvStore *db.Backend |
| mutex sync.RWMutex |
| versionMutex sync.RWMutex |
| Version int64 |
| isStored bool |
| } |
| |
| // NewPersistedRevision creates a new instance of a PersistentRevision structure |
| func NewPersistedRevision(branch *Branch, data interface{}, children map[string][]Revision) Revision { |
| pr := &PersistedRevision{} |
| pr.kvStore = branch.Node.GetRoot().KvStore |
| pr.Version = 1 |
| pr.Revision = NewNonPersistedRevision(nil, branch, data, children) |
| return pr |
| } |
| |
| func (pr *PersistedRevision) getVersion() int64 { |
| pr.versionMutex.RLock() |
| defer pr.versionMutex.RUnlock() |
| return pr.Version |
| } |
| |
| func (pr *PersistedRevision) setVersion(version int64) { |
| pr.versionMutex.Lock() |
| defer pr.versionMutex.Unlock() |
| pr.Version = version |
| } |
| |
| // Finalize is responsible of saving the revision in the persistent storage |
| func (pr *PersistedRevision) Finalize(ctx context.Context, skipOnExist bool) { |
| pr.store(ctx, skipOnExist) |
| } |
| |
| func (pr *PersistedRevision) store(ctx context.Context, skipOnExist bool) { |
| if pr.GetBranch().Txid != "" { |
| return |
| } |
| |
| logger.Debugw("ready-to-store-revision", log.Fields{"hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetData()}) |
| |
| // clone the revision data to avoid any race conditions with processes |
| // accessing the same data |
| cloned := proto.Clone(pr.GetConfig().Data.(proto.Message)) |
| |
| if blob, err := proto.Marshal(cloned); err != nil { |
| logger.Errorw("problem-to-marshal", log.Fields{"error": err, "hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetData()}) |
| } else { |
| if pr.Compress { |
| var b bytes.Buffer |
| w := gzip.NewWriter(&b) |
| if _, err := w.Write(blob); err != nil { |
| logger.Errorw("Unable to write a compressed form of p to the underlying io.Writer.", log.Fields{"error": err}) |
| } |
| w.Close() |
| blob = b.Bytes() |
| } |
| |
| getRevCache().Set(pr.GetName(), pr) |
| if err := pr.kvStore.Put(ctx, pr.GetName(), blob); err != nil { |
| logger.Warnw("problem-storing-revision", log.Fields{"error": err, "hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetConfig().Data}) |
| } else { |
| logger.Debugw("storing-revision", log.Fields{"hash": pr.GetHash(), "name": pr.GetName(), "data": pr.GetConfig().Data, "version": pr.getVersion()}) |
| pr.isStored = true |
| } |
| } |
| } |
| |
| // UpdateData modifies the information in the data model and saves it in the persistent storage |
| func (pr *PersistedRevision) UpdateData(ctx context.Context, data interface{}, branch *Branch) Revision { |
| logger.Debugw("updating-persisted-data", log.Fields{"hash": pr.GetHash()}) |
| |
| newNPR := pr.Revision.UpdateData(ctx, data, branch) |
| |
| newPR := &PersistedRevision{ |
| Revision: newNPR, |
| Compress: pr.Compress, |
| kvStore: pr.kvStore, |
| events: pr.events, |
| Version: pr.getVersion(), |
| } |
| |
| if newPR.GetHash() != pr.GetHash() { |
| newPR.isStored = false |
| pr.Drop(branch.Txid, false) |
| pr.Drop(branch.Txid, false) |
| } else { |
| newPR.isStored = true |
| } |
| |
| return newPR |
| } |
| |
| // UpdateChildren modifies the children of a revision and of a specific component and saves it in the persistent storage |
| func (pr *PersistedRevision) UpdateChildren(ctx context.Context, name string, children []Revision, branch *Branch) Revision { |
| logger.Debugw("updating-persisted-children", log.Fields{"hash": pr.GetHash()}) |
| |
| newNPR := pr.Revision.UpdateChildren(ctx, name, children, branch) |
| |
| newPR := &PersistedRevision{ |
| Revision: newNPR, |
| Compress: pr.Compress, |
| kvStore: pr.kvStore, |
| events: pr.events, |
| Version: pr.getVersion(), |
| } |
| |
| if newPR.GetHash() != pr.GetHash() { |
| newPR.isStored = false |
| pr.Drop(branch.Txid, false) |
| } else { |
| newPR.isStored = true |
| } |
| |
| return newPR |
| } |
| |
| // UpdateAllChildren modifies the children for all components of a revision and saves it in the peristent storage |
| func (pr *PersistedRevision) UpdateAllChildren(ctx context.Context, children map[string][]Revision, branch *Branch) Revision { |
| logger.Debugw("updating-all-persisted-children", log.Fields{"hash": pr.GetHash()}) |
| |
| newNPR := pr.Revision.UpdateAllChildren(ctx, children, branch) |
| |
| newPR := &PersistedRevision{ |
| Revision: newNPR, |
| Compress: pr.Compress, |
| kvStore: pr.kvStore, |
| events: pr.events, |
| Version: pr.getVersion(), |
| } |
| |
| if newPR.GetHash() != pr.GetHash() { |
| newPR.isStored = false |
| pr.Drop(branch.Txid, false) |
| } else { |
| newPR.isStored = true |
| } |
| |
| return newPR |
| } |
| |
| // Drop takes care of eliminating a revision hash that is no longer needed |
| // and its associated config when required |
| func (pr *PersistedRevision) Drop(txid string, includeConfig bool) { |
| pr.Revision.Drop(txid, includeConfig) |
| } |
| |
| // StorageDrop takes care of eliminating a revision hash that is no longer needed |
| // and its associated config when required |
| func (pr *PersistedRevision) StorageDrop(ctx context.Context, txid string, includeConfig bool) { |
| logger.Debugw("dropping-revision", log.Fields{"txid": txid, "hash": pr.GetHash(), "config-hash": pr.GetConfig().Hash, "key": pr.GetName(), "isStored": pr.isStored}) |
| |
| pr.mutex.Lock() |
| defer pr.mutex.Unlock() |
| if pr.kvStore != nil && txid == "" { |
| if err := pr.kvStore.Delete(ctx, pr.GetName()); err != nil { |
| logger.Errorw("failed-to-remove-revision", log.Fields{"hash": pr.GetHash(), "error": err.Error()}) |
| } else { |
| pr.isStored = false |
| } |
| } else { |
| if includeConfig { |
| logger.Debugw("attempted-to-remove-transacted-revision-config", log.Fields{"hash": pr.GetConfig().Hash, "txid": txid}) |
| } |
| logger.Debugw("attempted-to-remove-transacted-revision", log.Fields{"hash": pr.GetHash(), "txid": txid}) |
| } |
| |
| pr.Revision.Drop(txid, includeConfig) |
| } |
| |
| // verifyPersistedEntry validates if the provided data is available or not in memory and applies updates as required |
| func (pr *PersistedRevision) verifyPersistedEntry(ctx context.Context, data interface{}, typeName string, keyName string, |
| keyValue string, txid string, version int64) (response Revision) { |
| // Parent which holds the current node entry |
| parent := pr.GetBranch().Node.GetRoot() |
| |
| // Get a copy of the parent's children |
| children := make([]Revision, len(parent.GetBranch(NONE).Latest.GetChildren(typeName))) |
| copy(children, parent.GetBranch(NONE).Latest.GetChildren(typeName)) |
| |
| // Verify if a child with the provided key value can be found |
| if childIdx, childRev := pr.getNode().findRevByKey(children, keyName, keyValue); childRev != nil { |
| // A child matching the provided key exists in memory |
| // Verify if the data differs from what was retrieved from persistence |
| // Also check if we are treating a newer revision of the data or not |
| if childRev.GetData().(proto.Message).String() != data.(proto.Message).String() && childRev.getVersion() < version { |
| logger.Debugw("revision-data-is-different", log.Fields{ |
| "key": childRev.GetHash(), |
| "name": childRev.GetName(), |
| "data": childRev.GetData(), |
| "in-memory-version": childRev.getVersion(), |
| "persisted-version": version, |
| }) |
| |
| // |
| // Data has changed; replace the child entry and update the parent revision |
| // |
| |
| // BEGIN Lock child -- prevent any incoming changes |
| childRev.GetBranch().LatestLock.Lock() |
| |
| // Update child |
| updatedChildRev := childRev.UpdateData(ctx, data, childRev.GetBranch()) |
| |
| updatedChildRev.getNode().SetProxy(childRev.getNode().GetProxy()) |
| updatedChildRev.SetLastUpdate() |
| updatedChildRev.(*PersistedRevision).setVersion(version) |
| |
| // Update cache |
| getRevCache().Set(updatedChildRev.GetName(), updatedChildRev) |
| childRev.Drop(txid, false) |
| |
| childRev.GetBranch().LatestLock.Unlock() |
| // END lock child |
| |
| // Update child entry |
| children[childIdx] = updatedChildRev |
| |
| // BEGIN lock parent -- Update parent |
| parent.GetBranch(NONE).LatestLock.Lock() |
| |
| updatedRev := parent.GetBranch(NONE).GetLatest().UpdateChildren(ctx, typeName, children, parent.GetBranch(NONE)) |
| parent.GetBranch(NONE).Node.makeLatest(parent.GetBranch(NONE), updatedRev, nil) |
| |
| parent.GetBranch(NONE).LatestLock.Unlock() |
| // END lock parent |
| |
| // Drop the previous child revision |
| parent.GetBranch(NONE).Latest.ChildDrop(typeName, childRev.GetHash()) |
| |
| if updatedChildRev != nil { |
| logger.Debugw("verify-persisted-entry--adding-child", log.Fields{ |
| "key": updatedChildRev.GetHash(), |
| "name": updatedChildRev.GetName(), |
| "data": updatedChildRev.GetData(), |
| }) |
| response = updatedChildRev |
| } |
| } else { |
| logger.Debugw("keeping-revision-data", log.Fields{ |
| "key": childRev.GetHash(), |
| "name": childRev.GetName(), |
| "data": childRev.GetData(), |
| "in-memory-version": childRev.getVersion(), |
| "persistence-version": version, |
| }) |
| |
| // Update timestamp to reflect when it was last read and to reset tracked timeout |
| childRev.SetLastUpdate() |
| if childRev.getVersion() < version { |
| childRev.(*PersistedRevision).setVersion(version) |
| } |
| getRevCache().Set(childRev.GetName(), childRev) |
| response = childRev |
| } |
| |
| } else { |
| // There is no available child with that key value. |
| // Create a new child and update the parent revision. |
| logger.Debugw("no-such-revision-entry", log.Fields{ |
| "key": keyValue, |
| "name": typeName, |
| "data": data, |
| "version": version, |
| }) |
| |
| // BEGIN child lock |
| pr.GetBranch().LatestLock.Lock() |
| |
| // Construct a new child node with the retrieved persistence data |
| childRev = pr.GetBranch().Node.MakeNode(data, txid).Latest(txid) |
| |
| // We need to start watching this entry for future changes |
| childRev.SetName(typeName + "/" + keyValue) |
| childRev.(*PersistedRevision).setVersion(version) |
| |
| // Add entry to cache |
| getRevCache().Set(childRev.GetName(), childRev) |
| |
| pr.GetBranch().LatestLock.Unlock() |
| // END child lock |
| |
| // |
| // Add the child to the parent revision |
| // |
| |
| // BEGIN parent lock |
| parent.GetBranch(NONE).LatestLock.Lock() |
| children = append(children, childRev) |
| updatedRev := parent.GetBranch(NONE).GetLatest().UpdateChildren(ctx, typeName, children, parent.GetBranch(NONE)) |
| updatedRev.getNode().SetProxy(parent.GetBranch(NONE).Node.GetProxy()) |
| parent.GetBranch(NONE).Node.makeLatest(parent.GetBranch(NONE), updatedRev, nil) |
| parent.GetBranch(NONE).LatestLock.Unlock() |
| // END parent lock |
| |
| // Child entry is valid and can be included in the response object |
| if childRev != nil { |
| logger.Debugw("adding-revision-to-response", log.Fields{ |
| "key": childRev.GetHash(), |
| "name": childRev.GetName(), |
| "data": childRev.GetData(), |
| }) |
| response = childRev |
| } |
| } |
| |
| return response |
| } |
| |
| // LoadFromPersistence retrieves data from kv store at the specified location and refreshes the memory |
| // by adding missing entries, updating changed entries and ignoring unchanged ones |
| func (pr *PersistedRevision) LoadFromPersistence(ctx context.Context, path string, txid string, blobs map[string]*kvstore.KVPair) ([]Revision, error) { |
| pr.mutex.Lock() |
| defer pr.mutex.Unlock() |
| |
| logger.Debugw("loading-from-persistence", log.Fields{"path": path, "txid": txid}) |
| |
| var response []Revision |
| var err error |
| |
| for strings.HasPrefix(path, "/") { |
| path = path[1:] |
| } |
| |
| if pr.kvStore != nil && path != "" { |
| if len(blobs) == 0 { |
| logger.Debugw("retrieve-from-kv", log.Fields{"path": path, "txid": txid}) |
| |
| if blobs, err = pr.kvStore.List(ctx, path); err != nil { |
| logger.Errorw("failed-to-retrieve-data-from-kvstore", log.Fields{"error": err}) |
| return nil, err |
| } |
| } |
| |
| partition := strings.SplitN(path, "/", 2) |
| name := partition[0] |
| |
| var nodeType interface{} |
| if len(partition) < 2 { |
| path = "" |
| nodeType = pr.GetBranch().Node.Type |
| } else { |
| path = partition[1] |
| nodeType = pr.GetBranch().Node.GetRoot().Type |
| } |
| |
| field := ChildrenFields(nodeType)[name] |
| |
| if field != nil && field.IsContainer { |
| logger.Debugw("parsing-data-blobs", log.Fields{ |
| "path": path, |
| "name": name, |
| "size": len(blobs), |
| }) |
| |
| for _, blob := range blobs { |
| output := blob.Value.([]byte) |
| |
| data := reflect.New(field.ClassType.Elem()) |
| |
| if err := proto.Unmarshal(output, data.Interface().(proto.Message)); err != nil { |
| logger.Errorw("failed-to-unmarshal", log.Fields{ |
| "path": path, |
| "txid": txid, |
| "error": err, |
| }) |
| } else if path == "" { |
| if field.Key != "" { |
| logger.Debugw("no-path-with-container-key", log.Fields{ |
| "path": path, |
| "txid": txid, |
| "data": data.Interface(), |
| }) |
| |
| // Retrieve the key identifier value from the data structure |
| // based on the field's key attribute |
| _, key := GetAttributeValue(data.Interface(), field.Key, 0) |
| |
| if entry := pr.verifyPersistedEntry(ctx, data.Interface(), name, field.Key, key.String(), txid, blob.Version); entry != nil { |
| response = append(response, entry) |
| } |
| } else { |
| logger.Debugw("path-with-no-container-key", log.Fields{ |
| "path": path, |
| "txid": txid, |
| "data": data.Interface(), |
| }) |
| } |
| |
| } else if field.Key != "" { |
| logger.Debugw("path-with-container-key", log.Fields{ |
| "path": path, |
| "txid": txid, |
| "data": data.Interface(), |
| }) |
| // The request is for a specific entry/id |
| partition := strings.SplitN(path, "/", 2) |
| key := partition[0] |
| if len(partition) < 2 { |
| path = "" |
| } else { |
| path = partition[1] |
| } |
| keyValue := field.KeyFromStr(key) |
| |
| if entry := pr.verifyPersistedEntry(ctx, data.Interface(), name, field.Key, keyValue.(string), txid, blob.Version); entry != nil { |
| response = append(response, entry) |
| } |
| } |
| } |
| |
| logger.Debugw("no-more-data-blobs", log.Fields{"path": path, "name": name}) |
| } else { |
| logger.Debugw("cannot-process-field", log.Fields{ |
| "type": pr.GetBranch().Node.Type, |
| "name": name, |
| }) |
| } |
| } |
| |
| return response, nil |
| } |