request

Documentation

Index

• type AstraBundleConfig
• type AstraCqlStoreConfig
  • func NewAstraCqlStoreConfig(logger klog.Logger, config *AstraBundleConfig) *AstraCqlStoreConfig
• type Buffer
• type BufferConfig
• type BufferInput
• type BufferOutput
• type CheckpointStore
• type CqlStore
  • func NewAstraCqlStore(logger klog.Logger, bundle *AstraBundleConfig) *CqlStore
  • func NewCqlStore(cluster *gocql.ClusterConfig, logger klog.Logger) *CqlStore
  • func (cqls *CqlStore) ReadCheckpoint(algorithm string, id string) (*models.CheckpointedRequest, error)
  • func (cqls *CqlStore) ReadCheckpoints(requestTag string) ([]models.CheckpointedRequest, error)
  • func (cqls *CqlStore) ReadMetadata(checkpoint *models.CheckpointedRequest) (*models.SubmissionBufferEntry, error)
  • func (cqls *CqlStore) UpsertCheckpoint(checkpoint *models.CheckpointedRequest) error
  • func (cqls *CqlStore) UpsertMetadata(entry *models.SubmissionBufferEntry) error
• type DefaultBuffer
  • func NewDefaultBuffer(ctx context.Context, config *BufferConfig, astraConfig *AstraBundleConfig) *DefaultBuffer
  • func (buffer *DefaultBuffer) Add(requestId string, algorithmName string, request *models.AlgorithmRequest, ...) error
  • func (buffer *DefaultBuffer) Start(submitter pipeline.StageActor[*BufferOutput, types.UID])
• type MetadataStore

Types

type AstraBundleConfig

type AstraBundleConfig struct {
	SecureConnectionBundleBase64 string `mapstructure:"secure-connection-bundle-base64"`
	GatewayUser                  string `mapstructure:"gateway-user"`
	GatewayPassword              string `mapstructure:"gateway-password"`
}

type AstraCqlStoreConfig

type AstraCqlStoreConfig struct {
	GatewayHost string
	GatewayPort string
	GatewayUser string
	GatewayPass string
	TlsConfig   *tls.Config
}

func NewAstraCqlStoreConfig

func NewAstraCqlStoreConfig(logger klog.Logger, config *AstraBundleConfig) *AstraCqlStoreConfig

type Buffer

type Buffer interface {
	BufferRequest(ctx context.Context, requestId string, request *models.AlgorithmRequest, config *v1.MachineLearningAlgorithmSpec) (*models.SubmissionBufferEntry, error)
}

type BufferConfig

type BufferConfig struct {
	PayloadStoragePath         string        `mapstructure:"payload-storage-path"`
	PayloadValidFor            time.Duration `mapstructure:"payload-valid-for"`
	FailureRateBaseDelay       time.Duration `mapstructure:"failure-rate-base-delay"`
	FailureRateMaxDelay        time.Duration `mapstructure:"failure-rate-max-delay"`
	RateLimitElementsPerSecond int           `mapstructure:"rate-limit-elements-per-second"`
	RateLimitElementsBurst     int           `mapstructure:"rate-limit-elements-burst"`
	Workers                    int           `mapstructure:"workers"`
}

type BufferInput

type BufferInput struct {
	Checkpoint        *models.CheckpointedRequest
	SerializedPayload []byte
	Config            *v1.MachineLearningAlgorithmSpec
}

type BufferOutput

type BufferOutput struct {
	Checkpoint *models.CheckpointedRequest
	Entry      *models.SubmissionBufferEntry
}

type CheckpointStore

type CheckpointStore interface {
	UpsertCheckpoint(checkpoint *models.CheckpointedRequest) error
	ReadCheckpoint(algorithm string, id string) (*models.CheckpointedRequest, error)
	ReadCheckpoints(requestTag string) ([]models.CheckpointedRequest, error)
}

type CqlStore

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

func NewAstraCqlStore

func NewAstraCqlStore(logger klog.Logger, bundle *AstraBundleConfig) *CqlStore

NewAstraCqlStore creates a CqlStore connected to DataStax AstraDB serverless instance

func NewCqlStore

func NewCqlStore(cluster *gocql.ClusterConfig, logger klog.Logger) *CqlStore

NewCqlStore creates a generic connected CqlStore (Apache Cassandra/Scylla)

func (*CqlStore) ReadCheckpoint

func (cqls *CqlStore) ReadCheckpoint(algorithm string, id string) (*models.CheckpointedRequest, error)

func (*CqlStore) ReadCheckpoints

func (cqls *CqlStore) ReadCheckpoints(requestTag string) ([]models.CheckpointedRequest, error)

func (*CqlStore) ReadMetadata

func (cqls *CqlStore) ReadMetadata(checkpoint *models.CheckpointedRequest) (*models.SubmissionBufferEntry, error)

func (*CqlStore) UpsertCheckpoint

func (cqls *CqlStore) UpsertCheckpoint(checkpoint *models.CheckpointedRequest) error

func (*CqlStore) UpsertMetadata

func (cqls *CqlStore) UpsertMetadata(entry *models.SubmissionBufferEntry) error

type DefaultBuffer

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

func NewDefaultBuffer

func NewDefaultBuffer(ctx context.Context, config *BufferConfig, astraConfig *AstraBundleConfig) *DefaultBuffer

NewDefaultBuffer creates a default buffer that uses Astra DB for checkpointing and S3-compatible storage for payload persistence

func (*DefaultBuffer) Add

func (buffer *DefaultBuffer) Add(requestId string, algorithmName string, request *models.AlgorithmRequest, config *v1.MachineLearningAlgorithmSpec) error

func (*DefaultBuffer) Start

func (buffer *DefaultBuffer) Start(submitter pipeline.StageActor[*BufferOutput, types.UID])

type MetadataStore

type MetadataStore interface {
	UpsertMetadata(entry *models.SubmissionBufferEntry) error
	ReadMetadata(checkpoint *models.CheckpointedRequest) (*models.SubmissionBufferEntry, error)
}