From 68b2e2a1e11b777ff9f59ffba8dc67008a846442 Mon Sep 17 00:00:00 2001
From: Ardiea <mas48@mit.edu>
Date: Fri, 18 Feb 2022 11:55:54 -0500
Subject: [PATCH 1/2] Initial refactor of the vector configuration management
 code to support extra configuration files and global sink definitions.

Adding vault lookups for the loki configuration.

Initial commit of odlvideo vector monitoring configuration file.

Fixing data structure layout of odlvideo vector configs.

Initial commit of reddit vector monitoring configuration file.

Further development

Finished validation of monitoring configuration for all phase one apps.
---
 pillar/vector/cas.sls       | 145 +++++++++++++++++++++++++++++++
 pillar/vector/init.sls      |  80 ++++++++++++-----
 pillar/vector/odlvideo.sls  | 140 ++++++++++++++++++++++++++++++
 pillar/vector/rabbitmq.sls  |  51 +++++++++++
 pillar/vector/reddit.sls    | 165 ++++++++++++++++++++++++++++++++++++
 pillar/vector/xqwatcher.sls | 115 +++++++++++++++++++++++++
 salt/vector/configure.sls   |  17 +++-
 7 files changed, 689 insertions(+), 24 deletions(-)
 create mode 100644 pillar/vector/cas.sls
 create mode 100644 pillar/vector/odlvideo.sls
 create mode 100644 pillar/vector/rabbitmq.sls
 create mode 100644 pillar/vector/reddit.sls
 create mode 100644 pillar/vector/xqwatcher.sls

