ingestjob

README

ingest job

ingestjob library used by connectors to handle logging

How it works :

• when connector start it should create new log record using Write function with a unique identity based on connector, configuration and creation date with status in progress.

• once connector finished it should call Write function again with the same identity connector, configuration and creation date which internally update the same log record to done status.

• in case connector failed to complete its run, it should call Write function again with the same identity which internally update the same log record to failed status.

Ingest job has 4 functionalities:

1- Write create a log record if it is not exist. if record exists update it.

2- Read get a list of logs based on connector and status.

3- Count which count connector logs in a specific status.

4- Filter which filter log records based on any of status, configuration, from and to date.

Documentation

Index

• Constants
• type ESLogProvider
• type Hits
• type Log
• type Logger
  • func NewLogger(esClient ESLogProvider, environment string) (*Logger, error)
  • func (s *Logger) Count(connector string, status string) (int, error)
  • func (s *Logger) Filter(log *Log) ([]Log, error)
  • func (s *Logger) Read(connector string, status string) ([]Log, error)
  • func (s *Logger) Write(log *Log) error
  • func (s *Logger) WriteTask(log *TaskLog) error
• type NestedHits
• type TaskHits
• type TaskLog
• type TaskNestedHits
• type TaskTopHits
• type TopHits

Constants

const (
	InProgress = "inprogress"
	Failed     = "failed"
	Done       = "done"
	Internal   = "internal"

	QuerySize = 1000
)

Types

type ESLogProvider

type ESLogProvider interface {
	CreateDocument(index, documentID string, body []byte) ([]byte, error)
	Get(index string, query map[string]interface{}, result interface{}) error
	BackOffGet(index string, query map[string]interface{}, result interface{}, attempts uint, delay time.Duration) error
	UpdateDocument(index string, id string, body interface{}) ([]byte, error)
	Count(index string, query map[string]interface{}) (int, error)
}

ESLogProvider used in connecting to ES logging server

type Hits

type Hits struct {
	Hits []NestedHits `json:"hits"`
}

Hits result

type Log

type Log struct {
	Connector     string              `json:"connector"`
	Configuration []map[string]string `json:"configuration"`
	Status        string              `json:"status"`
	CreatedAt     time.Time           `json:"created_at"`
	UpdatedAt     time.Time           `json:"updated_at"`
	Message       string              `json:"message"`
	TaskARN       string              `json:"task_arn"`
	From          *time.Time          `json:"from,omitempty"`
	To            *time.Time          `json:"to,omitempty"`
}

Log ...

type Logger

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

Logger ...

func NewLogger

func NewLogger(esClient ESLogProvider, environment string) (*Logger, error)

NewLogger ...

func (*Logger) Count

func (s *Logger) Count(connector string, status string) (int, error)

Count ...

func (*Logger) Filter

func (s *Logger) Filter(log *Log) ([]Log, error)

Filter connector logs based on status, configuration and creation date

func (*Logger) Read

func (s *Logger) Read(connector string, status string) ([]Log, error)

Read ...

func (*Logger) Write

func (s *Logger) Write(log *Log) error

Write ...

func (*Logger) WriteTask

func (s *Logger) WriteTask(log *TaskLog) error

WriteTask ...

type NestedHits

type NestedHits struct {
	ID     string `json:"_id"`
	Source Log    `json:"_source"`
}

NestedHits is the actual hit data

type TaskHits

type TaskHits struct {
	Hits []TaskNestedHits `json:"hits"`
}

TaskHits result

type TaskLog

type TaskLog struct {
	Id            string              `json:"id"`            // task id
	Connector     string              `json:"connector"`     // name of connector as https://github.com/LF-Engineering/lfx-event-schema/blob/main/service/insights/shared.go#L32-L49
	EndpointId    string              `json:"endpoint_id"`   // generated ID of the endpoint per lfx-event-schema generator functions
	Configuration []map[string]string `json:"configuration"` // metadata
	CreatedAt     *time.Time          `json:"created_at,omitempty"`
}

type TaskNestedHits

type TaskNestedHits struct {
	ID     string  `json:"_id"`
	Source TaskLog `json:"_source"`
}

TaskNestedHits is the actual hit data

type TaskTopHits

type TaskTopHits struct {
	Hits TaskHits `json:"hits"`
}

TaskTopHits result

type TopHits

type TopHits struct {
	Hits Hits `json:"hits"`
}

TopHits result