numaflowtypes

Documentation

Index

• func CheckIfPipelineWontPause(ctx context.Context, pipeline *unstructured.Unstructured, ...) bool
• func CheckMonoVertexPhase(ctx context.Context, monovertex *unstructured.Unstructured, ...) bool
• func CheckPipelineDrained(ctx context.Context, pipeline *unstructured.Unstructured) (bool, error)
• func CheckPipelinePhase(ctx context.Context, pipeline *unstructured.Unstructured, ...) bool
• func GetISBServiceChildResourceHealth(conditions []metav1.Condition) (metav1.ConditionStatus, string)
• func GetISBSvcStatefulSetFromK8s(ctx context.Context, c client.Client, isbsvc *unstructured.Unstructured, ...) (*appsv1.StatefulSet, error)
• func GetMonoVertexDesiredPhase(monovertex *unstructured.Unstructured) (string, error)
• func GetPipelineDesiredPhase(pipeline *unstructured.Unstructured) (string, error)
• func GetPipelineISBSVCName(pipeline *unstructured.Unstructured) (string, error)
• func GetPipelineVertexDefinitions(pipeline *unstructured.Unstructured) ([]interface{}, error)
• func GetPipelineVertices(ctx context.Context, c client.Client, pipeline *unstructured.Unstructured) (map[string]*unstructured.Unstructured, error)
• func GetPipelinesForRollout(ctx context.Context, c client.Client, pipelineRollout *apiv1.PipelineRollout, ...) (unstructured.UnstructuredList, error)
• func GetRolloutForPipeline(ctx context.Context, c client.Client, pipeline *unstructured.Unstructured) (*apiv1.PipelineRollout, error)
• func IsPipelinePausedOrWontPause(ctx context.Context, pipeline *unstructured.Unstructured, ...) (bool, error)
• func MinimizePipelineVertexReplicas(ctx context.Context, c client.Client, pipeline *unstructured.Unstructured) error
• func PipelineWithDesiredPhase(pipeline *unstructured.Unstructured, phase string) error
• func PipelineWithISBServiceName(pipeline *unstructured.Unstructured, isbsvcName string) error
• func PipelineWithoutDesiredPhase(pipeline *unstructured.Unstructured)
• func PipelineWithoutScaleMinMax(pipeline *unstructured.Unstructured) error
• type AbstractVertex
• type Lifecycle
• type MonoVertexStatus
  • func ParseMonoVertexStatus(monoVertex *unstructured.Unstructured) (MonoVertexStatus, error)
• type PipelineSpec
  • func (pipeline PipelineSpec) GetISBSvcName() string
• type PipelineStatus
  • func ParsePipelineStatus(pipeline *unstructured.Unstructured) (PipelineStatus, error)
• type Scale

Functions

func CheckIfPipelineWontPause

func CheckIfPipelineWontPause(ctx context.Context, pipeline *unstructured.Unstructured, pipelineRollout *apiv1.PipelineRollout) bool

func CheckMonoVertexPhase

func CheckMonoVertexPhase(ctx context.Context, monovertex *unstructured.Unstructured, phase numaflowv1.PipelinePhase) bool

func CheckPipelineDrained

func CheckPipelineDrained(ctx context.Context, pipeline *unstructured.Unstructured) (bool, error)

func CheckPipelinePhase

func CheckPipelinePhase(ctx context.Context, pipeline *unstructured.Unstructured, phase numaflowv1.PipelinePhase) bool

func GetISBServiceChildResourceHealth

func GetISBServiceChildResourceHealth(conditions []metav1.Condition) (metav1.ConditionStatus, string)

func GetISBSvcStatefulSetFromK8s

func GetISBSvcStatefulSetFromK8s(ctx context.Context, c client.Client, isbsvc *unstructured.Unstructured, checkLive bool) (*appsv1.StatefulSet, error)

Each ISBService has one underlying StatefulSet Find it Depending on value "checkLive", either check K8S API directly or go to informer cache

func GetMonoVertexDesiredPhase

func GetMonoVertexDesiredPhase(monovertex *unstructured.Unstructured) (string, error)

func GetPipelineDesiredPhase

func GetPipelineDesiredPhase(pipeline *unstructured.Unstructured) (string, error)

func GetPipelineISBSVCName

func GetPipelineISBSVCName(pipeline *unstructured.Unstructured) (string, error)

func GetPipelineVertexDefinitions

func GetPipelineVertexDefinitions(pipeline *unstructured.Unstructured) ([]interface{}, error)

