팀 프로젝트) FluentBit + Fluentd

Fluentbit 설정

namespace: efkstack

FluentBit 설정을 위한 configmap 파일

• 현재용

fluent-bit-config.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluent-bit-config
  namespace: efkstack
data:
  fluent-bit.conf: |
    [SERVICE]
        Flush                 1
        Daemon                Off
        Log_Level             info
        storage.path          /var/fluent-bit/storage/
        storage.sync          normal
        storage.checksum      off
        storage.max_chunks_up 256
        Parsers_File          parsers.conf

    # =======================
    # INPUTS (containerd/CRI)
    # =======================
    # 서비스 로그만: /var/log/containers/*_service_*.log
    [INPUT]
        Name                  tail
        Path                  /var/log/containers/*_service_*.log
        Tag                   kube.*
        Parser                cri
        Mem_Buf_Limit         10MB
        Skip_Long_Lines       On
        storage.type          filesystem
        DB                    /var/fluent-bit/storage/tail-service.db
        DB.sync               normal

    # 시스템 로그만: /var/log/containers/*_system_*.log
    [INPUT]
        Name                  tail
        Path                  /var/log/containers/*_system_*.log
        Tag                   kube.*
        Parser                cri
        Mem_Buf_Limit         10MB
        Skip_Long_Lines       On
        storage.type          filesystem
        DB                    /var/fluent-bit/storage/tail-system.db
        DB.sync               normal

    # ==========================
    # FILTERS (K8s meta + tagging)
    # ==========================
    # K8s 메타데이터 병합 (JSON이면 필드 병합)
    [FILTER]
        Name                  kubernetes
        Match                 kube.var.log.containers.*
        Kube_URL              https://kubernetes.default.svc:443
        Kube_Tag_Prefix       kube.var.log.containers.
        Merge_Log             On
        Keep_Log              Off
        K8S-Logging.Exclude   On
        Buffer_Size           0

    # 서비스: app=service → 최종 태그 service.application.log
    [FILTER]
        Name                  rewrite_tag
        Match                 kube.var.log.containers.service*
        Rule                  $kubernetes['labels']['app'] ^service$ service.application.log false
        Rule                  $log ^.*$ service.unmatched.log false

    # 시스템: system 라벨값에 따라 최종 태그 분기:
    [FILTER]
        Name                  rewrite_tag
        Match                 kube.var.log.containers.*system*
        Rule                  $kubernetes['labels']['system'] .*auth.* system.auth.log false
        Rule                  $kubernetes['labels']['system'] .*kmsg.* system.kmsg.log false
        Rule                  $log ^.*$ system.unmatched.log false

    [FILTER]
        Name                   modify
        Match                  *
        Remove                 kubernetes

    # =======================
    # OUTPUTS
    # =======================
    # 전부 Fluentd로 전달
    [OUTPUT]
        Name                  forward
        Match                 *
        Host                  fluentd-aggregator.efkstack.svc.cluster.local
        Port                  24224
        Retry_Limit           False
        net.keepalive         On
        net.connect_timeout   10
        net.dns.resolver      LEGACY
    [OUTPUT]
        Name                  stdout
        Match                 *  

  parsers.conf: |
    # CRI (containerd) 단일 라인 파서
    [PARSER]
        Name        cri
        Format      regex
        Regex       ^(?<time>[^ ]+) (?<stream>stdout|stderr) [^ ]* (?<log>.*)$
        Time_Key    time
        Time_Format %Y-%m-%dT%H:%M:%S.%L%z

    # (참고) JSON 파서가 필요하면 사용 가능
    [PARSER]
        Name        json
        Format      json

rewrite_tag 규칙의 평가 시점 문제

• rewrite_tag는 들어온 레코드의 문자열 표현을 기준으로 매칭한다.
• service의 경우 라벨이 단일 키(app: service)라서 내부 표현이 단순 → ^service$ 매칭 성공.
• system의 경우 라벨이 여러 개(pod-template-hash + system)가 붙으면서 내부적으로 파싱된 JSON Path 접근($kubernetes['labels']['system'])이 정확히 기대대로 해석되지 못하는 경우가 있다.
  그래서 부분 일치(.*auth.*)로만 잡히는 것처럼 보이는 것

FluentBit DaemonSet

• 모든 노드에 하나씩 붙어서 해당 노드에서 생산되는 로그 수집을 담당할 친구들
• 리소스 제한은 임의로 붙여뒀으므로 필요시 수정 요망

fluent-bit-deployment.yaml

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluent-bit
  namespace: efkstack
spec:
  selector:
    matchLabels:
      name: fluent-bit
  template:
    metadata:
      labels:
        name: fluent-bit
    spec:
      serviceAccountName: fluent-bit
      containers:
      - name: fluent-bit
        image: fluent/fluent-bit:2.2.2
        resources:
          requests:
            cpu: 100m
            memory: 100Mi
          limits:
            cpu: 500m
            memory: 500Mi
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: fluent-bit-config
          mountPath: /fluent-bit/etc/
        - name: storage
          mountPath: /var/fluent-bit/storage/
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: state
        emptyDir: {}
      - name: fluent-bit-config
        configMap:
          name: fluent-bit-config
      - name: storage
        hostPath:
          path: /var/lib/fluent-bit/storage/ # /var/log 대신 /var/lib 경로 사용 권장

 

RBAC(Role Based Access Control)

• fluentbit이 모든 stdout 로그가 아닌 서비스 로그만 수집하기 위해 kubernetes에서 네임스페이스, 파드 라벨 등의 메타데이터를 읽어 이를 기준으로 필터링해야 하는데 이 API서버에 가서 메타데이터를 읽을 권한을 주기 위한 설정
• ServiceAccount, ClusterRole, ClusterRoleBinding을 한 매니페스트에서 정의

fluent-bit-RBAC.yaml

# fluentbit-rbac.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluent-bit
  namespace: efkstack
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluent-bit-read
rules:
- apiGroups: [""]
  resources: ["namespaces", "pods"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluent-bit-read
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: fluent-bit-read
subjects:
- kind: ServiceAccount
  name: fluent-bit
  namespace: efkstack

 

Fluentd-config.yaml

• stdout으로 나오던걸 kafka로 연결 완성!! → kafka clustrer의 LB IP 사용
• flush 간격을 1초로 지정해서 실시간성 확보 (기본 60s)

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-config
  namespace: efkstack
data:
  fluent.conf: |
    ############################################################
    # 1) Fluent Bit → Fluentd 입력 (forward)
    ############################################################
    <source>
      @type forward
      port 24224
      bind 0.0.0.0
    </source>

    ############################################################
    # 2) 서비스 애플리케이션 로그 → Kafka 전송 (service-topic)
    ############################################################
    <match service.application.log>
      @type kafka2
      brokers 192.168.1.210:9094
      topic service-topic
      use_event_time true

      <format>
        @type json
      </format>

      <buffer>
        @type memory
        flush_interval 1s
        flush_at_shutdown true
        flush_thread_count 2
        chunk_limit_size 1m
        retry_max_interval 30
        retry_forever true
      </buffer>
    </match>

    ############################################################
    # 3-1) kmsg  애플리케이션 로그 → Kafka 전송 (system-kmsg-topic)
    ############################################################
    <match system.kmsg.log>
      @type kafka2
      brokers 192.168.1.210:9094
      topic system-kmsg-topic
      use_event_time true

      <format>
        @type json
      </format>

      <buffer>
        @type memory
        flush_interval 1s
        flush_at_shutdown true
        flush_thread_count 2
        chunk_limit_size 1m
        retry_max_interval 30
        retry_forever true
      </buffer>
    </match>

    ############################################################
    # 3-2) 시스템 로그 → AUTH (system-auth-topic)
    ############################################################
    <filter system.auth.log>
      @type parser
      key_name log
      reserve_data true
      remove_key_name_field true
      <parse>
        @type regexp
        expression /^(?<timestamp>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}\.\d+\+\d{2}:\d{2})\s+(?<host>\S+)\s+sshd\[(?<pid>\d+)\]:\s+(?<auth_result>Accepted|Failed)\s+password\s+for\s+(?:invalid user\s+)?(?<user>\S+)\s+from\s+(?<ip>[0-9\.]+)\s+port\s+(?<port>\d+)\s+ssh2$/
      </parse>
    </filter>
    
    <match system.auth.log>
      @type kafka2
      brokers 192.168.1.210:9094
      topic system-auth-topic
      use_event_time true

      <format>
        @type json
      </format>

      <buffer>
        @type memory
        flush_interval 1s
        flush_at_shutdown true
        flush_thread_count 2
        chunk_limit_size 1m
        retry_max_interval 30
        retry_forever true
      </buffer>
    </match>

    ############################################################
    # 4) 매칭 안 된 나머지 (디버깅용)
    ############################################################
    <match **>
      @type stdout
    </match>

fluentd-statefulset.yaml

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: fluentd-aggregator
  namespace: efkstack
spec:
  serviceName: fluentd-aggregator
  replicas: 2   # 안정성을 위해 2개 이상 권장
  selector:
    matchLabels:
      app: fluentd-aggregator
  template:
    metadata:
      labels:
        app: fluentd-aggregator
    spec:
      terminationGracePeriodSeconds: 30
      containers:
      - name: fluentd
        image: fluent/fluentd-kubernetes-daemonset:v1-debian-kafka
        ports:
        - containerPort: 24224
          name: forward
        - containerPort: 24224
          name: forward-udp
          protocol: UDP
        - containerPort: 2020
          name: monitor
        volumeMounts:
        - name: config
          mountPath: /fluentd/etc
        - name: buffer
          mountPath: /fluentd/buffer
      volumes:
      - name: config
        configMap:
          name: fluentd-config
  volumeClaimTemplates:
  - metadata:
      name: buffer
    spec:
      accessModes: [ "ReadWriteMany" ]
      storageClassName: nfs-client
      resources:
        requests:
          storage: 5Gi   # 필요에 따라 조정 가능
---
apiVersion: v1
kind: Service
metadata:
  name: fluentd-aggregator
  namespace: efkstack
spec:
  selector:
    app: fluentd-aggregator
  ports:
  - name: forward
    protocol: TCP
    port: 24224
    targetPort: 24224
  - name: forward-udp
    protocol: UDP
    port: 24224
    targetPort: 24224
  - name: monitor
    protocol: TCP
    port: 2020
    targetPort: 2020

 

fluentd-netpol.yaml

# fluentd-netpol.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-fluentbit-to-fluentd
  namespace: efkstack # Fluentd가 있는 네임스페이스
spec:
  # 이 정책은 fluentd-aggregator 레이블을 가진 파드에 적용됩니다.
  podSelector:
    matchLabels:
      app: fluentd-aggregator
  policyTypes:
  - Ingress
  # Ingress (들어오는 트래픽) 규칙
  ingress:
  # 아래 조건을 만족하는 트래픽을 허용합니다.
  - from:
    # 1. efkstack 네임스페이스 내부의 모든 파드
    - podSelector: {}
    # 2. (선택적) 다른 네임스페이스의 fluent-bit를 위해 추가
    - namespaceSelector:
        matchLabels:
          # 모든 네임스페이스에서 들어오는 것을 허용하려면, 
          # 각 네임스페이스에 특별한 레이블을 추가하고 여기에 명시하거나, 
          # 더 간단하게는 from 필드를 비워두어 모든 소스를 허용할 수 있습니다.
          # 여기서는 우선 같은 네임스페이스 내 통신을 보장합니다.
          kubernetes.io/metadata.name: efkstack
    ports:
    # fluentd의 24224 포트로 들어오는 TCP 트래픽을 허용합니다.
    - protocol: TCP
      port: 24224

---