Merge pull request #115 from kube-logging/feat/persistence

feat: add persistence

.github/workflows/ci.yaml

@@ -87,4 +87,4 @@ jobs:
       contents: read
       packages: write
       id-token: write
-      security-events: write
+      security-events: write

.go-version

@@ -1 +1 @@
-1.23.2
+1.23.3

Makefile

@@ -56,16 +56,19 @@ help: ## Display this help.
 
 ##@ Development
 
+.PHONY: generate
+generate: codegen manifests fmt ## Generate code, documentation, etc
+
+.PHONY: codegen
+codegen: controller-gen ## Generate code containing DeepCopy, DeepCopyInto, and DeepCopyObject method implementations.
+	$(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./api/..."
+
 .PHONY: manifests
 manifests: controller-gen ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects.
 	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="./api/..." output:crd:artifacts:config=config/crd/bases
 	$(CONTROLLER_GEN) rbac:roleName=manager-role paths="./internal/controller/telemetry/..." output:rbac:artifacts:config=./config/rbac
 	cp config/crd/bases/* charts/telemetry-controller/crds/
 
-.PHONY: generate
-generate: controller-gen ## Generate code containing DeepCopy, DeepCopyInto, and DeepCopyObject method implementations.
-	$(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./api/..."
-
 .PHONY: fmt
 fmt: ## Run go fmt against code.
 	go fmt ./...

api/telemetry/v1alpha1/otlp_config.go

@@ -29,43 +29,35 @@ type TimeoutSettings struct {
 
 // QueueSettings defines configuration for queueing batches before sending to the consumerSender.
 type QueueSettings struct {
-	// Enabled indicates whether to not enqueue batches before sending to the consumerSender.
-	Enabled bool `json:"enabled,omitempty"`
-
 	// NumConsumers is the number of consumers from the queue.
-	NumConsumers int `json:"num_consumers,omitempty"`
+	NumConsumers *int `json:"num_consumers,omitempty"`
 
 	// QueueSize is the maximum number of batches allowed in queue at a given time.
-	QueueSize int `json:"queue_size,omitempty"`
-
-	// StorageID if not empty, enables the persistent storage and uses the component specified
-	// as a storage extension for the persistent queue
-	StorageID string `json:"storage,omitempty"` //TODO this is *component.ID at Otel
+	// Default value is 100.
+	QueueSize *int `json:"queue_size,omitempty"`
 }
 
 // BackOffConfig defines configuration for retrying batches in case of export failure.
 // The current supported strategy is exponential backoff.
 type BackOffConfig struct {
-	// Enabled indicates whether to not retry sending batches in case of export failure.
-	Enabled bool `json:"enabled,omitempty"`
-
 	// InitialInterval the time to wait after the first failure before retrying.
-	InitialInterval time.Duration `json:"initial_interval,omitempty"`
+	InitialInterval *time.Duration `json:"initial_interval,omitempty"`
 
 	// RandomizationFactor is a random factor used to calculate next backoffs
 	// Randomized interval = RetryInterval * (1 ± RandomizationFactor)
-	RandomizationFactor string `json:"randomization_factor,omitempty"`
+	RandomizationFactor *string `json:"randomization_factor,omitempty"`
 
 	// Multiplier is the value multiplied by the backoff interval bounds
-	Multiplier string `json:"multiplier,omitempty"`
+	Multiplier *string `json:"multiplier,omitempty"`
 
 	// MaxInterval is the upper bound on backoff interval. Once this value is reached the delay between
 	// consecutive retries will always be `MaxInterval`.
-	MaxInterval time.Duration `json:"max_interval,omitempty"`
+	MaxInterval *time.Duration `json:"max_interval,omitempty"`
 
 	// MaxElapsedTime is the maximum amount of time (including retries) spent trying to send a request/batch.
 	// Once this value is reached, the data is discarded. If set to 0, the retries are never stopped.
-	MaxElapsedTime time.Duration `json:"max_elapsed_time,omitempty"`
+	// Default value is 0 to ensure that the data is never discarded.
+	MaxElapsedTime *time.Duration `json:"max_elapsed_time,omitempty"`
 }
 
 // KeepaliveClientConfig exposes the keepalive.ClientParameters to be used by the exporter.

api/telemetry/v1alpha1/output_types.go

@@ -99,35 +99,26 @@ type OTLPHTTP struct {
 	HTTPClientConfig `json:",inline"`
 }
 
-// Configuration for the fluentforward exporter.
-type Fluentforward struct {
-	TCPClientSettings `json:",inline"`
-
-	// RequireAck enables the acknowledgement feature.
-	RequireAck *bool `json:"require_ack,omitempty"`
-
-	// The Fluent tag parameter used for routing
-	Tag *string `json:"tag,omitempty"`
-
-	// CompressGzip enables gzip compression for the payload.
-	CompressGzip *bool `json:"compress_gzip,omitempty"`
-
-	// DefaultLabelsEnabled is a map of default attributes to be added to each log record.
-	DefaultLabelsEnabled *map[string]bool `json:"default_labels_enabled,omitempty"`
-
-	QueueConfig *QueueSettings      `json:"sending_queue,omitempty"`
-	RetryConfig *BackOffConfig      `json:"retry_on_failure,omitempty"`
-	Kubernetes  *KubernetesMetadata `json:"kubernetes_metadata,omitempty"`
+type Endpoint struct {
+	// TCPAddr is the address of the server to connect to.
+	TCPAddr *string `json:"tcp_addr"`
+	// Controls whether to validate the tcp address.
+	// Turning this ON may result in the collector failing to start if it came up faster then the endpoint.
+	// default is false.
+	ValidateTCPResolution bool `json:"validate_tcp_resolution"`
 }
 
 type KubernetesMetadata struct {
 	Key              string `json:"key"`
 	IncludePodLabels bool   `json:"include_pod_labels"`
 }
 
+// TCPClientSettings defines common settings for a TCP client.
 type TCPClientSettings struct {
-	// The target endpoint URI to send data to (e.g.: some.url:24224).
-	Endpoint *string `json:"endpoint,omitempty"`
+	// +kubebuilder:validation:Required
+
+	// Endpoint to send logs to.
+	*Endpoint `json:"endpoint"`
 
 	// +kubebuilder:validation:Format=duration
 
@@ -141,6 +132,27 @@ type TCPClientSettings struct {
 	SharedKey *string `json:"shared_key,omitempty"`
 }
 
+// Configuration for the fluentforward exporter.
+type Fluentforward struct {
+	TCPClientSettings `json:",inline"`
+
+	// RequireAck enables the acknowledgement feature.
+	RequireAck *bool `json:"require_ack,omitempty"`
+
+	// The Fluent tag parameter used for routing
+	Tag *string `json:"tag,omitempty"`
+
+	// CompressGzip enables gzip compression for the payload.
+	CompressGzip *bool `json:"compress_gzip,omitempty"`
+
+	// DefaultLabelsEnabled is a map of default attributes to be added to each log record.
+	DefaultLabelsEnabled *map[string]bool `json:"default_labels_enabled,omitempty"`
+
+	QueueConfig *QueueSettings      `json:"sending_queue,omitempty"`
+	RetryConfig *BackOffConfig      `json:"retry_on_failure,omitempty"`
+	Kubernetes  *KubernetesMetadata `json:"kubernetes_metadata,omitempty"`
+}
+
 // OutputStatus defines the observed state of Output
 type OutputStatus struct {
 }

api/telemetry/v1alpha1/tenant_types.go

@@ -28,7 +28,7 @@ type TransformStatement struct {
 	Statements []string `json:"statements,omitempty"`
 }
 
-// Transform represents the Transform processor, which modifies telemetry based on its configuration
+// Transform represents the Transform processor, which modifies telemetry based on its configuration.
 type Transform struct {
 	// Name of the Transform processor
 	Name string `json:"name,omitempty"`
@@ -50,7 +50,7 @@ type Transform struct {
 }
 
 // RouteConfig defines the routing configuration for a tenant
-// it will be used to generate routing connectors
+// it will be used to generate routing connectors.
 type RouteConfig struct {
 	// Contains the list of pipelines to use when a record does not meet any of specified conditions.
 	DefaultPipelines []string `json:"defaultPipelines,omitempty"` // TODO: Provide users with a guide to determine generated pipeline names
@@ -67,17 +67,36 @@ type RouteConfig struct {
 	MatchOnce bool `json:"matchOnce,omitempty"`
 }
 
+// Configuration for persistence, will be used to generate
+// the filestorage extension.
+type PersistenceConfig struct {
+	// Determines whether file storage is enabled or not.
+	EnableFileStorage bool `json:"enableFileStorage,omitempty"`
+
+	// The directory where logs will be persisted.
+	// If unset or an invalid path is given, then an OS specific
+	// default value will be used.
+	// The cluster administrator must ensure that the directory
+	// is unique for each tenant.
+	// If unset /var/lib/otelcol/file_storage/<tenant_name> will be used.
+	Directory string `json:"directory,omitempty"`
+}
+
 // TenantSpec defines the desired state of Tenant
 type TenantSpec struct {
 	// Determines the namespaces from which subscriptions are collected by this tenant.
 	SubscriptionNamespaceSelectors []metav1.LabelSelector `json:"subscriptionNamespaceSelectors,omitempty"`
 
 	// Determines the namespaces from which logs are collected by this tenant.
-	// If initialized with an empty list, logs from all namespaces are collected.
-	// If uninitialized, no logs are collected.
+	// Cannot be used together with SelectFromAllNamespaces.
 	LogSourceNamespaceSelectors []metav1.LabelSelector `json:"logSourceNamespaceSelectors,omitempty"`
-	Transform                   `json:"transform,omitempty"`
-	RouteConfig                 `json:"routeConfig,omitempty"`
+
+	// If true, logs are collected from all namespaces.
+	// Cannot be used together with LogSourceNamespaceSelectors.
+	SelectFromAllNamespaces bool `json:"selectFromAllNamespaces,omitempty"`
+	Transform               `json:"transform,omitempty"`
+	RouteConfig             `json:"routeConfig,omitempty"`
+	PersistenceConfig       `json:"persistenceConfig,omitempty"`
 }
 
 // TenantStatus defines the observed state of Tenant