controller

Documentation

Overview

Copyright 2024 Illumio, Inc. All Rights Reserved.

Index

• Variables
• func ConnectStreams(ctx context.Context, logger *zap.SugaredLogger, envMap EnvironmentConfig, ...)
• func GetClusterID(ctx context.Context, logger *zap.SugaredLogger) (string, error)
• func IsRunningInCluster() bool
• func NewAuthenticatedConnection(ctx context.Context, logger *zap.SugaredLogger, envMap EnvironmentConfig) (*grpc.ClientConn, pb.KubernetesInfoServiceClient, error)
• func NewClientSet() (*kubernetes.Clientset, error)
• func NewGRPClogger(grpcSyncer *BufferedGrpcWriteSyncer) *zap.SugaredLogger
• func ServerIsHealthy() bool
• func SetUpOAuthConnection(ctx context.Context, logger *zap.SugaredLogger, tokenURL string, ...) (*grpc.ClientConn, error)
• type Authenticator
  • func (authn *Authenticator) DoesK8sSecretExist(ctx context.Context, secretName string) bool
  • func (authn *Authenticator) GetOnboardingCredentials(ctx context.Context, clientID string, clientSecret string) (Credentials, error)
  • func (authn *Authenticator) ReadCredentialsK8sSecrets(ctx context.Context, secretName string) (string, string, error)
  • func (authn *Authenticator) WriteK8sSecret(ctx context.Context, keyData OnboardResponse, ClusterCreds string) error
• type BufferedGrpcWriteSyncer
  • func NewBufferedGrpcWriteSyncer() *BufferedGrpcWriteSyncer
  • func (b *BufferedGrpcWriteSyncer) Close() error
  • func (b *BufferedGrpcWriteSyncer) ListenToLogStream() error
  • func (b *BufferedGrpcWriteSyncer) UpdateClient(client pb.KubernetesInfoService_SendLogsClient, conn ClientConnInterface)
• type Cache
• type CiliumFlowCollector
• type ClientConnInterface
• type Credentials
• type EnvironmentConfig
• type OnboardResponse
  • func Onboard(ctx context.Context, TlsSkipVerify bool, OnboardingEndpoint string, ...) (OnboardResponse, error)
• type ResourceManager
  • func (r *ResourceManager) DyanmicListAndWatchResources(ctx context.Context, cancel context.CancelFunc, resource string, ...)
  • func (r *ResourceManager) DynamicListResources(ctx context.Context, resource string, apiGroup string) (string, Cache, error)

Variables

var (
	ErrHubbleNotFound   = errors.New("hubble Relay service not found; disabling Cilium flow collection")
	ErrNoPortsAvailable = errors.New("hubble Relay service has no ports; disabling Cilium flow collection")
)

var ErrStopRetries = errors.New("stop retries")

Functions

func ConnectStreams

func ConnectStreams(ctx context.Context, logger *zap.SugaredLogger, envMap EnvironmentConfig, bufferedGrpcSyncer *BufferedGrpcWriteSyncer)

ConnectStreams will continue to reboot and restart the main operations within the operator if any disconnects or errors occur.

func GetClusterID

func GetClusterID(ctx context.Context, logger *zap.SugaredLogger) (string, error)

GetClusterID returns the uid of the k8s cluster's kube-system namespace, which is used as the cluster's globally unique ID.

func IsRunningInCluster

func IsRunningInCluster() bool

IsRunningInCluster helps determine if the application is running inside a Kubernetes cluster.

func NewAuthenticatedConnection

func NewAuthenticatedConnection(ctx context.Context, logger *zap.SugaredLogger, envMap EnvironmentConfig) (*grpc.ClientConn, pb.KubernetesInfoServiceClient, error)

NewAuthenticatedConnection gets a valid token and creats a connection to CloudSecure.

func NewClientSet

func NewClientSet() (*kubernetes.Clientset, error)

NewClientSet returns a new Kubernetes clientset based on the execution environment.

func NewGRPClogger

func NewGRPClogger(grpcSyncer *BufferedGrpcWriteSyncer) *zap.SugaredLogger

NewGRPClogger will define a new zap logger with multiple writesyncs one to stdout and one for GRPC writestream

func ServerIsHealthy

func ServerIsHealthy() bool

ServerIsHealthy checks if a deadlock has occured within the threaded resource listing process.

func SetUpOAuthConnection

func SetUpOAuthConnection(ctx context.Context, logger *zap.SugaredLogger, tokenURL string, TlsSkipVerify bool, clientID string, clientSecret string) (*grpc.ClientConn, error)

SetUpOAuthConnection establishes a gRPC connection using OAuth credentials and logging the process.

