Alloy 1.8.1 - loki.source.file / local.file_match - decompressor gz - Panic

[Nachtfalkeaw]

What's wrong?

Hello,
I use local.file_match to detect .gz files. loki.source.file and decompressor to decode .gz files.
If I delete a gz file which is on the disk I get a panic.

Steps to reproduce

Delete a .gz file which is monitored by local.file_match and loki.source.file

System information

RHEL 7 - Appliance

Software version

Grafana Alloy 1.8.1

Configuration

This is the part of my alloy config

local.file_match "filename" {

            // Dateien, die erkannt werden sollen
  path_targets = [
    {__path__ = "/var/log/filename.1.gz"},
    {__path__ = "/var/log/filename.2.gz"},
  ]

  sync_period = "10s"
}

//
// ==========================================================================================================================
//

loki.source.file "filename" {

  targets = local.file_match.filename.targets

  file_watch {
    min_poll_frequency = "1s"
    max_poll_frequency = "16s"
  }

  decompression {
    enabled       = true
    initial_delay = "10s"
    format        = "gz"
  }

  forward_to = [loki.relabel.filename.receiver]
}

Logs

panic: close of nil channel

goroutine 26318 [running]:
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition.func1()
        /drone/src/internal/component/loki/source/file/decompresser.go:190 +0x105
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition(0xc00cb0d800)
        /drone/src/internal/component/loki/source/file/decompresser.go:201 +0x267
created by github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).Run in goroutine 26315
        /drone/src/internal/component/loki/source/file/decompresser.go:176 +0x4d1
panic: close of nil channel

goroutine 38433 [running]:
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition.func1()
        /drone/src/internal/component/loki/source/file/decompresser.go:190 +0x105
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition(0xc00cb0d900)
        /drone/src/internal/component/loki/source/file/decompresser.go:201 +0x267
created by github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).Run in goroutine 26314
        /drone/src/internal/component/loki/source/file/decompresser.go:176 +0x4d1

[Nachtfalkeaw]

Some more panic logs:

panic: close of nil channel

goroutine 19184 [running]:
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition.func1()
        /drone/src/internal/component/loki/source/file/decompresser.go:190 +0x105
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition(0xc007f04200)
        /drone/src/internal/component/loki/source/file/decompresser.go:201 +0x267
created by github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).Run in goroutine 390
        /drone/src/internal/component/loki/source/file/decompresser.go:176 +0x4d1
panic: close of nil channel

goroutine 28851 [running]:
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition.func1()
        /drone/src/internal/component/loki/source/file/decompresser.go:190 +0x105
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition(0xc007f04200)
        /drone/src/internal/component/loki/source/file/decompresser.go:201 +0x267
created by github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).Run in goroutine 390
        /drone/src/internal/component/loki/source/file/decompresser.go:176 +0x4d1
panic: close of nil channel

goroutine 1115 [running]:
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition.func1()
        /drone/src/internal/component/loki/source/file/decompresser.go:190 +0x105
github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).updatePosition(0xc007f04200)
        /drone/src/internal/component/loki/source/file/decompresser.go:201 +0x267
created by github.com/grafana/alloy/internal/component/loki/source/file.(*decompressor).Run in goroutine 390
        /drone/src/internal/component/loki/source/file/decompresser.go:176 +0x4d1

[aglees]

We're seeing an issue with standard Kubernetes logs, and have had to rollback v1.8.1 to v1.6.1 in our churning DEV environment because logs on short-lived nodes are being missed. We're seeing something like the initial Daemonset pods getting logs captured, but it appears that we're still getting a panic and no further container logs being recorded.

[kalleep]

@aglees are you experiencing panic with decompression too? I managed to reproduce it locally with the config and steps provided by @Nachtfalkeaw so just wanted to make sure there is not other problems as well

[Nachtfalkeaw]

I do not receive a panic when alloy is decompressing the file, however alloy seems to stop decompression if this issue occurs.
#2736

As far as I can see it starts decompression, parsing the logs and if the error occurs it stops to continue to decompress and parse the other logs.

