---

dmitriev-aal@VDI-Dmitriev-A:~$ kubectl --context rcsdstbl -n vector get cm vector-agent \
>   -o jsonpath='{.data.vector\.yaml}'
api:
  address: 0.0.0.0:8686
  enabled: true
data_dir: /var/lib/vector
sinks:
  elasticsearch_journald_sink:
    api_version: v7
    auth:
      password: pe8jefowU5/huK4uc76fh+pHmRgYCdelW7MV3GMjCKc=
      strategy: basic
      user: fluentd
    batch:
      max_bytes: 10485760
    buffer:
      max_size: 314572800
      type: disk
      when_full: block
    bulk:
      index: journald-%Y.%m.%d
    compression: none
    endpoints:
    - https://elastic-data.rshbdev.ru
    inputs:
    - sanitizeJournalD
    mode: normal
    request:
      concurrency: adaptive
    tls:
      ca_file: /usr/share/cacert/rshb_bundle.pem
    type: elasticsearch
  elasticsearch_platform_sink:
    api_version: v7
    auth:
      password: pe8jefowU5/huK4uc76fh+pHmRgYCdelW7MV3GMjCKc=
      strategy: basic
      user: fluentd
    batch:
      max_bytes: 20971520
    buffer:
      max_size: 314572800
      type: disk
      when_full: block
    bulk:
      index: platform-%Y.%m.%d
    compression: none
    endpoints:
    - https://elastic-data.rshbdev.ru
    inputs:
    - platformLogs
    mode: normal
    request:
      concurrency: adaptive
    tls:
      ca_file: /usr/share/cacert/rshb_bundle.pem
    type: elasticsearch
  elasticsearch_sink:
    api_version: v7
    auth:
      password: pe8jefowU5/huK4uc76fh+pHmRgYCdelW7MV3GMjCKc=
      strategy: basic
      user: fluentd
    batch:
      max_bytes: 20971520
    buffer:
      max_size: 1610612736
      type: disk
      when_full: block
    bulk:
      index: logstash-%Y.%m.%d
    compression: none
    endpoints:
    - https://elastic-data.rshbdev.ru
    inputs:
    - throttle
    mode: normal
    request:
      concurrency: adaptive
    tls:
      ca_file: /usr/share/cacert/rshb_bundle.pem
    type: elasticsearch
  metrics:
    address: 0.0.0.0:9598
    inputs:
    - internal_metrics
    type: prometheus_exporter
sources:
  internal_metrics:
    scrape_interval_secs: 5
    type: internal_metrics
  journaldSource:
    batch_size: 8
    current_boot_only: false
    include_matches:
      _SYSTEMD_UNIT:
      - kubelet.service
      - containerd.service
      - nginx.service
      _TRANSPORT:
      - kernel
    since_now: true
    type: journald
  kubernetesLogs:
    extra_label_selector: app.kubernetes.io/name!=grafana-operator,type!=adapter
    glob_minimum_cooldown_ms: 100
    max_line_bytes: 8388608
    max_read_bytes: 8388608
    pod_annotation_fields:
      pod_labels: kubernetes.labels
      pod_namespace: kubernetes.namespace_name
      pod_node_name: kubernetes.node_name
    type: kubernetes_logs
