Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
886008a
commit 3fac67e
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package transformer | ||
|
|
||
| import ( | ||
| "github.com/rudderlabs/rudder-server/warehouse/internal/model" | ||
| "github.com/rudderlabs/rudder-server/warehouse/transformer/internal/utils" | ||
| whutils "github.com/rudderlabs/rudder-server/warehouse/utils" | ||
| ) | ||
|
|
||
| func dataTypeFor(destType, key string, val any, isJSONKey bool) string { | ||
| if typeName := primitiveType(val); typeName != "" { | ||
| return typeName | ||
| } | ||
| if strVal, ok := val.(string); ok && utils.ValidTimestamp(strVal) { | ||
| return model.DateTimeDataType | ||
| } | ||
| if override := dataTypeOverride(destType, key, val, isJSONKey); override != "" { | ||
| return override | ||
| } | ||
| return model.StringDataType | ||
| } | ||
|
|
||
| func primitiveType(val any) string { | ||
| switch v := val.(type) { | ||
| case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64: | ||
| return model.IntDataType | ||
| case float64: | ||
| return getFloatType(v) | ||
| case float32: | ||
| return getFloatType(float64(v)) | ||
| case bool: | ||
| return model.BooleanDataType | ||
| default: | ||
| return "" | ||
| } | ||
| } | ||
|
|
||
| func getFloatType(v float64) string { | ||
| if v == float64(int64(v)) { | ||
| return model.IntDataType | ||
| } | ||
| return model.FloatDataType | ||
| } | ||
|
|
||
| func dataTypeOverride(destType, key string, val any, isJSONKey bool) string { | ||
| switch destType { | ||
| case whutils.POSTGRES: | ||
| return overrideForPostgres(key, isJSONKey) | ||
| case whutils.SNOWFLAKE, whutils.SnowpipeStreaming: | ||
| return overrideForSnowflake(key, isJSONKey) | ||
| case whutils.RS: | ||
| return overrideForRedshift(val, isJSONKey) | ||
| default: | ||
| return "" | ||
| } | ||
| } | ||
|
|
||
| func overrideForPostgres(key string, isJSONKey bool) string { | ||
| if key == violationErrors || isJSONKey { | ||
| return model.JSONDataType | ||
| } | ||
| return model.StringDataType | ||
| } | ||
|
|
||
| func overrideForSnowflake(key string, isJSONKey bool) string { | ||
| if key == violationErrors || isJSONKey { | ||
| return model.JSONDataType | ||
| } | ||
| return model.StringDataType | ||
| } | ||
|
|
||
| func overrideForRedshift(val any, isJSONKey bool) string { | ||
| if isJSONKey { | ||
| return model.JSONDataType | ||
| } | ||
| if val == nil { | ||
| return model.StringDataType | ||
| } | ||
| if jsonVal, _ := json.Marshal(val); len(jsonVal) > redshiftStringLimit { | ||
| return model.TextDataType | ||
| } | ||
| return model.StringDataType | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| package transformer | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "sync" | ||
|
|
||
| "github.com/google/uuid" | ||
| "github.com/samber/lo" | ||
|
|
||
| "github.com/rudderlabs/rudder-go-kit/config" | ||
| "github.com/rudderlabs/rudder-go-kit/logger" | ||
| "github.com/rudderlabs/rudder-go-kit/stringify" | ||
|
|
||
| ptrans "github.com/rudderlabs/rudder-server/processor/transformer" | ||
| "github.com/rudderlabs/rudder-server/utils/misc" | ||
| "github.com/rudderlabs/rudder-server/utils/types" | ||
| ) | ||
|
|
||
| type DebugLogger struct { | ||
| logger logger.Logger | ||
| maxLoggedEvents config.ValueLoader[int] | ||
| eventLogMutex sync.Mutex | ||
| currentLogFileName string | ||
| loggedEvents int64 | ||
| } | ||
|
|
||
| func NewDebugLogger(conf *config.Config, logger logger.Logger) *DebugLogger { | ||
| logFileName := generateLogFileName() | ||
|
|
||
| return &DebugLogger{ | ||
| logger: logger.Child("debugLogger").With("currentLogFileName", logFileName), | ||
| maxLoggedEvents: conf.GetReloadableIntVar(10000, 1, "Processor.maxLoggedEvents"), | ||
| currentLogFileName: logFileName, | ||
| } | ||
| } | ||
|
|
||
| func generateLogFileName() string { | ||
| return fmt.Sprintf("warehouse_transformations_debug_%s.log", uuid.NewString()) | ||
| } | ||
|
|
||
| func (d *DebugLogger) LogEvents(events []types.SingularEventT, commonMedata *ptrans.Metadata) error { | ||
| if len(events) == 0 { | ||
| return nil | ||
| } | ||
| d.eventLogMutex.Lock() | ||
| defer d.eventLogMutex.Unlock() | ||
|
|
||
| if d.loggedEvents >= int64(d.maxLoggedEvents.Load()) { | ||
| return nil | ||
| } | ||
|
|
||
| logEntries := lo.Map(events, func(item types.SingularEventT, index int) string { | ||
| return stringify.Any(ptrans.TransformerEvent{ | ||
| Message: item, | ||
| Metadata: *commonMedata, | ||
| }) | ||
| }) | ||
|
|
||
| if err := d.writeLogEntries(logEntries); err != nil { | ||
| return fmt.Errorf("logging events: %w", err) | ||
| } | ||
|
|
||
| d.logger.Infon("Successfully logged events", logger.NewIntField("event_count", int64(len(logEntries)))) | ||
| d.loggedEvents += int64(len(logEntries)) | ||
| return nil | ||
| } | ||
|
|
||
| func (d *DebugLogger) writeLogEntries(entries []string) error { | ||
| writer, err := misc.CreateBufferedWriter(d.currentLogFileName) | ||
| if err != nil { | ||
| return fmt.Errorf("creating buffered writer: %w", err) | ||
| } | ||
| defer func() { _ = writer.Close() }() | ||
|
|
||
| for _, entry := range entries { | ||
| if _, err := writer.Write([]byte(entry + "\n")); err != nil { | ||
| return fmt.Errorf("writing log entry: %w", err) | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
Oops, something went wrong.