role: Agent
image:
  repository: "{{ .ImageRepo }}"
  tag: "{{ .ImageTag }}"
resources: {{ .Resources }}
podPriorityClassName: "{{ .PriorityClassName }}"
service:
  enabled: false
secrets:
  generic:
    awsAccessKeyId: "{{ .AccessKeyID }}"
    awsSecretAccessKey: "{{ .SecretAccessKey }}"
env:
  - name: AWS_ACCESS_KEY_ID
    valueFrom:
      secretKeyRef:
        key: awsAccessKeyId
        name: o11y-vector
  - name: AWS_SECRET_ACCESS_KEY
    valueFrom:
      secretKeyRef:
        key: awsSecretAccessKey
        name: o11y-vector
  - name: TENANT_ID
    value: "{{ .TenantID }}"
  - name: PROJECT_ID
    value: "{{ .ProjectID }}"
  - name: K8S_CLUSTER_NAME
    value: "{{ .ProjectID }}"
tolerations:
  - operator: Exists
containerPorts:
  - name: prom-exporter
    containerPort: 9598
    protocol: TCP
  - name: api
    containerPort: 8686
    protocol: TCP
readinessProbe:
  httpGet:
    path: /health
    port: api
  initialDelaySeconds: 30
  timeoutSeconds: 30
livenessProbe:
  httpGet:
    path: /health
    port: api
  initialDelaySeconds: 30
  timeoutSeconds: 30
podSecurityContext:
  runAsNonRoot: true
  runAsUser: 1000
  runAsGroup: 0
  supplementalGroups:
    - 190
securityContext:
  readOnlyRootFilesystem: true
  allowPrivilegeEscalation: false
  privileged: false
  capabilities:
    drop:
    - NET_RAW
initContainers:
  - name: volume-permission-fix
    image: "{{ .InitContainerImage }}"
    command: ['sh', '-c', 'chown -R 1000:2000 /vector-data-dir']
    volumeMounts:
    - name: data
      mountPath: /vector-data-dir
    securityContext:
      runAsNonRoot: false
      runAsUser: 0
      runAsGroup: 0
      readOnlyRootFilesystem: false
      allowPrivilegeEscalation: true
persistence:
  enabled: true
  hostPath:
    enabled: true
    path: "/var/lib/vector/02"
dataDir: /vector-data-dir
podAnnotations:
  prometheus.io/path: /metrics
  prometheus.io/port: "9598"
  prometheus.io/scrape: "true"