diff --git a/pillar/vector/cas.sls b/pillar/vector/cas.sls
new file mode 100644
index 000000000..1e06b7576
--- /dev/null
+++ b/pillar/vector/cas.sls
@@ -0,0 +1,145 @@
+vector:
+  extra_configurations:
+  - name: cas_logs
+    content:
+      log_schema:
+        timestamp_key: vector_timestamp
+        host_key: log_host
+      sources:
+        collect_cas_nginx_access_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/nginx/access.log
+        collect_cas_nginx_error_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/nginx/error.log
+        collect_cas_application_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /opt/log/django.log
+          multiline:
+            start_pattern: '^\['
+            condition_pattern: '^\['
+            mode: 'halt_before'
+            timeout_ms: 5000
+        collect_auth_logs:
+        {{ salt.pillar.get('vector:base_auth_log_collection')|yaml(False)|indent(8) }}
+      transforms:
+        # Transforms for NGINX logs
+        parse_cas_nginx_access_logs:
+          type: remap
+          inputs:
+          - 'collect_cas_nginx_access_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^time=(?P<time>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+\d{2}:\d{2})\sclient=(?P<client>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})\smethod=(?P<method>\S*)\srequest="(?P<request>.*)"\srequest_length=(?P<request_length>\d+)\sstatus=(?P<status>\d+)\sbytes_sent=(?P<bytes_sent>\d+)\sbody_bytes_sent=(?P<body_bytes_sent>\d+)\sreferer=(?P<referer>.*)\suser_agent="(?P<user_agent>.+)"\supstream_addr=(?P<upstream_addr>.+)\supstream_status=(?P<upstream_status>.+)\srequest_time=(?P<request_time>.+)\srequest_id=(?P<request_id>\w+)\supstream_response_time=(?P<upstream_response_time>.+)\supstream_connect_time=(?P<upstream_connect_time>.+)\supstream_header_time=(?P<upstream_header_time>.*)$')
+            if err != null {
+              .parse_error = err
+            }
+            err = null
+            . = merge(., parsed)
+            .log_process = "nginx"
+            .log_type = "cas.nginx.access"
+            .environment = "${ENVIRONMENT}"
+            parsed_bs, err = to_int(.bytes_sent)
+            if err == null {
+              .bytes_sent = parsed_bs
+            }
+            err = null
+            parsed_bbs, err = to_int(.body_bytes_sent)
+            if err == null {
+              .body_bytes_sent = parsed_bbs
+            }
+            err = null
+            parsed_rl, err = to_int(.request_length)
+            if err == null {
+              .request_length = parsed_rl
+            }
+            err = null
+            parsed_rt, err = to_float(.request_time)
+            if err == null {
+              .request_time = parsed_rt
+            }
+            err = null
+            parsed_status, err = to_int(.status)
+            if err == null {
+              .status = parsed_status
+            }
+            err = null
+            parsed_usct, err = to_float(.upstream_connect_time)
+            if err == null {
+              .upstream_connect_time = parsed_usct
+            }
+            err = null
+            parsed_usht, err = to_float(.upstream_header_time)
+            if err == null {
+              .upstream_header_time = parsed_usht
+            }
+            err = null
+            parsed_uprt, err = to_float(.upstream_response_time)
+            if err == null {
+              .upstream_response_time = parsed_uprt
+            }
+            err = null
+            parsed_ups, err = to_int(.upstream_response)
+            if err == null {
+              .upstream_status = parsed_ups
+            }
+            err = null
+        filter_healthchecks_cas_nginx_access_logs:
+          inputs:
+          - 'parse_cas_nginx_access_logs'
+          type: filter
+          condition: '! contains!(.http_user_agent, "ELB-HealthChecker")'
+        parse_cas_nginx_error_logs:
+          type: remap
+          inputs:
+          - 'collect_cas_nginx_error_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^(?P<time>\d{4}/\d{2}/\d{2}\s\d{2}:\d{2}:\d{2})\s\[(?P<severity>.*)\]\s(?P<pid>\d*)#(?P<tid>\d*):\s\*(?P<cid>\d*)\s(?P<message>.*),\sclient:\s(?P<client>.*),\sserver:(?P<server>.*)(?P<additional_content>.*)$')
+            . = merge(., parsed)
+            if err != null {
+              .parse_error = err
+            }
+            .log_process = "nginx"
+            .log_type = "cas.nginx.error"
+            .environment = "${ENVIRONMENT}"
+        parse_cas_application_logs:
+          type: remap
+          inputs:
+          - 'collect_cas_application_logs'
+          source: |
+            parsed = parse_regex!(.message, r'^\[(?P<time>\d{4}-\d{2}-\d{2}\w+:\d{2}:\d{2})\] (?P<log_level>\w+) \[(?P<module_name>[a-zA-Z0-9-_.]+):(?P<line_number>\d+)\] (?P<message>.*)')
+            if err != null {
+              .parse_error = err
+            }
+            . = merge(., parsed)
+            .log_process = "cas"
+            .log_type = "cas.application"
+            .environment = "${ENVIRONMENT}"
+        enrich_cas_application_logs:
+          type: aws_ec2_metadata
+          inputs:
+          - 'parse_cas_application_logs'
+          namespace: ec2
+        parse_auth_logs:
+          {{ salt.pillar.get('vector:base_auth_log_parse_source')|yaml(False)|indent(10) }}
+      sinks:
+        ship_cas_logs_to_grafana_cloud:
+          inputs:
+          - 'filter_healthchecks_cas_nginx_access_logs'
+          - 'parse_cas_nginx_error_logs'
+          - 'enrich_cas_application_logs'
+          - 'parse_auth_logs'
+          type: loki
+          labels:
+            application: cas
+            environment: ${ENVIRONMENT}
+            service: cas
+          {{ salt.pillar.get('vector:base_loki_configuration')|yaml(False)|indent(10) }}
diff --git a/pillar/vector/init.sls b/pillar/vector/init.sls
index fc8fb38ad..ee24821e3 100644
--- a/pillar/vector/init.sls
+++ b/pillar/vector/init.sls
@@ -1,42 +1,80 @@
 {% set ENVIRONMENT = salt.grains.get('environment', 'dev') %}
+
 vector:
