pipeline

Documentation

Index

• Constants
• func ConvertRequestToV2Bytes(data []byte, modelName string, modelVersion string) ([]byte, error)
• func ConvertV2ResponseBytesToJson(res []byte) ([]byte, error)
• type ConsumerManager
  • func NewConsumerManager(logger log.FieldLogger, consumerConfig *config.KafkaConfig, ...) *ConsumerManager
  • func (cm *ConsumerManager) GetNumModels() int
  • func (cm *ConsumerManager) Stop()
• type GatewayGrpcServer
  • func NewGatewayGrpcServer(port int, logger log.FieldLogger, gateway PipelineInferer, ...) *GatewayGrpcServer
  • func (g *GatewayGrpcServer) ModelInfer(ctx context.Context, r *v2.ModelInferRequest) (*v2.ModelInferResponse, error)
  • func (g *GatewayGrpcServer) ModelReady(ctx context.Context, req *v2.ModelReadyRequest) (*v2.ModelReadyResponse, error)
  • func (g *GatewayGrpcServer) Start() error
  • func (g *GatewayGrpcServer) Stop()
• type GatewayHttpServer
  • func NewGatewayHttpServer(port int, logger log.FieldLogger, gateway PipelineInferer, ...) *GatewayHttpServer
  • func (g *GatewayHttpServer) Start() error
  • func (g *GatewayHttpServer) Stop() error
• type InferenceRequest
• type InferenceResponse
• type KafkaManager
  • func NewKafkaManager(logger logrus.FieldLogger, namespace string, kafkaConfig *config.KafkaConfig, ...) (*KafkaManager, error)
  • func (km *KafkaManager) Infer(ctx context.Context, resourceName string, isModel bool, data []byte, ...) (*Request, error)
  • func (km *KafkaManager) Stop()
• type MultiTopicsKafkaConsumer
  • func NewMultiTopicsKafkaConsumer(logger log.FieldLogger, consumerConfig *config.KafkaConfig, id string, ...) (*MultiTopicsKafkaConsumer, error)
  • func (c *MultiTopicsKafkaConsumer) AddTopic(topic string, cb kafka.RebalanceCb) error
  • func (c *MultiTopicsKafkaConsumer) Close() error
  • func (c *MultiTopicsKafkaConsumer) GetNumTopics() int
  • func (c *MultiTopicsKafkaConsumer) RemoveTopic(topic string) error
• type NamedTensor
  • func (nt *NamedTensor) MarshalJSON() ([]byte, error)
• type Pipeline
• type PipelineInferer
• type Request
• type RequestOutput
• type TLSDetails
• type TensorData

Constants

const (
	EnvMaxNumConsumers             = "PIPELINEGATEWAY_MAX_NUM_CONSUMERS"
	EnvMaxNumTopicPerConsumer      = "PIPELINEGATEWAY_MAX_NUM_TOPICS_PER_CONSUMER"
	DefaultMaxNumTopicsPerConsumer = 100
	DefaultMaxNumConsumers         = 200
)

const (
	ResourceNameVariable = "model"
)

Functions

func ConvertRequestToV2Bytes

func ConvertRequestToV2Bytes(data []byte, modelName string, modelVersion string) ([]byte, error)

func ConvertV2ResponseBytesToJson

func ConvertV2ResponseBytesToJson(res []byte) ([]byte, error)

Types

type ConsumerManager

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

func NewConsumerManager

func NewConsumerManager(logger log.FieldLogger, consumerConfig *config.KafkaConfig, maxNumTopicsPerConsumer, maxNumConsumers int, tracer trace.Tracer) *ConsumerManager

func (*ConsumerManager) GetNumModels

func (cm *ConsumerManager) GetNumModels() int

func (*ConsumerManager) Stop

func (cm *ConsumerManager) Stop()

type GatewayGrpcServer

type GatewayGrpcServer struct {
	v2.UnimplementedGRPCInferenceServiceServer
	// contains filtered or unexported fields
}

func NewGatewayGrpcServer

func NewGatewayGrpcServer(port int,
	logger log.FieldLogger,
	gateway PipelineInferer,
	metricsHandler metrics.PipelineMetricsHandler,
	tlsOptions *util.TLSOptions,
	piplineReadyChecker status2.PipelineReadyChecker) *GatewayGrpcServer

func (*GatewayGrpcServer) ModelInfer

