Architecture Overview
The proposed solution utilizes a DaemonSet to run the log collector on every node in the cluster. This ensures that all pod outputs are captured locally before transmission. To prevent downstream components from becoming overwhelmed, logs are buffered into Apache Kafka topics, where they are segregated according to the originating namespace. This decoupling allows for independant scaling of ingestion versus storage layers.
Container Log Locations
Containerized aplications typically write standard output (stdout) and error streams to JSON files stored under /var/lib/docker/containers. If the Docker daemon configuration is customized, this path may vary. Kubernetes creates symbolic links within /var/log/containers and /var/log/pods to simplify access for collectors.
Standard naming conventions follow this pattern:
<PodName>_<Namespace>_<ContainerID>.log
For instance:
service-api_v1_2c4d5e6f7g8h.log
nginx-ingress_controller-9b3a2c1d0e9f.log
Regardless of the workload type (Deployment, StatefulSet, or Job), the presence of _namespace_ in the filename remains consistent, serving as a reliable identifier for categorization rules.
Message Broker Setup
Operator Installation
Strimzi provides a robust operator for managing Kafka clusters on Kubernetes. Depending on data volume, storage can be provisioned via NFS for smaller setups or local Persistent Volumes (PV) for high-throughput scenarios.
Helm Deployment
Download the chart archive and deploy using Helm:
tar -xvf strimzi-kafka-operator-helm-chart-0.35.0.tgz
cd charts/
helm install my-kafka-operator . --namespace kafka-system
Configuration Manifests
Extract the examples to define the cluster topology. Recommended configurations include kafka-persistent.yaml which provisions ZooKeeper and broker nodes with disk retention.
Inside the extracted directory:
root@node:~ # ls examples/kafka/
kafka-ephemeral.yaml kafka-jbod.yaml
kafka-ephemeral-single.yaml kafka-persistent.yaml
Storage Provisioning
PersistentVolumeClaims (PVCs) must be created prior to applying the cluster manifest to ensure persistent data storage. The example below defines resources for three ZooKeeper instances and three Kafka brokers using an NFS backend.
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: log-cluster-zookeeper-data-0
namespace: kafka-system
spec:
storageClassName: nfs-storage
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 50Gi
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: log-cluster-broker-data-0
namespace: kafka-system
spec:
storageClassName: nfs-storage
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 50Gi
# Additional PVCs required for other brokers/zookeepers
Monitoring Interface
A UI tool simplifies cluster management. Run the following container for dashboard access:
docker run -d \
--name kafka-monitor \
-p 9097:8080 \
-e KAFKA_CLUSTERS_NAME=prod-cluster \
-e KAFKA_CLUSTERS_BOOTSTRAPSERVERS=broker-list.kafka-system.svc:9092 \
dushixiang/kafka-ui:latest
Collector Configuration
Filebeat runs as a DaemonSet to scan node filesystems. Configuration requires specific RBAC permissions to read pods and namespaces.
Permissions
Define roles to allow listing and watching necessary resources:
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: filebeat-collector
subjects:
- kind: ServiceAccount
name: beat-user
namespace: kube-system
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: ClusterRole
name: filebeat-collector
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: filebeat-collector
rules:
- apiGroups: [""]
resources: ["namespaces", "pods", "nodes"]
verbs: ["get", "list", "watch"]
- apiGroups: ["apps"]
resources: ["replicasets"]
verbs: ["get", "list", "watch"]
Processing Pipeline
A ConfigMap holds the processing logic. Paths are matched by namespace patterns, and multiline regexes handle log aggregation. Custom JavaScript processors flatten metadata structures before sending to Kafka.
apiVersion: v1
kind: ConfigMap
metadata:
name: log-collector-config
namespace: kube-system
data:
filebeat.yml: |
filebeat.inputs:
- type: container
enabled: true
paths:
- /var/log/containers/*_prod_*log
fields:
stream_group: prod_logs
environment: production
multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
multiline.negate: true
multiline.match: after
processors:
- add_kubernetes_metadata:
host: ${NODE_NAME}
matchers:
- logs_path:
logs_path: "/var/log/containers/"
# JavaScript Logic to Normalize Data
- script:
lang: javascript
id: extract_time_info
tag: enable
source: |
function process(event) {
var msg = event.Get("message");
var timestamp_regex = /^(\[[^\]]+\]|\d{2}:\d{2}:\d{2})/;
var match = msg.match(timestamp_regex);
if (match) {
event.Put("raw_time", match[0]);
}
}
output.kafka:
hosts: ["my-kafka-cluster-headless.kafka-system.svc:9092"]
topic: '%{[fields.stream_group]}'
partition.round_robin:
reachable_only: true
compression: gzip
Apply manifests using:
kubectl apply -f filebeat-rbac.yaml
kubectl apply -f filebeat-daemonset.yaml
Verify running collectors:
kubectl get pods -n kube-system | grep filebeat
Stream Processing
Logstash transforms raw events before persisting them to the search engine.
Installation Steps
Install the RPM package on the processing node:
rpm -ivh logstash-8.x.x.rpm
systemctl enable logstash
export PATH=$PATH:/usr/share/logstash/bin
Transformation Rules
The pipeline consumes from Kafka, parses diverse log formats, and outputs to Elasticsearch.
input { kafka {
bootstrap_servers => "kafka-headless.kafka-system.svc:9092"
auto_offset_reset => "earliest"
topics => ["prod_logs", "dev_logs"]
consumer_threads => 2
codec => json
}}
filter { grok {
match => { "message" => [
"%{TIMESTAMP_ISO8601:ts} +%{LOGLEVEL:lvl} +%{DATA:class} - %{GREEDYDATA:text}",
"%{TIME:ts} \[%{DATA:thread}\] +%{LOGLEVEL:lvl}.*%{GREEDYDATA:text}"
]}
}
mutate {
remove_field => [ "agent", "ecs", "host" ]
rename => { "[kubernetes][container][image]" => "image_ref" }
}
date {
match => ["ts", "yyyy-MM-dd HH:mm:ss", "yyyy-MM-dd'T'HH:mm:ssZ"]
target => "@timestamp"
timezone => UTC
}}
output elasticsearch {
hosts => ["https://es-master-0:9200", "https://es-master-1:9200"]
index => "app-logs-%{+YYYY.MM.dd}"
user => "elastic"
password => "secure_password"
ssl => false
}
stdout { codec => rubydebug }
Start the service:
systemctl start logstash
Performance Optimization
Preprocessing at the agent layer significantly reduces load on the central index. By performing heavy lifting in Filebeat—such as removing redundant ECS fields and flattening nested objects—the payload size transmitted to Logstash and stored in Elasticsearch deccreases. Testing indicates that aggressive field pruning can reduce total storage footprint by nearly 50%. Additionally, maintaining version control over Filebeat YAML files aids in tracking configuration drift across different environments.