bitflow

Documentation

Index

• Constants
• func Add(mgr manager.Manager, watchNamespace string) error
• func CompareOneToOneSpec(source *bitflowv1.BitflowSource, pod *corev1.Pod, step *bitflowv1.BitflowStep) bool
• func CompareSingletonSpec(pod *corev1.Pod, step *bitflowv1.BitflowStep) bool
• func CompareSources(required *bitflowv1.BitflowSource, existing *bitflowv1.BitflowSource) bool
• func ConstructReproduciblePodName(stepName, sourceName string) string
• func ConstructSingletonPodName(stepName string) string
• func ConstructSourceName(podName, sourceName string) string
• func CreateOutputSource(step *bitflowv1.BitflowStep, pod *corev1.Pod, out *bitflowv1.StepOutput, ...) *bitflowv1.BitflowSource
• func MergeLabels(sources []*bitflowv1.BitflowSource, newStepName string) map[string]string
• func NewRequeueError(err error) error
• func PatchOneToOnePod(pod *corev1.Pod, step *bitflowv1.BitflowStep, source *bitflowv1.BitflowSource, ...)
• func PatchSingletonPod(pod *corev1.Pod, step *bitflowv1.BitflowStep, ...)
• func ReplyError(c *gin.Context, statusCode int, err error)
• func ReplyJSON(c *gin.Context, obj interface{}, err error)
• type BitflowReconciler
  • func (r *BitflowReconciler) GetAllToOnePod(step *bitflowv1.BitflowStep, matchedSources []*bitflowv1.BitflowSource) ([]*corev1.Pod, error)
  • func (r *BitflowReconciler) GetOneToOnePods(step *bitflowv1.BitflowStep, matchedSources []*bitflowv1.BitflowSource) ([]*corev1.Pod, error)
  • func (r *BitflowReconciler) GetSingletonPod(step *bitflowv1.BitflowStep, matchedSources []*bitflowv1.BitflowSource) ([]*corev1.Pod, error)
  • func (r *BitflowReconciler) IsControlledObject(obj metav1.Object) bool
  • func (r *BitflowReconciler) IsWatchedObject(obj metav1.Object) bool
  • func (r *BitflowReconciler) Reconcile(req reconcile.Request) (reconcile.Result, error)
  • func (r *BitflowReconciler) SetupRestInterface(engine *gin.Engine)
• type ControllerParameters
• type PodCreation
  • func (c *PodCreation) Log(node *corev1.Node) *log.Entry
• type PodOutputPair
• type ReconcileStatisticData
• type ReconcileStatistics
  • func (stat *ReconcileStatistics) ErrorOccurred()
  • func (stat *ReconcileStatistics) GetData() ReconcileStatisticData
  • func (stat *ReconcileStatistics) NodeResourcesReconciled(duration time.Duration)
  • func (stat *ReconcileStatistics) PodRespawned()
  • func (stat *ReconcileStatistics) PodsReconciled(duration time.Duration)
• type RequeueError
  • func (err *RequeueError) Error() string

Constants

const (
	// In addition to these env vars, the variable WATCH_NAMESPACE and POD_NAME are important for the Operator framework
	EnvOperatorName        = "OPERATOR_NAME" // This variable is also used internally by the Kubernetes SDK
	EnvConfigMapName       = "CONFIG_MAP"
	EnvOwnPodIp            = "POD_IP"
	EnvRestApiPort         = "API_LISTEN_PORT"
	EnvConcurrentReconcile = "CONCURRENT_RECONCILE"
	EnvPodIdLabels         = "POD_ID_LABELS"
	EnvRecordStatistics    = "RECORD_STATISTICS"
)

const (
	// TODO make these prefix/suffixes configurable
	STEP_OUTPUT_PREFIX = "output"
	STEP_POD_SUFFIX    = "pod"
	POD_SEPARATOR      = "-"
	SOURCE_SEPARATOR   = "."
)

const (
	PodEnvStepType             = "BITFLOW_STEP_TYPE"
	PodEnvStepName             = "BITFLOW_STEP_NAME"
	PodEnvDataSource           = "BITFLOW_SOURCE"
	PodEnvOneToOneSourceUrl    = "BITFLOW_SINGLE_SOURCE_URL"
	PodEnvOneToOneSourceName   = "BITFLOW_SINGLE_SOURCE_NAME"
	PodEnvOneToOneSourceLabels = "BITFLOW_SINGLE_SOURCE_LABELS"
)

const (
	RestApiDataSourcesPath     = "/dataSources"
	RestApiMatchingSourcesPath = "/matchingSources"
)

const STATE_VALIDATION_FAKE_STEP_NAME = "bitflow-validate-state-identifier-fake-step"

TODO if possible, avoid this hack with the fake step name. It is necessary to execute a regularly recurring reconcile.

Functions

