router

Documentation

Index

• Variables
• func Init()
• func InitRouterAdmin()
• func RegisterAdminHandlers(readonlyRouterDB, readonlyBatchRouterDB jobsdb.ReadonlyJobsDB)
• type Admin
  • func (ra *Admin) Status() interface{}
• type DSStats
• type DiagnosticT
• type ErrorCodeCountsByDestination
• type Factory
  • func (f *Factory) New(destination *backendconfig.DestinationT, identifier string) *HandleT
• type HandleDestOAuthRespParamsT
• type HandleT
  • func (rt *HandleT) ExecDisableDestination(destination *backendconfig.DestinationT, ...) (int, string)
  • func (rt *HandleT) HandleOAuthDestResponse(params *HandleDestOAuthRespParamsT) (int, string)
  • func (rt *HandleT) Setup(backendConfig backendconfig.BackendConfig, jobsDB jobsdb.MultiTenantJobsDB, ...)
  • func (rt *HandleT) Shutdown()
  • func (rt *HandleT) Start()
• type JSONResponseHandler
  • func (handler *JSONResponseHandler) IsSuccessStatus(respCode int, respBody string) (returnCode int)
• type JobCountByConnections
• type JobCountsByStateAndDestination
• type JobParametersT
• type JobResponse
• type LatestJobStatusCounts
• type NetHandleI
• type NetHandleT
  • func (network *NetHandleT) SendPost(ctx context.Context, structData integrations.PostParametersT) *utils.SendPostResponse
  • func (network *NetHandleT) Setup(destID string, netClientTimeout time.Duration)
• type ResponseHandlerI
  • func New(responseRules map[string]interface{}) ResponseHandlerI
• type RouterRpcHandler
  • func (r *RouterRpcHandler) GetDSFailedJobs(arg string, result *string) (err error)
  • func (r *RouterRpcHandler) GetDSJobCount(arg string, result *string) (err error)
  • func (r *RouterRpcHandler) GetDSList(_ string, result *string) (err error)
  • func (r *RouterRpcHandler) GetDSStats(dsName string, result *string) (err error)
  • func (r *RouterRpcHandler) GetJobByID(arg string, result *string) (err error)
  • func (r *RouterRpcHandler) GetJobIDStatus(arg string, result *string) (err error)
• type TXTResponseHandler
  • func (*TXTResponseHandler) IsSuccessStatus(respCode int, _ string) (returnCode int)

Variables

var (
	Diagnostics diagnostics.DiagnosticsI
)

Functions

func Init

func Init()

func InitRouterAdmin

func InitRouterAdmin()

func RegisterAdminHandlers

func RegisterAdminHandlers(readonlyRouterDB, readonlyBatchRouterDB jobsdb.ReadonlyJobsDB)

Types

type Admin

type Admin struct {
	// contains filtered or unexported fields
}

func (*Admin) Status

func (ra *Admin) Status() interface{}

Status function is used for debug purposes by the admin interface

type DSStats

type DSStats struct {
	JobCountsByStateAndDestination []JobCountsByStateAndDestination
	ErrorCodeCountsByDestination   []ErrorCodeCountsByDestination
	JobCountByConnections          []JobCountByConnections
	LatestJobStatusCounts          []LatestJobStatusCounts
	UnprocessedJobCounts           int
}

type DiagnosticT

type DiagnosticT struct {
	// contains filtered or unexported fields
}

type ErrorCodeCountsByDestination

type ErrorCodeCountsByDestination struct {
	Count         int
	ErrorCode     string
	Destination   string
	DestinationID string
}

type Factory

type Factory struct {
	Reporting        reporter
	Multitenant      tenantStats
	BackendConfig    backendconfig.BackendConfig
	RouterDB         jobsdb.MultiTenantJobsDB
	ProcErrorDB      jobsdb.JobsDB
	TransientSources transientsource.Service
	RsourcesService  rsources.JobService
	ThrottlerFactory *throttler.Factory
}

func (*Factory) New

func (f *Factory) New(destination *backendconfig.DestinationT, identifier string) *HandleT

type HandleDestOAuthRespParamsT

type HandleDestOAuthRespParamsT struct {
	// contains filtered or unexported fields
}

type HandleT

type HandleT struct {
	MultitenantI tenantStats

	Reporting reporter
	// contains filtered or unexported fields
}

HandleT is the handle to this module.

func (*HandleT) ExecDisableDestination

func (rt *HandleT) ExecDisableDestination(destination *backendconfig.DestinationT, workspaceID, destResBody, rudderAccountId string) (int, string)

func (*HandleT) HandleOAuthDestResponse

func (rt *HandleT) HandleOAuthDestResponse(params *HandleDestOAuthRespParamsT) (int, string)

func (*HandleT) Setup

