compact fail or not

[Baboulinet-33]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

Hello,

I'm trying to install vector via helm with Agent role (chart version 0.33.0), but I don't understand the error with compact function:

error[E103]: unhandled fallible assignment
   ┌─ :38:15
   │
38 │           . = compact(., true)
   │           --- ^^^^^^^^^^^^^^^^ this expression is fallible because at least one argument's type cannot be verified to be valid
   │           │
   │           or change this to an infallible assignment:
   │           ., err = compact(., true)
   │

When I change, I obtain:

error[E104]: unnecessary error assignment
 ┌─ :38:14
 │
38 │           ., err = compact(., true)
 │           -  ^^^   ---------------- because this expression can't fail
 │           │  │
 │           │  this error assignment is unnecessary
 │           use: . = compact(., true)

What am I suppose to do ?

Configuration

No response

Version

0.38.0-distroless-libc

Debug Output

No response

Example Data

No response

Additional Context

No response

References

No response

[jszwedko]

Hi @Baboulinet-33 !

I'm having trouble reproducing this. Could you share the full configuration?

[Baboulinet-33]

The full config:

expire_metrics_secs = 60
data_dir = "/var/lib/vector/openshift-logging/forwarder"

[api]
enabled = true

[sources.internal_metrics]
type = "internal_metrics"

# Logs from containers (including openshift containers)
[sources.input_cncs2_integ_container]
type = "kubernetes_logs"
max_read_bytes = 3145728
glob_minimum_cooldown_ms = 15000
auto_partial_merge = true
include_paths_glob_patterns = ["/var/log/pods/mi-*/*/*.log"]
exclude_paths_glob_patterns = ["/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.log.*", "/var/log/pods/*/*/*.tmp", "/var/log/pods/default_*/*/*.log", "/var/log/pods/kube*_*/*/*.log", "/var/log/pods/openshift*_*/*/*.log"]
pod_annotation_fields.pod_labels = "kubernetes.labels"
pod_annotation_fields.pod_namespace = "kubernetes.namespace_name"
pod_annotation_fields.pod_annotations = "kubernetes.annotations"
pod_annotation_fields.pod_uid = "kubernetes.pod_id"
pod_annotation_fields.pod_node_name = "hostname"
namespace_annotation_fields.namespace_uid = "kubernetes.namespace_id"
rotate_wait_ms = 5000

[transforms.input_cncs2_integ_container_viaq]
type = "remap"
inputs = ["input_cncs2_integ_container"]
source = '''
  .openshift.cluster_id = "${OPENSHIFT_CLUSTER_ID:-}"
  if !exists(.level) {
    .level = "default"
    if match!(.message, r'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"|<warn>') {
      .level = "warn"
    } else if match!(.message, r'Error|ERROR|^E[0-9]+|level=error|Value:error|"level":"error"|<error>') {
      .level = "error"
    } else if match!(.message, r'Critical|CRITICAL|^C[0-9]+|level=critical|Value:critical|"level":"critical"|<critical>') {
      .level = "critical"
    } else if match!(.message, r'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"|<debug>') {
      .level = "debug"
    } else if match!(.message, r'Notice|NOTICE|^N[0-9]+|level=notice|Value:notice|"level":"notice"|<notice>') {
      .level = "notice"
    } else if match!(.message, r'Alert|ALERT|^A[0-9]+|level=alert|Value:alert|"level":"alert"|<alert>') {
      .level = "alert"
    } else if match!(.message, r'Emergency|EMERGENCY|^EM[0-9]+|level=emergency|Value:emergency|"level":"emergency"|<emergency>') {
      .level = "emergency"
    } else if match!(.message, r'(?i)\b(?:info)\b|^I[0-9]+|level=info|Value:info|"level":"info"|<info>') {
      .level = "info"
      }
  }
  pod_name = string!(.kubernetes.pod_name)
  if starts_with(pod_name, "eventrouter-") {
    parsed, err = parse_json(.message)
    if err != null {
      log("Unable to process EventRouter log: " + err, level: "info")
    } else {
      ., err = merge(.,parsed)
      if err == null && exists(.event) && is_object(.event) {
          if exists(.verb) {
            .event.verb = .verb
            del(.verb)
          }
          .kubernetes.event = del(.event)
          .message = del(.kubernetes.event.message)
          set!(., ["@timestamp"], .kubernetes.event.metadata.creationTimestamp)
          del(.kubernetes.event.metadata.creationTimestamp)
          . = compact(., nullish: true)
      } else {
        log("Unable to merge EventRouter log message into record: " + err, level: "info")
      }
    }
  }
  del(.source_type)
  del(.stream)
  del(.kubernetes.pod_ips)
  del(.kubernetes.node_labels)
  del(.timestamp_end)
  ts = del(.timestamp); if !exists(."@timestamp") {."@timestamp" = ts}
'''

