elasticsearch

Documentation

Index

• Constants
• type Document
• type DocumentSearch
  • func (ds *DocumentSearch) String() string
• type FilterParams
  • func (fp *FilterParams) Query(latestTimestamp time.Time) (map[string]any, error)
• type Index
  • func NewIndex() *Index
  • func (i *Index) DocumentAlreadySeen(document *Document) bool
  • func (i *Index) UpdateLatestTimestampLastRun()
• type IndexStatus
• type Indices
  • func (indices *Indices) GetProcessedDocumentCount() int
  • func (indices *Indices) Update(ctx context.Context, client *es.TypedClient) error
  • func (indices *Indices) UpdateProcessedDocumentCount(additionalDocumentsProcessed int)
• type PointInTime
• type SearchRequestBody
• type Source
  • func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, ...) error
  • func (s *Source) Init(aCtx context.Context, name string, jobId sources.JobID, ...) error
  • func (s *Source) JobID() sources.JobID
  • func (s *Source) SourceID() sources.SourceID
  • func (s *Source) Type() sourcespb.SourceType
• type UnitOfWork
  • func NewUnitOfWork(maxDocumentCount int) UnitOfWork

Constants

const PAGE_SIZE = 10

const SourceType = sourcespb.SourceType_SOURCE_TYPE_ELASTICSEARCH

Types

type Document

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

type DocumentSearch

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

func (*DocumentSearch) String

func (ds *DocumentSearch) String() string

type FilterParams

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

func (*FilterParams) Query

func (fp *FilterParams) Query(latestTimestamp time.Time) (map[string]any, error)

type Index

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

func NewIndex

func NewIndex() *Index

func (*Index) DocumentAlreadySeen

func (i *Index) DocumentAlreadySeen(document *Document) bool

func (*Index) UpdateLatestTimestampLastRun

func (i *Index) UpdateLatestTimestampLastRun()

type IndexStatus

type IndexStatus int

type Indices

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

func (*Indices) GetProcessedDocumentCount

func (indices *Indices) GetProcessedDocumentCount() int

Returns the number of documents processed within these indices

func (*Indices) Update

func (indices *Indices) Update(
	ctx context.Context,
	client *es.TypedClient,
) error

Updates a set of indices from an Elasticsearch cluster. If an index has been deleted it will be removed; if it's been added it'll be added; if its document count has changed (based on filterParams and latestTimestamp) it'll be updated.

func (*Indices) UpdateProcessedDocumentCount

func (indices *Indices) UpdateProcessedDocumentCount(additionalDocumentsProcessed int)

Adds documents processed to the count, used for progress

type PointInTime

type PointInTime struct {
	ID        string `json:"id"`
	KeepAlive string `json:"keep_alive"`
}

type SearchRequestBody

type SearchRequestBody struct {
	PIT         PointInTime    `json:"pit"`
	Sort        []string       `json:"sort"`
	SearchAfter []int          `json:"search_after,omitempty"`
	Query       map[string]any `json:"query,omitempty"`
}

type Source

type Source struct {
	sources.Progress
	// contains filtered or unexported fields
}

func (*Source) Chunks

func (s *Source) Chunks(
	ctx context.Context,
	chunksChan chan *sources.Chunk,
	targets ...sources.ChunkingTarget,
) error

Chunks emits chunks of bytes over a channel.

func (*Source) Init

func (s *Source) Init(
	aCtx context.Context,
	name string,
	jobId sources.JobID,
	sourceId sources.SourceID,
	verify bool,
	connection *anypb.Any,
	concurrency int,
) error

Init returns an initialized Elasticsearch source

func (*Source) JobID

func (s *Source) JobID() sources.JobID

func (*Source) SourceID

func (s *Source) SourceID() sources.SourceID

func (*Source) Type

func (s *Source) Type() sourcespb.SourceType

type UnitOfWork

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

func NewUnitOfWork

func NewUnitOfWork(maxDocumentCount int) UnitOfWork