func (rt *HandleT) Setup(backendConfig backendconfig.BackendConfig, jobsDB jobsdb.MultiTenantJobsDB, errorDB jobsdb.JobsDB, destinationConfig destinationConfig, transientSources transientsource.Service, rsourcesService rsources.JobService)

Setup initializes this module

func (*HandleT) Shutdown

func (rt *HandleT) Shutdown()

func (*HandleT) Start

func (rt *HandleT) Start()

type JSONResponseHandler

type JSONResponseHandler struct {
	// contains filtered or unexported fields
}

JSONResponseHandler handler for json response

func (*JSONResponseHandler) IsSuccessStatus

func (handler *JSONResponseHandler) IsSuccessStatus(respCode int, respBody string) (returnCode int)

IsSuccessStatus - returns the status code based on the response code and body

type JobCountByConnections

type JobCountByConnections struct {
	Count         int
	SourceId      string
	DestinationId string
}

type JobCountsByStateAndDestination

type JobCountsByStateAndDestination struct {
	Count       int
	State       string
	Destination string
}

type JobParametersT

type JobParametersT struct {
	SourceID                string      `json:"source_id"`
	DestinationID           string      `json:"destination_id"`
	ReceivedAt              string      `json:"received_at"`
	TransformAt             string      `json:"transform_at"`
	SourceBatchID           string      `json:"source_batch_id"`
	SourceTaskID            string      `json:"source_task_id"`
	SourceTaskRunID         string      `json:"source_task_run_id"`
	SourceJobID             string      `json:"source_job_id"`
	SourceJobRunID          string      `json:"source_job_run_id"`
	SourceDefinitionID      string      `json:"source_definition_id"`
	DestinationDefinitionID string      `json:"destination_definition_id"`
	SourceCategory          string      `json:"source_category"`
	RecordID                interface{} `json:"record_id"`
	MessageID               string      `json:"message_id"`
	WorkspaceID             string      `json:"workspaceId"`
	RudderAccountID         string      `json:"rudderAccountId"`
}

JobParametersT struct holds source id and destination id of a job

type JobResponse

type JobResponse struct {
	// contains filtered or unexported fields
}

type LatestJobStatusCounts

type LatestJobStatusCounts struct {
	Count int
	State string
	Rank  int
}

type NetHandleI

type NetHandleI interface {
	SendPost(ctx context.Context, structData integrations.PostParametersT) *utils.SendPostResponse
}

Network interface

type NetHandleT

type NetHandleT struct {
	// contains filtered or unexported fields
}

NetHandleT is the wrapper holding private variables

func (*NetHandleT) SendPost

func (network *NetHandleT) SendPost(ctx context.Context, structData integrations.PostParametersT) *utils.SendPostResponse

SendPost takes the EventPayload of a transformed job, gets the necessary values from the payload and makes a call to destination to push the event to it this returns the statusCode, status and response body from the response of the destination call

func (*NetHandleT) Setup

func (network *NetHandleT) Setup(destID string, netClientTimeout time.Duration)

Setup initializes the module

type ResponseHandlerI

type ResponseHandlerI interface {
	IsSuccessStatus(respCode int, respBody string) (returnCode int)
}

ResponseHandlerI - handle destination response

func New

func New(responseRules map[string]interface{}) ResponseHandlerI

New returns a destination response handler. Can be nil(Check before using this)

type RouterRpcHandler

type RouterRpcHandler struct {
	// contains filtered or unexported fields
}

func (*RouterRpcHandler) GetDSFailedJobs

func (r *RouterRpcHandler) GetDSFailedJobs(arg string, result *string) (err error)

func (*RouterRpcHandler) GetDSJobCount

func (r *RouterRpcHandler) GetDSJobCount(arg string, result *string) (err error)

func (*RouterRpcHandler) GetDSList

func (r *RouterRpcHandler) GetDSList(_ string, result *string) (err error)

func (*RouterRpcHandler) GetDSStats

func (r *RouterRpcHandler) GetDSStats(dsName string, result *string) (err error)

GetDSStats group_by job_status group by custom_val Get all errors = distinct (error), count(*) where state=failed Distinct (src_id, dst_id) Router jobs status flow ⇒ ordered by rank unprocessed_params ⇒ Num jobs not yet picked

func (*RouterRpcHandler) GetJobByID

func (r *RouterRpcHandler) GetJobByID(arg string, result *string) (err error)

func (*RouterRpcHandler) GetJobIDStatus

func (r *RouterRpcHandler) GetJobIDStatus(arg string, result *string) (err error)

type TXTResponseHandler

type TXTResponseHandler struct {
	// contains filtered or unexported fields
}

TXTResponseHandler handler for text response

func (*TXTResponseHandler) IsSuccessStatus

func (*TXTResponseHandler) IsSuccessStatus(respCode int, _ string) (returnCode int)

IsSuccessStatus - returns the status code based on the response code and body