# Set log_type
[transforms.input_cncs2_integ_viaq_logtype]
type = "remap"
inputs = ["input_cncs2_integ_container_viaq"]
source = '''
  .log_type = "application"
'''

[transforms.pipeline_cncs2_integ_archive_openshiftlabels_0]
type = "remap"
inputs = ["input_cncs2_integ_viaq_logtype"]
source = '''
  .openshift.labels = {"clustername":"integ-cncs2"}
'''

[transforms.output_httpout_app_normalize]
type = "remap"
inputs = ["pipeline_cncs2_integ_archive_openshiftlabels_0"]
source = '''
  del(.file)
'''

[transforms.output_httpout_app_dedot]
type = "remap"
inputs = ["output_httpout_app_normalize"]
source = '''
  .openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  if exists(.kubernetes.namespace_labels) {
      for_each(object!(.kubernetes.namespace_labels)) -> |key,value| { 
        newkey = replace(key, r'[\./]', "_") 
        .kubernetes.namespace_labels = set!(.kubernetes.namespace_labels,[newkey],value)
        if newkey != key {
          .kubernetes.namespace_labels = remove!(.kubernetes.namespace_labels,[key],true)
        }
      }
  }
  if exists(.kubernetes.labels) {
      for_each(object!(.kubernetes.labels)) -> |key,value| { 
        newkey = replace(key, r'[\./]', "_") 
        .kubernetes.labels = set!(.kubernetes.labels,[newkey],value)
        if newkey != key {
          .kubernetes.labels = remove!(.kubernetes.labels,[key],true)
        }
      }
  }
'''

[sinks.output_httpout_app]
type = "http"
inputs = ["output_httpout_app_dedot"]
uri = "http://vdebug.infra-vector.svc.cluster.local:8080"
method = "post"

[sinks.output_httpout_app.encoding]
codec = "json"

[sinks.output_httpout_app.request]
headers = {"h1"="v1","h2"="v2"}

[transforms.add_nodename_to_metric]
type = "remap"
inputs = ["internal_metrics"]
source = '''
.tags.hostname = get_env_var!("VECTOR_SELF_NODE_NAME")
'''

[sinks.prometheus_output]
type = "prometheus_exporter"
inputs = ["add_nodename_to_metric"]
address = "[::]:24231"
default_namespace = "collector"

# [sinks.prometheus_output.tls]
# enabled = true
# key_file = "/etc/collector/metrics/tls.key"
# crt_file = "/etc/collector/metrics/tls.crt"
# min_tls_version = "VersionTLS12"
# ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"

[jszwedko]

I get different errors running that configuration with v0.38.0 like:

2024-06-17T18:22:34.540205Z ERROR vector::cli: Configuration error. error=unknown field namespace_uid, expected namespace_labels

Are you sure that's the config you are running and that the version is v0.38.0?

[Baboulinet-33]

Hop sorry, I've take the old one generated from openshift.

