mirror of https://github.com/hashicorp/boundary
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
490 lines
16 KiB
490 lines
16 KiB
// Copyright (c) HashiCorp, Inc.
|
|
// SPDX-License-Identifier: BUSL-1.1
|
|
|
|
package oplog
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
|
|
"github.com/golang/protobuf/ptypes/wrappers"
|
|
"github.com/hashicorp/boundary/internal/errors"
|
|
"github.com/hashicorp/boundary/internal/oplog/store"
|
|
"github.com/hashicorp/boundary/internal/util"
|
|
"github.com/hashicorp/go-dbw"
|
|
wrapping "github.com/hashicorp/go-kms-wrapping/v2"
|
|
"github.com/hashicorp/go-kms-wrapping/v2/extras/structwrapping"
|
|
"google.golang.org/protobuf/proto"
|
|
"google.golang.org/protobuf/types/known/structpb"
|
|
)
|
|
|
|
// Version of oplog entries (among other things, it's used to manage upgrade
|
|
// compatibility when replicating)
|
|
//
|
|
// v1: initial version
|
|
// v2: adds the new Message.Opts
|
|
const Version = "v2"
|
|
|
|
// Message wraps a proto.Message with some other bits like operation type,
|
|
// paths and options.
|
|
type Message struct {
|
|
proto.Message
|
|
TypeName string
|
|
OpType OpType
|
|
FieldMaskPaths []string
|
|
SetToNullPaths []string
|
|
Opts []dbw.Option
|
|
}
|
|
|
|
// Entry represents an oplog entry
|
|
type Entry struct {
|
|
*store.Entry
|
|
Wrapper wrapping.Wrapper `gorm:"-"`
|
|
Ticketer Ticketer `gorm:"-"`
|
|
}
|
|
|
|
// Metadata provides meta information about the Entry
|
|
type Metadata map[string][]string
|
|
|
|
// NewEntry creates a new Entry
|
|
func NewEntry(ctx context.Context, aggregateName string, metadata Metadata, wrapper wrapping.Wrapper, ticketer Ticketer) (*Entry, error) {
|
|
const op = "oplog.NewEntry"
|
|
if util.IsNil(wrapper) {
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, "nil wrapper")
|
|
}
|
|
keyId, err := wrapper.KeyId(ctx)
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
entry := Entry{
|
|
Entry: &store.Entry{
|
|
AggregateName: aggregateName,
|
|
Version: Version,
|
|
KeyId: keyId,
|
|
},
|
|
Wrapper: wrapper,
|
|
Ticketer: ticketer,
|
|
}
|
|
if len(metadata) > 0 {
|
|
storeMD := []*store.Metadata{}
|
|
for k, v := range metadata {
|
|
if len(v) > 0 {
|
|
for _, vv := range v {
|
|
storeMD = append(storeMD, &store.Metadata{Key: k, Value: vv})
|
|
}
|
|
continue
|
|
}
|
|
// this metadata just has a key with no values
|
|
storeMD = append(storeMD, &store.Metadata{Key: k})
|
|
}
|
|
entry.Metadata = storeMD
|
|
}
|
|
if err := entry.validate(ctx); err != nil {
|
|
return nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
return &entry, nil
|
|
}
|
|
|
|
func (e *Entry) validate(ctx context.Context) error {
|
|
const op = "oplog.(Entry).validate"
|
|
if util.IsNil(e.Wrapper) {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "nil wrapper")
|
|
}
|
|
if util.IsNil(e.Ticketer) {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "nil ticketer")
|
|
}
|
|
if e.Entry == nil {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "nil entry")
|
|
}
|
|
if e.Entry.Version == "" {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "missing entry version")
|
|
}
|
|
if e.Entry.AggregateName == "" {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "missing entry aggregate name")
|
|
}
|
|
if e.Entry.KeyId == "" {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "missing entry key id")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// UnmarshalData the data attribute from []byte (treated as a FIFO QueueBuffer) to a []proto.Message
|
|
func (e *Entry) UnmarshalData(ctx context.Context, types *TypeCatalog) ([]Message, error) {
|
|
const op = "oplog.(Entry).UnmarshalData"
|
|
if types == nil {
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, "nil type catalog")
|
|
}
|
|
if len(e.Data) == 0 {
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing data")
|
|
}
|
|
msgs := []Message{}
|
|
queue := Queue{
|
|
Buffer: *bytes.NewBuffer(e.Data),
|
|
Catalog: types,
|
|
}
|
|
for {
|
|
item, err := queue.remove(ctx)
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error removing item from queue"))
|
|
}
|
|
name, err := types.GetTypeName(ctx, item.msg)
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
dbwOpts, err := convertToDbwOpts(ctx, item.operationOptions)
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
msgs = append(msgs, Message{
|
|
Message: item.msg,
|
|
TypeName: name,
|
|
OpType: item.opType,
|
|
FieldMaskPaths: item.fieldMask,
|
|
SetToNullPaths: item.setToNullPaths,
|
|
Opts: dbwOpts,
|
|
})
|
|
}
|
|
return msgs, nil
|
|
}
|
|
|
|
func convertToDbwOpts(ctx context.Context, opts *OperationOptions) ([]dbw.Option, error) {
|
|
const op = "oplog.convertToDbwOpts"
|
|
if opts == nil {
|
|
return []dbw.Option{}, nil
|
|
}
|
|
dbwOpts := []dbw.Option{
|
|
dbw.WithSkipVetForWrite(opts.GetWithSkipVetForWrite()),
|
|
}
|
|
if opts.GetWithVersion() != nil {
|
|
dbwOpts = append(dbwOpts, dbw.WithVersion(&opts.GetWithVersion().Value))
|
|
}
|
|
if opts.GetWithOnConflict() != nil {
|
|
c := &dbw.OnConflict{}
|
|
|
|
switch targetType := opts.GetWithOnConflict().GetTarget().(type) {
|
|
case *WithOnConflict_Columns:
|
|
c.Target = dbw.Columns(opts.GetWithOnConflict().GetColumns().GetNames())
|
|
case *WithOnConflict_Constraint:
|
|
c.Target = dbw.Constraint(opts.GetWithOnConflict().GetConstraint())
|
|
default:
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("not a supported target type: %T", targetType))
|
|
}
|
|
|
|
switch actionType := opts.GetWithOnConflict().GetAction().(type) {
|
|
case *WithOnConflict_DoNothing:
|
|
c.Action = dbw.DoNothing(opts.GetWithOnConflict().GetDoNothing())
|
|
case *WithOnConflict_UpdateAll:
|
|
c.Action = dbw.UpdateAll(opts.GetWithOnConflict().GetUpdateAll())
|
|
case *WithOnConflict_ColumnValues:
|
|
cvAction := make([]dbw.ColumnValue, 0, len(actionType.ColumnValues.GetValues()))
|
|
for _, cv := range actionType.ColumnValues.GetValues() {
|
|
newColVal := dbw.ColumnValue{
|
|
Column: cv.GetName(),
|
|
}
|
|
switch pbVal := cv.Value.(type) {
|
|
case *ColumnValue_Raw:
|
|
newColVal.Value = pbVal.Raw.AsInterface()
|
|
case *ColumnValue_ExprValue:
|
|
expr := dbw.ExprValue{
|
|
Sql: pbVal.ExprValue.GetSql(),
|
|
}
|
|
args := make([]any, 0, len(pbVal.ExprValue.GetArgs()))
|
|
for _, a := range pbVal.ExprValue.GetArgs() {
|
|
args = append(args, a.AsInterface())
|
|
}
|
|
if len(args) > 0 {
|
|
expr.Vars = args
|
|
}
|
|
newColVal.Value = expr
|
|
case *ColumnValue_Column:
|
|
newColVal.Value = dbw.Column{
|
|
Name: pbVal.Column.GetName(),
|
|
Table: pbVal.Column.GetTable(),
|
|
}
|
|
default:
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("not a supported column value type: %T", pbVal))
|
|
}
|
|
cvAction = append(cvAction, newColVal)
|
|
}
|
|
c.Action = cvAction
|
|
default:
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("not a supported action type: %T", actionType))
|
|
}
|
|
dbwOpts = append(dbwOpts, dbw.WithOnConflict(c))
|
|
}
|
|
if opts.GetWithWhereClause() != "" {
|
|
sql := opts.GetWithWhereClause()
|
|
args := make([]any, 0, len(opts.GetWithWhereClauseArgs()))
|
|
for _, a := range opts.GetWithWhereClauseArgs() {
|
|
args = append(args, a.AsInterface())
|
|
}
|
|
dbwOpts = append(dbwOpts, dbw.WithWhere(sql, args...))
|
|
|
|
}
|
|
return dbwOpts, nil
|
|
}
|
|
|
|
// convertFromDbwOpts converts dbw options to an OperationalOptions for an Entry
|
|
func convertFromDbwOpts(ctx context.Context, opts dbw.Options) (*OperationOptions, error) {
|
|
const op = "oplog.convertFromDbwOptions"
|
|
pbOpts := &OperationOptions{
|
|
WithSkipVetForWrite: opts.WithSkipVetForWrite,
|
|
WithWhereClause: opts.WithWhereClause,
|
|
}
|
|
if opts.WithVersion != nil {
|
|
pbOpts.WithVersion = &wrappers.UInt32Value{Value: *opts.WithVersion}
|
|
}
|
|
|
|
clauseValues := make([]*structpb.Value, 0, len(opts.WithWhereClauseArgs))
|
|
for _, arg := range opts.WithWhereClauseArgs {
|
|
v, err := structpb.NewValue(arg)
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
clauseValues = append(clauseValues, v)
|
|
}
|
|
pbOpts.WithWhereClauseArgs = clauseValues
|
|
|
|
if opts.WithOnConflict != nil {
|
|
c := &WithOnConflict{}
|
|
switch target := opts.WithOnConflict.Target.(type) {
|
|
case dbw.Constraint:
|
|
c.Target = &WithOnConflict_Constraint{string(target)}
|
|
case dbw.Columns:
|
|
c.Target = &WithOnConflict_Columns{
|
|
&Columns{
|
|
Names: []string(target),
|
|
},
|
|
}
|
|
default:
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("not a supported target type: %T", target))
|
|
}
|
|
switch action := opts.WithOnConflict.Action.(type) {
|
|
case dbw.DoNothing:
|
|
c.Action = &WithOnConflict_DoNothing{bool(action)}
|
|
case dbw.UpdateAll:
|
|
c.Action = &WithOnConflict_UpdateAll{bool(action)}
|
|
case []dbw.ColumnValue:
|
|
colVals := make([]*ColumnValue, 0, len(action))
|
|
for _, cv := range action {
|
|
pbColVal := &ColumnValue{
|
|
Name: cv.Column,
|
|
}
|
|
switch cvVal := cv.Value.(type) {
|
|
case dbw.ExprValue:
|
|
args := make([]*structpb.Value, 0, len(cvVal.Vars))
|
|
for _, vv := range cvVal.Vars {
|
|
pbVar, err := structpb.NewValue(vv)
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to create expr arg"))
|
|
}
|
|
args = append(args, pbVar)
|
|
}
|
|
pbColVal.Value = &ColumnValue_ExprValue{&ExprValue{
|
|
Sql: cvVal.Sql,
|
|
Args: args,
|
|
}}
|
|
case dbw.Column:
|
|
pbColVal.Value = &ColumnValue_Column{
|
|
&Column{
|
|
Name: cvVal.Name,
|
|
Table: cvVal.Table,
|
|
},
|
|
}
|
|
default:
|
|
exprVal, err := structpb.NewValue(cv.Value)
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to create column value"))
|
|
}
|
|
pbColVal.Value = &ColumnValue_Raw{exprVal}
|
|
}
|
|
colVals = append(colVals, pbColVal)
|
|
}
|
|
c.Action = &WithOnConflict_ColumnValues{ColumnValues: &ColumnValues{Values: colVals}}
|
|
default:
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("not a valid supported action: %T", action))
|
|
}
|
|
|
|
pbOpts.WithOnConflict = c
|
|
}
|
|
return pbOpts, nil
|
|
}
|
|
|
|
// WriteEntryWith the []proto.Message marshaled into the entry data as a FIFO QueueBuffer
|
|
// if Wrapper != nil then the data is authentication encrypted
|
|
func (e *Entry) WriteEntryWith(ctx context.Context, tx *Writer, ticket *store.Ticket, msgs ...*Message) error {
|
|
const op = "oplog.(Entry).WriteEntryWith"
|
|
if tx == nil {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "nil writer")
|
|
}
|
|
if err := e.validate(ctx); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if ticket == nil {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "nil ticket")
|
|
}
|
|
if ticket.Version == 0 {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "missing ticket version")
|
|
}
|
|
queue := Queue{}
|
|
for _, m := range msgs {
|
|
if m == nil {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "nil message")
|
|
}
|
|
if err := queue.add(ctx, m.Message, m.TypeName, m.OpType, WithOperationOptions(m.Opts...), WithFieldMaskPaths(m.FieldMaskPaths), WithSetToNullPaths(m.SetToNullPaths)); err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("error adding message to queue"))
|
|
}
|
|
}
|
|
e.Data = append(e.Data, queue.Bytes()...)
|
|
|
|
if e.Wrapper != nil {
|
|
if err := e.encryptData(ctx); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
rw := dbw.New(tx.DB)
|
|
if err := rw.Create(ctx, e); err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("error writing data to storage"))
|
|
}
|
|
if err := e.Ticketer.Redeem(ctx, ticket); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Write the entry as is with whatever it has for e.Data marshaled into a FIFO QueueBuffer
|
|
// If Wrapper != nil then the data is authentication encrypted
|
|
func (e *Entry) Write(ctx context.Context, tx *Writer, ticket *store.Ticket) error {
|
|
const op = "oplog.(Entry).Write"
|
|
if err := e.validate(ctx); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if ticket == nil {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "nil ticket")
|
|
}
|
|
if ticket.Version == 0 {
|
|
return errors.New(ctx, errors.InvalidParameter, op, "missing ticket version")
|
|
}
|
|
if e.Wrapper != nil {
|
|
if err := e.encryptData(ctx); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
rw := dbw.New(tx.DB)
|
|
if err := rw.Create(ctx, e); err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("error writing data to storage"))
|
|
}
|
|
if err := e.Ticketer.Redeem(ctx, ticket); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// encryptData the entry's data using its Wrapper (wrapping.Wrapper)
|
|
func (e *Entry) encryptData(ctx context.Context) error {
|
|
const op = "oplog.(Entry).EncryptData"
|
|
// structwrapping doesn't support embedding, so we'll pass in the store.Entry directly
|
|
if err := structwrapping.WrapStruct(ctx, e.Wrapper, e.Entry, nil); err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithCode(errors.Encrypt))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DecryptData will decrypt the entry's data using its Wrapper (wrapping.Wrapper)
|
|
func (e *Entry) DecryptData(ctx context.Context) error {
|
|
const op = "oplog.(Entry).DecryptData"
|
|
// structwrapping doesn't support embedding, so we'll pass in the store.Entry directly
|
|
if err := structwrapping.UnwrapStruct(ctx, e.Wrapper, e.Entry, nil); err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithCode(errors.Decrypt))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Replay provides the ability to replay an entry. you must initialize any new tables ending with the tableSuffix before
|
|
// calling Replay, otherwise you'll get "a table doesn't exist" error.
|
|
func (e *Entry) Replay(ctx context.Context, tx *Writer, types *TypeCatalog, tableSuffix string) error {
|
|
const op = "oplog.(Entry).Replay"
|
|
msgs, err := e.UnmarshalData(ctx, types)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
for _, m := range msgs {
|
|
em, ok := m.Message.(ReplayableMessage)
|
|
if !ok {
|
|
return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("%T is not a replayable message", m.Message))
|
|
}
|
|
origTableName := em.TableName()
|
|
defer em.SetTableName(origTableName)
|
|
|
|
/*
|
|
how replay will be implemented for snapshots is still very much under discussion.
|
|
when we go to implement snapshots we may very well need to refactor
|
|
this create table choice... there are many issues with doing the
|
|
"create" in this manner:
|
|
* the perms needed to create a table and possible security issues
|
|
* the fk references would be to the original tables, not the new
|
|
replay tables.
|
|
* we need a way to create a replay table that contains the existing
|
|
named constraints. Currently, on conflicts using a constraint
|
|
target won't work.
|
|
It may be a better choice to just create separate schemas for replay
|
|
named blue and green since we need at min of two replay tables
|
|
definitions. if we went with separate schemas they could be create
|
|
with a boundary cli cmd that had appropriate privs (reducing
|
|
security issues) and the separate schemas wouldn't have the fk
|
|
reference issues mentioned above.
|
|
|
|
|
|
*/
|
|
replayTable := origTableName + tableSuffix
|
|
hasTable, err := tx.hasTable(ctx, replayTable)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if !hasTable {
|
|
if err := tx.createTableLike(ctx, origTableName, replayTable); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
rw := dbw.New(tx.DB)
|
|
|
|
m.Opts = append(m.Opts, dbw.WithTable(replayTable))
|
|
switch m.OpType {
|
|
case OpType_OP_TYPE_CREATE:
|
|
if err := rw.Create(ctx, m.Message, m.Opts...); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
case OpType_OP_TYPE_CREATE_ITEMS:
|
|
// TODO: jimlambrt 12/2021 -> while this will work for
|
|
// CreateItems(...) it's hardly efficient. We'll need to refactor
|
|
// oplog quite a bit to support a multi-message operation.
|
|
if err := rw.CreateItems(ctx, []any{m.Message}, m.Opts...); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
case OpType_OP_TYPE_UPDATE:
|
|
if _, err := rw.Update(ctx, m.Message, m.FieldMaskPaths, m.SetToNullPaths, m.Opts...); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
case OpType_OP_TYPE_DELETE:
|
|
if _, err := rw.Delete(ctx, m.Message, m.Opts...); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
case OpType_OP_TYPE_DELETE_ITEMS:
|
|
// TODO: jimlambrt 12/2021 -> while this will work for
|
|
// DeleteItems(...) it's hardly efficient. We'll need to refactor
|
|
// oplog quite a bit to support a multi-message operation.
|
|
if _, err := rw.DeleteItems(ctx, []any{m.Message}, m.Opts...); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
default:
|
|
return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("invalid operation %T", m.OpType))
|
|
}
|
|
}
|
|
return nil
|
|
}
|