adx

Documentation

Index

• Constants
• func NewDispatcher(uploaders []Uploader) *dispatcher
• func NewUploader(kustoCli *kusto.Client, opts UploaderOpts) *uploader
• type DropUnusedTablesTask
  • func NewDropUnusedTablesTask(kustoCli StatementExecutor) *DropUnusedTablesTask
  • func (t *DropUnusedTablesTask) Run(ctx context.Context) error
• type IngestionMapping
• type SampleType
• type StatementExecutor
• type SyncFunctionsTask
  • func NewSyncFunctionsTask(store storage.Functions, kustoCli StatementExecutor) *SyncFunctionsTask
  • func (t *SyncFunctionsTask) Run(ctx context.Context) error
• type Syncer
  • func NewSyncer(kustoCli mgmt, database string, defaultMapping schema.SchemaMapping, ...) *Syncer
  • func (s *Syncer) Close() error
  • func (s *Syncer) EnsureDefaultMapping(table string) (string, error)
  • func (s *Syncer) EnsureDefaultTable(table string) error
  • func (s *Syncer) EnsureMapping(table string, mapping schema.SchemaMapping) (string, error)
  • func (s *Syncer) EnsureTable(table string, mapping schema.SchemaMapping) error
  • func (s *Syncer) Open(ctx context.Context) error
• type Table
• type TableDetail
• type Uploader
  • func NewFakeUploader(db string) Uploader
• type UploaderOpts

Constants

const ConcurrentUploads = 50

Functions

func NewDispatcher

func NewDispatcher(uploaders []Uploader) *dispatcher

func NewUploader

func NewUploader(kustoCli *kusto.Client, opts UploaderOpts) *uploader

Types

type DropUnusedTablesTask

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

func NewDropUnusedTablesTask

func NewDropUnusedTablesTask(kustoCli StatementExecutor) *DropUnusedTablesTask

func (*DropUnusedTablesTask) Run

func (t *DropUnusedTablesTask) Run(ctx context.Context) error

type IngestionMapping

type IngestionMapping struct {
	Name          string    `kusto:"Name"`
	Kind          string    `kusto:"Kind"`
	Mapping       string    `kusto:"Mapping"`
	LastUpdatedOn time.Time `kusto:"LastUpdatedOn"`
	Database      string    `kusto:"Database"`
	Table         string    `kusto:"Table"`
}

type SampleType

type SampleType int

const (
	PromMetrics SampleType = iota
	OTLPLogs
)

type StatementExecutor

type StatementExecutor interface {
	Database() string
	Mgmt(ctx context.Context, query kusto.Statement, options ...kusto.MgmtOption) (*kusto.RowIterator, error)
}

type SyncFunctionsTask

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

func NewSyncFunctionsTask

func NewSyncFunctionsTask(store storage.Functions, kustoCli StatementExecutor) *SyncFunctionsTask

func (*SyncFunctionsTask) Run

func (t *SyncFunctionsTask) Run(ctx context.Context) error

type Syncer

type Syncer struct {
	KustoCli mgmt
	// contains filtered or unexported fields
}

func NewSyncer

func NewSyncer(kustoCli mgmt, database string, defaultMapping schema.SchemaMapping, st SampleType) *Syncer

func (*Syncer) Close

func (s *Syncer) Close() error

func (*Syncer) EnsureDefaultMapping

func (s *Syncer) EnsureDefaultMapping(table string) (string, error)

EnsureMapping creates a schema mapping for the specified table if it does not exist. It returns the name of the mapping.

func (*Syncer) EnsureDefaultTable

func (s *Syncer) EnsureDefaultTable(table string) error

EnsureDefaultTable creates a table with the default schema mapping if it does not exist.

func (*Syncer) EnsureMapping

func (s *Syncer) EnsureMapping(table string, mapping schema.SchemaMapping) (string, error)

EnsureMapping creates a schema mapping for the specified table if it does not exist. It returns the name of the mapping.

func (*Syncer) EnsureTable

func (s *Syncer) EnsureTable(table string, mapping schema.SchemaMapping) error

EnsureTable creates a table with the specified schema mapping if it does not exist.

func (*Syncer) Open

func (s *Syncer) Open(ctx context.Context) error

type Table

type Table struct {
	TableName string `kusto:"TableName"`
}

type TableDetail

type TableDetail struct {
	TableName       string  `kusto:"TableName"`
	HotExtentSize   float64 `kusto:"HotExtentSize"`
	TotalExtentSize float64 `kusto:"TotalExtentSize"`
	TotalExtents    int64   `kusto:"TotalExtents"`
	HotRowCount     int64   `kusto:"HotRowCount"`
	TotalRowCount   int64   `kusto:"TotalRowCount"`
}

type Uploader

type Uploader interface {
	service.Component

	Database() string

	// UploadQueue returns a channel that can be used to upload files to kusto.
	UploadQueue() chan *cluster.Batch

	// Mgmt executes a management query against the database.
	Mgmt(ctx context.Context, query kusto.Statement, options ...kusto.MgmtOption) (*kusto.RowIterator, error)
}

func NewFakeUploader

func NewFakeUploader(db string) Uploader

type UploaderOpts

type UploaderOpts struct {
	StorageDir        string
	Database          string
	ConcurrentUploads int
	Dimensions        []string
	DefaultMapping    adxschema.SchemaMapping
	SampleType        SampleType
}