Fluentbit support for filesystem buffer storage and rolling upgrade o…

…n config change (#213) * Fluentbit support for filesystem buffer storage and rolling upgrade on config change * Separate in-container path of the buffer from the hostpath used for the mount. Setup postionDb by default for fluent-bit. Add docs. * add some lint checks back and fix lint errors * restore sample config, reword docs
kube-logging · Nov 6, 2019 · 76bbe22 · 76bbe22
1 parent c689b53
commit 76bbe22
Show file tree

Hide file tree

Showing 22 changed files with 238 additions and 76 deletions.
diff --git a/.golangci.yml b/.golangci.yml
@@ -33,10 +33,8 @@ linters:
         - stylecheck
         - staticcheck
         - gosimple
-        - whitespace
         - gocritic
         - godox
-        - interfacer
         - gocyclo
 
 issues:

diff --git a/api/v1beta1/fluentbit_types.go b/api/v1beta1/fluentbit_types.go
@@ -24,21 +24,23 @@ import (
 
 // FluentbitSpec defines the desired state of Fluentbit
 type FluentbitSpec struct {
-	Annotations        map[string]string           `json:"annotations,omitempty"`
-	Image              ImageSpec                   `json:"image,omitempty"`
-	TLS                FluentbitTLS                `json:"tls,omitempty"`
-	TargetHost         string                      `json:"targetHost,omitempty"`
-	TargetPort         int32                       `json:"targetPort,omitempty"`
-	Resources          corev1.ResourceRequirements `json:"resources,omitempty"`
-	Parser             string                      `json:"parser,omitempty"`
-	Tolerations        []corev1.Toleration         `json:"tolerations,omitempty"`
-	Metrics            *Metrics                    `json:"metrics,omitempty"`
-	Security           *Security                   `json:"security,omitempty"`
-	PositionDB         *KubernetesStorage          `json:"position_db,omitempty"`
-	MountPath          string                      `json:"mountPath,omitempty"`
-	InputTail          InputTail                   `json:"inputTail,omitempty"`
-	FilterKubernetes   FilterKubernetes            `json:"filterKubernetes,omitempty"`
-	CustomConfigSecret string                      `json:"customConfigSecret,omitempty"`
+	Annotations         map[string]string           `json:"annotations,omitempty"`
+	Image               ImageSpec                   `json:"image,omitempty"`
+	TLS                 FluentbitTLS                `json:"tls,omitempty"`
+	TargetHost          string                      `json:"targetHost,omitempty"`
+	TargetPort          int32                       `json:"targetPort,omitempty"`
+	Resources           corev1.ResourceRequirements `json:"resources,omitempty"`
+	Parser              string                      `json:"parser,omitempty"`
+	Tolerations         []corev1.Toleration         `json:"tolerations,omitempty"`
+	Metrics             *Metrics                    `json:"metrics,omitempty"`
+	Security            *Security                   `json:"security,omitempty"`
+	PositionDB          *KubernetesStorage          `json:"position_db,omitempty"`
+	MountPath           string                      `json:"mountPath,omitempty"`
+	InputTail           InputTail                   `json:"inputTail,omitempty"`
+	FilterKubernetes    FilterKubernetes            `json:"filterKubernetes,omitempty"`
+	BufferStorage       BufferStorage               `json:"bufferStorage,omitempty"`
+	BufferStorageVolume *KubernetesStorage          `json:"bufferStorageVolume,omitempty"`
+	CustomConfigSecret  string                      `json:"customConfigSecret,omitempty"`
 }
 
 // +kubebuilder:object:generate=true
@@ -63,8 +65,22 @@ func (spec FluentbitSpec) GetPrometheusPortFromAnnotation() int32 {
 	return int32(port)
 }
 
+// BufferStorage is the Service Section Configuration of fluent-bit
+type BufferStorage struct {
+	// Set an optional location in the file system to store streams and chunks of data. If this parameter is not set, Input plugins can only use in-memory buffering.
+	StoragePath string `json:"storage.path,omitempty"`
+	// Configure the synchronization mode used to store the data into the file system. It can take the values normal or full. (default:normal)
+	StorageSync string `json:"storage.sync,omitempty"`
+	// Enable the data integrity check when writing and reading data from the filesystem. The storage layer uses the CRC32 algorithm. (default:Off)
+	StorageChecksum string `json:"storage.checksum,omitempty"`
+	// If storage.path is set, Fluent Bit will look for data chunks that were not delivered and are still in the storage layer, these are called backlog data. This option configure a hint of maximum value of memory to use when processing these records. (default:5M)
+	StorageBacklogMemLimit string `json:"storage.backlog.mem_limit,omitempty"`
+}
+
 // InputTail defines Fluentbit tail input configuration The tail input plugin allows to monitor one or several text files. It has a similar behavior like tail -f shell command.
 type InputTail struct {
+	// Specify the buffering mechanism to use. It can be memory or filesystem. (default:memory)
+	StorageType string `json:"storage.type,omitempty"`
 	// Set the buffer size for HTTP client when reading responses from Kubernetes API server. The value must be according to the Unit Size specification. (default:32k)
 	BufferChunkSize string `json:"Buffer_Chunk_Size,omitempty"`
 	// Set the limit of the buffer size per monitored file. When a buffer needs to be increased (e.g: very long lines), this value is used to restrict how much the memory buffer can grow. If reading a file exceed this limit, the file is removed from the monitored file list. The value must be according to the Unit Size specification. (default:Buffer_Chunk_Size)

diff --git a/api/v1beta1/logging_types.go b/api/v1beta1/logging_types.go
@@ -23,6 +23,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const (
+	BuffersPath    = "/opt/fluent-bit/%s/buf"
+	PositionDbPath = "/opt/fluent-bit/%s/pos"
+)
+
 // EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
 // NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
 
@@ -114,7 +119,6 @@ func (l *Logging) SetDefaults() *Logging {
 			}
 
 			if copy.Spec.FluentdSpec.Metrics.PrometheusAnnotations {
-
 				copy.Spec.FluentdSpec.Annotations["prometheus.io/scrape"] = "true"
 
 				copy.Spec.FluentdSpec.Annotations["prometheus.io/path"] = copy.Spec.FluentdSpec.Metrics.Path
@@ -226,7 +230,6 @@ func (l *Logging) SetDefaults() *Logging {
 				copy.Spec.FluentbitSpec.Metrics.Interval = "15s"
 			}
 			if copy.Spec.FluentbitSpec.Metrics.PrometheusAnnotations {
-
 				copy.Spec.FluentbitSpec.Annotations["prometheus.io/scrape"] = "true"
 				copy.Spec.FluentbitSpec.Annotations["prometheus.io/path"] = copy.Spec.FluentbitSpec.Metrics.Path
 				copy.Spec.FluentbitSpec.Annotations["prometheus.io/port"] = string(copy.Spec.FluentbitSpec.Metrics.Port)
@@ -235,7 +238,23 @@ func (l *Logging) SetDefaults() *Logging {
 		if copy.Spec.FluentbitSpec.MountPath == "" {
 			copy.Spec.FluentbitSpec.MountPath = "/var/lib/docker/containers"
 		}
-
+		if copy.Spec.FluentbitSpec.PositionDB == nil {
+			copy.Spec.FluentbitSpec.PositionDB = &KubernetesStorage{
+				HostPath: &v1.HostPathVolumeSource{
+					Path: fmt.Sprintf(PositionDbPath, copy.Name),
+				},
+			}
+		}
+		if copy.Spec.FluentbitSpec.BufferStorage.StoragePath == "" {
+			copy.Spec.FluentbitSpec.BufferStorage.StoragePath = "/buffers"
+		}
+		if copy.Spec.FluentbitSpec.BufferStorageVolume == nil {
+			copy.Spec.FluentbitSpec.BufferStorageVolume = &KubernetesStorage{
+				HostPath: &v1.HostPathVolumeSource{
+					Path: fmt.Sprintf(BuffersPath, copy.Name),
+				},
+			}
+		}
 	}
 	return copy
 }

diff --git a/api/v1beta1/zz_generated.deepcopy.go b/api/v1beta1/zz_generated.deepcopy.go
diff --git a/cmd/docs.go b/cmd/docs.go
@@ -70,7 +70,6 @@ func (d *doc) checkNodes(n ast.Node) bool {
 					d.append(fmt.Sprintf("| %s | %s | %s | %s | %s |", name, normaliseType(item.Type), required, def, com))
 				}
 			}
-
 		}
 	}
 
@@ -169,7 +168,6 @@ func main() {
 	}
 
 	index.generate()
-
 }
 
 func getPrefixedLine(origin, expression string) string {
@@ -249,7 +247,6 @@ func getValuesFromItem(item *ast.Field) (name, comment, def, required string) {
 	}
 
 	return nameResult, getLink(commentWithDefault), "-", required
-
 }
 
 func getDocumentParser(file plugin) *doc {

diff --git a/config/crd/bases/logging.banzaicloud.io_loggings.yaml b/config/crd/bases/logging.banzaicloud.io_loggings.yaml
@@ -47,6 +47,53 @@ spec:
                   additionalProperties:
                     type: string
                   type: object
+                bufferStorage:
+                  description: BufferStorage is the Service Section Configuration
+                    of fluent-bit
+                  properties:
+                    storage.backlog.mem_limit:
+                      description: If storage.path is set, Fluent Bit will look for
+                        data chunks that were not delivered and are still in the storage
+                        layer, these are called backlog data. This option configure
+                        a hint of maximum value of memory to use when processing these
+                        records. (default:5M)
+                      type: string
+                    storage.checksum:
+                      description: Enable the data integrity check when writing and
+                        reading data from the filesystem. The storage layer uses the
+                        CRC32 algorithm. (default:Off)
+                      type: string
+                    storage.path:
+                      description: Set an optional location in the file system to
+                        store streams and chunks of data. If this parameter is not
+                        set, Input plugins can only use in-memory buffering.
+                      type: string
+                    storage.sync:
+                      description: Configure the synchronization mode used to store
+                        the data into the file system. It can take the values normal
+                        or full. (default:normal)
+                      type: string
+                  type: object
+                bufferStorageVolume:
+                  properties:
+                    host_path:
+                      description: Represents a host path mapped into a pod. Host
+                        path volumes do not support ownership management or SELinux
+                        relabeling.
+                      properties:
+                        path:
+                          description: 'Path of the directory on the host. If the
+                            path is a symlink, it will follow the link to the real
+                            path. More info: https://kubernetes.io/docs/concepts/storage/volumes#hostpath'
+                          type: string
+                        type:
+                          description: 'Type for HostPath Volume Defaults to "" More
+                            info: https://kubernetes.io/docs/concepts/storage/volumes#hostpath'
+                          type: string
+                      required:
+                      - path
+                      type: object
+                  type: object
                 customConfigSecret:
                   type: string
                 filterKubernetes:
@@ -278,6 +325,10 @@ spec:
                     Tag_Regex:
                       description: Set a regex to extract fields from the file.
                       type: string
+                    storage.type:
+                      description: Specify the buffering mechanism to use. It can
+                        be memory or filesystem. (default:memory)
+                      type: string
                   type: object
                 metrics:
                   description: Metrics defines the service monitor endpoints

diff --git a/docs/crds.md b/docs/crds.md
@@ -125,9 +125,11 @@ spec:
 | tolerations | [Toleration](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#toleration-v1-core) | {} | Pod toleration |
 | metrics | [Metrics](./logging-operator-monitoring.md#metrics-variables) | {} | Metrics defines the service monitor endpoints |
 | security | [Security](./security#security-variables) | {} | Security defines Fluentd, Fluentbit deployment security properties |
-| position_db |  [KubernetesStorage](#KubernetesStorage) | nil | Add position db storage support |
+| position_db |  [KubernetesStorage](#KubernetesStorage) | nil | Add position db storage support. If nothing is configured a `hostPath` volume is used with the path `/opt/fluent-bit/<name of the logging CR>/pos`. |
 | inputTail | [InputTail](./fluentbit.md#tail-inputtail) | {} | The tail input plugin allows to monitor one or several text files.  |
 | filterKubernetes | [FilterKubernetes](./fluentbit.md#kubernetes-filterkubernetes) | {} | Fluent Bit Kubernetes Filter allows to enrich your log files with Kubernetes metadata. |
+| bufferStorage | [BufferStorage](./fluentbit.md#bufferstorage) |  | Buffer Storage configures persistent buffer to avoid losing data in case of a failure |
+| bufferStorageVolume | [KubernetesStorage](#KubernetesStorage) | nil | Volume definition for the Buffer Storage. If nothing is configured a `hostPath` volume is used with the path `/opt/fluent-bit/<name of the logging CR>/buf`. |
 | customConfigSecret | string | "" | Custom secret to use as fluent-bit config.<br /> It must include all the config files necessary to run fluent-bit (_fluent-bit.conf_, _parsers*.conf_) |
 
 **`logging` with custom fluent-bit annotations** 

diff --git a/docs/fluentbit.md b/docs/fluentbit.md
@@ -81,6 +81,7 @@ The plugin supports the following configuration parameters:
 
 | Key | Description | Default |
 | :--- | :--- | :--- |
+| storage.type | Specify the buffering mechanism to use. It can be memory or filesystem. | memory |
 | Buffer\_Chunk\_Size | Set the initial buffer size to read files data. This value is used too to increase buffer size. The value must be according to the [Unit Size](../configuration/unit_sizes.md) specification. | 32k |
 | Buffer\_Max\_Size | Set the limit of the buffer size per monitored file. When a buffer needs to be increased \(e.g: very long lines\), this value is used to restrict how much the memory buffer can grow. If reading a file exceed this limit, the file is removed from the monitored file list. The value must be according to the [Unit Size](../configuration/unit_sizes.md) specification. | Buffer\_Chunk\_Size |
 | Path | Pattern specifying a specific log files or multiple ones through the use of common wildcards. |  |
@@ -102,4 +103,35 @@ The plugin supports the following configuration parameters:
 | Parser\_Firstline | Name of the parser that matchs the beginning of a multiline message. Note that the regular expression defined in the parser must include a group name \(named capture\) |  |
 | Parser\_N | Optional-extra parser to interpret and structure multiline entries. This option can be used to define multiple parsers, e.g: Parser\_1 ab1,  Parser\_2 ab2, Parser\_N abN. |  |
 | Docker\_Mode | If enabled, the plugin will recombine split Docker log lines before passing them to any parser as configured above. This mode cannot be used at the same time as Multiline. | Off |
-| Docker\_Mode\_Flush | Wait period time in seconds to flush queued unfinished split lines. | 4 |
+| Docker\_Mode\_Flush | Wait period time in seconds to flush queued unfinished split lines. | 4 |
+
+## Buffering
+
+### BufferStorage
+A mechanism to place processed data into a temporal location until is ready to be shipped. [More Info](https://docs.fluentbit.io/manual/configuration/buffering)
+
+
+| Key | Description | Default | 
+| :--- | :--- | :--- |
+| storage.path | Set an optional location in the file system to store streams and chunks of data. If this parameter is not set, Input plugins can only use in-memory buffering. | |
+| storage.sync | Configure the synchronization mode used to store the data into the file system. It can take the values normal or full. | normal |
+| storage.checksum | Enable the data integrity check when writing and reading data from the filesystem. The storage layer uses the CRC32 algorithm. | Off |
+| storage.backlog.mem_limit | If storage.path is set, Fluent Bit will look for data chunks that were not delivered and are still in the storage layer, these are called backlog data. This option configure a hint of maximum value of memory to use when processing these records. | 5M |
+
+
+#### Default configuration
+
+If nothing is set, by default it configures the `storage.path` explicitly to use `/buffers` and leaves fluent-bit defaults for the other options. 
+
+```
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Logging
+metadata:
+  name: default-logging-simple
+spec:
+  fluentd: {}
+  fluentbit:
+    bufferStorage:
+       storage.path: /buffers
+  controlNamespace: logging
+```
diff --git a/pkg/k8sutil/resource.go b/pkg/k8sutil/resource.go
@@ -122,7 +122,6 @@ func (r *GenericResourceReconciler) ReconcileResource(desired runtime.Object, de
 			}
 			log.Info("resource updated", "resource", desired.GetObjectKind().GroupVersionKind())
 		}
-
 	}
 	return nil
 }
@@ -141,7 +140,6 @@ func (r *GenericResourceReconciler) createIfNotExists(desired runtime.Object) (b
 		if err != nil {
 			return false, current, err
 		}
-
 	}
 	key, err := runtimeClient.ObjectKeyFromObject(current)
 	if err != nil {
@@ -183,9 +181,7 @@ func (r *GenericResourceReconciler) delete(desired runtime.Object) (bool, error)
 					"resource", desired.GetObjectKind().GroupVersionKind(), "type", reflect.TypeOf(desired))
 			}
 			return false, nil
-
 		}
-
 	}
 
 	key, err := runtimeClient.ObjectKeyFromObject(current)
@@ -200,9 +196,7 @@ func (r *GenericResourceReconciler) delete(desired runtime.Object) (bool, error)
 		} else {
 			log.Info("resource not found skipping delete", "resource", current.GetObjectKind().GroupVersionKind())
 			return false, nil
-
 		}
-
 	}
 	err = r.Client.Delete(context.TODO(), current)
 	if err != nil {

diff --git a/pkg/model/render/fluent.go b/pkg/model/render/fluent.go
@@ -78,7 +78,6 @@ func (f *FluentRender) indented(indent int, format string, values ...interface{}
 			fmt.Fprintln(f.Out, "")
 		}
 	}
-
 }
 
 func tag(tag string) string {

diff --git a/pkg/model/render/fluent_test.go b/pkg/model/render/fluent_test.go
@@ -31,7 +31,6 @@ import (
 )
 
 func TestRenderDirective(t *testing.T) {
-
 	var tests = []struct {
 		name      string
 		directive types.Directive
@@ -266,7 +265,6 @@ func TestRenderDirective(t *testing.T) {
 				t.Errorf("[%s] Result does not match (-actual vs +expected):\n%v", test.name, diff.LineDiff(a, e))
 			}
 		}
-
 	}
 }
 
@@ -492,7 +490,7 @@ func TestRenderS3(t *testing.T) {
 	}
 	for _, item := range table {
 		t.Logf("> %s\n", item.name)
-		err := ValidateRenderS3(t, item.s3Config, item.expected)
+		err := ValidateRenderS3(t, &item.s3Config, item.expected)
 		if item.err != "" {
 			if err == nil {
 				t.Errorf("expected error: %s", item.err)
@@ -510,7 +508,7 @@ func TestRenderS3(t *testing.T) {
 	}
 }
 
-func ValidateRenderS3(t *testing.T, s3Config output.S3OutputConfig, expected string) error {
+func ValidateRenderS3(t *testing.T, s3Config plugins.DirectiveConverter, expected string) error {
 	system := types.NewSystem(toDirective(t, input.NewTailInputConfig("input.log")), types.NewRouter("test"))
 
 	s3Plugin, err := s3Config.ToDirective(secret.NewSecretLoader(nil, "", "", nil), "test")

diff --git a/pkg/model/types/stringmaps.go b/pkg/model/types/stringmaps.go
@@ -194,7 +194,6 @@ func (s *StructToStringMapper) fillMap(value reflect.Value, out map[string]strin
 					}
 				}
 			}
-
 		}
 	}
 	return multierror
-Original file line number
+Diff line change
@@ Expand Up / @@ -70,7 +70,6 @@ func (d *doc) checkNodes(n ast.Node) bool { @@
     					d.append(fmt.Sprintf("| %s | %s | %s | %s | %s |", name, normaliseType(item.Type), required, def, com))
     				}
     			}
     		}
     	}
@@ Expand Down Expand Up / @@ -169,7 +168,6 @@ func main() { @@
     	}
     	index.generate()
     }
     func getPrefixedLine(origin, expression string) string {
@@ Expand Down Expand Up @@
     	}
     	return nameResult, getLink(commentWithDefault), "-", required
     }
     func getDocumentParser(file plugin) *doc {
@@ Expand Down @@