-  configuration:
+  base_auth_log_collection:
+    type: file
+    file_key: log_file
+    read_from: end
+    include:
+    - /var/log/auth.log
+  base_auth_log_parse_source:
+    type: remap
+    inputs:
+    - 'collect_auth_logs'
+    source: |
+      parsed, err = parse_syslog(.message)
+      if err != null {
+        .parse_error = err
+      }
+      . = merge(., parsed)
+      .log_process = "authlog"
+      .environment = "${ENVIRONMENT}"
+
+  # These two are intentionally incomplete sink configurations. The type, inputs, and labels 
+  #  need to be provided on a configuration-by-configuration basis.
+  base_loki_configuration:
+    auth:
+      strategy: basic
+      password: __vault__::secret-operations/global/grafana-cloud-credentials>data>api_key
+      user: __vault__::secret-operations/global/grafana-cloud-credentials>data>loki_user
+    endpoint: https://logs-prod-us-central1.grafana.net
+    encoding:
+      codec: json
+    out_of_order_action: rewrite_timestamp
+  base_cortex_configuration:
+    endpoint: https://prometheus-prod-10-prod-us-central-0.grafana.net/api/prom/push
+    healthcheck: false
+    auth:
+      strategy: basic
+      user: __vault__::secret-operations/global/grafana-cloud-credentials>data>prometheus_user
+      password: __vault__::secret-operations/global/grafana-cloud-credentials>data>api_key
+
+  # By default, there are no extra vector configurations to add
+  extra_configurations: []
+
+  # Call out host metrics in their own area because they will be enabled globally
+  host_metrics_configuration:
     sources:
-      host_metrics:
+      collect_host_metrics:
         type: host_metrics
         scrape_interval_secs: 60
         collectors:
           - cpu
-          - disk
           - filesystem
           - load
           - host
           - memory
           - network
-
     transforms:
-      host_metrics_relabel:
+      cleanup_host_metrics:
         type: remap
         inputs:
-        - host_metrics
+        - 'collect_host_metrics'
         source: |
-          .tags.job = "integrations/linux_host"
-
-      add_labels_to_metrics:
+          # Drop all the not-real filesystems metrics
+          abort_match_filesystem, err = !(match_any(.tags.filesystem, [r'ext.', r'btrfs', r'xfs']))
+          if abort_match_filesystem {
+            abort
+          }
+      add_labels_to_host_metrics:
         type: remap
         inputs:
-        - '*metrics_relabel'
+        - 'cleanup_host_metrics'
         source: |
-          .tags.environment = "{{ ENVIRONMENT }}"
-
+          .tags.environment = "${ENVIRONMENT}"
+          .tags.job = "integrations/linux_host"
     sinks:
-      grafana_cortex_metrics:
+      ship_host_metrics_to_grafana_cloud:
         inputs:
