DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Plug new envoy to record import endpoints

Browse Source
This commit is contained in:
Tomaž Jerman
2023-10-09 14:45:01 +02:00
parent 38580e549d
commit 399f9a7cc4
9 changed files with 341 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/compose/envoy/record_datasource.go
+24 -14
View File
@@ -2,6 +2,7 @@ package envoy
import (
"context"
"strconv"
"strings"
"github.com/cortezaproject/corteza/server/pkg/dal"
@@ -13,8 +14,10 @@ type (
// RecordDatasource provides a mechanism for you to access large
// record datasets optimally
RecordDatasource struct {
mapping datasourceMapping
provider envoyx.Provider
Mapping envoyx.DatasourceMapping
Provider envoyx.Provider
currentIndex int
// Reusable buffer for reading records
rowCache map[string]string
@@ -36,11 +39,11 @@ type (
)
func (rd *RecordDatasource) SetProvider(s envoyx.Provider) bool {
if rd.mapping.SourceIdent != s.Ident() {
if rd.Mapping.SourceIdent != s.Ident() {
return false
}
rd.provider = s
rd.Provider = s
return true
}
@@ -49,27 +52,34 @@ func (rd *RecordDatasource) Next(ctx context.Context, out map[string]string) (id
rd.rowCache = make(map[string]string)
}
more, err = rd.provider.Next(ctx, rd.rowCache)
more, err = rd.Provider.Next(ctx, rd.rowCache)
if err != nil || !more {
return
}
rd.applyMapping(rd.rowCache, out)
for _, k := range rd.mapping.KeyField {
ident = append(ident, rd.rowCache[k])
if len(rd.Mapping.KeyField) == 0 {
ident = append(ident, strconv.FormatInt(int64(rd.currentIndex), 10))
} else {
for _, k := range rd.Mapping.KeyField {
ident = append(ident, rd.rowCache[k])
}
}
rd.currentIndex++
return
}
func (rd *RecordDatasource) Reset(ctx context.Context) (err error) {
return rd.provider.Reset(ctx)
rd.currentIndex = 0
return rd.Provider.Reset(ctx)
}
func (rd *RecordDatasource) applyMapping(in, out map[string]string) {
if len(rd.mapping.Mapping.m) == 0 {
if !rd.mapping.Defaultable {
if len(rd.Mapping.Mapping.Map) == 0 {
if !rd.Mapping.Defaultable {
return
}
@@ -79,7 +89,7 @@ func (rd *RecordDatasource) applyMapping(in, out map[string]string) {
return
}
if rd.mapping.Defaultable {
if rd.Mapping.Defaultable {
rd.applyMappingWithDefaults(in, out)
} else {
rd.applyMappingWoDefaults(in, out)
@@ -87,8 +97,8 @@ func (rd *RecordDatasource) applyMapping(in, out map[string]string) {
}
func (rd *RecordDatasource) applyMappingWithDefaults(in, out map[string]string) {
maps := make(map[string]mapEntry)
for k, v := range rd.mapping.Mapping.m {
maps := make(map[string]envoyx.MapEntry)
for k, v := range rd.Mapping.Mapping.Map {
maps[k] = v
}
@@ -105,7 +115,7 @@ func (rd *RecordDatasource) applyMappingWithDefaults(in, out map[string]string)
}
func (rd *RecordDatasource) applyMappingWoDefaults(in, out map[string]string) {
for _, m := range rd.mapping.Mapping.m {
for _, m := range rd.Mapping.Mapping.Map {
if m.Skip {
continue
}
server/compose/envoy/store_decode.go
+2 -2
View File
@@ -240,10 +240,10 @@ func (d StoreDecoder) decodeRecordDatasource(ctx context.Context, s store.Storer
}
ou := &RecordDatasource{
provider: &iteratorProvider{iter: iter},
Provider: &iteratorProvider{iter: iter},
refToID: make(map[string]uint64),
// @todo consider providing defaults from the outside
mapping: datasourceMapping{
Mapping: envoyx.DatasourceMapping{
KeyField: []string{"id"},
Defaultable: true,
},
server/compose/envoy/store_encode.go
+12
View File
@@ -13,6 +13,18 @@ import (
)
func (e StoreEncoder) encode(ctx context.Context, p envoyx.EncodeParams, s store.Storer, rt string, nn envoyx.NodeSet, tree envoyx.Traverser) (err error) {
dl, err := e.grabDal(p)
if err != nil {
return
}
switch rt {
case ComposeRecordDatasourceAuxType:
err = e.encodeRecordDatasources(ctx, p, s, dl, nn, tree)
if err != nil {
return
}
}
return
}
server/compose/envoy/yaml_decode.go
+2 -87
View File
@@ -12,32 +12,6 @@ import (
"gopkg.in/yaml.v3"
)
type (
mapEntry struct {
Column string
Field string
Skip bool
}
fieldMapping struct {
m map[string]mapEntry
}
datasourceMapping struct {
SourceIdent string `yaml:"source"`
KeyField []string `yaml:"key"`
References map[string]string
Scope map[string]string
// Defaultable indicates wether the mapping should keep the values where
// the ident is not explicitly mapped.
//
// When true, the value is assigned to the given identifier.
Defaultable bool `yaml:"defaultable"`
Mapping fieldMapping
}
)
const (
ComposeRecordDatasourceAuxType = "corteza::compose:record-datasource"
)
@@ -432,7 +406,7 @@ func (d *auxYamlDoc) unmarshalPagesExtendedNode(dctx documentContext, n *yaml.No
}
func (d *auxYamlDoc) unmarshalSourceExtendedNode(dctx documentContext, n *yaml.Node, meta ...*yaml.Node) (out envoyx.NodeSet, err error) {
var r datasourceMapping
var r envoyx.DatasourceMapping
// @todo we're omitting errors because there will be a bunch due to invalid
// resource field types. This might be a bit unstable as other errors may
@@ -474,7 +448,7 @@ func (d *auxYamlDoc) unmarshalSourceExtendedNode(dctx documentContext, n *yaml.N
// @todo for now we only support record datasources; extend when needed
auxN := &envoyx.Node{
Datasource: &RecordDatasource{
mapping: r,
Mapping: r,
},
ResourceType: ComposeRecordDatasourceAuxType,
@@ -486,65 +460,6 @@ func (d *auxYamlDoc) unmarshalSourceExtendedNode(dctx documentContext, n *yaml.N
return
}
// UnmarshalYAML is used to get the yaml parsed into a series of nodes so
// we can easily pass it down
func (d *fieldMapping) UnmarshalYAML(n *yaml.Node) (err error) {
d.m = make(map[string]mapEntry)
if y7s.IsSeq(n) {
err = y7s.EachSeq(n, func(n *yaml.Node) error {
a, err := d.unmarshalMappingNode(n)
d.m[a.Column] = a
return err
})
} else {
err = y7s.EachMap(n, func(k, n *yaml.Node) error {
a, err := d.unmarshalMappingNode(n)
if a.Column == "" {
err = y7s.DecodeScalar(k, "fieldMapping column", &a.Column)
if err != nil {
return err
}
}
d.m[a.Column] = a
return err
})
}
return
}
func (d *fieldMapping) unmarshalMappingNode(n *yaml.Node) (out mapEntry, err error) {
if y7s.IsKind(n, yaml.ScalarNode) {
err = y7s.DecodeScalar(n, "Column", &out.Column)
if err != nil {
return
}
err = y7s.DecodeScalar(n, "Field", &out.Field)
return
}
// @todo we're omitting errors because there will be a bunch due to invalid
// resource field types. This might be a bit unstable as other errors may
// also get ignored.
//
// A potential fix would be to firstly unmarshal into an any, check errors
// and then unmarshal into the resource while omitting errors.
n.Decode(&out)
err = y7s.EachMap(n, func(k, v *yaml.Node) error {
switch strings.ToLower(k.Value) {
case "skip":
if v.Value == "/" {
out.Skip = true
}
}
return nil
})
return
}
func (d *auxYamlDoc) procMappingRefs(in map[string]string) (out map[string]envoyx.Ref, scope envoyx.Scope) {
out = make(map[string]envoyx.Ref)
server/compose/rest/record.go
+181 -86
View File
@@ -11,6 +11,7 @@ import (
"strings"
"time"
composeEnvoy "github.com/cortezaproject/corteza/server/compose/envoy"
"github.com/cortezaproject/corteza/server/compose/rest/request"
"github.com/cortezaproject/corteza/server/compose/service"
"github.com/cortezaproject/corteza/server/compose/types"
@@ -19,9 +20,9 @@ import (
"github.com/cortezaproject/corteza/server/pkg/dal"
"github.com/cortezaproject/corteza/server/pkg/envoy"
"github.com/cortezaproject/corteza/server/pkg/envoy/csv"
ejson "github.com/cortezaproject/corteza/server/pkg/envoy/json"
"github.com/cortezaproject/corteza/server/pkg/envoy/resource"
envoyJson "github.com/cortezaproject/corteza/server/pkg/envoy/json"
estore "github.com/cortezaproject/corteza/server/pkg/envoy/store"
"github.com/cortezaproject/corteza/server/pkg/envoyx"
"github.com/cortezaproject/corteza/server/pkg/filter"
"github.com/cortezaproject/corteza/server/pkg/revisions"
"github.com/cortezaproject/corteza/server/store"
@@ -393,113 +394,207 @@ func (ctrl *Record) ImportInit(ctx context.Context, r *request.RecordImportInit)
return ctrl.importSession.Create(ctx, f, r.Upload.Filename, ct, r.NamespaceID, r.ModuleID)
}
func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) (interface{}, error) {
// @todo :')
func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) (_ interface{}, err error) {
var (
err error
ns *types.Namespace
mod *types.Module
)
// Access control.
if _, err = ctrl.module.FindByID(ctx, r.NamespaceID, r.ModuleID); err != nil {
if mod, err = ctrl.module.FindByID(ctx, r.NamespaceID, r.ModuleID); err != nil {
return nil, err
}
if ns, err = ctrl.namespace.FindByID(ctx, r.NamespaceID); err != nil {
return nil, err
}
// Check if session ok
ses, err := ctrl.importSession.FindByID(ctx, r.SessionID)
if err != nil {
return nil, err
}
ses.Fields = make(map[string]string)
err = json.Unmarshal(r.Fields, &ses.Fields)
if err != nil {
return nil, err
}
ses.OnError = r.OnError
// Errors are presented in the session
var importSession *service.RecordImportSession
err = func() (err error) {
if ses.Progress.StartedAt != nil {
return fmt.Errorf("unable to start import: import session already active")
// Check if session ok
{
importSession, err = ctrl.importSession.FindByID(ctx, r.SessionID)
if err != nil {
return
}
if importSession.Progress.StartedAt != nil {
return fmt.Errorf("unable to start import: import session already active")
}
}
sa := time.Now()
ses.Progress.StartedAt = &sa
importSession.Progress.StartedAt = &sa
// Prepare additional metadata
tpl := resource.NewComposeRecordTemplate(
strconv.FormatUint(ses.ModuleID, 10),
strconv.FormatUint(ses.NamespaceID, 10),
ses.Name,
false,
resource.MapToMappingTplSet(ses.Fields),
ses.Key,
// Some prereq
{
importSession.Fields = make(map[string]string)
err = json.Unmarshal(r.Fields, &importSession.Fields)
if err != nil {
return err
}
importSession.OnError = r.OnError
}
// Prep envoy bits
var (
envoySvc *envoyx.Service
encodeParams envoyx.EncodeParams
nodeScope envoyx.Scope
nodes envoyx.NodeSet
node *envoyx.Node
storeEncoder = composeEnvoy.StoreEncoder{}
)
{
envoySvc = envoyx.New()
// @todo add when/if needed
envoySvc.AddEncoder(envoyx.EncodeTypeStore,
storeEncoder,
)
// Shape the data
ses.Resources = append(ses.Resources, tpl)
rt := resource.ComposeRecordShaper()
ses.Resources, err = resource.Shape(ses.Resources, rt)
encodeParams = envoyx.EncodeParams{
Type: envoyx.EncodeTypeStore,
Params: map[string]any{
"storer": service.DefaultStore,
"dal": dal.Service(),
},
DeferOk: func() {
importSession.Progress.Completed++
},
DeferNok: func(err error) error {
importSession.Progress.Failed++
// Build
cfg := &estore.EncoderConfig{
// For now the identifier is ignored, so this will never occur
OnExisting: resource.Skip,
DeferOk: func() {
ses.Progress.Completed++
},
}
cfg.DeferNok = func(err error) error {
ses.Progress.Failed++
if ses.Progress.FailLog == nil {
ses.Progress.FailLog = &service.FailLog{
Errors: make(service.ErrorIndex),
}
}
if rve, is := err.(*types.RecordValueErrorSet); is {
for _, ve := range rve.Set {
for k, v := range ve.Meta {
ses.Progress.FailLog.Errors.Add(fmt.Sprintf("%s %s %v", ve.Kind, k, v))
if importSession.Progress.FailLog == nil {
importSession.Progress.FailLog = &service.FailLog{
Errors: make(service.ErrorIndex),
}
}
if rve, is := err.(*types.RecordValueErrorSet); is {
for _, ve := range rve.Set {
for k, v := range ve.Meta {
importSession.Progress.FailLog.Errors.Add(fmt.Sprintf("%s %s %v", ve.Kind, k, v))
}
}
} else {
importSession.Progress.FailLog.Errors.Add(err.Error())
}
if len(importSession.Progress.FailLog.Records) < service.IMPORT_ERROR_MAX_INDEX_COUNT {
// +1 because we indexed them with 1 before
importSession.Progress.FailLog.Records = append(importSession.Progress.FailLog.Records, int(importSession.Progress.Completed)+1)
} else {
importSession.Progress.FailLog.RecordsTruncated = true
}
if importSession.OnError == service.IMPORT_ON_ERROR_SKIP {
return nil
}
return err
},
}
nodeScope = envoyx.Scope{
ResourceType: types.NamespaceResourceType,
Identifiers: envoyx.MakeIdentifiers(importSession.NamespaceID),
}
fieldMapping := map[string]envoyx.MapEntry{}
for c, f := range importSession.Fields {
fieldMapping[c] = envoyx.MapEntry{
Column: c,
Field: f,
}
} else {
ses.Progress.FailLog.Errors.Add(err.Error())
}
if len(ses.Progress.FailLog.Records) < service.IMPORT_ERROR_MAX_INDEX_COUNT {
// +1 because we indexed them with 1 before
ses.Progress.FailLog.Records = append(ses.Progress.FailLog.Records, int(ses.Progress.Completed)+1)
} else {
ses.Progress.FailLog.RecordsTruncated = true
nsNode := &envoyx.Node{
Resource: ns,
ResourceType: types.NamespaceResourceType,
Identifiers: envoyx.MakeIdentifiers(ns.Slug, ns.ID),
Scope: nodeScope,
Placeholder: true,
}
if ses.OnError == service.IMPORT_ON_ERROR_SKIP {
return nil
modNode := &envoyx.Node{
Resource: mod,
ResourceType: types.ModuleResourceType,
Identifiers: envoyx.MakeIdentifiers(mod.Handle, mod.ID),
Scope: nodeScope,
References: map[string]envoyx.Ref{
"NamespaceID": {
ResourceType: types.NamespaceResourceType,
Identifiers: nsNode.Identifiers,
Scope: nsNode.Scope,
},
},
Placeholder: true,
}
return err
}
se := estore.NewStoreEncoder(service.DefaultStore, dal.Service(), cfg)
bld := envoy.NewBuilder(se)
g, err := bld.Build(ctx, ses.Resources...)
if err != nil {
return err
node = &envoyx.Node{
Datasource: &composeEnvoy.RecordDatasource{
Mapping: envoyx.DatasourceMapping{
SourceIdent: importSession.Name,
References: map[string]string{},
Scope: map[string]string{},
Defaultable: false,
Mapping: envoyx.FieldMapping{
Map: fieldMapping,
},
},
},
ResourceType: composeEnvoy.ComposeRecordDatasourceAuxType,
Identifiers: envoyx.MakeIdentifiers(importSession.Name),
References: map[string]envoyx.Ref{
"ModuleID": {
ResourceType: types.ModuleResourceType,
Identifiers: envoyx.MakeIdentifiers(importSession.ModuleID),
Scope: nodeScope,
},
"NamespaceID": {
ResourceType: types.NamespaceResourceType,
Identifiers: envoyx.MakeIdentifiers(importSession.NamespaceID),
Scope: nodeScope,
},
},
Scope: nodeScope,
}
nodes = envoyx.NodeSet{nsNode, modNode, node}
}
// Encode
err = envoy.Encode(ctx, g, se)
now := time.Now()
ses.Progress.FinishedAt = &now
if err != nil {
ses.Progress.FailReason = err.Error()
return err
}
// encoding stuff
var (
depGraph *envoyx.DepGraph
)
{
depGraph, err = envoySvc.Bake(ctx, encodeParams, importSession.Providers, nodes...)
if err != nil {
return
}
return
{
// @todo this is temporary because the service's logic is a bit flawed for this case
err = storeEncoder.Prepare(ctx, encodeParams, composeEnvoy.ComposeRecordDatasourceAuxType, envoyx.NodeSet{node})
if err != nil {
return
}
err = storeEncoder.Encode(ctx, encodeParams, composeEnvoy.ComposeRecordDatasourceAuxType, envoyx.NodeSet{node}, depGraph)
// @note err is handled lower down; bare with
}
// err = envoySvc.Encode(ctx, encodeParams, depGraph)
now := time.Now()
importSession.Progress.FinishedAt = &now
if err != nil {
importSession.Progress.FailReason = err.Error()
return
}
return
}
}()
return ses, ctrl.record.RecordImport(ctx, err)
return importSession, ctrl.record.RecordImport(ctx, err)
}
func (ctrl *Record) ImportProgress(ctx context.Context, r *request.RecordImportProgress) (interface{}, error) {
@@ -559,7 +654,7 @@ func (ctrl *Record) Export(ctx context.Context, r *request.RecordExport) (interf
switch strings.ToLower(r.Ext) {
case "json", "jsonl", "ldjson", "ndjson":
contentType = "application/jsonl"
encoder = ejson.NewBulkRecordEncoder(&ejson.EncoderConfig{
encoder = envoyJson.NewBulkRecordEncoder(&envoyJson.EncoderConfig{
Fields: fx,
Timezone: r.Timezone,
})
server/compose/service/import_session.go
+25 -37
View File
@@ -9,23 +9,29 @@ import (
"time"
"github.com/cortezaproject/corteza/server/pkg/auth"
"github.com/cortezaproject/corteza/server/pkg/envoy"
"github.com/cortezaproject/corteza/server/pkg/envoy/csv"
"github.com/cortezaproject/corteza/server/pkg/envoy/json"
"github.com/cortezaproject/corteza/server/pkg/envoy/resource"
"github.com/cortezaproject/corteza/server/pkg/envoyx"
envoyCsv "github.com/cortezaproject/corteza/server/pkg/envoyx/csv"
envoyJson "github.com/cortezaproject/corteza/server/pkg/envoyx/json"
)
type (
recordSet []*recordImportSession
recordSet []*RecordImportSession
importSession struct {
l sync.Mutex
records recordSet
}
countableProvider interface {
envoyx.Provider
Count() uint64
Fields() []string
}
ImportSessionService interface {
Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (*recordImportSession, error)
FindByID(ctx context.Context, sessionID uint64) (*recordImportSession, error)
Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (*RecordImportSession, error)
FindByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error)
DeleteByID(ctx context.Context, sessionID uint64) error
}
)
@@ -46,12 +52,12 @@ func (svc *importSession) indexOf(userID, sessionID uint64) int {
return -1
}
func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (*recordImportSession, error) {
func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, contentType string, namespaceID, moduleID uint64) (_ *RecordImportSession, err error) {
svc.l.Lock()
defer svc.l.Unlock()
// Prepare the session
sh := &recordImportSession{
sh := &RecordImportSession{
Name: name,
SessionID: nextID(),
UserID: auth.GetIdentityFromContext(ctx).Identity(),
@@ -66,49 +72,31 @@ func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, con
UpdatedAt: time.Now(),
}
// Decoders; We only need to do csv & yaml here
cd := csv.Decoder()
jd := json.Decoder()
// This will really be at most 1
var err error
do := &envoy.DecoderOpts{
Name: name,
Path: "",
}
sh.Resources, err = func() ([]resource.Interface, error) {
if cd.CanDecodeFile(f) || cd.CanDecodeMime(contentType) {
rd, err := func() (pp countableProvider, err error) {
if envoyCsv.CanDecodeFile(f) || envoyCsv.CanDecodeMime(contentType) {
f.Seek(0, 0)
return cd.Decode(ctx, f, do)
return envoyCsv.Decoder(f, name)
}
f.Seek(0, 0)
if jd.CanDecodeFile(f) || jd.CanDecodeMime(contentType) {
if envoyJson.CanDecodeFile(f) || envoyJson.CanDecodeMime(contentType) {
f.Seek(0, 0)
return jd.Decode(ctx, f, do)
return envoyJson.Decoder(f, name)
}
return nil, fmt.Errorf("compose.service.RecordImportFormatNotSupported")
}()
if err != nil {
return nil, err
return
}
// Get some metadata
n, ok := (sh.Resources[0]).(*resource.ResourceDataset)
if !ok {
// @todo move this logic to service and use action/error pattern
return nil, fmt.Errorf("compose.service.RecordImportFormatNotSupported")
}
sh.Providers = append(sh.Providers, rd)
prepKey := func(k string) string {
return strings.TrimSpace(strings.ToLower(k))
}
sh.Progress.EntryCount = n.P.Count()
for _, f := range n.P.Fields() {
sh.Progress.EntryCount = rd.Count()
for _, f := range rd.Fields() {
sh.Fields[f] = ""
// @todo improve this bit
@@ -123,7 +111,7 @@ func (svc *importSession) Create(ctx context.Context, f io.ReadSeeker, name, con
return sh, nil
}
func (svc *importSession) FindByID(ctx context.Context, sessionID uint64) (*recordImportSession, error) {
func (svc *importSession) FindByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) {
svc.l.Lock()
defer svc.l.Unlock()
server/compose/service/record.go
+3 -1
View File
@@ -10,6 +10,7 @@ import (
"strings"
"time"
"github.com/cortezaproject/corteza/server/pkg/envoyx"
"github.com/cortezaproject/corteza/server/pkg/filter"
"github.com/cortezaproject/corteza/server/pkg/revisions"
"github.com/spf13/cast"
@@ -153,7 +154,7 @@ type (
Country() string
}
recordImportSession struct {
RecordImportSession struct {
Name string `json:"-"`
SessionID uint64 `json:"sessionID,string"`
UserID uint64 `json:"userID,string"`
@@ -169,6 +170,7 @@ type (
UpdatedAt time.Time `json:"updatedAt"`
Resources []resource.Interface `json:"-"`
Providers []envoyx.Provider `json:"-"`
}
RecordImportProgress struct {
server/pkg/envoyx/datasource.go
+91 -1
View File
@@ -1,6 +1,12 @@
package envoyx
import "context"
import (
"context"
"strings"
"github.com/cortezaproject/corteza/server/pkg/y7s"
"gopkg.in/yaml.v3"
)
type (
Provider interface {
@@ -15,6 +21,31 @@ type (
Reset(ctx context.Context) error
SetProvider(Provider) bool
}
MapEntry struct {
Column string
Field string
Skip bool
}
FieldMapping struct {
// @note This had to be like so to simplify decoding
Map map[string]MapEntry
}
DatasourceMapping struct {
SourceIdent string `yaml:"source"`
KeyField []string `yaml:"key"`
References map[string]string
Scope map[string]string
// Defaultable indicates wether the mapping should keep the values where
// the ident is not explicitly mapped.
//
// When true, the value is assigned to the given identifier.
Defaultable bool `yaml:"defaultable"`
Mapping FieldMapping
}
)
func SetDecoderSources(nn NodeSet, dd ...Provider) {
@@ -30,3 +61,62 @@ func SetDecoderSources(nn NodeSet, dd ...Provider) {
}
}
}
// UnmarshalYAML is used to get the yaml parsed into a series of nodes so
// we can easily pass it down
func (d *FieldMapping) UnmarshalYAML(n *yaml.Node) (err error) {
d.Map = make(map[string]MapEntry)
if y7s.IsSeq(n) {
err = y7s.EachSeq(n, func(n *yaml.Node) error {
a, err := d.unmarshalMappingNode(n)
d.Map[a.Column] = a
return err
})
} else {
err = y7s.EachMap(n, func(k, n *yaml.Node) error {
a, err := d.unmarshalMappingNode(n)
if a.Column == "" {
err = y7s.DecodeScalar(k, "fieldMapping column", &a.Column)
if err != nil {
return err
}
}
d.Map[a.Column] = a
return err
})
}
return
}
func (d *FieldMapping) unmarshalMappingNode(n *yaml.Node) (out MapEntry, err error) {
if y7s.IsKind(n, yaml.ScalarNode) {
err = y7s.DecodeScalar(n, "Column", &out.Column)
if err != nil {
return
}
err = y7s.DecodeScalar(n, "Field", &out.Field)
return
}
// @todo we're omitting errors because there will be a bunch due to invalid
// resource field types. This might be a bit unstable as other errors may
// also get ignored.
//
// A potential fix would be to firstly unmarshal into an any, check errors
// and then unmarshal into the resource while omitting errors.
n.Decode(&out)
err = y7s.EachMap(n, func(k, v *yaml.Node) error {
switch strings.ToLower(k.Value) {
case "skip":
if v.Value == "/" {
out.Skip = true
}
}
return nil
})
return
}
server/pkg/envoyx/envoy.go
+1
View File
@@ -87,6 +87,7 @@ type (
Encoder EncoderConfig
// @note these are only used by records since v1 did just that
// @todo make these more expanded
DeferOk func()
DeferNok func(error) error
Defer func()