mirror of
https://github.com/yunionio/cloudpods.git
synced 2026-06-04 08:24:26 +08:00
322 lines
9.2 KiB
Go
322 lines
9.2 KiB
Go
// Copyright 2019 Yunion
|
|
//
|
|
// 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 db
|
|
|
|
import (
|
|
"context"
|
|
"reflect"
|
|
"runtime/debug"
|
|
"time"
|
|
|
|
"yunion.io/x/jsonutils"
|
|
"yunion.io/x/log"
|
|
"yunion.io/x/pkg/errors"
|
|
"yunion.io/x/pkg/gotypes"
|
|
"yunion.io/x/sqlchemy"
|
|
|
|
api "yunion.io/x/onecloud/pkg/apis/notify"
|
|
"yunion.io/x/onecloud/pkg/cloudcommon/informer"
|
|
"yunion.io/x/onecloud/pkg/util/nopanic"
|
|
"yunion.io/x/onecloud/pkg/util/splitable"
|
|
)
|
|
|
|
type ITableSpec interface {
|
|
Name() string
|
|
Columns() []sqlchemy.IColumnSpec
|
|
PrimaryColumns() []sqlchemy.IColumnSpec
|
|
DataType() reflect.Type
|
|
// CreateSQL() string
|
|
Instance() *sqlchemy.STable
|
|
ColumnSpec(name string) sqlchemy.IColumnSpec
|
|
Insert(ctx context.Context, dt interface{}) error
|
|
InsertOrUpdate(ctx context.Context, dt interface{}) error
|
|
Update(ctx context.Context, dt interface{}, doUpdate func() error) (sqlchemy.UpdateDiffs, error)
|
|
Fetch(dt interface{}) error
|
|
// FetchAll(dest interface{}) error
|
|
SyncSQL() []string
|
|
Sync() error
|
|
DropForeignKeySQL() []string
|
|
AddIndex(unique bool, cols ...string) bool
|
|
Increment(ctx context.Context, diff interface{}, target interface{}) error
|
|
Decrement(ctx context.Context, diff interface{}, target interface{}) error
|
|
|
|
GetSplitTable() *splitable.SSplitTableSpec
|
|
|
|
GetTableSpec() *sqlchemy.STableSpec
|
|
|
|
GetDBName() sqlchemy.DBName
|
|
|
|
InformUpdate(ctx context.Context, dt interface{}, oldObj *jsonutils.JSONDict)
|
|
}
|
|
|
|
type sTableSpec struct {
|
|
sqlchemy.ITableSpec
|
|
}
|
|
|
|
func newTableSpec(model interface{}, tableName string, indexField string, dateField string, maxDuration time.Duration, maxSegments int, dbName sqlchemy.DBName) ITableSpec {
|
|
var itbl sqlchemy.ITableSpec
|
|
if len(indexField) > 0 && len(dateField) > 0 {
|
|
var err error
|
|
itbl, err = splitable.NewSplitTableSpec(model, tableName, indexField, dateField, maxDuration, maxSegments, dbName)
|
|
if err != nil {
|
|
log.Errorf("NewSplitTableSpec %s %s", tableName, err)
|
|
return nil
|
|
} else {
|
|
log.Debugf("table %s maxDuration %d hour maxSegements %d", tableName, maxDuration/time.Hour, maxSegments)
|
|
}
|
|
} else if len(dbName) > 0 {
|
|
itbl = sqlchemy.NewTableSpecFromStructWithDBName(model, tableName, dbName)
|
|
} else {
|
|
itbl = sqlchemy.NewTableSpecFromStruct(model, tableName)
|
|
}
|
|
return &sTableSpec{
|
|
ITableSpec: itbl,
|
|
}
|
|
}
|
|
|
|
func newClickhouseTableSpecFromMySQL(spec ITableSpec, name string, dbName sqlchemy.DBName, extraOpts sqlchemy.TableExtraOptions) ITableSpec {
|
|
itbl := sqlchemy.NewTableSpecFromISpecWithDBName(spec.(*sTableSpec).ITableSpec, name, dbName, extraOpts)
|
|
return &sTableSpec{
|
|
ITableSpec: itbl,
|
|
}
|
|
}
|
|
|
|
func (ts *sTableSpec) GetSplitTable() *splitable.SSplitTableSpec {
|
|
sts, ok := ts.ITableSpec.(*splitable.SSplitTableSpec)
|
|
if ok {
|
|
return sts
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (ts *sTableSpec) GetDBName() sqlchemy.DBName {
|
|
sts, ok := ts.ITableSpec.(*sqlchemy.STableSpec)
|
|
if ok {
|
|
dbName := sts.DBName()
|
|
return dbName
|
|
}
|
|
return sqlchemy.DefaultDB
|
|
}
|
|
|
|
func (ts *sTableSpec) newInformerModel(dt interface{}) (*informer.ModelObject, error) {
|
|
obj, ok := dt.(IModel)
|
|
if !ok {
|
|
return nil, errors.Errorf("informer model is not IModel")
|
|
}
|
|
if obj.GetVirtualObject() == nil {
|
|
return nil, errors.Errorf("object %#v virtual object is nil", obj)
|
|
}
|
|
if obj.GetModelManager() == nil {
|
|
return nil, errors.Errorf("object %#v model manager is nil", obj)
|
|
}
|
|
jointObj, isJoint := obj.(IJointModel)
|
|
if isJoint {
|
|
mObj := JointMaster(jointObj)
|
|
if gotypes.IsNil(mObj) {
|
|
return nil, errors.Errorf("object %#v master is nil", obj)
|
|
}
|
|
sObj := JointSlave(jointObj)
|
|
if gotypes.IsNil(sObj) {
|
|
return nil, errors.Errorf("object %#v slave is nil", obj)
|
|
}
|
|
return informer.NewJointModel(jointObj, jointObj.KeywordPlural(), mObj.GetId(), sObj.GetId()), nil
|
|
}
|
|
return informer.NewModel(obj, obj.KeywordPlural(), obj.GetId()), nil
|
|
}
|
|
|
|
func (ts *sTableSpec) isMarkDeleted(dt interface{}) (bool, error) {
|
|
if vObj, ok := dt.(IVirtualModel); ok {
|
|
if vObj.GetPendingDeleted() {
|
|
return true, nil
|
|
}
|
|
}
|
|
obj, ok := dt.(IModel)
|
|
if !ok {
|
|
return false, errors.Errorf("informer model is not IModel")
|
|
}
|
|
return obj.GetDeleted(), nil
|
|
}
|
|
|
|
func (ts *sTableSpec) rejectRecordChecksumAfterInsert(model IModel) error {
|
|
obj, ok := IsModelEnableRecordChecksum(model)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
return UpdateModelChecksum(obj)
|
|
}
|
|
|
|
func (ts *sTableSpec) Insert(ctx context.Context, dt interface{}) error {
|
|
if err := ts.ITableSpec.Insert(dt); err != nil {
|
|
return err
|
|
}
|
|
ts.rejectRecordChecksumAfterInsert(dt.(IModel))
|
|
ts.inform(ctx, dt, informer.Create)
|
|
return nil
|
|
}
|
|
|
|
func (ts *sTableSpec) GetTableSpec() *sqlchemy.STableSpec {
|
|
return ts.ITableSpec.(*sqlchemy.STableSpec)
|
|
}
|
|
|
|
func (ts *sTableSpec) calculateRecordChecksum(dt interface{}) (string, error) {
|
|
return "", errors.ErrNotImplemented
|
|
}
|
|
|
|
func (ts *sTableSpec) InsertOrUpdate(ctx context.Context, dt interface{}) error {
|
|
if err := ts.ITableSpec.InsertOrUpdate(dt); err != nil {
|
|
return err
|
|
}
|
|
ts.rejectRecordChecksumAfterInsert(dt.(IModel))
|
|
ts.inform(ctx, dt, informer.Create)
|
|
return nil
|
|
}
|
|
|
|
func (ts *sTableSpec) CheckRecordChanged(dbObj IModel) error {
|
|
return ts.CheckRecordChecksumConsistent(dbObj)
|
|
}
|
|
|
|
func (ts *sTableSpec) CheckRecordChecksumConsistent(model IModel) error {
|
|
obj, ok := IsModelEnableRecordChecksum(model)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
calChecksum, err := CalculateModelChecksum(obj)
|
|
if err != nil {
|
|
return errors.Wrap(err, "CalculateModelChecksum")
|
|
}
|
|
savedChecksum := obj.GetRecordChecksum()
|
|
if calChecksum != savedChecksum {
|
|
log.Errorf("Record %s(%s) checksum changed, expected(%s) != calculated(%s)", obj.Keyword(), obj.GetId(), savedChecksum, calChecksum)
|
|
return errors.Errorf("Record %s(%s) checksum changed, expected(%s) != calculated(%s)", obj.Keyword(), obj.GetId(), savedChecksum, calChecksum)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func checksumTestNotify(ctx context.Context, action api.SAction, resType string, obj jsonutils.JSONObject) {
|
|
|
|
}
|
|
|
|
func (ts *sTableSpec) Update(ctx context.Context, dt interface{}, doUpdate func() error) (sqlchemy.UpdateDiffs, error) {
|
|
model := dt.(IModel)
|
|
dbObj, isEnableRecordChecksum := IsModelEnableRecordChecksum(model)
|
|
if isEnableRecordChecksum {
|
|
if err := ts.CheckRecordChanged(dbObj); err != nil {
|
|
log.Errorf("checkRecordChanged when update error: %s", err)
|
|
return nil, errors.Wrap(err, "checkRecordChanged when update")
|
|
}
|
|
}
|
|
|
|
oldObj := jsonutils.Marshal(dt)
|
|
diffs, err := ts.ITableSpec.Update(dt, func() error {
|
|
if err := doUpdate(); err != nil {
|
|
return err
|
|
}
|
|
if isEnableRecordChecksum {
|
|
dbObj = dt.(IRecordChecksumModel)
|
|
updateChecksum, err := CalculateModelChecksum(dbObj)
|
|
if err != nil {
|
|
return errors.Wrap(err, "CalculateModelChecksum for update")
|
|
}
|
|
dbObj.SetRecordChecksum(updateChecksum)
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if diffs == nil {
|
|
// no data to update
|
|
return nil, nil
|
|
}
|
|
isDeleted, err := ts.isMarkDeleted(dt)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "check is mark deleted")
|
|
}
|
|
if isDeleted {
|
|
ts.inform(ctx, dt, informer.Delete)
|
|
} else {
|
|
ts.informUpdate(ctx, dt, oldObj.(*jsonutils.JSONDict))
|
|
}
|
|
return diffs, nil
|
|
}
|
|
|
|
func (ts *sTableSpec) Increment(ctx context.Context, diff, target interface{}) error {
|
|
oldObj := jsonutils.Marshal(target)
|
|
err := ts.ITableSpec.Increment(diff, target)
|
|
if err != nil {
|
|
return errors.Wrap(err, "Increment")
|
|
}
|
|
ts.informUpdate(ctx, target, oldObj.(*jsonutils.JSONDict))
|
|
return nil
|
|
}
|
|
|
|
func (ts *sTableSpec) Decrement(ctx context.Context, diff, target interface{}) error {
|
|
oldObj := jsonutils.Marshal(target)
|
|
err := ts.ITableSpec.Decrement(diff, target)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ts.informUpdate(ctx, target, oldObj.(*jsonutils.JSONDict))
|
|
return nil
|
|
}
|
|
|
|
func (ts *sTableSpec) inform(ctx context.Context, dt interface{}, f func(ctx context.Context, obj *informer.ModelObject) error) {
|
|
if !informer.IsInit() {
|
|
return
|
|
}
|
|
nf := func() {
|
|
obj, err := ts.newInformerModel(dt)
|
|
if err != nil {
|
|
log.Warningf("newInformerModel error: %v", err)
|
|
debug.PrintStack()
|
|
return
|
|
}
|
|
if err := f(ctx, obj); err != nil {
|
|
if errors.Cause(err) == informer.ErrBackendNotInit {
|
|
log.V(4).Warningf("informer backend not init")
|
|
} else {
|
|
log.Errorf("call informer func error: %v", err)
|
|
}
|
|
}
|
|
}
|
|
nopanic.Run(nf)
|
|
}
|
|
|
|
func (ts *sTableSpec) informUpdate(ctx context.Context, dt interface{}, oldObj *jsonutils.JSONDict) {
|
|
if !informer.IsInit() {
|
|
return
|
|
}
|
|
nf := func() {
|
|
obj, err := ts.newInformerModel(dt)
|
|
if err != nil {
|
|
log.Warningf("newInformerModel error: %v", err)
|
|
debug.PrintStack()
|
|
return
|
|
}
|
|
if err := informer.Update(ctx, obj, oldObj); err != nil {
|
|
if errors.Cause(err) == informer.ErrBackendNotInit {
|
|
log.V(4).Warningf("informer backend not init")
|
|
} else {
|
|
log.Errorf("call informer update func error: %v", err)
|
|
}
|
|
}
|
|
}
|
|
nopanic.Run(nf)
|
|
}
|
|
|
|
func (ts *sTableSpec) InformUpdate(ctx context.Context, dt interface{}, oldObj *jsonutils.JSONDict) {
|
|
ts.informUpdate(ctx, dt, oldObj)
|
|
}
|