numaflowtypes

Documentation

Index

• func CheckIfPipelineWontPause(ctx context.Context, pipeline *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 GetPipelineDesiredPhase(pipeline *unstructured.Unstructured) (string, error)
• func GetPipelineISBSVCName(pipeline *unstructured.Unstructured) (string, 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 MonoVertexWithoutReplicas(monoVertex *unstructured.Unstructured) (map[string]interface{}, error)
• func PipelineWithDesiredPhase(pipeline *unstructured.Unstructured, phase string) error
• func PipelineWithISBServiceName(pipeline *unstructured.Unstructured, isbsvcName string) error
• func PipelineWithoutDesiredPhase(pipeline *unstructured.Unstructured)
• 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)

Functions

func CheckIfPipelineWontPause

func CheckIfPipelineWontPause(ctx context.Context, pipeline *unstructured.Unstructured, pipelineRollout *apiv1.PipelineRollout) 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 GetPipelineDesiredPhase

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

func GetPipelineISBSVCName

func GetPipelineISBSVCName(pipeline *unstructured.Unstructured) (string, 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 MonoVertexWithoutReplicas

func MonoVertexWithoutReplicas(monoVertex *unstructured.Unstructured) (map[string]interface{}, error)

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

Types

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"`
}

PipelineSpec keeps track of minimum number of fields we need to know about

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"`
}

func ParsePipelineStatus

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