(Edit: the whole manifest used is here: https://github.com/Baboulinet-33/vector-log-forwarder-test/tree/main)

The "correct" one:

    expire_metrics_secs: 60
    data_dir: /vector-data-dir
    api:
      enabled: true
      address: 127.0.0.1:8686
    sources:
      internal_metrics:
        type: internal_metrics
      input_cncs2_integ_container:
        type: kubernetes_logs
        max_read_bytes: 3145728
        glob_minimum_cooldown_ms: 15000
        auto_partial_merge: true
        include_paths_glob_patterns:
          - /var/log/pods/my-projects_*/*/*.log
        exclude_paths_glob_patterns:
          - /var/log/pods/*/*/*.gz
          - /var/log/pods/*/*/*.log.*
          - /var/log/pods/*/*/*.tmp
          - /var/log/pods/default_*/*/*.log
          - /var/log/pods/kube*_*/*/*.log
          - /var/log/pods/openshift*_*/*/*.log
        pod_annotation_fields:
          pod_labels: kubernetes.labels
          pod_namespace: kubernetes.namespace_name
          pod_annotations: kubernetes.annotations
          pod_uid: kubernetes.pod_id
          pod_node_name: hostname
        namespace_annotation_fields:
          namespace_labels: kubernetes.namespace_id
        rotate_wait_secs: 5
    transforms:
      input_cncs2_integ_container_viaq:
        type: remap
        inputs:
          - input_cncs2_integ_container
        source: >2
            .openshift.cluster_id = "${OPENSHIFT_CLUSTER_ID:-}"
            if !exists(.level) {
              .level = "default"
              if match!(.message, r'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"|<warn>') {
                .level = "warn"
              } else if match!(.message, r'Error|ERROR|^E[0-9]+|level=error|Value:error|"level":"error"|<error>') {
                .level = "error"
              } else if match!(.message, r'Critical|CRITICAL|^C[0-9]+|level=critical|Value:critical|"level":"critical"|<critical>') {
                .level = "critical"
              } else if match!(.message, r'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"|<debug>') {
                .level = "debug"
              } else if match!(.message, r'Notice|NOTICE|^N[0-9]+|level=notice|Value:notice|"level":"notice"|<notice>') {
                .level = "notice"
              } else if match!(.message, r'Alert|ALERT|^A[0-9]+|level=alert|Value:alert|"level":"alert"|<alert>') {
                .level = "alert"
              } else if match!(.message, r'Emergency|EMERGENCY|^EM[0-9]+|level=emergency|Value:emergency|"level":"emergency"|<emergency>') {
                .level = "emergency"
              } else if match!(.message, r'(?i)\b(?:info)\b|^I[0-9]+|level=info|Value:info|"level":"info"|<info>') {
                .level = "info"
              }
            }
            pod_name = string!(.kubernetes.pod_name)
            if starts_with(pod_name, "eventrouter-") {
              parsed, err = parse_json(.message)
              if err != null {
                log("Unable to process EventRouter log: " + err, level: "info")
              } else {
                ., err = merge(.,parsed)
                if err == null && exists(.event) && is_object(.event) {
                    if exists(.verb) {
                      .event.verb = .verb
                      del(.verb)
                    }
                    .kubernetes.event = del(.event)
                    .message = del(.kubernetes.event.message)
                    set(., ["@timestamp"], .kubernetes.event.metadata.creationTimestamp)
                    del(.kubernetes.event.metadata.creationTimestamp)
                    . = compact(., nullish: true)
                } else {
                  log("Unable to merge EventRouter log message into record: " + err, level: "info")
                }
              }
            }
            del(.source_type)
            del(.stream)
            del(.kubernetes.pod_ips)
            del(.kubernetes.node_labels)
            del(.timestamp_end)
            ts = del(.timestamp); if !exists(."@timestamp") {."@timestamp" = ts}
      input_cncs2_integ_viaq_logtype:
        type: remap
        inputs:
          - input_cncs2_integ_container_viaq
        source: |2
            .log_type = "application"
      pipeline_cncs2_integ_archive_openshiftlabels_0:
        type: remap
        inputs:
          - input_cncs2_integ_viaq_logtype
        source: |2
            .openshift.labels = {"clustername":"integ-cncs2"}
      output_httpout_app_normalize:
        type: remap
        inputs:
          - pipeline_cncs2_integ_archive_openshiftlabels_0
        source: |2
            del(.file)
      output_httpout_app_dedot:
        type: remap
        inputs:
          - output_httpout_app_normalize
        source: >2
            .openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
            if exists(.kubernetes.namespace_labels) {
                for_each(object!(.kubernetes.namespace_labels)) -> |key,value| { 
                  newkey = replace(key, r'[\./]', "_") 
                  .kubernetes.namespace_labels = set!(.kubernetes.namespace_labels,[newkey],value)
                  if newkey != key {
                    .kubernetes.namespace_labels = remove!(.kubernetes.namespace_labels,[key],true)
                  }
                }
            }
            if exists(.kubernetes.labels) {
                for_each(object!(.kubernetes.labels)) -> |key,value| { 
                  newkey = replace(key, r'[\./]', "_") 
                  .kubernetes.labels = set!(.kubernetes.labels,[newkey],value)
                  if newkey != key {
                    .kubernetes.labels = remove!(.kubernetes.labels,[key],true)
                  }
                }
            }
      add_nodename_to_metric:
        type: remap
        inputs:
          - internal_metrics
        source: |
          .tags.hostname = get_env_var!("VECTOR_SELF_NODE_NAME")
    sinks:
      output_httpout_app:
        type: http
        inputs:
          - output_httpout_app_dedot
        uri: http://vector.infra-vector.svc.cluster.local:8080
        method: post
        encoding:
          codec: json

[jszwedko]

Thanks @Baboulinet-33 . This looks like a type inference bug in VRL caused by the set function. I'm not sure why it is happening, but you can workaround it by changing:

set(., ["@timestamp"], .kubernetes.event.metadata.creationTimestamp)

to

."@timestamp" = .kubernetes.event.metadata.creationTimestamp

As a note, set(., ["@timestamp"], .kubernetes.event.metadata.creationTimestamp) also actually won't do anything because it returns a modified copy rather than modifying in place. You would want to use . = set(., ["@timestamp"], .kubernetes.event.metadata.creationTimestamp), but that fails to compile still too.

[Baboulinet-33]

Thanks for your time, it's now working :)