-        - add_labels_to_metrics
-        type: prometheus_remote_write
-        endpoint: https://prometheus-prod-10-prod-us-central-0.grafana.net/api/prom/push
-        healthcheck: false
-        auth:
-          strategy: basic
-          user: __vault__::secret-operations/global/grafana-cloud-credentials>data>prometheus_user
-          password: __vault__::secret-operations/global/grafana-cloud-credentials>data>api_key
+        - 'add_labels_to_host_metrics'
+        {{ salt.pillar.get('vector:base_cortex_configuration')|yaml(False)|indent(8) }}
diff --git a/pillar/vector/odlvideo.sls b/pillar/vector/odlvideo.sls
new file mode 100644
index 000000000..f7c461953
--- /dev/null
+++ b/pillar/vector/odlvideo.sls
@@ -0,0 +1,140 @@
+vector:
+  extra_configurations:
+  - name: odlvideo_logs
+    content:
+      log_schema:
+        timestamp_key: vector_timestamp
+        host_key: log_host
+      sources:
+        collect_odlvideo_nginx_access_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/nginx/access.log
+        collect_odlvideo_nginx_error_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/nginx/error.log
+        collect_odlvideo_django_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/odl-video/django.log
+        collect_auth_logs:
+        {{ salt.pillar.get('vector:base_auth_log_collection')|yaml(False)|indent(8) }}
+      transforms:
+        parse_odlvideo_nginx_access_logs:
+          type: remap
+          inputs:
+          - 'collect_odlvideo_nginx_access_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^time=(?P<time>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+\d{2}:\d{2})\sclient=(?P<client>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})\smethod=(?P<method>\S*)\srequest="(?P<request>.*)"\srequest_length=(?P<request_length>\d+)\sstatus=(?P<status>\d+)\sbytes_sent=(?P<bytes_sent>\d+)\sbody_bytes_sent=(?P<body_bytes_sent>\d+)\sreferer=(?P<referer>.*)\suser_agent="(?P<user_agent>.+)"\supstream_addr=(?P<upstream_addr>.+)\supstream_status=(?P<upstream_status>.+)\srequest_time=(?P<request_time>.+)\srequest_id=(?P<request_id>\w+)\supstream_response_time=(?P<upstream_response_time>.+)\supstream_connect_time=(?P<upstream_connect_time>.+)\supstream_header_time=(?P<upstream_header_time>.*)$')
+            if err != null {
+              .parse_error = err
+            }
+            err = null
+
+            . = merge(., parsed)
+            .log_process = "nginx"
+            .log_type = "odl.nginx.access"
+            .environment = "${ENVIRONMENT}"
+      
+            parsed_bs, err = to_int(.bytes_sent)
+            if err == null {
+              .bytes_sent = parsed_bs
+            }
+            err = null
+            parsed_bbs, err = to_int(.body_bytes_sent)
+            if err == null {
+              .body_bytes_sent = parsed_bbs
+            }
+            err = null
+            parsed_rl, err = to_int(.request_length)
+            if err == null {
+              .request_length = parsed_rl
+            }
+            err = null
+            parsed_rt, err = to_float(.request_time)
+            if err == null {
+              .request_time = parsed_rt
+            }
+            err = null
+            parsed_status, err = to_int(.status)
+            if err == null {
+              .status = parsed_status
+            }
+            err = null
+            parsed_usct, err = to_float(.upstream_connect_time)
+            if err == null {
+              .upstream_connect_time = parsed_usct
+            }
+            err = null
+            parsed_usht, err = to_float(.upstream_header_time)
+            if err == null {
+              .upstream_header_time = parsed_usht
+            }
+            err = null
+            parsed_uprt, err = to_float(.upstream_response_time)
+            if err == null {
+              .upstream_response_time = parsed_uprt
+            }
+            err = null
+            parsed_ups, err = to_int(.upstream_response)
+            if err == null {
+              .upstream_status = parsed_ups
+            }
+            err = null
+        filter_healthchecks_odlvideo_nginx_access_logs:
+          inputs:
+          - 'parse_odlvideo_nginx_access_logs'
+          type: filter
+          condition: '! contains!(.user_agent, "ELB-HealthChecker")'
+        parse_odlvideo_nginx_error_logs:
+          type: remap
+          inputs:
+          - 'collect_odlvideo_nginx_error_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^(?P<time>\d{4}/\d{2}/\d{2}\s\d{2}:\d{2}:\d{2})\s\[(?P<severity>.*)\]\s(?P<pid>\d*)#(?P<tid>\d*):\s\*(?P<cid>\d*)\s(?P<message>.*),\sclient:\s(?P<client>.*),\sserver:(?P<server>.*)(?P<additional_content>.*)$')
+            . = merge(., parsed)
+            if err != null {
+              .parse_error = err
+            }
+            .log_process = "nginx"
+            .log_type = "odlvideo.nginx.error"
+            .environment = "${ENVIRONMENT}"
+        parse_odlvideo_django_logs:
+          type: remap
+          inputs:
+          - 'collect_odlvideo_django_logs'
+          source: |
+            event, err = parse_json(.message)
+            if event != null {
+              . = merge!(., event)
+              .log_process = "odlvideo"
+              .log_type = "odlvideo.application"
+              .environment = "${ENVIRONMENT}"
+            }
+        enrich_odlvideo_django_logs:
+          type: aws_ec2_metadata
+          inputs:
+          - 'parse_odlvideo_django_logs'
+          namespace: ec2
+        parse_auth_logs:
+          {{ salt.pillar.get('vector:base_auth_log_parse_source')|yaml(False)|indent(10) }}
+      sinks:
+        ship_odlvideo_logs_to_grafana_cloud:
+          inputs:
+          - 'enrich_odlvideo_django_logs'
+          - 'filter_healthchecks_odlvideo_nginx_access_logs'
+          - 'parse_odlvideo_nginx_error_logs'
+          - 'parse_auth_logs'
+          type: loki
+          labels:
+            environment: ${ENVIRONMENT}
+            application: odlvideo
+            service: odlvideo
+          {{ salt.pillar.get('vector:base_loki_configuration')|yaml(False)|indent(10) }}
diff --git a/pillar/vector/rabbitmq.sls b/pillar/vector/rabbitmq.sls
new file mode 100644
index 000000000..fb1ab0f94
--- /dev/null
+++ b/pillar/vector/rabbitmq.sls
@@ -0,0 +1,51 @@
+vector:
+  extra_configurations:
+  - name: rabbitmq_logs
+    content:
+      log_schema:
+        timestamp_key: vector_timestamp
+        host_key: log_host
+      sources:
+        collect_rabbitmq_application_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          glob_minimum_cooldown_ms: 20000
+          include:
+          - /var/log/rabbitmq/rabbit*.log
+          multiline:
+            start_pattern: r'^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+?) \[(\w+)\]'
+            condition_pattern: r'^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+?) \[(\w+)\] (.*)$'
+            mode: 'halt_before'
+            timeout_ms: 5000
+        collect_auth_logs:
+        {{ salt.pillar.get('vector:base_auth_log_collection')|yaml(False)|indent(8) }}
+      transforms:
+        parse_rabbitmq_application_logs:
+          type: remap
+          inputs:
+          - 'collect_rabbitmq_application_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^(?P<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+?) \[(?P<type>\w+)\] (?P<message>.*)$')
+            if err != null {
+              .parse_error = err
+            }
+            . = merge(., parsed)
+            .log_process = "rabbitmq"
+            .log_type = "rabbitmq.application"
+            .environment = "${ENVIRONMENT}"
+
+        parse_auth_logs:
+          {{ salt.pillar.get('vector:base_auth_log_parse_source')|yaml(False)|indent(10) }}
+
+      sinks:
+        ship_rabbitmq_logs_to_grafana_cloud:
+          inputs:
+          - 'parse_rabbitmq_application_logs'
+          - 'parse_auth_logs'
+          labels:
+            environment: ${ENVIRONMENT}
+            application: rabbitmq
+            service: rabbitmq
+          type: loki
+          {{ salt.pillar.get('vector:base_loki_configuration')|yaml(False)|indent(10) }}
diff --git a/pillar/vector/reddit.sls b/pillar/vector/reddit.sls
new file mode 100644
index 000000000..036957979
--- /dev/null
+++ b/pillar/vector/reddit.sls
@@ -0,0 +1,165 @@
+vector:
+  extra_configurations:
+  - name: reddit_logs
+    content:
+      log_schema:
+        timestamp_key: vector_timestamp
+        host_key: log_host
+      sources:
+        collect_reddit_nginx_access_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/nginx/access.log
+        collect_reddit_nginx_error_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/nginx/error.log
+        collect_reddit_mcrouter_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/mcrouter/mcrouter.log
+        collect_reddit_application_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          include:
+          - /var/log/reddit/reddit.log
+          multiline:
+            start_pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            condition_pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            mode: halt_before
+            timeout_ms: 5000
+        collect_auth_logs:
+        {{ salt.pillar.get('vector:base_auth_log_collection')|yaml(False)|indent(8) }}
+      transforms:
+        # Transforms for NGINX logs
+        parse_reddit_nginx_access_logs:
+          type: remap
+          inputs:
+          - 'collect_reddit_nginx_access_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^time=(?P<time>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+\d{2}:\d{2})\sclient=(?P<client>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})\smethod=(?P<method>\S*)\srequest="(?P<request>.*)"\srequest_length=(?P<request_length>\d+)\sstatus=(?P<status>\d+)\sbytes_sent=(?P<bytes_sent>\d+)\sbody_bytes_sent=(?P<body_bytes_sent>\d+)\sreferer=(?P<referer>.*)\suser_agent="(?P<user_agent>.+)"\supstream_addr=(?P<upstream_addr>.+)\supstream_status=(?P<upstream_status>.+)\srequest_time=(?P<request_time>.+)\srequest_id=(?P<request_id>\w+)\supstream_response_time=(?P<upstream_response_time>.+)\supstream_connect_time=(?P<upstream_connect_time>.+)\supstream_header_time=(?P<upstream_header_time>.*)$')
+            if err != null {
+              .parse_error = err
+            }
+            err = null
+            . = merge(., parsed)
+            .log_process = "nginx"
+            .log_type = "reddit.nginx.access"
+            .environment = "${ENVIRONMENT}"
+            parsed_bs, err = to_int(.bytes_sent)
+            if err == null {
+              .bytes_sent = parsed_bs
+            }
+            err = null
+            parsed_bbs, err = to_int(.body_bytes_sent)
+            if err == null {
+              .body_bytes_sent = parsed_bbs
+            }
+            err = null
+            parsed_rl, err = to_int(.request_length)
+            if err == null {
+              .request_length = parsed_rl
+            }
+            err = null
+            parsed_rt, err = to_float(.request_time)
+            if err == null {
+              .request_time = parsed_rt
+            }
+            err = null
+            parsed_status, err = to_int(.status)
+            if err == null {
+              .status = parsed_status
+            }
+            err = null
+            parsed_usct, err = to_float(.upstream_connect_time)
+            if err == null {
+              .upstream_connect_time = parsed_usct
+            }
+            err = null
+            parsed_usht, err = to_float(.upstream_header_time)
+            if err == null {
+              .upstream_header_time = parsed_usht
+            }
+            err = null
+            parsed_uprt, err = to_float(.upstream_response_time)
+            if err == null {
+              .upstream_response_time = parsed_uprt
+            }
+            err = null
+            parsed_ups, err = to_int(.upstream_response)
+            if err == null {
+              .upstream_status = parsed_ups
+            }
+            err = null
+        filter_healthchecks_reddit_nginx_access_logs:
+          inputs:
+          - 'parse_reddit_nginx_access_logs'
+          type: filter
+          condition: '! contains!(.http_user_agent, "ELB-HealthChecker")'
+        parse_reddit_nginx_error_logs:
+          type: remap
+          inputs:
+          - 'collect_reddit_nginx_error_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^(?P<time>\d{4}/\d{2}/\d{2}\s\d{2}:\d{2}:\d{2})\s\[(?P<severity>.*)\]\s(?P<pid>\d*)#(?P<tid>\d*):\s\*(?P<cid>\d*)\s(?P<message>.*),\sclient:\s(?P<client>.*),\sserver:(?P<server>.*)(?P<additional_content>.*)$')
+            . = merge(., parsed)
+            if err != null {
+              .parse_error = err
+            }
+            .log_process = "nginx"
+            .log_type = "reddit.nginx.error"
+            .environment = "${ENVIRONMENT}"
+        parse_reddit_mcrouter_logs:
+          type: remap
+          inputs:
+          - 'collect_reddit_mcrouter_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'(?P<time>\w\d{4}\s\d{2}:\d{2}:\d{2}.\d{6})\s*(?P<code_value>\d+)\s*(?P<file_name>.*):(?P<line_num>\d+)\]\s*(?P<message>.*)')
+            . = merge(., parsed)
+            if err != null {
+              .parse_error = err
+            }
+            .log_process = "mcrouter"
+            .log_type = "reddit.mcrouter"
+            .environment = "${ENVIRONMENT}"
+        parse_reddit_application_logs:
+          type: remap
+          inputs:
+          - 'collect_reddit_application_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'(?P<asctime>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d{3})\s-(?P<file_name>.*)(?P<line_num>\d+)\s--\s(?P<func_name>\w+)\s(?P<level_name>\[\w+\]):(?s)(?P<message>.*)')
+            . = merge(., parsed)
+            if err != null {
+              .parse_error = err
+            }
+            .log_process = "reddit"
+            .log_type = "reddit.application"
+            .environment = "${ENVIRONMENT}"
+        enrich_reddit_application_logs:
+          type: aws_ec2_metadata
+          inputs:
+          - 'parse_reddit_application_logs'
+          namespace: ec2
+        parse_auth_logs:
+          {{ salt.pillar.get('vector:base_auth_log_parse_source')|yaml(False)|indent(10) }}
+      sinks:
+        ship_reddit_logs_to_grafana_cloud:
+          inputs:
+          - 'filter_healthchecks_reddit_nginx_access_logs'
+          - 'parse_reddit_nginx_error_logs'
+          - 'parse_reddit_mcrouter_logs'
+          - 'enrich_reddit_application_logs'
+          - 'parse_auth_logs'
+          type: loki
+          labels:
+            application: reddit
+            environment: ${ENVIRONMENT}
+            service: reddit
+          {{ salt.pillar.get('vector:base_loki_configuration')|yaml(False)|indent(10) }}
diff --git a/pillar/vector/xqwatcher.sls b/pillar/vector/xqwatcher.sls
new file mode 100644
index 000000000..4da724432
--- /dev/null
+++ b/pillar/vector/xqwatcher.sls
@@ -0,0 +1,115 @@
+vector:
+  extra_configurations:
+  - name: xqueuewatcher_logs
+    content:
+      log_schema:
+        timestamp_key: vector_timestamp
+        host_key: log_host
+      sources:
+        collect_xqwatcher_application_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          glob_minimum_cooldown_ms: 20000
+          include:
+          - /edx/var/log/xqwatcher/xqwatcher.log
+          multiline:
+            start_pattern: r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            condition_pattern: r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            mode: halt_before
+            timeout_ms: 5000
+        collect_xqwatcher_stderr_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          glob_minimum_cooldown_ms: 20000
+          include:
+          - /edx/var/log/supervisor/xqwatcher-stderr.log
+          multiline:
+            start_pattern: r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            condition_pattern: r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            mode: halt_before
+            timeout_ms: 5000
+        collect_xqwatcher_supervisord_logs:
+          type: file
+          read_from: end
+          file_key: log_file
+          glob_minimum_cooldown_ms: 20000
+          include:
+          - /edx/var/log/supervisor/supervisord.log
+          multiline:
+            start_pattern: r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            condition_pattern: r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+'
+            mode: halt_before
+            timeout_ms: 5000
+        collect_auth_logs:
+        {{ salt.pillar.get('vector:base_auth_log_collection')|yaml(False)|indent(8) }}
+      transforms:
+        parse_xqwatcher_application_logs:
+          type: remap
+          inputs:
+          - 'collect_xqwatcher_application_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^(?P<time>^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+)\s-\s(?P<file_name>.*):(?P<line_number>\d+)\s--\s(?P<function_name>\w+)\s\[(?P<log_level>\w+)\]:\s(?P<message>.*)')
+            if err != null {
+              .parse_error = err
+            }
+            . = merge(., parsed)
+            .log_process = "xqwatcher"
+            .log_type = "xqwatcher.application"
+            .environment = "${ENVIRONMENT}"
+        filter_debug_xqwatcher_application_logs:
+          inputs:
+          - 'parse_xqwatcher_application_logs'
+          type: filter
+          condition: .log_level != "DEBUG"
+        parse_xqwatcher_stderr_logs:
+          type: remap
+          inputs:
+          - 'collect_xqwatcher_stderr_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^(?P<time>^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+)\s-\s(?P<file_name>.*):(?P<line_number>\d+)\s--\s(?P<function_name>\w+)\s\[(?P<log_level>\w+)\]:\s(?P<message>.*)')
+            if err != null {
+              .parse_error = err
+            }
+            . = merge(., parsed)
+            .log_process = "xqwatcher"
+            .log_type = "xqwatcher.stderr"
+            .environment = "${ENVIRONMENT}"
+        filter_debug_xqwatcher_stderr_logs:
+          inputs:
+          - 'parse_xqwatcher_stderr_logs'
+          type: filter
+          condition: .log_level != "DEBUG"
+        parse_xqwatcher_supervisord_logs:
+          type: remap
+          inputs:
+          - 'collect_xqwatcher_supervisord_logs'
+          source: |
+            parsed, err = parse_regex(.message, r'^(?P<time>^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+)\s(?P<log_level>\w+)\s(?P<message>.*)')
+            if err != null {
+              .parse_error = err
+            }
+            . = merge(., parsed)
+            .log_process = "xqwatcher"
+            .log_type = "xqwatcher.supervisord"
+            .environment = "${ENVIRONMENT}"
+        filter_debug_xqwatcher_supervisord_logs:
+          inputs:
+          - 'parse_xqwatcher_supervisord_logs'
+          type: filter
+          condition: .log_level != "DEBUG"
+        parse_auth_logs:
+          {{ salt.pillar.get('vector:base_auth_log_parse_source')|yaml(False)|indent(10) }}
+      sinks:
+        ship_xqwatcher_logs_to_grafana_cloud:
+          type: loki
+          inputs:
+          - 'filter_debug_xqwatcher_application_logs'
+          - 'filter_debug_xqwatcher_stderr_logs'
+          - 'filter_debug_xqwatcher_supervisord_logs'
+          - 'parse_auth_logs'
+          labels:
+            environment: ${ENVIRONMENT}
+            applicaiton: xqwatcher
+            service: xqwatcher
diff --git a/salt/vector/configure.sls b/salt/vector/configure.sls
index 4480dd568..f57aabfa0 100644
--- a/salt/vector/configure.sls
+++ b/salt/vector/configure.sls
@@ -5,10 +5,21 @@ ensure_absence_of_default_toml_configuration:
   file.absent:
     - name: /etc/vector/vector.toml
 
