Skip to content

Commit 8a38327

Browse files
flarcoflarco
committed
improve col type logic, telemetry
1 parent 0793b16 commit 8a38327

5 files changed

Lines changed: 116 additions & 74 deletions

File tree

‎cmd/sling/sling.go‎

Lines changed: 46 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,18 @@ var ctx = g.NewContext(context.Background())
3232
var telemetryMap = g.M("begin_time", time.Now().UnixMicro())
3333
var telemetry = true
3434
var interrupted = false
35+
var machineID = ""
36+
var sentryOptions = sentry.ClientOptions{
37+
// Either set your DSN here or set the SENTRY_DSN environment variable.
38+
Dsn: "https://abb36e36341a4a2fa7796b6f9a0b3766@o881232.ingest.sentry.io/5835484",
39+
// Either set environment and release here or set the SENTRY_ENVIRONMENT
40+
// and SENTRY_RELEASE environment variables.
41+
Environment: lo.Ternary(core.Version == "dev", "Development", "Production"),
42+
Release: "sling@" + core.Version,
43+
// Enable printing of SDK debug messages.
44+
// Useful when getting started or trying to figure something out.
45+
Debug: false,
46+
}
3547

3648
var cliRun = &g.CliSC{
3749
Name: "run",
@@ -228,17 +240,8 @@ func init() {
228240
// cliUi.Make().Add()
229241

230242
if telemetry {
231-
sentry.Init(sentry.ClientOptions{
232-
// Either set your DSN here or set the SENTRY_DSN environment variable.
233-
Dsn: "https://abb36e36341a4a2fa7796b6f9a0b3766@o881232.ingest.sentry.io/5835484",
234-
// Either set environment and release here or set the SENTRY_ENVIRONMENT
235-
// and SENTRY_RELEASE environment variables.
236-
Environment: lo.Ternary(core.Version == "dev", "Development", "Production"),
237-
Release: "sling@" + core.Version,
238-
// Enable printing of SDK debug messages.
239-
// Useful when getting started or trying to figure something out.
240-
Debug: false,
241-
})
243+
machineID, _ = machineid.ProtectedID("sling")
244+
sentry.Init(sentryOptions)
242245
}
243246
}
244247

@@ -275,9 +278,8 @@ func Track(event string, props ...map[string]interface{}) {
275278
}
276279
}
277280

