Chainer

Learn about the internal Chainer API in Seldon Core, including model chaining, tensor mapping, and pipeline configuration.

This API is for communication between the Seldon Scheduler and the Seldon dataflow component which manages pipelines.

Proto Definition

https://github.com/SeldonIO/seldon-core/blob/v2/apis/mlops/chainer/chainer.proto

syntax = "proto3";

package seldon.mlops.chainer;

option go_package = "github.com/seldonio/seldon-core/apis/go/v2/mlops/chainer";
option java_package = "io.seldon.mlops.chainer";

message PipelineSubscriptionRequest {
  string name = 1;
}

message PipelineUpdateMessage {
  enum PipelineOperation {
    Unknown = 0;
    Create = 1;
    Delete = 2;
  }
  PipelineOperation op = 1;
  string pipeline = 2;
  uint32 version = 3;
  string uid = 4;
  repeated PipelineStepUpdate updates = 5;
  string pipelineOutputTopic = 6;
  string pipelineErrorTopic = 7;
  bool allowCycles = 8;
  uint32 maxStepRevisits = 9;
  uint64 timestamp = 10;
  string stream = 11;
}

message PipelineStepUpdate {
  enum PipelineJoinType {
    Unknown = 0;
    Inner = 1;
    Outer = 2;
    Any = 3;
  }
  // https://docs.google.com/document/d/1tX-uaOvngx1RpEyWEZ4EbEcU8D0OgYuRWVb2UAi85n4/edit
  // Pipeline Resource example, e.g. transform.outputs.traffic
  //    seldon.<namespace>.<model name>.<inputs|outputs>.<tensor name>
  repeated PipelineTopic sources = 1;
  repeated PipelineTopic triggers = 2;
  PipelineTopic sink = 3;
  PipelineJoinType inputJoinTy = 4;
  PipelineJoinType triggersJoinTy = 5;
  bool passEmptyResponses = 6; // Forward empty response to following steps, default false
  optional uint32 joinWindowMs = 7; // Join window millisecs, some nozero default (TBD)
  repeated PipelineTensorMapping tensorMap = 8; // optional list of tensor name mappings
  Batch batch = 9; // Batch settings
}

message PipelineTensorMapping {
  string pipelineName = 1;
  string topicAndTensor = 2;
  string tensorName = 3;
}

message PipelineTopic {
  string pipelineName = 1;
  string topicName = 2;
  optional string tensor = 3;
}

message Batch {
  optional uint32 size = 1;
  optional uint32 windowMs = 2;
  bool rolling = 3;
}

message PipelineUpdateStatusMessage {
  // TODO - include `name` to identify transformer message comes from
  PipelineUpdateMessage update = 1;
  bool success = 2;
  string reason = 3;
}

message PipelineUpdateStatusResponse {
}

service Chainer {
  rpc SubscribePipelineUpdates(PipelineSubscriptionRequest) returns (stream PipelineUpdateMessage) {};
  rpc PipelineUpdateEvent(PipelineUpdateStatusMessage) returns (PipelineUpdateStatusResponse) {};
}