flink

package
v0.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 31, 2019 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	JobManagerDefaultReplicaCount = 1
	TaskManagerDefaultSlots       = 16
	RPCDefaultPort                = 6123
	QueryDefaultPort              = 6124
	BlobDefaultPort               = 6125
	UIDefaultPort                 = 8081
	MetricsQueryDefaultPort       = 50101
	OffHeapMemoryDefaultFraction  = 0.5
)
View Source 
const (
	AppName                          = "APP_NAME"
	AwsMetadataServiceTimeoutKey     = "AWS_METADATA_SERVICE_TIMEOUT"
	AwsMetadataServiceNumAttemptsKey = "AWS_METADATA_SERVICE_NUM_ATTEMPTS"
	AwsMetadataServiceTimeout        = "5"
	AwsMetadataServiceNumAttempts    = "20"
	OperatorFlinkConfig              = "OPERATOR_FLINK_CONFIG"
	FlinkDeploymentType              = "flink-deployment-type"
	FlinkDeploymentTypeJobmanager    = "jobmanager"
	FlinkDeploymentTypeTaskmanager   = "taskmanager"
	FlinkAppHash                     = "flink-app-hash"
	FlinkJobProperties               = "flink-job-properties"
	RestartNonce                     = "restart-nonce"
)
View Source 
const (
	JobManagerNameFormat                = "%s-%s-jm"
	JobManagerPodNameFormat             = "%s-%s-jm-pod"
	JobManagerContainerName             = "jobmanager"
	JobManagerArg                       = "jobmanager"
	JobManagerReadinessPath             = "/config"
	JobManagerReadinessInitialDelaySec  = 10
	JobManagerReadinessTimeoutSec       = 1
	JobManagerReadinessSuccessThreshold = 1
	JobManagerReadinessFailureThreshold = 2
	JobManagerReadinessPeriodSec        = 5
)
View Source 
const (
	FlinkRPCPortName            = "rpc"
	FlinkQueryPortName          = "query"
	FlinkBlobPortName           = "blob"
	FlinkUIPortName             = "ui"
	FlinkInternalMetricPortName = "metrics"
)
View Source 
const (
	TaskManagerNameFormat     = "%s-%s-tm"
	TaskManagerPodNameFormat  = "%s-%s-tm-pod"
	TaskManagerContainerName  = "taskmanager"
	TaskManagerArg            = "taskmanager"
	TaskManagerHostnameEnvVar = "TASKMANAGER_HOSTNAME"
)

Variables

View Source 
var JobManagerDefaultResources = coreV1.ResourceRequirements{
	Requests: coreV1.ResourceList{
		coreV1.ResourceCPU:    resource.MustParse("4"),
		coreV1.ResourceMemory: resource.MustParse("3072Mi"),
	},
	Limits: coreV1.ResourceList{
		coreV1.ResourceCPU:    resource.MustParse("4"),
		coreV1.ResourceMemory: resource.MustParse("3072Mi"),
	},
}
View Source 
var TaskManagerDefaultResources = coreV1.ResourceRequirements{
	Requests: coreV1.ResourceList{
		coreV1.ResourceCPU:    resource.MustParse("2"),
		coreV1.ResourceMemory: resource.MustParse("1024Mi"),
	},
	Limits: coreV1.ResourceList{
		coreV1.ResourceCPU:    resource.MustParse("2"),
		coreV1.ResourceMemory: resource.MustParse("1024Mi"),
	},
}

Functions

func DeploymentIsJobmanager 

func DeploymentIsJobmanager(deployment *v1.Deployment) bool

func DeploymentIsTaskmanager 

func DeploymentIsTaskmanager(deployment *v1.Deployment) bool

func DeploymentsEqual 

func DeploymentsEqual(a *appsv1.Deployment, b *appsv1.Deployment) bool

Returns true if there are no relevant differences between the deployments. This should be used only to determine that two deployments correspond to the same FlinkApplication, not as a general notion of equality.

func FetchJobManagerContainerObj 

func FetchJobManagerContainerObj(application *v1alpha1.FlinkApplication) *coreV1.Container

func FetchJobManagerIngressCreateObj 

func FetchJobManagerIngressCreateObj(app *v1alpha1.FlinkApplication) *v1beta1.Ingress

func FetchJobManagerServiceCreateObj 

func FetchJobManagerServiceCreateObj(app *v1alpha1.FlinkApplication, hash string) *coreV1.Service

func FetchJobMangerDeploymentCreateObj 

