Documentation
¶
Overview ¶
Package kubernetes provides Kubernetes environment detection and integration.
Package kubernetes provides Kubernetes environment detection and integration.
Package kubernetes provides Kubernetes environment detection and integration.
Index ¶
- Constants
- Variables
- func TokenReviewMiddleware(auth *TokenReviewAuthenticator) func(http.Handler) http.Handler
- type CRDController
- func (c *CRDController) Start(ctx context.Context) error
- func (c *CRDController) Stop()
- func (c *CRDController) UpdateData(ctx context.Context, data []DataEntry) error
- func (c *CRDController) UpdateMetadata(ctx context.Context, metadata map[string]string) error
- func (c *CRDController) UpdateSecretIndex(ctx context.Context, refs []SecretRef, secretData map[string][]byte) error
- type Config
- type DataEntry
- type DefaultDetector
- type EnvironmentDetector
- type HTTPTokenReviewClient
- type K8sAuditLogReader
- type KubeClient
- type KubeSecret
- type KubernetesEnvironment
- type PlexdHook
- type PlexdHookController
- type PlexdHookEvent
- type PlexdHookJobTemplate
- type PlexdHookParam
- type PlexdHookSpec
- type PlexdHookStatus
- type PlexdJob
- type PlexdJobContainer
- type PlexdJobOwnerRef
- type PlexdNodeState
- type PlexdNodeStateEvent
- type PlexdNodeStateSpec
- type PlexdNodeStateStatus
- type ReportNotifier
- type SecretRef
- type TokenReviewAuthenticator
- type TokenReviewClient
- type TokenReviewResult
Constants ¶
const DefaultAuditLogPath = "/var/log/kubernetes/audit/audit.log"
DefaultAuditLogPath is the default path to the Kubernetes audit log.
const DefaultCACertPath = ServiceAccountBasePath + "/ca.crt"
DefaultCACertPath is the default path to the Kubernetes cluster CA certificate.
const DefaultCRDSyncInterval = 10 * time.Second
DefaultCRDSyncInterval is the default interval for syncing CRD state.
const DefaultNamespacePath = ServiceAccountBasePath + "/namespace"
DefaultNamespacePath is the default path to the service account namespace file.
const DefaultTokenPath = ServiceAccountBasePath + "/token"
DefaultTokenPath is the default path to the service account token.
const ServiceAccountBasePath = "/var/run/secrets/kubernetes.io/serviceaccount"
ServiceAccountBasePath is the base path for Kubernetes service account secrets.
Variables ¶
var ( // ErrNotFound is returned when a requested resource does not exist. ErrNotFound = errors.New("kubernetes: resource not found") // ErrAlreadyExists is returned when creating a resource that already exists. ErrAlreadyExists = errors.New("kubernetes: resource already exists") ErrUnauthorized = errors.New("kubernetes: unauthorized") )
Sentinel errors for Kubernetes operations.
Functions ¶
func TokenReviewMiddleware ¶
func TokenReviewMiddleware(auth *TokenReviewAuthenticator) func(http.Handler) http.Handler
TokenReviewMiddleware returns an HTTP middleware that authenticates requests using the TokenReviewAuthenticator. It extracts the Bearer token from the Authorization header, validates it, and attaches the identity to the request context. Returns 401 Unauthorized on failure.
Types ¶
type CRDController ¶
type CRDController struct {
// contains filtered or unexported fields
}
CRDController manages the lifecycle of a PlexdNodeState CRD resource: creates on startup, updates spec on state changes, watches status for workload-written report entries, and manages associated Kubernetes Secrets.
func NewCRDController ¶
func NewCRDController(client KubeClient, cfg Config, nodeID, meshIP, namespace string, reportNotifier ReportNotifier, logger *slog.Logger) *CRDController
NewCRDController creates a new CRDController.
func (*CRDController) Start ¶
func (c *CRDController) Start(ctx context.Context) error
Start creates or updates the PlexdNodeState resource and starts the status watcher goroutine. It blocks until the context is cancelled.
func (*CRDController) Stop ¶
func (c *CRDController) Stop()
Stop cancels the status watch and returns.
func (*CRDController) UpdateData ¶
func (c *CRDController) UpdateData(ctx context.Context, data []DataEntry) error
UpdateData updates the PlexdNodeState .spec.data via the KubeClient.
func (*CRDController) UpdateMetadata ¶
UpdateMetadata updates the PlexdNodeState .spec.metadata via the KubeClient.
func (*CRDController) UpdateSecretIndex ¶
func (c *CRDController) UpdateSecretIndex(ctx context.Context, refs []SecretRef, secretData map[string][]byte) error
UpdateSecretIndex updates the PlexdNodeState .spec.secretRefs and manages associated Kubernetes Secrets with ownerReferences. New refs get Secrets created; removed refs get Secrets deleted.
type Config ¶
type Config struct {
// Enabled controls whether Kubernetes integration is active.
// Default: false (must be explicitly enabled).
Enabled bool
// CRDEnabled controls whether CRD management is active.
// Default: true when Enabled is true.
CRDEnabled bool
// Namespace overrides the auto-detected namespace. If empty, the
// namespace is read from the service account metadata at runtime.
Namespace string
// AuditLogPath is the filesystem path to the Kubernetes audit log.
// Default: /var/log/kubernetes/audit/audit.log.
AuditLogPath string
// CRDSyncInterval controls how often CRD state is reconciled.
// Must be at least 1s. Default: 10s.
CRDSyncInterval time.Duration
// TokenPath is the filesystem path to the service account token.
// Default: /var/run/secrets/kubernetes.io/serviceaccount/token.
TokenPath string
}
Config holds the configuration for the Kubernetes integration.
func (*Config) ApplyDefaults ¶
func (c *Config) ApplyDefaults(env *KubernetesEnvironment)
ApplyDefaults sets default values for zero-valued fields. If env is non-nil and InCluster is true, auto-detected values are used for unset fields.
type DataEntry ¶
type DataEntry struct {
Key string `json:"key"`
ContentType string `json:"contentType,omitempty"`
Payload any `json:"payload,omitempty"`
Version int `json:"version,omitempty"`
UpdatedAt string `json:"updatedAt,omitempty"`
}
DataEntry represents a single data item in the PlexdNodeState spec or status report.
type DefaultDetector ¶
DefaultDetector implements EnvironmentDetector using real environment variables and filesystem paths.
func (*DefaultDetector) Detect ¶
func (d *DefaultDetector) Detect() *KubernetesEnvironment
Detect checks environment variables and filesystem paths to determine whether the process is running inside a Kubernetes pod. Returns a KubernetesEnvironment with InCluster=false if KUBERNETES_SERVICE_HOST is not set, or if the service account token file does not exist.
type EnvironmentDetector ¶
type EnvironmentDetector interface {
// Detect returns information about the Kubernetes environment.
// Returns nil if the process is not running inside a Kubernetes pod.
Detect() *KubernetesEnvironment
}
EnvironmentDetector detects whether the process is running inside a Kubernetes cluster and returns environment metadata.
type HTTPTokenReviewClient ¶
type HTTPTokenReviewClient struct {
// contains filtered or unexported fields
}
HTTPTokenReviewClient implements TokenReviewClient using the Kubernetes API server.
func NewHTTPTokenReviewClient ¶
func NewHTTPTokenReviewClient(apiServer, saTokenPath string) *HTTPTokenReviewClient
NewHTTPTokenReviewClient creates a new client that calls the TokenReview API on the given apiServer. The saTokenPath is the filesystem path to the service account token used to authenticate with the API server.
func (*HTTPTokenReviewClient) Review ¶
func (c *HTTPTokenReviewClient) Review(ctx context.Context, token string) (*TokenReviewResult, error)
Review validates a bearer token by calling the Kubernetes TokenReview API.
type K8sAuditLogReader ¶
type K8sAuditLogReader struct {
// contains filtered or unexported fields
}
K8sAuditLogReader implements auditfwd.K8sAuditReader by reading Kubernetes audit log files in JSON-lines format. It tracks the file position between reads so only new entries are returned on each call.
func NewK8sAuditLogReader ¶
func NewK8sAuditLogReader(path string, logger *slog.Logger) *K8sAuditLogReader
NewK8sAuditLogReader creates a new reader for the audit log at the given path.
func (*K8sAuditLogReader) ReadEvents ¶
func (r *K8sAuditLogReader) ReadEvents(_ context.Context) ([]auditfwd.K8sAuditEntry, error)
ReadEvents reads new audit log entries since the last call. It returns only entries appended after the previously recorded offset. If the file does not exist, it returns nil, nil. If the file has been truncated (current size is smaller than the stored offset), it resets to the beginning.
type KubeClient ¶
type KubeClient interface {
// GetNodeState retrieves the PlexdNodeState CRD for the given name and namespace.
// Returns ErrNotFound if the resource does not exist.
GetNodeState(ctx context.Context, namespace, name string) (*PlexdNodeState, error)
// CreateNodeState creates a new PlexdNodeState CRD resource.
// Returns ErrAlreadyExists if the resource already exists.
CreateNodeState(ctx context.Context, state *PlexdNodeState) error
// UpdateNodeState updates an existing PlexdNodeState CRD resource spec.
// Returns ErrNotFound if the resource does not exist.
UpdateNodeState(ctx context.Context, state *PlexdNodeState) error
// DeleteNodeState deletes a PlexdNodeState CRD resource.
// Returns nil if the resource does not exist (idempotent).
DeleteNodeState(ctx context.Context, namespace, name string) error
// WatchNodeState starts a watch on a specific PlexdNodeState resource.
// Returns a channel that receives events until the context is cancelled.
WatchNodeState(ctx context.Context, namespace, name string) (<-chan PlexdNodeStateEvent, error)
// CreateSecret creates a Kubernetes Secret.
// Returns ErrAlreadyExists if the secret already exists.
CreateSecret(ctx context.Context, secret *KubeSecret) error
// UpdateSecret updates an existing Kubernetes Secret.
// Returns ErrNotFound if the secret does not exist.
UpdateSecret(ctx context.Context, secret *KubeSecret) error
// DeleteSecret deletes a Kubernetes Secret.
// Returns nil if the secret does not exist (idempotent).
DeleteSecret(ctx context.Context, namespace, name string) error
// WatchPlexdHooks starts a watch on PlexdHook resources in the given namespace.
// Returns a channel that receives events until the context is cancelled.
WatchPlexdHooks(ctx context.Context, namespace string) (<-chan PlexdHookEvent, error)
// UpdatePlexdHookStatus updates the status subresource of a PlexdHook resource.
// Returns ErrNotFound if the resource does not exist.
UpdatePlexdHookStatus(ctx context.Context, hook *PlexdHook) error
// CreateJob creates a Kubernetes batch/v1 Job.
// Returns ErrAlreadyExists if the Job already exists.
CreateJob(ctx context.Context, job *PlexdJob) error
}
KubeClient abstracts Kubernetes API interactions for testability. All methods that modify state must be idempotent: repeating an operation that is already applied returns nil.
type KubeSecret ¶
type KubeSecret struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
Labels map[string]string `json:"labels,omitempty"`
Data map[string][]byte `json:"data,omitempty"`
OwnerRefName string `json:"ownerRefName,omitempty"`
OwnerRefUID string `json:"ownerRefUID,omitempty"`
}
KubeSecret represents a Kubernetes Secret managed by plexd.
type KubernetesEnvironment ¶
type KubernetesEnvironment struct {
// InCluster is true when plexd is running inside a Kubernetes pod.
InCluster bool
// Namespace is the Kubernetes namespace of the pod.
Namespace string
// PodName is the name of the pod, typically sourced from the HOSTNAME env var.
PodName string
// NodeName is the Kubernetes node name, sourced from the MY_NODE_NAME env var
// (set via the downward API).
NodeName string
// ServiceAccountToken is the filesystem path to the service account token.
ServiceAccountToken string
}
KubernetesEnvironment holds information about the Kubernetes environment in which plexd is running. A nil value indicates that the process is not running inside a Kubernetes pod.
type PlexdHook ¶
type PlexdHook struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
UID string `json:"uid,omitempty"`
ResourceVersion string `json:"resourceVersion,omitempty"`
Labels map[string]string `json:"labels,omitempty"`
Spec PlexdHookSpec `json:"spec"`
Status PlexdHookStatus `json:"status,omitempty"`
}
PlexdHook represents the CRD resource for a hook execution request.
type PlexdHookController ¶
type PlexdHookController struct {
// contains filtered or unexported fields
}
PlexdHookController watches PlexdHook CRD resources and creates Kubernetes Jobs to execute hook workloads on the local node.
func NewPlexdHookController ¶
func NewPlexdHookController(client KubeClient, cfg Config, namespace, nodeName string, logger *slog.Logger) *PlexdHookController
NewPlexdHookController creates a new PlexdHookController.
func (*PlexdHookController) Start ¶
func (c *PlexdHookController) Start(ctx context.Context) error
Start watches PlexdHook resources and creates Jobs for new hooks. It blocks until the context is cancelled or the watch channel is closed.
func (*PlexdHookController) Stop ¶
func (c *PlexdHookController) Stop()
Stop cancels the watch and returns.
type PlexdHookEvent ¶
PlexdHookEvent represents a watch event for a PlexdHook resource.
type PlexdHookJobTemplate ¶
type PlexdHookJobTemplate struct {
Image string `json:"image"`
Command []string `json:"command,omitempty"`
Args []string `json:"args,omitempty"`
}
PlexdHookJobTemplate holds the job template for a PlexdHook.
type PlexdHookParam ¶
PlexdHookParam represents a name/value parameter for a PlexdHook.
type PlexdHookSpec ¶
type PlexdHookSpec struct {
HookName string `json:"hookName"`
JobTemplate *PlexdHookJobTemplate `json:"jobTemplate,omitempty"`
Parameters []PlexdHookParam `json:"parameters,omitempty"`
Privileged bool `json:"privileged,omitempty"`
}
PlexdHookSpec holds the spec fields of a PlexdHook CRD resource.
type PlexdHookStatus ¶
type PlexdHookStatus struct {
JobName string `json:"jobName,omitempty"`
Phase string `json:"phase,omitempty"`
Message string `json:"message,omitempty"`
StartedAt *time.Time `json:"startedAt,omitempty"`
CompletedAt *time.Time `json:"completedAt,omitempty"`
}
PlexdHookStatus holds the status fields of a PlexdHook CRD resource.
type PlexdJob ¶
type PlexdJob struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
Labels map[string]string `json:"labels,omitempty"`
OwnerReferences []PlexdJobOwnerRef `json:"ownerReferences,omitempty"`
NodeSelector map[string]string `json:"nodeSelector,omitempty"`
ServiceAccountName string `json:"serviceAccountName,omitempty"`
Containers []PlexdJobContainer `json:"containers"`
RestartPolicy string `json:"restartPolicy"`
}
PlexdJob represents a Kubernetes batch/v1 Job managed by plexd.
type PlexdJobContainer ¶
type PlexdJobContainer struct {
Name string `json:"name"`
Image string `json:"image"`
Command []string `json:"command,omitempty"`
Args []string `json:"args,omitempty"`
Env map[string]string `json:"env,omitempty"`
Privileged bool `json:"privileged,omitempty"`
ReadOnlyRootFS bool `json:"readOnlyRootFilesystem,omitempty"`
DropCapabilities []string `json:"dropCapabilities,omitempty"`
}
PlexdJobContainer represents a container in a PlexdJob.
type PlexdJobOwnerRef ¶
type PlexdJobOwnerRef struct {
APIVersion string `json:"apiVersion"`
Kind string `json:"kind"`
Name string `json:"name"`
UID string `json:"uid"`
Controller bool `json:"controller"`
BlockOwnerDeletion bool `json:"blockOwnerDeletion"`
}
PlexdJobOwnerRef represents an owner reference for a PlexdJob.
type PlexdNodeState ¶
type PlexdNodeState struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
UID string `json:"uid,omitempty"`
ResourceVersion string `json:"resourceVersion,omitempty"`
Labels map[string]string `json:"labels,omitempty"`
Spec PlexdNodeStateSpec `json:"spec"`
Status PlexdNodeStateStatus `json:"status,omitempty"`
LastUpdate time.Time `json:"lastUpdate"`
}
PlexdNodeState represents the CRD resource for a node's state.
type PlexdNodeStateEvent ¶
type PlexdNodeStateEvent struct {
Type string `json:"type"` // ADDED, MODIFIED, DELETED
State *PlexdNodeState `json:"state"`
}
PlexdNodeStateEvent represents a watch event for a PlexdNodeState resource.
type PlexdNodeStateSpec ¶
type PlexdNodeStateSpec struct {
NodeID string `json:"nodeId"`
MeshIP string `json:"meshIp,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
Data []DataEntry `json:"data,omitempty"`
SecretRefs []SecretRef `json:"secretRefs,omitempty"`
}
PlexdNodeStateSpec holds the spec fields of a PlexdNodeState CRD resource.
type PlexdNodeStateStatus ¶
type PlexdNodeStateStatus struct {
Report []DataEntry `json:"report,omitempty"`
}
PlexdNodeStateStatus holds the status fields of a PlexdNodeState CRD resource.
type ReportNotifier ¶
type ReportNotifier interface {
NotifyChange()
}
ReportNotifier is called when the CRDController detects new or changed report entries in the status subresource.
type SecretRef ¶
type SecretRef struct {
Key string `json:"key"`
SecretName string `json:"secretName"`
Version int `json:"version,omitempty"`
}
SecretRef represents a reference to a Kubernetes Secret holding encrypted data.
type TokenReviewAuthenticator ¶
type TokenReviewAuthenticator struct {
// contains filtered or unexported fields
}
TokenReviewAuthenticator validates bearer tokens using the Kubernetes TokenReview API.
func NewTokenReviewAuthenticator ¶
func NewTokenReviewAuthenticator(client TokenReviewClient, logger *slog.Logger, audiences []string) *TokenReviewAuthenticator
NewTokenReviewAuthenticator creates a new authenticator that validates tokens via the provided TokenReviewClient. If audiences is non-empty, the result must contain at least one matching audience for authentication to succeed.
func (*TokenReviewAuthenticator) Authenticate ¶
func (a *TokenReviewAuthenticator) Authenticate(ctx context.Context, token string) (*TokenReviewResult, error)
Authenticate validates the given bearer token. Returns the review result if the token is valid, or an error if authentication fails. If audiences are configured, the result must contain at least one matching audience.
type TokenReviewClient ¶
type TokenReviewClient interface {
// Review validates a bearer token via the Kubernetes TokenReview API.
// Returns the authenticated identity if valid, or an error if invalid or unreachable.
Review(ctx context.Context, token string) (*TokenReviewResult, error)
}
TokenReviewClient abstracts Kubernetes TokenReview API access for testability.
type TokenReviewResult ¶
type TokenReviewResult struct {
Authenticated bool
Username string
UID string
Groups []string
Audiences []string
}
TokenReviewResult contains the result of a TokenReview API call.
func IdentityFromContext ¶
func IdentityFromContext(ctx context.Context) *TokenReviewResult
IdentityFromContext retrieves the TokenReviewResult from the request context. Returns nil if no identity is present.