Documentation
¶
Index ¶
- Constants
- Variables
- func AddFileToZip(zipWriter *zip.Writer, filename string) error
- func AppendError(callingMethodName string, firstError, secondError *error)
- func BugsnagNotify(ctx context.Context, team string) func()
- func ConcatErrors(givenErrors []error) error
- func ContainsInt(slice []int, val int) bool
- func ContainsInt64(slice []int64, val int64) bool
- func ContainsString(slice []string, str string) bool
- func ConvertInterfaceToStringArray(input []interface{}) []string
- func ConvertStringInterfaceToIntArray(interfaceT interface{}) ([]int64, error)
- func Copy(dst, src interface{})
- func CopyStringMap(originalMap map[string]string) map[string]string
- func CreateTMPDIR() (string, error)
- func FastUUID() uuid.UUID
- func GetChronologicalTimeStamp(receivedAt, sentAt, originalTimestamp time.Time) time.Time
- func GetDatabricksVersion() (version string)
- func GetHash(s string) int
- func GetIPFromReq(req *http.Request) string
- func GetJsonSchemaDTFromGoDT(goType string) string
- func GetMD5Hash(input string) string
- func GetMD5UUID(str string) (uuid.UUID, error)
- func GetMacAddress() string
- func GetMandatoryJSONFieldNames(st interface{}) []string
- func GetNodeID() string
- func GetObjectStorageConfig(opts ObjectStorageOptsT) map[string]interface{}
- func GetOutboundIP() (net.IP, error)
- func GetParsedTimestamp(input interface{}) (time.Time, bool)
- func GetRudderEventVal(key string, rudderEvent types.SingularEventT) (interface{}, bool)
- func GetRudderID(event types.SingularEventT) (string, bool)
- func GetRudderObjectStorageAccessKeys() (accessKeyID, accessKey string)
- func GetRudderObjectStorageConfig(prefixOverride string) (storageConfig map[string]interface{})
- func GetRudderObjectStoragePrefix() (prefix string)
- func GetSpacesLocation(location string) (region string)
- func GetStringifiedData(data interface{}) string
- func GetTagName(id string, names ...string) string
- func GetWarehouseURL() (url string)
- func HTTPCallWithRetry(url string, payload []byte) ([]byte, int)
- func HTTPCallWithRetryWithTimeout(url string, payload []byte, timeout time.Duration) ([]byte, int)
- func HasAWSKeysInConfig(config interface{}) bool
- func HasAWSRegionInConfig(config interface{}) bool
- func IncrementMapByKey(m map[string]int, key string, increment int)
- func Init()
- func IntArrayToString(a []int64, delim string) string
- func IsConfiguredToUseRudderObjectStorage(storageConfig map[string]interface{}) bool
- func IsValidUUID(uuid string) bool
- func KeepProcessAlive()
- func LoadDestinations() ([]string, []string)
- func MakeHTTPRequestWithTimeout(url string, payload io.Reader, timeout time.Duration) ([]byte, int, error)
- func MakeJSONArray(bytesArray [][]byte) []byte
- func MakeRetryablePostRequest(url, endpoint string, data interface{}) (response []byte, statusCode int, err error)
- func MapLookup(mapToLookup map[string]interface{}, keys ...string) interface{}
- func MapStringKeys(input map[string]interface{}) []string
- func MaxInt(a, b int) int
- func MergeMaps(maps ...map[string]interface{}) map[string]interface{}
- func MinInt(a, b int) int
- func NestedMapLookup(m map[string]interface{}, ks ...string) (rval interface{}, err error)
- func ParseRudderEventBatch(eventPayload json.RawMessage) ([]types.SingularEventT, bool)
- func QuoteLiteral(literal string) string
- func ReadLines(path string) ([]string, error)
- func RecordAppError(err error)
- func RemoveEmptyFolderStructureForFilePath(fp string)
- func RemoveFilePaths(filePaths ...string)
- func ReplaceDB(dbName, targetName string)
- func ReplaceMultiRegex(str string, expList map[string]string) (string, error)
- func RetryWith(parentContext context.Context, timeout time.Duration, maxAttempts int, ...) error
- func ReverseInt(s []int) []int
- func RunWithConcurrency(config *RWCConfig)
- func RunWithTimeout(f, onTimeout func(), d time.Duration)
- func SingleQuoteLiteralJoin(slice []string) string
- func SleepCtx(ctx context.Context, delay time.Duration) bool
- func SortMap(inputMap map[string]metric.MovingAverage) []string
- func SortedMapKeys(input interface{}) []string
- func SortedStructSliceValues(input interface{}, filedName string) []string
- func StringKeys(input interface{}) []string
- func TailTruncateStr(str string, count int) string
- func TruncateStr(str string, limit int) string
- func Unique(stringSlice []string) []string
- func UpdateJSONWithNewKeyVal(params []byte, key string, val interface{}) []byte
- func UseFairPickup() bool
- func WithBugsnag(fn func() error) func() error
- func WithBugsnagForWarehouse(fn func() error) func() error
- func ZipFiles(filename string, files []string) error
- type BufferedWriter
- type DefaultString
- type ErrorStoreT
- type GZipWriter
- type ObjectStorageOptsT
- type PerfStats
- type RFP
- type RWCConfig
- type RWCJob
- type RudderError
- type WaitGroup
Constants ¶
const ( // RFC3339Milli with milli sec precision RFC3339Milli = "2006-01-02T15:04:05.000Z07:00" POSTGRESTIMEFORMATPARSE = "2006-01-02T15:04:05" )
const ( RudderAsyncDestinationLogs = "rudder-async-destination-logs" RudderArchives = "rudder-archives" RudderWarehouseStagingUploads = "rudder-warehouse-staging-uploads" RudderRawDataDestinationLogs = "rudder-raw-data-destination-logs" RudderWarehouseLoadUploadsTmp = "rudder-warehouse-load-uploads-tmp" RudderIdentityMergeRulesTmp = "rudder-identity-merge-rules-tmp" RudderIdentityMappingsTmp = "rudder-identity-mappings-tmp" RudderRedshiftManifests = "rudder-redshift-manifests" RudderWarehouseJsonUploadsTmp = "rudder-warehouse-json-uploads-tmp" RudderTestPayload = "rudder-test-payload" )
Variables ¶
var (
AppStartTime int64
)
Functions ¶
func AddFileToZip ¶
AddFileToZip adds file to zip including size header stats
func AppendError ¶
AppendError creates or appends second error to first error
func BugsnagNotify ¶
func ConcatErrors ¶
func ContainsInt ¶
func ContainsInt64 ¶
func ContainsString ¶
func ConvertInterfaceToStringArray ¶
func ConvertInterfaceToStringArray(input []interface{}) []string
func Copy ¶
func Copy(dst, src interface{})
Copy copies the exported fields from src to dest Used for copying the default transport
func CreateTMPDIR ¶
CreateTMPDIR creates tmp dir at path configured via RUDDER_TMPDIR env var
func GetChronologicalTimeStamp ¶
Returns chronological timestamp of the event using the formula timestamp = receivedAt - (sentAt - originalTimestamp)
func GetDatabricksVersion ¶
func GetDatabricksVersion() (version string)
func GetIPFromReq ¶
GetIPFromReq gets ip address from request
func GetJsonSchemaDTFromGoDT ¶
GetJsonSchemaDTFromGoDT returns the json schema supported data types from go lang supported data types. References: 1. Go supported types: https://golangbyexample.com/all-data-types-in-golang-with-examples/ 2. Json schema supported types: https://json-schema.org/understanding-json-schema/reference/type.html
func GetMD5Hash ¶
GetMD5Hash returns EncodeToString(md5 hash of the input string)
func GetMD5UUID ¶
GetMD5UUID hashes the given string into md5 and returns it as auuid
func GetMacAddress ¶
func GetMacAddress() string
func GetMandatoryJSONFieldNames ¶
func GetMandatoryJSONFieldNames(st interface{}) []string
GetMandatoryJSONFieldNames returns all the json field names defined against the json tag for each field.
func GetObjectStorageConfig ¶
func GetObjectStorageConfig(opts ObjectStorageOptsT) map[string]interface{}
func GetOutboundIP ¶
GetOutboundIP returns preferred outbound ip of this machine https://stackoverflow.com/a/37382208
func GetParsedTimestamp ¶
GetParsedTimestamp returns the parsed timestamp
func GetRudderEventVal ¶
func GetRudderEventVal(key string, rudderEvent types.SingularEventT) (interface{}, bool)
GetRudderEventVal returns the value corresponding to the key in the message structure
func GetRudderID ¶
func GetRudderID(event types.SingularEventT) (string, bool)
GetRudderID return the UserID from the object
func GetRudderObjectStorageAccessKeys ¶
func GetRudderObjectStorageAccessKeys() (accessKeyID, accessKey string)
func GetRudderObjectStoragePrefix ¶
func GetRudderObjectStoragePrefix() (prefix string)
func GetSpacesLocation ¶
func GetStringifiedData ¶
func GetStringifiedData(data interface{}) string
func GetTagName ¶
GetTagName gets the tag name using a uuid and name
func GetWarehouseURL ¶
func GetWarehouseURL() (url string)
func HasAWSKeysInConfig ¶
func HasAWSKeysInConfig(config interface{}) bool
func HasAWSRegionInConfig ¶
func HasAWSRegionInConfig(config interface{}) bool
func IncrementMapByKey ¶
IncrementMapByKey starts with 1 and increments the counter of a key
func IntArrayToString ¶
func IsValidUUID ¶
IsValidUUID will check if provided string is a valid UUID
func KeepProcessAlive ¶
func KeepProcessAlive()
func LoadDestinations ¶
func MakeJSONArray ¶
func MakeRetryablePostRequest ¶
func MakeRetryablePostRequest(url, endpoint string, data interface{}) (response []byte, statusCode int, err error)
MakeRetryablePostRequest is Util function to make a post request.
func MapStringKeys ¶
func NestedMapLookup ¶
NestedMapLookup m: a map from strings to other maps or values, of arbitrary depth ks: successive keys to reach an internal or leaf node (variadic) If an internal node is reached, will return the internal map
Returns: (Exactly one of these will be nil) rval: the target node (if found) err: an error created by fmt.Errorf
func ParseRudderEventBatch ¶
func ParseRudderEventBatch(eventPayload json.RawMessage) ([]types.SingularEventT, bool)
ParseRudderEventBatch looks for the batch structure inside event
func QuoteLiteral ¶
func RecordAppError ¶
func RecordAppError(err error)
RecordAppError appends the error occurred to error_store.json
func RemoveEmptyFolderStructureForFilePath ¶
func RemoveEmptyFolderStructureForFilePath(fp string)
RemoveEmptyFolderStructureForFilePath recursively cleans up everything till it reaches the stage where the folders are not empty or parent.
func RemoveFilePaths ¶
func RemoveFilePaths(filePaths ...string)
RemoveFilePaths removes filePaths as well as cleans up the empty folder structure.
func ReplaceDB ¶
func ReplaceDB(dbName, targetName string)
ReplaceDB : Rename the OLD DB and create a new one. Since we are not journaling, this should be idemponent
func ReplaceMultiRegex ¶
func RunWithConcurrency ¶
func RunWithConcurrency(config *RWCConfig)
RunWithConcurrency runs provided function f with concurrency provided by the factor factor.
func RunWithTimeout ¶
RunWithTimeout runs provided function f until provided timeout d. If the timeout is reached, onTimeout callback will be called.
func SingleQuoteLiteralJoin ¶
func SortedMapKeys ¶
func SortedMapKeys(input interface{}) []string
func SortedStructSliceValues ¶
func StringKeys ¶
func StringKeys(input interface{}) []string
func TailTruncateStr ¶
TailTruncateStr returns the last `count` digits of a string
func TruncateStr ¶
func UpdateJSONWithNewKeyVal ¶
UpdateJSONWithNewKeyVal enhances the json passed with key, val
func UseFairPickup ¶
func UseFairPickup() bool
func WithBugsnag ¶
func WithBugsnagForWarehouse ¶
Types ¶
type BufferedWriter ¶
func CreateBufferedWriter ¶
func CreateBufferedWriter(s string) (w BufferedWriter, err error)
func (BufferedWriter) Close ¶
func (b BufferedWriter) Close() error
func (BufferedWriter) GetFile ¶
func (b BufferedWriter) GetFile() *os.File
type DefaultString ¶
type DefaultString string
DefaultString is a utility type for providing default values using one-liners in otherwise multi-line scenarios.
E.g. this
v := misc.DefaultString("unknown").OnError(os.Hostname())
is the equivalent of
v, err := os.Hostname()
if err != nil {
v = "unknown"
}
type ErrorStoreT ¶
type ErrorStoreT struct {
Errors []RudderError
}
ErrorStoreT : DS to store the app errors
type GZipWriter ¶
func CreateGZ ¶
func CreateGZ(s string) (w GZipWriter, err error)
func (GZipWriter) Close ¶
func (w GZipWriter) Close() error
func (GZipWriter) CloseGZ ¶
func (w GZipWriter) CloseGZ() error
func (GZipWriter) GetLoadFile ¶
func (w GZipWriter) GetLoadFile() *os.File
func (GZipWriter) WriteGZ ¶
func (w GZipWriter) WriteGZ(s string) error
func (GZipWriter) WriteRow ¶
func (w GZipWriter) WriteRow(row []interface{}) error
type ObjectStorageOptsT ¶
type PerfStats ¶
type PerfStats struct {
// contains filtered or unexported fields
}
PerfStats is the class for managing performance stats. Not multi-threaded safe now
func (*PerfStats) End ¶
End marks the end of one round of stat collection. events is number of events processed since start
type RFP ¶
type RFP struct {
// contains filtered or unexported fields
}
func GetReservedFolderPaths ¶
func GetReservedFolderPaths() []*RFP
GetReservedFolderPaths returns all temporary folder paths.
type RWCJob ¶
type RWCJob interface{}
RWCConfig config for RunWithConcurrency factor: number of concurrent job jobs: range of jobs you need to provide runJob: caller function for the concurrent job
type RudderError ¶
type RudderError struct {
StartTime int64
CrashTime int64
ReadableStartTime string
ReadableCrashTime string
Message string
StackTrace string
Code int
}
RudderError : to store rudder error
type WaitGroup ¶
type WaitGroup struct {
// contains filtered or unexported fields
}
func NewWaitGroup ¶
func NewWaitGroup() *WaitGroup
Source Files