func FetchJobMangerDeploymentCreateObj(app *v1alpha1.FlinkApplication, hash string) *v1.Deployment

func FetchJobMangerDeploymentDeleteObj 

func FetchJobMangerDeploymentDeleteObj(app *v1alpha1.FlinkApplication, hash string) *v1.Deployment

func FetchTaskManagerContainerObj 

func FetchTaskManagerContainerObj(application *v1alpha1.FlinkApplication) *coreV1.Container

func FetchTaskMangerDeploymentCreateObj 

func FetchTaskMangerDeploymentCreateObj(app *v1alpha1.FlinkApplication, hash string) *v1.Deployment

func FetchTaskMangerDeploymentDeleteObj 

func FetchTaskMangerDeploymentDeleteObj(app *v1alpha1.FlinkApplication, hash string) *v1.Deployment

func FetchVersionedJobManagerServiceDeleteObj 

func FetchVersionedJobManagerServiceDeleteObj(app *v1alpha1.FlinkApplication, hash string) *coreV1.Service

func GetAWSServiceEnv 

func GetAWSServiceEnv() []v1.EnvVar

func GetActiveFlinkJob 

func GetActiveFlinkJob(jobs []client.FlinkJob) *client.FlinkJob

func GetAppHashSelector 

func GetAppHashSelector(app *v1alpha1.FlinkApplication) map[string]string

func GetAppHashSelectorWithHash 

func GetAppHashSelectorWithHash(hash string) map[string]string

func GetFlinkContainerEnv 

func GetFlinkContainerEnv(app *v1alpha1.FlinkApplication) []v1.EnvVar

func GetFlinkUIIngressURL 

func GetFlinkUIIngressURL(jobName string) string

func GetTaskManagerPorts 

func GetTaskManagerPorts(app *v1alpha1.FlinkApplication) []coreV1.ContainerPort

func HashForApplication 

func HashForApplication(app *v1alpha1.FlinkApplication) string

Returns an 8 character hash sensitive to the application name, labels, annotations, and spec. TODO: we may need to add collision-avoidance to this

func ImagePullPolicy 

func ImagePullPolicy(app *v1alpha1.FlinkApplication) v1.PullPolicy

func InjectHashesIntoConfig 

func InjectHashesIntoConfig(deployment *appsv1.Deployment, app *v1alpha1.FlinkApplication, hash string)

func JobManagerDeploymentMatches 

func JobManagerDeploymentMatches(deployment *v1.Deployment, application *v1alpha1.FlinkApplication) bool

func ReplaceJobURL 

func ReplaceJobURL(value string, input string) string

func TaskManagerDeploymentMatches 

func TaskManagerDeploymentMatches(deployment *v1.Deployment, application *v1alpha1.FlinkApplication) bool

func VersionedJobManagerService 

func VersionedJobManagerService(app *v1alpha1.FlinkApplication, hash string) string

Types

type Controller 

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

func (*Controller) CancelWithSavepoint 

