Documentation
¶
Index ¶
- Constants
- type GatewayReadinessProgress
- type K8sService
- func (s *K8sService) ApplyGatewayYAML(ctx context.Context, namespace, gatewayName string, yamlData []byte) error
- func (s *K8sService) CheckPermissions(ctx context.Context, verb, resource, group, namespace string) (bool, error)
- func (s *K8sService) GetGatewayPodUIDs(ctx context.Context, namespace, gatewayName string) (map[types.UID]struct{}, error)
- func (s *K8sService) GetGatewayYAML(ctx context.Context, namespace, gatewayName string) ([]byte, error)
- func (s *K8sService) ValidateGatewayCRs(initialYAML, fencedYAML, switchoverYAML []byte) error
- func (s *K8sService) WaitForGatewayPods(ctx context.Context, namespace, gatewayName string, ...) error
- func (s *K8sService) WaitForGatewayReady(ctx context.Context, namespace, gatewayName string, ...) error
- type PodRolloutProgress
- type Service
Constants ¶
const ( // ConfluentNamespace = "kcp" GatewayGroup = "platform.confluent.io" GatewayVersion = "v1beta1" GatewayResourcePlural = "gateways" )
Kubernetes resource constants
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type GatewayReadinessProgress ¶ added in v0.8.1
type GatewayReadinessProgress struct {
InitialPodCount int
PodsReady int
Elapsed time.Duration
RolloutDetected bool
Ready bool
}
GatewayReadinessProgress reports the current state of a gateway readiness wait. The gate (whether the wait returns) comes from the underlying apps/v1 Deployment's rollout status; pod counts come from Deployment.status.readyReplicas.
type K8sService ¶
type K8sService struct {
// contains filtered or unexported fields
}
K8sService implements gateway operations using Kubernetes clients
func NewK8sService ¶
func NewK8sService(kubeConfigPath string) *K8sService
NewK8sService creates a new gateway service
func (*K8sService) ApplyGatewayYAML ¶
func (s *K8sService) ApplyGatewayYAML(ctx context.Context, namespace, gatewayName string, yamlData []byte) error
ApplyGatewayYAML applies a complete gateway CR YAML to the cluster using server-side apply
func (*K8sService) CheckPermissions ¶
func (s *K8sService) CheckPermissions(ctx context.Context, verb, resource, group, namespace string) (bool, error)
CheckPermissions checks if the user has the required Kubernetes permissions
func (*K8sService) GetGatewayPodUIDs ¶
func (s *K8sService) GetGatewayPodUIDs(ctx context.Context, namespace, gatewayName string) (map[types.UID]struct{}, error)
GetGatewayPodUIDs returns a set of UIDs for the current gateway pods. This should be called BEFORE patching the gateway to capture the initial pod state.
func (*K8sService) GetGatewayYAML ¶
func (s *K8sService) GetGatewayYAML(ctx context.Context, namespace, gatewayName string) ([]byte, error)
GetGatewayYAML retrieves the gateway resource as YAML
func (*K8sService) ValidateGatewayCRs ¶
func (s *K8sService) ValidateGatewayCRs(initialYAML, fencedYAML, switchoverYAML []byte) error
ValidateGatewayCRs validates that the initial, fenced, and switchover gateway CRs are consistent
func (*K8sService) WaitForGatewayPods ¶
func (s *K8sService) WaitForGatewayPods(ctx context.Context, namespace, gatewayName string, initialPodUIDs map[types.UID]struct{}, pollInterval, timeout time.Duration, onProgress func(PodRolloutProgress)) error
WaitForGatewayPods waits for all gateway pods to be completely replaced after a config change.
After patching the Gateway CRD, the Confluent operator triggers a rolling restart of gateway pods. This method polls until all initial pods are replaced with new pods and all are ready, ensuring the rollout is truly complete even when maxSurge temporarily creates extra pods.
The initialPodUIDs parameter must contain the pod UIDs captured BEFORE the gateway patch is applied. This is critical to avoid race conditions where the rollout completes before we capture the initial state.
The method works in two phases:
- Wait for change detection: Wait up to 10 seconds to detect if rollout starts
- Wait for complete replacement: Ensure all initial pods are replaced and new pods are ready
This prevents returning prematurely when maxSurge creates extra pods during the rollout.
func (*K8sService) WaitForGatewayReady ¶ added in v0.8.1
func (s *K8sService) WaitForGatewayReady(ctx context.Context, namespace, gatewayName string, pollInterval, timeout time.Duration, onProgress func(GatewayReadinessProgress)) error
WaitForGatewayReady polls the underlying apps/v1 Deployment's rollout status until the Deployment reports a complete rollout, or until ctx is cancelled. When timeout > 0, a hard deadline is applied via context; a timeout of 0 means no deadline.
The wait runs in two phases:
- Detection (up to 10s): if the Deployment is already at rollout-complete state for the entire window, no pod restart was required — onProgress is invoked once with RolloutDetected=false and the wait returns nil.
- Convergence: polls until the Deployment reports rollout complete (observedGeneration >= generation, updatedReplicas == replicas, availableReplicas == replicas, replicas > 0), calling onProgress on every poll tick with monotonically increasing Elapsed.
type PodRolloutProgress ¶
type PodRolloutProgress struct {
InitialPodCount int
NewPodsReady int
OldPodsRemaining int
RolloutDetected bool
}
PodRolloutProgress reports the current state of a pod rollout
type Service ¶
type Service interface {
GetGatewayYAML(ctx context.Context, namespace, gatewayName string) ([]byte, error)
ValidateGatewayCRs(initialYAML, fencedYAML, switchoverYAML []byte) error
CheckPermissions(ctx context.Context, verb, resource, group, namespace string) (bool, error)
ApplyGatewayYAML(ctx context.Context, namespace, gatewayName string, yamlData []byte) error
GetGatewayPodUIDs(ctx context.Context, namespace, gatewayName string) (map[types.UID]struct{}, error)
WaitForGatewayPods(ctx context.Context, namespace, gatewayName string, initialPodUIDs map[types.UID]struct{}, pollInterval, timeout time.Duration, onProgress func(PodRolloutProgress)) error
WaitForGatewayReady(ctx context.Context, namespace, gatewayName string, pollInterval, timeout time.Duration, onProgress func(GatewayReadinessProgress)) error
}
Service defines gateway operations
Source Files