transforms:
  parseJSON:
    inputs:
    - sanitizeEvents
    source: |
      json_message,err = parse_json(.message)

      ## https://gitlab.rshbdev.ru/rshbintech/integrations/ckpr/board/-/issues/2336
      ## Replace @timestamp with .timestamp, or use timestamp if it exists
      if exists(json_message.timestamp) {
        del(json_message.@timestamp)
      } else {
        json_message.timestamp = del(json_message.@timestamp)
      }

      if err == null {
        pod_name = .kubernetes.container_name
        if exists(.kubernetes.labels.app) && .kubernetes.labels.app != pod_name {
          app_name = .kubernetes.labels.app
          . = set!(value: ., path: ["parsed", pod_name, app_name ], data: json_message)
        } else {
          . = set!(value: ., path: ["parsed", pod_name ], data: json_message)
        }
        del(.message)
      }

      . = map_keys(., recursive: true) -> |key| { replace(key, ".", "_") }
    type: remap
  platformLogs:
    condition:
      source: .kubernetes.namespace_name == "platform"
      type: vrl
    inputs:
    - parseJSON
    type: filter
  sanitizeEvents:
    inputs:
    - sanitizeKubernetes
    source: |
      if .kubernetes.container_name == "calico-node" {
        if match(.message, r'INFO') ?? false {
          abort
        }
      }

      if .kubernetes.namespace_name == "topolvm-system" {
        if match(.message, r'"level":"info"') ?? false {
          abort
        }
      }
    type: remap
  sanitizeJournalD:
    inputs:
    - journaldSource
    source: |
      del(.PRIORITY)
      del(.SYSLOG_FACILITY)
      del(._BOOT_ID)
      del(._CAP_EFFECTIVE)
      del(._EXE)
      del(._COMM)
      del(._GID)
      del(._MACHINE_ID)
      del(._SELINUX_CONTEXT)
      del(._STREAM_ID)
      del(._SYSTEMD_INVOCATION_ID)
      del(.__MONOTONIC_TIMESTAMP)
      del(.__REALTIME_TIMESTAMP)
      del(._UID)
      . = map_keys(., recursive: true) -> |key| { replace(key, r'^_', "") }
    type: remap
  sanitizeKubernetes:
    inputs:
    - kubernetesLogs
    source: |
      del(.kubernetes.namespace_labels)
      del(.source_type)
      .@timestamp = del(.timestamp)
      .kubernetes.cluster_name = "rcsdstbl"
    type: remap
  sanitizeParsedJSON:
    inputs:
    - parseJSON
    source: |
      ## Drop nginx ingress-controller success requests
      if .kubernetes.container_name == "controller" && .kubernetes.namespace_name == "ingress-nginx" && exists(.parsed.controller.httpRequest.status) {
        if .parsed.controller.httpRequest.status < 299 ?? false {
          abort
        }
      }

      ## Drop metallb debug info
      if .kubernetes.container_name == "speaker" && .kubernetes.namespace_name == "metallb-system" && exists(.parsed.speaker.msg) {
        if match(.parsed.speaker.msg, r'DEBUG') ?? false {
          abort
        }
      }
      ## Drop blackbox-monitoring
      if .kubernetes.namespace_name == "blackbox-monitoring"{
        abort
      }
      ## Drop balanciaga logs
      if .kubernetes.namespace_name == "vector-business" && .kubernetes.pod_owner == "DaemonSet/balanciaga" {
        if match(.message, r'POST / HTTP/1.1') ?? false {
          abort
        }
      }
      if .kubernetes.namespace_name == "platform" {
        abort
      }

      ## SRE-371, режим structural: одно правило для любого сервиса без списка нарушителей.
      ## Корень сервиса — .parsed.<container>[.<app>], ровно так, как его кладёт parseJSON.
      ## Объект первого уровня под корнем -> JSON-строка, кроме структурных контейнеров логирующих
      ## библиотек из normalizeParsed.structuredKeys; внутри них всё глубже одного уровня -> JSON-строка.
      ## Массивы скаляров не трогаем, массивы с объектами внутри -> строка.
      ## Сервисы из normalizeParsed.excludeServices (по метке app или имени контейнера) не трогаем вовсе.
      excluded = includes(["esia-auth-service","es-opendistro","airflow","git-sync","git-sync-init"], .kubernetes.container_name) || includes(["esia-auth-service","es-opendistro","airflow","git-sync","git-sync-init"], .kubernetes.labels.app)
      if is_object(.parsed) && exists(.kubernetes.container_name) && !excluded {
        svc_path = ["parsed", replace(string!(.kubernetes.container_name), ".", "_")]
        if exists(.kubernetes.labels.app) && .kubernetes.labels.app != .kubernetes.container_name {
          svc_path = push(svc_path, replace(string!(.kubernetes.labels.app), ".", "_"))
        }
        svc, err = get(., svc_path)
        if err == null && is_object(svc) {
          svc = object!(svc)
          kept = {}
          if is_object(svc.properties) {
            kept.properties = map_values(object!(svc.properties)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.properties)
          }
          if is_object(svc.exception) {
            kept.exception = map_values(object!(svc.exception)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.exception)
          }
          if is_object(svc.mdc) {
            kept.mdc = map_values(object!(svc.mdc)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.mdc)
          }
          if is_object(svc.ecs) {
            kept.ecs = map_values(object!(svc.ecs)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.ecs)
          }
          if is_object(svc.log) {
            kept.log = map_values(object!(svc.log)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.log)
          }
          if is_object(svc.process) {
            kept.process = map_values(object!(svc.process)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.process)
          }
          if is_object(svc.service) {
            kept.service = map_values(object!(svc.service)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.service)
          }
          if is_object(svc.map) {
            kept.map = map_values(object!(svc.map)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.map)
          }
          if is_object(svc.EventId) {
            kept.EventId = map_values(object!(svc.EventId)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.EventId)
          }
          if is_object(svc.http_obj) {
            kept.http_obj = map_values(object!(svc.http_obj)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.http_obj)
          }
          if is_object(svc.httpRequest) {
            kept.httpRequest = map_values(object!(svc.httpRequest)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.httpRequest)
          }
          if is_object(svc.Headers) {
            kept.Headers = map_values(object!(svc.Headers)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.Headers)
          }
          if is_object(svc.error) {
            kept.error = map_values(object!(svc.error)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.error)
          }
          if is_object(svc.ExceptionDetail) {
            kept.ExceptionDetail = map_values(object!(svc.ExceptionDetail)) -> |v| {
              if is_object(v) || is_array(v) { encode_json(v) } else { v }
            }
            del(svc.ExceptionDetail)
          }
          svc = map_values(svc) -> |v| {
            if is_object(v) {
              encode_json(v)
            } else if is_array(v) && length(filter(array!(v)) -> |_i, e| { is_object(e) || is_array(e) }) > 0 {
              encode_json(v)
            } else {
              v
            }
          }
          . = set!(value: ., path: svc_path, data: merge(svc, kept))
        }
      }
      ## Правило 2: явный список путей относительно .parsed (normalizeParsed.stringifyPaths) -> JSON-строка целиком.
      ##   Список пересобирается по логам vector (grep mapper_parsing_exception), не по маппингу индекса.
      if is_object(.parsed."victoria-metrics-operator"."vmalert") || is_array(.parsed."victoria-metrics-operator"."vmalert") {
        .parsed."victoria-metrics-operator"."vmalert" = encode_json(.parsed."victoria-metrics-operator"."vmalert")
      }
      if is_object(.parsed."app"."coll-cti-module-service"."messageParams") || is_array(.parsed."app"."coll-cti-module-service"."messageParams") {
        .parsed."app"."coll-cti-module-service"."messageParams" = encode_json(.parsed."app"."coll-cti-module-service"."messageParams")
      }
      if is_object(.parsed."app"."axiom-api"."request_body") || is_array(.parsed."app"."axiom-api"."request_body") {
        .parsed."app"."axiom-api"."request_body" = encode_json(.parsed."app"."axiom-api"."request_body")
      }
      if is_object(.parsed."app"."marketrates"."Key") || is_array(.parsed."app"."marketrates"."Key") {
        .parsed."app"."marketrates"."Key" = encode_json(.parsed."app"."marketrates"."Key")
      }
      if is_object(.parsed."app"."ges-enterprise"."error") || is_array(.parsed."app"."ges-enterprise"."error") {
        .parsed."app"."ges-enterprise"."error" = encode_json(.parsed."app"."ges-enterprise"."error")
      }

      if exists(.parsed.app.@timestamp) { del(.parsed.app.@timestamp) }
    type: remap
  throttle:
    exclude:
      source: |
        .kubernetes.namespace_name == "kube-system" || .kubernetes.namespace_name == "istio-system" || .kubernetes.namespace_name == "isys-suz-rc" || .kubernetes.namespace_name == "isys-suz"
      type: vrl
    inputs:
    - sanitizeParsedJSON
    key_field: '{{ kubernetes.pod_name }}-{{ kubernetes.container_name }}'
    threshold: 100
    type: throttle
    window_secs: 10
dmitriev-aal@VDI-Dmitriev-A:~$