func (g *GatewayGrpcServer) ModelInfer(ctx context.Context, r *v2.ModelInferRequest) (*v2.ModelInferResponse, error)

func (*GatewayGrpcServer) ModelReady

func (g *GatewayGrpcServer) ModelReady(ctx context.Context, req *v2.ModelReadyRequest) (*v2.ModelReadyResponse, error)

This is presently used for pipeline ready use cases but the v2 protocol only has the concept of model ready calls

func (*GatewayGrpcServer) Start

func (g *GatewayGrpcServer) Start() error

func (*GatewayGrpcServer) Stop

func (g *GatewayGrpcServer) Stop()

type GatewayHttpServer

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

func NewGatewayHttpServer

func NewGatewayHttpServer(port int, logger log.FieldLogger,
	gateway PipelineInferer,
	metrics metrics.PipelineMetricsHandler,
	tlsOptions *util.TLSOptions,
	pipelineReadyChecker status.PipelineReadyChecker) *GatewayHttpServer

func (*GatewayHttpServer) Start

func (g *GatewayHttpServer) Start() error

func (*GatewayHttpServer) Stop

func (g *GatewayHttpServer) Stop() error

type InferenceRequest

type InferenceRequest struct {
	Id         string                 `json:"id,omitempty"`
	Parameters map[string]interface{} `json:"parameters,omitempty"`
	Inputs     []*NamedTensor         `json:"inputs"`
	Outputs    []*RequestOutput       `json:"outputs,omitempty"`
}

type InferenceResponse

type InferenceResponse struct {
	ModelName         string                 `json:"model_name"`
	ModelVersion      string                 `json:"model_version,omitempty"`
	Id                string                 `json:"id,omitempty"`
	Parameters        map[string]interface{} `json:"parameters,omitempty"`
	Outputs           []*NamedTensor         `json:"outputs,omitempty"`
	RawOutputContents [][]byte               `json:"rawOutputContents,omitempty"`
}

type KafkaManager

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

func NewKafkaManager

func NewKafkaManager(logger logrus.FieldLogger, namespace string, kafkaConfig *config.KafkaConfig, traceProvider *seldontracer.TracerProvider, maxNumConsumers, maxNumTopicsPerConsumer int) (*KafkaManager, error)

func (*KafkaManager) Infer

func (km *KafkaManager) Infer(ctx context.Context, resourceName string, isModel bool, data []byte, headers []kafka.Header, requestId string) (*Request, error)

func (*KafkaManager) Stop

func (km *KafkaManager) Stop()

type MultiTopicsKafkaConsumer

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

func NewMultiTopicsKafkaConsumer

func NewMultiTopicsKafkaConsumer(logger log.FieldLogger, consumerConfig *config.KafkaConfig, id string, tracer trace.Tracer) (*MultiTopicsKafkaConsumer, error)

func (*MultiTopicsKafkaConsumer) AddTopic

func (c *MultiTopicsKafkaConsumer) AddTopic(topic string, cb kafka.RebalanceCb) error

func (*MultiTopicsKafkaConsumer) Close

func (c *MultiTopicsKafkaConsumer) Close() error

func (*MultiTopicsKafkaConsumer) GetNumTopics

func (c *MultiTopicsKafkaConsumer) GetNumTopics() int

func (*MultiTopicsKafkaConsumer) RemoveTopic

func (c *MultiTopicsKafkaConsumer) RemoveTopic(topic string) error

type NamedTensor

type NamedTensor struct {
	Name       string                 `json:"name"`
	Shape      []int64                `json:"shape"`
	Datatype   string                 `json:"datatype"`
	Parameters map[string]interface{} `json:"parameters,omitempty"`
	Data       json.RawMessage        `json:"data"`
	// contains filtered or unexported fields
}

func (*NamedTensor) MarshalJSON

func (nt *NamedTensor) MarshalJSON() ([]byte, error)

type Pipeline

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

type PipelineInferer

type PipelineInferer interface {
	Infer(ctx context.Context, resourceName string, isModel bool, data []byte, headers []kafka.Header, requestId string) (*Request, error)
}

type Request

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

type RequestOutput

type RequestOutput struct {
	Name       string                 `json:"name"`
	Parameters map[string]interface{} `json:"parameters"`
}

type TLSDetails

type TLSDetails struct {
	CertMountPath string
	CertFilename  string
	KeyFilename   string
}

type TensorData

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