Types

type Authenticator

type Authenticator struct {
	Logger *zap.SugaredLogger
}

Authenticator keeps a logger for its own methods.

func (*Authenticator) DoesK8sSecretExist

func (authn *Authenticator) DoesK8sSecretExist(ctx context.Context, secretName string) bool

func (*Authenticator) GetOnboardingCredentials

func (authn *Authenticator) GetOnboardingCredentials(ctx context.Context, clientID string, clientSecret string) (Credentials, error)

GetOnboardingCredentials returns credentials to onboard this cluster with CloudSecure.

func (*Authenticator) ReadCredentialsK8sSecrets

func (authn *Authenticator) ReadCredentialsK8sSecrets(ctx context.Context, secretName string) (string, string, error)

ReadK8sSecret takes a secretName and reads the file.

func (*Authenticator) WriteK8sSecret

func (authn *Authenticator) WriteK8sSecret(ctx context.Context, keyData OnboardResponse, ClusterCreds string) error

WriteK8sSecret takes an OnboardResponse and writes it to a locally kept secret.

type BufferedGrpcWriteSyncer

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

BufferedGrpcWriteSyncer is a custom zap writesync that writes to a grpc stream In case stream is not connected it will buffer to memory

func NewBufferedGrpcWriteSyncer

func NewBufferedGrpcWriteSyncer() *BufferedGrpcWriteSyncer

NewBufferedGrpcWriteSyncer returns a new BufferedGrpcWriteSyncer

func (*BufferedGrpcWriteSyncer) Close

func (b *BufferedGrpcWriteSyncer) Close() error

Close flushes buffered log data into grpc stream if possible, and closes the connection.

func (*BufferedGrpcWriteSyncer) ListenToLogStream

func (b *BufferedGrpcWriteSyncer) ListenToLogStream() error

ListenToLogStream will wait for responses from server and will update log level depending on response contents

func (*BufferedGrpcWriteSyncer) UpdateClient

func (b *BufferedGrpcWriteSyncer) UpdateClient(client pb.KubernetesInfoService_SendLogsClient, conn ClientConnInterface)

UpdateClient will update BufferedGrpcWriteSyncer with new client stream and GRPC connection

type Cache

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

Cache contains the cache that is used to store seen events.

type CiliumFlowCollector

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

CiliumFlowCollector collects flows from Cilium Hubble Relay running in this cluster.

type ClientConnInterface

type ClientConnInterface interface {
	GetState() connectivity.State
	Close() error
}

type Credentials

type Credentials struct {
	ClientID     string `json:"client_id"`
	ClientSecret string `json:"client_secret"`
}

Credentials contains attributes that are needed for onboarding.

type EnvironmentConfig

type EnvironmentConfig struct {
	// Namspace of Cilium.
	CiliumNamespace string
	// K8s cluster secret name.
	ClusterCreds string
	// Client ID for onboarding. "" if not specified, i.e. if the operator is not meant to onboard itself.
	OnboardingClientId string
	// Client secret for onboarding. "" if not specified, i.e. if the operator is not meant to onboard itself.
	OnboardingClientSecret string
	// URL of the onboarding endpoint.
	OnboardingEndpoint string
	// URL of the token endpoint.
	TokenEndpoint string
	// Whether to skip TLS certificate verification when starting a stream.
	TlsSkipVerify bool
}

type OnboardResponse

type OnboardResponse struct {
	ClusterClientId     string `json:"cluster_client_id"`
	ClusterClientSecret string `json:"cluster_client_secret"`
}

func Onboard

func Onboard(ctx context.Context, TlsSkipVerify bool, OnboardingEndpoint string, credentials Credentials, logger *zap.SugaredLogger) (OnboardResponse, error)

Onboard onboards this cluster with CloudSecure using the onboarding credentials and obtains OAuth 2 credentials for this cluster.

type ResourceManager

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

ResourceManager encapsulates components for listing and managing Kubernetes resources.

func (*ResourceManager) DyanmicListAndWatchResources

func (r *ResourceManager) DyanmicListAndWatchResources(ctx context.Context, cancel context.CancelFunc, resource string, apiGroup string, allResourcesSnapshotted *sync.WaitGroup, snapshotCompleted *sync.WaitGroup)

DynamicListAndWatchResources lists and watches the specified resource dynamically, managing context cancellation and synchronization with wait groups.

func (*ResourceManager) DynamicListResources

func (r *ResourceManager) DynamicListResources(ctx context.Context, resource string, apiGroup string) (string, Cache, error)

DynamicListResources lists a specifed resource dynamically and sends down the current gRPC stream.