Rabbit MQ - Fluentd - Opensearch

Rabbit MQ와 Opensearch는 helm을 통해 설치

helm repo add my-repo https://charts.bitnami.com/bitnami

helm install my-release my-repo/rabbitmq

 

helm repo add opensearch https://opensearch-project.github.io/helm-charts/

helm install my-deployment opensearch/opensearch

helm install my-release opensearch/opensearch-dashboards

 

RabbitMQ는 exchange와 queue를 생성해 bind 시켜준 상황

fluentd는 아래 yaml 파일 통해 설정

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluentd
  namespace: oscmp-dev

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluentd
rules:
- apiGroups:
  - ""
  resources:
  - pods
  - namespaces
  verbs:
  - get
  - list
  - watch
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: fluentd
roleRef:
  kind: ClusterRole
  name: fluentd
  apiGroup: rbac.authorization.k8s.io
subjects:
- kind: ServiceAccount
  name: fluentd
  namespace: oscmp-dev

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fluentd
  namespace: oscmp-dev
  labels:
    k8s-app: fluentd-logging
    version: v1
spec:
  replicas: 1
  selector:
    matchLabels:
      k8s-app: fluentd-logging
      version: v1
  template:
    metadata:
      labels:
        k8s-app: fluentd-logging
        version: v1
    spec:
      serviceAccount: fluentd
      serviceAccountName: fluentd
      containers:
      - name: fluentd
        image: fluentd-rabbit-open:latest
        imagePullPolicy: IfNotPresent
        env:
          - name: K8S_NODE_NAME
            valueFrom:
              fieldRef:
                fieldPath: spec.nodeName
          - name:  FLUENT_OPENSEARCH_HOST
            value: "opensearch-cluster-master.default.svc.cluster.local"
          - name:  FLUENT_OPENSEARCH_PORT
            value: "9200"
          - name: FLUENT_OPENSEARCH_SCHEME
            value: "https"
          # Option to configure opensearch plugin with self signed certs
          # ================================================================
          - name: FLUENT_OPENSEARCH_SSL_VERIFY
            value: "false"
          # Option to configure opensearch plugin with tls
          # ================================================================
          - name: FLUENT_OPENSEARCH_SSL_VERSION
            value: "TLSv1_2"
          - name: FLUENTD_SYSTEMD_CONF
            value: disable
          - name: FLUENT_OPENSEARCH_USER
            value: "admin"
          - name: FLUENT_OPENSEARCH_PASS
            value: "admin"
          - name: RABBITMQ_HOST
            value: "my-release-rabbitmq.default.svc.cluster.local"
          - name: RABBITMQ_PASS
            value: "admin"
          - name: RABBITMQ_USER
            value: "admin"
          - name: RABBITMQ_VHOST
            value: "/"
        resources:
          limits:
            memory: 1000Mi
          requests:
            cpu: 100m
            memory: 200Mi
        volumeMounts:
        - name: config
          mountPath: /fluentd/etc/fluent.conf
          subPath: fluent.conf
        - name: buffer
          mountPath: /var/log/fluentd-buffer
#        command: ['/bin/sh', '-c', 'fluent-gem install fluent-plugin-rabbitmq']
      terminationGracePeriodSeconds: 30
      volumes:
      - name: config
        configMap:
          name: fluentd-config
      - name: buffer
        emptyDir: {}
      nodeSelector:
        node: node1

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-config
  namespace: oscmp-dev
data:
  tail_container_parse.conf: |-
    <parse>
      @type cri
    </parse>
  fluent.conf: |-
    <source>
      @type rabbitmq
      @label @rabbitmq_noti
      tag "notification"
      host "#{ENV['RABBITMQ_HOST']}"
      user "#{ENV['RABBITMQ_USER']}"
      pass "#{ENV['RABBITMQ_PASS']}"
      vhost "#{ENV['RABBITMQ_VHOST']}"
      exchange "tran.exchange"
      queue "tran.queue"
      routing_key "notification.*"
      heartbeat 10
      <parse>
        @type json
        time_key time
        keep_time_key false
      </parse>
    </source>

    <label @rabbitmq_noti>
      <match **>
        @type opensearch
        include_tag_key true
        host "#{ENV['FLUENT_OPENSEARCH_HOST']}"
        port "#{ENV['FLUENT_OPENSEARCH_PORT']}"
        path "#{ENV['FLUENT_OPENSEARCH_PATH']}"
        scheme "#{ENV['FLUENT_OPENSEARCH_SCHEME'] || 'https'}"
        ssl_verify "#{ENV['FLUENT_OPENSEARCH_SSL_VERIFY'] || 'false'}"
        ssl_version "#{ENV['FLUENT_OPENSEARCH_SSL_VERSION'] || 'TLSv1_2'}"
        user "#{ENV['FLUENT_OPENSEARCH_USER']}"
        password "#{ENV['FLUENT_OPENSEARCH_PASS']}"
        index_name "#{ENV['FLUENT_OPENSEARCH_INDEX_NAME'] || 'fluentd-2'}"
        logstash_dateformat "#{ENV['FLUENT_OPENSEARCH_LOGSTASH_DATEFORMAT'] || '%Y.%m.%d'}"
        <buffer>
          @type file
          path /var/log/fluentd-buffer/notification.buffer
          flush_mode interval
          retry_type exponential_backoff
          flush_thread_count 1
          flush_interval 15s
          retry_forever
          retry_max_interval 30
          chunk_limit_size 16MB
          total_limit_size 128MB
          overflow_action block
        </buffer>
      </match>
    </label>

 

fluent-gem install fluent-plugin-rabbitmq로 fluentd plugin 설치

rabbitmq UI를 사용하기 위해선 rabbitmq pod에 접속하여 rabbitmq-plugins enable rabbitmq_management명령어로 허용시킨 후 15672 port를 포트 포워딩

conf yaml에서 source가 input, match가 output으로 판단하며, rabbitmq를 받아서 opensearch로 전달하도록 구성

fluentd를 설치하면 queue에 연결된 consumer가 늘어나는 것을 확인할 수 있기에 연결되었다고 판단

삭제하면 cosumer가 사라짐

fluent-2라는 이름의 index로 지정하였기 때문에 해당 이름으로 데이터가 들어올 경우 opensearch에서 확인 할 수 있게 됨 ( 인덱스를 지정하더라도 들어온 데이터가 아예 없으면 리스트에 생성되지 않음 )

fluentd-2로 생성되었기 때문에 fluentd가 수집한 정보는 맞으나, 안에 rabbitmq에서 전달 받은 메시지를 확인할 수 없었기 때문에 해결 필요

 

해결

opensearch에서 받는 형식의 log가 아니였기에 거부된 것

{"log":"notinoti ss\n","stream":"stdout","time":"2022-10-31T06:47:52.021674475Z"}

아래와 같은 형식으로 넣어본 결과 들어간 것을 확인할 수 있음

filter와 parse를 구성하여 내용에 대한 정리가 필요