blob: fa35eca8614ad9f2e86ade4265622ed4e854180c [file] [log] [blame]
/*
* 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"
"github.com/golang/protobuf/proto"
"github.com/opencord/voltha-go/common/log"
"github.com/opencord/voltha-go/db/kvstore"
"reflect"
"runtime/debug"
"strings"
"sync"
)
// PersistedRevision holds information of revision meant to be saved in a persistent storage
type PersistedRevision struct {
Revision
Compress bool
events chan *kvstore.Event `json:"-"`
kvStore *Backend `json:"-"`
mutex sync.RWMutex `json:"-"`
isStored bool
isWatched 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.Revision = NewNonPersistedRevision(nil, branch, data, children)
return pr
}
// Finalize is responsible of saving the revision in the persistent storage
func (pr *PersistedRevision) Finalize(skipOnExist bool) {
pr.store(skipOnExist)
}
type revData struct {
Children map[string][]string
Config string
}
func (pr *PersistedRevision) store(skipOnExist bool) {
if pr.GetBranch().Txid != "" {
return
}
if pair, _ := pr.kvStore.Get(pr.GetHash()); pair != nil && skipOnExist {
log.Debugw("revision-config-already-exists", log.Fields{"hash": pr.GetHash()})
return
}
if blob, err := proto.Marshal(pr.GetConfig().Data.(proto.Message)); err != nil {
// TODO report error
} else {
if pr.Compress {
var b bytes.Buffer
w := gzip.NewWriter(&b)
w.Write(blob)
w.Close()
blob = b.Bytes()
}
if err := pr.kvStore.Put(pr.GetHash(), blob); err != nil {
log.Warnw("problem-storing-revision-config",
log.Fields{"error": err, "hash": pr.GetHash(), "data": pr.GetConfig().Data})
} else {
log.Debugw("storing-revision-config",
log.Fields{"hash": pr.GetHash(), "data": pr.GetConfig().Data})
pr.isStored = true
}
}
}
func (pr *PersistedRevision) SetupWatch(key string) {
if pr.events == nil {
pr.events = make(chan *kvstore.Event)
log.Debugw("setting-watch", log.Fields{"key": key})
pr.events = pr.kvStore.CreateWatch(key)
pr.isWatched = true
// Start watching
go pr.startWatching()
}
}
func (pr *PersistedRevision) startWatching() {
log.Debugw("starting-watch", log.Fields{"key": pr.GetHash()})
StopWatchLoop:
for {
select {
case event, ok := <-pr.events:
if !ok {
log.Errorw("event-channel-failure: stopping watch loop", log.Fields{"key": pr.GetHash()})
break StopWatchLoop
}
log.Debugw("received-event", log.Fields{"type": event.EventType})
switch event.EventType {
case kvstore.DELETE:
log.Debugw("delete-from-memory", log.Fields{"key": pr.GetHash()})
pr.Revision.Drop("", true)
break StopWatchLoop
case kvstore.PUT:
log.Debugw("update-in-memory", log.Fields{"key": pr.GetHash()})
if dataPair, err := pr.kvStore.Get(pr.GetHash()); err != nil || dataPair == nil {
log.Errorw("update-in-memory--key-retrieval-failed", log.Fields{"key": pr.GetHash(), "error": err})
} else {
data := reflect.New(reflect.TypeOf(pr.GetData()).Elem())
if err := proto.Unmarshal(dataPair.Value.([]byte), data.Interface().(proto.Message)); err != nil {
log.Errorw("update-in-memory--unmarshal-failed", log.Fields{"key": pr.GetHash(), "error": err})
} else {
// Apply changes to current revision
branch := pr.GetBranch()
rev := branch.GetLatest()
updatedRev := rev.UpdateData(data.Interface(), branch)
// ensure that we keep the previous hash value
updatedRev.SetHash(rev.GetHash())
// Save revision
branch.SetLatest(updatedRev)
}
}
default:
log.Debugw("unhandled-event", log.Fields{"key": pr.GetHash(), "type": event.EventType})
}
}
}
log.Debugw("exiting-watch", log.Fields{"key": pr.GetHash()})
}
func (pr *PersistedRevision) LoadFromPersistence(path string, txid string) []Revision {
log.Debugw("loading-from-persistence", log.Fields{"path": path, "txid": txid})
var response []Revision
var rev Revision
rev = pr
if pr.kvStore != nil {
blobMap, _ := pr.kvStore.List(path)
partition := strings.SplitN(path, "/", 2)
name := partition[0]
if len(partition) < 2 {
path = ""
} else {
path = partition[1]
}
field := ChildrenFields(rev.GetBranch().Node.Type)[name]
if field.IsContainer {
var children []Revision
children = make([]Revision, len(rev.GetChildren(name)))
copy(children, rev.GetChildren(name))
existChildMap := make(map[string]int)
for i, child := range rev.GetChildren(name) {
existChildMap[child.GetHash()] = i
}
for _, blob := range blobMap {
output := blob.Value.([]byte)
data := reflect.New(field.ClassType.Elem())
if err := proto.Unmarshal(output, data.Interface().(proto.Message)); err != nil {
log.Errorw(
"loading-from-persistence--failed-to-unmarshal",
log.Fields{"path": path, "txid": txid, "error": err},
)
} else if field.Key != "" {
var key reflect.Value
var keyValue interface{}
var keyStr string
if path == "" {
// e.g. /logical_devices --> path="" name=logical_devices key=""
_, key = GetAttributeValue(data.Interface(), field.Key, 0)
keyStr = key.String()
} else {
// e.g.
// /logical_devices/abcde --> path="abcde" name=logical_devices
// /logical_devices/abcde/image_downloads --> path="abcde/image_downloads" name=logical_devices
partition := strings.SplitN(path, "/", 2)
key := partition[0]
if len(partition) < 2 {
path = ""
} else {
path = partition[1]
}
keyValue = field.KeyFromStr(key)
keyStr = keyValue.(string)
if idx, childRev := rev.GetBranch().Node.findRevByKey(children, field.Key, keyValue); childRev != nil {
// Key is memory, continue recursing path
newChildRev := childRev.LoadFromPersistence(path, txid)
children[idx] = newChildRev[0]
rev := rev.UpdateChildren(name, rev.GetChildren(name), rev.GetBranch())
rev.GetBranch().Node.makeLatest(rev.GetBranch(), rev, nil)
response = append(response, newChildRev[0])
continue
}
}
childRev := rev.GetBranch().Node.MakeNode(data.Interface(), txid).Latest(txid)
childRev.SetHash(name + "/" + keyStr)
// Do not process a child that is already in memory
if _, childExists := existChildMap[childRev.GetHash()]; !childExists {
// Create watch for <component>/<key>
childRev.SetupWatch(childRev.GetHash())
children = append(children, childRev)
rev = rev.UpdateChildren(name, children, rev.GetBranch())
rev.GetBranch().Node.makeLatest(rev.GetBranch(), rev, nil)
}
response = append(response, childRev)
continue
}
}
}
}
return response
}
// UpdateData modifies the information in the data model and saves it in the persistent storage
func (pr *PersistedRevision) UpdateData(data interface{}, branch *Branch) Revision {
log.Debugw("updating-persisted-data", log.Fields{"hash": pr.GetHash()})
newNPR := pr.Revision.UpdateData(data, branch)
newPR := &PersistedRevision{
Revision: newNPR,
Compress: pr.Compress,
kvStore: pr.kvStore,
}
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(name string, children []Revision, branch *Branch) Revision {
log.Debugw("updating-persisted-children", log.Fields{"hash": pr.GetHash()})
newNPR := pr.Revision.UpdateChildren(name, children, branch)
newPR := &PersistedRevision{
Revision: newNPR,
Compress: pr.Compress,
kvStore: pr.kvStore,
}
return newPR
}
// UpdateAllChildren modifies the children for all components of a revision and saves it in the peristent storage
func (pr *PersistedRevision) UpdateAllChildren(children map[string][]Revision, branch *Branch) Revision {
log.Debugw("updating-all-persisted-children", log.Fields{"hash": pr.GetHash()})
newNPR := pr.Revision.UpdateAllChildren(children, branch)
newPR := &PersistedRevision{
Revision: newNPR,
Compress: pr.Compress,
kvStore: pr.kvStore,
}
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) {
log.Debugw("dropping-revision",
log.Fields{"txid": txid, "hash": pr.GetHash(), "config-hash": pr.GetConfig().Hash, "stack": string(debug.Stack())})
pr.mutex.Lock()
defer pr.mutex.Unlock()
if pr.kvStore != nil && txid == "" {
if pr.isStored {
if includeConfig {
if err := pr.kvStore.Delete(pr.GetConfig().Hash); err != nil {
log.Errorw("failed-to-remove-revision-config", log.Fields{"hash": pr.GetConfig().Hash, "error": err.Error()})
}
}
if err := pr.kvStore.Delete(pr.GetHash()); err != nil {
log.Errorw("failed-to-remove-revision", log.Fields{"hash": pr.GetHash(), "error": err.Error()})
} else {
pr.isStored = false
}
if pr.isWatched {
pr.kvStore.DeleteWatch(pr.GetHash(), pr.events)
pr.isWatched = false
}
}
} else {
if includeConfig {
log.Debugw("attempted-to-remove-transacted-revision-config", log.Fields{"hash": pr.GetConfig().Hash, "txid": txid})
}
log.Debugw("attempted-to-remove-transacted-revision", log.Fields{"hash": pr.GetHash(), "txid": txid})
}
pr.Revision.Drop(txid, includeConfig)
}