-
Notifications
You must be signed in to change notification settings - Fork 322
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
5fd9fba
commit ca8ec5a
Showing
45 changed files
with
12,959 additions
and
11 deletions.
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,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 | ||
| } |
60 changes: 60 additions & 0 deletions
60
warehouse/transformer/internal/reservedkeywords/reservedkeywords.go
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,60 @@ | ||
| package reservedkeywords | ||
|
|
||
| import ( | ||
| "embed" | ||
| "log" | ||
| "strings" | ||
|
|
||
| jsoniter "github.com/json-iterator/go" | ||
| "github.com/samber/lo" | ||
| ) | ||
|
|
||
| var ( | ||
| //go:embed reservedtablescolumns.json | ||
| tablesColumnsFile embed.FS | ||
|
|
||
| //go:embed reservednamespaces.json | ||
| namespacesFile embed.FS | ||
|
|
||
| reservedTablesColumns, reservedNamespaces map[string]map[string]struct{} | ||
|
|
||
| json = jsoniter.ConfigCompatibleWithStandardLibrary | ||
| ) | ||
|
|
||
| func init() { | ||
| reservedTablesColumns = load(tablesColumnsFile, "reservedtablescolumns.json") | ||
| reservedNamespaces = load(namespacesFile, "reservednamespaces.json") | ||
| } | ||
|
|
||
| func load(file embed.FS, fileName string) map[string]map[string]struct{} { | ||
| data, err := file.ReadFile(fileName) | ||
| if err != nil { | ||
| log.Fatalf("failed to load reserved keywords from %s: %v", fileName, err) | ||
| } | ||
|
|
||
| var tempKeywords map[string][]string | ||
| if err := json.Unmarshal(data, &tempKeywords); err != nil { | ||
| log.Fatalf("failed to parse reserved keywords from %s: %v", fileName, err) | ||
| } | ||
|
|
||
| return lo.MapValues(tempKeywords, func(keywords []string, _ string) map[string]struct{} { | ||
| return lo.SliceToMap(keywords, func(k string) (string, struct{}) { | ||
| return strings.ToUpper(k), struct{}{} | ||
| }) | ||
| }) | ||
| } | ||
|
|
||
| // IsTableOrColumn checks if the given keyword is a reserved table/column keyword for the destination type. | ||
| func IsTableOrColumn(destType, keyword string) bool { | ||
| return isKeywordReserved(reservedTablesColumns, destType, keyword) | ||
| } | ||
|
|
||
| // IsNamespace checks if the given keyword is a reserved namespace keyword for the destination type. | ||
| func IsNamespace(destType, keyword string) bool { | ||
| return isKeywordReserved(reservedNamespaces, destType, keyword) | ||
| } | ||
|
|
||
| func isKeywordReserved(keywords map[string]map[string]struct{}, destType, keyword string) bool { | ||
| _, exists := keywords[destType][strings.ToUpper(keyword)] | ||
| return exists | ||
| } |
Oops, something went wrong.