adx

Documentation

Index

• Constants
• func NewDispatcher(uploaders []Uploader) *dispatcher
• func NewUploader(kustoCli ingest.QueryClient, opts UploaderOpts) *uploader
• type DropUnusedTablesTask
  • func NewDropUnusedTablesTask(kustoCli StatementExecutor) *DropUnusedTablesTask
  • func (t *DropUnusedTablesTask) Run(ctx context.Context) error
• type FunctionStore
• type IngestionMapping
• type SampleType
• type StatementExecutor
• type SyncFunctionsTask
  • func NewSyncFunctionsTask(store FunctionStore, 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) EnsureView(ctx context.Context, table string) error
  • func (s *Syncer) Open(ctx context.Context) error
• type Table
• type TableDetail
• type Uploader
  • func NewFakeUploader(db string) Uploader
• type UploaderOpts
• type ViewStore

Constants

const ConcurrentUploads = 50

Functions

func NewDispatcher

func NewDispatcher(uploaders []Uploader) *dispatcher

func NewUploader

func NewUploader(kustoCli ingest.QueryClient, 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 FunctionStore

type FunctionStore interface {
	Functions() []*v1.Function
	UpdateStatus(ctx context.Context, fn *v1.Function) 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 FunctionStore, kustoCli StatementExecutor) *SyncFunctionsTask

func (*SyncFunctionsTask) Run

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

type Syncer

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

func NewSyncer

func NewSyncer(kustoCli mgmt, database string, defaultMapping schema.SchemaMapping, st SampleType, vs ViewStore) *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) EnsureView

func (s *Syncer) EnsureView(ctx context.Context, table string) error

EnsureView will create or update a KQL View for the specified Table if one exists.

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
	ViewStore         ViewStore
}

type ViewStore

type ViewStore interface {
	View(database, table string) (*v1.Function, bool)
	UpdateStatus(ctx context.Context, fn *v1.Function) error
}