func (f *Controller) CancelWithSavepoint(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (string, error)

func (*Controller) CompareAndUpdateClusterStatus 

func (f *Controller) CompareAndUpdateClusterStatus(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (bool, error)

Gets and updates the cluster status

func (*Controller) CompareAndUpdateJobStatus 

func (f *Controller) CompareAndUpdateJobStatus(ctx context.Context, app *v1alpha1.FlinkApplication, hash string) (bool, error)

func (*Controller) CreateCluster 

func (f *Controller) CreateCluster(ctx context.Context, application *v1alpha1.FlinkApplication) error

func (*Controller) DeleteCluster 

func (f *Controller) DeleteCluster(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) error

func (*Controller) FindExternalizedCheckpoint 

func (f *Controller) FindExternalizedCheckpoint(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (string, error)

func (*Controller) ForceCancel 

func (f *Controller) ForceCancel(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) error

func (*Controller) GetCurrentAndOldDeploymentsForApp 

func (f *Controller) GetCurrentAndOldDeploymentsForApp(ctx context.Context,
	application *v1alpha1.FlinkApplication) (*common.FlinkDeployment, []common.FlinkDeployment, error)

Gets the current deployment and any other deployments for the application. The current deployment will be the one that matches the FlinkApplication, unless the FailedDeployHash is set, in which case it will be the one with that hash.

func (*Controller) GetJobsForApplication 

func (f *Controller) GetJobsForApplication(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) ([]client.FlinkJob, error)

func (*Controller) GetSavepointStatus 

func (f *Controller) GetSavepointStatus(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (*client.SavepointResponse, error)

func (*Controller) IsClusterReady 

func (f *Controller) IsClusterReady(ctx context.Context, application *v1alpha1.FlinkApplication) (bool, error)

func (*Controller) IsServiceReady 

func (f *Controller) IsServiceReady(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (bool, error)

func (*Controller) LogEvent 

func (f *Controller) LogEvent(ctx context.Context, app *v1alpha1.FlinkApplication, fieldPath string, eventType string, message string)

func (*Controller) StartFlinkJob 

func (f *Controller) StartFlinkJob(ctx context.Context, application *v1alpha1.FlinkApplication, hash string,
	jarName string, parallelism int32, entryClass string, programArgs string) (string, error)

type ControllerInterface 

type ControllerInterface interface {
	// Creates a Flink cluster with necessary Job Manager, Task Managers and services for UI
	CreateCluster(ctx context.Context, application *v1alpha1.FlinkApplication) error

	// Deletes a Flink cluster based on the hash
	DeleteCluster(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) error

	// Cancels the running/active jobs in the Cluster for the Application after savepoint is created
	CancelWithSavepoint(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (string, error)

	// Force cancels the running/active job without taking a savepoint
	ForceCancel(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) error

	// Starts the Job in the Flink Cluster
	StartFlinkJob(ctx context.Context, application *v1alpha1.FlinkApplication, hash string,
		jarName string, parallelism int32, entryClass string, programArgs string) (string, error)

	// Savepoint creation is asynchronous.
	// Polls the status of the Savepoint, using the triggerID
	GetSavepointStatus(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (*client.SavepointResponse, error)

	// Check if the Flink Kubernetes Cluster is Ready.
	// Checks if all the pods of task and job managers are ready.
	IsClusterReady(ctx context.Context, application *v1alpha1.FlinkApplication) (bool, error)

	// Checks to see if the Flink Cluster is ready to handle API requests
	IsServiceReady(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (bool, error)

	// Returns the list of Jobs running on the Flink Cluster for the Application
	GetJobsForApplication(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) ([]client.FlinkJob, error)

	// For the application, a deployment corresponds to an image. This returns the current and older deployments for the app.
	GetCurrentAndOldDeploymentsForApp(ctx context.Context, application *v1alpha1.FlinkApplication) (*common.FlinkDeployment, []common.FlinkDeployment, error)

	// Attempts to find an externalized checkpoint for the job. This can be used to recover an application that is not
	// able to savepoint for some reason.
	FindExternalizedCheckpoint(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (string, error)

	// Logs an event to the FlinkApplication resource and to the operator log
	LogEvent(ctx context.Context, app *v1alpha1.FlinkApplication, fieldPath string, eventType string, message string)

	// Compares and updates new cluster status with current cluster status
	// Returns true if there is a change in ClusterStatus
	CompareAndUpdateClusterStatus(ctx context.Context, application *v1alpha1.FlinkApplication, hash string) (bool, error)

	// Compares and updates new job status with current job status
	// Returns true if there is a change in JobStatus
	CompareAndUpdateJobStatus(ctx context.Context, app *v1alpha1.FlinkApplication, hash string) (bool, error)
}

Interface to manage Flink Application in Kubernetes

func NewController 

func NewController(k8sCluster k8.ClusterInterface, config config.RuntimeConfig) ControllerInterface

type JobManagerController 

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

func (*JobManagerController) CreateIfNotExist 

func (j *JobManagerController) CreateIfNotExist(ctx context.Context, application *v1alpha1.FlinkApplication) (bool, error)

type JobManagerControllerInterface 

type JobManagerControllerInterface interface {
	CreateIfNotExist(ctx context.Context, application *v1alpha1.FlinkApplication) (bool, error)
}

func NewJobManagerController 

func NewJobManagerController(k8sCluster k8.ClusterInterface, config config.RuntimeConfig) JobManagerControllerInterface

type TaskManagerController 

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

func (*TaskManagerController) CreateIfNotExist 

func (t *TaskManagerController) CreateIfNotExist(ctx context.Context, application *v1alpha1.FlinkApplication) (bool, error)

type TaskManagerControllerInterface 

type TaskManagerControllerInterface interface {
	CreateIfNotExist(ctx context.Context, application *v1alpha1.FlinkApplication) (bool, error)
}

func NewTaskManagerController 

func NewTaskManagerController(k8sCluster k8.ClusterInterface, config config.RuntimeConfig) TaskManagerControllerInterface

Source Files

View all Source files

Directories

Path Synopsis
mock

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.