PS:
Decompression has a disadvantage in my environment why I don't use it:
my log file is plain text (filename.log) and after log rotation it is immediately moved into "filename.log.1.gz".
The decompressor block or local.file_match does not allow to use "mixed" logs (compressed and uncompressed")in the same config.

[kalleep]

I did not meant that this happen during decompression, just that I have only managed to get the panic when its enabled but haven't looked into why this happens yet

[aglees]

@kalleep sadly with rolling Alloy back to v1.6.1 in DEV we're going to have a harder time gathering those logs; our lower environments don't churn at anything like the same rate, therefore new nodes are spawned infrequently which is when we're going to see the errors.

In terms of configuration, I can share this, which I think is pretty vanilla as far as the loki.source.file is concerned:

loki.process "logs_default_journal" {
  forward_to = [loki.write.logs_default.receiver]

  stage.drop {
    drop_counter_reason = "kubelet_getters_path_does_not_exist"
    expression          = ".*Path does not exist.*"
  }

  stage.drop {
    drop_counter_reason = "kubelet_cadvisor_failed_update_stats"
    expression          = ".*Failed to update stats for container.*"
  }

  stage.drop {
    drop_counter_reason = "kubelet_cadvisor_failed_filesystem_stats"
    expression          = ".*failed to collect filesystem stats.*"
  }

  stage.structured_metadata {
      values = {
        priority  = "",
      }
  }

  stage.label_drop {
    values = [ "priority" ]
  }

  stage.static_labels {
    values = {
      source = "systemd",
    }
  }
}

discovery.relabel "logs_default_journal" {
  targets = []

  rule {
    source_labels = ["__journal__hostname"]
    target_label  = "hostname"
  }

  rule {
    source_labels = ["__journal__systemd_unit"]
    target_label  = "unit"
  }

  rule {
    source_labels = ["unit"]
    regex         = "kubelet\\.service|container1_runtime\\.service|container2_runtime\\.service"
    action        = "keep"
  }

  rule {
    source_labels = ["__journal_priority_keyword"]
    target_label  = "priority"
  }
}

loki.source.journal "logs_default_journal" {
  max_age       = "12h0m0s"
  path          = "/var/log/journal"
  relabel_rules = discovery.relabel.logs_default_journal.rules
  forward_to    = [loki.process.logs_default_journal.receiver]
  labels        = {
    job = "systemd-journal",
  }
}

discovery.kubelet "logs_kubelet_pod_logs" {
  bearer_token_file = "/var/run/secrets/kubernetes.io/serviceaccount/token"

  tls_config {
    insecure_skip_verify = true
  }
}

discovery.relabel "logs_default_pod_logs" {
  targets = discovery.kubelet.logs_kubelet_pod_logs.targets

  rule {
    source_labels = ["__meta_kubernetes_pod_node_name"]
    target_label  = "k8s_node_name"
  }

  rule {
    source_labels = ["__meta_kubernetes_pod_label_name"]
    target_label  = "name"
  }

  rule {
    regex        = "__meta_kubernetes_pod_label_(app|component|k8s_app|tier)"
    target_label = "$1"
    action       = "labelmap"
  }

  rule {
    regex        = "__meta_kubernetes_pod_label_app_kubernetes_io_(component|instance|managed_by|name|part_of|version)"
    target_label = "$1"
    action       = "labelmap"
  }

  rule {
    regex        = "__meta_kubernetes_pod_label_app_acmecompany_com_(environment|manifest|node_group|customerName|tier)"
    target_label = "$1"
    action       = "labelmap"
  }

  rule {
    regex        = "__meta_kubernetes_node_label_node_kubernetes_io_(.+)"
    target_label = "$1"
    action       = "labelmap"
  }

  rule {
    regex        = "__meta_kubernetes_node_label_topology_kubernetes_io_(.+)"
    target_label = "$1"
    action       = "labelmap"
  }

  rule {
    source_labels = ["__meta_kubernetes_pod_annotation_kubernetes_io_config_source"]
    target_label  = "config_source"
  }

  rule {
    source_labels = ["__meta_kubernetes_node_label_workload"]
    target_label  = "node_workload"
  }

  rule {
    source_labels = ["__meta_kubernetes_node_label_node_asg"]
    target_label  = "node_asg"
  }

  rule {
    regex  = "__meta_kubernetes_pod_label_(.+)"
    action = "labeldrop"
  }

  rule {
    source_labels = ["__meta_kubernetes_namespace"]
    target_label  = "namespace"
  }

  rule {
    source_labels = ["__meta_kubernetes_pod_name"]
    target_label  = "pod"
  }

  rule {
    source_labels = ["__meta_kubernetes_pod_container_name"]
    target_label  = "container"
  }

  rule {
    source_labels = ["__meta_kubernetes_pod_controller_kind"]
    target_label  = "workload_type"
  }

  rule {
    source_labels = ["__meta_kubernetes_pod_uid", "__meta_kubernetes_pod_container_name"]
    separator     = "/"
    target_label  = "__path__"
    replacement   = "/var/log/pods/*$1/*.log"
  }

  rule {
    source_labels = ["container", "component"]
    separator     = ","
    regex         = "reverseproxy,gateway"
    action        = "drop"
  }
}

local.file_match "logs_default_pod_logs" {
  path_targets = discovery.relabel.logs_default_pod_logs.output
}

loki.process "logs_default_pod_logs" {
  forward_to = [loki.write.logs_default.receiver]

  stage.cri { }

  stage.match {
    selector            = "{namespace=~\"alternate-logging-tool|logging\"}"
    action              = "drop"
    drop_counter_reason = "logging_ns"
  }

  stage.match {
    selector = "{namespace=\"loki\", part_of=\"memberlist\"} |~ \"info\""

    stage.static_labels {
      values = {
        level          = "info",
        loki_info_logs = "true",
      }
    }
  }

  stage.match {
    selector = "{namespace=~\".+\"} |~ \"(.*kube-probe.*|.*/ping.*|.*/heartbeat.*)\""

    stage.static_labels {
      values = {
        low_retention = "true",
        flag          = "low_retention",
      }
    }
  }

  stage.match {
    selector = "{namespace=~\".+\"}"

    stage.labels {
      values = {
        service_name = "k8s_app",
      }
    }

    stage.structured_metadata {
      values = {
        instance = "",
        k8s_node_name = "",
        pod = "",
        stream = "",
        version = "",
      }
    }
  }

  stage.match {
    selector = "{namespace=~\"customer-.+\"}"

    stage.json {
      expressions = {
        event = "",
        level = "",
        tags  = "",
      }
    }

    stage.labels {
      values = {
        level = "",
        tags  = "",
      }
    }

    stage.structured_metadata {
      values = {
        container = "",
        event     = "",
      }
    }
  }

  stage.match {
    selector = "{namespace=~\"customer-.+\"} |~ \"application-tag1|application-tag2\""

    stage.static_labels {
      values = {
        features = "true",
        flag     = "features",
      }
    }
  }

  stage.match {
    selector = "{config_source=\"file\"}"

    stage.labels {
      values = {
        app = "tier",
        service_name = "component",
      }
    }

    stage.static_labels {
      values = {
        workload_type = "Pod",
      }
    }

    stage.label_drop {
      values = [ "tier" ]
    }
  }

  stage.static_labels {
    values = {
      source = "k8s",
    }
  }

  stage.label_drop {
    values = [ "ami_id", "config_source", "container", "filename", "instance", "instance_type", "k8s_app", "k8s_node_name", "node", "node_asg", "node_name", "pod", "stream", "version", "zone" ]
  }

  stage.match {
    selector = "{service_name=\"cniprovider-node\"}"

    stage.static_labels {
      values = {
        low_retention = "true",
        flag          = "low_retention",
      }
    }
  }
}

loki.source.file "logs_default_pod_logs" {
  targets               = local.file_match.logs_default_pod_logs.targets
  forward_to            = [loki.process.logs_default_pod_logs.receiver]
  legacy_positions_file = "/wal/logs/positions.yaml"
}

local.file_match "match_k8s_audit_logs" {
    path_targets = [{"__path__" = "/var/log/apiserver/audit.log"}]
}

loki.source.file "source_k8s_audit_logs" {
    targets    = local.file_match.match_k8s_audit_logs.targets
    legacy_positions_file = "/wal/logs/positions.yaml"

    forward_to = [loki.process.mutate_labels_k8s_audit_logs.receiver]
}

loki.process "mutate_labels_k8s_audit_logs" {

  stage.static_labels {
    values = {
      low_retention = "true",
      source        = "k8s-audit",
      service_name  = "k8s-audit",
    }
  }

  stage.label_drop {
    values = ["filename"]
  }

  forward_to = [loki.write.logs_default.receiver]
}

loki.write "logs_default" {
  endpoint {
    url       = "{{ .Values.lokiUrl }}"
    customer_id = "{{ .Values.clusterName }}"

    basic_auth {
      username      = "{{ .Values.basicAuth.loki.username }}"
      password_file = "/etc/basic-auth/password"
    }
  }
  external_labels = {
    cluster = "{{ .Values.clusterName }}",
  }
}

prometheus.exporter.self "alloy" {
}

logging {
  level  = "{{ .Values.node.logging.level | default "info" }}"
  format  = "{{ .Values.node.logging.format | default "logfmt" }}"
}

[kalleep]

Thanks for the config, pretty sure the issues was introduced in #3191

[kalleep]

Fixed in #3475.

I plan to release 1.8.3 on Monday and include this fix and #3488