customConfig:
  api:
    address: 0.0.0.0:8686
    enabled: true
    playground: false
  data_dir: /vector-data-dir
  sinks:
    loki_pod_logs:
      batch:
        max_bytes: 10240
        timeout_secs: 1
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "{{ .LokiURL }}"
      inputs:
      - ensure_cluster_id
      - route_logs.infra_logs
      - ensure_component.dropped
      - route_logs._unmatched
      - ensure_cluster_id.dropped
      labels:
        source: o11y
        app: {{ "'{{ `{{ kubernetes.pod_labels.app }}` }}'" }}
        cluster: "{{ .ClusterName }}-cluster"
        cluster_env: "{{ .ClusterEnv }}"
        cluster_id: {{ "'{{ `{{ o11y.cluster_id }}` }}'" }}
        container: {{ "'{{ `{{ kubernetes.container_name }}` }}'" }}
        instance: {{ "'{{ `{{ kubernetes.pod_name }}` }}'" }}
        namespace: {{ "'{{ `{{ kubernetes.pod_namespace }}` }}'" }}
        node: {{ "'{{ `{{ kubernetes.pod_node_name }}` }}'" }}
        project_id: "{{ .ProjectID }}"
        release: {{ "'{{ `{{ kubernetes.pod_labels.release }}` }}'" }}
        stream: {{ "'{{ `{{ stream }}` }}'" }}
        tanent_id: "{{ .TenantID }}"
      out_of_order_action: accept
      type: loki
    loki_sys_logs:
      batch:
        max_bytes: 10240
        timeout_secs: 1
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "{{ .LokiURL }}"
      inputs:
      - journald
      labels:
        source: o11y
        cluster: "{{ .ClusterName }}"
        cluster_env: "{{ .ClusterEnv }}"
        node: {{ "'{{ `{{ host }}` }}'" }}
        project_id: "{{ .ProjectID }}"
        syslog: {{ "'{{ `{{ SYSLOG_IDENTIFIER }}` }}'" }}
        tanent_id: "{{ .TenantID }}"
      out_of_order_action: accept
      type: loki
    s3_infra_logs:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - enrich_infra_log_message
      key_prefix: 1/${TENANT_ID}/${PROJECT_ID}/k8s-infra/logs/{{ "{{ `{{ o11y.component }}` }}" }}/
      region: "{{ .Region }}"
      force_path_style: true
      type: aws_s3
    s3_journald_logs:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - journald
      key_prefix: 1/${TENANT_ID}/${PROJECT_ID}/k8s-infra/logs/journald/{{ "{{ `{{ host }}` }}" }}/
      region: "{{ .Region }}"
      force_path_style: true
      type: aws_s3
    s3_serverless_tidb_audit_logs:
      batch:
        max_bytes: 104857600
        timeout_secs: 120
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
      compression: none
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - ensure_serverless_audit_logs
      key_prefix: 0/${TENANT_ID}/auditlog/{{ "{{ `{{ serverless_tenant_id }}` }}" }}/{{ "{{ `{{ serverless_project_id }}` }}" }}/{{ "{{ `{{ serverless_cluster_id }}` }}" }}/{{ "{{ `{{ year }}` }}" }}/{{ "{{ `{{ month }}` }}" }}/{{ "{{ `{{ day }}` }}" }}/
      region: "{{ .Region }}"
      force_path_style: true
      tags:
        category: audit_logs
      type: aws_s3
    s3_serverless_tidb_slow_logs:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - ensure_serverless_slow_logs_empty_id
      key_prefix: 0/${TENANT_ID}/serverless/{{ "{{ `{{ serverless_tenant_id }}` }}" }}/{{ "{{ `{{ serverless_project_id }}` }}" }}/{{ "{{ `{{ serverless_cluster_id }}` }}" }}/logs/slowlog/
      region: "{{ .Region }}"
      force_path_style: true
      type: aws_s3
    s3_serverless_tidb_slow_logs_all_in_one:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - ensure_serverless_slow_logs_empty_id
      key_prefix: 0/${TENANT_ID}/serverless-all-in-one/{{ "{{ `{{ o11y.pod_namespace }}` }}" }}/logs/slowlog/
      region: "{{ .Region }}"
      force_path_style: true
      type: aws_s3
    s3_serverless_tidb_statements:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - ensure_serverless_statements
      key_prefix: 0/${TENANT_ID}/serverless/{{ "{{ `{{ serverless_tenant_id }}` }}" }}/{{ "{{ `{{ serverless_project_id }}` }}" }}/{{ "{{ `{{ serverless_cluster_id }}` }}" }}/logs/statement/
      region: "{{ .Region }}"
      force_path_style: true
      tags:
        category: statements
      type: aws_s3
    s3_serverless_tidb_statements_all_in_one:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - ensure_serverless_statements
      key_prefix: 0/${TENANT_ID}/serverless-all-in-one/{{ "{{ `{{ o11y.pod_namespace }}` }}" }}/logs/statement/
      region: "{{ .Region }}"
      force_path_style: true
      tags:
        category: statements
      type: aws_s3
    s3_tidb_cluster_logs:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - enrich_tidb_cluster_log_message
      key_prefix: 0/${TENANT_ID}/${PROJECT_ID}/{{ "{{ `{{ o11y.cluster_id }}` }}" }}/logs/{{ "{{ `{{ o11y.component }}` }}" }}/
      region: "{{ .Region }}"
      force_path_style: true
      type: aws_s3
    s3_tidb_gateway_access_logs:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - enrich_tidb_gateway_access_logs
      key_prefix: 1/${TENANT_ID}/${PROJECT_ID}/k8s-infra/logs/tidb-gateway-access-logs/
      region: "{{ .Region }}"
      force_path_style: true
      type: aws_s3
    s3_unmatched_kubernetes_logs:
      batch:
        max_bytes: 104857600
        timeout_secs: 180
      bucket: "{{ .Bucket }}"
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: json
      endpoint: "http://{{ .Bucket }}.oss-{{ .Region }}-internal.aliyuncs.com"
      inputs:
      - ensure_component.dropped
      - route_logs._unmatched
      - ensure_cluster_id.dropped
      - enrich_tidb_cluster_log_message.dropped
      - enrich_infra_log_message.dropped
      key_prefix: 1/${TENANT_ID}/${PROJECT_ID}/k8s-infra/logs/_unmatched/
      region: "{{ .Region }}"
      force_path_style: true
      type: aws_s3
    self_logs_sink:
      batch:
        max_bytes: 10240
        timeout_secs: 1
      buffer:
        max_size: 268435488
        type: disk
        when_full: drop_newest
      compression: gzip
      encoding:
        codec: text
      endpoint: "{{ .LokiURL }}"
      inputs:
      - self_logs
      labels:
        app: vector_log_agent
        cluster: "{{ .ClusterName }}"
        cluster_env: "{{ .ClusterEnv }}"
        container: {{ "'{{ `{{ kubernetes.container_name }}` }}'" }}
        instance: {{ "'{{ `{{ kubernetes.pod_name }}` }}'" }}
        namespace: {{ "'{{ `{{ kubernetes.pod_namespace }}` }}'" }}
        node: {{ "'{{ `{{ kubernetes.pod_node_name }}` }}'" }}
        project_id: "{{ .ProjectID }}"
        release: {{ "'{{ `{{ kubernetes.pod_labels.release }}` }}'" }}
        stream: {{ "'{{ `{{ stream }}` }}'" }}
        tanent_id: "{{ .TenantID }}"
      out_of_order_action: accept
      type: loki
    self_metrics_sink:
      address: 0.0.0.0:9598
      inputs:
      - self_metrics
      type: prometheus_exporter
  sources:
    all_logs:
      delay_deletion_ms: 10000
      glob_minimum_cooldown_ms: 30000
      type: kubernetes_logs
    journald:
      type: journald
    self_logs:
      type: internal_logs
    self_metrics:
      type: internal_metrics
  transforms:
    enrich_infra_log_message:
      drop_on_error: true
      inputs:
      - route_logs.infra_logs
      reroute_dropped: true
      source: |-
        t = to_string!(.timestamp)
        node = .kubernetes.pod_node_name
        pod_namespace = .kubernetes.pod_namespace
        pod_name = .kubernetes.pod_name
        container_name = .kubernetes.container_name
        prefix = join!([t, node, pod_namespace, pod_name, container_name], ";")
        new_message, err = prefix + " " + .message
        if err == null {
          .message = new_message
        }
      type: remap
    enrich_tidb_cluster_log_message:
      drop_on_error: true
      inputs:
      - ensure_cluster_id
      reroute_dropped: true
      source: |-
        t = to_string!(.timestamp)
        node = .kubernetes.pod_node_name
        pod_namespace = .kubernetes.pod_namespace
        pod_name = .kubernetes.pod_name
        container_name = .kubernetes.container_name
        prefix = join!([t, node, pod_namespace, pod_name, container_name], ";")
        new_message, err = prefix + " " + .message
        if err == null {
          .message = new_message
        }
      type: remap
    enrich_tidb_gateway_access_logs:
      drop_on_error: true
      inputs:
      - tidb_gateway_access_logs
      reroute_dropped: true
      source: |-
        t = to_string!(.timestamp)
        node = .kubernetes.pod_node_name
        pod_namespace = .kubernetes.pod_namespace
        pod_name = .kubernetes.pod_name
        container_name = .kubernetes.container_name
        prefix = join!([t, node, pod_namespace, pod_name, container_name], ";")
        new_message, err = prefix + " " + .message
        if err == null {
          .message = new_message
        }
      type: remap
    ensure_cluster_id:
      drop_on_error: true
      inputs:
      - multiline_slow_logs
      - route_logs.tidb_cluster_general_logs
      reroute_dropped: true
      source: |-
        if !is_null(.kubernetes.pod_labels."tags.tidbcloud.com/cluster") {
          .o11y.cluster_id = .kubernetes.pod_labels."tags.tidbcloud.com/cluster"
        } else {
          parsed, err = parse_grok(.kubernetes.pod_namespace, "tidb%{GREEDYDATA:cluster_id}")
          if err == null {
            .o11y.cluster_id = parsed.cluster_id
          } else {
            .o11y.cluster_id = "unknown"
          }
          if .kubernetes.pod_namespace == "tidb-pool" || .kubernetes.pod_namespace == "vip-tidb-pool" || .kubernetes.pod_namespace == "tidb-serverless" || .kubernetes.pod_namespace == "tidb-admin" || .kubernetes.pod_namespace == "canary-tidb-pool" || .kubernetes.pod_namespace == "canary-vip-tidb-pool" || .kubernetes.pod_namespace == "super-vip-tidb-pool" || .kubernetes.pod_namespace == "canary-super-vip-tidb-pool" {
            .o11y.cluster_id = .kubernetes.pod_namespace
          }
        }
      type: remap
    ensure_component:
      drop_on_error: true
      inputs:
      - all_logs
      reroute_dropped: true
      source: |-
        if !is_null(.kubernetes.pod_labels."app.kubernetes.io/component") {
          .o11y.component = .kubernetes.pod_labels."app.kubernetes.io/component"
        } else if !is_null(.kubernetes.container_name) {
          .o11y.component = .kubernetes.container_name
        } else {
          .o11y.component = "unknown"
        }
        if !is_null(.kubernetes.container_name) {
          if .kubernetes.container_name == "slowlog" {
            .o11y.component = "slowlog"
          } else if .kubernetes.container_name == "statementlog" {
            .o11y.component = "statementlog"
          } else if .kubernetes.container_name == "auditlog" {
            .o11y.component = "auditlog"
          }
        }
      type: remap
    ensure_serverless_audit_logs:
      inputs:
      - route_logs.tidb_cluster_audit_logs
      source: |-
        audit = parse_json!(string!(.message))
        .serverless_tenant_id = string(audit.SERVERLESS_TENANT_ID) ?? "0"
        .serverless_project_id = string(audit.SERVERLESS_PROJECT_ID) ?? "0"
        .serverless_cluster_id = string(audit.SERVERLESS_CLUSTER_ID) ?? "0"
        parsed_timestamp, err = parse_timestamp(audit.TIME, format: "%Y/%m/%d %H:%M:%S.%f %z")
          if err == null {
          .year = format_timestamp!(parsed_timestamp, format: "%Y")
          .month = format_timestamp!(parsed_timestamp, format: "%m")
          .day = format_timestamp!(parsed_timestamp, format: "%d")
        } else {
          .year = 0
          .month = 0
          .day = 0
        }
      type: remap
    ensure_serverless_slow_logs:
      inputs:
      - route_logs.tidb_cluster_slow_logs
      source: |-
        if starts_with(string!(.message), "# Serverless_tenant_ID: ") {
          .serverless_tenant_id = strip_whitespace!(slice!(.message, 24))
        }
        if starts_with(string!(.message), "# Serverless_project_ID: ") {
          .serverless_project_id = strip_whitespace!(slice!(.message, 25))
        }
        if starts_with(string!(.message), "# Serverless_cluster_ID: ") {
          .serverless_cluster_id = strip_whitespace!(slice!(.message, 25))
        }
      type: remap
    ensure_serverless_slow_logs_empty_id:
      inputs:
      - multiline_slow_logs
      source: |-
        if .serverless_tenant_id == null {
          .serverless_tenant_id = 0
        }
        if .serverless_project_id == null {
          .serverless_project_id = 0
        }
        if .serverless_cluster_id == null {
          .serverless_cluster_id = 0
        }
        pod_namespace = string(.kubernetes.pod_namespace) ?? ""
        if pod_namespace == "" {
          .o11y.pod_namespace = "unknown"
        } else {
          .o11y.pod_namespace = pod_namespace
        }
      type: remap
    ensure_serverless_statements:
      inputs:
      - route_logs.tidb_cluster_statements
      source: |-
        stmt = parse_json!(string!(.message))
        .serverless_tenant_id = string(stmt.serverless_tenant_id) ?? "0"
        .serverless_project_id = string(stmt.serverless_project_id) ?? "0"
        .serverless_cluster_id = string(stmt.serverless_cluster_id) ?? "0"
        pod_namespace = string(.kubernetes.pod_namespace) ?? ""
        if pod_namespace == "" {
          .o11y.pod_namespace = "unknown"
        } else {
          .o11y.pod_namespace = pod_namespace
        }
      type: remap
    multiline_slow_logs:
      inputs:
      - ensure_serverless_slow_logs
      merge_strategies:
        message: concat_newline
      starts_when:
        source: |-
          res, err = starts_with(.message, "# Time: ")
          if err != null {
            false
          } else {
            res
          }
        type: vrl
      type: reduce
    route_logs:
      inputs:
      - ensure_component
      route:
        infra_logs: '!includes(["tidb","serverless-tidb","tikv","pd","ticdc","tiflash","serverless-tiflash","tiflash-learner","db-tidb-extra-access","backup","restore","tidb-lightning","dm-master","dm-worker","extra-tidb","slowlog","br"],
          .o11y.component)'
        tidb_cluster_audit_logs: .o11y.component == "auditlog"
        tidb_cluster_general_logs: includes(["tidb","serverless-tidb","tikv","pd","ticdc","tiflash","serverless-tiflash","tiflash-learner","db-tidb-extra-access","backup","restore","tidb-lightning","dm-master","dm-worker","extra-tidb","br"],
          .o11y.component)
        tidb_cluster_slow_logs: .o11y.component == "slowlog"
        tidb_cluster_statements: .o11y.component == "statementlog"
        tidb_gateway_logs: .o11y.component == "tidb-gateway"
      type: route
    tidb_gateway_access_logs:
      condition: contains(string!(.message), "gw_access_log")
      inputs:
      - route_logs.tidb_gateway_logs
      type: filter