util

Documentation

Index

• func BuildConfig(kubeconfig string) (*rest.Config, error)
• func EnrichSpecWithLogger(logger *log.Entry, cluster spec.Kafkacluster) *log.Entry
• func GetBrokerAdressess(cluster spec.Kafkacluster) []string
• func NewKubeClient(kubeCfgFile string) (*k8sclient.Clientset, error)
• func ReflectOptionsStruct(v interface{}) []v1.EnvVar
• type ClientUtil
  • func New(kubeConfigFile, masterHost string) (*ClientUtil, error)
  • func (c *ClientUtil) BrokerStatefulSetExist(cluster spec.Kafkacluster) bool
  • func (c *ClientUtil) CheckIfAnyEndpointIsReady(serviceName string, namespace string) bool
  • func (c *ClientUtil) CleanupKafkaCluster(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) CreateBrokerService(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) CreateBrokerStatefulSet(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) CreateDirectBrokerService(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) CreatePersistentVolumes(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) CreatePersistentVolumesTODODELETE(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) CreateStorage(cluster spec.KafkaclusterSpec)
  • func (c *ClientUtil) DeleteKafkaCluster(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) DeleteOffsetMonitor(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) DeployOffsetMonitor(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) GenerateExporterDeployment(cluster spec.Kafkacluster) *appsv1Beta1.Deployment
  • func (c *ClientUtil) GenerateHeadlessService(cluster spec.Kafkacluster) *v1.Service
  • func (c *ClientUtil) GenerateKafkaOptions(cluster spec.Kafkacluster) []v1.EnvVar
  • func (c *ClientUtil) GetBrokerStates(cluster spec.Kafkacluster) ([]string, error)
  • func (c *ClientUtil) GetBrokersWithState(cluster spec.Kafkacluster, state spec.KafkaBrokerState) (int32, error)
  • func (c *ClientUtil) GetMaxHeap(cluster spec.Kafkacluster) *resource.Quantity
  • func (c *ClientUtil) GetMaxHeapJavaString(cluster spec.Kafkacluster) string
  • func (c *ClientUtil) GetPodAnnotations(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) GetReadyEndpoints(serviceName string, namespace string) []string
  • func (c *ClientUtil) SetBrokerState(cluster spec.Kafkacluster, brokerId int32, state spec.KafkaBrokerState) error
  • func (c *ClientUtil) UpdateBrokerImage(cluster spec.Kafkacluster) error
  • func (c *ClientUtil) UpsizeBrokerStS(cluster spec.Kafkacluster) error

Functions

func BuildConfig

func BuildConfig(kubeconfig string) (*rest.Config, error)

func EnrichSpecWithLogger

func EnrichSpecWithLogger(logger *log.Entry, cluster spec.Kafkacluster) *log.Entry

func GetBrokerAdressess

func GetBrokerAdressess(cluster spec.Kafkacluster) []string

func NewKubeClient

func NewKubeClient(kubeCfgFile string) (*k8sclient.Clientset, error)

TODO refactor for config *rest.Config :)

func ReflectOptionsStruct

func ReflectOptionsStruct(v interface{}) []v1.EnvVar

Types

type ClientUtil

type ClientUtil struct {
	KubernetesClient *k8sclient.Clientset
	MasterHost       string
	DefaultOption    metav1.GetOptions
}

func New

func New(kubeConfigFile, masterHost string) (*ClientUtil, error)

func (*ClientUtil) BrokerStatefulSetExist

func (c *ClientUtil) BrokerStatefulSetExist(cluster spec.Kafkacluster) bool

TODO caching of the STS

func (*ClientUtil) CheckIfAnyEndpointIsReady

func (c *ClientUtil) CheckIfAnyEndpointIsReady(serviceName string, namespace string) bool

TODO check if client already has function

func (*ClientUtil) CleanupKafkaCluster

func (c *ClientUtil) CleanupKafkaCluster(cluster spec.Kafkacluster) error

func (*ClientUtil) CreateBrokerService

func (c *ClientUtil) CreateBrokerService(cluster spec.Kafkacluster) error

TODO refactor, into headless svc and direct svc

func (*ClientUtil) CreateBrokerStatefulSet

func (c *ClientUtil) CreateBrokerStatefulSet(cluster spec.Kafkacluster) error

func (*ClientUtil) CreateDirectBrokerService

func (c *ClientUtil) CreateDirectBrokerService(cluster spec.Kafkacluster) error

func (*ClientUtil) CreatePersistentVolumes

func (c *ClientUtil) CreatePersistentVolumes(cluster spec.Kafkacluster) error

Creates PV if no dynamicProvisioner is aviable for that class HostPath with statefulset need manuel creation of PV

func (*ClientUtil) CreatePersistentVolumesTODODELETE

func (c *ClientUtil) CreatePersistentVolumesTODODELETE(cluster spec.Kafkacluster) error

TODO delete

func (*ClientUtil) CreateStorage

func (c *ClientUtil) CreateStorage(cluster spec.KafkaclusterSpec)

func (*ClientUtil) DeleteKafkaCluster

func (c *ClientUtil) DeleteKafkaCluster(cluster spec.Kafkacluster) error

func (*ClientUtil) DeleteOffsetMonitor

func (c *ClientUtil) DeleteOffsetMonitor(cluster spec.Kafkacluster) error

Deletes the offset checker for the given kafka cluster. Return error if any problems occurs. (Except if monitor dosnt exist)

func (*ClientUtil) DeployOffsetMonitor

func (c *ClientUtil) DeployOffsetMonitor(cluster spec.Kafkacluster) error

Deploys the OffsetMonitor as an extra Pod inside the Cluster

func (*ClientUtil) GenerateExporterDeployment

func (c *ClientUtil) GenerateExporterDeployment(cluster spec.Kafkacluster) *appsv1Beta1.Deployment

func (*ClientUtil) GenerateHeadlessService

func (c *ClientUtil) GenerateHeadlessService(cluster spec.Kafkacluster) *v1.Service

func (*ClientUtil) GenerateKafkaOptions

func (c *ClientUtil) GenerateKafkaOptions(cluster spec.Kafkacluster) []v1.EnvVar

func (*ClientUtil) GetBrokerStates

func (c *ClientUtil) GetBrokerStates(cluster spec.Kafkacluster) ([]string, error)

func (*ClientUtil) GetBrokersWithState

func (c *ClientUtil) GetBrokersWithState(cluster spec.Kafkacluster, state spec.KafkaBrokerState) (int32, error)

TODO return multiple not LAst? Return the Last Broker with the Desired State

func (*ClientUtil) GetMaxHeap

func (c *ClientUtil) GetMaxHeap(cluster spec.Kafkacluster) *resource.Quantity

func (*ClientUtil) GetMaxHeapJavaString

func (c *ClientUtil) GetMaxHeapJavaString(cluster spec.Kafkacluster) string

func (*ClientUtil) GetPodAnnotations

func (c *ClientUtil) GetPodAnnotations(cluster spec.Kafkacluster) error

func (*ClientUtil) GetReadyEndpoints

func (c *ClientUtil) GetReadyEndpoints(serviceName string, namespace string) []string

func (*ClientUtil) SetBrokerState

func (c *ClientUtil) SetBrokerState(cluster spec.Kafkacluster, brokerId int32, state spec.KafkaBrokerState) error

func (*ClientUtil) UpdateBrokerImage

func (c *ClientUtil) UpdateBrokerImage(cluster spec.Kafkacluster) error

func (*ClientUtil) UpsizeBrokerStS

func (c *ClientUtil) UpsizeBrokerStS(cluster spec.Kafkacluster) error