278-
id, _ := machineid.ProtectedID("sling")
279281
rsClient.Enqueue(analytics.Track{
280-
UserId: id,
282+
UserId: machineID,
281283
Event: event,
282284
Properties: properties,
283285
})
@@ -341,9 +343,8 @@ func cliInit() int {
341343
ok, err := g.CliProcess()
342344
if ok {
343345
if err != nil {
344-
sentry.CaptureException(err)
345346

346-
if g.In(g.CliObj.Name, "conns", "update") {
347+
if g.In(g.CliObj.Name, "conns", "update") || telemetryMap["error"] == nil {
347348
telemetryMap["error"] = getErrString(err)
348349

349350
eventName := g.CliObj.Name
@@ -352,6 +353,36 @@ func cliInit() int {
352353
}
353354
Track(eventName)
354355
}
356+
357+
// sentry details
358+
if telemetry {
359+
360+
evt := sentry.NewEvent()
361+
evt.Environment = sentryOptions.Environment
362+
evt.Release = sentryOptions.Release
363+
evt.Level = sentry.LevelError
364+
evt.Exception = []sentry.Exception{
365+
{
366+
Type: err.Error(),
367+
// Value: err.Error(),
368+
Stacktrace: sentry.ExtractStacktrace(err),
369+
},
370+
}
371+
372+
E, ok := err.(*g.ErrType)
373+
if ok {
374+
evt.Exception[0].Type = E.Err
375+
evt.Exception[0].Value = E.Full()
376+
}
377+
378+
sentry.ConfigureScope(func(scope *sentry.Scope) {
379+
scope.SetUser(sentry.User{ID: machineID})
380+
scope.SetTransaction(E.Err)
381+
})
382+
383+
sentry.CaptureEvent(evt)
384+
// eid := sentry.CaptureException(err)
385+
}
355386
}
356387
g.LogFatal(err)
357388
} else {

‎cmd/sling/sling_logic.go‎

Lines changed: 18 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"runtime"
1010
"sort"
1111
"strings"
12+
"time"
1213

1314
"github.com/integrii/flaggy"
1415
"github.com/samber/lo"
@@ -176,18 +177,7 @@ func processRun(c *g.CliSC) (ok bool, err error) {
176177
}
177178

178179
func runTask(cfg *sling.Config) (err error) {
179-
err = cfg.Prepare()
180-
if err != nil {
181-
return g.Error(err, "could not set task configuration")
182-
}
183-
184-
task := sling.NewTask(0, cfg)
185-
if task.Err != nil {
186-
return g.Error(task.Err)
187-
}
188-
189-
// set context
190-
task.Context = &ctx
180+
var task *sling.TaskExecution
191181

192182
// track usage
193183
defer func() {
@@ -216,15 +206,27 @@ func runTask(cfg *sling.Config) (err error) {
216206
Track("run")
217207
}()
218208

209+
err = cfg.Prepare()
210+
if err != nil {
211+
err = g.Error(err, "could not set task configuration")
212+
return
213+
}
214+
215+
task = sling.NewTask(0, cfg)
216+
if task.Err != nil {
217+
err = g.Error(task.Err)
218+
return
219+
}
220+
221+
// set context
222+
task.Context = &ctx
223+
219224
// run task
220225
err = task.Execute()
221226
if err != nil {
222227
return g.Error(err)
223228
}
224229

225-
// g.PP(task.ProcStatsStart)
226-
// g.PP(g.GetProcStats(os.Getpid()))
227-
228230
return nil
229231
}
230232

@@ -292,6 +294,7 @@ func runReplication(cfgPath string) (err error) {
292294
g.LogError(err)
293295
eG.Capture(err)
294296
}
297+
telemetryMap = g.M("begin_time", time.Now().UnixMicro()) // reset map
295298
}
296299

297300
return eG.Err()

‎core/sling/config.go‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -111,45 +111,45 @@ func (cfg *Config) SetDefault() {
111111
}
112112

113113
// set target options
114-
var targetptions TargetOptions
114+
var targetOptions TargetOptions
115115
switch cfg.TgtConn.Type.Kind() {
116116
case dbio.KindFile:
117-
targetptions = TargetFileOptionsDefault
117+
targetOptions = TargetFileOptionsDefault
118118
case dbio.KindDatabase, dbio.KindAPI, dbio.KindAirbyte:
119-
targetptions = TargetDBOptionsDefault
119+
targetOptions = TargetDBOptionsDefault
120120
default:
121-
targetptions = TargetDBOptionsDefault
121+
targetOptions = TargetDBOptionsDefault
122122
}
123123

124124
if cfg.Target.Options == nil {
125-
cfg.Target.Options = &targetptions
125+
cfg.Target.Options = &targetOptions
126126
}
127127
if cfg.Target.Options.Header == nil {
128-
cfg.Target.Options.Header = targetptions.Header
128+
cfg.Target.Options.Header = targetOptions.Header
129129
}
130130
if cfg.Target.Options.Compression == nil {
131-
cfg.Target.Options.Compression = targetptions.Compression
131+
cfg.Target.Options.Compression = targetOptions.Compression
132132
}
133133
if cfg.Target.Options.Concurrency == 0 {
134-
cfg.Target.Options.Concurrency = targetptions.Concurrency
134+
cfg.Target.Options.Concurrency = targetOptions.Concurrency
135135
}
136136
if cfg.Target.Options.FileMaxRows == 0 {
137-
cfg.Target.Options.FileMaxRows = targetptions.FileMaxRows
137+
cfg.Target.Options.FileMaxRows = targetOptions.FileMaxRows
138138
}
139139
if cfg.Target.Options.UseBulk == nil {
140-
cfg.Target.Options.UseBulk = targetptions.UseBulk
140+
cfg.Target.Options.UseBulk = targetOptions.UseBulk
141141
}
142142
if cfg.Target.Options.AdjustColumnType == nil {
143-
cfg.Target.Options.AdjustColumnType = targetptions.AdjustColumnType
143+
cfg.Target.Options.AdjustColumnType = targetOptions.AdjustColumnType
144144
}
145145
if cfg.Target.Options.DatetimeFormat == "" {
146-
cfg.Target.Options.DatetimeFormat = targetptions.DatetimeFormat
146+
cfg.Target.Options.DatetimeFormat = targetOptions.DatetimeFormat
147147
}
148148
if cfg.Target.Options.Delimiter == "" {
149-
cfg.Target.Options.Delimiter = targetptions.Delimiter
149+
cfg.Target.Options.Delimiter = targetOptions.Delimiter
150150
}
151151
if cfg.Target.Options.MaxDecimals == nil {
152-
cfg.Target.Options.MaxDecimals = targetptions.MaxDecimals
152+
cfg.Target.Options.MaxDecimals = targetOptions.MaxDecimals
153153
}
154154

155155
}

‎core/sling/elt.go‎

Lines changed: 37 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -609,8 +609,7 @@ func (t *TaskExecution) runDbToDb() (err error) {
609609
return
610610
}
611611

612-
tgtProps := g.MapToKVArr(t.Config.TgtConn.DataS())
613-
tgtConn, err := database.NewConnContext(t.Context.Ctx, t.Config.TgtConn.URL(), tgtProps...)
612+
tgtConn, err := t.getTgtDBConn()
614613
if err != nil {
615614
err = g.Error(err, "Could not initialize target connection")
616615
return
@@ -991,12 +990,16 @@ func (t *TaskExecution) WriteToDb(cfg *Config, df *iop.Dataflow, tgtConn databas
991990
sampleData := iop.NewDataset(df.Columns)
992991
sampleData.Rows = df.Buffer
993992
sampleData.SafeInference = true
993+
sampleData.InferColumnTypes()
994+
df.Columns = sampleData.Columns
995+
994996
_, err = createTableIfNotExists(tgtConn, sampleData, cfg.Target.Options.TableTmp, "")
995997
if err != nil {
996998
err = g.Error(err, "could not create temp table "+cfg.Target.Options.TableTmp)
997999
return
9981000
}
9991001
cfg.Target.TmpTableCreated = true
1002+
df.Columns = sampleData.Columns
10001003
t.AddCleanupTask(func() {
10011004
err := tgtConn.DropTable(cfg.Target.Options.TableTmp)
10021005
g.LogError(err)
@@ -1008,32 +1011,36 @@ func (t *TaskExecution) WriteToDb(cfg *Config, df *iop.Dataflow, tgtConn databas
10081011
return
10091012
}
10101013

1014+
adjustColumnType := cfg.Target.Options.AdjustColumnType != nil && *cfg.Target.Options.AdjustColumnType
1015+
10111016
// set OnSchemaChange
1012-
df.OnSchemaChange = func(i int, newType iop.ColumnType) error {
1013-
df.Context.Lock()
1014-
defer df.Context.Unlock()
1017+
if adjustColumnType {
1018+
df.OnSchemaChange = func(i int, newType iop.ColumnType) error {
1019+
df.Context.Lock()
1020+
defer df.Context.Unlock()
10151021

1016-
table, err := database.ParseTableName(cfg.Target.Options.TableTmp, tgtConn.GetType())
1017-
if err != nil {
1018-
return g.Error(err, "could not get temp table name for schema change")
1019-
}
1020-
table.Columns, err = tgtConn.GetColumns(cfg.Target.Options.TableTmp)
1021-
if err != nil {
1022-
return g.Error(err, "could not get table columns for schema change")
1023-
}
1022+
table, err := database.ParseTableName(cfg.Target.Options.TableTmp, tgtConn.GetType())
1023+
if err != nil {
1024+
return g.Error(err, "could not get temp table name for schema change")
1025+
}
1026+
table.Columns, err = tgtConn.GetColumns(cfg.Target.Options.TableTmp)
1027+
if err != nil {
1028+
return g.Error(err, "could not get table columns for schema change")
1029+
}
10241030

1025-
df.Columns[i].Type = newType
1026-
ok, err := tgtConn.OptimizeTable(&table, df.Columns)
1027-
if err != nil {
1028-
return g.Error(err, "could not change table schema")
1029-
} else if ok {
1030-
cfg.Target.columns = table.Columns
1031-
for i := range df.Columns {
1032-
df.Columns[i].Type = table.Columns[i].Type
1031+
df.Columns[i].Type = newType
1032+
ok, err := tgtConn.OptimizeTable(&table, df.Columns)
1033+
if err != nil {
1034+
return g.Error(err, "could not change table schema")
1035+
} else if ok {
1036+
cfg.Target.columns = table.Columns
1037+
for i := range df.Columns {
1038+
df.Columns[i].Type = table.Columns[i].Type
1039+
}
10331040
}
1034-
}
10351041

1036-
return nil
1042+
return nil
1043+
}
10371044
}
10381045

10391046
t.SetProgress("streaming data")
@@ -1119,9 +1126,14 @@ func (t *TaskExecution) WriteToDb(cfg *Config, df *iop.Dataflow, tgtConn databas
11191126
t.SetProgress("created table %s", targetTable)
11201127
}
11211128

1129+
table, err := database.ParseTableName(targetTable, tgtConn.GetType())
1130+
if err != nil {
1131+
return cnt, g.Error(err, "could not get table name for optimization")
1132+
}
1133+
11221134
if !created && cfg.Mode != FullRefreshMode {
11231135
if cfg.Target.Options.AddNewColumns {
1124-
ok, err := database.AddMissingColumns(tgtConn, targetTable, sample.Columns)
1136+
ok, err := database.AddMissingColumns(tgtConn, table, sample.Columns)
11251137
if err != nil {
11261138
return cnt, g.Error(err, "could not add missing columns")
11271139
} else if ok {
@@ -1132,11 +1144,7 @@ func (t *TaskExecution) WriteToDb(cfg *Config, df *iop.Dataflow, tgtConn databas
11321144
}
11331145
}
11341146

1135-
if cfg.Target.Options.AdjustColumnType != nil && *cfg.Target.Options.AdjustColumnType {
1136-
table, err := database.ParseTableName(targetTable, tgtConn.GetType())
1137-
if err != nil {
1138-
return cnt, g.Error(err, "could not get table name for optimization")
1139-
}
1147+
if adjustColumnType {
11401148

11411149
table.Columns, err = pullTargetTableColumns(t.Config, tgtConn, false)
11421150
if err != nil {

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ require (
66
github.com/c-bata/go-prompt v0.2.6
77
github.com/denisbrodbeck/machineid v1.0.1
88
github.com/dustin/go-humanize v1.0.0
9-
github.com/flarco/dbio v0.3.128
9+
github.com/flarco/dbio v0.3.129
1010
github.com/flarco/g v0.1.45
1111
github.com/getsentry/sentry-go v0.11.0
1212
github.com/integrii/flaggy v1.4.4

0 commit comments

Comments
 (0)