-manage_vector_configuration_file:
+manage_vector_base_configuration:
   file.managed:
-    - name: /etc/vector/vector.yaml
+    - name: /etc/vector/host_metrics.yaml
     - contents: |
-        {{ salt.pillar.get('vector:configuration')|yaml(False)|indent(8) }}
+        {{ salt.pillar.get('vector:host_metrics_configuration')|yaml(False)|indent(8) }}
     - onchanges_in:
       - service: vector_service_running
+
+manage_vector_extra_configurations:
+{% set extra_configs = salt.pillar.get('vector:extra_configurations') %}
+{$ for cfg in extra_configs -&
+  file.managed:
+  - name: /etc/vector/{{ cfg.name }}.yaml
+  - contents: |
+      {{ cfg.content|yaml(False)|indent(8) }}
+  - onchanges_in:
+    - service: vector_service_running
+{% endfor %}

From 7f71465d9562ee704ca52acdf23f17acf642b2a9 Mon Sep 17 00:00:00 2001
From: Ardiea <mas48@mit.edu>
Date: Thu, 24 Feb 2022 15:04:43 -0500
Subject: [PATCH 2/2] Adding new vector configuration to the top level
 salt-stack definition.

---
 pillar/top.sls | 8 ++++----
 salt/top.sls   | 4 ++++
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/pillar/top.sls b/pillar/top.sls
index b40397010..a33123ec5 100644
--- a/pillar/top.sls
+++ b/pillar/top.sls
@@ -15,7 +15,7 @@ base:
     - elastic_stack.version_production
   'roles:auth_server':
     - match: grain
-    - fluentd.cas
+    - vector.cas
   'G@roles:elasticsearch and not P@environment:operations*':
     - match: compound
     - fluentd.elasticsearch
@@ -116,7 +116,7 @@ base:
     - nginx
     - nginx.reddit
     - reddit
-    - fluentd.reddit
+    - vector.reddit
   'G@environment:operations and G@roles:redash':
     - match: compound
     - nginx
@@ -183,7 +183,7 @@ base:
   'roles:xqwatcher':
     - match: grain
     - edx.xqwatcher
-    - fluentd.xqwatcher
+    - vector.xqwatcher
   'lightsail-xqwatcher-686':
     - match: glob
     - edx.xqwatcher
@@ -220,7 +220,7 @@ base:
   'roles:rabbitmq':
     - match: grain
     - rabbitmq
-    - fluentd.rabbitmq
+    - vector.rabbitmq
     - consul.rabbitmq
   'roles:tika':
     - match: grain
diff --git a/salt/top.sls b/salt/top.sls
index f56176506..2c5397fd8 100644
--- a/salt/top.sls
+++ b/salt/top.sls
@@ -9,6 +9,7 @@ base:
   'roles:xqwatcher':
     - match: grain
     - edx.xqwatcher
+    - vector
   'roles:amps-redirect':
     - match: grain
     - letsencrypt
@@ -37,6 +38,7 @@ base:
     - match: grain
     - rabbitmq
     - rabbitmq.tests
+    - vector
   'roles:dagster':
     - match: grain
     - mongodb.repository
@@ -58,6 +60,7 @@ base:
     - reddit
     - nginx
     - nginx.certificates
+    - vector
   'roles:cassandra':
     - match: grain
     - cassandra
@@ -72,6 +75,7 @@ base:
     - match: compound
     - utils.configure_debian_source_repos
     - consul
+    - vector
     - python
     - node
     - nginx-shibboleth