func Add

func Add(mgr manager.Manager, watchNamespace string) error

Add creates a new Bitflow Controller and adds it to the Manager. The Manager will set fields on the Controller and Start it when the Manager is Started.

func CompareOneToOneSpec

func CompareOneToOneSpec(source *bitflowv1.BitflowSource, pod *corev1.Pod, step *bitflowv1.BitflowStep) bool

func CompareSingletonSpec

func CompareSingletonSpec(pod *corev1.Pod, step *bitflowv1.BitflowStep) bool

func CompareSources

func CompareSources(required *bitflowv1.BitflowSource, existing *bitflowv1.BitflowSource) bool

func ConstructReproduciblePodName

func ConstructReproduciblePodName(stepName, sourceName string) string

func ConstructSingletonPodName

func ConstructSingletonPodName(stepName string) string

func ConstructSourceName

func ConstructSourceName(podName, sourceName string) string

func CreateOutputSource

func CreateOutputSource(step *bitflowv1.BitflowStep, pod *corev1.Pod, out *bitflowv1.StepOutput, matchedInputSources []*bitflowv1.BitflowSource, extraLabels map[string]string) *bitflowv1.BitflowSource

func MergeLabels

func MergeLabels(sources []*bitflowv1.BitflowSource, newStepName string) map[string]string

func NewRequeueError

func NewRequeueError(err error) error

func PatchOneToOnePod

func PatchOneToOnePod(pod *corev1.Pod, step *bitflowv1.BitflowStep, source *bitflowv1.BitflowSource, extraLabels, extraEnv map[string]string, apiIP string, apiPort int)

func PatchSingletonPod

func PatchSingletonPod(pod *corev1.Pod, step *bitflowv1.BitflowStep, extraLabels, extraEnvVars map[string]string, apiIP string, apiPort int)

func ReplyError

func ReplyError(c *gin.Context, statusCode int, err error)

func ReplyJSON

func ReplyJSON(c *gin.Context, obj interface{}, err error)

Types

type BitflowReconciler

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

func (*BitflowReconciler) GetAllToOnePod

func (r *BitflowReconciler) GetAllToOnePod(step *bitflowv1.BitflowStep, matchedSources []*bitflowv1.BitflowSource) ([]*corev1.Pod, error)

func (*BitflowReconciler) GetOneToOnePods

func (r *BitflowReconciler) GetOneToOnePods(step *bitflowv1.BitflowStep, matchedSources []*bitflowv1.BitflowSource) ([]*corev1.Pod, error)

func (*BitflowReconciler) GetSingletonPod

func (r *BitflowReconciler) GetSingletonPod(step *bitflowv1.BitflowStep, matchedSources []*bitflowv1.BitflowSource) ([]*corev1.Pod, error)

func (*BitflowReconciler) IsControlledObject

func (r *BitflowReconciler) IsControlledObject(obj metav1.Object) bool

func (*BitflowReconciler) IsWatchedObject

func (r *BitflowReconciler) IsWatchedObject(obj metav1.Object) bool

func (*BitflowReconciler) Reconcile

func (r *BitflowReconciler) Reconcile(req reconcile.Request) (reconcile.Result, error)

Reconcile checks all cluster resources associated with a BitflowStep object and makes sure that the cluster corresponds to the desired state. Note: The Controller will requeue the Request to be processed again if the returned error is non-nil or Result.Requeue is true, otherwise upon completion it will remove the work from the queue.

func (*BitflowReconciler) SetupRestInterface

func (r *BitflowReconciler) SetupRestInterface(engine *gin.Engine)

type ControllerParameters

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

type PodCreation

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

func (*PodCreation) Log

func (c *PodCreation) Log(node *corev1.Node) *log.Entry

type PodOutputPair

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

type ReconcileStatisticData

type ReconcileStatisticData struct {
	PodReconcileLoops                 int
	PodReconcileLoopDuration          onlinestats.Running
	NodeResourceReconcileLoops        int
	NodeResourceReconcileLoopDuration onlinestats.Running
	RestartedPods                     int
	Errors                            int
}

type ReconcileStatistics

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

func (*ReconcileStatistics) ErrorOccurred

func (stat *ReconcileStatistics) ErrorOccurred()

func (*ReconcileStatistics) GetData

func (stat *ReconcileStatistics) GetData() ReconcileStatisticData

func (*ReconcileStatistics) NodeResourcesReconciled

func (stat *ReconcileStatistics) NodeResourcesReconciled(duration time.Duration)

func (*ReconcileStatistics) PodRespawned

func (stat *ReconcileStatistics) PodRespawned()

func (*ReconcileStatistics) PodsReconciled

func (stat *ReconcileStatistics) PodsReconciled(duration time.Duration)

type RequeueError

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

func (*RequeueError) Error

func (err *RequeueError) Error() string