util

Documentation

Index

• Variables
• func CompareSlice(operator v1alpha1.LogicOperator, sa []string, sb []string) bool
• func ConnectToServer(udsSockAddr string, tcpSockAddr string, serverInfo *info.ServerInfo, ...) (*grpc.ClientConn, error)
• func GetConfigMapFromVolume(selector *corev1.ConfigMapKeySelector) (string, error)
• func GetConfigMapVolumePath(selector *corev1.ConfigMapKeySelector) (string, error)
• func GetGSSAPIConfig(config *dfv1.GSSAPI) (*sarama.GSSAPIConfig, error)
• func GetIsbSvcEnvVars(isbSvcConfig dfv1.BufferServiceConfig) (dfv1.ISBSvcType, []corev1.EnvVar)
• func GetSASL(saslConfig *dfv1.SASL) (..., error)
• func GetSaramaConfigFromYAMLString(yaml string) (*sarama.Config, error)
• func GetSecretFromVolume(selector *corev1.SecretKeySelector) (string, error)
• func GetSecretVolumePath(selector *corev1.SecretKeySelector) (string, error)
• func GetTLSConfig(config *dfv1.TLS) (*tls.Config, error)
• func K8sRestConfig() (*rest.Config, error)
• func LookupEnvBoolOr(key string, defaultValue bool) bool
• func LookupEnvIntOr(key string, defaultValue int) int
• func LookupEnvStringOr(key, defaultValue string) string
• func MustHash(v interface{}) string
• func MustJSON(in interface{}) string
• func MustUnJSON(v interface{}, in interface{})
• func RandomLowerCaseString(length int) string
• func RandomString(length int) string
• func StringSliceContains(list []string, str string) bool
• func ToUDFErr(name string, err error) error
• func VolumesFromSecretsAndConfigMaps(obj interface{}) ([]corev1.Volume, []corev1.VolumeMount)
• func WaitForServerInfo(timeout time.Duration, filePath string) (*info.ServerInfo, error)

Variables

var DefaultRetryBackoff = wait.Backoff{
	Steps:    10,
	Duration: 5 * time.Second,
	Factor:   2.0,
	Jitter:   0.1,
}

Functions

func CompareSlice

func CompareSlice(operator v1alpha1.LogicOperator, sa []string, sb []string) bool

CompareSlice compares two slices based on operator

func ConnectToServer

func ConnectToServer(udsSockAddr string, tcpSockAddr string, serverInfo *info.ServerInfo, maxMessageSize int) (*grpc.ClientConn, error)

ConnectToServer connects to the server with the given socket address based on the server info protocol.

func GetConfigMapFromVolume

func GetConfigMapFromVolume(selector *corev1.ConfigMapKeySelector) (string, error)

GetConfigMapFromVolume retrieves the value of mounted config map volume "/var/numaflow/config/${configMapRef.name}/${configMapRef.key}" is expected to be the file path

func GetConfigMapVolumePath

func GetConfigMapVolumePath(selector *corev1.ConfigMapKeySelector) (string, error)

GetConfigMapVolumePath returns the path of the mounted configmap

func GetGSSAPIConfig

func GetGSSAPIConfig(config *dfv1.GSSAPI) (*sarama.GSSAPIConfig, error)

GetGSSAPIConfig A utility function to get sasl.gssapi.Config

func GetIsbSvcEnvVars

func GetIsbSvcEnvVars(isbSvcConfig dfv1.BufferServiceConfig) (dfv1.ISBSvcType, []corev1.EnvVar)

GetIsbSvcEnvVars is helper function to get the ISB service type and generate corresponnding environment variables

func GetSASL

func GetSASL(saslConfig *dfv1.SASL) (*struct {
	Enable                   bool
	Mechanism                sarama.SASLMechanism
	Version                  int16
	Handshake                bool
	AuthIdentity             string
	User                     string
	Password                 string
	SCRAMAuthzID             string
	SCRAMClientGeneratorFunc func() sarama.SCRAMClient
	TokenProvider            sarama.AccessTokenProvider
	GSSAPI                   sarama.GSSAPIConfig
}, error)

GetSASLConfig A utility function to get sarama.Config.Net.SASL

func GetSaramaConfigFromYAMLString

func GetSaramaConfigFromYAMLString(yaml string) (*sarama.Config, error)

GetSaramaConfigFromYAMLString parse yaml string to sarama.config

func GetSecretFromVolume

func GetSecretFromVolume(selector *corev1.SecretKeySelector) (string, error)

GetSecretFromVolume retrieves the value of mounted secret volume "/var/numaflow/secrets/${secretRef.name}/${secretRef.key}" is expected to be the file path

func GetSecretVolumePath

func GetSecretVolumePath(selector *corev1.SecretKeySelector) (string, error)

GetSecretVolumePath returns the path of the mounted secret

func GetTLSConfig

func GetTLSConfig(config *dfv1.TLS) (*tls.Config, error)

A utility function to get tls.Config

func K8sRestConfig

func K8sRestConfig() (*rest.Config, error)

K8sRestConfig returns a rest config for the kubernetes cluster.

func LookupEnvBoolOr

func LookupEnvBoolOr(key string, defaultValue bool) bool

func LookupEnvIntOr

func LookupEnvIntOr(key string, defaultValue int) int

func LookupEnvStringOr

func LookupEnvStringOr(key, defaultValue string) string

func MustHash

func MustHash(v interface{}) string

func MustJSON

func MustJSON(in interface{}) string

func MustUnJSON

func MustUnJSON(v interface{}, in interface{})

MustUnJSON unmarshalls JSON or panics. v - must be []byte or string in - must be a pointer.

func RandomLowerCaseString

func RandomLowerCaseString(length int) string

func RandomString

func RandomString(length int) string

RandomString generate a random string with given length

func StringSliceContains

func StringSliceContains(list []string, str string) bool

func ToUDFErr

func ToUDFErr(name string, err error) error

ToUDFErr converts gRPC error to UDF Error

func VolumesFromSecretsAndConfigMaps

func VolumesFromSecretsAndConfigMaps(obj interface{}) ([]corev1.Volume, []corev1.VolumeMount)

VolumesFromSecretsOrConfigMaps builds volumes and volumeMounts spec based on the obj and its children's secretKeyselector and configMapKeySelector

func WaitForServerInfo

func WaitForServerInfo(timeout time.Duration, filePath string) (*info.ServerInfo, error)

WaitForServerInfo waits until the server info is ready. It returns an error if the server info is not ready within the given timeout.