func GetPipelineVertices

func GetPipelineVertices(ctx context.Context, c client.Client, pipeline *unstructured.Unstructured) (map[string]*unstructured.Unstructured, error)

find all the Pipeline Vertices in K8S using the Pipeline's definition: return a map of vertex name to resource found for any Vertices that can't be found, return an entry mapped to nil

func GetPipelinesForRollout

func GetPipelinesForRollout(ctx context.Context, c client.Client, pipelineRollout *apiv1.PipelineRollout, live bool) (unstructured.UnstructuredList, error)

func GetRolloutForPipeline

func GetRolloutForPipeline(ctx context.Context, c client.Client, pipeline *unstructured.Unstructured) (*apiv1.PipelineRollout, error)

func IsPipelinePausedOrWontPause

func IsPipelinePausedOrWontPause(ctx context.Context, pipeline *unstructured.Unstructured, pipelineRollout *apiv1.PipelineRollout, requireDrained bool) (bool, error)

either pipeline must be:

• Paused
• Failed (contract with Numaflow is that unpausible Pipelines are "Failed" pipelines)
• PipelineRollout parent Annotated to allow data loss

func MinimizePipelineVertexReplicas

func MinimizePipelineVertexReplicas(ctx context.Context, c client.Client, pipeline *unstructured.Unstructured) error

MinimizePipelineVertexReplicas clears out the `replicas` field from each Vertex of a Pipeline, which has the effect in Numaflow of resetting to "scale.min" value

func PipelineWithDesiredPhase

func PipelineWithDesiredPhase(pipeline *unstructured.Unstructured, phase string) error

func PipelineWithISBServiceName

func PipelineWithISBServiceName(pipeline *unstructured.Unstructured, isbsvcName string) error

func PipelineWithoutDesiredPhase

func PipelineWithoutDesiredPhase(pipeline *unstructured.Unstructured)

remove 'lifecycle.desiredPhase' key/value pair from spec also remove 'lifecycle' if it's an empty map

func PipelineWithoutScaleMinMax

func PipelineWithoutScaleMinMax(pipeline *unstructured.Unstructured) error

Types

type AbstractVertex

type AbstractVertex struct {
	Name  string `json:"name"`
	Scale Scale  `json:"scale,omitempty"`
}

AbstractVertex keeps track of minimum number of fields we need to know about in Numaflow's AbstractVertex, which are presumed not to change from version to version

type Lifecycle

type Lifecycle struct {
	// DesiredPhase used to bring the pipeline from current phase to desired phase
	// +kubebuilder:default=Running
	// +optional
	DesiredPhase string `json:"desiredPhase,omitempty"`
}

type MonoVertexStatus

type MonoVertexStatus = kubernetes.GenericStatus

func ParseMonoVertexStatus

func ParseMonoVertexStatus(monoVertex *unstructured.Unstructured) (MonoVertexStatus, error)

type PipelineSpec

type PipelineSpec struct {
	InterStepBufferServiceName string           `json:"interStepBufferServiceName"`
	Lifecycle                  Lifecycle        `json:"lifecycle,omitempty"`
	Vertices                   []AbstractVertex `json:"vertices,omitempty"`
}

PipelineSpec keeps track of minimum number of fields we need to know about in Numaflow's PipelineSpec, which are presumed not to change from version to version

func (PipelineSpec) GetISBSvcName

func (pipeline PipelineSpec) GetISBSvcName() string

type PipelineStatus

type PipelineStatus struct {
	Phase              numaflowv1.PipelinePhase `json:"phase,omitempty"`
	Conditions         []metav1.Condition       `json:"conditions,omitempty"`
	ObservedGeneration int64                    `json:"observedGeneration,omitempty"`
	DrainedOnPause     bool                     `json:"drainedOnPause,omitempty" protobuf:"bytes,12,opt,name=drainedOnPause"`
}

PipelineStatus keeps track of minimum number of fields we need to know about in Numaflow's PipelineStatus, which are presumed not to change from version to version

func ParsePipelineStatus

func ParsePipelineStatus(pipeline *unstructured.Unstructured) (PipelineStatus, error)

type Scale

type Scale struct {
	// Minimum replicas.
	Min *int32 `json:"min,omitempty"`
	// Maximum replicas.
	Max *int32 `json:"max,omitempty"`
}

Scale keeps track of minimum number of fields we need to know about in Numaflow's Scale struct, which are presumed not to change from version to version